From d803f79efec754a382c18191dea206e0ab8484f3 Mon Sep 17 00:00:00 2001 From: Your Name Date: Fri, 12 Jun 2026 09:26:59 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20Bin=C3=A4rdateien=20auf=20=C3=84nderung?= =?UTF-8?q?en=20pr=C3=BCfen=20(check-assets=20/=20approve-assets)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Neue Befehle: - `python -m scanner check-assets` — SHA-256-Hashes aller JS/CSS/Bild-Assets aus dem letzten Snapshot fetchen und mit Baseline vergleichen - `python -m scanner approve-assets` — aktuelle Hashes als neue Baseline setzen Scoring: geändertes JS 40 Pkt, CSS 30 Pkt, Bild/PDF 20 Pkt. Asset-Baseline liegt in data/baseline/asset_hashes.json. Co-Authored-By: Claude Sonnet 4.6 --- config.yaml | 3 + scanner/__main__.py | 220 ++++++++++++++++++++++++++++++++++++++++++- scanner/baseline.py | 22 +++++ scanner/checker.py | 43 +++++++++ scanner/config.py | 3 + scanner/differ.py | 72 ++++++++++++++ tests/test_assets.py | 144 ++++++++++++++++++++++++++++ 7 files changed, 506 insertions(+), 1 deletion(-) create mode 100644 tests/test_assets.py diff --git a/config.yaml b/config.yaml index ceff972..e234505 100644 --- a/config.yaml +++ b/config.yaml @@ -51,6 +51,9 @@ scoring: cloaking_extra_link: 60 # Link nur im Bot-Crawl sichtbar cloaking_extra_text: 40 # signifikant mehr Text im Bot-Crawl (>100 Zeichen) cloaking_vary_ua: 25 # Vary: User-Agent Header (ohne sonstigen Unterschied) + changed_script: 40 # JavaScript-Datei inhaltlich geändert + changed_stylesheet: 30 # CSS-Datei inhaltlich geändert + changed_asset: 20 # Bild / PDF / sonstige Binärdatei geändert thresholds: yellow: 20 # ab diesem Score: Warnung (gelb) diff --git a/scanner/__main__.py b/scanner/__main__.py index a3c49dc..4069988 100644 --- a/scanner/__main__.py +++ b/scanner/__main__.py @@ -27,12 +27,20 @@ from .checker import ( check_external_links, check_links_against_whitelist, check_security_headers, + fetch_asset_hashes, is_suspicious_url, load_whitelist, ) from .config import load_config, resolve_paths from .crawler import Crawler -from .differ import compare_cloaking, compare_snapshots, score_cloak_diff, score_diff +from .differ import ( + compare_asset_hashes, + compare_cloaking, + compare_snapshots, + score_asset_diff, + score_cloak_diff, + score_diff, +) from .extractor import extract_page @@ -459,6 +467,115 @@ def _print_ext_links_summary(report: dict) -> None: print(f" → {page}") +def _collect_asset_urls(snap: dict) -> dict[str, str]: + """Return {url: type} for all asset links across all pages in a snapshot.""" + assets: dict[str, str] = {} + for page in snap["pages"].values(): + if page.get("status") != 200: + continue + for ltype in ("script", "link_rel", "img"): + for link in page.get("links", {}).get(ltype, []): + if isinstance(link, dict): + url = link.get("url", "") + if url and url not in assets: + assets[url] = ltype + return assets + + +def _write_asset_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) / f"{ts}_assets" + 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_asset_markdown(report), encoding="utf-8") + return json_path, md_path + + +def _render_asset_markdown(r: dict) -> str: + a = r.get("assessment", {}) + d = r.get("asset_diff", {}) + level = a.get("level", "?").upper() + level_badge = {"GREEN": "[OK]", "YELLOW": "[WARNUNG]", "RED": "[ALARM]"}.get(level, level) + + lines = [ + f"# Asset-Check — {r.get('target', '?')}", + "", + f"**Erstellt:** {r.get('generated_at', '?')} ", + f"**Level:** {level_badge} ", + f"**Score:** {a.get('score', 0)}", + "", + "## Zusammenfassung", + "", + "| Kennzahl | Wert |", + "|---|---|", + f"| Geprüfte Assets | {d.get('total_checked', 0)} |", + f"| Geändert | {len(d.get('changed', []))} |", + f"| Neu (nicht in Baseline) | {len(d.get('new', []))} |", + f"| Fehlt (war in Baseline) | {len(d.get('missing', []))} |", + f"| Abruffehler | {len(d.get('fetch_errors', []))} |", + "", + "## Risikobewertung", + "", + ] + for reason in a.get("reasons", []): + lines.append(f"- {reason}") + if not a.get("reasons"): + lines.append("- Keine Änderungen in Binärdateien gefunden.") + + if d.get("changed"): + lines += ["", "## Geänderte Dateien", ""] + for entry in d["changed"]: + size_info = f" ({entry['size_diff']:+d} Bytes)" if entry.get("size_diff") else "" + lines.append(f"- **[{entry['type']}]** `{entry['url']}`{size_info}") + lines.append(f" - vorher: `{entry['old_sha256']}`") + lines.append(f" - jetzt: `{entry['new_sha256']}`") + + if d.get("new"): + lines += ["", "## Neue Assets (nicht in Baseline)", ""] + for url in d["new"]: + lines.append(f"- `{url}`") + + if d.get("missing"): + lines += ["", "## Fehlende Assets (waren in Baseline)", ""] + for url in d["missing"]: + lines.append(f"- `{url}`") + + if d.get("fetch_errors"): + lines += ["", "## Abruffehler", ""] + for entry in d["fetch_errors"]: + lines.append(f"- `{entry['url']}`: {entry['error']}") + + lines += ["", "---", "", "## Nächste Schritte", ""] + lev = a.get("level", "green") + if lev in ("red", "yellow"): + lines += [ + "1. Geänderte Dateien mit Backup vergleichen.", + "2. Bei unbekannten Änderungen: Hoster kontaktieren.", + "3. Nach Prüfung freigeben: `python -m scanner approve-assets`", + ] + else: + lines += ["Keine Aktion erforderlich."] + + return "\n".join(lines) + "\n" + + +def _print_asset_summary(assessment: dict, asset_diff: dict) -> None: + level = assessment.get("level", "?").upper() + score = assessment.get("score", 0) + print() + print(f"=== Asset-Check: {level} (Score {score}) ===") + for reason in assessment.get("reasons", []): + print(f" * {reason}") + if not assessment.get("reasons"): + print(" Keine Änderungen.") + print(f" Geprüfte Assets: {asset_diff.get('total_checked', 0)}") + print(f" Geändert: {len(asset_diff.get('changed', []))}") + print(f" Neu: {len(asset_diff.get('new', []))}") + print(f" Fehlt: {len(asset_diff.get('missing', []))}") + + def _latest_report_path(reports_dir: str) -> Path | None: rd = Path(reports_dir) if not rd.exists(): @@ -865,6 +982,100 @@ def cmd_check_ext_links(args: argparse.Namespace, cfg: dict) -> int: return 0 if not broken and not conn_errors else 1 +# --------------------------------------------------------------------------- +# Asset check +# --------------------------------------------------------------------------- + +def cmd_check_assets(args: argparse.Namespace, cfg: dict) -> int: + logger = logging.getLogger("scanner.assets") + bm = BaselineManager(cfg["data_dir"]) + + snap = bm.load_snapshot() + if not snap: + print( + "FEHLER: Kein Snapshot vorhanden. Zuerst `python -m scanner crawl` ausführen.", + file=sys.stderr, + ) + return 1 + + asset_urls = _collect_asset_urls(snap) + if not asset_urls: + print("Keine Assets im letzten Snapshot gefunden.") + return 0 + + print(f"Hashe {len(asset_urls)} Assets (script: {sum(1 for t in asset_urls.values() if t=='script')}, " + f"css: {sum(1 for t in asset_urls.values() if t=='link_rel')}, " + f"img: {sum(1 for t in asset_urls.values() if t=='img')}) ...") + + raw_hashes = fetch_asset_hashes( + list(asset_urls.keys()), + timeout=cfg.get("request_timeout", 15), + ) + # Attach link type to each result for scoring + current_hashes = { + url: {**r, "type": asset_urls[url]} + for url, r in raw_hashes.items() + } + + baseline_hashes = bm.load_asset_hashes() + asset_diff = compare_asset_hashes(baseline_hashes, current_hashes) + assessment = score_asset_diff(asset_diff, cfg) + + report = { + "generated_at": datetime.now(timezone.utc).isoformat(), + "target": cfg["target"], + "type": "asset-check", + "assessment": assessment, + "asset_diff": asset_diff, + } + json_path, md_path = _write_asset_report(report, cfg["reports_dir"]) + logger.info("Asset-Report: %s", md_path) + + _print_asset_summary(assessment, asset_diff) + + if not baseline_hashes: + print() + print("Noch keine Asset-Baseline vorhanden.") + print("Führen Sie aus: python -m scanner approve-assets") + + return assessment["exit_code"] + + +def cmd_approve_assets(args: argparse.Namespace, cfg: dict) -> int: + logger = logging.getLogger("scanner.approve-assets") + bm = BaselineManager(cfg["data_dir"]) + + snap = bm.load_snapshot() + if not snap: + print( + "FEHLER: Kein Snapshot vorhanden. Zuerst `python -m scanner crawl` ausführen.", + file=sys.stderr, + ) + return 1 + + asset_urls = _collect_asset_urls(snap) + if not asset_urls: + print("Keine Assets gefunden.") + return 0 + + print(f"Hashe {len(asset_urls)} Assets für neue Baseline ...") + raw_hashes = fetch_asset_hashes( + list(asset_urls.keys()), + timeout=cfg.get("request_timeout", 15), + ) + hashes_to_save = { + url: {**r, "type": asset_urls[url]} + for url, r in raw_hashes.items() + if r.get("sha256") # only store successfully fetched assets + } + bm.save_asset_hashes(hashes_to_save) + note = getattr(args, "note", "") or "" + print(f"Asset-Baseline gesetzt: {len(hashes_to_save)} Dateien.") + if note: + print(f"Notiz: {note}") + return 0 + + # --------------------------------------------------------------------------- # Output helper # --------------------------------------------------------------------------- @@ -922,6 +1133,11 @@ def _build_parser() -> argparse.ArgumentParser: sub.add_parser("check-ext-links", help="Externe Links auf Erreichbarkeit prüfen (4xx/5xx)") + sub.add_parser("check-assets", help="Binärdateien (JS/CSS/Bilder) auf Änderungen prüfen") + + aa = sub.add_parser("approve-assets", help="Aktuelle Asset-Hashes als neue Baseline setzen") + aa.add_argument("--note", default="", help="Freitext-Notiz") + return p @@ -949,6 +1165,8 @@ def main() -> None: "status": cmd_status, "cloak-check": cmd_cloak_check, "check-ext-links": cmd_check_ext_links, + "check-assets": cmd_check_assets, + "approve-assets": cmd_approve_assets, } handler = commands.get(args.command) if not handler: diff --git a/scanner/baseline.py b/scanner/baseline.py index e0514c4..1cc574b 100644 --- a/scanner/baseline.py +++ b/scanner/baseline.py @@ -211,6 +211,28 @@ class BaselineManager: self.pages_dir.mkdir(parents=True, exist_ok=True) return self.approve_all(snap_dir=snap_dir, note=note, approved_by=approved_by) + # ------------------------------------------------------------------ + # Asset hash baseline + # ------------------------------------------------------------------ + + @property + def _asset_hashes_path(self) -> Path: + return self.baseline_dir / "asset_hashes.json" + + def load_asset_hashes(self) -> dict: + """Load approved asset hashes. Returns {} if none approved yet.""" + if not self._asset_hashes_path.exists(): + return {} + return json.loads(self._asset_hashes_path.read_text(encoding="utf-8")) + + def save_asset_hashes(self, hashes: dict) -> None: + """Persist asset hashes as the new approved state.""" + self.baseline_dir.mkdir(parents=True, exist_ok=True) + self._asset_hashes_path.write_text( + json.dumps(hashes, indent=2, ensure_ascii=False), encoding="utf-8" + ) + logger.info("Asset hashes saved: %d entries", len(hashes)) + # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ diff --git a/scanner/checker.py b/scanner/checker.py index f4da9b8..ddc0129 100644 --- a/scanner/checker.py +++ b/scanner/checker.py @@ -147,6 +147,49 @@ def check_external_links( return results +# --------------------------------------------------------------------------- +# Asset hashing +# --------------------------------------------------------------------------- + +def fetch_asset_hashes( + urls: list[str], + timeout: int = 15, + delay: float = 0.2, +) -> dict[str, dict]: + """ + Fetch binary assets and compute SHA-256 hashes via streaming GET. + + Returns {url: {"sha256": str|None, "size": int|None, "error": str|None}}. + """ + import hashlib + results: dict[str, dict] = {} + session = requests.Session() + session.headers["User-Agent"] = "integrity-scanner/1.0 (asset-check)" + + try: + for i, url in enumerate(urls): + if i > 0 and delay > 0: + time.sleep(delay) + logger.debug("Hashing asset: %s", url) + try: + resp = session.get(url, timeout=timeout, stream=True) + if resp.status_code == 200: + h = hashlib.sha256() + size = 0 + for chunk in resp.iter_content(chunk_size=65536): + h.update(chunk) + size += len(chunk) + results[url] = {"sha256": h.hexdigest(), "size": size, "error": None} + else: + results[url] = {"sha256": None, "size": None, "error": f"HTTP {resp.status_code}"} + except requests.exceptions.RequestException as exc: + results[url] = {"sha256": None, "size": None, "error": str(exc)} + finally: + session.close() + + return results + + def is_suspicious_url(url: str) -> bool: """Flag URLs with names typical for webshells / backdoors.""" filename = urlparse(url).path.rsplit("/", 1)[-1] diff --git a/scanner/config.py b/scanner/config.py index 3adcb87..3e87ba2 100644 --- a/scanner/config.py +++ b/scanner/config.py @@ -36,6 +36,9 @@ DEFAULT_CONFIG: dict = { "cloaking_extra_link": 60, "cloaking_extra_text": 40, "cloaking_vary_ua": 25, + "changed_script": 40, + "changed_stylesheet": 30, + "changed_asset": 20, }, "thresholds": { "yellow": 20, diff --git a/scanner/differ.py b/scanner/differ.py index b2afa97..8814d5b 100644 --- a/scanner/differ.py +++ b/scanner/differ.py @@ -467,6 +467,78 @@ def score_cloak_diff(cloak_diff: dict, cfg: dict) -> dict: return {"score": score, "level": level, "reasons": reasons, "exit_code": exit_code} +# --------------------------------------------------------------------------- +# Asset hash comparison +# --------------------------------------------------------------------------- + +def compare_asset_hashes(baseline: dict, current: dict) -> dict: + """ + Compare asset hash dicts {url: {sha256, size, type, error}}. + Returns {changed, new, missing, fetch_errors, total_checked}. + """ + b_hashed = {url for url, r in baseline.items() if r.get("sha256")} + c_hashed = {url for url, r in current.items() if r.get("sha256")} + + changed = [ + { + "url": url, + "type": current[url].get("type", "unknown"), + "old_sha256": baseline[url]["sha256"][:16] + "…", + "new_sha256": current[url]["sha256"][:16] + "…", + "size_diff": (current[url].get("size") or 0) - (baseline[url].get("size") or 0), + } + for url in sorted(b_hashed & c_hashed) + if baseline[url]["sha256"] != current[url]["sha256"] + ] + + return { + "changed": changed, + "new": sorted(c_hashed - b_hashed), + "missing": sorted(b_hashed - set(current)), + "fetch_errors": [ + {"url": url, "error": r["error"]} + for url, r in sorted(current.items()) + if r.get("error") + ], + "total_checked": len(c_hashed), + } + + +def score_asset_diff(asset_diff: dict, cfg: dict) -> dict: + """Score asset hash changes by file type.""" + sc = cfg.get("scoring", {}) + thr = cfg.get("thresholds", {"yellow": 20, "red": 60}) + score = 0 + reasons: list[str] = [] + + def add(pts: int, msg: str) -> None: + nonlocal score + score += pts + reasons.append(f"{msg} (+{pts})") + + for entry in asset_diff.get("changed", []): + asset_type = entry.get("type", "unknown") + if asset_type == "script": + pts = sc.get("changed_script", 40) + elif asset_type == "link_rel": + pts = sc.get("changed_stylesheet", 30) + else: + pts = sc.get("changed_asset", 20) + size_info = f", {entry['size_diff']:+d} Bytes" if entry.get("size_diff") else "" + add(pts, f"Geänderte Datei [{asset_type}]{size_info}: {entry['url']}") + + yellow = thr.get("yellow", 20) + red = thr.get("red", 60) + if score >= red: + level, exit_code = "red", 2 + elif score >= yellow: + level, exit_code = "yellow", 1 + else: + level, exit_code = "green", 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 diff --git a/tests/test_assets.py b/tests/test_assets.py new file mode 100644 index 0000000..8585748 --- /dev/null +++ b/tests/test_assets.py @@ -0,0 +1,144 @@ +"""Tests for asset hashing and comparison.""" +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from scanner.checker import fetch_asset_hashes +from scanner.differ import compare_asset_hashes, score_asset_diff + +CFG = {"scoring": {}, "thresholds": {}} + + +# --------------------------------------------------------------------------- +# fetch_asset_hashes +# --------------------------------------------------------------------------- + +class TestFetchAssetHashes: + def _mock_session(self, status=200, content=b"data"): + resp = MagicMock() + resp.status_code = status + resp.iter_content.return_value = [content] + resp.close = MagicMock() + inst = MagicMock() + inst.get.return_value = resp + inst.close = MagicMock() + return inst + + def test_successful_fetch_returns_sha256(self): + import hashlib + data = b"hello asset" + expected = hashlib.sha256(data).hexdigest() + + with patch("scanner.checker.requests.Session") as MockSession: + MockSession.return_value = self._mock_session(content=data) + results = fetch_asset_hashes(["https://x.de/app.js"], delay=0) + + assert results["https://x.de/app.js"]["sha256"] == expected + assert results["https://x.de/app.js"]["error"] is None + + def test_http_error_recorded(self): + with patch("scanner.checker.requests.Session") as MockSession: + MockSession.return_value = self._mock_session(status=404) + results = fetch_asset_hashes(["https://x.de/missing.js"], delay=0) + + assert results["https://x.de/missing.js"]["sha256"] is None + assert "HTTP 404" in results["https://x.de/missing.js"]["error"] + + def test_connection_error_recorded(self): + with patch("scanner.checker.requests.Session") as MockSession: + inst = MockSession.return_value + inst.get.side_effect = requests.exceptions.ConnectionError("refused") + inst.close = MagicMock() + results = fetch_asset_hashes(["https://unreachable.de/x.js"], delay=0) + + assert results["https://unreachable.de/x.js"]["sha256"] is None + assert results["https://unreachable.de/x.js"]["error"] is not None + + def test_empty_list_returns_empty(self): + with patch("scanner.checker.requests.Session"): + results = fetch_asset_hashes([], delay=0) + assert results == {} + + +# --------------------------------------------------------------------------- +# compare_asset_hashes +# --------------------------------------------------------------------------- + +class TestCompareAssetHashes: + def _h(self, sha, type_="img", size=100): + return {"sha256": sha, "size": size, "type": type_, "error": None} + + def test_no_changes_empty_diff(self): + h = {"https://x.de/img.jpg": self._h("aaa")} + result = compare_asset_hashes(h, h) + assert result["changed"] == [] + assert result["new"] == [] + assert result["missing"] == [] + + def test_changed_hash_detected(self): + baseline = {"https://x.de/img.jpg": self._h("aaa111")} + current = {"https://x.de/img.jpg": self._h("bbb222")} + result = compare_asset_hashes(baseline, current) + assert len(result["changed"]) == 1 + assert result["changed"][0]["url"] == "https://x.de/img.jpg" + + def test_new_asset_detected(self): + baseline = {} + current = {"https://x.de/new.jpg": self._h("abc")} + result = compare_asset_hashes(baseline, current) + assert "https://x.de/new.jpg" in result["new"] + + def test_missing_asset_detected(self): + baseline = {"https://x.de/old.jpg": self._h("abc")} + current = {} + result = compare_asset_hashes(baseline, current) + assert "https://x.de/old.jpg" in result["missing"] + + def test_fetch_error_not_in_changed(self): + baseline = {"https://x.de/img.jpg": self._h("aaa")} + current = {"https://x.de/img.jpg": {"sha256": None, "size": None, "type": "img", "error": "timeout"}} + result = compare_asset_hashes(baseline, current) + assert result["changed"] == [] + assert len(result["fetch_errors"]) == 1 + + +# --------------------------------------------------------------------------- +# score_asset_diff +# --------------------------------------------------------------------------- + +class TestScoreAssetDiff: + def test_no_changes_green(self): + result = score_asset_diff({"changed": [], "new": [], "missing": [], "fetch_errors": [], "total_checked": 5}, CFG) + assert result["level"] == "green" + assert result["score"] == 0 + + def test_changed_script_scores_40(self): + diff = {"changed": [{"url": "https://x.de/app.js", "type": "script", "old_sha256": "a…", "new_sha256": "b…", "size_diff": 0}], + "new": [], "missing": [], "fetch_errors": [], "total_checked": 1} + result = score_asset_diff(diff, CFG) + assert result["score"] == 40 + assert result["level"] == "yellow" + + def test_changed_stylesheet_scores_30(self): + diff = {"changed": [{"url": "https://x.de/style.css", "type": "link_rel", "old_sha256": "a…", "new_sha256": "b…", "size_diff": 0}], + "new": [], "missing": [], "fetch_errors": [], "total_checked": 1} + result = score_asset_diff(diff, CFG) + assert result["score"] == 30 + assert result["level"] == "yellow" + + def test_changed_image_scores_20(self): + diff = {"changed": [{"url": "https://x.de/logo.png", "type": "img", "old_sha256": "a…", "new_sha256": "b…", "size_diff": 100}], + "new": [], "missing": [], "fetch_errors": [], "total_checked": 1} + result = score_asset_diff(diff, CFG) + assert result["score"] == 20 + assert result["level"] == "yellow" + + def test_two_changed_scripts_red(self): + entry = {"url": "https://x.de/x.js", "type": "script", "old_sha256": "a…", "new_sha256": "b…", "size_diff": 0} + diff = {"changed": [entry, {**entry, "url": "https://x.de/y.js"}], + "new": [], "missing": [], "fetch_errors": [], "total_checked": 2} + result = score_asset_diff(diff, CFG) + assert result["score"] == 80 + assert result["level"] == "red" + assert result["exit_code"] == 2