dschlueter/integrity_scanner_fuer_statische_Webseiten
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

feat: Binärdateien auf Änderungen prüfen (check-assets / approve-assets)

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Your Name 2026-06-12 09:26:59 +02:00
commit d803f79efe
7 changed files with 506 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

220
scanner/__main__.py
View file

@ -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:

22
scanner/baseline.py
View file

@ -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
# ------------------------------------------------------------------

43
scanner/checker.py
View file

@ -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]

3
scanner/config.py
View file

@ -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,

72
scanner/differ.py
View file

@ -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