feat: externe Links auf Erreichbarkeit prüfen (check-ext-links)
Neuer Befehl `python -m scanner check-ext-links`: - Liest alle externen <a>-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/<ts>_ext-links/ - Exit-Code 1 bei kaputten Links, 0 wenn alle erreichbar Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Your Name
parent
0302d469b1
commit
ecd01abf98
3 changed files with 263 additions and 1 deletions
|
|
@ -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 <a> 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:
|
||||
|
|
|
|||
|
|
@ -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]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue