Aria-Software/geoblocking-python-script
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Port-/Protokoll-Einschraenkung fuers Geoblocking hinzufuegen (Selbstaussperr-Schutz beim Testen)

Browse Source
This commit is contained in:
ARIA
2026-07-21 13:31:47 +02:00
parent a608af9a27
commit 7d8aeabbd2
2 changed files with 167 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+40
View File
@@ -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`):
aria_geoblock.py
+127 -36
View File
@@ -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 <chain> -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,