Hacker-Software/webproxyserver-serviceclient
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Verbindungstest für Proxys hinzufügen

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
duffyduck
2026-10-05 14:34:44 +02:00
co-authored by Claude Opus 5.5
parent 3968c1a7b8
commit 9270994e82
2 changed files with 619 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+62
View File
@@ -196,6 +196,7 @@ Markierungen: **G** = nutzt globalen Proxy, **E** = eigene Werte,
| Entf / d | Proxy vom Dienst entfernen | | Entf / d | Proxy vom Dienst entfernen |
| / | Filter (Name oder Beschreibung), Esc löscht den Filter | | / | Filter (Name oder Beschreibung), Esc löscht den Filter |
| c | nur konfigurierte Dienste anzeigen (an/aus) | | 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 | | r | gewählten Dienst sofort neu starten |
| F5 | Dienstliste neu einlesen | | F5 | Dienstliste neu einlesen |
| F1 / ? | Hilfe | | 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“) | | ← → / Leertaste | Modus bzw. Proxy-Typ wählen (beim Typ gibt es zusätzlich „global“) |
| Enter | nächstes Feld bzw. Schaltfläche auslösen | | Enter | nächstes Feld bzw. Schaltfläche auslösen |
| F2 | Speichern | | F2 | Speichern |
| F3 / „Testen“ | Verbindungstest mit den eingegebenen, noch nicht gespeicherten Werten |
| Strg+U | Feld leeren | | Strg+U | Feld leeren |
| Esc | Abbrechen | | Esc | Abbrechen |
@@ -236,6 +238,7 @@ svc-proxy [BEFEHL] [OPTIONEN]
| `set DIENST... [FELDER] [--reset]` | Proxy für Dienst(e) setzen/ändern | | `set DIENST... [FELDER] [--reset]` | Proxy für Dienst(e) setzen/ändern |
| `unset DIENST...` / `unset --all` | Proxy entfernen | | `unset DIENST...` / `unset --all` | Proxy entfernen |
| `apply` | alle Drop-ins aus der Konfiguration neu erzeugen | | `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 | | `help`, `-h`, `--help` | Hilfe mit Beispielen |
### Felder ### 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 ## Beispiele
```bash ```bash
@@ -334,6 +390,12 @@ sudo svc-proxy set myapp --reset
sudo svc-proxy unset myapp --restart sudo svc-proxy unset myapp --restart
sudo svc-proxy unset --all --no-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 # Anzeigen
sudo svc-proxy list sudo svc-proxy list
sudo svc-proxy list --all sudo svc-proxy list --all
svc-proxy
+557 -8
View File
@@ -8,6 +8,7 @@ Ausführliche Hilfe: svc-proxy --help
""" """
import argparse import argparse
import base64
import curses import curses
import ipaddress import ipaddress
import json import json
@@ -16,13 +17,16 @@ import os
import pwd import pwd
import re import re
import socket import socket
import ssl
import struct
import subprocess import subprocess
import sys import sys
import tempfile import tempfile
import time
from glob import glob as glob_paths from glob import glob as glob_paths
from urllib.parse import quote, unquote, urlsplit from urllib.parse import quote, unquote, urlsplit
VERSION = "1.1.0" VERSION = "1.2.0"
PROG = "svc-proxy" PROG = "svc-proxy"
DEFAULT_CONFIG_DIR = "/etc/svc-proxy" DEFAULT_CONFIG_DIR = "/etc/svc-proxy"
@@ -70,6 +74,29 @@ PROXYCHAINS_LOCALNETS = [
"[::]:53/0", "[::]: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:_.@\\-]+$") UNIT_RE = re.compile(r"^[A-Za-z0-9:_.@\\-]+$")
HOST_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.,:*/_\[\]-]*$") NO_PROXY_RE = re.compile(r"^[A-Za-z0-9.,:*/_\[\]-]*$")
@@ -107,6 +134,8 @@ BEFEHLE
unset DIENST... Proxy von Diensten entfernen unset DIENST... Proxy von Diensten entfernen
unset --all Proxy von allen Diensten entfernen unset --all Proxy von allen Diensten entfernen
apply Alle Drop-ins aus der Konfiguration neu schreiben 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 help Diese Hilfe anzeigen
FELDER (für 'global' und 'set') 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, Ändert man den globalen Proxy, werden alle Dienste, die ihn (teilweise) erben,
automatisch mit angepasst. 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 NEUSTART / RELOAD
Nach jeder Änderung wird 'systemctl daemon-reload' ausgeführt. Die neuen Nach jeder Änderung wird 'systemctl daemon-reload' ausgeführt. Die neuen
Werte gelten aber erst nach einem Neustart des Dienstes. Werte gelten aber erst nach einem Neustart des Dienstes.
@@ -194,6 +239,12 @@ BEISPIELE
# Proxy von einem Dienst entfernen # Proxy von einem Dienst entfernen
sudo {prog} unset myapp --restart 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 # Übersicht und Details
sudo {prog} list sudo {prog} list
sudo {prog} show docker sudo {prog} show docker
@@ -206,6 +257,7 @@ DATEIEN
Dienstnamen ohne Endung erhalten automatisch '.service'. Dienstnamen ohne Endung erhalten automatisch '.service'.
""".format(prog=PROG, version=VERSION, cfg=DEFAULT_CONFIG_DIR, """.format(prog=PROG, version=VERSION, cfg=DEFAULT_CONFIG_DIR,
web=", ".join(TEST_TARGETS_WEB), timeout=TEST_TIMEOUT,
sysd=DEFAULT_SYSTEMD_DIR, dropin=DROPIN_NAME) sysd=DEFAULT_SYSTEMD_DIR, dropin=DROPIN_NAME)
@@ -213,6 +265,10 @@ class ProxyError(Exception):
pass pass
class ProxyFatal(ProxyError):
"""Fehler, bei dem weitere Testziele sinnlos sind (falscher Typ, Anmeldung)."""
# -------------------------------------------------------------------------- # --------------------------------------------------------------------------
# Hilfsfunktionen # Hilfsfunktionen
# -------------------------------------------------------------------------- # --------------------------------------------------------------------------
@@ -737,6 +793,316 @@ def list_system_services():
return units 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 # ncurses-Oberfläche
# -------------------------------------------------------------------------- # --------------------------------------------------------------------------
@@ -932,8 +1298,8 @@ class TUI:
if self.filter_mode: if self.filter_mode:
self.key_bar("Filter eingeben Enter übernehmen Esc Filter löschen") self.key_bar("Filter eingeben Enter übernehmen Esc Filter löschen")
else: else:
self.key_bar("↑↓ Wählen Enter Bearbeiten g Global Entf Proxy entfernen " self.key_bar("↑↓ Wählen Enter Bearbeiten g Global Entf Entfernen "
"/ Filter c Nur konfigurierte r Neustart F1 Hilfe q Ende") "t Test / Filter c Nur konfigurierte r Neustart F1 Hilfe q Ende")
scr.refresh() scr.refresh()
def run(self): def run(self):
@@ -978,6 +1344,8 @@ class TUI:
elif k in ("c", "C"): elif k in ("c", "C"):
self.only_configured = not self.only_configured self.only_configured = not self.only_configured
self.apply_filter() self.apply_filter()
elif k in ("t", "T") and n:
self.test_service(self.visible[self.sel])
elif k == "r" and n: elif k == "r" and n:
self.restart_now([self.visible[self.sel]]) self.restart_now([self.visible[self.sel]])
elif k == curses.KEY_F5: elif k == curses.KEY_F5:
@@ -1110,6 +1478,122 @@ class TUI:
del win del win
self.scr.touchwin() 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): def confirm(self, text):
win, bh, bw = self.box("Bestätigen", [text]) win, bh, bw = self.box("Bestätigen", [text])
put(win, 1, 2, text, 0, bw - 4) 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.", " (z.B. Kerio SMTP/POP3). Lokale/private Netze und DNS gehen direkt.",
"", "",
"Tasten: Enter bearbeiten, g global, Entf/d entfernen, / filtern,", "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.", "Nach jeder Änderung wird 'systemctl daemon-reload' ausgeführt.",
"Beim Beenden kann man geänderte Dienste neu starten lassen.", "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], fields.append(FormField(key, FIELD_LABELS[key], settings[key],
secret=(key == "password"), placeholder=ph(key))) secret=(key == "password"), placeholder=ph(key)))
buttons = [FormField("save", "Speichern", "", kind="button"), buttons = [FormField("save", "Speichern", "", kind="button"),
FormField("test", "Testen", "", kind="button"),
FormField("cancel", "Abbrechen", "", kind="button")] FormField("cancel", "Abbrechen", "", kind="button")]
if exists: if exists:
buttons.append(FormField("remove", "Proxy entfernen" if not is_global else "Global löschen", buttons.append(FormField("remove", "Proxy entfernen" if not is_global else "Global löschen",
@@ -1303,7 +1789,7 @@ class TUI:
else: else:
put(scr, py + 3, 2, "Benutzer/Kennwort leer = Proxy ohne Anmeldung.", self.c_dim) 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": if cursor_pos and items[cur].kind == "text":
self.cursor(True) self.cursor(True)
try: try:
@@ -1327,6 +1813,8 @@ class TUI:
action = "cancel" action = "cancel"
elif k == curses.KEY_F2: elif k == curses.KEY_F2:
action = "save" action = "save"
elif k == curses.KEY_F3:
action = "test"
elif k in ("\t", curses.KEY_DOWN): elif k in ("\t", curses.KEY_DOWN):
cur = (cur + 1) % len(items) cur = (cur + 1) % len(items)
elif k in (curses.KEY_BTAB, curses.KEY_UP): 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.value = item.value[:item.pos] + k + item.value[item.pos:]
item.pos += len(k) 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": if action == "cancel":
self.cursor(False) self.cursor(False)
return None return None
@@ -1601,6 +2100,43 @@ def cmd_show(cfg, args):
return 0 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 build_parser():
def common(sub): def common(sub):
d = argparse.SUPPRESS if sub else None d = argparse.SUPPRESS if sub else None
@@ -1645,11 +2181,18 @@ def build_parser():
p.add_argument("units", nargs="*", metavar="DIENST") p.add_argument("units", nargs="*", metavar="DIENST")
p.add_argument("--all", action="store_true") p.add_argument("--all", action="store_true")
sub.add_parser("apply", add_help=False, parents=[c]) 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 return parser
def main(argv=None): def main(argv=None):
argv = sys.argv[1:] if argv is None else argv 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"): if "-h" in argv or "--help" in argv or (argv and argv[0] == "help"):
print(HELP_TEXT) print(HELP_TEXT)
return 0 return 0
@@ -1658,7 +2201,9 @@ def main(argv=None):
custom_dirs = bool(args.config_dir or args.systemd_dir) 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) 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) print("%s: Bitte als root ausführen (z.B. sudo %s)." % (PROG, PROG), file=sys.stderr)
return 1 return 1
try: try:
@@ -1667,14 +2212,18 @@ def main(argv=None):
pass pass
try: try:
cfg.load() try:
cfg.load()
except ProxyError:
if not adhoc_test:
raise
if command == "tui": if command == "tui":
if not sys.stdin.isatty() or not sys.stdout.isatty(): 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).") raise ProxyError("Das Menü braucht ein Terminal. Für Skripte die Befehle nutzen (siehe --help).")
return run_tui(cfg, args) return run_tui(cfg, args)
handler = { handler = {
"list": cmd_list, "show": cmd_show, "global": cmd_global, "set": cmd_set, "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] }[command]
return handler(cfg, args) return handler(cfg, args)
except ProxyError as e: except ProxyError as e: