TP23 Stufe 5 Redesign: pro-Person Self-Service (agh-exception.py + Bau-Spec)
Stufe 5 reopened — DAU-Requirements (pro Person, "zuletzt geblockt -> 1 Klick", temp+dauerhaft umschaltbar, scoped HA-Dashboard) nicht erfuellt → Stufe 6 gated. agh-exception.py: $client-scoped allow/unallow, list-exceptions, recent-blocked (selftest OK, reine Logik netzunabhaengig getestet). HA-Bau-Spec + Personen- Client-Modell dokumentiert. Bau/Abnahme on-network offen. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -2,22 +2,37 @@
|
||||
"""
|
||||
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
|
||||
Von der HA-Self-Service-Schicht (shell_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.
|
||||
Aufrufer (HA-Timer) bzw. der RouterOS-Backstop (Stufe 6). Diese CLI ist reines add/remove/read.
|
||||
|
||||
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
|
||||
allow-domain <domain> [--client <name|ip[,ip]>] → user_rules += @@||domain^[$client=…]
|
||||
unallow-domain <domain> [--client <name|ip[,ip]>] → dieselbe Regel wieder entfernen
|
||||
list-exceptions [--client <name>] [--json] → alle @@-Regeln, nach Client gruppiert
|
||||
recent-blocked --client <name|ip[,ip]> [--limit N] [--json]
|
||||
→ zuletzt geblockte Domains DIESER Person
|
||||
unfilter <ip> → Client (dessen ids <ip> enthält) filtering_enabled=false (grob!)
|
||||
refilter <ip> → denselben Client wieder filtern (use_global_settings=true)
|
||||
status → user_rules + Client-Policy beider Instanzen
|
||||
|
||||
--client:
|
||||
* ein AGH-Client-NAME (empfohlen, stabil bei Gerätewechsel) → Regel bekommt $client='Name',
|
||||
IPs werden für recent-blocked aus /control/clients (ids) aufgelöst.
|
||||
* ODER eine/mehrere IP/CIDR, komma-getrennt → $client=<ip>|<ip>, recent-blocked filtert darauf.
|
||||
ohne --client bei allow/unallow = GLOBAL (für alle) — nur für Admin-Zwecke.
|
||||
|
||||
🔴 ANNAHME, on-network zu verifizieren, BEVOR Stufe 6 scharf geht: dass AGH $client= auf
|
||||
ALLOW-Regeln (@@) in der laufenden Version wirkt. Die Regelkonstruktion steht isoliert in
|
||||
scoped_rule()/parse_allow_rule() — falls die Syntax abweicht, nur dort anpassen.
|
||||
|
||||
Auth: AGH_AUTOMATION_USER/_PASS aus ~/.secrets/credentials.env (nie in argv).
|
||||
"""
|
||||
import sys, os, json, base64, urllib.request, urllib.error, pathlib, re
|
||||
import sys, os, json, base64, urllib.request, urllib.error, urllib.parse, pathlib, re, ipaddress, argparse
|
||||
|
||||
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}$")
|
||||
CIDR_RE = re.compile(r"^\d{1,3}(\.\d{1,3}){3}/\d{1,2}$")
|
||||
# Ein Client-„spec" ist IP-artig, wenn JEDES komma-Glied IP oder CIDR ist; sonst = Name.
|
||||
|
||||
INST = ["10.40.20.36", "10.40.20.37"]
|
||||
# Creds-Quellen in Reihenfolge: erste, die User+Pass liefert, gewinnt.
|
||||
@@ -50,14 +65,20 @@ def creds():
|
||||
sys.exit("AGH_AUTOMATION_USER/_PASS: weder Env noch eine Creds-Datei gefunden")
|
||||
|
||||
|
||||
AUTH = "Basic " + base64.b64encode(":".join(creds()).encode()).decode()
|
||||
_AUTH = None
|
||||
def auth_header():
|
||||
# lazy: erst beim ersten echten API-Call — so bleibt das Modul ohne Creds importier-/testbar.
|
||||
global _AUTH
|
||||
if _AUTH is None:
|
||||
_AUTH = "Basic " + base64.b64encode(":".join(creds()).encode()).decode()
|
||||
return _AUTH
|
||||
|
||||
|
||||
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,
|
||||
headers={"Authorization": auth_header(),
|
||||
"Content-Type": "application/json"})
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=8) as r:
|
||||
@@ -69,26 +90,129 @@ def api(ip, path, method="GET", body=None):
|
||||
sys.exit(f"{ip} {path} → {e}")
|
||||
|
||||
|
||||
def rule(domain):
|
||||
return f"@@||{domain}^"
|
||||
# ---------------------------------------------------------------------------
|
||||
# Reine Regel-Logik (netzunabhängig, self-testbar). Bei AGH-Syntax-Drift NUR hier anfassen.
|
||||
# ---------------------------------------------------------------------------
|
||||
def is_ip_spec(spec):
|
||||
"""True, wenn spec eine (Komma-)Liste aus lauter IPs/CIDRs ist (→ IP-spec), sonst Name."""
|
||||
parts = [s.strip() for s in spec.split(",") if s.strip()]
|
||||
return bool(parts) and all(IP_RE.match(p) or CIDR_RE.match(p) for p in parts)
|
||||
|
||||
|
||||
def allow(domain, remove=False):
|
||||
def client_modifier(spec):
|
||||
"""$client-Modifier-Wert für eine Regel. IP-spec → ip|ip ; Name → 'Name' (quoted)."""
|
||||
if is_ip_spec(spec):
|
||||
return "|".join(s.strip() for s in spec.split(",") if s.strip())
|
||||
return "'" + spec.replace("'", "") + "'"
|
||||
|
||||
|
||||
def scoped_rule(domain, client=None):
|
||||
"""@@||domain^ bzw. @@||domain^$client=… (client = Name oder ip[,ip])."""
|
||||
base = f"@@||{domain}^"
|
||||
return base if not client else f"{base}$client={client_modifier(client)}"
|
||||
|
||||
|
||||
_ALLOW_RE = re.compile(r"^@@\|\|(?P<domain>[^\^]+)\^(?:\$client=(?P<client>.+))?$")
|
||||
def parse_allow_rule(rule):
|
||||
"""@@-Regel → {'domain':…, 'client': <str oder None>, 'raw': rule} oder None."""
|
||||
m = _ALLOW_RE.match(rule.strip())
|
||||
if not m:
|
||||
return None
|
||||
return {"domain": m.group("domain"), "client": m.group("client"), "raw": rule.strip()}
|
||||
|
||||
|
||||
def ip_in_specs(ip, specs):
|
||||
"""Ist ip (aus dem Query-Log) von einer der specs (IP oder CIDR) erfasst?"""
|
||||
try:
|
||||
addr = ipaddress.ip_address(ip)
|
||||
except ValueError:
|
||||
return False
|
||||
for s in specs:
|
||||
s = s.strip()
|
||||
try:
|
||||
if "/" in s:
|
||||
if addr in ipaddress.ip_network(s, strict=False):
|
||||
return True
|
||||
elif ip == s:
|
||||
return True
|
||||
except ValueError:
|
||||
continue
|
||||
return False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
def resolve_client_ips(spec):
|
||||
"""spec → Liste von IP/CIDR-strings. IP-spec direkt; Name → ids des AGH-Clients (von .36)."""
|
||||
if is_ip_spec(spec):
|
||||
return [s.strip() for s in spec.split(",") if s.strip()]
|
||||
data = api(INST[0], "/control/clients")
|
||||
for c in data.get("clients", []):
|
||||
if c.get("name") == spec:
|
||||
return list(c.get("ids", []))
|
||||
sys.exit(f"kein AGH-Client mit name={spec!r} (für IP-spec Komma-Liste aus IP/CIDR angeben)")
|
||||
|
||||
|
||||
def allow(domain, client=None, remove=False):
|
||||
domain = domain.strip().lower()
|
||||
if not DOMAIN_RE.match(domain):
|
||||
sys.exit(f"ungültige Domain: {domain!r}")
|
||||
r = rule(domain)
|
||||
r = scoped_rule(domain, client)
|
||||
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]
|
||||
rules = [x for x in rules if x.strip() != 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 list_exceptions(client=None, as_json=False):
|
||||
st = api(INST[0], "/control/filtering/status") # .36 == .37 (Invariante, via agh-sync geprüft)
|
||||
items = []
|
||||
for raw in st.get("user_rules", []):
|
||||
p = parse_allow_rule(raw)
|
||||
if not p:
|
||||
continue
|
||||
if client is not None and (p["client"] or "") != client_modifier(client):
|
||||
continue
|
||||
items.append(p)
|
||||
if as_json:
|
||||
print(json.dumps(items, ensure_ascii=False))
|
||||
return
|
||||
if not items:
|
||||
print("— keine Ausnahmen —"); return
|
||||
for p in items:
|
||||
who = p["client"] or "GLOBAL"
|
||||
print(f" {p['domain']:40} client={who}")
|
||||
|
||||
|
||||
def recent_blocked(client, limit=20, as_json=False):
|
||||
specs = resolve_client_ips(client)
|
||||
q = urllib.parse.urlencode({"response_status": "blocked", "limit": max(limit * 8, 100)})
|
||||
data = api(INST[0], f"/control/querylog?{q}")
|
||||
seen = {} # domain -> {count, last}
|
||||
for e in data.get("data", []):
|
||||
c = e.get("client", "")
|
||||
if not ip_in_specs(c, specs):
|
||||
continue
|
||||
dom = (e.get("question", {}) or {}).get("name", "").rstrip(".").lower()
|
||||
if not dom:
|
||||
continue
|
||||
rec = seen.setdefault(dom, {"domain": dom, "count": 0, "last": e.get("time", "")})
|
||||
rec["count"] += 1
|
||||
if e.get("time", "") > rec["last"]:
|
||||
rec["last"] = e.get("time", "")
|
||||
out = sorted(seen.values(), key=lambda r: r["last"], reverse=True)[:limit]
|
||||
if as_json:
|
||||
print(json.dumps(out, ensure_ascii=False)); return
|
||||
if not out:
|
||||
print("— nichts geblockt —"); return
|
||||
for r in out:
|
||||
print(f" {r['domain']:40} x{r['count']:<4} {r['last']}")
|
||||
|
||||
|
||||
def set_filter(ip_addr, enabled):
|
||||
if not IP_RE.match(ip_addr):
|
||||
sys.exit(f"ungültige IP: {ip_addr!r}")
|
||||
@@ -117,13 +241,47 @@ def status():
|
||||
print(f" {c['name']:16} ids={c['ids']} global={c['use_global_settings']} filt={c['filtering_enabled']}")
|
||||
|
||||
|
||||
def _selftest():
|
||||
"""Netzunabhängiger Test der reinen Regel-Logik."""
|
||||
assert is_ip_spec("10.40.10.82") and is_ip_spec("10.40.10.82,10.40.10.183")
|
||||
assert is_ip_spec("10.40.10.0/24") and not is_ip_spec("christoph")
|
||||
assert scoped_rule("x.com") == "@@||x.com^"
|
||||
assert scoped_rule("x.com", "10.40.10.82,10.40.10.183") == "@@||x.com^$client=10.40.10.82|10.40.10.183"
|
||||
assert scoped_rule("x.com", "christoph") == "@@||x.com^$client='christoph'"
|
||||
p = parse_allow_rule("@@||x.com^$client='christoph'")
|
||||
assert p and p["domain"] == "x.com" and p["client"] == "'christoph'"
|
||||
p2 = parse_allow_rule("@@||y.com^")
|
||||
assert p2 and p2["client"] is None
|
||||
assert parse_allow_rule("||block.com^") is None
|
||||
assert ip_in_specs("10.40.10.82", ["10.40.10.82"]) is True
|
||||
assert ip_in_specs("10.40.10.5", ["10.40.10.0/24"]) is True
|
||||
assert ip_in_specs("10.40.20.5", ["10.40.10.0/24"]) is False
|
||||
# round-trip: eine scoped Regel wird von list-exceptions --client wieder getroffen
|
||||
assert (parse_allow_rule(scoped_rule("z.com", "christoph"))["client"]
|
||||
== client_modifier("christoph"))
|
||||
print("selftest OK")
|
||||
|
||||
|
||||
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__}")
|
||||
ap = argparse.ArgumentParser(add_help=False)
|
||||
ap.add_argument("cmd", nargs="?")
|
||||
ap.add_argument("arg", nargs="?")
|
||||
ap.add_argument("--client")
|
||||
ap.add_argument("--permanent", action="store_true") # semantisch = kein Timer; wirkt in HA, nicht hier
|
||||
ap.add_argument("--limit", type=int, default=20)
|
||||
ap.add_argument("--json", action="store_true")
|
||||
a = ap.parse_args()
|
||||
|
||||
if not a.cmd:
|
||||
sys.exit(__doc__)
|
||||
if a.cmd == "allow-domain": allow(a.arg, a.client)
|
||||
elif a.cmd == "unallow-domain": allow(a.arg, a.client, remove=True)
|
||||
elif a.cmd == "list-exceptions": list_exceptions(a.client, a.json)
|
||||
elif a.cmd == "recent-blocked":
|
||||
if not a.client: sys.exit("recent-blocked braucht --client <name|ip[,ip]>")
|
||||
recent_blocked(a.client, a.limit, a.json)
|
||||
elif a.cmd == "unfilter": set_filter(a.arg, False)
|
||||
elif a.cmd == "refilter": set_filter(a.arg, True)
|
||||
elif a.cmd == "status": status()
|
||||
elif a.cmd == "selftest": _selftest()
|
||||
else: sys.exit(f"unbekannt: {a.cmd}\n{__doc__}")
|
||||
|
||||
Reference in New Issue
Block a user