feat: initial implementation of website integrity scanner

Python-Package zur unabhängigen Überwachung statischer Websites gegen
SEO-Spam und unbefugte Manipulationen. Läuft außerhalb des Hosters.

Kernfunktionen:
- Reiner Python-Crawler (bs4+lxml), kein wget
- Baseline-Management mit manuellem Freigabe-Workflow (niemals automatisch)
- Text-/Link-/Meta-Diff mit Risiko-Scoring (grün/gelb/rot)
- Hidden-Content-Erkennung (CSS inline, noscript, Kommentar-Links)
- Whitelist für externe Domains und interne Pfade
- E-Mail + Webhook Alarmierung
- CLI: init, crawl, check, scan, approve, report, status
- 79 Unit-Tests (pytest), alle grün

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Your Name 2026-06-12 00:51:41 +02:00
commit 98e8d11eb5
24 changed files with 4295 additions and 0 deletions

45
.gitignore vendored Normal file
View file

@ -0,0 +1,45 @@
# Python
__pycache__/
*.py[cod]
*.pyo
*.pyd
.Python
*.egg-info/
dist/
build/
.eggs/
*.egg
# Virtual environments
venv/
.venv/
env/
.env/
# Tests / Coverage
.pytest_cache/
.coverage
htmlcov/
.tox/
# Runtime data (never commit — may contain crawled content or sensitive diffs)
data/snapshots/
data/baseline/
data/pending/
reports/
logs/
# Editor
.idea/
.vscode/
*.swp
*.swo
*~
# OS
.DS_Store
Thumbs.db
# Secrets (extra safeguard)
*.env
.env.*

83
CLAUDE.md Normal file
View file

@ -0,0 +1,83 @@
# CLAUDE.md
This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
## Commands
```bash
# Install dependencies
pip install -r requirements.txt
# Run all tests
pytest tests/ -v
# Run a single test file
pytest tests/test_extractor.py -v
# Run a single test
pytest tests/test_differ.py::TestScoreDiff::test_green_for_no_changes -v
# CLI usage (always run from project root where config.yaml lives)
python -m scanner status
python -m scanner init
python -m scanner scan
python -m scanner approve --all --note "Reason"
python -m scanner approve --url https://bredelar.info/page/
python -m scanner report
python -m scanner --verbose scan
```
## Architecture
The scanner is a Python package (`scanner/`) invoked via `python -m scanner`. All subcommands share one data flow:
```
Crawler → Extractor → BaselineManager (snapshot)
differ.compare_snapshots
differ.score_diff → alerter.send_alert
__main__._write_report (JSON + MD)
```
**Key design constraint**: The baseline is immutable except through an explicit `approve` call. `compare_snapshots` and `score_diff` are pure functions — they never write anything.
### Module responsibilities
- **`config.py`** — loads `config.yaml`, deep-merges onto `DEFAULT_CONFIG`. All paths (data_dir, reports_dir, logs_dir, config_dir) are resolved to absolute paths by `resolve_paths()` before being passed to other modules.
- **`crawler.py`** — breadth-first crawl using `requests`. `normalize_url()` is the canonical URL form used everywhere (strips fragment, lowercases host, removes noise query params, sorts remaining params). `should_crawl()` filters out binary assets and tracking URLs.
- **`extractor.py`** — parses HTML with bs4+lxml. Returns a single dict per page with keys: `text`, `links`, `metadata`, `hidden_content`, `inline_scripts`, `jsonld`, `comment_links`. The `links` dict has typed sub-lists (`a`, `link_rel`, `script`, `img`, `iframe`, `form`, `meta_refresh`, `canonical`).
- **`baseline.py`** — `BaselineManager` handles two separate stores: `data/snapshots/YYYYMMDD_HHMMSS/` (raw crawl output) and `data/baseline/` (approved reference). Each page is stored as `<url_key(url)>.json`. `approve_*` methods are the only path into the baseline.
- **`differ.py`** — `compare_snapshots(baseline_pages, current_pages, cfg)` returns a flat diff dict. `score_diff(diff, cfg)` maps it to `{score, level, reasons, exit_code}`. Thresholds: yellow ≥ 20, red ≥ 60 (configurable). A single new external domain = 50 pts (yellow); combined with hidden content = 90 pts (red).
- **`checker.py`** — stateless helper functions for whitelist enforcement, security-header auditing, canonical hijack detection. Called from `__main__.cmd_check`, not from `differ`.
- **`alerter.py`** — sends alert only when `level >= min_level`. SMTP password is read from the env var named in `smtp_password_env` (default: `SCANNER_SMTP_PASSWORD`).
- **`__main__.py`** — wires everything together. `cmd_scan` = `cmd_crawl` (crawl + save snapshot) + `cmd_check` (load baseline + snapshot, diff, score, report, alert). Exit codes: 0=green, 1=yellow, 2=red.
### Data layout
```
data/baseline/manifest.json # approved_at, approved_by, note, urls[]
data/baseline/pages/<16hex>.json # one file per approved URL
data/snapshots/YYYYMMDD_HHMMSS/ # one dir per crawl run
reports/YYYYMMDD_HHMMSS/report.{json,md}
config/allowed_external.yaml # set of permitted external domains
config/allowed_paths.yaml # set of permitted internal paths (optional)
config/ignore_rules.yaml # normalization selectors + regex patterns
```
### Scoring weights (defaults, all configurable in `config.yaml`)
| Event | Points |
|---|---|
| New external domain | 50 |
| Hidden content (CSS-hidden element with links/text) | 40 |
| Meta-Refresh | 40 |
| Unexpected canonical | 35 |
| Unexpected JSON-LD @type | 35 |
| New internal URL | 30 |
| New suspicious inline script | 30 |
| Comment-embedded link | 25 |
| Large text addition (>200 chars) | 20 |
| Missing internal URL | 20 |
| Missing security header | 5 |

249
Keller/PROMPT.md Normal file
View file

@ -0,0 +1,249 @@
Erstelle bitte ein produktionsreifes, möglichst schlankes Kommandozeilen-Kontroll-Programm in Python zur unabhängigen Überwachung der kleinen, statischen Website https://bredelar.info (einstellbar) gegen Manipulationen. Es soll manuell oder als Cron-Programm einsetzbar sein.
Ziel
Das Programm soll im Verantwortungsbereich des Betreibers laufen, nicht beim beauftragten Web-Dienstleister. Es soll regelmäßig, z. B. täglich per Cron um 03:00 Uhr (einstellbar) , die gesamte Web-Repräsentanz von bredelar.info überwachen und Alarm auslösen, wenn unerwartete Änderungen, neue Seiten, unerlaubte Links oder problematische Inhalte auftauchen. Der Fokus liegt auf robuster Erkennung bei möglichst geringem Betriebsaufwand. Die Website ist klein und soll klein bleiben.
Wichtige Rahmenbedingungen
- Die Website ist klein, vermutlich mit wenigen Seiten.
- Es gab bereits einen SEO-Spam-/Parasite-SEO-artigen Vorfall.
- Angreifer können Texte thematisch passend tarnen; reine Keyword-Heuristik reicht daher nicht.
- Das System soll unabhängig vom Dienstleister arbeiten.
- Das System soll auf einem kleinen Linux-VHost oder Minimalserver laufen können.
- Bevorzugte Sprachen/Tools: Python 3 oder Shell, aber bitte pragmatisch entscheiden.
- Ergebnis soll einfach installierbar, wartbar und nachvollziehbar sein.
- Keine unnötig komplexe Infrastruktur, keine Kubernetes-, Cloud-, Queue- oder Datenbank-Overkill-Lösung.
Die Lösung soll diese drei Kernprobleme adressieren
1. Integritätsprüfung / Diff-Monitoring
- Prüfe regelmäßig alle relevanten Seiten von bredelar.info.
- Erstelle Snapshots einer „gesunden“ Referenzversion.
- Vergleiche spätere Abrufe mit dieser Referenz.
- Alarmiere bei jeder relevanten Abweichung.
- Alarm muss detailliert sein: betroffene URL, Art der Änderung, alter/neuer Zustand, neue/entfernte Links, neue/entfernte Textelemente, neue/entfernte HTML-Elemente, Risiko-Einstufung.
- Änderungen durch legitime Pflege müssen ebenfalls erkannt werden; sie dürfen erst nach manueller Freigabe zur neuen Referenz/Baseline werden.
- Wichtig: Das System darf kompromittierte Inhalte niemals automatisch als neue „gesunde“ Basis übernehmen.
2. Link-Kontrolle / Whitelist
- Finde alle offenen und verdeckten Links auf der gesamten Web-Repräsentanz.
- Prüfe sichtbare Links, versteckte Links, Links in Metadaten, Canonical, hreflang, redirects, iframe/src, script/src, img/src, form action, meta refresh, JSON-LD, strukturierte Daten, noscript-Bereiche, Kommentare, data-Attribute, CSS-basierte Hinweise, falls sinnvoll.
- Führe eine Whitelist erlaubter externer Ziele und eine Whitelist erlaubter interner URL-Pfade.
- Jede neue externe Ziel-Domain oder unerwartete interne URL soll Alarm auslösen.
- SEO-Spam braucht Links; daher ist das ein zentraler Kontrollpunkt.
3. Problematische Inhalte
- Erkenne auch problematische Inhalte ohne Links, z. B. Pornografie, Dirty Talk, politische Propaganda, diffamierende oder strafbare Inhalte, widersprüchliche Aussagen, versteckte Spam-Texte oder thematisch unpassende kommerzielle Inhalte.
- Nutze dafür einfache robuste Regeln plus optional eine zusätzliche semantische/KI-gestützte Analyse.
- Aber: KI/semantische Analyse ist Ergänzung, nicht Kern. Kern bleibt Integritäts- und Link-Monitoring.
- Gute Angreifer können Dorf-/Heimat-Begriffe einbauen; daher bitte nicht auf simple White-/Blacklist-Keywords vertrauen.
Ganz wichtige Zusatzanforderung: „Gesunde“ Datenausgangslage und Pflege
Das System muss sauber mit einer Referenz-/Baseline-Verwaltung umgehen:
A. Initiale gesunde Ausgangslage erzeugen
- Implementiere einen expliziten Initialisierungsmodus.
- Das System soll die Domain crawlen, bekannte Seiten erfassen und daraus eine erste Referenz erzeugen.
- Diese Referenz darf nicht blind übernommen werden; es muss einen Review-/Freigabeprozess geben.
- Liefere dafür einen klaren Workflow:
1. Crawl der Domain
2. Fundliste aller internen Seiten
3. Extraktion aller externen Links
4. Erkennung möglicher Auffälligkeiten
5. Manuelle Prüfung
6. Erst nach Freigabe: Speichern als „gesunde“ Baseline
B. Pflege der gesunden Daten
- Änderungen an legitimen Inhalten sollen nicht automatisch in die Baseline einfließen.
- Stattdessen:
- Monitoring meldet Änderung
- Mensch prüft Änderung
- separater „approve baseline“-Schritt übernimmt die neue Version
- Baselines sollen versioniert und nachvollziehbar abgelegt werden.
- Es soll möglich sein, pro URL eine neue Referenz freizugeben.
- Es soll auch möglich sein, die gesamte Site-Referenz neu aufzubauen, wenn es große legitime Änderungen gibt.
- Es soll erkennbar bleiben, wer/was/wann eine Baseline freigegeben hat.
Crawler- und Erkennungslogik
Bitte entwerfe das System so, dass es nicht nur bekannte Seiten prüft, sondern die Domain aktiv crawlt:
- Start mit einer Seed-Liste
- Danach internes Follow aller erlaubten Links
- Begrenzung auf bredelar.info
- Erkennung neuer interner URLs
- Erkennung von Seiten, die plötzlich verschwunden sind
- Optional: robots.txt respektieren konfigurierbar
- Schutz gegen Crawl-Explosion, Parameter-Müll und Kalender-/Session-URLs
Das System soll auf mehreren Ebenen vergleichen:
- Roh-HTML-Diff
- bereinigter DOM-/Text-Diff
- Link-Diff
- URL-Diff der gefundenen internen Seiten
- Metadaten-Diff (title, meta description, canonical, hreflang, robots, og-tags)
- Hidden-Content-Indikatoren
- Optional Header-Checks
Wichtig: Normalisierung gegen Fehlalarme
Bitte implementiere vor dem Diff sinnvolle Normalisierung, damit harmlose Änderungen nicht dauernd Alarm auslösen:
- Whitespace normalisieren
- irrelevante Timestamps, Cache-Buster, Nonces, zufällige IDs, Tracking-Parameter optional ignorieren
- HTML in eine stabile Form bringen
- konfigurierbare Ignore-Regeln für Seitenbereiche oder Selektoren
- getrennte Betrachtung von Roh-HTML und sichtbar extrahiertem Inhalt
Hidden-/SEO-Spam-spezifische Checks
Bitte berücksichtige speziell typische SEO-Spam-/Parasite-SEO-Indikatoren:
- neue interne Unterseiten
- neue externe Links
- versteckte Links
- display:none / visibility:hidden / opacity:0 / off-screen positioning
- meta refresh
- unerwartete canonical-Tags
- unerwartete strukturierte Daten / JSON-LD mit fremden Entities
- starke Änderungen in Title/H1/H2
- längere neue Textblöcke
- eingebettete kommerzielle Blöcke innerhalb eigentlich lokaler Inhalte
- auffällige Script-/Iframe-Einbindungen
- Kommentar- oder noscript-Missbrauch
- falls praktikabel: einfache Scoring-Logik für Verdachtsfälle
Alarmierung
Das System soll Alarmmeldungen verschicken können, idealerweise modular:
- E-Mail als Mindeststandard
- optional Webhook/Signal/Matrix/Telegram/SMS als Erweiterung
- Alarmmeldungen sollen kompakt, aber forensisch brauchbar sein
- Beispielinhalt:
- Zeitpunkt
- URL
- Schweregrad
- neue interne URLs
- neue externe Domains
- Link-Diff
- Text-Diff
- Hidden-Content-Hinweise
- Empfehlung: nur beobachten / prüfen / sofort handeln
Technische Zielarchitektur
Bitte schlage eine minimalistische, aber saubere Struktur vor, z. B.:
- ein Hauptprogramm mit Subcommands oder mehrere kleine Skripte
- Cron-tauglich
- Konfigurationsdateien in YAML oder JSON
- lokale Dateistruktur für:
- seeds
- erlaubte interne URLs
- erlaubte externe Links/Domains
- snapshots
- baselines
- reports
- state
- logs
- bitte ohne unnötige externe Dienste, wenn nicht zwingend erforderlich
Gewünschte Artefakte
Ich möchte von dir eine vollständige, direkt nutzbare Lösung erzeugt bekommen. Bitte liefere:
1. Architekturüberblick
- kurz und präzise
- welche Komponenten gibt es und warum
2. Bedrohungsmodell
- was die Lösung erkennt
- was sie nicht erkennt
- welche Annahmen gelten
3. Projektstruktur
- konkrete Dateibaum-Struktur
4. Konfigurationsdateien
- vollständig, mit sinnvollen Beispielen
- z. B. config.yaml, seeds.yaml, allowed_external_domains.yaml, allowed_internal_paths.yaml, ignore_rules.yaml
5. Vollständigen Quellcode
- möglichst in Python 3
- sauber strukturiert
- modular
- mit CLI-Subcommands wie z. B.:
- init-baseline
- crawl
- check
- approve
- report
- robuste Fehlerbehandlung
- keine unnötige Magie
- gut lesbar
6. HTML-Parsing und Extraktion
- stabile Extraktion von:
- sichtbarem Text
- Links
- Assets/Referenzen
- Metadaten
- Hidden-Content-Hinweisen
- bitte geeignete Bibliotheken wählen und begründen
7. Diff-Logik
- wie Roh-HTML-, Text-, Link- und URL-Diffs erzeugt werden
- Ausgabe menschenlesbar und maschinenlesbar
8. Alarmierungsmodul
- zunächst mindestens E-Mail
- optional Webhook-Schnittstelle abstrahiert
- Beispielkonfiguration
9. Cron-Setup
- konkrete Cron-Beispiele
- z. B. täglicher Check 03:00 Uhr
- Log-Rotation oder Umgang mit Logs
10. Workflow-Dokumentation
- Erstinitialisierung der „gesunden“ Daten
- Review/Freigabe
- täglicher Betrieb
- Umgang mit legitimen Änderungen
- Incident-Fall: was tun bei Alarm
11. Sicherheits- und Qualitätsverbesserungen
- sinnvolle optionale Erweiterungen
- aber nur pragmatisch, nicht überengineert
12. Tests
- mindestens ein paar einfache Unit- oder Integrations-Tests
- insbesondere für:
- URL-Normalisierung
- Link-Extraktion
- Hidden-Content-Erkennung
- Baseline-Freigabeprozess
13. Beispielausgabe
- Beispiel für normalen Bericht ohne Auffälligkeiten
- Beispiel für Alarm bei neuer SEO-Spam-Seite
- Beispiel für Alarm bei neuem externen Link
- Beispiel für Alarm bei legitimer Änderung, die anschließend freigegeben wird
Wichtige Designprinzipien
- So einfach wie möglich, aber nicht einfacher.
- Kleine Website, kleine Lösung.
- Hohe Nachvollziehbarkeit.
- Keine automatische Selbstheilung.
- Keine automatische Baseline-Aktualisierung.
- Lieber wenige robuste Regeln als viele fragile.
- Unabhängige Kontrolle des Dienstleisters.
- Gute Wartbarkeit für technisch versierte Betreiber.
Wichtige technische Präferenzen
- Linux-tauglich
- Python 3 bevorzugt
- Wenige, verbreitete Bibliotheken
- Keine schweren externen Datenbanken, wenn Dateispeicherung reicht
- UTF-8 überall
- Saubere Trennung zwischen Konfiguration, Zustand, Referenzdaten und Reports
Bitte beginne mit:
1. einer kurzen Zielarchitektur,
2. danach der konkreten Projektstruktur,
3. dann allen Dateien mit vollständigem Inhalt,
4. und zum Schluss einer knappen Betriebsanleitung.
Wichtig:
- Bitte keine abstrakte Beratung allein, sondern vollständige umsetzbare Artefakte.
- Falls du Annahmen treffen musst, triff pragmatische Annahmen und dokumentiere sie.
- Der Code soll direkt als Ausgangspunkt verwendbar sein.

754
Keller/site_audit_tool_v4.py Executable file
View file

@ -0,0 +1,754 @@
#!/usr/bin/env python3
import argparse
import hashlib
import json
import os
import re
import subprocess
import sys
from collections import defaultdict
from html.parser import HTMLParser
from pathlib import Path
from urllib.parse import urljoin, urlparse
import requests
DEFAULT_ALLOWED = {
'html','htm','css','js','mjs','json','txt','xml','svg',
'png','jpg','jpeg','gif','webp','ico','map',
'woff','woff2','ttf','otf','eot','pdf','webmanifest'
}
SUSPICIOUS_NAME_RE = re.compile(
r'(shell|backdoor|cmd|upload|gate|mailer|wso|c99|r57|b374k)',
re.I
)
SUSPICIOUS_CONTENT_PATTERNS = [
r'eval\s*\(',
r'base64_decode',
r'shell_exec\s*\(',
r'passthru\s*\(',
r'system\s*\(',
r'exec\s*\(',
r'cmd\.exe',
r'powershell',
r'document\.createElement\s*\(\s*[\"\']script[\"\']\s*\)',
r'new\s+Image\s*\(\s*\)\.src',
r'fromCharCode',
r'atob\s*\(',
r'fetch\s*\(',
r'XMLHttpRequest'
]
HEADERS = {"User-Agent": "site-audit/4.0 (+security review)"}
class AssetParser(HTMLParser):
def __init__(self):
super().__init__()
self.links = []
self.inline_scripts = []
self.inline_script = []
self.in_script = False
def handle_starttag(self, tag, attrs):
attrs = dict(attrs)
if tag == 'a' and attrs.get('href'):
self.links.append(('a', attrs['href']))
elif tag == 'link' and attrs.get('href'):
self.links.append(('link', attrs['href']))
elif tag == 'script' and attrs.get('src'):
self.links.append(('script', attrs['src']))
elif tag == 'img' and attrs.get('src'):
self.links.append(('img', attrs['src']))
elif tag == 'source' and attrs.get('src'):
self.links.append(('source', attrs['src']))
elif tag == 'iframe' and attrs.get('src'):
self.links.append(('iframe', attrs['src']))
elif tag == 'script' and not attrs.get('src'):
self.in_script = True
self.inline_script = []
def handle_endtag(self, tag):
if tag == 'script' and self.in_script:
self.inline_scripts.append(''.join(self.inline_script))
self.in_script = False
def handle_data(self, data):
if self.in_script:
self.inline_script.append(data)
def sha256_file(path: Path):
h = hashlib.sha256()
with path.open('rb') as f:
for chunk in iter(lambda: f.read(65536), b''):
h.update(chunk)
return h.hexdigest()
def ensure_dir(p: Path):
p.mkdir(parents=True, exist_ok=True)
def is_same_host(base, target):
b = urlparse(base)
t = urlparse(target)
return (not t.netloc) or (t.netloc == b.netloc)
def clean_url(url):
parsed = urlparse(url)
return parsed._replace(fragment='').geturl()
def mirror_with_wget(url: str, outdir: Path):
ensure_dir(outdir)
cmd = [
'wget',
'--mirror',
'--convert-links',
'--adjust-extension',
'--page-requisites',
'--no-parent',
'--directory-prefix',
str(outdir),
url,
]
proc = subprocess.run(cmd, capture_output=True, text=True)
return {
'returncode': proc.returncode,
'stdout': proc.stdout[-4000:],
'stderr': proc.stderr[-4000:],
}
def collect_files(root: Path):
return sorted([p for p in root.rglob('*') if p.is_file()])
def scan_tree(root: Path, allowed_exts):
findings = defaultdict(list)
for p in collect_files(root):
rel = str(p.relative_to(root))
ext = p.suffix.lower().lstrip('.')
if ext and ext not in allowed_exts:
findings['unexpected_extensions'].append(rel)
if SUSPICIOUS_NAME_RE.search(p.name):
findings['suspicious_filenames'].append(rel)
try:
data = p.read_text(encoding='utf-8', errors='ignore')
except Exception:
continue
for pat in SUSPICIOUS_CONTENT_PATTERNS:
if re.search(pat, data, re.I):
findings['suspicious_content'].append(
{'file': rel, 'pattern': pat}
)
if '<script' in data.lower() and ('http://' in data.lower() or 'https://' in data.lower()):
findings['external_refs_in_text'].append(rel)
return findings
def normalize_internal_url(base_url: str, ref: str):
full = clean_url(urljoin(base_url, ref))
parsed = urlparse(full)
if parsed.scheme not in ('http', 'https'):
return None
return full
def should_crawl(url: str, base_url: str):
if not is_same_host(base_url, url):
return False
parsed = urlparse(url)
path = parsed.path.lower()
if path.endswith((
'.jpg','.jpeg','.png','.gif','.webp','.svg','.ico',
'.pdf','.zip','.woff','.woff2','.ttf','.otf','.eot',
'.mp4','.mp3','.webm','.avi','.mov'
)):
return False
return True
def crawl_site(start_url: str, max_pages: int = 200):
seen = set()
queue = [clean_url(start_url)]
pages = []
externals = []
page_errors = []
while queue and len(pages) < max_pages:
url = queue.pop(0)
if url in seen:
continue
seen.add(url)
try:
r = requests.get(url, headers=HEADERS, timeout=20)
status = r.status_code
ctype = r.headers.get('content-type', '')
final_url = clean_url(r.url)
if 'text/html' not in ctype:
pages.append({
'url': url,
'final_url': final_url,
'status': status,
'content_type': ctype,
'title': None,
'inline_hits': [],
'internal_links': [],
'external_links': [],
})
continue
parser = AssetParser()
parser.feed(r.text)
title = None
m = re.search(r'<title>(.*?)</title>', r.text, re.I | re.S)
if m:
title = re.sub(r'\s+', ' ', m.group(1)).strip()
internal_links = []
external_links = []
for kind, ref in parser.links:
full = normalize_internal_url(final_url, ref)
if not full:
continue
item = {'kind': kind, 'url': full}
if is_same_host(start_url, full):
internal_links.append(item)
if (
kind == 'a'
and should_crawl(full, start_url)
and full not in seen
and full not in queue
):
queue.append(full)
else:
external_links.append(item)
externals.append({'from': final_url, **item})
inline_hits = []
for idx, script in enumerate(parser.inline_scripts, 1):
for pat in SUSPICIOUS_CONTENT_PATTERNS:
if re.search(pat, script, re.I):
inline_hits.append({
'script_index': idx,
'pattern': pat
})
pages.append({
'url': url,
'final_url': final_url,
'status': status,
'content_type': ctype,
'title': title,
'inline_hits': inline_hits,
'internal_links': internal_links,
'external_links': external_links,
})
except Exception as e:
page_errors.append({'url': url, 'error': str(e)})
return {'pages': pages, 'externals': externals, 'page_errors': page_errors}
def extract_remote_assets(start_url: str):
r = requests.get(start_url, headers=HEADERS, timeout=20)
r.raise_for_status()
parser = AssetParser()
parser.feed(r.text)
assets = []
externals = []
for kind, ref in parser.links:
full = clean_url(urljoin(start_url, ref))
if is_same_host(start_url, full):
assets.append({'kind': kind, 'url': full})
else:
externals.append({'kind': kind, 'url': full})
inline_hits = []
for idx, script in enumerate(parser.inline_scripts, 1):
for pat in SUSPICIOUS_CONTENT_PATTERNS:
if re.search(pat, script, re.I):
inline_hits.append({'script_index': idx, 'pattern': pat})
return {
'page_url': start_url,
'assets': assets,
'externals': externals,
'inline_hits': inline_hits,
}
def fetch_headers(url: str):
try:
r = requests.get(url, headers=HEADERS, timeout=20)
headers = {k: v for k, v in r.headers.items()}
interesting_keys = [
'Content-Security-Policy',
'Strict-Transport-Security',
'X-Content-Type-Options',
'X-Frame-Options',
'Referrer-Policy',
'Permissions-Policy',
'Cache-Control',
'Content-Type',
'Server',
]
interesting = {
k: headers.get(k)
for k in interesting_keys
if headers.get(k) is not None
}
missing = [
k for k in [
'Content-Security-Policy',
'Strict-Transport-Security',
'X-Content-Type-Options',
'Referrer-Policy',
]
if k not in interesting
]
return {
'url': r.url,
'status': r.status_code,
'headers': interesting,
'missing_recommended': missing,
}
except Exception as e:
return {
'url': url,
'error': str(e),
'headers': {},
'missing_recommended': [],
}
def scan_dom_text_for_indicators(html: str):
indicators = []
checks = {
'base_tag_present': r'<base\b',
'meta_refresh_present': r'<meta[^>]+http-equiv=[\"\']refresh[\"\']',
'iframe_present': r'<iframe\b',
'data_uri_present': r'data:(?:text|application)/',
'inline_event_handlers': r'\bon(?:click|load|error|mouseover|submit)\s*=',
'obfuscated_long_hex': r'\b[a-f0-9]{80,}\b',
'long_base64_like': r'\b[A-Za-z0-9+/]{200,}={0,2}\b',
}
for name, pat in checks.items():
if re.search(pat, html, re.I | re.S):
indicators.append(name)
return indicators
def compare_with_baseline(current_hashes: dict, baseline_file: Path):
if not baseline_file or not baseline_file.exists():
return None
baseline = json.loads(baseline_file.read_text(encoding='utf-8'))
old_hashes = baseline.get('hashes', {})
added = sorted(set(current_hashes) - set(old_hashes))
removed = sorted(set(old_hashes) - set(current_hashes))
changed = sorted([
k for k in set(current_hashes).intersection(old_hashes)
if current_hashes[k] != old_hashes[k]
])
return {
'added': added,
'removed': removed,
'changed': changed,
'baseline_source': str(baseline_file),
}
def compare_with_local_dir(current_root: Path, local_dir: Path):
if not local_dir or not local_dir.exists():
return None
local_files = collect_files(local_dir)
local_hashes = {
str(p.relative_to(local_dir)): sha256_file(p)
for p in local_files
}
current_files = collect_files(current_root)
current_hashes = {
str(p.relative_to(current_root)): sha256_file(p)
for p in current_files
}
added = sorted(set(current_hashes) - set(local_hashes))
missing_on_server = sorted(set(local_hashes) - set(current_hashes))
changed = sorted([
k for k in set(current_hashes).intersection(local_hashes)
if current_hashes[k] != local_hashes[k]
])
return {
'local_dir': str(local_dir),
'added_on_server': added,
'missing_on_server': missing_on_server,
'changed': changed,
}
def export_baseline(outdir: Path, report: dict):
baseline = {
'target': report['target'],
'created_from': report['mirror_dir'],
'hashes': report['hashes'],
}
path = outdir / 'baseline_hashes.json'
path.write_text(json.dumps(baseline, indent=2, ensure_ascii=False),
encoding='utf-8')
return path
def score_report(report: dict):
score = 0
reasons = []
if report['summary'].get('unexpected_extensions', 0) > 0:
score += 30
reasons.append('unerwartete Dateiendungen gefunden')
if report['summary'].get('suspicious_filenames', 0) > 0:
score += 40
reasons.append('verdächtige Dateinamen gefunden')
if report['summary'].get('suspicious_content', 0) > 0:
score += 50
reasons.append('verdächtige Inhaltsmuster gefunden')
if report['summary'].get('dom_indicators', 0) > 0:
score += 10
reasons.append('DOM-Indikatoren erkannt')
if report['summary'].get('header_missing', 0) >= 2:
score += 10
reasons.append('mehrere empfohlene Security-Header fehlen')
if report.get('baseline_diff'):
delta = len(report['baseline_diff'].get('added', [])) + \
len(report['baseline_diff'].get('changed', []))
if delta > 0:
score += min(40, delta * 5)
reasons.append('Abweichungen zur Baseline gefunden')
if report.get('local_diff'):
delta = len(report['local_diff'].get('added_on_server', [])) + \
len(report['local_diff'].get('changed', []))
if delta > 0:
score += min(40, delta * 5)
reasons.append('Abweichungen zum lokalen Build gefunden')
if score >= 70:
level = 'red'
exit_code = 2
elif score >= 20:
level = 'yellow'
exit_code = 1
else:
level = 'green'
exit_code = 0
return {
'score': score,
'level': level,
'exit_code': exit_code,
'reasons': reasons,
}
def send_webhook(url: str, payload: dict):
try:
r = requests.post(url, json=payload, headers=HEADERS, timeout=20)
return {
'status_code': r.status_code,
'ok': r.ok,
'text': r.text[:500],
}
except Exception as e:
return {'ok': False, 'error': str(e)}
def write_report(outdir: Path, report: dict):
ensure_dir(outdir)
json_path = outdir / 'audit_report.json'
md_path = outdir / 'audit_report.md'
json_path.write_text(
json.dumps(report, indent=2, ensure_ascii=False),
encoding='utf-8'
)
lines = []
lines.append("# Website-Audit\n")
lines.append(f"- Ziel: {report['target']}\n")
lines.append(f"- Mirror-Verzeichnis: `{report['mirror_dir']}`\n")
lines.append(
f"- Dateien im Mirror: {report['summary']['file_count']}\n"
)
lines.append(
f"- Unerwartete Endungen: {report['summary']['unexpected_extensions']}\n"
)
lines.append(
f"- Verdächtige Dateinamen: {report['summary']['suspicious_filenames']}\n"
)
lines.append(
f"- Verdächtige Inhaltsmuster: {report['summary']['suspicious_content']}\n"
)
lines.append(
f"- Externe Ressourcen auf Startseite: {report['summary']['external_assets']}\n"
)
lines.append(
f"- Treffer in Inline-Skripten: {report['summary']['inline_hits']}\n"
)
lines.append(
f"- Gecrawlte Seiten: {report['summary']['crawled_pages']}\n"
)
lines.append(
f"- Externe Ressourcen im Crawl: {report['summary']['crawl_externals']}\n"
)
lines.append(
f"- Crawl-Fehler: {report['summary']['crawl_errors']}\n"
)
lines.append(
f"- Header-Warnungen: {report['summary']['header_missing']}\n"
)
lines.append(
f"- DOM-Indikatoren auf Startseite: {report['summary']['dom_indicators']}\n"
)
lines.append(
f"- Ampelstatus: {report['assessment']['level']}\n"
)
lines.append(
f"- Risikoscore: {report['assessment']['score']}\n"
)
if report.get('assessment'):
lines.append("\n## Bewertung\n")
lines.append(
f"- Level: `{report['assessment']['level']}`"
)
lines.append(
f"- Score: `{report['assessment']['score']}`"
)
lines.append(
f"- Exit-Code: `{report['assessment']['exit_code']}`"
)
for item in report['assessment'].get('reasons', []):
lines.append(f"- Grund: {item}")
if report.get('headers'):
lines.append("\n## Header-Check\n")
if report['headers'].get('error'):
lines.append(f"- Fehler: {report['headers']['error']}")
else:
lines.append(f"- URL: {report['headers']['url']}")
lines.append(f"- Status: {report['headers']['status']}")
for k, v in report['headers']['headers'].items():
lines.append(f"- {k}: `{v}`")
if report['headers']['missing_recommended']:
lines.append(
f"- Fehlend empfohlen: "
f"{', '.join(report['headers']['missing_recommended'])}"
)
if report['findings']['unexpected_extensions']:
lines.append("\n## Unerwartete Endungen\n")
for x in report['findings']['unexpected_extensions'][:200]:
lines.append(f"- `{x}`")
if report['findings']['suspicious_filenames']:
lines.append("\n## Verdächtige Dateinamen\n")
for x in report['findings']['suspicious_filenames'][:200]:
lines.append(f"- `{x}`")
if report['findings']['suspicious_content']:
lines.append("\n## Verdächtige Inhaltsmuster\n")
for x in report['findings']['suspicious_content'][:200]:
lines.append(f"- `{x['file']}` → `{x['pattern']}`")
if report['remote']['externals']:
lines.append("\n## Externe Ressourcen\n")
for x in report['remote']['externals'][:200]:
lines.append(f"- {x['kind']}: {x['url']}")
if report['remote']['inline_hits']:
lines.append("\n## Inline-Skript-Treffer\n")
for x in report['remote']['inline_hits'][:200]:
lines.append(
f"- Inline-Skript {x['script_index']} → `{x['pattern']}`"
)
if report.get('crawl', {}).get('pages'):
lines.append("\n## Gecrawlte Seiten\n")
for x in report['crawl']['pages'][:200]:
title = f"{x['title']}" if x.get('title') else ""
lines.append(
f"- {x['status']} {x['final_url']}{title}"
)
if report.get('crawl', {}).get('externals'):
lines.append("\n## Externe Ressourcen im Crawl\n")
for x in report['crawl']['externals'][:200]:
lines.append(
f"- {x['from']} -> {x['kind']}: {x['url']}"
)
if report.get('dom_indicators'):
lines.append("\n## DOM-Indikatoren\n")
for item in report['dom_indicators']:
lines.append(f"- {item}")
if report.get('baseline_diff'):
lines.append("\n## Baseline-Diff\n")
lines.append(
f"- Baseline: `{report['baseline_diff']['baseline_source']}`"
)
for key in ('added', 'removed', 'changed'):
if report['baseline_diff'][key]:
lines.append(
f"- {key}: {len(report['baseline_diff'][key])}"
)
for item in report['baseline_diff'][key][:200]:
lines.append(f" - `{item}`")
if report.get('local_diff'):
lines.append("\n## Vergleich mit lokalem Build\n")
lines.append(
f"- Lokal: `{report['local_diff']['local_dir']}`"
)
for key in (
'added_on_server',
'missing_on_server',
'changed',
):
if report['local_diff'][key]:
lines.append(
f"- {key}: {len(report['local_diff'][key])}"
)
for item in report['local_diff'][key][:200]:
lines.append(f" - `{item}`")
md_path.write_text('\n'.join(lines), encoding='utf-8')
baseline_path = export_baseline(outdir, report)
return json_path, md_path, baseline_path
def main():
ap = argparse.ArgumentParser(
description='Mirror + Scan für statische Websites'
)
ap.add_argument('url')
ap.add_argument('--out', default='output/site_audit')
ap.add_argument('--allow-ext', action='append', default=[])
ap.add_argument('--max-pages', type=int, default=200)
ap.add_argument('--baseline-json', default=None)
ap.add_argument('--local-dir', default=None)
ap.add_argument('--webhook-url', default=None)
ap.add_argument('--fail-on-yellow', action='store_true')
args = ap.parse_args()
outdir = Path(args.out)
mirror = outdir / 'mirror'
allowed = set(DEFAULT_ALLOWED)
for item in args.allow_ext:
for part in item.split(','):
part = part.strip().lower().lstrip('.')
if part:
allowed.add(part)
wget_result = mirror_with_wget(args.url, mirror)
remote = extract_remote_assets(args.url)
crawl = crawl_site(args.url, max_pages=args.max_pages)
headers_report = fetch_headers(args.url)
html = requests.get(args.url, headers=HEADERS, timeout=20).text
dom_indicators = scan_dom_text_for_indicators(html)
file_root_candidates = [
p for p in mirror.rglob('*')
if p.is_dir() and p.name == urlparse(args.url).netloc
]
file_root = file_root_candidates[0] if file_root_candidates else mirror
findings = scan_tree(file_root, allowed)
files = collect_files(file_root)
hashes = {
str(p.relative_to(file_root)): sha256_file(p)
for p in files
}
baseline_diff = (
compare_with_baseline(hashes, Path(args.baseline_json))
if args.baseline_json else None
)
local_diff = (
compare_with_local_dir(file_root, Path(args.local_dir))
if args.local_dir else None
)
report = {
'target': args.url,
'mirror_dir': str(file_root),
'wget': wget_result,
'summary': {
'file_count': len(files),
'unexpected_extensions':
len(findings['unexpected_extensions']),
'suspicious_filenames':
len(findings['suspicious_filenames']),
'suspicious_content':
len(findings['suspicious_content']),
'external_assets': len(remote['externals']),
'inline_hits': len(remote['inline_hits']),
'crawled_pages': len(crawl['pages']),
'crawl_externals': len(crawl['externals']),
'crawl_errors': len(crawl['page_errors']),
'header_missing':
len(headers_report.get('missing_recommended', [])),
'dom_indicators': len(dom_indicators),
},
'findings': findings,
'remote': remote,
'crawl': crawl,
'headers': headers_report,
'dom_indicators': dom_indicators,
'baseline_diff': baseline_diff,
'local_diff': local_diff,
'hashes': hashes,
}
assessment = score_report(report)
if args.fail_on_yellow and assessment['level'] == 'yellow':
assessment['exit_code'] = 2
report['assessment'] = assessment
if args.webhook_url:
report['webhook_result'] = send_webhook(
args.webhook_url,
{
'target': report['target'],
'summary': report['summary'],
'assessment': report['assessment'],
},
)
write_report(outdir, report)
sys.exit(report['assessment']['exit_code'])
if __name__ == '__main__':
main()

203
README.md Normal file
View file

@ -0,0 +1,203 @@
# Website Integrity Scanner
Unabhängiges Überwachungswerkzeug für statische Websites gegen SEO-Spam
und unbefugte Manipulationen. Läuft im Verantwortungsbereich des Betreibers,
nicht beim Webhoster.
## Schnellstart
```bash
cd integrity_scanner_fuer_statische_Webseiten
python -m venv venv
source venv/bin/activate
pip install -r requirements.txt
# 1. Erste Baseline anlegen
python -m scanner init
# 2. Ergebnis prüfen, dann freigeben
python -m scanner approve --all --note "Initiale Baseline nach Sichtprüfung"
# 3. Täglicher Betrieb (oder per Cron)
python -m scanner scan
```
---
## Subcommands
| Befehl | Funktion |
|---|---|
| `init` | Erstcrawl + Vorschau, kein Baseline-Update |
| `crawl` | Crawlt die Website, speichert Snapshot |
| `check` | Vergleicht letzten Snapshot mit Baseline |
| `scan` | `crawl` + `check` in einem (für Cron) |
| `approve --all` | Alle Snapshot-URLs als neue Baseline freigeben |
| `approve --url URL` | Einzelne URL freigeben |
| `approve --rebuild` | Gesamte Baseline ersetzen (nach großer Änderung) |
| `report` | Letzten Report anzeigen |
| `status` | Baseline-Datum, letzter Scan, Risiko-Level |
Alle Befehle akzeptieren `--config path/to/config.yaml` und `--verbose`.
---
## Workflow: Erstinitialisierung
```
python -m scanner init
```
Ausgabe zeigt:
- alle gefundenen internen URLs
- alle externen Links (mit Whitelist-Status)
- potenzielle Auffälligkeiten (Hidden Content, Meta-Refresh, ...)
**Wichtig**: Die Ausgabe manuell prüfen, bevor die Baseline gesetzt wird.
Insbesondere externe Links in `config/allowed_external.yaml` eintragen.
```
python -m scanner approve --all --note "Initiale Baseline, geprüft am $(date)"
```
---
## Workflow: Täglicher Betrieb (Cron)
```
python -m scanner scan
```
Exit-Codes:
- `0` = Grün (keine Auffälligkeiten)
- `1` = Gelb (Warnung, manuelle Prüfung empfohlen)
- `2` = Rot (Alarm, sofortige Prüfung notwendig)
---
## Workflow: Legitime Änderung freigeben
```bash
# 1. Scan durchgeführt, Warnung auf /impressum/
python -m scanner report # Report ansehen
python -m scanner approve --url https://bredelar.info/impressum/ --note "Impressum aktualisiert"
# 2. Oder alle Änderungen auf einmal (wenn alles geprüft)
python -m scanner approve --all --note "Redesign vom 2026-06-12 freigegeben"
# 3. Nach großem Redesign: komplette neue Baseline
python -m scanner approve --rebuild --note "Komplettes Redesign, alle Seiten neu"
```
---
## Workflow: Incident (Alarm)
1. Report lesen: `python -m scanner report`
2. Verdächtige URLs manuell im Browser prüfen
3. Quelltext der betroffenen Seiten ansehen (insb. `display:none`-Bereiche)
4. Dienstleister kontaktieren, Zugriff auf Hosting prüfen
5. Nach Bereinigung: `python -m scanner scan` — muss grün sein
6. Dann: `python -m scanner approve --rebuild --note "Nach Incident bereinigt"`
**Wichtig**: Niemals `approve` auf eine kompromittierte Seite anwenden.
---
## Cron-Setup
```cron
# Täglich 03:00 Uhr
0 3 * * * cd /opt/scanner && /opt/scanner/venv/bin/python -m scanner scan >> /opt/scanner/logs/cron.log 2>&1
```
Log-Rotation (`/etc/logrotate.d/scanner`):
```
/opt/scanner/logs/*.log {
daily
rotate 30
compress
missingok
notifempty
}
```
---
## Konfiguration
### E-Mail aktivieren
In `config.yaml`:
```yaml
alerting:
email:
enabled: true
smtp_host: mail.example.com
smtp_port: 587
smtp_user: scanner@example.com
smtp_password_env: SCANNER_SMTP_PASSWORD
from: scanner@bredelar.info
to:
- admin@bredelar.info
```
Passwort als Umgebungsvariable setzen (nicht in config.yaml):
```bash
export SCANNER_SMTP_PASSWORD="geheimesPasswort"
```
### Webhook aktivieren
```yaml
alerting:
webhook:
enabled: true
url: "https://hooks.slack.com/services/..."
```
### Externe Domains whitelisten
Nach dem ersten `init` alle legitimen externen Domains in
`config/allowed_external.yaml` eintragen:
```yaml
- fonts.googleapis.com
- maps.googleapis.com
```
---
## Verzeichnisstruktur
```
data/
baseline/ Aktuell freigegebene Referenz (NIEMALS automatisch aktualisiert)
manifest.json Wer/was/wann freigegeben hat
pages/ Pro URL eine JSON-Datei
snapshots/ Zeitgestempelte Crawl-Ergebnisse
YYYYMMDD_HHMMSS/
reports/ Generierte Reports (JSON + Markdown)
logs/ scanner.log
config/
allowed_external.yaml
allowed_paths.yaml
ignore_rules.yaml
```
---
## Tests
```bash
pytest tests/ -v
```
---
## Erweiterungen (optional, nicht implementiert)
- Ollama-Integration: Neue Textblöcke semantisch auf Spam-Kategorien prüfen
- Telegram-Bot: direkter Alert-Kanal
- Diff-Anzeige im Terminal: `python -m scanner report --diff-only`
- Automatische Archivierung alter Snapshots nach N Tagen

112
config.yaml Normal file
View file

@ -0,0 +1,112 @@
# Integrity Scanner — Hauptkonfiguration
# Alle Werte sind optional; fehlende werden aus den Defaults übernommen.
target: "https://bredelar.info"
user_agent: "integrity-scanner/1.0 (+security-monitoring)"
request_timeout: 20
data_dir: "data"
reports_dir: "reports"
logs_dir: "logs"
config_dir: "config"
crawl:
max_pages: 200
delay_seconds: 1.0
respect_robots_txt: false
skip_extensions:
- ".jpg"
- ".jpeg"
- ".png"
- ".gif"
- ".webp"
- ".svg"
- ".ico"
- ".pdf"
- ".zip"
- ".woff"
- ".woff2"
- ".ttf"
- ".otf"
- ".eot"
- ".mp4"
- ".mp3"
- ".webm"
- ".avi"
- ".mov"
scoring:
new_external_domain: 50
new_internal_url: 30
missing_internal_url: 20
hidden_content: 40
unexpected_canonical: 35
meta_refresh: 40
unexpected_jsonld: 35
large_text_addition: 20
new_inline_script_suspicious: 30
missing_security_header: 5
suspicious_content_pattern: 50
comment_links: 25
noscript_links: 25
new_external_link_unlisted: 40
thresholds:
yellow: 20 # ab diesem Score: Warnung (gelb)
red: 60 # ab diesem Score: Alarm (rot)
large_text_block_chars: 200 # Textblock ab dieser Länge wird als „groß" gewertet
alerting:
min_level: "yellow" # "yellow" oder "red"
email:
enabled: false
smtp_host: "localhost"
smtp_port: 587
smtp_tls: true
smtp_user: ""
smtp_password_env: "SCANNER_SMTP_PASSWORD" # Passwort aus Umgebungsvariable
from: "scanner@bredelar.info"
to:
- "admin@bredelar.info"
webhook:
enabled: false
url: "" # z.B. Telegram-Bot-Webhook, Slack-Incoming-Webhook
normalization:
strip_html_comments: true
collapse_whitespace: true
# Selektoren, die vor dem Diff ignoriert werden (z.B. Datums-/Cache-Elemente)
ignore_selectors:
# - "#last-modified"
# - ".timestamp"
# Regex-Muster im Text, die vor dem Diff entfernt werden
ignore_patterns:
# - '\d{1,2}\.\d{1,2}\.\d{4}' # Datumsangaben
# - 'Stand:\s+\S+'
# Erlaubte @type-Werte in JSON-LD (alles andere löst Alarm aus)
allowed_jsonld_types:
- "Organization"
- "WebSite"
- "WebPage"
- "Article"
- "BreadcrumbList"
- "ItemList"
- "LocalBusiness"
- "Place"
- "Person"
- "Event"
- "FAQPage"
- "Question"
- "Answer"
- "ImageObject"
- "SiteLinksSearchBox"
- "ContactPage"
- "AboutPage"
# Security-Header, die auf jeder Seite erwartet werden
security_headers:
- "Content-Security-Policy"
- "Strict-Transport-Security"
- "X-Content-Type-Options"
- "Referrer-Policy"

View file

@ -0,0 +1,18 @@
# Whitelist erlaubter externer Domains
#
# Jede externe Domain, die auf der Website verlinkt ist, muss hier eingetragen sein.
# Neue externe Domains, die NICHT hier stehen, lösen einen Alarm aus.
#
# Format: eine Domain pro Zeile (ohne Protokoll)
# Beispiele:
# Häufig verwendete externe Ressourcen — nur eintragen, was wirklich genutzt wird:
# - fonts.googleapis.com
# - fonts.gstatic.com
# - cdnjs.cloudflare.com
# - maps.googleapis.com
# - www.google-analytics.com
# - www.googletagmanager.com
# Für bredelar.info: Bitte nach erstem `init` prüfen und befüllen.
# Vorerst leer = alle externen Links lösen Alarm aus.

17
config/allowed_paths.yaml Normal file
View file

@ -0,0 +1,17 @@
# Whitelist erlaubter interner URL-Pfade (optional)
#
# Wenn diese Liste befüllt ist, lösen interne Links auf nicht gelistete
# Pfade einen Alarm aus.
# Leer lassen = keine Pfad-Whitelist-Prüfung (nur Baseline-Diff zählt).
#
# Format: ein Pfad pro Zeile (absoluter Pfad, ohne Domain)
# Beispiel für bredelar.info:
#
# - /
# - /index.html
# - /kontakt/
# - /kontakt/index.html
# - /impressum/
# - /datenschutz/
# - /ueber-uns/
# - /aktuelles/

23
config/ignore_rules.yaml Normal file
View file

@ -0,0 +1,23 @@
# Normalisierungsregeln für den Diff
#
# Diese Einstellungen reduzieren Fehlalarme durch harmlose, automatisch
# generierte Inhaltsänderungen.
# CSS-Selektoren, die vor dem Diff aus dem DOM entfernt werden.
# Nützlich für Elemente mit automatischen Zeitstempeln oder Zählern.
ignore_selectors:
# - "#last-modified"
# - ".timestamp"
# - "[data-cache-buster]"
# - ".cookie-banner"
# Reguläre Ausdrücke, die im normalisierten Text durch [IGNORED] ersetzt werden.
# Nützlich für Datumsangaben, Versionsnummern, etc.
ignore_patterns:
# Datumsformate (deutsch: 12.06.2026 oder 12. Juni 2026)
# - '\b\d{1,2}\.\s*(?:Januar|Februar|März|April|Mai|Juni|Juli|August|September|Oktober|November|Dezember)\s+\d{4}\b'
# - '\b\d{1,2}\.\d{1,2}\.\d{4}\b'
# "Stand: ..." Angaben
# - 'Stand:\s+\S+'
# Cache-Buster in URLs (werden im HTML-Text sichtbar)
# - '\?v=[\w.]+\b'

5
requirements.txt Normal file
View file

@ -0,0 +1,5 @@
requests>=2.31
beautifulsoup4>=4.12
lxml>=5.0
PyYAML>=6.0
pytest>=7.0

0
scanner/__init__.py Normal file
View file

672
scanner/__main__.py Normal file
View file

@ -0,0 +1,672 @@
"""
integrity-scanner CLI entry point.
Usage:
python -m scanner <subcommand> [options]
Subcommands:
init First-time crawl + baseline setup workflow
crawl Crawl site, save snapshot (no comparison)
check Compare latest snapshot against baseline, report + alert
scan crawl + check in one step (use this for cron)
approve Approve snapshot as new baseline
report Display / export latest report
status Show baseline age, last scan, open changes
"""
import argparse
import json
import logging
import sys
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlparse
from .alerter import send_alert
from .baseline import BaselineManager
from .checker import (
canonical_is_hijacked,
check_jsonld_types,
check_links_against_whitelist,
check_security_headers,
is_suspicious_url,
load_whitelist,
)
from .config import load_config, resolve_paths
from .crawler import Crawler
from .differ import compare_snapshots, normalize_text, score_diff
from .extractor import extract_page
# ---------------------------------------------------------------------------
# Logging
# ---------------------------------------------------------------------------
def _setup_logging(logs_dir: str, verbose: bool) -> None:
level = logging.DEBUG if verbose else logging.INFO
logs_path = Path(logs_dir)
logs_path.mkdir(parents=True, exist_ok=True)
log_file = logs_path / "scanner.log"
fmt = "%(asctime)s %(levelname)-8s %(name)s: %(message)s"
handlers: list[logging.Handler] = [
logging.StreamHandler(sys.stderr),
logging.FileHandler(log_file, encoding="utf-8"),
]
logging.basicConfig(level=level, format=fmt, handlers=handlers, force=True)
# ---------------------------------------------------------------------------
# Crawl + extract helpers
# ---------------------------------------------------------------------------
def _crawl_and_extract(cfg: dict) -> tuple[list[dict], list[dict]]:
"""Run crawler, extract each HTML page. Returns (pages, errors)."""
logger = logging.getLogger(__name__)
crawler = Crawler(cfg)
raw = crawler.crawl()
pages: list[dict] = []
for raw_page in raw["pages"]:
if raw_page.get("html"):
extracted = extract_page(raw_page["html"], raw_page["url"], cfg)
page = {**raw_page, **extracted}
else:
page = dict(raw_page)
pages.append(page)
logger.info("Extracted %d pages", len(pages))
return pages, raw["errors"]
# ---------------------------------------------------------------------------
# Whitelist loading
# ---------------------------------------------------------------------------
def _load_whitelists(cfg: dict) -> tuple[set[str], set[str]]:
config_dir = Path(cfg["config_dir"])
allowed_ext = load_whitelist(config_dir / "allowed_external.yaml")
allowed_int = load_whitelist(config_dir / "allowed_paths.yaml")
return allowed_ext, allowed_int
# ---------------------------------------------------------------------------
# Report building
# ---------------------------------------------------------------------------
def _build_report(
target: str,
diff: dict,
assessment: dict,
whitelist_violations: list[dict],
missing_sec_headers: dict,
crawl_errors: list[dict],
snap_dir: Path,
) -> dict:
return {
"generated_at": datetime.now(timezone.utc).isoformat(),
"target": target,
"snapshot_dir": str(snap_dir),
"assessment": assessment,
"diff": diff,
"whitelist_violations": whitelist_violations,
"missing_security_headers": missing_sec_headers,
"crawl_errors": crawl_errors,
}
def _write_report(report: dict, reports_dir: str) -> tuple[Path, Path]:
ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
out = Path(reports_dir) / ts
out.mkdir(parents=True, exist_ok=True)
json_path = out / "report.json"
md_path = out / "report.md"
json_path.write_text(
json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8"
)
md_path.write_text(_render_markdown(report), encoding="utf-8")
return json_path, md_path
def _render_markdown(r: dict) -> str:
a = r.get("assessment", {})
d = r.get("diff", {})
level = a.get("level", "?").upper()
score = a.get("score", 0)
level_badge = {"GREEN": "[OK]", "YELLOW": "[WARNUNG]", "RED": "[ALARM]"}.get(level, level)
lines = [
f"# Integrity Report — {r.get('target', '?')}",
"",
f"**Erstellt:** {r.get('generated_at', '?')} ",
f"**Level:** {level_badge} ",
f"**Score:** {score}",
"",
"## Zusammenfassung",
"",
f"| Kennzahl | Wert |",
f"|---|---|",
f"| Geprüfte Seiten | {d.get('total_pages_checked', '?')} |",
f"| Geänderte Seiten | {d.get('changed_pages', 0)} |",
f"| Neue interne URLs | {len(d.get('new_internal_urls', []))} |",
f"| Fehlende URLs | {len(d.get('missing_internal_urls', []))} |",
f"| Neue externe Domains | {len(d.get('new_external_domains', []))} |",
f"| Whitelist-Verstösse | {len(r.get('whitelist_violations', []))} |",
f"| Crawl-Fehler | {len(r.get('crawl_errors', []))} |",
"",
"## Risikobewertung",
"",
]
for reason in a.get("reasons", []):
lines.append(f"- {reason}")
if not a.get("reasons"):
lines.append("- Keine Auffälligkeiten.")
if d.get("new_external_domains"):
lines += ["", "## Neue externe Domains", ""]
for dom in d["new_external_domains"]:
lines.append(f"- `{dom}`")
if d.get("new_internal_urls"):
lines += ["", "## Neue interne URLs", ""]
for url in d["new_internal_urls"]:
lines.append(f"- {url}")
if d.get("missing_internal_urls"):
lines += ["", "## Fehlende URLs (waren in Baseline)", ""]
for url in d["missing_internal_urls"]:
lines.append(f"- {url}")
if d.get("page_diffs"):
lines += ["", "## Geänderte Seiten", ""]
for pd in d["page_diffs"]:
lines += [f"### {pd['url']}", ""]
if pd.get("added_chars"):
lines.append(f"- Neuer Text: +{pd['added_chars']} Zeichen")
if pd.get("meta_diff"):
for field, change in pd["meta_diff"].items():
if isinstance(change, dict) and "old" in change:
lines.append(f"- {field}: `{change['old']}` → `{change['new']}`")
else:
lines.append(f"- {field}: {change}")
if pd.get("canonical_changed"):
lines.append(
f"- Canonical geändert: `{pd.get('old_canonical')}` → `{pd.get('new_canonical')}`"
)
if pd.get("new_hidden_content"):
for h in pd["new_hidden_content"]:
lines.append(f"- Hidden Content ({h.get('type')}): `{', '.join(h.get('indicators', []))}`")
if pd.get("new_meta_refresh"):
for mr in pd["new_meta_refresh"]:
lines.append(f"- Meta-Refresh → `{mr}`")
if pd.get("new_comment_links"):
for cl in pd["new_comment_links"]:
lines.append(f"- Kommentar-Link: `{cl}`")
if pd.get("new_inline_scripts"):
for s in pd["new_inline_scripts"]:
lines.append(f"- Inline-Script: `{', '.join(s.get('patterns', []))}`")
if pd.get("link_diff"):
for ltype, ld in pd["link_diff"].items():
for added in ld.get("added", []):
lines.append(f"- {ltype} hinzugefügt: `{added}`")
for removed in ld.get("removed", []):
lines.append(f"- {ltype} entfernt: `{removed}`")
if pd.get("text_diff"):
lines += ["", "```diff"]
lines += pd["text_diff"][:40]
if len(pd["text_diff"]) > 40:
lines.append(f"... ({len(pd['text_diff']) - 40} weitere Zeilen)")
lines.append("```")
lines.append("")
if d.get("new_pages_analysis"):
lines += ["", "## Neue Seiten (Erstanalyse)", ""]
for npa in d["new_pages_analysis"]:
problems = (
npa.get("hidden_content")
or npa.get("inline_scripts")
or npa.get("meta_refresh")
or npa.get("unexpected_jsonld")
or npa.get("comment_links")
)
flag = " [VERDAECHTIG]" if problems else ""
lines.append(f"- {npa['url']}{flag}")
if r.get("whitelist_violations"):
lines += ["", "## Whitelist-Verstösse", ""]
for v in r["whitelist_violations"][:50]:
lines.append(f"- [{v.get('type')}] {v.get('url', '')} (Domain: {v.get('domain', '')})")
if r.get("missing_security_headers"):
lines += ["", "## Fehlende Security-Header", ""]
for url, hdrs in r["missing_security_headers"].items():
for h in hdrs:
lines.append(f"- `{url}`: {h}")
if r.get("crawl_errors"):
lines += ["", "## Crawl-Fehler", ""]
for e in r["crawl_errors"][:20]:
lines.append(f"- `{e.get('url', '?')}`: {e.get('error', '?')}")
lines += [
"",
"---",
"",
"## Nächste Schritte",
"",
]
lev = a.get("level", "green")
if lev == "red":
lines += [
"1. **Sofort**: Website-Backend überprüfen, Dateien mit Backup vergleichen.",
"2. Dienstleister kontaktieren.",
"3. Nach Bereinigung: `python -m scanner approve --rebuild`",
]
elif lev == "yellow":
lines += [
"1. Alle markierten URLs manuell prüfen.",
"2. Legitime Änderungen freigeben: `python -m scanner approve --url <URL>`",
"3. Alle freigeben (wenn geprüft): `python -m scanner approve --all`",
]
else:
lines += ["Keine Aktion erforderlich."]
return "\n".join(lines) + "\n"
def _latest_report_path(reports_dir: str) -> Path | None:
rd = Path(reports_dir)
if not rd.exists():
return None
dirs = sorted(d for d in rd.iterdir() if d.is_dir())
if not dirs:
return None
return dirs[-1] / "report.md"
# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------
def cmd_init(args: argparse.Namespace, cfg: dict) -> int:
logger = logging.getLogger("scanner.init")
print("=== Initialisierung: Erster Crawl ===")
print(f"Ziel: {cfg['target']}")
pages, errors = _crawl_and_extract(cfg)
if not pages:
print("FEHLER: Kein Seiten gecrawlt.", file=sys.stderr)
return 1
bm = BaselineManager(cfg["data_dir"])
snap_dir = bm.save_snapshot(pages)
print(f"Snapshot gespeichert: {snap_dir}")
print(f" Seiten: {len(pages)}")
print(f" Fehler: {len(errors)}")
print()
allowed_ext, allowed_int = _load_whitelists(cfg)
base_netloc = urlparse(cfg["target"]).netloc
print("=== Gefundene interne URLs ===")
for p in pages:
print(f" {p['url']}")
print()
print("=== Externe Links (alle) ===")
seen_domains: set[str] = set()
for p in pages:
for ltype, links in p.get("links", {}).items():
if not isinstance(links, list):
continue
for l in links:
if isinstance(l, dict) and l.get("class") == "external":
domain = urlparse(l.get("url", "")).netloc
if domain and domain not in seen_domains:
seen_domains.add(domain)
in_wl = " [in Whitelist]" if domain in allowed_ext else " [NICHT in Whitelist]"
print(f" {domain}{in_wl}")
print()
auffaelligkeiten = 0
for p in pages:
if p.get("hidden_content"):
print(f" [!] Hidden Content auf {p['url']}: {len(p['hidden_content'])} Fund(e)")
auffaelligkeiten += len(p["hidden_content"])
if p.get("inline_scripts"):
print(f" [!] Verdächtige Inline-Scripts auf {p['url']}: {len(p['inline_scripts'])} Fund(e)")
auffaelligkeiten += len(p["inline_scripts"])
if p.get("links", {}).get("meta_refresh"):
print(f" [!] Meta-Refresh auf {p['url']}")
auffaelligkeiten += 1
if p.get("comment_links"):
print(f" [!] Links in HTML-Kommentaren auf {p['url']}")
auffaelligkeiten += len(p["comment_links"])
if errors:
print()
print("=== Crawl-Fehler ===")
for e in errors:
print(f" {e['url']}: {e['error']}")
print()
if auffaelligkeiten:
print(f"[WARNUNG] {auffaelligkeiten} potenzielle Auffälligkeit(en) gefunden.")
print(" Bitte alle URLs oben manuell prüfen, bevor Sie freigeben.")
else:
print("[OK] Keine offensichtlichen Auffälligkeiten gefunden.")
print()
print("=== Naechste Schritte ===")
print(" 1. Prüfen Sie alle oben aufgelisteten externen Links und Auffälligkeiten.")
print(" 2. Wenn alles sauber ist, Baseline freigeben:")
print(" python -m scanner approve --all")
print(" 3. Danach ist der tägliche Betrieb möglich:")
print(" python -m scanner scan")
return 0
def cmd_crawl(args: argparse.Namespace, cfg: dict) -> int:
logger = logging.getLogger("scanner.crawl")
print(f"Crawle {cfg['target']} ...")
pages, errors = _crawl_and_extract(cfg)
bm = BaselineManager(cfg["data_dir"])
snap_dir = bm.save_snapshot(pages)
print(f"Snapshot: {snap_dir}")
print(f" Seiten: {len(pages)}")
print(f" Fehler: {len(errors)}")
return 0
def cmd_check(args: argparse.Namespace, cfg: dict) -> int:
logger = logging.getLogger("scanner.check")
bm = BaselineManager(cfg["data_dir"])
if not bm.baseline_exists():
print(
"FEHLER: Keine Baseline vorhanden. Zuerst `python -m scanner init` ausführen.",
file=sys.stderr,
)
return 1
snap = bm.load_snapshot()
if not snap:
print(
"FEHLER: Kein Snapshot vorhanden. Zuerst `python -m scanner crawl` ausführen.",
file=sys.stderr,
)
return 1
baseline = bm.load_baseline()
logger.info(
"Vergleiche %d Baseline-Seiten mit %d Snapshot-Seiten ...",
len(baseline["pages"]),
len(snap["pages"]),
)
diff = compare_snapshots(baseline["pages"], snap["pages"], cfg)
# Whitelist checks on all current pages
allowed_ext, allowed_int = _load_whitelists(cfg)
base_netloc = urlparse(cfg["target"]).netloc
wl_violations: list[dict] = []
missing_sec: dict = {}
for page in snap["pages"].values():
violations = check_links_against_whitelist(page, allowed_ext, base_netloc)
if violations:
wl_violations.extend({"page": page["url"], **v} for v in violations)
hdrs = page.get("response_headers", {})
missing = check_security_headers(hdrs, cfg.get("security_headers", []))
if missing:
missing_sec[page["url"]] = missing
# Unexpected JSON-LD types (absolute check, not just diff-based)
unexp = check_jsonld_types(page, cfg.get("allowed_jsonld_types", []))
for j in unexp:
if not any(
pd.get("url") == page["url"] and j in pd.get("new_jsonld", [])
for pd in diff.get("page_diffs", [])
):
# Add as new_jsonld to the page_diff if not already there
pd_for_url = next(
(pd for pd in diff.get("page_diffs", []) if pd["url"] == page["url"]),
None,
)
if pd_for_url:
pd_for_url.setdefault("new_jsonld", []).append({**j, "unexpected": True})
# Score with whitelist-violation contribution
for v in wl_violations:
if v.get("type") == "unlisted_external_domain":
existing = diff.get("new_external_domains", [])
domain = v.get("domain", "")
if domain and domain not in existing:
existing.append(domain)
diff["new_external_domains"] = existing
assessment = score_diff(diff, cfg)
crawl_errors: list[dict] = []
if getattr(args, "with_errors", False):
# errors were not persisted in snapshot, skip
pass
report = _build_report(
cfg["target"], diff, assessment, wl_violations, missing_sec,
crawl_errors, snap["dir"],
)
json_path, md_path = _write_report(report, cfg["reports_dir"])
logger.info("Report: %s", md_path)
_print_summary(assessment, diff)
send_alert(report, cfg)
return assessment["exit_code"]
def cmd_scan(args: argparse.Namespace, cfg: dict) -> int:
"""crawl + check."""
logger = logging.getLogger("scanner.scan")
print(f"Scanne {cfg['target']} ...")
pages, errors = _crawl_and_extract(cfg)
bm = BaselineManager(cfg["data_dir"])
snap_dir = bm.save_snapshot(pages)
logger.info("Snapshot: %s (%d Seiten, %d Fehler)", snap_dir, len(pages), len(errors))
if not bm.baseline_exists():
print()
print("Noch keine Baseline vorhanden — Scan abgeschlossen, kein Vergleich.")
print("Führen Sie aus: python -m scanner approve --all")
return 0
return cmd_check(args, cfg)
def cmd_approve(args: argparse.Namespace, cfg: dict) -> int:
logger = logging.getLogger("scanner.approve")
bm = BaselineManager(cfg["data_dir"])
snap_dir = Path(args.snapshot) if getattr(args, "snapshot", None) else None
note = getattr(args, "note", "") or ""
if getattr(args, "rebuild", False):
approved = bm.rebuild_baseline(snap_dir=snap_dir, note=note)
print(f"Baseline komplett neu aufgebaut: {len(approved)} URLs freigegeben.")
return 0
if getattr(args, "all", False):
approved = bm.approve_all(snap_dir=snap_dir, note=note)
print(f"Alle {len(approved)} URLs freigegeben.")
for u in approved:
print(f" + {u}")
return 0
url = getattr(args, "url", None)
if url:
ok = bm.approve_url(url, snap_dir=snap_dir, note=note)
if ok:
print(f"Freigegeben: {url}")
return 0
else:
print(f"FEHLER: URL nicht im Snapshot gefunden: {url}", file=sys.stderr)
return 1
print("FEHLER: --url, --all oder --rebuild angeben.", file=sys.stderr)
return 1
def cmd_report(args: argparse.Namespace, cfg: dict) -> int:
fmt = getattr(args, "format", "md")
rd = Path(cfg["reports_dir"])
if not rd.exists():
print("Noch keine Reports vorhanden.")
return 0
dirs = sorted(d for d in rd.iterdir() if d.is_dir())
if not dirs:
print("Noch keine Reports vorhanden.")
return 0
latest = dirs[-1]
if fmt == "json":
p = latest / "report.json"
else:
p = latest / "report.md"
if not p.exists():
print(f"Report-Datei nicht gefunden: {p}", file=sys.stderr)
return 1
print(p.read_text(encoding="utf-8"))
return 0
def cmd_status(args: argparse.Namespace, cfg: dict) -> int:
bm = BaselineManager(cfg["data_dir"])
print(f"Ziel: {cfg['target']}")
if bm.baseline_exists():
manifest = json.loads(
(bm.baseline_dir / "manifest.json").read_text(encoding="utf-8")
)
print(f"Baseline: {manifest.get('approved_at', '?')} (von: {manifest.get('approved_by', '?')})")
print(f" Seiten: {len(manifest.get('urls', []))}")
if manifest.get("note"):
print(f" Notiz: {manifest['note']}")
else:
print("Baseline: KEINE — 'python -m scanner init' ausführen")
snap_dir = bm.latest_snapshot_dir()
if snap_dir:
snap_manifest = json.loads((snap_dir / "manifest.json").read_text(encoding="utf-8"))
print(f"Letzter Scan: {snap_manifest.get('timestamp', '?')} ({snap_manifest.get('page_count', '?')} Seiten)")
else:
print("Letzter Scan: KEINER")
rd = Path(cfg["reports_dir"])
if rd.exists():
dirs = sorted(d for d in rd.iterdir() if d.is_dir())
if dirs:
rp = dirs[-1] / "report.json"
if rp.exists():
r = json.loads(rp.read_text(encoding="utf-8"))
a = r.get("assessment", {})
print(f"Letzter Report: {dirs[-1].name} — Level: {a.get('level','?').upper()} (Score: {a.get('score','?')})")
else:
print(f"Letzter Report: {dirs[-1].name}")
return 0
# ---------------------------------------------------------------------------
# Output helper
# ---------------------------------------------------------------------------
def _print_summary(assessment: dict, diff: dict) -> None:
level = assessment.get("level", "?").upper()
score = assessment.get("score", 0)
print()
print(f"=== Ergebnis: {level} (Score {score}) ===")
for reason in assessment.get("reasons", []):
print(f" * {reason}")
if not assessment.get("reasons"):
print(" Keine Auffälligkeiten.")
print(f" Neue URLs: {len(diff.get('new_internal_urls', []))}")
print(f" Fehlende URLs: {len(diff.get('missing_internal_urls', []))}")
print(f" Neue externe Domains:{len(diff.get('new_external_domains', []))}")
print(f" Geänderte Seiten: {diff.get('changed_pages', 0)}")
# ---------------------------------------------------------------------------
# Argument parser
# ---------------------------------------------------------------------------
def _build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
prog="scanner",
description="Website-Integrity-Scanner für statische Webseiten",
)
p.add_argument("--config", default="config.yaml", help="Pfad zur config.yaml")
p.add_argument("--verbose", "-v", action="store_true", help="Debug-Ausgabe")
sub = p.add_subparsers(dest="command", required=True)
sub.add_parser("init", help="Erstcrawl + Baseline-Workflow starten")
sub.add_parser("crawl", help="Nur crawlen, Snapshot speichern")
sub.add_parser("check", help="Snapshot vs. Baseline vergleichen")
sub.add_parser("scan", help="crawl + check (für Cron)")
ap = sub.add_parser("approve", help="Snapshot als neue Baseline freigeben")
ap.add_argument("--all", action="store_true", help="Alle URLs freigeben")
ap.add_argument("--url", help="Einzelne URL freigeben")
ap.add_argument("--rebuild", action="store_true", help="Baseline komplett ersetzen")
ap.add_argument("--snapshot", help="Pfad zu einem bestimmten Snapshot-Verzeichnis")
ap.add_argument("--note", default="", help="Freitext-Notiz für die Baseline-Versionierung")
rp = sub.add_parser("report", help="Letzten Report anzeigen")
rp.add_argument("--format", choices=["md", "json"], default="md")
sub.add_parser("status", help="Übersichtsstatus anzeigen")
return p
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main() -> None:
parser = _build_parser()
args = parser.parse_args()
cfg = load_config(args.config)
base_dir = Path(args.config).parent.resolve()
resolve_paths(cfg, base_dir)
_setup_logging(cfg["logs_dir"], args.verbose)
commands = {
"init": cmd_init,
"crawl": cmd_crawl,
"check": cmd_check,
"scan": cmd_scan,
"approve": cmd_approve,
"report": cmd_report,
"status": cmd_status,
}
handler = commands.get(args.command)
if not handler:
parser.print_help()
sys.exit(1)
sys.exit(handler(args, cfg))
if __name__ == "__main__":
main()

183
scanner/alerter.py Normal file
View file

@ -0,0 +1,183 @@
"""
Alert dispatch: e-mail (smtplib) and HTTP webhook.
Alerts are only sent when the assessment level meets or exceeds min_level.
"""
import logging
import os
import smtplib
from datetime import datetime, timezone
from email.message import EmailMessage
import requests
logger = logging.getLogger(__name__)
_LEVEL_ORDER = {"green": 0, "yellow": 1, "red": 2}
def send_alert(report: dict, cfg: dict) -> None:
"""Send alert if the report's level is at or above min_level."""
alerting = cfg.get("alerting", {})
level = report.get("assessment", {}).get("level", "green")
min_level = alerting.get("min_level", "yellow")
if _LEVEL_ORDER.get(level, 0) < _LEVEL_ORDER.get(min_level, 1):
logger.debug("Alert suppressed: level=%s < min_level=%s", level, min_level)
return
subject = _make_subject(report)
body = _format_body(report)
email_cfg = alerting.get("email", {})
if email_cfg.get("enabled"):
_send_email(subject, body, email_cfg)
webhook_cfg = alerting.get("webhook", {})
if webhook_cfg.get("enabled") and webhook_cfg.get("url"):
_send_webhook(webhook_cfg["url"], report, subject)
# ---------------------------------------------------------------------------
# Formatting
# ---------------------------------------------------------------------------
def _make_subject(report: dict) -> str:
level = report.get("assessment", {}).get("level", "?").upper()
target = report.get("target", "?")
ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
return f"[{level}] Integrity Alert: {target}{ts}"
def _format_body(report: dict) -> str:
lines = [
"=" * 60,
"WEBSITE INTEGRITY ALERT",
"=" * 60,
f"Ziel: {report.get('target', '?')}",
f"Zeitpunkt: {datetime.now(timezone.utc).isoformat()}",
]
assessment = report.get("assessment", {})
level = assessment.get("level", "?").upper()
score = assessment.get("score", "?")
lines += [
f"Level: {level}",
f"Score: {score}",
"",
"--- Gruende ---",
]
for reason in assessment.get("reasons", []):
lines.append(f" * {reason}")
diff = report.get("diff", {})
_section(lines, "Neue externe Domains", diff.get("new_external_domains", []), "+")
_section(lines, "Neue interne URLs", diff.get("new_internal_urls", []), "+")
_section(lines, "Fehlende URLs", diff.get("missing_internal_urls", []), "-")
for pd in diff.get("page_diffs", [])[:10]:
lines += ["", f"--- Aenderung: {pd['url']} ---"]
if pd.get("text_diff"):
lines.append(f" Text-Diff: {pd['added_chars']} Zeichen zugefügt")
lines += [f" {l}" for l in pd["text_diff"][:30]]
if pd.get("new_hidden_content"):
lines.append(f" Hidden Content: {len(pd['new_hidden_content'])} neue(r) Fund(e)")
if pd.get("new_meta_refresh"):
lines.append(f" Meta-Refresh: {pd['new_meta_refresh']}")
if pd.get("canonical_changed"):
lines.append(f" Canonical geaendert: {pd.get('old_canonical')} -> {pd.get('new_canonical')}")
if pd.get("new_comment_links"):
lines.append(f" Kommentar-Links: {pd['new_comment_links']}")
lines += [
"",
"--- Empfehlung ---",
]
level_lower = assessment.get("level", "green")
if level_lower == "red":
lines += [
" SOFORT HANDELN: Website moeglicherweise kompromittiert.",
" Zugriff auf Hosting-Backend prüfen, Dateien vergleichen.",
]
elif level_lower == "yellow":
lines += [
" PRÜFEN: Unerwartete Aenderungen gefunden.",
" Alle markierten URLs manuell kontrollieren.",
" Falls legitim: python -m scanner approve --all",
]
else:
lines.append(" OK — nur zur Information.")
lines += [
"",
"--- Workflow ---",
" Details: python -m scanner report",
" Freigabe: python -m scanner approve --url <URL>",
" Rebuild: python -m scanner approve --rebuild",
"=" * 60,
]
return "\n".join(lines)
def _section(lines: list, title: str, items: list, prefix: str) -> None:
if not items:
return
lines += ["", f"--- {title} ---"]
for item in items[:50]:
lines.append(f" {prefix} {item}")
if len(items) > 50:
lines.append(f" ... und {len(items) - 50} weitere")
# ---------------------------------------------------------------------------
# Transport
# ---------------------------------------------------------------------------
def _send_email(subject: str, body: str, cfg: dict) -> None:
password = os.environ.get(cfg.get("smtp_password_env", "SCANNER_SMTP_PASSWORD"), "")
recipients = cfg.get("to", [])
if not recipients:
logger.warning("E-Mail alert enabled but no recipients configured.")
return
msg = EmailMessage()
msg["Subject"] = subject
msg["From"] = cfg.get("from", "scanner@localhost")
msg["To"] = ", ".join(recipients)
msg.set_content(body)
host = cfg.get("smtp_host", "localhost")
port = cfg.get("smtp_port", 587)
use_tls = cfg.get("smtp_tls", True)
try:
with smtplib.SMTP(host, port) as server:
if use_tls:
server.starttls()
user = cfg.get("smtp_user", "")
if user and password:
server.login(user, password)
server.send_message(msg)
logger.info("Alert e-mail sent to %s", recipients)
except Exception as exc:
logger.error("E-mail alert failed: %s", exc)
def _send_webhook(url: str, report: dict, subject: str) -> None:
payload = {
"text": subject,
"target": report.get("target"),
"level": report.get("assessment", {}).get("level"),
"score": report.get("assessment", {}).get("score"),
"reasons": report.get("assessment", {}).get("reasons", []),
"new_external_domains": report.get("diff", {}).get("new_external_domains", []),
"new_internal_urls": report.get("diff", {}).get("new_internal_urls", []),
"missing_internal_urls": report.get("diff", {}).get("missing_internal_urls", []),
"changed_pages": report.get("diff", {}).get("changed_pages", 0),
}
try:
resp = requests.post(url, json=payload, timeout=10)
logger.info("Webhook delivered: HTTP %s", resp.status_code)
except Exception as exc:
logger.error("Webhook alert failed: %s", exc)

231
scanner/baseline.py Normal file
View file

@ -0,0 +1,231 @@
"""
Snapshot and baseline management.
Snapshots are timestamped crawl results. The baseline is a manually
approved reference state. Compromised content must never flow into the
baseline automatically every update requires an explicit `approve` call.
"""
import hashlib
import json
import logging
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
logger = logging.getLogger(__name__)
def url_key(url: str) -> str:
"""Stable filename-safe key for a URL (first 16 hex chars of SHA-256)."""
return hashlib.sha256(url.encode("utf-8")).hexdigest()[:16]
class BaselineManager:
def __init__(self, data_dir: str | Path):
self.data_dir = Path(data_dir)
self.baseline_dir = self.data_dir / "baseline"
self.pages_dir = self.baseline_dir / "pages"
self.snapshots_dir = self.data_dir / "snapshots"
# ------------------------------------------------------------------
# Snapshot operations
# ------------------------------------------------------------------
def save_snapshot(self, pages: list[dict]) -> Path:
"""
Persist a crawl+extraction result as a timestamped snapshot.
Returns the snapshot directory path.
"""
ts = _utc_stamp()
snap_dir = self.snapshots_dir / ts
snap_pages_dir = snap_dir / "pages"
snap_pages_dir.mkdir(parents=True, exist_ok=True)
manifest = {
"timestamp": ts,
"page_count": len(pages),
"urls": [p["url"] for p in pages],
}
(snap_dir / "manifest.json").write_text(
json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8"
)
for page in pages:
fname = url_key(page["url"]) + ".json"
(snap_pages_dir / fname).write_text(
json.dumps(page, indent=2, ensure_ascii=False), encoding="utf-8"
)
logger.info("Snapshot saved: %s (%d pages)", snap_dir, len(pages))
return snap_dir
def latest_snapshot_dir(self) -> Optional[Path]:
"""Return the most recent snapshot directory, or None."""
if not self.snapshots_dir.exists():
return None
dirs = sorted(
(d for d in self.snapshots_dir.iterdir() if d.is_dir()),
key=lambda d: d.name,
)
return dirs[-1] if dirs else None
def load_snapshot(self, snap_dir: Optional[Path] = None) -> dict:
"""
Load a snapshot (default: latest).
Returns {manifest, pages: {url: page_dict}, dir: Path}.
"""
if snap_dir is None:
snap_dir = self.latest_snapshot_dir()
if snap_dir is None or not snap_dir.exists():
return {}
manifest = json.loads((snap_dir / "manifest.json").read_text(encoding="utf-8"))
pages: dict[str, dict] = {}
pages_dir = snap_dir / "pages"
for f in pages_dir.glob("*.json"):
page = json.loads(f.read_text(encoding="utf-8"))
pages[page["url"]] = page
return {"manifest": manifest, "pages": pages, "dir": snap_dir}
# ------------------------------------------------------------------
# Baseline operations
# ------------------------------------------------------------------
def baseline_exists(self) -> bool:
return (self.baseline_dir / "manifest.json").exists()
def load_baseline(self) -> dict:
"""
Load the current approved baseline.
Returns {manifest, pages: {url: page_dict}}.
"""
if not self.baseline_exists():
return {"manifest": {}, "pages": {}}
manifest = json.loads(
(self.baseline_dir / "manifest.json").read_text(encoding="utf-8")
)
pages: dict[str, dict] = {}
if self.pages_dir.exists():
for f in self.pages_dir.glob("*.json"):
page = json.loads(f.read_text(encoding="utf-8"))
pages[page["url"]] = page
return {"manifest": manifest, "pages": pages}
# ------------------------------------------------------------------
# Approve operations (the only way baseline content is ever updated)
# ------------------------------------------------------------------
def approve_all(
self,
snap_dir: Optional[Path] = None,
note: str = "",
approved_by: str = "cli",
) -> list[str]:
"""
Approve every page in the snapshot as the new baseline.
Existing baseline pages NOT present in snapshot are kept unless
--rebuild is requested (caller responsibility to wipe baseline_dir first).
"""
snap = self.load_snapshot(snap_dir)
if not snap:
raise RuntimeError("No snapshot found to approve.")
self.pages_dir.mkdir(parents=True, exist_ok=True)
approved: list[str] = []
for url, page in snap["pages"].items():
fname = url_key(url) + ".json"
(self.pages_dir / fname).write_text(
json.dumps(page, indent=2, ensure_ascii=False), encoding="utf-8"
)
approved.append(url)
self._write_manifest(approved, note, approved_by, snap.get("dir"))
logger.info("Approved %d URLs into baseline.", len(approved))
return approved
def approve_url(
self,
url: str,
snap_dir: Optional[Path] = None,
note: str = "",
approved_by: str = "cli",
) -> bool:
"""Approve a single URL from snapshot into the baseline."""
snap = self.load_snapshot(snap_dir)
if url not in snap.get("pages", {}):
logger.error("URL not found in snapshot: %s", url)
return False
self.pages_dir.mkdir(parents=True, exist_ok=True)
fname = url_key(url) + ".json"
(self.pages_dir / fname).write_text(
json.dumps(snap["pages"][url], indent=2, ensure_ascii=False),
encoding="utf-8",
)
# Update manifest incrementally
if self.baseline_exists():
manifest = json.loads(
(self.baseline_dir / "manifest.json").read_text(encoding="utf-8")
)
else:
manifest = {"urls": []}
if url not in manifest.get("urls", []):
manifest.setdefault("urls", []).append(url)
manifest["last_updated"] = datetime.now(timezone.utc).isoformat()
manifest["last_note"] = note
manifest["last_approved_by"] = approved_by
(self.baseline_dir / "manifest.json").write_text(
json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8"
)
logger.info("Approved single URL into baseline: %s", url)
return True
def rebuild_baseline(
self,
snap_dir: Optional[Path] = None,
note: str = "",
approved_by: str = "cli",
) -> list[str]:
"""
Wipe the entire baseline and replace it with the given snapshot.
Use when a major legitimate redesign has taken place.
"""
import shutil
if self.baseline_dir.exists():
shutil.rmtree(self.baseline_dir)
self.pages_dir.mkdir(parents=True, exist_ok=True)
return self.approve_all(snap_dir=snap_dir, note=note, approved_by=approved_by)
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
def _write_manifest(
self,
urls: list[str],
note: str,
approved_by: str,
source_snap: Optional[Path],
) -> None:
self.baseline_dir.mkdir(parents=True, exist_ok=True)
manifest = {
"approved_at": datetime.now(timezone.utc).isoformat(),
"approved_by": approved_by,
"note": note,
"source_snapshot": str(source_snap) if source_snap else "",
"urls": urls,
}
(self.baseline_dir / "manifest.json").write_text(
json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8"
)
def _utc_stamp() -> str:
return datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")

164
scanner/checker.py Normal file
View file

@ -0,0 +1,164 @@
"""
Additional integrity checks beyond baseline diff:
- link whitelist enforcement
- security-header audit
- canonical hijack detection
- suspicious URL filenames (from v4 heritage)
"""
import re
from pathlib import Path
from urllib.parse import urlparse
import yaml
_SUSPICIOUS_FILENAME_RE = re.compile(
r"(shell|backdoor|cmd|upload|gate|mailer|wso|c99|r57|b374k)",
re.I,
)
# ---------------------------------------------------------------------------
# Whitelist checks
# ---------------------------------------------------------------------------
def load_whitelist(path: str | Path) -> set[str]:
"""Load a YAML list file into a set. Returns empty set if missing."""
p = Path(path)
if not p.exists():
return set()
data = yaml.safe_load(p.read_text(encoding="utf-8")) or []
if isinstance(data, list):
return {str(s).strip() for s in data if s}
return set()
def check_links_against_whitelist(
page: dict,
allowed_external_domains: set[str],
base_netloc: str,
) -> list[dict]:
"""
Return violations: external links to domains not in allowed_external_domains.
Also flags noscript links and comment links to external domains.
"""
violations: list[dict] = []
for link_type, links in page.get("links", {}).items():
if not isinstance(links, list):
continue
for link in links:
if not isinstance(link, dict):
continue
if link.get("class") != "external":
continue
domain = urlparse(link.get("url", "")).netloc
if domain and domain not in allowed_external_domains:
violations.append({
"type": "unlisted_external_domain",
"link_type": link_type,
"url": link.get("url", ""),
"domain": domain,
})
# Comment links
for href in page.get("comment_links", []):
domain = urlparse(href).netloc
if domain and domain != base_netloc and domain not in allowed_external_domains:
violations.append({
"type": "comment_link_to_unlisted_domain",
"link_type": "comment",
"url": href,
"domain": domain,
})
# noscript links
for hc in page.get("hidden_content", []):
if hc.get("type") == "noscript_links":
for href in hc.get("links", []):
domain = urlparse(href).netloc
if domain and domain != base_netloc and domain not in allowed_external_domains:
violations.append({
"type": "noscript_link_to_unlisted_domain",
"link_type": "noscript",
"url": href,
"domain": domain,
})
return violations
def check_internal_paths(
page: dict,
allowed_paths: set[str],
) -> list[dict]:
"""
Flag internal links to paths not in the allowed_paths whitelist.
If allowed_paths is empty, the check is skipped.
"""
if not allowed_paths:
return []
violations: list[dict] = []
for link in page.get("links", {}).get("a", []):
if link.get("class") != "internal":
continue
path = urlparse(link.get("url", "")).path
if path and path not in allowed_paths:
violations.append({
"type": "unlisted_internal_path",
"url": link.get("url", ""),
"path": path,
})
return violations
# ---------------------------------------------------------------------------
# Security headers
# ---------------------------------------------------------------------------
def check_security_headers(response_headers: dict, required: list[str]) -> list[str]:
"""Return list of missing recommended security headers."""
# Header lookup is case-insensitive
present = {k.lower() for k in response_headers}
return [h for h in required if h.lower() not in present]
# ---------------------------------------------------------------------------
# Canonical hijack
# ---------------------------------------------------------------------------
def canonical_is_hijacked(page: dict, base_netloc: str) -> bool:
"""
Return True if the canonical URL points to a different domain.
Canonical appearing in both <link rel=canonical> and metadata is checked.
"""
canonical = (
page.get("links", {}).get("canonical")
or page.get("metadata", {}).get("canonical")
)
if not canonical:
return False
netloc = urlparse(canonical).netloc
return bool(netloc) and netloc != base_netloc
# ---------------------------------------------------------------------------
# Suspicious filenames
# ---------------------------------------------------------------------------
def is_suspicious_url(url: str) -> bool:
"""Flag URLs with names typical for webshells / backdoors."""
filename = urlparse(url).path.rsplit("/", 1)[-1]
return bool(_SUSPICIOUS_FILENAME_RE.search(filename))
# ---------------------------------------------------------------------------
# JSON-LD type check
# ---------------------------------------------------------------------------
def check_jsonld_types(page: dict, allowed_types: list[str]) -> list[dict]:
"""Return JSON-LD entries whose @type is not in the allowed list."""
allowed = set(allowed_types)
return [
j for j in page.get("jsonld", [])
if j.get("type") not in allowed
]

110
scanner/config.py Normal file
View file

@ -0,0 +1,110 @@
"""Config loading with deep-merge against defaults."""
from pathlib import Path
import yaml
DEFAULT_CONFIG: dict = {
"target": "https://bredelar.info",
"user_agent": "integrity-scanner/1.0 (+security-monitoring)",
"request_timeout": 20,
"data_dir": "data",
"reports_dir": "reports",
"logs_dir": "logs",
"config_dir": "config",
"crawl": {
"max_pages": 200,
"delay_seconds": 1.0,
"respect_robots_txt": False,
"skip_extensions": [
".jpg", ".jpeg", ".png", ".gif", ".webp", ".svg", ".ico",
".pdf", ".zip", ".woff", ".woff2", ".ttf", ".otf", ".eot",
".mp4", ".mp3", ".webm", ".avi", ".mov",
],
},
"scoring": {
"new_external_domain": 50,
"new_internal_url": 30,
"missing_internal_url": 20,
"hidden_content": 40,
"unexpected_canonical": 35,
"meta_refresh": 40,
"unexpected_jsonld": 35,
"large_text_addition": 20,
"new_inline_script_suspicious": 30,
"missing_security_header": 5,
"suspicious_content_pattern": 50,
"comment_links": 25,
"noscript_links": 25,
"new_external_link_unlisted": 40,
},
"thresholds": {
"yellow": 20,
"red": 60,
"large_text_block_chars": 200,
},
"alerting": {
"min_level": "yellow",
"email": {
"enabled": False,
"smtp_host": "localhost",
"smtp_port": 587,
"smtp_tls": True,
"smtp_user": "",
"smtp_password_env": "SCANNER_SMTP_PASSWORD",
"from": "scanner@example.com",
"to": [],
},
"webhook": {
"enabled": False,
"url": "",
},
},
"normalization": {
"strip_html_comments": True,
"collapse_whitespace": True,
"ignore_selectors": [],
"ignore_patterns": [],
},
"allowed_jsonld_types": [
"Organization", "WebSite", "WebPage", "Article", "BreadcrumbList",
"ItemList", "LocalBusiness", "Place", "Person", "Event",
"FAQPage", "Question", "Answer", "ImageObject", "SiteLinksSearchBox",
"ContactPage", "AboutPage",
],
"security_headers": [
"Content-Security-Policy",
"Strict-Transport-Security",
"X-Content-Type-Options",
"Referrer-Policy",
],
}
def load_config(config_file: str = "config.yaml") -> dict:
"""Load config from YAML, deep-merged on top of defaults."""
cfg = _deep_copy(DEFAULT_CONFIG)
p = Path(config_file)
if p.exists():
with p.open(encoding="utf-8") as f:
user_cfg = yaml.safe_load(f) or {}
_deep_update(cfg, user_cfg)
return cfg
def resolve_paths(cfg: dict, base_dir: Path) -> dict:
"""Resolve relative paths in config against base_dir."""
for key in ("data_dir", "reports_dir", "logs_dir", "config_dir"):
cfg[key] = str(base_dir / cfg[key])
return cfg
def _deep_copy(d: dict) -> dict:
import copy
return copy.deepcopy(d)
def _deep_update(base: dict, override: dict) -> None:
for k, v in override.items():
if isinstance(v, dict) and isinstance(base.get(k), dict):
_deep_update(base[k], v)
else:
base[k] = v

150
scanner/crawler.py Normal file
View file

@ -0,0 +1,150 @@
"""Pure-Python crawler — no wget dependency."""
import logging
import re
import time
from urllib.parse import parse_qs, urlencode, urljoin, urlparse, urlunparse
import requests
from bs4 import BeautifulSoup
logger = logging.getLogger(__name__)
# Query parameters that indicate calendar/session/tracking noise
_NOISE_PARAM_RE = re.compile(
r"^(year|month|day|date|week|session|token|sid|nonce|_|v|cache|utm_|fbclid)",
re.I,
)
def normalize_url(url: str) -> str | None:
"""
Normalise a URL: lowercase scheme+host, drop fragment, sort and filter
query params that look like cache-busters or tracking noise.
Returns None for non-http(s) URLs.
"""
try:
p = urlparse(url.strip())
if p.scheme not in ("http", "https"):
return None
netloc = p.netloc.lower()
path = p.path or "/"
# Remove fragment
p = p._replace(netloc=netloc, path=path, fragment="")
# Keep only non-noise query params, sorted for stable comparison
if p.query:
clean_params = sorted(
(k, v)
for k, vs in parse_qs(p.query, keep_blank_values=True).items()
for v in vs
if not _NOISE_PARAM_RE.match(k)
)
p = p._replace(query=urlencode(clean_params))
return urlunparse(p)
except Exception:
return None
def should_crawl(url: str, base_netloc: str, skip_extensions: list[str]) -> bool:
"""Return True if url should be followed during a crawl."""
p = urlparse(url)
if p.netloc != base_netloc:
return False
path_lower = p.path.lower()
for ext in skip_extensions:
if path_lower.endswith(ext):
return False
# Skip URLs whose path looks like calendar/session noise
if _NOISE_PARAM_RE.search(p.query):
return False
return True
class Crawler:
def __init__(self, cfg: dict):
self.target = cfg["target"].rstrip("/")
self.base_netloc = urlparse(self.target).netloc
self.max_pages: int = cfg["crawl"]["max_pages"]
self.delay: float = cfg["crawl"]["delay_seconds"]
self.skip_ext: list[str] = cfg["crawl"]["skip_extensions"]
self.timeout: int = cfg["request_timeout"]
self.headers = {"User-Agent": cfg["user_agent"]}
def crawl(self) -> dict:
"""
Crawl the target site breadth-first.
Returns a dict with:
- pages: list of page-dicts (url, final_url, status, html, headers, ...)
- errors: list of {url, error}
"""
seen: set[str] = set()
queue: list[str] = [normalize_url(self.target + "/") or self.target + "/"]
pages: list[dict] = []
errors: list[dict] = []
while queue and len(pages) < self.max_pages:
url = queue.pop(0)
if url in seen:
continue
seen.add(url)
if self.delay > 0 and pages:
time.sleep(self.delay)
logger.debug("Crawling %s", url)
try:
resp = requests.get(
url,
headers=self.headers,
timeout=self.timeout,
allow_redirects=True,
)
final_url = normalize_url(resp.url) or resp.url
content_type = resp.headers.get("content-type", "")
is_html = "text/html" in content_type and resp.status_code == 200
page: dict = {
"url": url,
"final_url": final_url,
"status": resp.status_code,
"content_type": content_type,
"response_headers": dict(resp.headers),
"html": resp.text if is_html else None,
"redirected": final_url != url,
}
pages.append(page)
if is_html:
for link_url in self._extract_follow_links(resp.text, final_url):
if link_url not in seen and link_url not in queue:
queue.append(link_url)
except requests.exceptions.RequestException as exc:
errors.append({"url": url, "error": str(exc)})
logger.warning("Error crawling %s: %s", url, exc)
logger.info(
"Crawl finished: %d pages, %d errors, %d queued but skipped",
len(pages),
len(errors),
len(queue),
)
return {
"target": self.target,
"pages": pages,
"errors": errors,
}
def _extract_follow_links(self, html: str, base_url: str) -> list[str]:
"""Extract internal <a href> links worth following."""
result = []
try:
soup = BeautifulSoup(html, "lxml")
for tag in soup.find_all("a", href=True):
href = tag["href"].strip()
abs_url = normalize_url(urljoin(base_url, href))
if abs_url and should_crawl(abs_url, self.base_netloc, self.skip_ext):
result.append(abs_url)
except Exception as exc:
logger.debug("Link extraction failed on %s: %s", base_url, exc)
return result

338
scanner/differ.py Normal file
View file

@ -0,0 +1,338 @@
"""
Diff logic: text, link, metadata, and URL-set comparison.
Scoring maps diff results to a risk level (green/yellow/red).
"""
import difflib
import hashlib
import re
from urllib.parse import urlparse
# ---------------------------------------------------------------------------
# Normalisation
# ---------------------------------------------------------------------------
def normalize_text(text: str, cfg: dict) -> str:
"""Apply noise-reduction before diffing."""
norm_cfg = cfg.get("normalization", {})
if norm_cfg.get("collapse_whitespace", True):
text = re.sub(r"\s+", " ", text).strip()
for pattern in norm_cfg.get("ignore_patterns", []):
text = re.sub(pattern, "[IGNORED]", text)
return text
def text_hash(text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()
# ---------------------------------------------------------------------------
# Atomic diffs
# ---------------------------------------------------------------------------
def diff_text(old: str, new: str, context: int = 2) -> list[str]:
"""Unified diff of two normalised texts."""
return list(difflib.unified_diff(
old.splitlines(),
new.splitlines(),
fromfile="baseline",
tofile="current",
lineterm="",
n=context,
))
def diff_link_set(old_list: list[dict], new_list: list[dict]) -> dict:
"""Compare two link lists by URL."""
old_urls = {l["url"] for l in old_list if isinstance(l, dict) and "url" in l}
new_urls = {l["url"] for l in new_list if isinstance(l, dict) and "url" in l}
return {
"added": sorted(new_urls - old_urls),
"removed": sorted(old_urls - new_urls),
}
def diff_metadata(old: dict, new: dict) -> dict:
"""Compare metadata fields that matter for integrity monitoring."""
changes: dict = {}
for key in ("title", "description", "canonical", "robots"):
if old.get(key) != new.get(key):
changes[key] = {"old": old.get(key), "new": new.get(key)}
for heading in ("h1", "h2"):
old_set = set(old.get(heading, []))
new_set = set(new.get(heading, []))
if old_set != new_set:
changes[heading] = {
"added": sorted(new_set - old_set),
"removed": sorted(old_set - new_set),
}
return changes
def _all_external_domains(pages: dict) -> set[str]:
"""Collect every external domain referenced across all pages."""
domains: set[str] = set()
for page in pages.values():
for link_type, links in page.get("links", {}).items():
if not isinstance(links, list):
continue
for l in links:
if isinstance(l, dict) and l.get("class") == "external":
netloc = urlparse(l.get("url", "")).netloc
if netloc:
domains.add(netloc)
return domains
# ---------------------------------------------------------------------------
# Full snapshot comparison
# ---------------------------------------------------------------------------
def compare_snapshots(baseline_pages: dict, current_pages: dict, cfg: dict) -> dict:
"""
Compare two page sets (url page_dict) and produce a structured diff.
Returns a dict with:
new_internal_urls, missing_internal_urls,
new_external_domains, removed_external_domains,
page_diffs (list of per-page change records),
new_pages_analysis (freshly seen pages checked for immediate issues)
"""
b_urls = set(baseline_pages)
c_urls = set(current_pages)
new_urls = sorted(c_urls - b_urls)
missing_urls = sorted(b_urls - c_urls)
common_urls = sorted(b_urls & c_urls)
page_diffs: list[dict] = []
for url in common_urls:
old = baseline_pages[url]
new = current_pages[url]
old_text = normalize_text(old.get("text", ""), cfg)
new_text = normalize_text(new.get("text", ""), cfg)
text_changed = text_hash(old_text) != text_hash(new_text)
# Quick-skip if nothing changed (text + links + meta all identical)
if (
not text_changed
and old.get("links") == new.get("links")
and old.get("metadata") == new.get("metadata")
and old.get("hidden_content") == new.get("hidden_content")
and old.get("inline_scripts") == new.get("inline_scripts")
and old.get("jsonld") == new.get("jsonld")
):
continue
text_diff = diff_text(old_text, new_text) if text_changed else []
# Per-type link diff
link_diff: dict = {}
for ltype in ("a", "script", "iframe", "link_rel", "form"):
old_links = old.get("links", {}).get(ltype, [])
new_links = new.get("links", {}).get(ltype, [])
ld = diff_link_set(old_links, new_links)
if ld["added"] or ld["removed"]:
link_diff[ltype] = ld
meta_diff = diff_metadata(old.get("metadata", {}), new.get("metadata", {}))
# New hidden content (compare by fingerprint to avoid dupe noise)
old_hc_fps = {_fingerprint(h) for h in old.get("hidden_content", [])}
new_hidden = [
h for h in new.get("hidden_content", [])
if _fingerprint(h) not in old_hc_fps
]
# New suspicious inline scripts
old_scripts_fps = {_fingerprint(s) for s in old.get("inline_scripts", [])}
new_scripts = [
s for s in new.get("inline_scripts", [])
if _fingerprint(s) not in old_scripts_fps
]
# New JSON-LD
old_jld_fps = {_fingerprint(j) for j in old.get("jsonld", [])}
new_jsonld = [
j for j in new.get("jsonld", [])
if _fingerprint(j) not in old_jld_fps
]
# New meta-refresh targets
old_mr = set(old.get("links", {}).get("meta_refresh", []))
new_mr = [u for u in new.get("links", {}).get("meta_refresh", []) if u not in old_mr]
# New canonical
old_canonical = old.get("links", {}).get("canonical") or old.get("metadata", {}).get("canonical")
new_canonical = new.get("links", {}).get("canonical") or new.get("metadata", {}).get("canonical")
canonical_changed = old_canonical != new_canonical
# New comment links
old_cl = set(old.get("comment_links", []))
new_cl = [u for u in new.get("comment_links", []) if u not in old_cl]
if any([
text_diff, link_diff, meta_diff, new_hidden, new_scripts,
new_jsonld, new_mr, canonical_changed, new_cl,
]):
added_chars = sum(
len(line[1:]) for line in text_diff
if line.startswith("+") and not line.startswith("+++")
)
page_diffs.append({
"url": url,
"text_diff": text_diff,
"added_chars": added_chars,
"link_diff": link_diff,
"meta_diff": meta_diff,
"new_hidden_content": new_hidden,
"new_inline_scripts": new_scripts,
"new_jsonld": new_jsonld,
"new_meta_refresh": new_mr,
"canonical_changed": canonical_changed,
"old_canonical": old_canonical,
"new_canonical": new_canonical,
"new_comment_links": new_cl,
})
# Site-wide external domain diff
old_ext = _all_external_domains(baseline_pages)
new_ext = _all_external_domains(current_pages)
# Analysis of brand-new pages (no baseline to compare against)
new_pages_analysis = [
_quick_check_new_page(current_pages[url])
for url in new_urls
if url in current_pages
]
return {
"new_internal_urls": new_urls,
"missing_internal_urls": missing_urls,
"new_external_domains": sorted(new_ext - old_ext),
"removed_external_domains": sorted(old_ext - new_ext),
"page_diffs": page_diffs,
"new_pages_analysis": new_pages_analysis,
"total_pages_checked": len(common_urls),
"changed_pages": len(page_diffs),
}
def _quick_check_new_page(page: dict) -> dict:
"""Summarise a new (not in baseline) page for immediate risk indicators."""
return {
"url": page["url"],
"hidden_content": page.get("hidden_content", []),
"inline_scripts": page.get("inline_scripts", []),
"unexpected_jsonld": [j for j in page.get("jsonld", []) if j.get("unexpected")],
"meta_refresh": page.get("links", {}).get("meta_refresh", []),
"comment_links": page.get("comment_links", []),
"external_link_count": sum(
1 for links in page.get("links", {}).values()
if isinstance(links, list)
for l in links
if isinstance(l, dict) and l.get("class") == "external"
),
}
# ---------------------------------------------------------------------------
# Scoring
# ---------------------------------------------------------------------------
def score_diff(diff: dict, cfg: dict) -> dict:
"""
Map a diff result to a risk score and level.
Returns {score, level, reasons, exit_code}.
"""
sc = cfg.get("scoring", {})
thr = cfg.get("thresholds", {"yellow": 20, "red": 60})
large_text = thr.get("large_text_block_chars", 200)
score = 0
reasons: list[str] = []
def add(pts: int, msg: str) -> None:
nonlocal score
score += pts
reasons.append(f"{msg} (+{pts})")
for domain in diff.get("new_external_domains", []):
add(sc.get("new_external_domain", 50), f"Neue externe Domain: {domain}")
for url in diff.get("new_internal_urls", []):
add(sc.get("new_internal_url", 30), f"Neue interne URL: {url}")
for url in diff.get("missing_internal_urls", []):
add(sc.get("missing_internal_url", 20), f"Fehlende URL: {url}")
for pd in diff.get("page_diffs", []):
u = pd["url"]
for h in pd.get("new_hidden_content", []):
add(sc.get("hidden_content", 40),
f"{u}: Hidden Content [{', '.join(h.get('indicators', []))}]")
for mr in pd.get("new_meta_refresh", []):
add(sc.get("meta_refresh", 40), f"{u}: Meta-Refresh → {mr}")
for s in pd.get("new_inline_scripts", []):
add(sc.get("new_inline_script_suspicious", 30),
f"{u}: Verdächtiges Inline-Script [{', '.join(s.get('patterns', []))}]")
for j in pd.get("new_jsonld", []):
if j.get("unexpected"):
add(sc.get("unexpected_jsonld", 35),
f"{u}: Unerwartetes JSON-LD @type={j.get('type')}")
if pd.get("canonical_changed"):
add(sc.get("unexpected_canonical", 35),
f"{u}: Canonical geändert → {pd.get('new_canonical')}")
for cl in pd.get("new_comment_links", []):
add(sc.get("comment_links", 25), f"{u}: Link in HTML-Kommentar: {cl}")
if pd.get("added_chars", 0) > large_text:
add(sc.get("large_text_addition", 20),
f"{u}: +{pd['added_chars']} Zeichen neuer Text")
# New pages: immediate suspicious content counts too
for npa in diff.get("new_pages_analysis", []):
u = npa["url"]
if npa.get("hidden_content"):
add(sc.get("hidden_content", 40), f"{u} (neu): Hidden Content")
if npa.get("inline_scripts"):
add(sc.get("new_inline_script_suspicious", 30), f"{u} (neu): Verdächtiges Script")
if npa.get("meta_refresh"):
add(sc.get("meta_refresh", 40), f"{u} (neu): Meta-Refresh")
if npa.get("comment_links"):
add(sc.get("comment_links", 25), f"{u} (neu): Kommentar-Link")
if npa.get("unexpected_jsonld"):
add(sc.get("unexpected_jsonld", 35), f"{u} (neu): Unerwartetes JSON-LD")
yellow = thr.get("yellow", 20)
red = thr.get("red", 60)
if score >= red:
level = "red"
exit_code = 2
elif score >= yellow:
level = "yellow"
exit_code = 1
else:
level = "green"
exit_code = 0
return {"score": score, "level": level, "reasons": reasons, "exit_code": exit_code}
def _fingerprint(obj: dict | list | str) -> str:
"""Stable hash for deduplicating diff entries."""
import json as _json
return hashlib.md5( # noqa: S324 — not crypto, just dedup
_json.dumps(obj, sort_keys=True, ensure_ascii=False).encode()
).hexdigest()

312
scanner/extractor.py Normal file
View file

@ -0,0 +1,312 @@
"""
HTML extraction with BeautifulSoup4+lxml.
Extracts: visible text, links (all types), metadata, hidden-content indicators,
inline script patterns, JSON-LD, and links buried in HTML comments.
"""
import json
import re
from urllib.parse import urljoin, urlparse
from bs4 import BeautifulSoup, Comment
# Inline-script patterns that warrant flagging
_SUSPICIOUS_INLINE = [
(r"eval\s*\(", "eval()"),
(r"atob\s*\(", "atob()"),
(r"fromCharCode", "fromCharCode"),
(r"document\.createElement\s*\(\s*[\"']script[\"']", "createElement(script)"),
(r"\bfetch\s*\(", "fetch()"),
(r"XMLHttpRequest", "XMLHttpRequest"),
(r"base64_decode", "base64_decode"),
]
# CSS-based hiding patterns (applied to inline style attributes)
_HIDDEN_STYLE_CHECKS = [
(r"display\s*:\s*none", "display:none"),
(r"visibility\s*:\s*hidden", "visibility:hidden"),
(r"opacity\s*:\s*0(?:\.0+)?(?!\d)", "opacity:0"),
(r"font-size\s*:\s*0", "font-size:0"),
(r"color\s*:\s*(?:white|#fff(?:fff)?|rgba?\(\s*255\s*,\s*255\s*,\s*255)", "white-text"),
(r"width\s*:\s*0(?:px)?[;\s]", "width:0"),
(r"height\s*:\s*0(?:px)?[;\s]", "height:0"),
(r"position\s*:\s*absolute", "position:absolute"), # combined with off-screen check below
(r"(?:left|top)\s*:\s*-\d{3,}", "off-screen-position"),
(r"clip\s*:\s*rect\s*\(\s*0\s*,?\s*0\s*,?\s*0\s*,?\s*0\s*\)", "clip:rect(0)"),
(r"text-indent\s*:\s*-\d{3,}", "text-indent-offscreen"),
]
def extract_page(html: str, url: str, cfg: dict | None = None) -> dict:
"""
Full extraction from page HTML.
Returns a structured dict with all data needed for diffing and checking.
"""
cfg = cfg or {}
soup = BeautifulSoup(html, "lxml")
# Remove ignored selectors before text extraction
ignore_selectors = cfg.get("normalization", {}).get("ignore_selectors", [])
for sel in ignore_selectors:
for el in soup.select(sel):
el.decompose()
return {
"url": url,
"text": _visible_text(soup),
"links": _extract_links(soup, url),
"metadata": _extract_metadata(soup),
"hidden_content": _find_hidden_content(soup),
"inline_scripts": _check_inline_scripts(soup),
"jsonld": _extract_jsonld(soup),
"comment_links": _find_comment_links(html, url),
}
# ---------------------------------------------------------------------------
# Text
# ---------------------------------------------------------------------------
def _visible_text(soup: BeautifulSoup) -> str:
"""Extract visible text — strips scripts, styles, noscript, head."""
work = soup.__copy__()
for tag in work(["script", "style", "noscript", "head", "meta", "link"]):
tag.decompose()
text = work.get_text(separator=" ")
return re.sub(r"\s+", " ", text).strip()
# ---------------------------------------------------------------------------
# Links
# ---------------------------------------------------------------------------
def _extract_links(soup: BeautifulSoup, base_url: str) -> dict:
"""
Extract all link categories:
a, link, script[src], img[src], iframe[src], form[action],
meta_refresh, canonical
Each entry: {url, class: 'internal'|'external', ...extra}
"""
base_netloc = urlparse(base_url).netloc
def classify(u: str) -> str:
return "internal" if urlparse(u).netloc == base_netloc else "external"
def resolve(href: str | None) -> str | None:
if not href:
return None
href = href.strip()
if href.startswith(("javascript:", "mailto:", "tel:", "data:", "#")):
return None
abs_url = urljoin(base_url, href)
p = urlparse(abs_url)
if p.scheme not in ("http", "https"):
return None
# Drop fragment
return abs_url.split("#")[0]
result: dict = {
"a": [],
"link_rel": [],
"script": [],
"img": [],
"iframe": [],
"form": [],
"meta_refresh": [],
"canonical": None,
}
for tag in soup.find_all("a", href=True):
u = resolve(tag["href"])
if u:
result["a"].append({
"url": u,
"text": tag.get_text(" ", strip=True)[:200],
"class": classify(u),
"rel": " ".join(tag.get("rel", [])),
})
for tag in soup.find_all("link", href=True):
u = resolve(tag["href"])
if not u:
continue
rel = " ".join(tag.get("rel", []))
if "canonical" in rel:
result["canonical"] = u
result["link_rel"].append({"url": u, "rel": rel, "class": classify(u)})
for tag in soup.find_all("script", src=True):
u = resolve(tag["src"])
if u:
result["script"].append({"url": u, "class": classify(u)})
for tag in soup.find_all("img", src=True):
u = resolve(tag["src"])
if u:
result["img"].append({"url": u, "class": classify(u)})
for tag in soup.find_all("iframe", src=True):
u = resolve(tag["src"])
if u:
result["iframe"].append({"url": u, "class": classify(u)})
for tag in soup.find_all("form", action=True):
u = resolve(tag["action"])
if u:
result["form"].append({"url": u, "class": classify(u)})
for tag in soup.find_all("meta"):
if tag.get("http-equiv", "").lower() == "refresh":
content = tag.get("content", "")
m = re.search(r"url\s*=\s*(.+)", content, re.I)
if m:
u = resolve(m.group(1).strip().strip("\"'"))
if u:
result["meta_refresh"].append(u)
return result
# ---------------------------------------------------------------------------
# Metadata
# ---------------------------------------------------------------------------
def _extract_metadata(soup: BeautifulSoup) -> dict:
"""Extract title, description, canonical, og-tags, robots, hreflang, H1/H2."""
meta: dict = {}
t = soup.find("title")
meta["title"] = t.get_text(strip=True) if t else None
canonical_tag = soup.find("link", rel="canonical")
if canonical_tag:
meta["canonical"] = canonical_tag.get("href", "").strip()
for tag in soup.find_all("meta"):
name = (tag.get("name") or "").lower().strip()
prop = (tag.get("property") or "").lower().strip()
content = tag.get("content", "")
if name == "description":
meta["description"] = content
elif name == "robots":
meta["robots"] = content
elif name == "keywords":
meta["keywords"] = content
elif prop.startswith("og:"):
meta[prop] = content
elif prop.startswith("twitter:"):
meta[prop] = content
hreflang = []
for tag in soup.find_all("link", rel="alternate"):
hl = tag.get("hreflang")
if hl:
hreflang.append({"lang": hl, "href": tag.get("href", "")})
meta["hreflang"] = hreflang
meta["h1"] = [h.get_text(" ", strip=True) for h in soup.find_all("h1")]
meta["h2"] = [h.get_text(" ", strip=True) for h in soup.find_all("h2")]
return meta
# ---------------------------------------------------------------------------
# Hidden content
# ---------------------------------------------------------------------------
def _find_hidden_content(soup: BeautifulSoup) -> list[dict]:
"""
Detect elements that are visually hidden but contain text or links.
Catches common SEO-spam hiding techniques.
"""
findings: list[dict] = []
for tag in soup.find_all(style=True):
style = tag.get("style", "")
matched_indicators = []
for pattern, label in _HIDDEN_STYLE_CHECKS:
if re.search(pattern, style, re.I):
matched_indicators.append(label)
if not matched_indicators:
continue
text = tag.get_text(" ", strip=True)[:300]
links = [a.get("href", "") for a in tag.find_all("a", href=True)]
if text or links:
findings.append({
"type": "inline_style_hidden",
"indicators": matched_indicators,
"tag": tag.name,
"text_preview": text,
"links": links,
})
# <noscript> with links
for tag in soup.find_all("noscript"):
links = re.findall(r'href=["\']([^"\']+)["\']', str(tag), re.I)
if links:
findings.append({
"type": "noscript_links",
"links": links,
"text_preview": tag.get_text(" ", strip=True)[:200],
})
return findings
# ---------------------------------------------------------------------------
# Inline scripts
# ---------------------------------------------------------------------------
def _check_inline_scripts(soup: BeautifulSoup) -> list[dict]:
"""Find inline <script> blocks that match suspicious patterns."""
findings: list[dict] = []
for i, tag in enumerate(soup.find_all("script"), 1):
if tag.get("src"):
continue
code = tag.get_text() or ""
hits = [label for pat, label in _SUSPICIOUS_INLINE if re.search(pat, code, re.I)]
if hits:
findings.append({
"script_index": i,
"patterns": hits,
"preview": code[:400],
})
return findings
# ---------------------------------------------------------------------------
# JSON-LD
# ---------------------------------------------------------------------------
def _extract_jsonld(soup: BeautifulSoup) -> list[dict]:
"""Extract JSON-LD blocks and flag unexpected @types."""
results: list[dict] = []
for tag in soup.find_all("script", type="application/ld+json"):
raw = tag.string or ""
try:
data = json.loads(raw)
except (json.JSONDecodeError, ValueError):
results.append({"type": "PARSE_ERROR", "unexpected": True, "preview": raw[:200]})
continue
items = data if isinstance(data, list) else [data]
for item in items:
t = item.get("@type", "Unknown")
results.append({"type": t, "preview": json.dumps(item, ensure_ascii=False)[:500]})
return results
# ---------------------------------------------------------------------------
# Comment links
# ---------------------------------------------------------------------------
def _find_comment_links(html: str, base_url: str) -> list[str]:
"""Find URLs hidden inside HTML comments."""
found: list[str] = []
for comment in re.findall(r"<!--(.*?)-->", html, re.S):
for href in re.findall(r'href=["\']([^"\']+)["\']', comment, re.I):
if href.startswith(("http://", "https://", "/")):
found.append(urljoin(base_url, href))
return found

0
tests/__init__.py Normal file
View file

161
tests/test_baseline.py Normal file
View file

@ -0,0 +1,161 @@
"""Tests for snapshot saving, baseline management and approval workflow."""
import json
import pytest
from pathlib import Path
from scanner.baseline import BaselineManager, url_key
def _sample_page(url: str, text: str = "Inhalt") -> dict:
return {
"url": url,
"final_url": url,
"status": 200,
"content_type": "text/html",
"response_headers": {},
"text": text,
"links": {"a": [], "canonical": None, "meta_refresh": []},
"metadata": {"title": "T", "h1": [], "h2": []},
"hidden_content": [],
"inline_scripts": [],
"jsonld": [],
"comment_links": [],
}
@pytest.fixture
def bm(tmp_path: Path) -> BaselineManager:
return BaselineManager(tmp_path / "data")
class TestUrlKey:
def test_deterministic(self):
assert url_key("https://example.com/") == url_key("https://example.com/")
def test_different_urls_different_keys(self):
assert url_key("https://a.com/") != url_key("https://b.com/")
def test_16_hex_chars(self):
k = url_key("https://example.com/")
assert len(k) == 16
assert all(c in "0123456789abcdef" for c in k)
class TestSaveSnapshot:
def test_creates_manifest(self, bm, tmp_path):
pages = [_sample_page("https://b.info/")]
snap_dir = bm.save_snapshot(pages)
assert (snap_dir / "manifest.json").exists()
def test_manifest_content(self, bm):
pages = [_sample_page("https://b.info/"), _sample_page("https://b.info/p/")]
snap_dir = bm.save_snapshot(pages)
manifest = json.loads((snap_dir / "manifest.json").read_text())
assert manifest["page_count"] == 2
assert "https://b.info/" in manifest["urls"]
def test_creates_page_files(self, bm):
pages = [_sample_page("https://b.info/")]
snap_dir = bm.save_snapshot(pages)
page_files = list((snap_dir / "pages").glob("*.json"))
assert len(page_files) == 1
def test_latest_snapshot_dir(self, bm):
assert bm.latest_snapshot_dir() is None
bm.save_snapshot([_sample_page("https://b.info/")])
assert bm.latest_snapshot_dir() is not None
class TestLoadSnapshot:
def test_loads_saved_snapshot(self, bm):
pages = [_sample_page("https://b.info/"), _sample_page("https://b.info/p/")]
bm.save_snapshot(pages)
snap = bm.load_snapshot()
assert "https://b.info/" in snap["pages"]
assert "https://b.info/p/" in snap["pages"]
def test_returns_empty_when_no_snapshot(self, bm):
result = bm.load_snapshot()
assert result == {}
class TestBaselineExists:
def test_false_before_approve(self, bm):
assert not bm.baseline_exists()
def test_true_after_approve(self, bm):
bm.save_snapshot([_sample_page("https://b.info/")])
bm.approve_all(note="test")
assert bm.baseline_exists()
class TestApproveAll:
def test_approve_all_creates_baseline(self, bm):
pages = [_sample_page("https://b.info/"), _sample_page("https://b.info/p/")]
bm.save_snapshot(pages)
approved = bm.approve_all(note="Initial baseline")
assert "https://b.info/" in approved
assert "https://b.info/p/" in approved
assert bm.baseline_exists()
def test_manifest_contains_approval_metadata(self, bm):
bm.save_snapshot([_sample_page("https://b.info/")])
bm.approve_all(note="test note", approved_by="testuser")
manifest = json.loads((bm.baseline_dir / "manifest.json").read_text())
assert manifest["note"] == "test note"
assert manifest["approved_by"] == "testuser"
assert "approved_at" in manifest
def test_approve_all_raises_without_snapshot(self, bm):
with pytest.raises(RuntimeError):
bm.approve_all()
class TestApproveUrl:
def test_approve_single_url(self, bm):
pages = [_sample_page("https://b.info/"), _sample_page("https://b.info/p/")]
bm.save_snapshot(pages)
result = bm.approve_url("https://b.info/")
assert result is True
assert bm.baseline_exists()
baseline = bm.load_baseline()
assert "https://b.info/" in baseline["pages"]
assert "https://b.info/p/" not in baseline["pages"]
def test_approve_missing_url_returns_false(self, bm):
bm.save_snapshot([_sample_page("https://b.info/")])
result = bm.approve_url("https://b.info/does-not-exist/")
assert result is False
class TestLoadBaseline:
def test_loads_approved_pages(self, bm):
pages = [_sample_page("https://b.info/", "Text A")]
bm.save_snapshot(pages)
bm.approve_all()
baseline = bm.load_baseline()
assert "https://b.info/" in baseline["pages"]
assert baseline["pages"]["https://b.info/"]["text"] == "Text A"
def test_never_auto_updates(self, bm):
"""Baseline must not change unless approve is explicitly called."""
bm.save_snapshot([_sample_page("https://b.info/", "Original")])
bm.approve_all()
# Second snapshot with different content
bm.save_snapshot([_sample_page("https://b.info/", "Manipuliert")])
# Baseline must still contain original content
baseline = bm.load_baseline()
assert baseline["pages"]["https://b.info/"]["text"] == "Original"
class TestRebuildBaseline:
def test_rebuild_replaces_baseline(self, bm):
bm.save_snapshot([_sample_page("https://b.info/", "Alt")])
bm.approve_all()
bm.save_snapshot([_sample_page("https://b.info/", "Neu")])
bm.rebuild_baseline(note="Major update")
baseline = bm.load_baseline()
assert baseline["pages"]["https://b.info/"]["text"] == "Neu"

63
tests/test_crawler.py Normal file
View file

@ -0,0 +1,63 @@
"""Tests for crawler URL normalization and crawl filtering."""
import pytest
from scanner.crawler import normalize_url, should_crawl
SKIP_EXT = [".jpg", ".jpeg", ".png", ".pdf", ".zip", ".css", ".js"]
class TestNormalizeUrl:
def test_strips_fragment(self):
assert normalize_url("https://example.com/page#section") == "https://example.com/page"
def test_lowercase_host(self):
assert normalize_url("https://EXAMPLE.COM/path") == "https://example.com/path"
def test_non_http_returns_none(self):
assert normalize_url("javascript:void(0)") is None
assert normalize_url("mailto:foo@bar.com") is None
assert normalize_url("ftp://example.com") is None
def test_sorts_query_params(self):
a = normalize_url("https://example.com/?z=1&a=2")
b = normalize_url("https://example.com/?a=2&z=1")
assert a == b
def test_strips_tracking_params(self):
url = normalize_url("https://example.com/?utm_source=google&page=1")
assert "utm_source" not in url
assert "page=1" in url
def test_strips_session_params(self):
url = normalize_url("https://example.com/?sid=abc123&id=5")
assert "sid" not in url
assert "id=5" in url
def test_adds_default_path(self):
result = normalize_url("https://example.com")
assert result.endswith("/") or "example.com" in result
def test_empty_string_returns_none(self):
assert normalize_url("") is None
class TestShouldCrawl:
def test_same_host_plain_page(self):
assert should_crawl("https://example.com/about", "example.com", SKIP_EXT)
def test_different_host_rejected(self):
assert not should_crawl("https://other.com/page", "example.com", SKIP_EXT)
def test_image_rejected(self):
assert not should_crawl("https://example.com/photo.jpg", "example.com", SKIP_EXT)
def test_pdf_rejected(self):
assert not should_crawl("https://example.com/doc.pdf", "example.com", SKIP_EXT)
def test_html_accepted(self):
assert should_crawl("https://example.com/page.html", "example.com", SKIP_EXT)
def test_trailing_slash_accepted(self):
assert should_crawl("https://example.com/section/", "example.com", SKIP_EXT)
def test_utm_query_param_rejected(self):
assert not should_crawl("https://example.com/?utm_source=x", "example.com", SKIP_EXT)

250
tests/test_differ.py Normal file
View file

@ -0,0 +1,250 @@
"""Tests for diff logic and scoring."""
import pytest
from scanner.differ import (
compare_snapshots,
diff_link_set,
diff_metadata,
diff_text,
normalize_text,
score_diff,
text_hash,
)
BASE_CFG = {
"scoring": {
"new_external_domain": 50,
"new_internal_url": 30,
"missing_internal_url": 20,
"hidden_content": 40,
"unexpected_canonical": 35,
"meta_refresh": 40,
"unexpected_jsonld": 35,
"large_text_addition": 20,
"new_inline_script_suspicious": 30,
"comment_links": 25,
"noscript_links": 25,
"missing_security_header": 5,
"suspicious_content_pattern": 50,
},
"thresholds": {"yellow": 20, "red": 60, "large_text_block_chars": 200},
"normalization": {"collapse_whitespace": True, "ignore_patterns": []},
}
class TestNormalizeText:
def test_collapses_whitespace(self):
result = normalize_text("Hello World\n\nFoo", BASE_CFG)
assert result == "Hello World Foo"
def test_applies_ignore_patterns(self):
cfg = dict(BASE_CFG)
cfg["normalization"] = {
"collapse_whitespace": True,
"ignore_patterns": [r"\d{2}\.\d{2}\.\d{4}"],
}
result = normalize_text("Stand: 12.06.2026 Text", cfg)
assert "[IGNORED]" in result
assert "12.06.2026" not in result
class TestDiffText:
def test_identical_texts_produce_empty_diff(self):
assert diff_text("Hello", "Hello") == []
def test_addition_shows_plus_line(self):
lines = diff_text("Hello", "Hello\nWorld")
assert any(l.startswith("+") and "World" in l for l in lines)
def test_removal_shows_minus_line(self):
lines = diff_text("Hello\nWorld", "Hello")
assert any(l.startswith("-") and "World" in l for l in lines)
class TestDiffLinkSet:
def test_detects_added_link(self):
old = [{"url": "https://example.com/a"}]
new = [{"url": "https://example.com/a"}, {"url": "https://example.com/b"}]
result = diff_link_set(old, new)
assert "https://example.com/b" in result["added"]
assert result["removed"] == []
def test_detects_removed_link(self):
old = [{"url": "https://example.com/a"}, {"url": "https://example.com/b"}]
new = [{"url": "https://example.com/a"}]
result = diff_link_set(old, new)
assert "https://example.com/b" in result["removed"]
def test_no_change(self):
links = [{"url": "https://example.com/a"}]
result = diff_link_set(links, links)
assert result == {"added": [], "removed": []}
class TestDiffMetadata:
def test_detects_title_change(self):
old = {"title": "Alte Seite", "h1": [], "h2": []}
new = {"title": "Neue Seite", "h1": [], "h2": []}
result = diff_metadata(old, new)
assert "title" in result
assert result["title"]["old"] == "Alte Seite"
def test_no_change(self):
meta = {"title": "Seite", "description": "Desc", "h1": [], "h2": []}
assert diff_metadata(meta, meta) == {}
def test_detects_canonical_change(self):
old = {"canonical": "https://bredelar.info/", "h1": [], "h2": []}
new = {"canonical": "https://spam.example.com/", "h1": [], "h2": []}
result = diff_metadata(old, new)
assert "canonical" in result
class TestCompareSnapshots:
def _make_page(self, url, text="Inhalt", links=None, hidden=None):
return {
"url": url,
"text": text,
"links": links or {"a": [], "script": [], "iframe": [], "link_rel": [], "form": [], "meta_refresh": [], "canonical": None},
"metadata": {"title": "T", "h1": [], "h2": []},
"hidden_content": hidden or [],
"inline_scripts": [],
"jsonld": [],
"comment_links": [],
}
def test_new_url_detected(self):
baseline = {"https://b.info/": self._make_page("https://b.info/")}
current = {
"https://b.info/": self._make_page("https://b.info/"),
"https://b.info/new/": self._make_page("https://b.info/new/"),
}
result = compare_snapshots(baseline, current, BASE_CFG)
assert "https://b.info/new/" in result["new_internal_urls"]
def test_missing_url_detected(self):
baseline = {
"https://b.info/": self._make_page("https://b.info/"),
"https://b.info/old/": self._make_page("https://b.info/old/"),
}
current = {"https://b.info/": self._make_page("https://b.info/")}
result = compare_snapshots(baseline, current, BASE_CFG)
assert "https://b.info/old/" in result["missing_internal_urls"]
def test_unchanged_page_not_in_diffs(self):
page = self._make_page("https://b.info/", "Gleicher Inhalt")
result = compare_snapshots({"https://b.info/": page}, {"https://b.info/": page}, BASE_CFG)
assert result["changed_pages"] == 0
def test_text_change_detected(self):
old = self._make_page("https://b.info/", "Alter Text")
new = self._make_page("https://b.info/", "Alter Text plus neuer Spam-Block im SEO-Kontext")
result = compare_snapshots({"https://b.info/": old}, {"https://b.info/": new}, BASE_CFG)
assert result["changed_pages"] == 1
def test_new_external_domain_detected(self):
def _page_with_ext(url, ext_url):
p = self._make_page(url)
p["links"]["a"] = [{"url": ext_url, "class": "external", "text": "x", "rel": ""}]
return p
baseline = {"https://b.info/": self._make_page("https://b.info/")}
current = {"https://b.info/": _page_with_ext("https://b.info/", "https://spam.example.com/")}
result = compare_snapshots(baseline, current, BASE_CFG)
assert "spam.example.com" in result["new_external_domains"]
class TestScoreDiff:
def test_green_for_no_changes(self):
result = score_diff({
"new_internal_urls": [],
"missing_internal_urls": [],
"new_external_domains": [],
"page_diffs": [],
"new_pages_analysis": [],
}, BASE_CFG)
assert result["level"] == "green"
assert result["exit_code"] == 0
def test_yellow_for_new_url(self):
result = score_diff({
"new_internal_urls": ["https://b.info/new/"],
"missing_internal_urls": [],
"new_external_domains": [],
"page_diffs": [],
"new_pages_analysis": [],
}, BASE_CFG)
assert result["level"] == "yellow"
assert result["exit_code"] == 1
def test_yellow_for_single_new_external_domain(self):
# One new external domain = 50 pts → yellow (red threshold is 60)
result = score_diff({
"new_internal_urls": [],
"missing_internal_urls": [],
"new_external_domains": ["pharmacy.example.com"],
"page_diffs": [],
"new_pages_analysis": [],
}, BASE_CFG)
assert result["level"] == "yellow"
assert result["exit_code"] == 1
def test_red_for_new_external_domain_plus_hidden_content(self):
# External domain (50) + hidden content (40) = 90 pts → red
result = score_diff({
"new_internal_urls": [],
"missing_internal_urls": [],
"new_external_domains": ["pharmacy.example.com"],
"page_diffs": [{
"url": "https://b.info/",
"text_diff": [],
"added_chars": 0,
"link_diff": {},
"meta_diff": {},
"new_hidden_content": [{"type": "inline_style_hidden", "indicators": ["display:none"]}],
"new_inline_scripts": [],
"new_jsonld": [],
"new_meta_refresh": [],
"canonical_changed": False,
"new_comment_links": [],
"old_canonical": None,
"new_canonical": None,
}],
"new_pages_analysis": [],
}, BASE_CFG)
assert result["level"] == "red"
assert result["exit_code"] == 2
def test_yellow_for_hidden_content_alone(self):
# One hidden content finding = 40 pts → yellow (below red threshold of 60)
result = score_diff({
"new_internal_urls": [],
"missing_internal_urls": [],
"new_external_domains": [],
"page_diffs": [{
"url": "https://b.info/",
"text_diff": [],
"added_chars": 0,
"link_diff": {},
"meta_diff": {},
"new_hidden_content": [{"type": "inline_style_hidden", "indicators": ["display:none"]}],
"new_inline_scripts": [],
"new_jsonld": [],
"new_meta_refresh": [],
"canonical_changed": False,
"new_comment_links": [],
"old_canonical": None,
"new_canonical": None,
}],
"new_pages_analysis": [],
}, BASE_CFG)
assert result["level"] == "yellow"
def test_reasons_include_domain(self):
result = score_diff({
"new_internal_urls": [],
"missing_internal_urls": [],
"new_external_domains": ["spam.com"],
"page_diffs": [],
"new_pages_analysis": [],
}, BASE_CFG)
assert any("spam.com" in r for r in result["reasons"])

152
tests/test_extractor.py Normal file
View file

@ -0,0 +1,152 @@
"""Tests for HTML extraction: links, metadata, hidden content, JSON-LD."""
import pytest
from scanner.extractor import extract_page
BASE_URL = "https://bredelar.info/"
def _extract(html: str, url: str = BASE_URL) -> dict:
return extract_page(html, url)
class TestVisibleText:
def test_strips_scripts(self):
result = _extract("<html><script>var x=1;</script><body>Hallo Welt</body></html>")
assert "var x" not in result["text"]
assert "Hallo Welt" in result["text"]
def test_strips_style(self):
result = _extract("<html><style>.cls{color:red}</style><body>Text</body></html>")
assert ".cls" not in result["text"]
assert "Text" in result["text"]
def test_collapses_whitespace(self):
result = _extract("<body><p>Hello World</p></body>")
assert "Hello World" in result["text"]
class TestLinkExtraction:
def test_extracts_anchor_links(self):
result = _extract('<body><a href="/page">Link</a></body>')
urls = [l["url"] for l in result["links"]["a"]]
assert any("/page" in u for u in urls)
def test_classifies_internal_vs_external(self):
html = (
'<body>'
'<a href="/internal">intern</a>'
'<a href="https://spam.example.com/x">extern</a>'
'</body>'
)
result = _extract(html)
classes = {l["url"]: l["class"] for l in result["links"]["a"]}
assert any(v == "internal" for v in classes.values())
assert any(v == "external" for v in classes.values())
def test_extracts_canonical(self):
html = '<head><link rel="canonical" href="https://bredelar.info/page"/></head>'
result = _extract(html)
assert result["links"]["canonical"] == "https://bredelar.info/page"
def test_extracts_meta_refresh(self):
html = '<head><meta http-equiv="refresh" content="0; url=https://malware.example.com"/></head>'
result = _extract(html)
assert any("malware.example.com" in u for u in result["links"]["meta_refresh"])
def test_ignores_javascript_hrefs(self):
result = _extract('<body><a href="javascript:void(0)">JS</a></body>')
assert result["links"]["a"] == []
def test_extracts_iframe_src(self):
result = _extract('<body><iframe src="https://evil.com/x"></iframe></body>')
iframe_urls = [l["url"] for l in result["links"]["iframe"]]
assert any("evil.com" in u for u in iframe_urls)
class TestMetadata:
def test_extracts_title(self):
result = _extract("<html><head><title>Bredelar</title></head></html>")
assert result["metadata"]["title"] == "Bredelar"
def test_extracts_description(self):
html = '<head><meta name="description" content="Ein Dorf in NRW"/></head>'
result = _extract(html)
assert result["metadata"]["description"] == "Ein Dorf in NRW"
def test_extracts_og_tags(self):
html = '<head><meta property="og:title" content="OG-Titel"/></head>'
result = _extract(html)
assert result["metadata"].get("og:title") == "OG-Titel"
def test_extracts_h1(self):
result = _extract("<body><h1>Hauptüberschrift</h1></body>")
assert "Hauptüberschrift" in result["metadata"]["h1"]
class TestHiddenContent:
def test_detects_display_none(self):
html = '<body><div style="display:none"><a href="https://spam.com">Spam</a></div></body>'
result = _extract(html)
assert len(result["hidden_content"]) > 0
assert any("display:none" in str(h.get("indicators", [])) for h in result["hidden_content"])
def test_detects_visibility_hidden(self):
html = '<body><span style="visibility:hidden">Spam Text</span></body>'
result = _extract(html)
assert len(result["hidden_content"]) > 0
def test_detects_offscreen_position(self):
html = '<body><div style="position:absolute; left:-9999px">Hidden</div></body>'
result = _extract(html)
assert len(result["hidden_content"]) > 0
def test_detects_noscript_links(self):
html = '<body><noscript><a href="https://spam.com/x">Spam</a></noscript></body>'
result = _extract(html)
noscript = [h for h in result["hidden_content"] if h.get("type") == "noscript_links"]
assert len(noscript) > 0
def test_visible_element_not_flagged(self):
html = '<body><div style="color:red">Sichtbarer Text</div></body>'
result = _extract(html)
# color:red should NOT be flagged (not a hiding technique)
assert not any(
"red" in str(h) for h in result["hidden_content"]
)
class TestCommentLinks:
def test_finds_link_in_comment(self):
html = '<!-- <a href="https://hidden-spam.com/x">click</a> --><body></body>'
result = _extract(html)
assert any("hidden-spam.com" in u for u in result["comment_links"])
def test_no_false_positives(self):
html = '<!-- This is a normal comment --><body>Text</body>'
result = _extract(html)
assert result["comment_links"] == []
class TestJsonLd:
def test_extracts_known_type(self):
html = (
'<script type="application/ld+json">'
'{"@type": "Organization", "name": "Bredelar"}'
'</script>'
)
result = _extract(html)
assert any(j["type"] == "Organization" for j in result["jsonld"])
def test_handles_parse_error(self):
html = '<script type="application/ld+json">{broken json}</script>'
result = _extract(html)
assert any(j["type"] == "PARSE_ERROR" for j in result["jsonld"])
def test_extracts_unexpected_type(self):
html = (
'<script type="application/ld+json">'
'{"@type": "Pharmacy", "name": "SpamShop"}'
'</script>'
)
result = _extract(html)
assert any(j["type"] == "Pharmacy" for j in result["jsonld"])