Christoph Emsenhuber
f507c50623
- A0: API-User 'automation' auf agh-bf1/.36 + agh-bf2/.37 (PW in ~/.secrets, TP30-Merkposten) - A: 8 clients.persistent (7 VLAN-CIDR-Gruppen + dns-bypass-exempt), Policy-Matrix (INFRA/MEDIA/MGMT filtering=aus, GUEST safe-search+parental an), live verifiziert - B: agh-sync.py (check/push), Repo-SSoT AdGuardHome.yaml mit REDIGIERTEN Hashes (Alt-Hash-Exposure im Git entfernt); Invariante .36==.37 (AGH normalisiert beim Restart) - C: agh-exception.py (allow-domain/unallow/unfilter/refilter/status) gegen beide Instanzen, end-to-end getestet; Contract in adguard/README.md - Stufe 6 (Bypass-Sperre) NICHT geschaltet; kein DHCP-/Firewall-Write Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
104 lines
4.0 KiB
Python
Executable File
104 lines
4.0 KiB
Python
Executable File
#!/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
|
|
|
|
INST = ["10.40.20.36", "10.40.20.37"]
|
|
SECRETS = pathlib.Path.home() / ".secrets/credentials.env"
|
|
|
|
|
|
def creds():
|
|
u = p = None
|
|
for line in SECRETS.read_text().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()
|
|
if not (u and p): sys.exit("AGH_AUTOMATION_USER/_PASS fehlen in ~/.secrets")
|
|
return u, p
|
|
|
|
|
|
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):
|
|
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):
|
|
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__}")
|