HitonabiandClaude Opus 5.5
ad8dbbc1b3
Jede Gerätekachel (Homelab-Cockpit und Übersicht) zeigt eine Live-Zahl, wo es eine gibt: Proxmox-Host „8 von 8 Gästen an“, PBS „letzte Sicherung vor 13 h“, NPMplus „10 Zertifikate gültig“ (aus der Zertifikats-Wache), Arcane „11 von 12 Containern an“ (Arcane-API mit dem vorhandenen Schlüssel) und AdGuard „12 % geblockt“. Für AdGuard gibt es unter Einstellungen → Homelab einen Zugang, der vor dem Speichern gegen AdGuard geprüft wird und nur im Homelab-Teil liegt (0600, das Passwort steht in keiner Antwort). Arcane und AdGuard werden höchstens alle zwei Minuten gefragt. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
109 lines
4.2 KiB
Python
109 lines
4.2 KiB
Python
"""AdGuard Home: der Anteil geblockter Anfragen für die Kachel (Ausbauplan Welle 1, seit 25.09.2026).
|
|
|
|
Die Statistik (GET /control/stats) gibt AdGuard nur mit Anmeldung heraus. Den Zugang trägt der User selbst unter
|
|
Einstellungen ein; vor dem Speichern fragt dieser Teil AdGuard damit einmal — lehnt es ab oder antwortet nicht, wird
|
|
nichts gespeichert. Er liegt in <Datenordner>/adguard.json (0600) und wird bei jedem Zugriff gelesen: ein neuer gilt
|
|
ohne Neustart. Das Passwort steht nie in einer Antwort, einem Protokoll oder einer Fehlermeldung; den Benutzernamen
|
|
zeigt die Oberfläche, damit man weiß, welcher hinterlegt ist. Am besten ein eigener Benutzer nur dafür.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
from kern.einstellungen import einstellungen
|
|
|
|
from services.homelab import apps, kanal
|
|
|
|
ZEITLIMIT = httpx.Timeout(10.0, connect=5.0)
|
|
CACHE_S = 120
|
|
_lock = threading.Lock()
|
|
_cache: dict[str, tuple[float, dict | None]] = {}
|
|
|
|
|
|
def _pfad() -> Path:
|
|
return einstellungen().daten_dir / "adguard.json"
|
|
|
|
|
|
def zugang() -> tuple[str, str] | None:
|
|
"""(Benutzer, Passwort) oder None."""
|
|
try:
|
|
daten = json.loads(_pfad().read_text(encoding="utf-8"))
|
|
except (OSError, ValueError, UnicodeDecodeError):
|
|
return None
|
|
if not isinstance(daten, dict):
|
|
return None
|
|
benutzer, passwort = daten.get("benutzer"), daten.get("passwort")
|
|
return (benutzer, passwort) if isinstance(benutzer, str) and isinstance(passwort, str) and benutzer else None
|
|
|
|
|
|
def zugang_speichern(benutzer: str, passwort: str) -> None:
|
|
"""Atomar und von Anfang an nur für den Dienst lesbar (0600)."""
|
|
pfad = _pfad()
|
|
tmp = pfad.with_name(pfad.name + ".tmp")
|
|
with _lock:
|
|
pfad.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp.unlink(missing_ok=True)
|
|
fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
|
json.dump({"benutzer": benutzer, "passwort": passwort}, f)
|
|
os.replace(tmp, pfad)
|
|
_cache.clear()
|
|
|
|
|
|
def zugang_loeschen() -> bool:
|
|
with _lock:
|
|
pfad = _pfad()
|
|
da = pfad.exists()
|
|
pfad.unlink(missing_ok=True)
|
|
_cache.clear()
|
|
return da
|
|
|
|
|
|
def url() -> str | None:
|
|
"""Die Oberfläche von AdGuard aus dem Bericht des Ausführers."""
|
|
for gast in ((kanal.bericht() or {}).get("bericht") or {}).get("gaeste") or []:
|
|
app = apps.app_fuer((gast.get("app") or {}).get("kennung"), gast.get("name"))
|
|
if app and app.kennung == "adguard" and (adresse := apps.adresse(app, gast.get("ip"))):
|
|
return adresse.rstrip("/")
|
|
return None
|
|
|
|
|
|
def statistik(adresse: str, benutzer: str, passwort: str) -> dict:
|
|
"""{"ok": True, "anfragen", "geblockt", "anteil"} oder {"ok": False, "status", "art": abgelehnt|http|netz|antwort}.
|
|
Nichts davon enthält das Passwort."""
|
|
try:
|
|
r = httpx.get(f"{adresse}/control/stats", auth=(benutzer, passwort), timeout=ZEITLIMIT)
|
|
r.raise_for_status()
|
|
daten = r.json()
|
|
anfragen, geblockt = int(daten["num_dns_queries"]), int(daten["num_blocked_filtering"])
|
|
except httpx.HTTPStatusError as exc:
|
|
status = exc.response.status_code
|
|
return {"ok": False, "status": status, "art": "abgelehnt" if status in (401, 403) else "http"}
|
|
except httpx.HTTPError:
|
|
return {"ok": False, "status": None, "art": "netz"}
|
|
except (ValueError, KeyError, TypeError):
|
|
return {"ok": False, "status": None, "art": "antwort"}
|
|
return {"ok": True, "anfragen": anfragen, "geblockt": geblockt,
|
|
"anteil": round(geblockt / anfragen * 100, 1) if anfragen else None}
|
|
|
|
|
|
def geblockt() -> dict | None:
|
|
"""Für die Kachel: die Statistik mit dem hinterlegten Zugang, höchstens alle CACHE_S neu gefragt. None ohne Zugang,
|
|
ohne AdGuard im Bericht oder wenn AdGuard nicht antwortet."""
|
|
z, adresse = zugang(), url()
|
|
if not z or not adresse:
|
|
return None
|
|
jetzt = time.time()
|
|
with _lock:
|
|
if (eintrag := _cache.get(adresse)) and jetzt - eintrag[0] < CACHE_S:
|
|
return eintrag[1]
|
|
ergebnis = statistik(adresse, *z)
|
|
wert = ergebnis if ergebnis["ok"] else None
|
|
with _lock:
|
|
_cache[adresse] = (jetzt, wert)
|
|
return wert
|