From 9270994e825c3ae6825eb0c36768cca105286bb7 Mon Sep 17 00:00:00 2001 From: duffyduck Date: Mon, 5 Oct 2026 14:34:44 +0200 Subject: [PATCH] =?UTF-8?q?Verbindungstest=20f=C3=BCr=20Proxys=20hinzuf?= =?UTF-8?q?=C3=BCgen?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Neuer Befehl 'svc-proxy test [DIENST]' sowie Taste 't' in der Liste und Schaltfläche/F3 im Formular (testet ungespeicherte Werte). - prüft Erreichbarkeit, Handshake und Anmeldung für SOCKS5/SOCKS4/HTTP(S) mit klaren Meldungen (Kennwort falsch, Proxy-Typ vertauscht, Port gesperrt) - baut Tunnel zu Testzielen auf und liest die Begrüßung (SMTP, POP3, IMAP, HTTP), TLS für 443/465/993/995 - --target HOST:PORT, --mail (SMTP 25/587/465, POP3S, IMAPS), --timeout - im proxychains-Modus zusätzlicher Test über die proxychains-Bibliothek mit der Konfiguration des Dienstes - Ad-hoc-Test mit --proxy ohne root und ohne Speichern - Hilfe und README ergänzt, Version 1.2.0 Co-Authored-By: Claude Opus 5.5 (1M context) --- README.md | 62 ++++++ svc-proxy | 565 +++++++++++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 619 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index a3d271d..3803dde 100644 --- a/README.md +++ b/README.md @@ -196,6 +196,7 @@ Markierungen: **G** = nutzt globalen Proxy, **E** = eigene Werte, | Entf / d | Proxy vom Dienst entfernen | | / | Filter (Name oder Beschreibung), Esc löscht den Filter | | c | nur konfigurierte Dienste anzeigen (an/aus) | +| t | Verbindungstest für den gewählten Dienst (ohne Proxy: globaler Proxy) | | r | gewählten Dienst sofort neu starten | | F5 | Dienstliste neu einlesen | | F1 / ? | Hilfe | @@ -209,6 +210,7 @@ Markierungen: **G** = nutzt globalen Proxy, **E** = eigene Werte, | ← → / Leertaste | Modus bzw. Proxy-Typ wählen (beim Typ gibt es zusätzlich „global“) | | Enter | nächstes Feld bzw. Schaltfläche auslösen | | F2 | Speichern | +| F3 / „Testen“ | Verbindungstest mit den eingegebenen, noch nicht gespeicherten Werten | | Strg+U | Feld leeren | | Esc | Abbrechen | @@ -236,6 +238,7 @@ svc-proxy [BEFEHL] [OPTIONEN] | `set DIENST... [FELDER] [--reset]` | Proxy für Dienst(e) setzen/ändern | | `unset DIENST...` / `unset --all` | Proxy entfernen | | `apply` | alle Drop-ins aus der Konfiguration neu erzeugen | +| `test [DIENST] [--target H:P] [--mail]` | Verbindungstest über den Proxy (siehe unten) | | `help`, `-h`, `--help` | Hilfe mit Beispielen | ### Felder @@ -287,6 +290,59 @@ praktisch zum Testen, z.B. zusammen mit `--no-reload`. --- +## Verbindungstest + +```bash +sudo svc-proxy test kerio-connect --mail +``` + +``` +Verbindungstest: Dienst kerio-connect.service + Proxy: socks5://max:****@192.168.1.50:1080 [proxychains] +[ OK ] Proxy erreichbar: 192.168.1.50:1080 (1 ms) +[FEHLER] gmail-smtp-in.l.google.com:25 → Proxy meldet: vom Regelwerk des Proxys nicht erlaubt (Port/Ziel gesperrt?) +[ OK ] smtp.gmail.com:587 → Antwort: 220 smtp.gmail.com ESMTP ... (46 ms) +[ OK ] smtp.gmail.com:465 → TLS ok, Antwort: 220 smtp.gmail.com ESMTP ... (61 ms) +[ OK ] pop.gmail.com:995 → TLS ok, Antwort: +OK Gpop ready for requests ... (67 ms) +[ OK ] imap.gmail.com:993 → TLS ok, Antwort: * OK Gimap ready for requests ... (105 ms) + Test über proxychains (so verbindet der Dienst): +[FEHLER] proxychains: gmail-smtp-in.l.google.com:25 → Verbindung abgelehnt +[ OK ] proxychains: smtp.gmail.com:587 → Antwort: 220 smtp.gmail.com ESMTP ... +... +Ergebnis: 2 Test(s) fehlgeschlagen. +``` + +Ablauf: + +1. **Proxy erreichbar?** TCP-Verbindung zum Proxy. +2. **Handshake und Anmeldung:** SOCKS5, SOCKS4 oder HTTP CONNECT (auch `https`-Proxy). + Verständliche Meldungen z.B. bei falschem Kennwort, fehlender Anmeldung, + falschem Proxy-Typ (socks/http vertauscht) oder einem vom Proxy gesperrten Port. +3. **Ziele:** Tunnel zu jedem Ziel aufbauen und die Begrüßung lesen. Das zeigt, + dass wirklich Daten fließen (SMTP `220`, POP3 `+OK`, IMAP `* OK`, HTTP-Status). + Die Ports 443, 465, 993 und 995 werden per TLS angesprochen (ohne + Zertifikatsprüfung, es geht nur um die Verbindung). +4. **Nur im Modus proxychains:** Derselbe Test läuft noch einmal als normales + Programm über die proxychains-Bibliothek mit der Konfiguration des Dienstes. + Damit ist der komplette Weg geprüft, den auch der Dienst nimmt. + +| Option | Bedeutung | +|---|---| +| `--target HOST:PORT` | Testziel, mehrfach möglich (Standard: `example.com:80`, `example.com:443`) | +| `--mail` | Mail-Ziele: SMTP 25/587/465, POP3S 995, IMAPS 993 (Gmail) | +| `--timeout SEK` | Zeitlimit pro Schritt (Standard 6 s) | +| `--proxy`, `--host`, … | Werte nur für den Test überschreiben, nichts wird gespeichert | + +Ohne `DIENST` wird der globale Proxy getestet. Mit `--proxy URL` (ohne DIENST) +läuft der Test auch ohne root und ohne gespeicherte Konfiguration. So lässt sich +ein Proxy prüfen, bevor man ihn einträgt. Rückgabewert: `0` = alles ok, +`1` = mindestens ein Test fehlgeschlagen, praktisch für Monitoring-Skripte. + +Hinweis zu Port 25: Viele Proxys und Provider sperren ausgehend Port 25. Der Test +zeigt genau das. Abhilfe ist ein SMTP-Relay (Smarthost) auf Port 587 oder 465. + +--- + ## Beispiele ```bash @@ -334,6 +390,12 @@ sudo svc-proxy set myapp --reset sudo svc-proxy unset myapp --restart sudo svc-proxy unset --all --no-restart +# Verbindungstests +sudo svc-proxy test # globaler Proxy +sudo svc-proxy test kerio-connect --mail # Mail-Ports über Kerios Proxy +sudo svc-proxy test kerio-connect --target mx.example.com:25 --target pop.example.com:995 +svc-proxy test --proxy socks5://192.0.2.10:1080 --mail # ad hoc, ohne Speichern, ohne root + # Anzeigen sudo svc-proxy list sudo svc-proxy list --all diff --git a/svc-proxy b/svc-proxy index 495a96e..dbe8d7c 100755 --- a/svc-proxy +++ b/svc-proxy @@ -8,6 +8,7 @@ Ausführliche Hilfe: svc-proxy --help """ import argparse +import base64 import curses import ipaddress import json @@ -16,13 +17,16 @@ import os import pwd import re import socket +import ssl +import struct import subprocess import sys import tempfile +import time from glob import glob as glob_paths from urllib.parse import quote, unquote, urlsplit -VERSION = "1.1.0" +VERSION = "1.2.0" PROG = "svc-proxy" DEFAULT_CONFIG_DIR = "/etc/svc-proxy" @@ -70,6 +74,29 @@ PROXYCHAINS_LOCALNETS = [ "[::]:53/0", ] +# Verbindungstest +TEST_TARGETS_WEB = ["example.com:80", "example.com:443"] +TEST_TARGETS_MAIL = [ + "gmail-smtp-in.l.google.com:25", # SMTP (MX-Zustellung) + "smtp.gmail.com:587", # SMTP Submission + "smtp.gmail.com:465", # SMTPS + "pop.gmail.com:995", # POP3S + "imap.gmail.com:993", # IMAPS +] +TEST_TIMEOUT = 6.0 +TLS_PORTS = {443, 465, 563, 636, 853, 993, 995, 8443} +HTTP_PORTS = {80, 443, 8080, 8443} +SOCKS5_ERRORS = { + 1: "allgemeiner Fehler des Proxys", + 2: "vom Regelwerk des Proxys nicht erlaubt (Port/Ziel gesperrt?)", + 3: "Netz nicht erreichbar", + 4: "Host nicht erreichbar", + 5: "Verbindung vom Ziel abgelehnt", + 6: "Zeitüberschreitung (TTL abgelaufen)", + 7: "Befehl nicht unterstützt", + 8: "Adresstyp nicht unterstützt", +} + UNIT_RE = re.compile(r"^[A-Za-z0-9:_.@\\-]+$") HOST_RE = re.compile(r"^[A-Za-z0-9._:-]+$") NO_PROXY_RE = re.compile(r"^[A-Za-z0-9.,:*/_\[\]-]*$") @@ -107,6 +134,8 @@ BEFEHLE unset DIENST... Proxy von Diensten entfernen unset --all Proxy von allen Diensten entfernen apply Alle Drop-ins aus der Konfiguration neu schreiben + test [DIENST] [ZIELE] Verbindungstest über den Proxy des Dienstes + (ohne DIENST: globaler Proxy, oder ad hoc mit --proxy) help Diese Hilfe anzeigen FELDER (für 'global' und 'set') @@ -139,6 +168,22 @@ VERERBUNG (Dienst <-> global) Ändert man den globalen Proxy, werden alle Dienste, die ihn (teilweise) erben, automatisch mit angepasst. +VERBINDUNGSTEST ('test') + Prüft der Reihe nach: + 1. Ist der Proxy erreichbar (TCP)? + 2. Klappt der Handshake inkl. Anmeldung (SOCKS5/SOCKS4/HTTP CONNECT)? + Klare Meldung bei falschem Kennwort, falschem Proxy-Typ, gesperrtem Port. + 3. Für jedes Ziel: Tunnel aufbauen und Begrüßung lesen (SMTP "220", + POP3 "+OK", IMAP "* OK", HTTP-Status). Ports 443/465/993/995 per TLS. + 4. Im Modus proxychains zusätzlich: derselbe Test über die proxychains- + Bibliothek mit der Konfiguration des Dienstes (so verbindet der Dienst). + --target HOST:PORT Testziel (mehrfach möglich). Standard: {web} + --mail Mail-Ziele testen: SMTP 25/587/465, POP3S 995, IMAPS 993 (Gmail) + --timeout SEK Zeitlimit pro Schritt (Standard: {timeout:g} s) + FELDER (--proxy, --host, ...) überschreiben die gespeicherten Werte nur für den + Test. Mit --proxy ohne DIENST geht der Test auch ohne root. + Rückgabewert: 0 = alles ok, 1 = mindestens ein Test fehlgeschlagen. + NEUSTART / RELOAD Nach jeder Änderung wird 'systemctl daemon-reload' ausgeführt. Die neuen Werte gelten aber erst nach einem Neustart des Dienstes. @@ -194,6 +239,12 @@ BEISPIELE # Proxy von einem Dienst entfernen sudo {prog} unset myapp --restart + # Verbindungstests + sudo {prog} test # globaler Proxy, Web-Ziele + sudo {prog} test kerio-connect --mail # Mail-Ports über Kerios Proxy + sudo {prog} test kerio-connect --target mx.example.com:25 --target pop.example.com:995 + {prog} test --proxy socks5://192.0.2.10:1080 --mail # ad hoc, ohne Speichern + # Übersicht und Details sudo {prog} list sudo {prog} show docker @@ -206,6 +257,7 @@ DATEIEN Dienstnamen ohne Endung erhalten automatisch '.service'. """.format(prog=PROG, version=VERSION, cfg=DEFAULT_CONFIG_DIR, + web=", ".join(TEST_TARGETS_WEB), timeout=TEST_TIMEOUT, sysd=DEFAULT_SYSTEMD_DIR, dropin=DROPIN_NAME) @@ -213,6 +265,10 @@ class ProxyError(Exception): pass +class ProxyFatal(ProxyError): + """Fehler, bei dem weitere Testziele sinnlos sind (falscher Typ, Anmeldung).""" + + # -------------------------------------------------------------------------- # Hilfsfunktionen # -------------------------------------------------------------------------- @@ -737,6 +793,316 @@ def list_system_services(): return units +# -------------------------------------------------------------------------- +# Verbindungstest +# -------------------------------------------------------------------------- + +def parse_target(text): + text = text.strip() + m = re.match(r"^\[([0-9A-Fa-f:.]+)\]:(\d+)$", text) or re.match(r"^([A-Za-z0-9._-]+):(\d+)$", text) + if not m or not 1 <= int(m.group(2)) <= 65535: + raise ProxyError("Ungültiges Ziel %r (Format: HOST:PORT)" % text) + return m.group(1), int(m.group(2)) + + +def format_target(host, port): + return "[%s]:%d" % (host, port) if ":" in host else "%s:%d" % (host, port) + + +def recv_exact(sock, n): + buf = b"" + while len(buf) < n: + chunk = sock.recv(n - len(buf)) + if not chunk: + raise ProxyError("Proxy hat die Verbindung unerwartet geschlossen") + buf += chunk + return buf + + +def resolve_local(host, ipv4_only=False): + family = socket.AF_INET if ipv4_only else 0 + try: + infos = socket.getaddrinfo(host, None, family, socket.SOCK_STREAM) + except socket.gaierror as e: + raise ProxyError("Ziel %s lokal nicht auflösbar (%s)" % (host, e.strerror)) + infos.sort(key=lambda i: i[0] != socket.AF_INET) + return infos[0][4][0] + + +def socks5_connect(sock, eff, host, port): + methods = b"\x00\x02" if eff["user"] else b"\x00" + sock.sendall(b"\x05" + bytes([len(methods)]) + methods) + try: + ver, method = recv_exact(sock, 2) + except socket.timeout: + raise ProxyFatal("Keine Antwort auf den SOCKS5-Handshake – ist der Proxy-Typ richtig (http statt socks5?)") + if ver != 5: + raise ProxyFatal("Keine SOCKS5-Antwort – ist der Proxy-Typ richtig (http statt socks5?)") + if method == 0xFF: + raise ProxyFatal("Proxy akzeptiert keine angebotene Anmeldemethode%s" + % ("" if eff["user"] else " – verlangt er Benutzer/Kennwort?")) + if method == 2: + if not eff["user"]: + raise ProxyFatal("Proxy verlangt Benutzer/Kennwort") + u, p = eff["user"].encode(), eff["password"].encode() + sock.sendall(b"\x01" + bytes([len(u)]) + u + bytes([len(p)]) + p) + if recv_exact(sock, 2)[1] != 0: + raise ProxyFatal("Anmeldung abgelehnt – Benutzer oder Kennwort falsch") + elif method != 0: + raise ProxyFatal("Proxy verlangt eine nicht unterstützte Anmeldemethode (%d)" % method) + + if eff["type"] == "socks5h": + name = host.encode("idna") + addr = b"\x03" + bytes([len(name)]) + name + else: + ip = resolve_local(host) + if ":" in ip: + addr = b"\x04" + socket.inet_pton(socket.AF_INET6, ip) + else: + addr = b"\x01" + socket.inet_aton(ip) + sock.sendall(b"\x05\x01\x00" + addr + struct.pack(">H", port)) + try: + _, rep, _, atyp = recv_exact(sock, 4) + except socket.timeout: + raise ProxyError("Proxy erreicht das Ziel nicht (Zeitüberschreitung) – Port/Ziel gesperrt?") + if rep != 0: + raise ProxyError("Proxy meldet: %s" % SOCKS5_ERRORS.get(rep, "Fehlercode %d" % rep)) + if atyp == 1: + recv_exact(sock, 4) + elif atyp == 3: + recv_exact(sock, recv_exact(sock, 1)[0]) + elif atyp == 4: + recv_exact(sock, 16) + recv_exact(sock, 2) + return b"" + + +def socks4_connect(sock, eff, host, port): + ip = resolve_local(host, ipv4_only=True) + sock.sendall(struct.pack(">BBH", 4, 1, port) + socket.inet_aton(ip) + eff["user"].encode() + b"\x00") + try: + reply = recv_exact(sock, 8) + except socket.timeout: + raise ProxyError("Keine Antwort vom Proxy – falscher Proxy-Typ oder Ziel nicht erreichbar?") + if reply[0] != 0: + raise ProxyFatal("Keine SOCKS4-Antwort – ist der Proxy-Typ richtig?") + if reply[1] != 0x5A: + reasons = {0x5B: "Anfrage abgelehnt oder Ziel nicht erreichbar", + 0x5C: "identd des Clients nicht erreichbar", + 0x5D: "identd-Benutzer passt nicht"} + raise ProxyError("Proxy meldet: %s" % reasons.get(reply[1], "Code %d" % reply[1])) + return b"" + + +def http_connect(sock, eff, host, port): + target = format_target(host, port) + req = "CONNECT %s HTTP/1.1\r\nHost: %s\r\nUser-Agent: %s/%s\r\n" % (target, target, PROG, VERSION) + if eff["user"]: + cred = base64.b64encode(("%s:%s" % (eff["user"], eff["password"])).encode()).decode() + req += "Proxy-Authorization: Basic %s\r\n" % cred + sock.sendall((req + "\r\n").encode()) + data = b"" + while b"\r\n\r\n" not in data: + try: + chunk = sock.recv(4096) + except socket.timeout: + raise ProxyError("Keine Antwort vom Proxy – falscher Proxy-Typ (socks statt http?) " + "oder Ziel nicht erreichbar") + if not chunk: + if not data: + raise ProxyFatal("Proxy hat ohne Antwort geschlossen – ist der Proxy-Typ richtig (socks statt http?)") + break + data += chunk + if len(data) > 65536: + raise ProxyFatal("Antwort des Proxys zu groß") + head, _, rest = data.partition(b"\r\n\r\n") + status = head.split(b"\r\n", 1)[0].decode("latin-1").strip() + parts = status.split(None, 2) + if len(parts) < 2 or not parts[0].startswith("HTTP/") or not parts[1].isdigit(): + raise ProxyFatal("Keine HTTP-Proxy-Antwort – ist der Proxy-Typ richtig?") + code = int(parts[1]) + if code == 407: + raise ProxyFatal("Anmeldung %s (407)" % ("abgelehnt – Benutzer/Kennwort falsch" if eff["user"] + else "erforderlich – Benutzer/Kennwort fehlt")) + if code != 200: + raise ProxyError("Proxy lehnt ab: %s" % status) + return rest + + +def proxy_connect(eff, host, port, timeout): + """Baut über den Proxy einen Tunnel zu host:port. Liefert (socket, schon empfangene Daten).""" + sock = socket.create_connection((eff["host"], int(eff["port"])), timeout) + try: + if eff["type"] == "https": + ctx = ssl.create_default_context() + sock = ctx.wrap_socket(sock, server_hostname=eff["host"]) + if eff["type"] in ("http", "https"): + rest = http_connect(sock, eff, host, port) + elif eff["type"] == "socks4": + rest = socks4_connect(sock, eff, host, port) + else: + rest = socks5_connect(sock, eff, host, port) + except BaseException: + sock.close() + raise + return sock, rest + + +def probe(sock, host, port, early=b"", timeout=TEST_TIMEOUT): + """Prüft, ob über die Verbindung echte Daten fließen. Liefert (ok, text).""" + sock.settimeout(timeout) + prefix = "" + if port in TLS_PORTS: + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE # nur Erreichbarkeit testen, keine Zertifikatsprüfung + try: + try: + ipaddress.ip_address(host) + is_ip = True + except ValueError: + is_ip = False + sock = ctx.wrap_socket(sock, server_hostname=None if is_ip else host) + except (ssl.SSLError, OSError) as e: + return False, "TLS-Handshake fehlgeschlagen (%s)" % (getattr(e, "reason", None) or e) + prefix = "TLS ok, " + early = b"" + if port in HTTP_PORTS: + sock.sendall(("HEAD / HTTP/1.0\r\nHost: %s\r\nUser-Agent: %s/%s\r\n\r\n" + % (host, PROG, VERSION)).encode()) + data = early + try: + while b"\n" not in data and len(data) < 1024: + chunk = sock.recv(1024) + if not chunk: + break + data += chunk + except socket.timeout: + pass + except OSError as e: + return False, prefix + "Verbindung abgebrochen (%s)" % e + finally: + try: + sock.close() + except OSError: + pass + if not data: + return True, prefix + "verbunden (keine Begrüßung innerhalb %d s)" % timeout + line = data.split(b"\n", 1)[0].decode("utf-8", "replace").strip() + return True, prefix + "Antwort: " + (line[:70] or "(leer)") + + +def probe_direct(targets, timeout): + """Wird im proxychains-Selbsttest im Kindprozess ausgeführt: normale Verbindungen + wie ein Programm ohne Proxy-Unterstützung – proxychains leitet sie um.""" + rc = 0 + for text in targets: + host, port = parse_target(text) + t0 = time.monotonic() + try: + sock = socket.create_connection((host, port), timeout) + ok, msg = probe(sock, host, port, timeout=timeout) + except (OSError, ProxyError) as e: + ok, msg = False, describe_error(e) + ms = (time.monotonic() - t0) * 1000 + print("%s\t%s → %s (%d ms)" % ("OK" if ok else "FAIL", format_target(host, port), msg, ms)) + rc |= 0 if ok else 1 + return rc + + +def describe_error(e): + if isinstance(e, ProxyError): + return str(e) + if isinstance(e, socket.timeout): + return "Zeitüberschreitung" + if isinstance(e, ConnectionRefusedError): + return "Verbindung abgelehnt" + if isinstance(e, ConnectionResetError): + return "Verbindung zurückgesetzt" + if isinstance(e, socket.gaierror): + return "Name nicht auflösbar (%s)" % e.strerror + if isinstance(e, ssl.SSLError): + return "TLS-Fehler (%s)" % (e.reason or e) + return getattr(e, "strerror", None) or str(e) + + +def proxychains_selftest(pc_conf, targets, timeout): + try: + lib = find_proxychains_lib() + except ProxyError as e: + return [("fail", "proxychains-Test: %s" % e)] + fd, path = tempfile.mkstemp(prefix="svc-proxy-test-", suffix=".conf") + try: + with os.fdopen(fd, "w") as f: + f.write(pc_conf) + env = dict(os.environ, LD_PRELOAD=lib, PROXYCHAINS_CONF_FILE=path, PROXYCHAINS_QUIET_MODE="1") + cmd = [sys.executable, os.path.abspath(__file__), "_probe", str(timeout)] + cmd += [format_target(h, p) for h, p in targets] + try: + r = subprocess.run(cmd, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + universal_newlines=True, timeout=timeout * (len(targets) + 1) + 10) + except subprocess.TimeoutExpired: + return [("fail", "proxychains-Test: Zeitüberschreitung")] + finally: + os.unlink(path) + results = [] + for line in r.stdout.splitlines(): + status, _, text = line.partition("\t") + results.append(("ok" if status == "OK" else "fail", "proxychains: " + text)) + if not results: + err = (r.stderr.strip().splitlines() or ["keine Ausgabe"])[-1] + results.append(("fail", "proxychains-Test fehlgeschlagen: %s" % err)) + return results + + +def run_proxy_test(eff, targets, cfg=None, unit=None, timeout=TEST_TIMEOUT): + """Testet den Proxy. Liefert Liste von (status, text), status: ok | fail | info.""" + results = [("info", "Proxy: %s%s" % (proxy_url(eff, mask=True), mode_suffix(eff)))] + t0 = time.monotonic() + try: + socket.create_connection((eff["host"], int(eff["port"])), timeout).close() + except socket.gaierror as e: + results.append(("fail", "Proxy-Host %s nicht auflösbar (%s)" % (eff["host"], e.strerror))) + return results + except ConnectionRefusedError: + results.append(("fail", "Proxy %s:%s lehnt die Verbindung ab – läuft dort ein Proxy auf diesem Port?" + % (eff["host"], eff["port"]))) + return results + except socket.timeout: + results.append(("fail", "Proxy %s:%s antwortet nicht (Zeitüberschreitung) – Firewall?" + % (eff["host"], eff["port"]))) + return results + except OSError as e: + results.append(("fail", "Proxy %s:%s nicht erreichbar: %s" % (eff["host"], eff["port"], describe_error(e)))) + return results + results.append(("ok", "Proxy erreichbar: %s:%s (%d ms)" + % (eff["host"], eff["port"], (time.monotonic() - t0) * 1000))) + + for host, port in targets: + t0 = time.monotonic() + label = format_target(host, port) + try: + sock, early = proxy_connect(eff, host, port, timeout) + ok, msg = probe(sock, host, port, early, timeout) + except ProxyFatal as e: + results.append(("fail", "%s → %s" % (label, e))) + return results + except (OSError, ProxyError) as e: + ok, msg = False, describe_error(e) + ms = (time.monotonic() - t0) * 1000 + results.append(("ok" if ok else "fail", "%s → %s (%d ms)" % (label, msg, ms))) + + if eff["mode"] == "proxychains": + try: + conf = (cfg or Config(DEFAULT_CONFIG_DIR, DEFAULT_SYSTEMD_DIR)).render_proxychains(unit or "test", eff) + except ProxyError as e: + results.append(("fail", "proxychains-Test: %s" % e)) + else: + results.append(("info", "Test über proxychains (so verbindet der Dienst):")) + results += proxychains_selftest(conf, targets, timeout) + return results + + # -------------------------------------------------------------------------- # ncurses-Oberfläche # -------------------------------------------------------------------------- @@ -932,8 +1298,8 @@ class TUI: if self.filter_mode: self.key_bar("Filter eingeben Enter übernehmen Esc Filter löschen") else: - self.key_bar("↑↓ Wählen Enter Bearbeiten g Global Entf Proxy entfernen " - "/ Filter c Nur konfigurierte r Neustart F1 Hilfe q Ende") + self.key_bar("↑↓ Wählen Enter Bearbeiten g Global Entf Entfernen " + "t Test / Filter c Nur konfigurierte r Neustart F1 Hilfe q Ende") scr.refresh() def run(self): @@ -978,6 +1344,8 @@ class TUI: elif k in ("c", "C"): self.only_configured = not self.only_configured self.apply_filter() + elif k in ("t", "T") and n: + self.test_service(self.visible[self.sel]) elif k == "r" and n: self.restart_now([self.visible[self.sel]]) elif k == curses.KEY_F5: @@ -1110,6 +1478,122 @@ class TUI: del win self.scr.touchwin() + def busy(self, text): + win, bh, bw = self.box("Bitte warten", [text], height_extra=0) + put(win, 1, 2, text, curses.A_BOLD, bw - 4) + win.refresh() + del win + + def prompt(self, title, label, value=""): + """Einzeilige Eingabe in einem Dialog. Liefert Text oder None (Esc).""" + while True: + win, bh, bw = self.box(title, [label, " " * 50]) + put(win, 1, 2, label, 0, bw - 4) + field_w = bw - 4 + vis = value[-(field_w - 1):] + put(win, 2, 2, vis.ljust(field_w), self.c_sel) + put(win, bh - 2, 2, "Enter übernehmen Esc abbrechen", self.c_dim) + self.cursor(True) + try: + win.move(2, 2 + len(vis)) + except curses.error: + pass + win.refresh() + k = win.get_wch() + del win + if is_enter(k): + self.cursor(False) + self.scr.touchwin() + return value + if k == "\x1b": + self.cursor(False) + self.scr.touchwin() + return None + if is_backspace(k): + value = value[:-1] + elif k == "\x15": + value = "" + elif isinstance(k, str) and k.isprintable(): + value += k + + def choose_targets(self): + lines = [ + "Welche Ziele sollen über den Proxy getestet werden?", + "", + " w Web: " + ", ".join(TEST_TARGETS_WEB), + " m Mail: SMTP 25/587/465, POP3S 995, IMAPS 993 (Gmail)", + " b beides", + " e eigenes Ziel (HOST:PORT, mehrere mit Komma)", + ] + win, bh, bw = self.box("Verbindungstest", lines) + for i, l in enumerate(lines): + put(win, 1 + i, 2, l, 0, bw - 4) + put(win, bh - 2, 2, "w / m / b / e wählen, Esc abbrechen", self.c_dim) + win.refresh() + while True: + k = win.get_wch() + if k in ("w", "W", "\n", "\r", curses.KEY_ENTER): + names = TEST_TARGETS_WEB + elif k in ("m", "M"): + names = TEST_TARGETS_MAIL + elif k in ("b", "B"): + names = TEST_TARGETS_WEB + TEST_TARGETS_MAIL + elif k in ("e", "E"): + del win + self.scr.touchwin() + text = self.prompt("Eigenes Ziel", "Ziel(e) als HOST:PORT, z.B. mail.example.com:25", + getattr(self, "last_target", "")) + if not text: + return None + self.last_target = text + names = [t for t in text.split(",") if t.strip()] + elif k in ("\x1b", "q"): + del win + self.scr.touchwin() + return None + else: + continue + try: + return [parse_target(t) for t in names] + except ProxyError as e: + self.message("Fehler", [str(e)], error=True) + return None + + def run_test(self, eff, unit, title): + targets = self.choose_targets() + if not targets: + return + self.busy("Verbindungstest läuft ... (bis zu %d s pro Ziel)" % TEST_TIMEOUT) + results = run_proxy_test(eff, targets, self.cfg, unit) + failed = sum(1 for st, _ in results if st == "fail") + results.append(("info", "")) + results.append(("ok" if not failed else "fail", + "Alles in Ordnung." if not failed else "%d Test(s) fehlgeschlagen." % failed)) + self.results_box(title, results) + + def results_box(self, title, results): + lines = [("[ OK ] " if st == "ok" else "[FEHL] " if st == "fail" else " ") + text + for st, text in results] + win, bh, bw = self.box(title, lines) + for i, (line, (st, _)) in enumerate(list(zip(lines, results))[:bh - 4]): + attr = self.c_ok if st == "ok" else self.c_err if st == "fail" else curses.A_BOLD + put(win, 1 + i, 2, line, attr, bw - 4) + put(win, bh - 2, 2, "[ OK – beliebige Taste ]", curses.A_REVERSE) + win.refresh() + win.get_wch() + del win + self.scr.touchwin() + + def test_service(self, unit): + svc = self.cfg.services.get(unit) + try: + eff, _ = resolve(svc or empty_settings(), self.cfg.global_) + except ProxyError as e: + self.message("Verbindungstest", [str(e)], error=True) + return + title = "Test %s%s" % (unit, "" if svc else " (globaler Proxy)") + self.run_test(eff, unit if svc else None, title) + def confirm(self, text): win, bh, bw = self.box("Bestätigen", [text]) put(win, 1, 2, text, 0, bw - 4) @@ -1181,7 +1665,8 @@ class TUI: " (z.B. Kerio SMTP/POP3). Lokale/private Netze und DNS gehen direkt.", "", "Tasten: Enter bearbeiten, g global, Entf/d entfernen, / filtern,", - "c nur konfigurierte, r Dienst neu starten, F5 Liste neu laden, q Ende.", + "c nur konfigurierte, r Dienst neu starten, t Verbindungstest,", + "F5 Liste neu laden, q Ende. Im Formular: F3 testet die eingegebenen Werte.", "Nach jeder Änderung wird 'systemctl daemon-reload' ausgeführt.", "Beim Beenden kann man geänderte Dienste neu starten lassen.", ] @@ -1213,6 +1698,7 @@ class TUI: fields.append(FormField(key, FIELD_LABELS[key], settings[key], secret=(key == "password"), placeholder=ph(key))) buttons = [FormField("save", "Speichern", "", kind="button"), + FormField("test", "Testen", "", kind="button"), FormField("cancel", "Abbrechen", "", kind="button")] if exists: buttons.append(FormField("remove", "Proxy entfernen" if not is_global else "Global löschen", @@ -1303,7 +1789,7 @@ class TUI: else: put(scr, py + 3, 2, "Benutzer/Kennwort leer = Proxy ohne Anmeldung.", self.c_dim) - self.key_bar("Tab/↑↓ Feld ←→ Auswahl Enter weiter F2 Speichern Strg+U Feld leeren Esc Abbrechen") + self.key_bar("Tab/↑↓ Feld ←→ Auswahl Enter weiter F2 Speichern F3 Testen Strg+U leeren Esc Abbrechen") if cursor_pos and items[cur].kind == "text": self.cursor(True) try: @@ -1327,6 +1813,8 @@ class TUI: action = "cancel" elif k == curses.KEY_F2: action = "save" + elif k == curses.KEY_F3: + action = "test" elif k in ("\t", curses.KEY_DOWN): cur = (cur + 1) % len(items) elif k in (curses.KEY_BTAB, curses.KEY_UP): @@ -1370,6 +1858,17 @@ class TUI: item.value = item.value[:item.pos] + k + item.value[item.pos:] item.pos += len(k) + if action == "test": + self.cursor(False) + try: + cand = validate_settings(collect(), is_global) + eff, _ = resolve(empty_settings() if is_global else cand, cand if is_global else g) + except ProxyError as e: + error = "Test nicht möglich: %s" % e + continue + self.run_test(eff, unit, "Test %s (ungespeicherte Werte)" % (unit or "globaler Proxy")) + error = "" + continue if action == "cancel": self.cursor(False) return None @@ -1601,6 +2100,43 @@ def cmd_show(cfg, args): return 0 +def print_test_results(results): + marks = {"ok": "[ OK ] ", "fail": "[FEHLER]", "info": " "} + for status, text in results: + print("%s %s" % (marks[status], text)) + failed = sum(1 for st, _ in results if st == "fail") + print() + print("Ergebnis: alles in Ordnung." if not failed else "Ergebnis: %d Test(s) fehlgeschlagen." % failed) + return 1 if failed else 0 + + +def test_targets(args): + targets = [parse_target(t) for t in (args.target or [])] + if args.mail: + targets += [parse_target(t) for t in TEST_TARGETS_MAIL] + if not targets: + targets = [parse_target(t) for t in TEST_TARGETS_WEB] + return targets + + +def cmd_test(cfg, args): + vals = fields_from_args(args) + targets = test_targets(args) + if args.unit: + unit = normalize_unit(args.unit) + base = dict(cfg.services.get(unit) or empty_settings()) + base.update(vals) + svc = validate_settings(base, is_global=False) + what = "Dienst %s%s" % (unit, "" if unit in cfg.services else " (nicht konfiguriert → globaler Proxy)") + else: + unit = None + svc = validate_settings(vals, is_global=False) + what = "angegebener Proxy" if any(svc[k] for k in FIELDS) else "globaler Proxy" + eff, _ = resolve(svc, cfg.global_) + print("Verbindungstest: %s" % what) + return print_test_results(run_proxy_test(eff, targets, cfg, unit, args.timeout)) + + def build_parser(): def common(sub): d = argparse.SUPPRESS if sub else None @@ -1645,11 +2181,18 @@ def build_parser(): p.add_argument("units", nargs="*", metavar="DIENST") p.add_argument("--all", action="store_true") sub.add_parser("apply", add_help=False, parents=[c]) + p = sub.add_parser("test", add_help=False, parents=[c, fields]) + p.add_argument("unit", nargs="?") + p.add_argument("--target", action="append", metavar="HOST:PORT") + p.add_argument("--mail", action="store_true") + p.add_argument("--timeout", type=float, default=TEST_TIMEOUT) return parser def main(argv=None): argv = sys.argv[1:] if argv is None else argv + if argv and argv[0] == "_probe": # intern: Kindprozess des proxychains-Selbsttests + return probe_direct(argv[2:], float(argv[1])) if "-h" in argv or "--help" in argv or (argv and argv[0] == "help"): print(HELP_TEXT) return 0 @@ -1658,7 +2201,9 @@ def main(argv=None): custom_dirs = bool(args.config_dir or args.systemd_dir) cfg = Config(args.config_dir or DEFAULT_CONFIG_DIR, args.systemd_dir or DEFAULT_SYSTEMD_DIR) - if os.geteuid() != 0 and not custom_dirs: + # 'test' mit eigenem --proxy/--host geht auch ohne root (ohne gespeicherte Konfiguration) + adhoc_test = (command == "test" and not args.unit and (args.proxy or args.host)) + if os.geteuid() != 0 and not custom_dirs and not adhoc_test: print("%s: Bitte als root ausführen (z.B. sudo %s)." % (PROG, PROG), file=sys.stderr) return 1 try: @@ -1667,14 +2212,18 @@ def main(argv=None): pass try: - cfg.load() + try: + cfg.load() + except ProxyError: + if not adhoc_test: + raise if command == "tui": if not sys.stdin.isatty() or not sys.stdout.isatty(): raise ProxyError("Das Menü braucht ein Terminal. Für Skripte die Befehle nutzen (siehe --help).") return run_tui(cfg, args) handler = { "list": cmd_list, "show": cmd_show, "global": cmd_global, "set": cmd_set, - "unset": cmd_unset, "apply": cmd_apply, + "unset": cmd_unset, "apply": cmd_apply, "test": cmd_test, }[command] return handler(cfg, args) except ProxyError as e: