252 lines
9.1 KiB
Python
Executable File
252 lines
9.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
aria_geoblock_testserver.py — kleiner Test-Webserver fuer aria-geoblock
|
|
|
|
ZWECK:
|
|
Ein minimaler HTTP-Server (nur Python-Stdlib, keine Abhaengigkeiten),
|
|
gegen den man von AUSSEN (Handy im Mobilfunknetz, VPN mit anderem Land,
|
|
Freund im Ausland, ...) connecten kann, um zu pruefen ob die Geoblocking-
|
|
Regeln (ipset/iptables aus aria_geoblock.py) wirklich greifen.
|
|
|
|
Die Logik ist bewusst simpel: der Server selbst weiss NICHTS von Laendern
|
|
oder Blocklisten. Er beantwortet einfach jeden Request, der ihn erreicht,
|
|
mit einer kleinen HTML-Seite (Client-IP, Zeitstempel, User-Agent) und
|
|
loggt den Request in eine Datei.
|
|
|
|
Der eigentliche TEST ist der Netzwerkeffekt: iptables droppt Pakete
|
|
geblockter Laender bereits VOR dem TCP-Handshake. Kommt eine Anfrage aus
|
|
einem geblockten Land, sieht dieser Server sie also gar nicht erst -
|
|
der Client bekommt stattdessen ein Timeout. Aus einem NICHT geblockten
|
|
Land/Netz laedt die Seite ganz normal.
|
|
|
|
Testablauf (siehe auch README.md):
|
|
1. Testserver starten (systemd-Service aria-geoblock-testserver, oder
|
|
manuell: siehe unten).
|
|
2. Von einem Client aus einem ERLAUBTEN Land/Netz aufrufen:
|
|
curl http://<host>:<port>/
|
|
-> sollte die Test-Seite liefern.
|
|
3. Ziel-Land testweise in aria_geoblock.ini eintragen (z.B. den eigenen
|
|
Ländercode DE), aria_geoblock.py --apply ausfuehren, dann von einem
|
|
Client MIT dieser Landes-IP erneut connecten
|
|
-> sollte jetzt in ein Timeout laufen (kein TCP-Connect möglich).
|
|
4. Land wieder aus der .ini entfernen, --apply, Zugriff geht wieder.
|
|
|
|
BENUTZUNG:
|
|
python3 aria_geoblock_testserver.py --config aria_geoblock.ini
|
|
-> Port/Bind-Adresse/Log-Pfad kommen aus dem Abschnitt [testserver]
|
|
der .ini (siehe aria_geoblock.ini im gleichen Repo)
|
|
|
|
python3 aria_geoblock_testserver.py --port 8899 --bind 0.0.0.0 \
|
|
--log-file /var/log/aria_geoblock/testserver.log
|
|
-> alternativ komplett ohne .ini, per CLI-Argumenten
|
|
|
|
CLI-Argumente ueberschreiben, falls angegeben, die Werte aus der .ini.
|
|
|
|
LOGGING:
|
|
Jeder eingehende Request wird als eine Zeile geloggt (Zeitstempel,
|
|
Client-IP, Methode, Pfad, User-Agent, Statuscode). Ohne --log-file/
|
|
ohne log_file in der .ini wird nur nach stderr (systemd-Journal)
|
|
geloggt.
|
|
|
|
AUTOSTART:
|
|
Siehe aria-geoblock-testserver.service im gleichen Repo sowie
|
|
install_aria_geoblock.sh, das den Service optional mit einrichtet.
|
|
"""
|
|
|
|
import argparse
|
|
import configparser
|
|
import datetime
|
|
import html
|
|
import ipaddress
|
|
import socket
|
|
import sys
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
DEFAULT_PORT = 8899
|
|
DEFAULT_BIND = "0.0.0.0"
|
|
|
|
# Wird in main() aus CLI/.ini gesetzt. 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-testserver] WARNUNG: konnte Log-Datei nicht schreiben "
|
|
f"({LOG_FILE}): {e}", file=sys.stderr)
|
|
|
|
|
|
def log(msg):
|
|
line = f"[geoblock-testserver] {msg}"
|
|
print(line, file=sys.stderr, flush=True)
|
|
_write_log_file(line)
|
|
|
|
|
|
def load_config(path):
|
|
"""Liest [testserver] aus der aria_geoblock.ini. Fehlt der Abschnitt
|
|
komplett, werden die Defaults benutzt (die .ini ist fuer den Testserver
|
|
also optional, im Gegensatz zu aria_geoblock.py)."""
|
|
cfg = configparser.ConfigParser()
|
|
read_ok = cfg.read(path)
|
|
if not read_ok:
|
|
raise FileNotFoundError(f"Konnte .ini nicht lesen: {path}")
|
|
|
|
if "testserver" not in cfg:
|
|
log("Abschnitt [testserver] nicht in der .ini gefunden - benutze Defaults "
|
|
f"(port={DEFAULT_PORT}, bind={DEFAULT_BIND}, kein log_file)")
|
|
return {"port": DEFAULT_PORT, "bind": DEFAULT_BIND, "log_file": ""}
|
|
|
|
sec = cfg["testserver"]
|
|
return {
|
|
"port": sec.getint("port", fallback=DEFAULT_PORT),
|
|
"bind": sec.get("bind", fallback=DEFAULT_BIND).strip(),
|
|
"log_file": sec.get("log_file", fallback="").strip(),
|
|
}
|
|
|
|
|
|
def _client_ip(handler):
|
|
"""Bevorzugt X-Forwarded-For (falls hinter einem Reverse-Proxy), sonst
|
|
die direkte Peer-IP der TCP-Verbindung."""
|
|
xff = handler.headers.get("X-Forwarded-For")
|
|
if xff:
|
|
# erster Eintrag = urspruenglicher Client
|
|
candidate = xff.split(",")[0].strip()
|
|
try:
|
|
ipaddress.ip_address(candidate)
|
|
return candidate
|
|
except ValueError:
|
|
pass
|
|
return handler.client_address[0]
|
|
|
|
|
|
PAGE_TMPL = """<!DOCTYPE html>
|
|
<html lang="de">
|
|
<head>
|
|
<meta charset="utf-8">
|
|
<title>ARIA Geoblock — Verbindungstest</title>
|
|
<style>
|
|
body {{ font-family: -apple-system, Segoe UI, Roboto, sans-serif;
|
|
background: #0b1220; color: #e6ebf5; margin: 0;
|
|
display: flex; align-items: center; justify-content: center;
|
|
min-height: 100vh; }}
|
|
.card {{ background: #131c2e; border: 1px solid #24304a; border-radius: 12px;
|
|
padding: 2rem 2.5rem; max-width: 520px; box-shadow: 0 8px 24px rgba(0,0,0,.4); }}
|
|
h1 {{ font-size: 1.3rem; margin-top: 0; color: #6ee7a8; }}
|
|
table {{ width: 100%; border-collapse: collapse; margin-top: 1rem; }}
|
|
td {{ padding: .4rem 0; border-bottom: 1px solid #24304a; font-size: .92rem; }}
|
|
td.k {{ color: #8fa1c7; width: 40%; }}
|
|
.ok {{ color: #6ee7a8; font-weight: 600; }}
|
|
p.hint {{ color: #8fa1c7; font-size: .82rem; margin-top: 1.2rem; line-height: 1.4; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<div class="card">
|
|
<h1>✓ Verbindung erfolgreich</h1>
|
|
<p class="ok">Diese Anfrage kam beim Server an — die Quelle ist also NICHT geblockt.</p>
|
|
<table>
|
|
<tr><td class="k">Client-IP</td><td>{client_ip}</td></tr>
|
|
<tr><td class="k">Zeitstempel</td><td>{ts}</td></tr>
|
|
<tr><td class="k">Methode / Pfad</td><td>{method} {path}</td></tr>
|
|
<tr><td class="k">User-Agent</td><td>{ua}</td></tr>
|
|
<tr><td class="k">Host-Header</td><td>{host}</td></tr>
|
|
</table>
|
|
<p class="hint">Kommt eine Anfrage aus einem in aria_geoblock.ini geblockten Land,
|
|
sieht dieser Server sie gar nicht erst — iptables droppt sie vorher.
|
|
Der Client bekommt dann ein Timeout statt dieser Seite.</p>
|
|
</div>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
|
|
class TestHandler(BaseHTTPRequestHandler):
|
|
server_version = "ARIA-Geoblock-Testserver/1.0"
|
|
|
|
def log_message(self, format, *args):
|
|
# Default-Logging von BaseHTTPRequestHandler unterdruecken,
|
|
# wir loggen selbst strukturiert in do_GET/do_HEAD.
|
|
pass
|
|
|
|
def _handle(self, send_body):
|
|
client_ip = _client_ip(self)
|
|
ua = self.headers.get("User-Agent", "-")
|
|
host_hdr = self.headers.get("Host", "-")
|
|
log(f'request client_ip={client_ip} method={self.command} path={self.path} '
|
|
f'host="{host_hdr}" ua="{ua}"')
|
|
|
|
body = PAGE_TMPL.format(
|
|
client_ip=html.escape(client_ip),
|
|
ts=html.escape(_timestamp()),
|
|
method=html.escape(self.command),
|
|
path=html.escape(self.path),
|
|
ua=html.escape(ua),
|
|
host=html.escape(host_hdr),
|
|
).encode("utf-8")
|
|
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/html; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.end_headers()
|
|
if send_body:
|
|
self.wfile.write(body)
|
|
|
|
def do_GET(self):
|
|
self._handle(send_body=True)
|
|
|
|
def do_HEAD(self):
|
|
self._handle(send_body=False)
|
|
|
|
|
|
class ThreadingHTTPServerV6Fallback(ThreadingHTTPServer):
|
|
"""Erlaubt Bind auf IPv6-Adressen (z.B. '::'), faellt aber sauber
|
|
auf IPv4 zurueck wenn eine IPv4-Adresse konfiguriert ist."""
|
|
def server_bind(self):
|
|
if ":" in self.server_address[0]:
|
|
self.address_family = socket.AF_INET6
|
|
super().server_bind()
|
|
|
|
|
|
def main():
|
|
global LOG_FILE
|
|
|
|
parser = argparse.ArgumentParser(
|
|
description="Kleiner Test-Webserver um Geoblocking-Regeln von aussen zu pruefen")
|
|
parser.add_argument("--config", help="Pfad zur aria_geoblock.ini (Abschnitt [testserver])")
|
|
parser.add_argument("--port", type=int, help="TCP-Port (ueberschreibt .ini)")
|
|
parser.add_argument("--bind", help="Bind-Adresse (ueberschreibt .ini)")
|
|
parser.add_argument("--log-file", help="Pfad zur Request-Log-Datei (ueberschreibt .ini)")
|
|
args = parser.parse_args()
|
|
|
|
if args.config:
|
|
cfg = load_config(args.config)
|
|
else:
|
|
cfg = {"port": DEFAULT_PORT, "bind": DEFAULT_BIND, "log_file": ""}
|
|
|
|
port = args.port if args.port is not None else cfg["port"]
|
|
bind = args.bind if args.bind else cfg["bind"]
|
|
LOG_FILE = args.log_file if args.log_file else (cfg["log_file"] or None)
|
|
|
|
log(f"Starte Test-Webserver auf {bind}:{port} "
|
|
f"(log_file={'aus' if not LOG_FILE else LOG_FILE})")
|
|
|
|
httpd = ThreadingHTTPServerV6Fallback((bind, port), TestHandler)
|
|
try:
|
|
httpd.serve_forever()
|
|
except KeyboardInterrupt:
|
|
log("Beende (Strg+C).")
|
|
finally:
|
|
httpd.server_close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|