duffyduckandClaude Opus 4.8
af2a75c443
purge_geoblock_rules() scannt jetzt die konfigurierten Chains PLUS die Standard- Filter-Chains INPUT/FORWARD/OUTPUT. Damit werden verwaiste geoblock-Regeln auch dann entfernt, wenn 'chain' zwischen zwei Laeufen geaendert wurde (z.B. INPUT -> FORWARD) -- vorher blieben die alten Regeln der nicht mehr konfigurierten Chain stehen. Es werden nur Regeln geloescht, die auf das eigene ipset verweisen. README: Abschnitt "Manuelle Nutzung" um die Aufraeum-Semantik ergaenzt (jeder --apply/Service-Restart entfernt alle geoblock-Regeln und setzt sie frisch, inkl. Reste aus geaenderter ports/protocol/chain-Config). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
576 lines
25 KiB
Python
Executable File
576 lines
25 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
geoblock.py — Geoblocking via ipset + iptables
|
|
|
|
Liest eine .ini-Datei mit einer Laenderliste, laedt die zugehoerigen
|
|
IP-Bereiche (CIDR) kostenlos von ipdeny.com (kein API-Key noetig) und
|
|
blockt sie am Linux-Kernel-Paketfilter (iptables + ipset).
|
|
|
|
WARUM ipset+iptables statt "Verbindung live pruefen":
|
|
- Live pro Connection ein GeoIP-Lookup zu machen ist langsam, braucht
|
|
eine (meist kostenpflichtige/registrierungspflichtige) GeoIP-DB und
|
|
blockt den Traffic erst NACH dem TCP-Handshake.
|
|
- ipset haelt die komplette IP-Range-Liste jedes Landes im Kernel vor;
|
|
iptables matched dagegen in O(1)/sehr schnell und droppt Pakete
|
|
schon auf Netzwerk-Ebene, bevor sie den Server ueberhaupt erreichen.
|
|
- Das ist der Standard-Ansatz den auch fail2ban/CSF-Firewalls nutzen.
|
|
|
|
VORAUSSETZUNGEN (auf dem Ziel-Host, NICHT im Container):
|
|
- root-Rechte (iptables/ipset aendern den Kernel-Netfilter)
|
|
- Pakete installiert: iptables, ipset
|
|
Debian/Ubuntu: apt install iptables ipset
|
|
- Internetzugang zum Laden der Zonefiles von ipdeny.com
|
|
(falls der Host selbst keine direkte Verbindung hat: 'proxy' in der
|
|
.ini setzen -- siehe README)
|
|
|
|
BENUTZUNG:
|
|
sudo python3 geoblock.py --config geoblock.ini --apply
|
|
-> laedt IP-Ranges, befuellt das ipset, setzt die iptables-Regel
|
|
|
|
sudo python3 geoblock.py --config geoblock.ini --apply --dry-run
|
|
-> zeigt nur an was gemacht wuerde, aendert nichts
|
|
|
|
sudo python3 geoblock.py --config geoblock.ini --remove
|
|
-> entfernt die iptables-Regel und loescht das ipset wieder
|
|
|
|
sudo python3 geoblock.py --config geoblock.ini --status
|
|
-> zeigt aktuellen Stand (Set vorhanden? wie viele Eintraege? Regel aktiv?)
|
|
|
|
.ini-Format siehe mitgelieferte geoblock.ini (Beispiel).
|
|
|
|
LOGGING:
|
|
Jeder Lauf schreibt nach stderr (systemd-Journal). Zusaetzlich kann in
|
|
der .ini unter "log_file" ein Pfad angegeben werden — dort wird bei
|
|
jedem Lauf eine Zeile angehaengt, inkl. einer maschinenlesbaren
|
|
"RESULT ..."-Zeile (status=OK/ERROR, Anzahl Laender/IP-Bereiche). Genau
|
|
diese Zeile wertet das mitgelieferte checkmk-Plugin aus.
|
|
|
|
AUTOMATISCH AKTUELL HALTEN (Systemstart + taeglich):
|
|
Siehe systemd/geoblock.service + geoblock.timer sowie
|
|
install_geoblock.sh im gleichen Repo — das richtet Timer fuer
|
|
"beim Booten" + "einmal taeglich" automatisch ein.
|
|
--apply ist idempotent: es flusht das bestehende Set und befuellt es neu und
|
|
entfernt vorher ALLE bestehenden geoblock-Regeln der Chain (auch verwaiste aus
|
|
frueheren Laeufen mit anderer ini), bevor es die Regeln frisch setzt. Die
|
|
DROP-/LOG-Regeln werden dabei per 'iptables -I' an den ANFANG der Chain gesetzt,
|
|
damit sie vor etwaigen ACCEPT-Regeln greifen (wichtig auf Hosts mit gemanagter
|
|
Firewall, die eigene ACCEPT-Regeln/Sub-Chains mitbringt).
|
|
|
|
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 geoblock.ini.
|
|
"""
|
|
|
|
import argparse
|
|
import configparser
|
|
import datetime
|
|
import ipaddress
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
ZONE_URL_TMPL = "https://www.ipdeny.com/ipblocks/data/countries/{cc}.zone"
|
|
|
|
# Standard-Filter-Chains, die beim Aufraeumen IMMER mitgescannt werden -- auch
|
|
# wenn sie gerade nicht in 'chain' konfiguriert sind. So werden verwaiste
|
|
# geoblock-Regeln entfernt, wenn 'chain' zwischen zwei Laeufen geaendert wurde
|
|
# (z.B. von INPUT auf FORWARD umgestellt). Rein additiv zum Scan: es werden nur
|
|
# Regeln geloescht, die auf UNSER ipset verweisen -- fremde Regeln bleiben unberuehrt.
|
|
STANDARD_FILTER_CHAINS = ("INPUT", "FORWARD", "OUTPUT")
|
|
|
|
# Wird in main() aus der .ini gesetzt (log_file). None = keine Datei-Logs.
|
|
LOG_FILE = None
|
|
|
|
|
|
def _timestamp():
|
|
return datetime.datetime.now().astimezone().isoformat(timespec="seconds")
|
|
|
|
|
|
def _write_log_file(line):
|
|
if not LOG_FILE:
|
|
return
|
|
try:
|
|
with open(LOG_FILE, "a", encoding="utf-8") as f:
|
|
f.write(f"{_timestamp()} {line}\n")
|
|
except OSError as e:
|
|
print(f"[geoblock] WARNUNG: konnte Log-Datei nicht schreiben ({LOG_FILE}): {e}",
|
|
file=sys.stderr)
|
|
|
|
|
|
def log(msg):
|
|
line = f"[geoblock] {msg}"
|
|
print(line, file=sys.stderr)
|
|
_write_log_file(line)
|
|
|
|
|
|
def log_result(status, action, countries=None, entries=None, msg=""):
|
|
"""Schreibt eine strukturierte, maschinenlesbare Ergebniszeile.
|
|
|
|
Wird vom checkmk-Plugin (geoblock_checkmk) ausgewertet, um den
|
|
Zustand des letzten Laufs zu beurteilen (OK/WARN/CRIT/UNKNOWN).
|
|
"""
|
|
parts = [f"RESULT ts={_timestamp()}", f"status={status}", f"action={action}"]
|
|
if countries is not None:
|
|
parts.append(f"countries={countries}")
|
|
if entries is not None:
|
|
parts.append(f"entries={entries}")
|
|
safe_msg = msg.replace('"', "'").replace("\n", " ")
|
|
parts.append(f'msg="{safe_msg}"')
|
|
line = "[geoblock] " + " ".join(parts)
|
|
print(line, file=sys.stderr)
|
|
_write_log_file(line)
|
|
|
|
|
|
def run(cmd, dry_run=False, check=True):
|
|
"""Fuehrt einen Shell-Befehl (als Liste) aus. Bei dry_run nur anzeigen."""
|
|
printable = " ".join(cmd)
|
|
if dry_run:
|
|
log(f"DRY-RUN: {printable}")
|
|
return None
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
if check and result.returncode != 0:
|
|
log(f"FEHLER bei: {printable}")
|
|
log(f"stderr: {result.stderr.strip()}")
|
|
raise RuntimeError(f"Befehl fehlgeschlagen: {printable}")
|
|
return result
|
|
|
|
|
|
def load_config(path):
|
|
cfg = configparser.ConfigParser()
|
|
read_ok = cfg.read(path)
|
|
if not read_ok:
|
|
raise FileNotFoundError(f"Konnte .ini nicht lesen: {path}")
|
|
if "geoblock" not in cfg:
|
|
raise ValueError("Abschnitt [geoblock] fehlt in der .ini-Datei")
|
|
|
|
sec = cfg["geoblock"]
|
|
countries_raw = sec.get("countries", "")
|
|
countries = [c.strip().lower() for c in countries_raw.split(",") if c.strip()]
|
|
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)")
|
|
|
|
# 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,
|
|
"chains": chains,
|
|
"interface": sec.get("interface", "").strip(),
|
|
"log": sec.getboolean("log", fallback=False),
|
|
"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,
|
|
"proxy": sec.get("proxy", "").strip(),
|
|
}
|
|
|
|
|
|
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 _build_url_opener(proxy):
|
|
"""Baut einen urllib-Opener. Mit 'proxy' gesetzt (z.B. 'http://host:3128' oder
|
|
'http://user:pass@host:3128') wird dieser fuer http UND https genutzt -- fuer
|
|
Server, die selbst keine direkte Internetverbindung haben und nur ueber
|
|
einen Vorgelagerten Proxy raus koennen."""
|
|
if not proxy:
|
|
return urllib.request.build_opener()
|
|
handler = urllib.request.ProxyHandler({"http": proxy, "https": proxy})
|
|
return urllib.request.build_opener(handler)
|
|
|
|
|
|
def fetch_country_cidrs(country_code, proxy=""):
|
|
"""Laedt die CIDR-Liste eines Landes von ipdeny.com. Gibt Liste von Strings zurueck.
|
|
|
|
'proxy' (optional): HTTP(S)-Proxy-URL, falls der Server selbst keine direkte
|
|
Internetverbindung hat und ipdeny.com nur ueber einen Proxy erreichen kann.
|
|
"""
|
|
url = ZONE_URL_TMPL.format(cc=country_code)
|
|
if proxy:
|
|
log(f"Lade IP-Ranges fuer '{country_code}' von {url} (ueber Proxy {proxy})")
|
|
else:
|
|
log(f"Lade IP-Ranges fuer '{country_code}' von {url}")
|
|
opener = _build_url_opener(proxy)
|
|
try:
|
|
with opener.open(url, timeout=20) as resp:
|
|
text = resp.read().decode("utf-8", errors="ignore")
|
|
except urllib.error.HTTPError as e:
|
|
raise RuntimeError(
|
|
f"Laendercode '{country_code}' unbekannt oder ipdeny nicht erreichbar "
|
|
f"(HTTP {e.code}). Pruefe den ISO-3166-1-alpha-2 Code."
|
|
)
|
|
except urllib.error.URLError as e:
|
|
raise RuntimeError(f"Konnte {url} nicht laden: {e}")
|
|
|
|
cidrs = []
|
|
for line in text.splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
try:
|
|
ipaddress.ip_network(line) # Validierung
|
|
except ValueError:
|
|
log(f" ignoriere ungueltige Zeile: {line}")
|
|
continue
|
|
cidrs.append(line)
|
|
log(f" -> {len(cidrs)} IP-Bereiche fuer '{country_code}'")
|
|
return cidrs
|
|
|
|
|
|
def ipset_exists(name):
|
|
result = subprocess.run(["ipset", "list", "-n"], capture_output=True, text=True)
|
|
return name in result.stdout.splitlines()
|
|
|
|
|
|
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 += _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
|
|
result = subprocess.run(check_cmd, capture_output=True, text=True)
|
|
return result.returncode == 0
|
|
|
|
|
|
def _list_chain_rules(chain):
|
|
"""Gibt die aktuellen Regeln einer iptables-Chain als '-A ...'-Zeilen zurueck
|
|
(Format von 'iptables -S <chain>', eine Zeile pro Regel)."""
|
|
result = subprocess.run(["iptables", "-S", chain], capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
return []
|
|
return [ln for ln in result.stdout.splitlines() if ln.startswith("-A ")]
|
|
|
|
|
|
def purge_geoblock_rules(cfg, dry_run=False):
|
|
"""Loescht in allen konfigurierten Chains PLUS den Standard-Filter-Chains
|
|
(INPUT/FORWARD/OUTPUT) 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 sauber entfernt: andere Ports/Protokolle (kennt build_rule_variants(cfg)
|
|
nicht mehr) UND andere Chains (z.B. nach Umstellung von INPUT auf FORWARD --
|
|
die verwaisten INPUT-Regeln werden dank Standard-Chain-Scan trotzdem gefunden).
|
|
Wir rekonstruieren das Loesch-Kommando direkt aus 'iptables -S' (per shlex,
|
|
damit z.B. das --log-prefix mit Leerzeichen erhalten bleibt), statt die Args
|
|
selbst nachzubauen -- so passt das -D immer exakt zur real existierenden Regel.
|
|
"""
|
|
name = cfg["ipset_name"]
|
|
needle = f"--match-set {name} src"
|
|
# Konfigurierte Chains PLUS die Standard-Filter-Chains scannen (dedupliziert,
|
|
# Reihenfolge erhalten). Faengt auch verwaiste Regeln ab, wenn 'chain' seit dem
|
|
# letzten Lauf geaendert wurde (z.B. INPUT -> FORWARD): die alten INPUT-Regeln
|
|
# werden trotzdem gefunden und entfernt.
|
|
scan_chains = list(dict.fromkeys(list(cfg["chains"]) + list(STANDARD_FILTER_CHAINS)))
|
|
removed = 0
|
|
for chain in scan_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 gefunden (gescannt: {', '.join(scan_chains)})")
|
|
return removed
|
|
|
|
|
|
def set_iptables_rules(cfg, dry_run=False):
|
|
"""Setzt die LOG-/DROP-Regel(n) mit -I an den ANFANG der Chain, damit sie VOR
|
|
etwaigen ACCEPT-Regeln greifen. Sonst akzeptiert eine weiter oben stehende
|
|
ACCEPT-Regel das Paket und eine ans Ende (-A) gehaengte DROP-Regel kommt nie
|
|
dran -- typisch auf Hosts mit gemanagter Firewall (Shorewall/firewalld/UFW/
|
|
UTM-Appliances mit eigenen Sub-Chains).
|
|
|
|
Setzt voraus, dass das ipset bereits existiert/befuellt ist UND die Chain
|
|
vorher per purge_geoblock_rules() bereinigt wurde (sonst entstehen Dubletten).
|
|
Pro Variante wird DROP zuerst an den Anfang gesetzt, dann LOG davor -> die
|
|
finale Reihenfolge in der Chain ist LOG vor DROP (erst loggen, dann droppen).
|
|
"""
|
|
name = cfg["ipset_name"]
|
|
set_match = ["-m", "set", "--match-set", name, "src"]
|
|
iface_opt = ["-i", cfg["interface"]] if cfg["interface"] else []
|
|
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", 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", 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 _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)
|
|
"""
|
|
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).
|
|
|
|
Gedacht fuer einen haeufigen Timer auf Hosts mit gemanagter Firewall, die die
|
|
Chain periodisch neu baut und dabei unsere Regeln nach unten schiebt oder
|
|
entfernt. Fehlt das ipset komplett, wird NICHT eingegriffen -- dann muss erst
|
|
ein voller --apply laufen, der das Set befuellt (der taegliche Timer erledigt das).
|
|
|
|
Rueckgabe: True, wenn korrigiert wurde; False, wenn nichts zu tun war.
|
|
"""
|
|
name = cfg["ipset_name"]
|
|
if not ipset_exists(name):
|
|
log(f"ipset '{name}' existiert nicht — reassert uebersprungen (erst voller --apply noetig).")
|
|
return False
|
|
|
|
on_top, present = geoblock_rules_on_top(cfg)
|
|
if on_top:
|
|
log("geoblock-Regeln stehen bereits am Anfang der Chain — nichts zu tun.")
|
|
return False
|
|
|
|
if present:
|
|
log("geoblock-Regeln nach unten gerutscht — setze sie neu an den Anfang der Chain.")
|
|
else:
|
|
log("geoblock-Regeln fehlen in der Chain — setze sie neu an den Anfang.")
|
|
purge_geoblock_rules(cfg, dry_run=dry_run) # evtl. verrutschte Reste einsammeln
|
|
set_iptables_rules(cfg, dry_run=dry_run)
|
|
return True
|
|
|
|
|
|
def apply_geoblock(cfg, dry_run=False):
|
|
name = cfg["ipset_name"]
|
|
|
|
# 1) ipset anlegen falls noetig
|
|
if dry_run or not ipset_exists(name):
|
|
run(["ipset", "create", name, "hash:net", "-exist"], dry_run=dry_run)
|
|
else:
|
|
log(f"ipset '{name}' existiert bereits, wird neu befuellt")
|
|
|
|
# 2) alle CIDRs aller konfigurierten Laender sammeln
|
|
all_cidrs = []
|
|
for cc in cfg["countries"]:
|
|
all_cidrs.extend(fetch_country_cidrs(cc, proxy=cfg.get("proxy", "")))
|
|
|
|
if not all_cidrs and not dry_run:
|
|
raise RuntimeError("Keine IP-Bereiche geladen — Abbruch, um kein leeres Set zu aktivieren")
|
|
|
|
# 3) Set atomar neu befuellen: temp-Set bauen, dann swap (kein Traffic-Loch)
|
|
tmp_name = f"{name}_tmp"
|
|
run(["ipset", "create", tmp_name, "hash:net", "-exist"], dry_run=dry_run)
|
|
run(["ipset", "flush", tmp_name], dry_run=dry_run)
|
|
for cidr in all_cidrs:
|
|
run(["ipset", "add", tmp_name, cidr, "-exist"], dry_run=dry_run, check=False)
|
|
run(["ipset", "create", name, "hash:net", "-exist"], dry_run=dry_run)
|
|
run(["ipset", "swap", tmp_name, name], dry_run=dry_run)
|
|
run(["ipset", "destroy", tmp_name], dry_run=dry_run, check=False)
|
|
|
|
log(f"ipset '{name}' befuellt mit {len(all_cidrs)} Bereichen "
|
|
f"aus Laendern: {', '.join(c.upper() for c in cfg['countries'])}")
|
|
|
|
# 4) Vor dem Neu-Setzen ALLE bestehenden geoblock-Regeln der Chain entfernen
|
|
# (auch verwaiste aus frueheren Laeufen mit anderer ini, z.B. andere Ports).
|
|
# Erst JETZT (nach dem Befuellen des Sets), damit waehrend des ipdeny-Downloads
|
|
# der bestehende Block aktiv bleibt und kein Traffic-Loch entsteht.
|
|
purge_geoblock_rules(cfg, dry_run=dry_run)
|
|
|
|
# 5) LOG-/DROP-Regel(n) frisch an den ANFANG der Chain setzen (siehe
|
|
# set_iptables_rules -- gleiche Logik nutzt auch --reassert). Der purge oben
|
|
# hat die Chain vorher von alten geoblock-Regeln befreit, daher keine Dubletten.
|
|
set_iptables_rules(cfg, dry_run=dry_run)
|
|
|
|
log("Fertig. Hinweis: Regeln sind NICHT reboot-persistent — "
|
|
"fuer Persistenz z.B. 'iptables-persistent' bzw. 'netfilter-persistent save' nutzen, "
|
|
"und das ipset per systemd-Unit / rc.local vor dem iptables-Restore neu befuellen.")
|
|
|
|
return len(cfg["countries"]), len(all_cidrs)
|
|
|
|
|
|
def remove_geoblock(cfg, dry_run=False):
|
|
name = cfg["ipset_name"]
|
|
# Loescht ALLE geoblock-Regeln der Chain (auch verwaiste aus frueheren Laeufen
|
|
# mit anderer ini), nicht nur die aktuell konfigurierten Varianten -- so bleiben
|
|
# keine Reste stehen, egal wie sich 'ports'/'protocol' zwischenzeitlich geaendert haben.
|
|
purge_geoblock_rules(cfg, dry_run=dry_run)
|
|
run(["ipset", "destroy", name], dry_run=dry_run, check=False)
|
|
log(f"Geoblock-Regeln entfernt, ipset '{name}' geloescht.")
|
|
|
|
|
|
def show_status(cfg):
|
|
name = cfg["ipset_name"]
|
|
if ipset_exists(name):
|
|
result = subprocess.run(["ipset", "list", name, "-t"], capture_output=True, text=True)
|
|
print(result.stdout)
|
|
else:
|
|
print(f"ipset '{name}' existiert nicht (noch nicht aktiv).")
|
|
|
|
variants = build_rule_variants(cfg)
|
|
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 = _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"]:
|
|
print(f"Port-Einschraenkung: {', '.join(cfg['ports'])} ({'/'.join(cfg['protocols'])})")
|
|
else:
|
|
print("Port-Einschraenkung: keine (alle Ports/Protokolle werden geblockt)")
|
|
|
|
|
|
def main():
|
|
global LOG_FILE
|
|
|
|
parser = argparse.ArgumentParser(description="Geoblocking per Laenderliste (.ini) via ipset+iptables")
|
|
parser.add_argument("--config", required=True, help="Pfad zur .ini-Konfigurationsdatei")
|
|
group = parser.add_mutually_exclusive_group(required=True)
|
|
group.add_argument("--apply", action="store_true", help="Laender laden und Block-Regeln setzen/aktualisieren")
|
|
group.add_argument("--reassert", action="store_true",
|
|
help="Nur pruefen ob die Block-Regeln noch am Anfang der Chain stehen und sie "
|
|
"ggf. wieder nach oben setzen -- OHNE Zonefiles neu zu laden (schneller Watchdog "
|
|
"fuer gemanagte Firewalls, die die Chain neu bauen). Setzt ein bestehendes ipset voraus.")
|
|
group.add_argument("--remove", action="store_true", help="Block-Regeln und ipset wieder entfernen")
|
|
group.add_argument("--status", action="store_true", help="Aktuellen Zustand anzeigen")
|
|
parser.add_argument("--dry-run", action="store_true", help="Nur anzeigen was passieren wuerde, nichts aendern")
|
|
args = parser.parse_args()
|
|
|
|
action = ("apply" if args.apply else "reassert" if args.reassert
|
|
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,
|
|
msg=f"Geoblock aktualisiert ({n_countries} Laender, {n_entries} IP-Bereiche)")
|
|
elif args.reassert:
|
|
corrected = reassert_rules(cfg, dry_run=args.dry_run)
|
|
log_result("OK", action,
|
|
msg="Regeln neu an den Anfang gesetzt" if corrected else "Regeln bereits am Anfang, nichts zu tun")
|
|
elif args.remove:
|
|
remove_geoblock(cfg, dry_run=args.dry_run)
|
|
log_result("OK", action, msg="Geoblock-Regeln entfernt")
|
|
elif args.status:
|
|
show_status(cfg)
|
|
except Exception as e:
|
|
log_result("ERROR", action, msg=str(e))
|
|
log(f"FEHLER: {e}")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|