geoblock: mehrere iptables-Chains unterstuetzen (chain = INPUT, FORWARD)
'chain' akzeptiert jetzt eine komma-getrennte Liste. Noetig, wenn der geschuetzte Dienst hinter einer Firewall/einem Router liegt (Portweiterleitung/DNAT): solcher Traffic laeuft durch die FORWARD-Chain und wird von einer reinen INPUT-Regel nie gesehen. "chain = INPUT, FORWARD" deckt beide Wege ab. - load_config: cfg["chain"] -> cfg["chains"] (Liste), Default ["INPUT"]. - apply/purge/set_iptables_rules/reassert/status/Watchdog arbeiten ueber alle konfigurierten Chains. geoblock_rules_on_top aggregiert (on_top nur, wenn in JEDER Chain oben; present, wenn in mindestens einer vorhanden) -- neuer Helper _chain_rules_position() prueft je Chain. --status zeigt Position pro Chain. - Doku: geoblock.ini.example + README-Abschnitt "INPUT vs FORWARD" (inkl. DNAT-Diagnose und Endian-Hinweis). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
duffyduck
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
867a8a1c6f
commit
da83f351d7
+67
-47
@@ -164,9 +164,16 @@ def load_config(path):
|
||||
if pr not in ("tcp", "udp"):
|
||||
raise ValueError(f"Ungueltiges Protokoll '{pr}' in 'protocol' (erlaubt: tcp, udp)")
|
||||
|
||||
# chain: eine oder (komma-getrennt) mehrere iptables-Chains. INPUT = Traffic ZU
|
||||
# diesem Host, FORWARD = durch diesen Host GEROUTETER Traffic (z.B. Portweiter-
|
||||
# leitung/DNAT auf einen internen Host hinter einer Firewall/Router). "INPUT, FORWARD"
|
||||
# deckt beides ab. Default: INPUT.
|
||||
chains_raw = sec.get("chain", "INPUT").strip()
|
||||
chains = [c.strip() for c in chains_raw.split(",") if c.strip()] or ["INPUT"]
|
||||
|
||||
return {
|
||||
"countries": countries,
|
||||
"chain": sec.get("chain", "INPUT").strip(),
|
||||
"chains": chains,
|
||||
"interface": sec.get("interface", "").strip(),
|
||||
"log": sec.getboolean("log", fallback=False),
|
||||
"log_prefix": sec.get("log_prefix", "GEOBLOCK-DROP:").strip(),
|
||||
@@ -301,8 +308,8 @@ def _list_chain_rules(chain):
|
||||
|
||||
|
||||
def purge_geoblock_rules(cfg, dry_run=False):
|
||||
"""Loescht ALLE Regeln in der Chain, die auf das geoblock-ipset verweisen
|
||||
(--match-set <name> src) -- unabhaengig von Proto/Port/Variante.
|
||||
"""Loescht in ALLEN konfigurierten Chains alle Regeln, die auf das geoblock-ipset
|
||||
verweisen (--match-set <name> src) -- unabhaengig von Proto/Port/Variante.
|
||||
|
||||
Dadurch werden auch alte Regeln aus frueheren --apply-Laeufen mit anderer
|
||||
ini (z.B. andere Ports/Protokolle) sauber entfernt, die build_rule_variants(cfg)
|
||||
@@ -312,20 +319,22 @@ def purge_geoblock_rules(cfg, dry_run=False):
|
||||
-D immer exakt zur real existierenden Regel.
|
||||
"""
|
||||
name = cfg["ipset_name"]
|
||||
chain = cfg["chain"]
|
||||
needle = f"--match-set {name} src"
|
||||
removed = 0
|
||||
for line in _list_chain_rules(chain):
|
||||
if needle not in line:
|
||||
continue
|
||||
tokens = shlex.split(line) # z.B. ["-A", "INPUT", "-m", "set", ...]
|
||||
# "-A <chain> <rest>" -> "iptables -D <chain> <rest>"
|
||||
run(["iptables", "-D"] + tokens[1:], dry_run=dry_run, check=False)
|
||||
removed += 1
|
||||
if removed:
|
||||
log(f"{removed} bestehende geoblock-Regel(n) aus Chain '{chain}' entfernt (Set '{name}')")
|
||||
else:
|
||||
log(f"Keine bestehenden geoblock-Regeln in Chain '{chain}' gefunden")
|
||||
for chain in cfg["chains"]:
|
||||
n = 0
|
||||
for line in _list_chain_rules(chain):
|
||||
if needle not in line:
|
||||
continue
|
||||
tokens = shlex.split(line) # z.B. ["-A", "INPUT", "-m", "set", ...]
|
||||
# "-A <chain> <rest>" -> "iptables -D <chain> <rest>"
|
||||
run(["iptables", "-D"] + tokens[1:], dry_run=dry_run, check=False)
|
||||
n += 1
|
||||
if n:
|
||||
log(f"{n} bestehende geoblock-Regel(n) aus Chain '{chain}' entfernt (Set '{name}')")
|
||||
removed += n
|
||||
if not removed:
|
||||
log(f"Keine bestehenden geoblock-Regeln in Chains {cfg['chains']} gefunden")
|
||||
return removed
|
||||
|
||||
|
||||
@@ -344,40 +353,50 @@ def set_iptables_rules(cfg, dry_run=False):
|
||||
name = cfg["ipset_name"]
|
||||
set_match = ["-m", "set", "--match-set", name, "src"]
|
||||
iface_opt = ["-i", cfg["interface"]] if cfg["interface"] else []
|
||||
for variant in build_rule_variants(cfg):
|
||||
proto_args = _variant_match_args(variant)
|
||||
port_args = _variant_port_args(variant)
|
||||
label = _variant_label(variant)
|
||||
for chain in cfg["chains"]:
|
||||
for variant in build_rule_variants(cfg):
|
||||
proto_args = _variant_match_args(variant)
|
||||
port_args = _variant_port_args(variant)
|
||||
label = _variant_label(variant)
|
||||
|
||||
run(["iptables", "-I", cfg["chain"]] + iface_opt + proto_args + set_match +
|
||||
port_args + ["-j", "DROP"], dry_run=dry_run)
|
||||
log(f"DROP-Regel am Anfang von Chain '{cfg['chain']}' aktiv fuer Set '{name}' ({label})")
|
||||
run(["iptables", "-I", chain] + iface_opt + proto_args + set_match +
|
||||
port_args + ["-j", "DROP"], dry_run=dry_run)
|
||||
log(f"DROP-Regel am Anfang von Chain '{chain}' aktiv fuer Set '{name}' ({label})")
|
||||
|
||||
if cfg["log"]:
|
||||
run(["iptables", "-I", cfg["chain"]] + iface_opt + proto_args + set_match +
|
||||
port_args + ["-j", "LOG", "--log-prefix", cfg["log_prefix"] + " "], dry_run=dry_run)
|
||||
log(f"LOG-Regel am Anfang von Chain '{cfg['chain']}' aktiv ({label})")
|
||||
if cfg["log"]:
|
||||
run(["iptables", "-I", chain] + iface_opt + proto_args + set_match +
|
||||
port_args + ["-j", "LOG", "--log-prefix", cfg["log_prefix"] + " "], dry_run=dry_run)
|
||||
log(f"LOG-Regel am Anfang von Chain '{chain}' aktiv ({label})")
|
||||
|
||||
|
||||
def geoblock_rules_on_top(cfg):
|
||||
"""Prueft, ob die geoblock-Regeln (noch) ganz am Anfang der Chain stehen.
|
||||
|
||||
Gibt (on_top, present) zurueck:
|
||||
def _chain_rules_position(chain, ipset_name):
|
||||
"""(on_top, present) fuer eine EINZELNE Chain.
|
||||
present : ob ueberhaupt geoblock-Regeln in der Chain vorhanden sind
|
||||
on_top : ob sie luueckenlos die ERSTEN Regeln der Chain sind (Index 0..k-1)
|
||||
|
||||
Ist present=True aber on_top=False, sind unsere Regeln nach unten gerutscht
|
||||
(z.B. weil eine gemanagte Firewall die Chain neu gebaut und eigene Regeln
|
||||
davor eingehaengt hat) -- dann sollten sie neu an den Anfang gesetzt werden.
|
||||
"""
|
||||
needle = f"--match-set {cfg['ipset_name']} src"
|
||||
rules = _list_chain_rules(cfg["chain"]) # geordnete Liste der '-A ...'-Regeln
|
||||
needle = f"--match-set {ipset_name} src"
|
||||
rules = _list_chain_rules(chain) # geordnete Liste der '-A ...'-Regeln
|
||||
idx = [i for i, ln in enumerate(rules) if needle in ln]
|
||||
present = bool(idx)
|
||||
on_top = present and idx == list(range(len(idx)))
|
||||
return on_top, present
|
||||
|
||||
|
||||
def geoblock_rules_on_top(cfg):
|
||||
"""Aggregiert die Regel-Position ueber ALLE konfigurierten Chains.
|
||||
|
||||
Gibt (on_top, present) zurueck:
|
||||
on_top : True nur, wenn in JEDER konfigurierten Chain die Regeln vorhanden
|
||||
UND ganz oben sind (Index 0..k-1). Sonst False -> --reassert sollte
|
||||
korrigieren (fehlende neu setzen, verrutschte wieder nach oben).
|
||||
present : True, wenn mindestens eine Chain geoblock-Regeln hat.
|
||||
"""
|
||||
results = [_chain_rules_position(ch, cfg["ipset_name"]) for ch in cfg["chains"]]
|
||||
on_top = all(on for on, _ in results)
|
||||
present = any(pr for _, pr in results)
|
||||
return on_top, present
|
||||
|
||||
|
||||
def reassert_rules(cfg, dry_run=False):
|
||||
"""Watchdog: stellt sicher, dass die geoblock-Regeln am ANFANG der Chain stehen
|
||||
-- OHNE die Zonefiles neu von ipdeny zu laden (nutzt das bestehende ipset).
|
||||
@@ -475,18 +494,19 @@ def show_status(cfg):
|
||||
print(f"ipset '{name}' existiert nicht (noch nicht aktiv).")
|
||||
|
||||
variants = build_rule_variants(cfg)
|
||||
for variant in variants:
|
||||
active = iptables_rule_exists(cfg["chain"], name, cfg["interface"], variant=variant)
|
||||
print(f"iptables DROP-Regel in Chain '{cfg['chain']}' ({_variant_label(variant)}) aktiv: {active}")
|
||||
for chain in cfg["chains"]:
|
||||
for variant in variants:
|
||||
active = iptables_rule_exists(chain, name, cfg["interface"], variant=variant)
|
||||
print(f"iptables DROP-Regel in Chain '{chain}' ({_variant_label(variant)}) aktiv: {active}")
|
||||
|
||||
on_top, present = geoblock_rules_on_top(cfg)
|
||||
if not present:
|
||||
print(f"Regel-Position in Chain '{cfg['chain']}': keine geoblock-Regeln vorhanden")
|
||||
elif on_top:
|
||||
print(f"Regel-Position in Chain '{cfg['chain']}': am Anfang (ok)")
|
||||
else:
|
||||
print(f"Regel-Position in Chain '{cfg['chain']}': NACH UNTEN GERUTSCHT "
|
||||
f"-- '--reassert' oder '--apply' setzt sie wieder an den Anfang")
|
||||
on_top, present = _chain_rules_position(chain, name)
|
||||
if not present:
|
||||
print(f"Regel-Position in Chain '{chain}': keine geoblock-Regeln vorhanden")
|
||||
elif on_top:
|
||||
print(f"Regel-Position in Chain '{chain}': am Anfang (ok)")
|
||||
else:
|
||||
print(f"Regel-Position in Chain '{chain}': NACH UNTEN GERUTSCHT "
|
||||
f"-- '--reassert' oder '--apply' setzt sie wieder an den Anfang")
|
||||
|
||||
print(f"Konfigurierte Laender: {', '.join(c.upper() for c in cfg['countries'])}")
|
||||
if cfg["ports"]:
|
||||
|
||||
Reference in New Issue
Block a user