From 68243a97b989337b9dacf116141a0e3ba42ebf26 Mon Sep 17 00:00:00 2001 From: Your Name Date: Fri, 12 Jun 2026 03:20:32 +0200 Subject: [PATCH] fix: Bugfixes, toten Code entfernt, Cache-Normalisierung & Session Echte Bugs: - extractor: JSON-LD-Crash bei Nicht-Dict-Items (isinstance-Guard) - baseline/__main__: Crawl-Fehler ins Snapshot-Manifest -> erscheinen im Report - __main__: Whitelist-/Header-/Webshell-Checks nur noch auf status==200 - crawler: Noise-Param-Regex auf Voll-Key (view=/value= nicht mehr verworfen) - differ/__main__: unerwartetes JSON-LD via eigenem Kanal, auch auf unveraenderten Seiten erkannt, kein Re-Alert auf bereits genehmigte Typen Aufgeraeumt: - checker: check_internal_paths, find_broken_pages, canonical_is_hijacked, check_jsonld_types entfernt (nicht verdrahtet) - allowed_paths.yaml-Logik und tote Scoring-Keys entfernt - tote Imports entfernt Neuer aktiver Schutz: - Webshell-/Backdoor-Dateinamen-Check verdrahtet (suspicious_filename: 60). Regex wortgrenzen-verankert gegen Fehlalarm auf PSEMailerAntispam.js Effizienz/Struktur: - crawler nutzt requests.Session (keep-alive) - Cache-Buster-Normalisierung an einer Stelle (Extraktion) -> stabile Snapshots, differ wieder reiner Set-Diff Tests: 111 gruen (neu: test_checker.py + Regressionstests) CLAUDE.md aktualisiert Co-Authored-By: Claude Opus 4.8 --- CLAUDE.md | 16 +++++---- config.yaml | 5 +-- scanner/__main__.py | 72 +++++++++++++++++++++------------------ scanner/baseline.py | 14 ++++++-- scanner/checker.py | 74 +++-------------------------------------- scanner/config.py | 5 +-- scanner/crawler.py | 67 ++++++++++++++++++++++++------------- scanner/differ.py | 37 ++++++++++++++++----- scanner/extractor.py | 22 ++++++++++-- tests/test_checker.py | 63 +++++++++++++++++++++++++++++++++++ tests/test_crawler.py | 10 ++++++ tests/test_differ.py | 62 ++++++++++++++++++++++++++++++++++ tests/test_extractor.py | 29 ++++++++++++++++ 13 files changed, 323 insertions(+), 153 deletions(-) create mode 100644 tests/test_checker.py diff --git a/CLAUDE.md b/CLAUDE.md index cc6ff25..06a586d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -46,11 +46,11 @@ Crawler → Extractor → BaselineManager (snapshot) ### 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 `.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`. +- **`crawler.py`** — breadth-first crawl using a shared `requests.Session` (keep-alive). `normalize_url()` is the canonical URL form used everywhere (strips fragment, lowercases host, removes noise query params via whole-key matching in `_is_noise_param`, 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`). `strip_cache_params()` removes cache-buster query params (e.g. `style.css?v=123`) from asset URLs (`link_rel`/`script`/`img`) **at extraction time**, so snapshots store stable URLs — this is the single source of truth for cache-buster normalisation. +- **`baseline.py`** — `BaselineManager` handles two separate stores: `data/snapshots/YYYYMMDD_HHMMSS/` (raw crawl output) and `data/baseline/` (approved reference). Each page is stored as `.json`. Snapshot `manifest.json` also records crawl `errors`. `approve_*` methods are the only path into the baseline. +- **`differ.py`** — `compare_snapshots(baseline_pages, current_pages, cfg)` returns a flat diff dict (incl. `unexpected_jsonld`, `new_broken_urls`, `known_broken_urls`). `score_diff(diff, cfg)` maps it to `{score, level, reasons, exit_code}`. Thresholds: yellow ≥ 20, red ≥ 60 (configurable). Unexpected JSON-LD `@type` is detected independently of page diffs (flags only types not allowed AND not already in the baseline for that URL). A single new external domain = 50 pts (yellow); combined with hidden content = 90 pts (red). +- **`checker.py`** — stateless helpers for whitelist enforcement (`check_links_against_whitelist`), security-header auditing (`check_security_headers`, informational only — not scored), and webshell/backdoor filename detection (`is_suspicious_url`, word-boundary anchored to avoid false positives). Called from `__main__.cmd_check` only on `status == 200` pages. - **`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. @@ -62,7 +62,6 @@ 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 ``` @@ -70,6 +69,7 @@ config/ignore_rules.yaml # normalization selectors + regex patterns | Event | Points | |---|---| +| Suspicious filename (webshell/backdoor link) | 60 | | New external domain | 50 | | Hidden content (CSS-hidden element with links/text) | 40 | | Meta-Refresh | 40 | @@ -80,4 +80,6 @@ config/ignore_rules.yaml # normalization selectors + regex patterns | Comment-embedded link | 25 | | Large text addition (>200 chars) | 20 | | Missing internal URL | 20 | -| Missing security header | 5 | +| New broken link (4xx/5xx, not in baseline) | 20 | + +Security headers are audited and reported but **not** scored. Known broken links (already in the baseline) are reported but not scored. diff --git a/config.yaml b/config.yaml index a9a29d1..8f15abd 100644 --- a/config.yaml +++ b/config.yaml @@ -45,12 +45,9 @@ scoring: 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 new_broken_link: 20 # neue 4xx/5xx-Seite, die in Baseline nicht vorhanden war + suspicious_filename: 60 # Link auf Webshell-/Backdoor-typischen Dateinamen thresholds: yellow: 20 # ab diesem Score: Warnung (gelb) diff --git a/scanner/__main__.py b/scanner/__main__.py index 3369246..e111209 100644 --- a/scanner/__main__.py +++ b/scanner/__main__.py @@ -24,8 +24,6 @@ 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, @@ -33,7 +31,7 @@ from .checker import ( ) from .config import load_config, resolve_paths from .crawler import Crawler -from .differ import compare_snapshots, normalize_text, score_diff +from .differ import compare_snapshots, score_diff from .extractor import extract_page @@ -82,11 +80,9 @@ def _crawl_and_extract(cfg: dict) -> tuple[list[dict], list[dict]]: # Whitelist loading # --------------------------------------------------------------------------- -def _load_whitelists(cfg: dict) -> tuple[set[str], set[str]]: +def _load_allowed_external(cfg: dict) -> 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 + return load_whitelist(config_dir / "allowed_external.yaml") # --------------------------------------------------------------------------- @@ -154,6 +150,8 @@ def _render_markdown(r: dict) -> str: f"| Neue externe Domains | {len(d.get('new_external_domains', []))} |", f"| Kaputte Links (neu) | {len(d.get('new_broken_urls', []))} |", f"| Kaputte Links (bekannt) | {len(d.get('known_broken_urls', []))} |", + f"| Unerwartetes JSON-LD | {len(d.get('unexpected_jsonld', []))} |", + f"| Verdächtige Dateinamen | {len(d.get('suspicious_filenames', []))} |", f"| Whitelist-Verstösse | {len(r.get('whitelist_violations', []))} |", f"| Crawl-Fehler | {len(r.get('crawl_errors', []))} |", "", @@ -244,6 +242,16 @@ def _render_markdown(r: dict) -> str: for entry in known_broken: lines.append(f"- [bekannt] HTTP {entry['status']} `{entry['url']}`") + if d.get("unexpected_jsonld"): + lines += ["", "## Unerwartetes JSON-LD", ""] + for entry in d["unexpected_jsonld"]: + lines.append(f"- `{entry['url']}`: @type={entry['type']}") + + if d.get("suspicious_filenames"): + lines += ["", "## Verdächtige Dateinamen", ""] + for entry in d["suspicious_filenames"]: + lines.append(f"- `{entry['page']}` → `{entry['url']}` ({entry['link_type']})") + if r.get("whitelist_violations"): lines += ["", "## Whitelist-Verstösse", ""] for v in r["whitelist_violations"][:50]: @@ -311,14 +319,13 @@ def cmd_init(args: argparse.Namespace, cfg: dict) -> int: return 1 bm = BaselineManager(cfg["data_dir"]) - snap_dir = bm.save_snapshot(pages) + snap_dir = bm.save_snapshot(pages, errors) 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 + allowed_ext = _load_allowed_external(cfg) print("=== Gefundene interne URLs ===") for p in pages: @@ -383,7 +390,7 @@ def cmd_crawl(args: argparse.Namespace, cfg: dict) -> int: print(f"Crawle {cfg['target']} ...") pages, errors = _crawl_and_extract(cfg) bm = BaselineManager(cfg["data_dir"]) - snap_dir = bm.save_snapshot(pages) + snap_dir = bm.save_snapshot(pages, errors) print(f"Snapshot: {snap_dir}") print(f" Seiten: {len(pages)}") print(f" Fehler: {len(errors)}") @@ -418,13 +425,18 @@ def cmd_check(args: argparse.Namespace, cfg: dict) -> int: diff = compare_snapshots(baseline["pages"], snap["pages"], cfg) - # Whitelist checks on all current pages - allowed_ext, allowed_int = _load_whitelists(cfg) + # Whitelist / header / webshell checks — only on successfully fetched HTML + # pages (status 200). 404 / asset responses would otherwise create noise. + allowed_ext = _load_allowed_external(cfg) base_netloc = urlparse(cfg["target"]).netloc wl_violations: list[dict] = [] missing_sec: dict = {} + suspicious_filenames: list[dict] = [] for page in snap["pages"].values(): + if page.get("status") != 200 or not page.get("links"): + continue + violations = check_links_against_whitelist(page, allowed_ext, base_netloc) if violations: wl_violations.extend({"page": page["url"], **v} for v in violations) @@ -434,20 +446,19 @@ def cmd_check(args: argparse.Namespace, cfg: dict) -> int: 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}) + # Webshell / backdoor filename in any referenced link + for ltype, links in page.get("links", {}).items(): + if not isinstance(links, list): + continue + for l in links: + if isinstance(l, dict) and is_suspicious_url(l.get("url", "")): + suspicious_filenames.append({ + "page": page["url"], + "url": l["url"], + "link_type": ltype, + }) + + diff["suspicious_filenames"] = suspicious_filenames # Score with whitelist-violation contribution for v in wl_violations: @@ -460,10 +471,7 @@ def cmd_check(args: argparse.Namespace, cfg: dict) -> int: assessment = score_diff(diff, cfg) - crawl_errors: list[dict] = [] - if getattr(args, "with_errors", False): - # errors were not persisted in snapshot, skip - pass + crawl_errors = snap.get("errors", []) report = _build_report( cfg["target"], diff, assessment, wl_violations, missing_sec, @@ -485,7 +493,7 @@ def cmd_scan(args: argparse.Namespace, cfg: dict) -> int: print(f"Scanne {cfg['target']} ...") pages, errors = _crawl_and_extract(cfg) bm = BaselineManager(cfg["data_dir"]) - snap_dir = bm.save_snapshot(pages) + snap_dir = bm.save_snapshot(pages, errors) logger.info("Snapshot: %s (%d Seiten, %d Fehler)", snap_dir, len(pages), len(errors)) if not bm.baseline_exists(): diff --git a/scanner/baseline.py b/scanner/baseline.py index 3e37162..e0514c4 100644 --- a/scanner/baseline.py +++ b/scanner/baseline.py @@ -31,9 +31,11 @@ class BaselineManager: # Snapshot operations # ------------------------------------------------------------------ - def save_snapshot(self, pages: list[dict]) -> Path: + def save_snapshot(self, pages: list[dict], errors: list[dict] | None = None) -> Path: """ Persist a crawl+extraction result as a timestamped snapshot. + Crawl errors (network failures) are recorded in the manifest so they + survive into the comparison/report stage. Returns the snapshot directory path. """ ts = _utc_stamp() @@ -45,6 +47,7 @@ class BaselineManager: "timestamp": ts, "page_count": len(pages), "urls": [p["url"] for p in pages], + "errors": errors or [], } (snap_dir / "manifest.json").write_text( json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8" @@ -71,7 +74,7 @@ class BaselineManager: def load_snapshot(self, snap_dir: Optional[Path] = None) -> dict: """ Load a snapshot (default: latest). - Returns {manifest, pages: {url: page_dict}, dir: Path}. + Returns {manifest, pages: {url: page_dict}, errors: list, dir: Path}. """ if snap_dir is None: snap_dir = self.latest_snapshot_dir() @@ -85,7 +88,12 @@ class BaselineManager: page = json.loads(f.read_text(encoding="utf-8")) pages[page["url"]] = page - return {"manifest": manifest, "pages": pages, "dir": snap_dir} + return { + "manifest": manifest, + "pages": pages, + "errors": manifest.get("errors", []), + "dir": snap_dir, + } # ------------------------------------------------------------------ # Baseline operations diff --git a/scanner/checker.py b/scanner/checker.py index 5b801cf..700de67 100644 --- a/scanner/checker.py +++ b/scanner/checker.py @@ -11,8 +11,11 @@ from urllib.parse import urlparse import yaml +# Webshell / backdoor filename markers. Word boundaries are essential: +# without them "mailer" matches the legitimate "PSEMailerAntispam.js" and +# "gate" matches "navigate", producing false positives on every scan. _SUSPICIOUS_FILENAME_RE = re.compile( - r"(shell|backdoor|cmd|upload|gate|mailer|wso|c99|r57|b374k)", + r"\b(shell|backdoor|webshell|c99|r57|wso|b374k|cmd|reverse_shell|bindshell)\b", re.I, ) @@ -87,30 +90,6 @@ def check_links_against_whitelist( 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 # --------------------------------------------------------------------------- @@ -122,38 +101,6 @@ def check_security_headers(response_headers: dict, required: list[str]) -> list[ return [h for h in required if h.lower() not in present] -# --------------------------------------------------------------------------- -# Broken links -# --------------------------------------------------------------------------- - -def find_broken_pages(pages: dict) -> list[dict]: - """Return all pages whose HTTP status is >= 400 (broken / server error).""" - return [ - {"url": p["url"], "status": p.get("status", 0)} - for p in pages.values() - if p.get("status", 0) >= 400 - ] - - -# --------------------------------------------------------------------------- -# 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 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 # --------------------------------------------------------------------------- @@ -162,16 +109,3 @@ 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 - ] diff --git a/scanner/config.py b/scanner/config.py index b8e5009..96aab38 100644 --- a/scanner/config.py +++ b/scanner/config.py @@ -30,12 +30,9 @@ DEFAULT_CONFIG: dict = { "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, "new_broken_link": 20, + "suspicious_filename": 60, }, "thresholds": { "yellow": 20, diff --git a/scanner/crawler.py b/scanner/crawler.py index c8bb177..813186b 100644 --- a/scanner/crawler.py +++ b/scanner/crawler.py @@ -1,6 +1,5 @@ """Pure-Python crawler — no wget dependency.""" import logging -import re import time from urllib.parse import parse_qs, urlencode, urljoin, urlparse, urlunparse @@ -9,11 +8,20 @@ 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, -) +# Query-param keys that indicate calendar / session / tracking / cache noise. +# Matched as whole keys (not prefixes) so legitimate params like ?view= or +# ?value= are never silently dropped. utm_* is a known tracking prefix family. +_NOISE_PARAM_KEYS = { + "year", "month", "day", "date", "week", + "session", "token", "sid", "nonce", "fbclid", + "v", "ver", "version", "cache", "cb", "_", "_t", +} + + +def _is_noise_param(key: str) -> bool: + """True if a query-param key is calendar/session/tracking/cache noise.""" + k = key.lower() + return k in _NOISE_PARAM_KEYS or k.startswith("utm_") def normalize_url(url: str) -> str | None: @@ -36,7 +44,7 @@ def normalize_url(url: str) -> str | None: (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) + if not _is_noise_param(k) ) p = p._replace(query=urlencode(clean_params)) return urlunparse(p) @@ -53,8 +61,8 @@ def should_crawl(url: str, base_netloc: str, skip_extensions: list[str]) -> bool 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): + # Skip URLs whose query carries calendar/session/tracking noise + if p.query and any(_is_noise_param(k) for k in parse_qs(p.query)): return False return True @@ -68,6 +76,8 @@ class Crawler: self.skip_ext: list[str] = cfg["crawl"]["skip_extensions"] self.timeout: int = cfg["request_timeout"] self.headers = {"User-Agent": cfg["user_agent"]} + self.session = requests.Session() + self.session.headers.update(self.headers) def crawl(self) -> dict: """ @@ -82,6 +92,30 @@ class Crawler: pages: list[dict] = [] errors: list[dict] = [] + try: + self._crawl_loop(queue, seen, pages, errors) + finally: + self.session.close() + + 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 _crawl_loop( + self, + queue: list[str], + seen: set[str], + pages: list[dict], + errors: list[dict], + ) -> None: while queue and len(pages) < self.max_pages: url = queue.pop(0) if url in seen: @@ -93,9 +127,8 @@ class Crawler: logger.debug("Crawling %s", url) try: - resp = requests.get( + resp = self.session.get( url, - headers=self.headers, timeout=self.timeout, allow_redirects=True, ) @@ -123,18 +156,6 @@ class Crawler: 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 links worth following.""" result = [] diff --git a/scanner/differ.py b/scanner/differ.py index f6fa857..f3e15a0 100644 --- a/scanner/differ.py +++ b/scanner/differ.py @@ -43,7 +43,11 @@ def diff_text(old: str, new: str, context: int = 2) -> list[str]: def diff_link_set(old_list: list[dict], new_list: list[dict]) -> dict: - """Compare two link lists by URL.""" + """Compare two link lists by URL. + + Cache-buster query params are stripped at extraction time + (extractor.strip_cache_params), so stored URLs are already stable here. + """ 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 { @@ -228,6 +232,20 @@ def compare_snapshots(baseline_pages: dict, current_pages: dict, cfg: dict) -> d if url in current_pages ] + # Unexpected JSON-LD @types — independent of page_diffs. + # Only flag types that are NOT allowed AND were not already in the baseline + # for that URL (avoids re-alerting on a pre-approved page). + allowed_jsonld = set(cfg.get("allowed_jsonld_types", [])) + unexpected_jsonld: list[dict] = [] + for url, page in current_pages.items(): + base_types = { + j.get("type") for j in baseline_pages.get(url, {}).get("jsonld", []) + } + for j in page.get("jsonld", []): + t = j.get("type") + if t not in allowed_jsonld and t not in base_types: + unexpected_jsonld.append({"url": url, "type": t}) + return { "new_internal_urls": new_urls, "missing_internal_urls": missing_urls, @@ -235,6 +253,7 @@ def compare_snapshots(baseline_pages: dict, current_pages: dict, cfg: dict) -> d "removed_external_domains": sorted(old_ext - new_ext), "page_diffs": page_diffs, "new_pages_analysis": new_pages_analysis, + "unexpected_jsonld": unexpected_jsonld, "new_broken_urls": new_broken, "known_broken_urls": known_broken, "total_pages_checked": len(common_urls), @@ -305,11 +324,6 @@ def score_diff(diff: dict, cfg: dict) -> dict: 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')}") @@ -326,6 +340,14 @@ def score_diff(diff: dict, cfg: dict) -> dict: f"Neuer kaputte Link: {entry['url']} (HTTP {entry['status']})") # known_broken_urls are reported but not scored — they were already in the baseline + for entry in diff.get("unexpected_jsonld", []): + add(sc.get("unexpected_jsonld", 35), + f"{entry['url']}: Unerwartetes JSON-LD @type={entry['type']}") + + for entry in diff.get("suspicious_filenames", []): + add(sc.get("suspicious_filename", 60), + f"{entry['page']}: Verdächtiger Dateiname {entry['url']}") + # New pages: immediate suspicious content counts too for npa in diff.get("new_pages_analysis", []): u = npa["url"] @@ -337,8 +359,7 @@ def score_diff(diff: dict, cfg: dict) -> dict: 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") + # Unexpected JSON-LD on new pages is covered by the unexpected_jsonld channel. yellow = thr.get("yellow", 20) red = thr.get("red", 60) diff --git a/scanner/extractor.py b/scanner/extractor.py index e684f3f..8963b1f 100644 --- a/scanner/extractor.py +++ b/scanner/extractor.py @@ -6,9 +6,22 @@ inline script patterns, JSON-LD, and links buried in HTML comments. """ import json import re -from urllib.parse import urljoin, urlparse +from urllib.parse import urlencode, urljoin, urlparse, parse_qsl, urlunparse -from bs4 import BeautifulSoup, Comment +from bs4 import BeautifulSoup + +# Query-param keys that are pure cache-busters on asset URLs (CSS/JS/img). +# They change on every request and must be stripped before storing/diffing. +CACHE_PARAM_KEYS = {"v", "ver", "version", "cache", "cb", "_", "_t"} + + +def strip_cache_params(url: str) -> str: + """Remove cache-buster query params (e.g. style.css?v=12345) from a URL.""" + p = urlparse(url) + if not p.query: + return url + kept = [(k, v) for k, v in parse_qsl(p.query) if k.lower() not in CACHE_PARAM_KEYS] + return urlunparse(p._replace(query=urlencode(kept))) # Inline-script patterns that warrant flagging _SUSPICIOUS_INLINE = [ @@ -134,16 +147,19 @@ def _extract_links(soup: BeautifulSoup, base_url: str) -> dict: rel = " ".join(tag.get("rel", [])) if "canonical" in rel: result["canonical"] = u + u = strip_cache_params(u) # assets carry cache-busters → normalise away 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: + u = strip_cache_params(u) result["script"].append({"url": u, "class": classify(u)}) for tag in soup.find_all("img", src=True): u = resolve(tag["src"]) if u: + u = strip_cache_params(u) result["img"].append({"url": u, "class": classify(u)}) for tag in soup.find_all("iframe", src=True): @@ -293,6 +309,8 @@ def _extract_jsonld(soup: BeautifulSoup) -> list[dict]: continue items = data if isinstance(data, list) else [data] for item in items: + if not isinstance(item, dict): + continue # JSON-LD must be an object; ignore stray scalars/arrays t = item.get("@type", "Unknown") results.append({"type": t, "preview": json.dumps(item, ensure_ascii=False)[:500]}) return results diff --git a/tests/test_checker.py b/tests/test_checker.py new file mode 100644 index 0000000..7e4fe8d --- /dev/null +++ b/tests/test_checker.py @@ -0,0 +1,63 @@ +"""Tests for stateless checker helpers.""" +import pytest +from scanner.checker import ( + check_links_against_whitelist, + check_security_headers, + is_suspicious_url, +) + + +class TestIsSuspiciousUrl: + @pytest.mark.parametrize("url", [ + "https://b.info/uploads/shell.php", + "https://b.info/c99.php", + "https://b.info/wso.aspx", + "https://b.info/admin/backdoor.php", + "https://b.info/webshell.php", + ]) + def test_flags_webshell_names(self, url): + assert is_suspicious_url(url) is True + + @pytest.mark.parametrize("url", [ + "https://b.info/index.html", + "https://b.info/css/app.css", + "https://b.info/impressum", + "https://b.info/team/photo.jpg", + # Regression: legitimate site assets must NOT be flagged + "https://b.info/assets/js/PSEMailerAntispam.js", + "https://b.info/navigate.php", + "https://b.info/uploads/document.pdf", + ]) + def test_ignores_normal_names(self, url): + assert is_suspicious_url(url) is False + + +class TestSecurityHeaders: + def test_reports_missing(self): + present = {"Content-Type": "text/html"} + missing = check_security_headers(present, ["Content-Security-Policy", "X-Content-Type-Options"]) + assert "Content-Security-Policy" in missing + + def test_case_insensitive(self): + present = {"content-security-policy": "default-src 'self'"} + missing = check_security_headers(present, ["Content-Security-Policy"]) + assert missing == [] + + +class TestWhitelist: + def _page(self, ext_url): + return { + "links": {"a": [{"url": ext_url, "class": "external"}]}, + "comment_links": [], + "hidden_content": [], + } + + def test_flags_unlisted_domain(self): + page = self._page("https://spam.example.com/x") + violations = check_links_against_whitelist(page, set(), "b.info") + assert any(v["domain"] == "spam.example.com" for v in violations) + + def test_allows_listed_domain(self): + page = self._page("https://good.example.com/x") + violations = check_links_against_whitelist(page, {"good.example.com"}, "b.info") + assert violations == [] diff --git a/tests/test_crawler.py b/tests/test_crawler.py index fc6aaf3..56f13d4 100644 --- a/tests/test_crawler.py +++ b/tests/test_crawler.py @@ -32,6 +32,16 @@ class TestNormalizeUrl: assert "sid" not in url assert "id=5" in url + def test_strips_cache_buster_v(self): + url = normalize_url("https://example.com/?v=1781223996&id=5") + assert "v=1781223996" not in url + assert "id=5" in url + + def test_keeps_non_noise_v_prefixed_param(self): + # 'view' must NOT be dropped just because it starts with 'v' + url = normalize_url("https://example.com/?view=gallery") + assert "view=gallery" in url + def test_adds_default_path(self): result = normalize_url("https://example.com") assert result.endswith("/") or "example.com" in result diff --git a/tests/test_differ.py b/tests/test_differ.py index a8675b6..fc38e09 100644 --- a/tests/test_differ.py +++ b/tests/test_differ.py @@ -79,6 +79,15 @@ class TestDiffLinkSet: result = diff_link_set(links, links) assert result == {"added": [], "removed": []} + def test_truly_new_url_detected(self): + # Cache-buster stripping happens at extraction time; diff_link_set is a + # plain set comparison on already-normalised URLs. + old = [{"url": "https://example.com/style.css"}] + new = [{"url": "https://example.com/style.css"}, + {"url": "https://example.com/extra.js"}] + result = diff_link_set(old, new) + assert "https://example.com/extra.js" in result["added"] + class TestDiffMetadata: def test_detects_title_change(self): @@ -153,6 +162,59 @@ class TestCompareSnapshots: assert "spam.example.com" in result["new_external_domains"] +class TestUnexpectedJsonLd: + CFG = {**BASE_CFG, "allowed_jsonld_types": ["Organization", "WebPage"]} + + def _page(self, url, jsonld): + return { + "url": url, "status": 200, "text": "Inhalt", + "links": {"a": [], "script": [], "iframe": [], "link_rel": [], "form": [], "meta_refresh": [], "canonical": None}, + "metadata": {"title": "T", "h1": [], "h2": []}, + "hidden_content": [], "inline_scripts": [], "jsonld": jsonld, "comment_links": [], + } + + def test_unexpected_type_flagged_without_page_diff(self): + # Page is otherwise unchanged (same text/links) but carries a bad @type. + base = {"https://b.info/": self._page("https://b.info/", [{"type": "Organization"}])} + cur = {"https://b.info/": self._page("https://b.info/", [{"type": "Pharmacy"}])} + result = compare_snapshots(base, cur, self.CFG) + types = [e["type"] for e in result["unexpected_jsonld"]] + assert "Pharmacy" in types + + def test_allowed_type_not_flagged(self): + base = {"https://b.info/": self._page("https://b.info/", [])} + cur = {"https://b.info/": self._page("https://b.info/", [{"type": "WebPage"}])} + result = compare_snapshots(base, cur, self.CFG) + assert result["unexpected_jsonld"] == [] + + def test_preexisting_unexpected_type_not_realerted(self): + # Bad type already in baseline → approved → should not re-alert. + base = {"https://b.info/": self._page("https://b.info/", [{"type": "Pharmacy"}])} + cur = {"https://b.info/": self._page("https://b.info/", [{"type": "Pharmacy"}])} + result = compare_snapshots(base, cur, self.CFG) + assert result["unexpected_jsonld"] == [] + + def test_unexpected_type_scores(self): + result = score_diff({ + "new_internal_urls": [], "missing_internal_urls": [], + "new_external_domains": [], "page_diffs": [], "new_pages_analysis": [], + "unexpected_jsonld": [{"url": "https://b.info/", "type": "Pharmacy"}], + }, BASE_CFG) + assert result["score"] == 35 + assert result["level"] == "yellow" + + +class TestSuspiciousFilenames: + def test_suspicious_filename_scores_red(self): + result = score_diff({ + "new_internal_urls": [], "missing_internal_urls": [], + "new_external_domains": [], "page_diffs": [], "new_pages_analysis": [], + "suspicious_filenames": [{"page": "https://b.info/", "url": "https://b.info/shell.php"}], + }, BASE_CFG) + assert result["score"] == 60 + assert result["level"] == "red" + + class TestBrokenLinks: def _make_page(self, url, text="Inhalt", status=200): return { diff --git a/tests/test_extractor.py b/tests/test_extractor.py index 3d84a35..5b62bec 100644 --- a/tests/test_extractor.py +++ b/tests/test_extractor.py @@ -150,3 +150,32 @@ class TestJsonLd: ) result = _extract(html) assert any(j["type"] == "Pharmacy" for j in result["jsonld"]) + + def test_non_dict_jsonld_does_not_crash(self): + # Valid JSON that is not an object (scalar / array of scalars) must be + # tolerated, not raise AttributeError. + for body in ("[1, 2, 3]", '"just a string"', "42"): + html = f'' + result = _extract(html) + assert isinstance(result["jsonld"], list) + + +class TestCacheBusterNormalization: + def test_link_rel_cache_param_stripped(self): + html = '' + result = _extract(html) + urls = [l["url"] for l in result["links"]["link_rel"]] + assert any(u.endswith("/css/app.css") for u in urls) + assert all("v=" not in u for u in urls) + + def test_script_cache_param_stripped(self): + html = '' + result = _extract(html) + urls = [l["url"] for l in result["links"]["script"]] + assert any(u.endswith("/js/app.js") for u in urls) + + def test_non_cache_param_kept_on_asset(self): + html = '' + result = _extract(html) + urls = [l["url"] for l in result["links"]["img"]] + assert any("size=large" in u for u in urls)