From ecd01abf98a4dba8ce0aee55b47a7caa8d215f63 Mon Sep 17 00:00:00 2001 From: Your Name Date: Fri, 12 Jun 2026 04:14:50 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20externe=20Links=20auf=20Erreichbarkeit?= =?UTF-8?q?=20pr=C3=BCfen=20(check-ext-links)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Neuer Befehl `python -m scanner check-ext-links`: - Liest alle externen -Links aus dem letzten Snapshot - Prüft jeden per HEAD-Request (Fallback auf GET bei HTTP 405) - Meldet 4xx/5xx und Verbindungsfehler mit Quelladressen - Report unter reports/_ext-links/ - Exit-Code 1 bei kaputten Links, 0 wenn alle erreichbar Co-Authored-By: Claude Sonnet 4.6 --- scanner/__main__.py | 138 ++++++++++++++++++++++++++++++++++++++++ scanner/checker.py | 44 ++++++++++++- tests/test_ext_links.py | 82 ++++++++++++++++++++++++ 3 files changed, 263 insertions(+), 1 deletion(-) create mode 100644 tests/test_ext_links.py diff --git a/scanner/__main__.py b/scanner/__main__.py index c07f8d3..a3c49dc 100644 --- a/scanner/__main__.py +++ b/scanner/__main__.py @@ -24,6 +24,7 @@ from urllib.parse import urlparse from .alerter import send_alert from .baseline import BaselineManager from .checker import ( + check_external_links, check_links_against_whitelist, check_security_headers, is_suspicious_url, @@ -390,6 +391,74 @@ def _print_cloak_summary(assessment: dict, cloak_diff: dict) -> None: print(f" Seiten mit Unterschied: {len(cloak_diff.get('findings', []))}") +def _write_ext_links_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}_ext-links" + 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_ext_links_markdown(report), encoding="utf-8") + return json_path, md_path + + +def _render_ext_links_markdown(r: dict) -> str: + lines = [ + f"# Externe Links — {r.get('target', '?')}", + "", + f"**Erstellt:** {r.get('generated_at', '?')}", + "", + "## Zusammenfassung", + "", + "| Kennzahl | Wert |", + "|---|---|", + f"| Geprüfte Links | {r.get('total', 0)} |", + f"| Erreichbar (2xx/3xx) | {r.get('ok', 0)} |", + f"| Kaputt (4xx/5xx) | {r.get('broken', 0)} |", + f"| Nicht erreichbar (Fehler) | {r.get('errors', 0)} |", + "", + ] + + broken = r.get("broken_links", []) + if broken: + lines += ["## Kaputte externe Links", ""] + for entry in broken: + lines.append(f"### HTTP {entry['status']} — `{entry['url']}`") + lines.append("") + lines.append("Gefunden auf:") + for page in entry["pages"]: + lines.append(f"- {page}") + lines.append("") + + errors = r.get("error_links", []) + if errors: + lines += ["## Nicht erreichbar (Verbindungsfehler)", ""] + for entry in errors: + lines.append(f"- `{entry['url']}` — {entry['error']}") + for page in entry["pages"]: + lines.append(f" - verlinkt von: {page}") + lines.append("") + + if not broken and not errors: + lines += ["## Ergebnis", "", "Alle externen Links sind erreichbar."] + + return "\n".join(lines) + "\n" + + +def _print_ext_links_summary(report: dict) -> None: + print() + print(f"=== Externe Links: {report['total']} geprüft ===") + print(f" Erreichbar: {report['ok']}") + print(f" Kaputt (4xx/5xx): {report['broken']}") + print(f" Verbindungsfehler: {report['errors']}") + if report.get("broken_links"): + print() + for entry in report["broken_links"]: + print(f" [HTTP {entry['status']}] {entry['url']}") + for page in entry["pages"][:3]: + print(f" → {page}") + + def _latest_report_path(reports_dir: str) -> Path | None: rd = Path(reports_dir) if not rd.exists(): @@ -730,6 +799,72 @@ def cmd_cloak_check(args: argparse.Namespace, cfg: dict) -> int: return assessment["exit_code"] +# --------------------------------------------------------------------------- +# External link check +# --------------------------------------------------------------------------- + +def cmd_check_ext_links(args: argparse.Namespace, cfg: dict) -> int: + logger = logging.getLogger("scanner.ext-links") + 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 + + # Collect unique external links → pages that link to them + ext_urls: dict[str, list[str]] = {} + for page in snap["pages"].values(): + if page.get("status") != 200: + continue + for link in page.get("links", {}).get("a", []): + if isinstance(link, dict) and link.get("class") == "external": + url = link.get("url", "") + if url: + ext_urls.setdefault(url, []).append(page["url"]) + + if not ext_urls: + print("Keine externen Links im letzten Snapshot gefunden.") + return 0 + + print(f"Prüfe {len(ext_urls)} externe Links ...") + results = check_external_links( + list(ext_urls.keys()), + timeout=cfg.get("request_timeout", 10), + ) + + broken = {url: r for url, r in results.items() if r.get("status") and r["status"] >= 400} + conn_errors = {url: r for url, r in results.items() if r.get("error")} + ok_count = len(results) - len(broken) - len(conn_errors) + + report = { + "generated_at": datetime.now(timezone.utc).isoformat(), + "target": cfg["target"], + "type": "ext-links", + "total": len(results), + "ok": ok_count, + "broken": len(broken), + "errors": len(conn_errors), + "broken_links": [ + {"url": url, "status": r["status"], "pages": ext_urls[url]} + for url, r in sorted(broken.items()) + ], + "error_links": [ + {"url": url, "error": r["error"], "pages": ext_urls[url]} + for url, r in sorted(conn_errors.items()) + ], + } + + json_path, md_path = _write_ext_links_report(report, cfg["reports_dir"]) + logger.info("Ext-Links-Report: %s", md_path) + + _print_ext_links_summary(report) + return 0 if not broken and not conn_errors else 1 + + # --------------------------------------------------------------------------- # Output helper # --------------------------------------------------------------------------- @@ -785,6 +920,8 @@ def _build_parser() -> argparse.ArgumentParser: sub.add_parser("cloak-check", help="Doppel-Crawl: Browser-UA vs. Googlebot-UA (Cloaking-Erkennung)") + sub.add_parser("check-ext-links", help="Externe Links auf Erreichbarkeit prüfen (4xx/5xx)") + return p @@ -811,6 +948,7 @@ def main() -> None: "report": cmd_report, "status": cmd_status, "cloak-check": cmd_cloak_check, + "check-ext-links": cmd_check_ext_links, } handler = commands.get(args.command) if not handler: diff --git a/scanner/checker.py b/scanner/checker.py index 700de67..f4da9b8 100644 --- a/scanner/checker.py +++ b/scanner/checker.py @@ -2,15 +2,20 @@ Additional integrity checks beyond baseline diff: - link whitelist enforcement - security-header audit -- canonical hijack detection - suspicious URL filenames (from v4 heritage) +- external link reachability check """ +import logging import re +import time from pathlib import Path from urllib.parse import urlparse +import requests import yaml +logger = logging.getLogger(__name__) + # 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. @@ -105,6 +110,43 @@ def check_security_headers(response_headers: dict, required: list[str]) -> list[ # Suspicious filenames # --------------------------------------------------------------------------- +# --------------------------------------------------------------------------- +# External link reachability +# --------------------------------------------------------------------------- + +def check_external_links( + urls: list[str], + timeout: int = 10, + delay: float = 0.3, +) -> dict[str, dict]: + """ + Check a list of URLs for reachability via HEAD (fallback: GET). + + Returns {url: {"status": int|None, "final_url": str|None, "error": str|None}}. + """ + results: dict[str, dict] = {} + session = requests.Session() + session.headers["User-Agent"] = "integrity-scanner/1.0 (external-link-check)" + + try: + for i, url in enumerate(urls): + if i > 0 and delay > 0: + time.sleep(delay) + logger.debug("Checking external link: %s", url) + try: + resp = session.head(url, timeout=timeout, allow_redirects=True) + if resp.status_code == 405: + resp = session.get(url, timeout=timeout, allow_redirects=True, stream=True) + resp.close() + results[url] = {"status": resp.status_code, "final_url": resp.url, "error": None} + except requests.exceptions.RequestException as exc: + results[url] = {"status": None, "final_url": 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/tests/test_ext_links.py b/tests/test_ext_links.py new file mode 100644 index 0000000..9730f62 --- /dev/null +++ b/tests/test_ext_links.py @@ -0,0 +1,82 @@ +"""Tests for external link reachability check.""" +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from scanner.checker import check_external_links + + +def _mock_response(status_code=200, url="https://example.com/"): + resp = MagicMock() + resp.status_code = status_code + resp.url = url + return resp + + +class TestCheckExternalLinks: + def test_reachable_link_recorded(self): + with patch("scanner.checker.requests.Session") as MockSession: + session = MockSession.return_value.__enter__.return_value + MockSession.return_value = MagicMock() + inst = MockSession.return_value + inst.head.return_value = _mock_response(200) + inst.close = MagicMock() + + results = check_external_links(["https://example.com/"], delay=0) + + assert results["https://example.com/"]["status"] == 200 + assert results["https://example.com/"]["error"] is None + + def test_broken_link_404_recorded(self): + with patch("scanner.checker.requests.Session") as MockSession: + inst = MockSession.return_value + inst.head.return_value = _mock_response(404, "https://example.com/gone") + inst.close = MagicMock() + + results = check_external_links(["https://example.com/gone"], delay=0) + + assert results["https://example.com/gone"]["status"] == 404 + + def test_connection_error_recorded(self): + with patch("scanner.checker.requests.Session") as MockSession: + inst = MockSession.return_value + inst.head.side_effect = requests.exceptions.ConnectionError("refused") + inst.close = MagicMock() + + results = check_external_links(["https://unreachable.example.com/"], delay=0) + + r = results["https://unreachable.example.com/"] + assert r["status"] is None + assert "refused" in r["error"] + + def test_head_405_falls_back_to_get(self): + with patch("scanner.checker.requests.Session") as MockSession: + inst = MockSession.return_value + inst.head.return_value = _mock_response(405) + get_resp = _mock_response(200) + get_resp.close = MagicMock() + inst.get.return_value = get_resp + inst.close = MagicMock() + + results = check_external_links(["https://example.com/no-head"], delay=0) + + inst.get.assert_called_once() + assert results["https://example.com/no-head"]["status"] == 200 + + def test_multiple_urls_all_checked(self): + urls = [f"https://example.com/{i}" for i in range(3)] + with patch("scanner.checker.requests.Session") as MockSession: + inst = MockSession.return_value + inst.head.return_value = _mock_response(200) + inst.close = MagicMock() + + results = check_external_links(urls, delay=0) + + assert len(results) == 3 + assert inst.head.call_count == 3 + + def test_empty_list_returns_empty(self): + with patch("scanner.checker.requests.Session"): + results = check_external_links([], delay=0) + assert results == {}