diff --git a/README.md b/README.md index d3d78b8..2edd745 100644 --- a/README.md +++ b/README.md @@ -118,6 +118,39 @@ Port lokal — ob er von aussen erreichbar ist, haengt von eurer sonstigen Firewall/Portweiterleitung ab (das ist bewusst getrennt von den Geoblock-DROP-Regeln, die ja genau diesen Port betreffen sollen). +## Gezieltes Blocken einzelner Ports (Selbstaussperr-Schutz) + +Per Default blockt `--apply` **alle** Ports/Protokolle fuer die gelisteten +Laender — das ist klassisches Geoblocking auf Host-Ebene. Das ist beim +Testen riskant: testet man testweise den eigenen Laendercode (siehe oben), +sperrt man sich damit auch von SSH & Co. aus, falls der Test-Client zufaellig +aus demselben Land connectet. + +Ueber `ports` (und optional `protocol`) in der `.ini` laesst sich das +Blocking auf einzelne Anwendungen eingrenzen. Beispiel: nur den +Testserver-Port blocken, SSH bleibt fuer alle Laender offen: + +```ini +[geoblock] +countries = de +... +ports = 8899 +protocol = tcp +``` + +`--apply` legt dann statt einer allgemeinen DROP-Regel eine mit +`-p tcp -m multiport --dports 8899` an — matched nur Pakete zu Port 8899, +alles andere (inkl. Port 22) bleibt unberuehrt. Mehrere Ports/Bereiche +gehen kommaseparat (`80,443,8000:9000`), mehrere Protokolle ebenso +(`protocol = tcp,udp`, legt dann eine Regel pro Protokoll an). + +`--status` zeigt an, ob gerade eine Port-Einschraenkung aktiv ist und +welche. `--remove` entfernt automatisch die zur aktuellen `.ini` passenden +Regel-Varianten — falls `ports`/`protocol` zwischen zwei Laeufen geaendert +wurden, lohnt sich vor dem naechsten `--apply` ein Blick mit +`iptables -L INPUT -n --line-numbers`, ob noch Altregeln mit der vorherigen +Variante stehen. + ## Logging In `aria_geoblock.ini` unter `log_file` einen Pfad eintragen (Default: @@ -156,6 +189,13 @@ Siehe Kommentare in der Datei selbst — kurz: - `log`: iptables-LOG-Eintrag vor dem DROP (dmesg/kern.log) an/aus - `ipset_name`: Name des ipset-Sets - `log_file`: Pfad zur script-eigenen Log-Datei (fuer checkmk) +- `ports`: optional, Komma-Liste von Ports/Bereichen (z.B. `80,443,8000:9000`). + Leer = alle Ports/Protokolle werden geblockt (klassisches Geoblocking). + Gesetzt = nur diese Ports werden fuer die gelisteten Laender geblockt, der + Rest des Hosts bleibt erreichbar. Siehe Abschnitt "Gezieltes Blocken + einzelner Ports" unten. +- `protocol`: nur relevant wenn `ports` gesetzt ist — `tcp` (Default), `udp` + oder `tcp,udp` Abschnitt `[testserver]` (fuer `aria_geoblock_testserver.py`): diff --git a/aria_geoblock.py b/aria_geoblock.py index ceacca3..13b8638 100755 --- a/aria_geoblock.py +++ b/aria_geoblock.py @@ -49,6 +49,14 @@ AUTOMATISCH AKTUELL HALTEN (Systemstart + taeglich): "beim Booten" + "einmal taeglich" automatisch ein. --apply ist idempotent: es flusht das bestehende Set und befuellt es neu, legt aber KEINE doppelte iptables-Regel an (wird vorher geprueft). + +PORT-EINSCHRAENKUNG (optional): + Per Default blockt das Script ALLE Ports/Protokolle fuer die gelisteten + Laender (klassisches Geoblocking). Ueber "ports" in der .ini kann man das + auf bestimmte Anwendungen/Ports eingrenzen — z.B. nur den Testserver + (8899) blocken, waehrend SSH (22) fuer die Laender offen bleibt. Praktisch + zum gefahrlosen Testen, damit man sich nicht selbst aussperrt. Siehe + Kommentare in aria_geoblock.ini. """ import argparse @@ -134,6 +142,21 @@ def load_config(path): if not countries: raise ValueError("Keine Laender in 'countries' angegeben") + # ports: leer/weggelassen = alle Ports blocken (klassisches Geoblocking). + # Angegeben (z.B. "80,443,8899") = nur diese Ports blocken. + ports_raw = sec.get("ports", "").strip() + ports = [p.strip() for p in ports_raw.split(",") if p.strip()] + for p in ports: + _validate_port_spec(p) + + # protocol: welche(s) Protokoll(e) die Port-Einschraenkung betreffen. + # Nur relevant wenn "ports" gesetzt ist. Default: tcp. Auch "tcp,udp" moeglich. + protocols_raw = sec.get("protocol", "tcp").strip() + protocols = [pr.strip().lower() for pr in protocols_raw.split(",") if pr.strip()] or ["tcp"] + for pr in protocols: + if pr not in ("tcp", "udp"): + raise ValueError(f"Ungueltiges Protokoll '{pr}' in 'protocol' (erlaubt: tcp, udp)") + return { "countries": countries, "chain": sec.get("chain", "INPUT").strip(), @@ -142,9 +165,54 @@ def load_config(path): "log_prefix": sec.get("log_prefix", "GEOBLOCK-DROP:").strip(), "ipset_name": sec.get("ipset_name", "geoblock").strip(), "log_file": sec.get("log_file", "").strip(), + "ports": ports, + "protocols": protocols, } +def _validate_port_spec(spec): + """Erlaubt Einzelport ('443') oder Bereich ('8000:9000'), wie iptables --dports.""" + parts = spec.split(":") + if len(parts) not in (1, 2): + raise ValueError(f"Ungueltige Port-Angabe '{spec}' in 'ports' (erlaubt: '443' oder '8000:9000')") + for part in parts: + if not part.isdigit() or not (0 <= int(part) <= 65535): + raise ValueError(f"Ungueltige Port-Angabe '{spec}' in 'ports' (Ports muessen 0-65535 sein)") + + +def build_rule_variants(cfg): + """Baut die Liste der (proto, dports)-Kombinationen fuer iptables-Regeln. + + Ohne 'ports' in der .ini: eine Variante ohne Proto-/Port-Einschraenkung + (= klassisches Geoblocking, alle Ports/Protokolle). Mit 'ports': eine + Variante pro konfiguriertem Protokoll (multiport braucht -p tcp ODER -p udp, + nicht beides gleichzeitig). + """ + if not cfg["ports"]: + return [{"proto": None, "dports": None}] + dports = ",".join(cfg["ports"]) + return [{"proto": proto, "dports": dports} for proto in cfg["protocols"]] + + +def _variant_match_args(variant): + args = [] + if variant["proto"]: + args += ["-p", variant["proto"]] + return args + + +def _variant_port_args(variant): + if variant["dports"]: + return ["-m", "multiport", "--dports", variant["dports"]] + return [] + + +def _variant_label(variant): + if not variant["dports"]: + return "alle Ports" + return f"{variant['proto']}/{variant['dports']}" + + def fetch_country_cidrs(country_code): """Laedt die CIDR-Liste eines Landes von ipdeny.com. Gibt Liste von Strings zurueck.""" url = ZONE_URL_TMPL.format(cc=country_code) @@ -180,11 +248,15 @@ def ipset_exists(name): return name in result.stdout.splitlines() -def iptables_rule_exists(chain, ipset_name, interface, dry_run=False): +def iptables_rule_exists(chain, ipset_name, interface, variant=None, dry_run=False): + variant = variant or {"proto": None, "dports": None} check_cmd = ["iptables", "-C", chain] if interface: check_cmd += ["-i", interface] - check_cmd += ["-m", "set", "--match-set", ipset_name, "src", "-j", "DROP"] + check_cmd += _variant_match_args(variant) + check_cmd += ["-m", "set", "--match-set", ipset_name, "src"] + check_cmd += _variant_port_args(variant) + check_cmd += ["-j", "DROP"] if dry_run: # Im Dry-Run wissen wir es nicht sicher, nehmen konservativ "existiert nicht" an return False @@ -222,26 +294,33 @@ def apply_geoblock(cfg, dry_run=False): log(f"ipset '{name}' befuellt mit {len(all_cidrs)} Bereichen " f"aus Laendern: {', '.join(c.upper() for c in cfg['countries'])}") - # 4) optionale LOG-Regel + DROP-Regel in iptables, nur wenn noch nicht vorhanden - base_match = ["-m", "set", "--match-set", name, "src"] + # 4) optionale LOG-Regel(n) + DROP-Regel(n) in iptables, nur wenn noch nicht vorhanden. + # Eine Variante pro Protokoll wenn 'ports' gesetzt ist, sonst eine Variante (alle Ports). + set_match = ["-m", "set", "--match-set", name, "src"] iface_opt = ["-i", cfg["interface"]] if cfg["interface"] else [] + variants = build_rule_variants(cfg) - if cfg["log"]: - log_check = ["iptables", "-C", cfg["chain"]] + iface_opt + base_match + [ - "-j", "LOG", "--log-prefix", cfg["log_prefix"] + " " - ] - exists = False if dry_run else subprocess.run(log_check, capture_output=True).returncode == 0 - if not exists: - run(["iptables", "-I", cfg["chain"]] + iface_opt + base_match + - ["-j", "LOG", "--log-prefix", cfg["log_prefix"] + " "], dry_run=dry_run) + for variant in variants: + proto_args = _variant_match_args(variant) + port_args = _variant_port_args(variant) + label = _variant_label(variant) + + if cfg["log"]: + log_check = (["iptables", "-C", cfg["chain"]] + iface_opt + proto_args + set_match + + port_args + ["-j", "LOG", "--log-prefix", cfg["log_prefix"] + " "]) + exists = False if dry_run else subprocess.run(log_check, capture_output=True).returncode == 0 + if not exists: + run(["iptables", "-I", cfg["chain"]] + iface_opt + proto_args + set_match + + port_args + ["-j", "LOG", "--log-prefix", cfg["log_prefix"] + " "], dry_run=dry_run) + else: + log(f"LOG-Regel ({label}) existiert bereits, ueberspringe") + + if iptables_rule_exists(cfg["chain"], name, cfg["interface"], variant=variant, dry_run=dry_run): + log(f"DROP-Regel ({label}) existiert bereits, ueberspringe (idempotent)") else: - log("LOG-Regel existiert bereits, ueberspringe") - - if iptables_rule_exists(cfg["chain"], name, cfg["interface"], dry_run=dry_run): - log("DROP-Regel existiert bereits, ueberspringe (idempotent)") - else: - run(["iptables", "-A", cfg["chain"]] + iface_opt + base_match + ["-j", "DROP"], dry_run=dry_run) - log(f"DROP-Regel in Chain '{cfg['chain']}' aktiv fuer Set '{name}'") + run(["iptables", "-A", cfg["chain"]] + iface_opt + proto_args + set_match + + port_args + ["-j", "DROP"], dry_run=dry_run) + log(f"DROP-Regel in Chain '{cfg['chain']}' aktiv fuer Set '{name}' ({label})") log("Fertig. Hinweis: Regeln sind NICHT reboot-persistent — " "fuer Persistenz z.B. 'iptables-persistent' bzw. 'netfilter-persistent save' nutzen, " @@ -253,23 +332,29 @@ def apply_geoblock(cfg, dry_run=False): def remove_geoblock(cfg, dry_run=False): name = cfg["ipset_name"] iface_opt = ["-i", cfg["interface"]] if cfg["interface"] else [] - base_match = ["-m", "set", "--match-set", name, "src"] + set_match = ["-m", "set", "--match-set", name, "src"] + # Entfernt fuer alle aktuell konfigurierten Varianten. Falls die .ini seit dem letzten + # --apply geaendert wurde (z.B. ports entfernt/hinzugefuegt), koennen dadurch alte + # Regeln mit anderer Variante stehen bleiben -- im Zweifel per 'iptables -L -n' + # pruefen und manuell aufraeumen. + for variant in build_rule_variants(cfg): + proto_args = _variant_match_args(variant) + port_args = _variant_port_args(variant) - # DROP-Regel entfernen (mehrfach versuchen falls doppelt vorhanden) - for _ in range(5): - result = run(["iptables", "-D", cfg["chain"]] + iface_opt + base_match + ["-j", "DROP"], - dry_run=dry_run, check=False) - if dry_run or result is None or result.returncode != 0: - break - - if cfg["log"]: for _ in range(5): - result = run(["iptables", "-D", cfg["chain"]] + iface_opt + base_match + - ["-j", "LOG", "--log-prefix", cfg["log_prefix"] + " "], - dry_run=dry_run, check=False) + result = run(["iptables", "-D", cfg["chain"]] + iface_opt + proto_args + set_match + + port_args + ["-j", "DROP"], dry_run=dry_run, check=False) if dry_run or result is None or result.returncode != 0: break + if cfg["log"]: + for _ in range(5): + result = run(["iptables", "-D", cfg["chain"]] + iface_opt + proto_args + set_match + + port_args + ["-j", "LOG", "--log-prefix", cfg["log_prefix"] + " "], + dry_run=dry_run, check=False) + if dry_run or result is None or result.returncode != 0: + break + run(["ipset", "destroy", name], dry_run=dry_run, check=False) log(f"Geoblock-Regeln entfernt, ipset '{name}' geloescht.") @@ -282,9 +367,15 @@ def show_status(cfg): else: print(f"ipset '{name}' existiert nicht (noch nicht aktiv).") - active = iptables_rule_exists(cfg["chain"], name, cfg["interface"]) - print(f"iptables DROP-Regel in Chain '{cfg['chain']}' aktiv: {active}") + 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}") print(f"Konfigurierte Laender: {', '.join(c.upper() for c in cfg['countries'])}") + if cfg["ports"]: + print(f"Port-Einschraenkung: {', '.join(cfg['ports'])} ({'/'.join(cfg['protocols'])})") + else: + print("Port-Einschraenkung: keine (alle Ports/Protokolle werden geblockt)") def main(): @@ -299,12 +390,12 @@ def main(): parser.add_argument("--dry-run", action="store_true", help="Nur anzeigen was passieren wuerde, nichts aendern") args = parser.parse_args() - cfg = load_config(args.config) - LOG_FILE = cfg["log_file"] or None - action = "apply" if args.apply else ("remove" if args.remove else "status") try: + cfg = load_config(args.config) + LOG_FILE = cfg["log_file"] or None + if args.apply: n_countries, n_entries = apply_geoblock(cfg, dry_run=args.dry_run) log_result("OK", action, countries=n_countries, entries=n_entries,