bf-infra/adguard/agh-exception.py

#!/usr/bin/env python3
"""
agh-exception.py — TP23 Stufe 5 Exception-Primitive gegen das AGH-Paar (.36 + .37).

Von der HA-Self-Service-Schicht (rest_command) ODER von Hand aufrufbar. Wirkt IMMER auf
BEIDE Instanzen (kein Sync im Datenpfad). KEIN eigenes Timing — Ablauf/Revert macht der
Aufrufer (HA-Timer) bzw. der RouterOS-Backstop (Stufe 6). Diese CLI ist reines add/remove.

  allow-domain   <domain>   → user_rules += @@||domain^   (Domain für ALLE entsperren)
  unallow-domain <domain>   → user_rules -= @@||domain^
  unfilter       <ip>       → Client (dessen ids <ip> enthält) filtering_enabled=false
  refilter       <ip>       → denselben Client wieder filtern (use_global_settings=true)
  status                    → user_rules + Client-Policy beider Instanzen

Auth: AGH_AUTOMATION_USER/_PASS aus ~/.secrets/credentials.env (nie in argv).
"""
import sys, os, json, base64, urllib.request, urllib.error, pathlib, re

DOMAIN_RE = re.compile(r"^(?=.{1,253}$)([a-z0-9](-?[a-z0-9]+)*\.)+[a-z]{2,}$", re.I)
IP_RE = re.compile(r"^\d{1,3}(\.\d{1,3}){3}$")

INST = ["10.40.20.36", "10.40.20.37"]
# Creds-Quellen in Reihenfolge: erste, die User+Pass liefert, gewinnt.
# So läuft derselbe Code auf dem Mac (~/.secrets) und auf HAOS (/config).
CRED_FILES = [
    os.environ.get("AGH_CREDS_FILE"),
    str(pathlib.Path.home() / ".secrets/credentials.env"),
    "/config/agh-exception.secrets",
]


def _from_file(path):
    u = p = None
    try:
        for line in open(path).read().splitlines():
            if line.startswith("AGH_AUTOMATION_USER="): u = line.split("=", 1)[1].strip()
            if line.startswith("AGH_AUTOMATION_PASS="): p = line.split("=", 1)[1].strip()
    except OSError:
        return None
    return (u, p) if (u and p) else None


def creds():
    eu, ep = os.environ.get("AGH_AUTOMATION_USER"), os.environ.get("AGH_AUTOMATION_PASS")
    if eu and ep:
        return eu, ep
    for f in CRED_FILES:
        if f and (r := _from_file(f)):
            return r
    sys.exit("AGH_AUTOMATION_USER/_PASS: weder Env noch eine Creds-Datei gefunden")


AUTH = "Basic " + base64.b64encode(":".join(creds()).encode()).decode()


def api(ip, path, method="GET", body=None):
    url = f"http://{ip}{path}"
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(url, data=data, method=method,
                                 headers={"Authorization": AUTH,
                                          "Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=8) as r:
            raw = r.read()
            return json.loads(raw) if raw else {}
    except urllib.error.HTTPError as e:
        sys.exit(f"{ip} {path} → HTTP {e.code}: {e.read().decode()[:200]}")
    except Exception as e:
        sys.exit(f"{ip} {path} → {e}")


def rule(domain):
    return f"@@||{domain}^"


def allow(domain, remove=False):
    domain = domain.strip().lower()
    if not DOMAIN_RE.match(domain):
        sys.exit(f"ungültige Domain: {domain!r}")
    r = rule(domain)
    for ip in INST:
        st = api(ip, "/control/filtering/status")
        rules = st.get("user_rules", [])
        if remove:
            rules = [x for x in rules if x != r]
        elif r not in rules:
            rules = rules + [r]
        api(ip, "/control/filtering/set_rules", "POST", {"rules": rules})
        print(f"{ip}: {'entfernt' if remove else 'gesetzt'} {r} (user_rules={len(rules)})")


def set_filter(ip_addr, enabled):
    if not IP_RE.match(ip_addr):
        sys.exit(f"ungültige IP: {ip_addr!r}")
    for ip in INST:
        data = api(ip, "/control/clients")
        c = None
        for cand in data.get("clients", []):
            if ip_addr in cand.get("ids", []):
                c = cand; break
        if not c:
            sys.exit(f"{ip}: kein Client mit id exakt {ip_addr} — für Gruppen die CIDR angeben")
        c = dict(c)
        c["use_global_settings"] = enabled
        c["filtering_enabled"] = enabled
        api(ip, "/control/clients/update", "POST", {"name": c["name"], "data": c})
        print(f"{ip}: {c['name']} filtering={'an' if enabled else 'AUS'}")


def status():
    for ip in INST:
        st = api(ip, "/control/filtering/status")
        allows = [r for r in st.get("user_rules", []) if r.startswith("@@")]
        cl = api(ip, "/control/clients").get("clients", [])
        print(f"== {ip} ==  allow-rules={allows or '—'}")
        for c in cl:
            print(f"   {c['name']:16} ids={c['ids']} global={c['use_global_settings']} filt={c['filtering_enabled']}")


if __name__ == "__main__":
    a = sys.argv[1:]
    if not a: sys.exit(__doc__)
    cmd = a[0]
    if cmd == "allow-domain":   allow(a[1])
    elif cmd == "unallow-domain": allow(a[1], remove=True)
    elif cmd == "unfilter":     set_filter(a[1], False)
    elif cmd == "refilter":     set_filter(a[1], True)
    elif cmd == "status":       status()
    else: sys.exit(f"unbekannt: {cmd}\n{__doc__}")