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

feat: KI-Absicherung gegen Falsch-Negative bei Dienstausfall

Browse source 
Bisher: Konnte die KI kein Verdikt liefern (alle Modellstufen scheitern),
ging der Inhalt still als grün durch — für ein Sicherheitswerkzeug ein
Falsch-Negativ-Risiko ("konnte nicht prüfen" ≠ "sauber").

Zwei Maßnahmen:
1. Retry: _classify wiederholt die ganze Modell-Kette bei Komplettausfall
   bis max_retries (Default 1) mit Backoff — fängt transiente Aussetzer ab.
2. Tracking: nicht prüfbare Inhalte landen in ai_result["unchecked"] (statt
   stillem skip) und werden sichtbar in Terminal, Report (Markdown), diff-only
   und E-Mail — jeweils mit URL. score_ai_findings nimmt sie als Grund auf;
   mit unchecked_level="yellow" heben sie das Level auf mindestens Gelb an
   (Default "warn" = nur Hinweis, Ampel unberührt).

Config: max_retries, retry_backoff_seconds, unchecked_level (ai_analysis).
Tests: 230 grün (+ Retry-Recovery, unchecked-Tracking, unchecked_level).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Dieter Schlüter 2026-06-13 00:50:36 +02:00
commit d06e3d3dfd
9 changed files with 138 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

17
BEDIENUNGSANLEITUNG.md
View file

@ -879,6 +879,23 @@ mittlerer Schwere werden gewertet — das hält Fehlalarme niedrig.
Fehlt der API-Key oder schlägt eine Abfrage fehl, wird die Analyse **übersprungen** und der
Scan läuft unverändert weiter. Die KI kann den normalen Betrieb nie blockieren.
### „Nicht geprüft" ≠ „sauber"
Kann ein Inhalt trotz Modell-Eskalation **und** Wiederholung (`max_retries`) nicht geprüft
werden, weil der Dienst gerade nicht erreichbar ist, wird er **nicht** still als unbedenklich
gewertet. Stattdessen erscheint er als **ungeprüft** im Terminal, im Report und in der E-Mail —
mit URL und dem Hinweis, beim nächsten Lauf erneut zu prüfen. Über `unchecked_level` steuern
Sie die Schärfe:
```yaml
ai_analysis:
max_retries: 1 # Wiederholungen der Kette bei Komplettausfall
unchecked_level: "warn" # "warn" = nur Hinweis | "yellow" = Level auf mindestens Gelb anheben
```
Für hohe Sicherheitsansprüche empfiehlt sich `"yellow"`: Dann fällt jeder nicht prüfbare
Inhalt sofort als gelbe Warnung auf, statt nur als Notiz zu erscheinen.
### Aktivieren
```yaml

5
README.md
View file

@ -237,6 +237,11 @@ KI-Funde sind **auf Gelb gedeckelt** — sie lösen nie allein ROT aus (das blei
Integritäts-Signalen vorbehalten). Fehlt der API-Key oder schlägt eine Abfrage fehl, wird
die Analyse übersprungen und der Scan läuft unverändert weiter.
**„Nicht geprüft" ≠ „sauber":** Kann ein Inhalt trotz Modell-Eskalation und Wiederholung
(`max_retries`) nicht geprüft werden, wird er nicht still als unbedenklich gewertet, sondern
als **ungeprüft** mit URL in Terminal/Report/E-Mail ausgewiesen. Mit `unchecked_level: "yellow"`
hebt ein nicht prüfbarer Inhalt das Level auf mindestens Gelb an (Default `"warn"` = nur Hinweis).
### Aktivieren
In `config.yaml`:

6
meine-seite.de/config.yaml
View file

@ -141,6 +141,12 @@ ai_analysis:
site_context: ""
ai_confidence_min: 0.7
attempt_timeout: 30 # Zeitlimit je Modell-Versuch (Sek.) → Auslöser der Eskalation
max_retries: 1 # Wiederholungen der ganzen Modell-Kette bei Komplettausfall
retry_backoff_seconds: 2.0
# Verhalten, wenn Inhalte trotz Retry NICHT geprüft werden konnten (KI nicht erreichbar):
# "warn" = sichtbarer Hinweis (Level bleibt unberührt)
# "yellow" = zusätzlich Gesamt-Level auf mindestens Gelb anheben (für hohe Sicherheit)
unchecked_level: "warn"
# Modell-Kette mit zweifacher Eskalation: Stufe 1+2 free, Stufe 3 günstig bezahlt.
# Schlägt ein Modell fehl oder antwortet zu langsam, wird zur nächsten Stufe eskaliert.
text:

11
scanner/__main__.py
View file

@ -414,6 +414,12 @@ def _render_markdown(r: dict) -> str:
f"- {ai.get('checked', 0)} Inhalt(e) geprüft, "
f"{ai.get('cache_hits', 0)} aus Cache, {ai.get('api_calls', 0)} API-Abfrage(n)."
)
if ai.get("unchecked"):
lines.append(f"- ⚠ {len(ai['unchecked'])} Inhalt(e) **NICHT geprüft** "
"(KI-Dienst nicht erreichbar) — „nicht geprüft“ ≠ „sauber“:")
for u in ai["unchecked"][:20]:
ziel = u.get("asset_url") or u.get("url", "?")
lines.append(f" - [{u.get('kind', '?')}] `{ziel}`")
for f in ai.get("findings", []):
asset = f.get("asset_url")
ziel = f"`{asset}` (auf {f['url']})" if asset else f"`{f['url']}`"
@ -1215,6 +1221,10 @@ def _print_diff_only(report: dict) -> None:
f"({f['confidence']:.0%}): {ziel}{C['rst']}")
if f.get("explanation"):
print(f" {f['explanation']}")
ai_unchecked = ai.get("unchecked", [])
for u in ai_unchecked:
ziel = u.get("asset_url") or u.get("url", "?")
print(f"{C['hdr']}? KI nicht geprüft [{u.get('kind', '?')}]: {ziel}{C['rst']}")
nothing = (
not d.get("new_external_domains")
@ -1223,6 +1233,7 @@ def _print_diff_only(report: dict) -> None:
and not d.get("new_broken_urls")
and not has_page_diffs
and not ai_findings
and not ai_unchecked
)
if nothing:
print("Keine Änderungen.")

51
scanner/ai_analyzer.py
View file

@ -17,6 +17,7 @@ KI darf den Kern-Scan nie brechen.
import json
import logging
import os
import time
from datetime import datetime, timezone
import requests
@ -79,7 +80,8 @@ def run_ai_analysis(cfg: dict, bm, snap: dict, diff: dict) -> dict:
"api_calls": int, "skipped": str|None}.
Bricht nie mit einer Exception — bei Problemen wird "skipped" gesetzt.
"""
result = {"findings": [], "checked": 0, "cache_hits": 0, "api_calls": 0, "skipped": None}
result = {"findings": [], "unchecked": [], "checked": 0, "cache_hits": 0,
"api_calls": 0, "skipped": None}
ai_cfg = cfg.get("ai_analysis", {})
if not ai_cfg.get("enabled"):
@ -139,6 +141,16 @@ def score_ai_findings(ai_result: dict, cfg: dict) -> dict:
else:
level, exit_code = "green", 0
# Ungeprüfte Inhalte: "konnte nicht prüfen" ≠ "sauber".
unchecked = ai_result.get("unchecked", [])
if unchecked:
reasons.append(
f"{len(unchecked)} Inhalt(e) konnten von der KI nicht geprüft werden "
"(Dienst nicht erreichbar) — bitte später erneut prüfen."
)
if cfg.get("ai_analysis", {}).get("unchecked_level") == "yellow" and level == "green":
level, exit_code = "yellow", 1
return {"score": score, "level": level, "reasons": reasons, "exit_code": exit_code}
@ -175,7 +187,8 @@ def _analyze_text(cfg, ai_cfg, api_key, snap, diff, entries, result) -> bool:
continue # Kostendeckel erreicht — nächster Scan holt den Rest nach
classified = _classify_text(text, site_context, models, ai_cfg, api_key)
if classified is None:
result["skipped"] = result["skipped"] or "API-Fehler (Text)"
# KI nicht erreichbar → Inhalt bleibt UNGEPRÜFT (nicht still als clean werten).
result["unchecked"].append({"kind": "text", "url": url})
continue
verdict, used_model = classified
entry = _make_entry("text", url, verdict, used_model)
@ -239,7 +252,8 @@ def _analyze_images(cfg, ai_cfg, api_key, snap, diff, entries, result) -> bool:
if entry is None:
classified = _classify_image(url, site_context, models, ai_cfg, api_key)
if classified is None:
result["skipped"] = result["skipped"] or "API-Fehler (Bild)"
# KI nicht erreichbar → Bild bleibt UNGEPRÜFT (nicht still als clean werten).
result["unchecked"].append({"kind": "image", "url": page_url, "asset_url": url})
continue
verdict, used_model = classified
entry = _make_entry("image", url, verdict, used_model)
@ -383,17 +397,28 @@ def _models_for(modality_cfg: dict) -> list[str]:
def _classify(messages, models, ai_cfg, api_key) -> tuple[dict, str] | None:
"""Versucht die Modell-Kette der Reihe nach (zweifache Eskalation).
Erste gültige Antwort → (verdict, model_used). Alle gescheitert → None.
Eskaliert auch bei Langsamkeit (Zeitlimit attempt_timeout je Versuch)."""
Eskaliert auch bei Langsamkeit (Zeitlimit attempt_timeout je Versuch).
Bei Komplettausfall der ganzen Kette wird bis max_retries wiederholt
(mit Backoff) — fängt transiente Aussetzer ab, bevor Inhalt ungeprüft bleibt."""
timeout = ai_cfg.get("attempt_timeout", 30)
for i, model in enumerate(models):
verdict = _openrouter_chat(model, messages, ai_cfg, api_key, timeout=timeout)
if verdict is not None:
if i > 0:
logger.info("KI-Modell-Eskalation: Stufe %d (%s) lieferte das Verdikt.", i + 1, model)
return verdict, model
if i + 1 < len(models):
logger.info("KI-Modell Stufe %d (%s) erfolglos — eskaliere zu Stufe %d.",
i + 1, model, i + 2)
max_retries = ai_cfg.get("max_retries", 1)
backoff = ai_cfg.get("retry_backoff_seconds", 2.0)
for attempt in range(max_retries + 1):
for i, model in enumerate(models):
verdict = _openrouter_chat(model, messages, ai_cfg, api_key, timeout=timeout)
if verdict is not None:
if i > 0 or attempt > 0:
logger.info("KI-Verdikt von Stufe %d (%s)%s.", i + 1, model,
f" nach Wiederholung {attempt}" if attempt else "")
return verdict, model
if i + 1 < len(models):
logger.info("KI-Modell Stufe %d (%s) erfolglos — eskaliere zu Stufe %d.",
i + 1, model, i + 2)
if attempt < max_retries:
logger.info("KI-Kette komplett erfolglos — Wiederholung %d/%d in %.1fs.",
attempt + 1, max_retries, backoff)
time.sleep(backoff)
return None

7
scanner/alerter.py
View file

@ -124,6 +124,13 @@ def _format_body(report: dict) -> str:
lines.append(f" Begründung: {f['explanation']}")
if f.get("fingerprint"):
lines.append(f" Quittieren: python -m scanner ai-dismiss --hash {f['fingerprint']}")
if ai.get("unchecked"):
lines += ["", "--- KI-Inhaltsanalyse: NICHT geprüfte Inhalte ---",
" Achtung: Der Prüfdienst war nicht erreichbar. „Nicht geprüft“ ≠ „sauber“ —"
" bitte beim nächsten Lauf erneut prüfen."]
for u in ai["unchecked"]:
ziel = u.get("asset_url") or u.get("url", "?")
lines.append(f" [{u.get('kind', '?')}] {ziel}")
lines += [
"",

7
scanner/config.py
View file

@ -89,6 +89,13 @@ DEFAULT_CONFIG: dict = {
"ai_confidence_min": 0.7, # Mindest-Konfidenz, damit ein Fund gewertet wird
"request_timeout": 60, # Ceiling (auch für Asset-Fetch der Bilder)
"attempt_timeout": 30, # Zeitlimit je Modell-Versuch → Eskalation bei Langsamkeit
"max_retries": 1, # Wiederholungen der ganzen Modell-Kette bei Komplettausfall
"retry_backoff_seconds": 2.0,
# Verhalten, wenn Inhalte trotz Retry NICHT geprüft werden konnten (KI nicht erreichbar):
# "warn" = sichtbarer Hinweis in Report/Terminal/E-Mail (Level bleibt unberührt)
# "yellow" = zusätzlich Gesamt-Level auf mindestens Gelb anheben
# Für ein Sicherheitswerkzeug gilt: "konnte nicht prüfen" ≠ "sauber".
"unchecked_level": "warn",
# Modell-Kette mit zweifacher Eskalation: Stufe 1+2 free, Stufe 3 günstig bezahlt.
# Schlägt ein Modell fehl oder antwortet zu langsam, wird zur nächsten Stufe eskaliert.
"text": {

7
scanner/plain.py
View file

@ -191,6 +191,13 @@ def klartext_befunde(
f"Auf {len(bild_funde)} Bild(ern) wurden möglicherweise problematische "
"Inhalte erkannt. Bitte prüfen Sie die betroffenen Bilder."
)
if ai_result and ai_result.get("unchecked"):
n = len(ai_result["unchecked"])
saetze.append(
f"{n} Inhalt(e) konnten nicht automatisch geprüft werden, weil der "
"Prüfdienst gerade nicht erreichbar war. „Nicht geprüft“ bedeutet nicht "
"„unbedenklich“ — bitte beim nächsten Lauf erneut prüfen."
)
if not saetze:
saetze.append("Es hat sich nichts Verdächtiges verändert.")

45
tests/test_ai_analyzer.py
View file

@ -70,14 +70,17 @@ class TestGating:
assert "kein API-Key" in res["skipped"]
m.assert_not_called()
def test_api_failure_is_graceful(self, tmp_path, monkeypatch):
def test_api_failure_tracks_unchecked_not_silent_clean(self, tmp_path, monkeypatch):
monkeypatch.setenv("OPENROUTER_API_KEY", "test")
monkeypatch.setattr("scanner.ai_analyzer.time.sleep", lambda *_: None)
bm = BaselineManager(tmp_path)
with patch("scanner.ai_analyzer._openrouter_chat", return_value=None):
res = run_ai_analysis(_cfg(), bm, _snap(_LONG), {})
assert res["findings"] == []
assert res["api_calls"] == 0
assert res["skipped"] # ein Hinweis wurde gesetzt, aber keine Exception
# Inhalt bleibt UNGEPRÜFT — darf nicht still als clean durchgehen
assert len(res["unchecked"]) == 1
assert res["unchecked"][0]["kind"] == "text"
# ---------------------------------------------------------------------------
@ -248,6 +251,21 @@ class TestScoring:
out = score_ai_findings({"findings": findings}, _cfg())
assert out["score"] == 40
def test_unchecked_level_yellow_raises_from_green(self):
cfg = _cfg()
cfg["ai_analysis"]["unchecked_level"] = "yellow"
out = score_ai_findings(
{"findings": [], "unchecked": [{"kind": "text", "url": "u"}]}, cfg)
assert out["level"] == "yellow"
assert out["exit_code"] == 1
def test_unchecked_level_warn_stays_green_but_notes_reason(self):
cfg = _cfg() # default "warn"
out = score_ai_findings(
{"findings": [], "unchecked": [{"kind": "text", "url": "u"}]}, cfg)
assert out["level"] == "green"
assert any("nicht geprüft" in r for r in out["reasons"])
# ---------------------------------------------------------------------------
# Model escalation chain
@ -291,18 +309,35 @@ class TestEscalation:
entry = next(iter(bm.load_ai_ledger()["entries"].values()))
assert entry["model"] == "paid-3"
def test_all_stages_fail_is_graceful(self, tmp_path, monkeypatch):
def test_all_stages_fail_tracks_unchecked(self, tmp_path, monkeypatch):
monkeypatch.setenv("OPENROUTER_API_KEY", "test")
monkeypatch.setattr("scanner.ai_analyzer.time.sleep", lambda *_: None)
bm = BaselineManager(tmp_path)
cfg = _cfg()
cfg["ai_analysis"]["text"]["models"] = ["free-1", "free-2", "paid-3"]
with patch("scanner.ai_analyzer._openrouter_chat", side_effect=[None, None, None]):
with patch("scanner.ai_analyzer._openrouter_chat", return_value=None):
res = run_ai_analysis(cfg, bm, _snap(_LONG), {})
assert res["findings"] == []
assert res["api_calls"] == 0
assert res["skipped"] # Hinweis gesetzt
assert len(res["unchecked"]) == 1 # ungeprüft erfasst
assert bm.load_ai_ledger()["entries"] == {} # nichts gespeichert
def test_retry_recovers_after_full_chain_failure(self, tmp_path, monkeypatch):
"""Erster Durchlauf der Kette scheitert komplett → Wiederholung rettet das Verdikt."""
monkeypatch.setenv("OPENROUTER_API_KEY", "test")
monkeypatch.setattr("scanner.ai_analyzer.time.sleep", lambda *_: None)
bm = BaselineManager(tmp_path)
cfg = _cfg()
cfg["ai_analysis"]["text"]["models"] = ["m1", "m2"]
cfg["ai_analysis"]["max_retries"] = 1
v = _verdict("clean", "none", 0.9)
# Durchlauf 1: m1,m2 → None,None. Wiederholung: m1 → verdict.
with patch("scanner.ai_analyzer._openrouter_chat", side_effect=[None, None, v]) as m:
res = run_ai_analysis(cfg, bm, _snap(_LONG), {})
assert m.call_count == 3
assert res["unchecked"] == []
assert res["api_calls"] == 1
# ---------------------------------------------------------------------------
# Fingerprint stability