Namen 'ARIA' aus Dateinamen und Texten entfernen (nur noch 'geoblock')
This commit is contained in:
ARIA
Executable
+251
@@ -0,0 +1,251 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
geoblock_testserver.py — kleiner Test-Webserver fuer 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 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 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 geoblock.ini eintragen (z.B. den eigenen
|
||||
Ländercode DE), 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 geoblock_testserver.py --config geoblock.ini
|
||||
-> Port/Bind-Adresse/Log-Pfad kommen aus dem Abschnitt [testserver]
|
||||
der .ini (siehe geoblock.ini im gleichen Repo)
|
||||
|
||||
python3 geoblock_testserver.py --port 8899 --bind 0.0.0.0 \
|
||||
--log-file /var/log/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 geoblock-testserver.service im gleichen Repo sowie
|
||||
install_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 geoblock.ini. Fehlt der Abschnitt
|
||||
komplett, werden die Defaults benutzt (die .ini ist fuer den Testserver
|
||||
also optional, im Gegensatz zu 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>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 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 = "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 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()
|
||||
Reference in New Issue
Block a user