#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ svc-proxy – Proxy-Einstellungen für systemd-Dienste verwalten. Bedienung per ncurses-Menü (ohne Parameter) oder per Kommandozeile. Ausführliche Hilfe: svc-proxy --help """ import argparse import curses import ipaddress import json import locale import os import pwd import re import socket import subprocess import sys import tempfile from glob import glob as glob_paths from urllib.parse import quote, unquote, urlsplit VERSION = "1.1.0" PROG = "svc-proxy" DEFAULT_CONFIG_DIR = "/etc/svc-proxy" DEFAULT_SYSTEMD_DIR = "/etc/systemd/system" DROPIN_NAME = "50-svc-proxy.conf" PROXY_TYPES = ["http", "https", "socks5", "socks5h", "socks4"] DEFAULT_NO_PROXY = "localhost,127.0.0.1,::1" FIELDS = ["type", "host", "port", "user", "password", "no_proxy"] FIELD_LABELS = { "type": "Proxy-Typ", "host": "Host", "port": "Port", "user": "Benutzer", "password": "Kennwort", "no_proxy": "No-Proxy", } PROXY_ENV_VARS = ["http_proxy", "https_proxy", "ftp_proxy", "all_proxy"] # Modus pro Dienst: Proxy-Variablen oder proxychains (LD_PRELOAD, alle TCP-Verbindungen) MODES = ["env", "proxychains"] MODE_LABELS = { "env": "Proxy-Variablen", "proxychains": "proxychains (alle TCP-Verbindungen)", } PROXYCHAINS_TYPES = {"http": "http", "socks4": "socks4", "socks5": "socks5", "socks5h": "socks5"} PROXYCHAINS_LIB_GLOBS = [ "/usr/lib/*/libproxychains.so.4", "/usr/lib/libproxychains.so.4", "/usr/lib*/libproxychains4.so", "/usr/lib*/proxychains-ng/libproxychains4.so", "/usr/local/lib/libproxychains4.so", ] # Ziele, die im proxychains-Modus immer direkt (ohne Proxy) erreicht werden PROXYCHAINS_LOCALNETS = [ "127.0.0.0/255.0.0.0", "::1/128", "10.0.0.0/255.0.0.0", "172.16.0.0/255.240.0.0", "192.168.0.0/255.255.0.0", "169.254.0.0/255.255.0.0", "fc00::/7", "fe80::/10", "0.0.0.0:53/0.0.0.0", "[::]:53/0", ] 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.,:*/_\[\]-]*$") HELP_TEXT = """\ {prog} {version} – Proxy für systemd-Dienste setzen Hängt an beliebige systemd-Dienste ein Drop-in an, das den Dienst über einen Proxy laufen lässt. Pro Dienst gibt es zwei Modi: env (Standard) Setzt http_proxy, https_proxy, ftp_proxy, all_proxy und no_proxy (auch in GROSSBUCHSTABEN). Wirkt bei allen Programmen, die diese Variablen beachten (curl, wget, apt, Python, Go, Docker, ...). proxychains Lädt proxychains-ng per LD_PRELOAD in den Dienst. Dadurch laufen ALLE ausgehenden TCP-Verbindungen über den Proxy, auch bei Programmen, die keine Proxy-Variablen kennen (z.B. Kerio Connect: SMTP, POP3, IMAP). Direkt (ohne Proxy) bleiben: localhost, private Netze (10/8, 172.16/12, 192.168/16, fc00::/7, Link-Local), DNS (Port 53) und alle IP-/Netz- Einträge aus der No-Proxy-Liste. Benötigt: apt install proxychains4 Proxy-Typen: http, socks4, socks5 (socks5h wird wie socks5 behandelt). AUFRUF {prog} ncurses-Menü starten (wie '{prog} tui') {prog} BEFEHL [OPTIONEN] Kommandozeilen-Modus (für Automatisierung) BEFEHLE tui Menü starten: alle Dienste auflisten, auswählen, bearbeiten list [--all] Dienste mit Proxy-Konfiguration anzeigen (--all: alle Dienste des Systems) show [DIENST] Effektive Einstellungen eines Dienstes anzeigen (ohne DIENST: den globalen Proxy) global [FELDER] Globalen Proxy setzen/ändern (ohne FELDER: anzeigen) global --clear Globalen Proxy löschen set DIENST... [FELDER] Proxy für einen oder mehrere Dienste setzen/ändern unset DIENST... Proxy von Diensten entfernen unset --all Proxy von allen Diensten entfernen apply Alle Drop-ins aus der Konfiguration neu schreiben help Diese Hilfe anzeigen FELDER (für 'global' und 'set') --proxy URL Komplette Proxy-URL, z.B. http://user:pass@proxy:3128 Setzt Typ, Host, Port, Benutzer und Kennwort auf einmal. Einzelne Felder dahinter überschreiben die URL-Werte. --type TYP http | https | socks5 | socks5h | socks4 --host HOST Hostname oder IP des Proxys --port PORT Port des Proxys (1-65535) --user NAME Benutzername (weglassen = Proxy ohne Anmeldung) --password KENNWORT Kennwort (Achtung: in der Prozessliste sichtbar!) --password-stdin Kennwort von der Standardeingabe lesen (empfohlen) --no-proxy LISTE Kommagetrennte Ausnahmen, z.B. localhost,10.0.0.0/8,.intern --mode MODUS (nur 'set') env | proxychains (Standard: env, wird nicht vom globalen Proxy geerbt) Ein Feld mit leerem Wert (z.B. --user "") wird geleert. Bei 'set' werden nur die angegebenen Felder geändert, die übrigen bleiben erhalten. --reset (nur 'set') Vorher alle Felder des Dienstes leeren. 'set DIENST --reset' = Dienst nutzt komplett den globalen Proxy. VERERBUNG (Dienst <-> global) Jedes leere Feld eines Dienstes übernimmt den globalen Wert: Typ, Host, Port, No-Proxy leer -> globaler Wert Benutzer/Kennwort leer -> globale Zugangsdaten, ABER nur wenn auch der Host vom globalen Proxy kommt. Hat der Dienst einen eigenen Host und keinen Benutzer, wird der Proxy ohne Anmeldung verwendet. Ändert man den globalen Proxy, werden alle Dienste, die ihn (teilweise) erben, automatisch mit angepasst. NEUSTART / RELOAD Nach jeder Änderung wird 'systemctl daemon-reload' ausgeführt. Die neuen Werte gelten aber erst nach einem Neustart des Dienstes. --restart Geänderte Dienste ohne Nachfrage neu starten (nur laufende Dienste, per 'systemctl try-restart') --no-restart Nicht neu starten und nicht nachfragen (keins von beiden) Im Terminal wird nachgefragt; ohne Terminal kein Neustart. Im Menü erscheint beim Beenden eine Auswahlliste. --no-reload Kein 'systemctl daemon-reload' ausführen WEITERE OPTIONEN --config-dir DIR Konfigurationsverzeichnis (Standard: {cfg}) --systemd-dir DIR Ziel für Drop-ins (Standard: {sysd}) -h, --help Diese Hilfe anzeigen -V, --version Version anzeigen BEISPIELE # Menü starten sudo {prog} # Globalen Proxy ohne Anmeldung setzen sudo {prog} global --proxy http://proxy.firma.local:3128 # Globalen Proxy mit Anmeldung, Kennwort sicher über stdin echo 'geheim' | sudo {prog} global --proxy http://proxy:3128 --user max --password-stdin # Ausnahmen für den globalen Proxy festlegen sudo {prog} global --no-proxy localhost,127.0.0.1,::1,.firma.local,10.0.0.0/8 # Dienste den globalen Proxy nutzen lassen und gleich neu starten sudo {prog} set docker nginx.service --restart # Eigener SOCKS-Proxy ohne Anmeldung für einen Dienst sudo {prog} set myapp --proxy socks5h://10.1.1.1:1080 # Kerio Connect: alle Mailverbindungen über SOCKS5, Proxy-Daten vom globalen Proxy sudo apt install proxychains4 sudo {prog} set kerio-connect --mode proxychains --restart # Kerio mit eigenem SOCKS5-Proxy inkl. Anmeldung, LAN-Netz 192.0.2.0/24 direkt sudo {prog} set kerio-connect --mode proxychains --proxy socks5://192.0.2.10:1080 \\ --user max --password-stdin --no-proxy 192.0.2.0/24 # Zurück zu den Proxy-Variablen sudo {prog} set kerio-connect --mode env # Nur den Port eines Dienstes ändern (Rest bleibt) sudo {prog} set myapp --port 1081 --no-restart # Dienst wieder komplett auf den globalen Proxy zurücksetzen sudo {prog} set myapp --reset # Proxy von einem Dienst entfernen sudo {prog} unset myapp --restart # Übersicht und Details sudo {prog} list sudo {prog} show docker DATEIEN {cfg}/config.json Konfiguration (global + Dienste, 0600) {cfg}/env/DIENST.env erzeugte Umgebungsdatei (0600) {cfg}/proxychains/DIENST.conf proxychains-Konfiguration (Modus proxychains) {sysd}/DIENST.d/{dropin} erzeugtes Drop-in Dienstnamen ohne Endung erhalten automatisch '.service'. """.format(prog=PROG, version=VERSION, cfg=DEFAULT_CONFIG_DIR, sysd=DEFAULT_SYSTEMD_DIR, dropin=DROPIN_NAME) class ProxyError(Exception): pass # -------------------------------------------------------------------------- # Hilfsfunktionen # -------------------------------------------------------------------------- def empty_settings(): s = {k: "" for k in FIELDS} s["mode"] = "" # nur bei Diensten benutzt, wird nicht vererbt return s def normalize_unit(name): name = name.strip() if not name or not UNIT_RE.match(name): raise ProxyError("Ungültiger Dienstname: %r" % name) if "." not in name: name += ".service" if name.endswith("@.service"): raise ProxyError("%s ist eine Vorlage – bitte eine Instanz angeben (z.B. name@xyz.service)" % name) return name def atomic_write(path, content, mode): directory = os.path.dirname(path) os.makedirs(directory, exist_ok=True) fd, tmp = tempfile.mkstemp(dir=directory, prefix=".tmp-svc-proxy-") try: with os.fdopen(fd, "w", encoding="utf-8") as f: f.write(content) os.chmod(tmp, mode) os.replace(tmp, path) except BaseException: try: os.unlink(tmp) except OSError: pass raise def write_if_different(path, content, mode): try: with open(path, encoding="utf-8") as f: if f.read() == content: return False except OSError: pass atomic_write(path, content, mode) return True def remove_file(path): try: os.unlink(path) return True except FileNotFoundError: return False def validate_settings(raw, is_global): """Prüft und bereinigt ein Feld-Dict. Wirft ProxyError.""" s = empty_settings() for k in FIELDS: v = raw.get(k) or "" if any(ord(c) < 32 for c in v): raise ProxyError("%s enthält Steuerzeichen" % FIELD_LABELS[k]) s[k] = v if k == "password" else v.strip() s["type"] = s["type"].lower() if s["type"] and s["type"] not in PROXY_TYPES: raise ProxyError("Unbekannter Proxy-Typ '%s' (erlaubt: %s)" % (s["type"], ", ".join(PROXY_TYPES))) host = s["host"] if host.startswith("[") and host.endswith("]"): host = host[1:-1] if host and not HOST_RE.match(host): raise ProxyError("Ungültiger Host: %s" % s["host"]) s["host"] = host if s["port"]: if not s["port"].isdigit() or not 1 <= int(s["port"]) <= 65535: raise ProxyError("Ungültiger Port: %s (erlaubt: 1-65535)" % s["port"]) s["port"] = str(int(s["port"])) if s["password"] and not s["user"]: raise ProxyError("Kennwort angegeben, aber kein Benutzername") s["no_proxy"] = re.sub(r"\s+", "", s["no_proxy"]) if not NO_PROXY_RE.match(s["no_proxy"]): raise ProxyError("Ungültige No-Proxy-Liste: %s" % s["no_proxy"]) s["mode"] = (raw.get("mode") or "").strip().lower() if s["mode"] and s["mode"] not in MODES: raise ProxyError("Unbekannter Modus '%s' (erlaubt: %s)" % (s["mode"], ", ".join(MODES))) if is_global: if s["mode"]: raise ProxyError("Den Modus gibt es nur pro Dienst, nicht global") if any(s[k] for k in ("host", "port", "user", "password")): if not s["host"]: raise ProxyError("Globaler Proxy: Host fehlt") if not s["port"]: raise ProxyError("Globaler Proxy: Port fehlt") if s["host"] and not s["type"]: s["type"] = "http" return s def resolve(svc, glob): """Effektive Werte eines Dienstes. Liefert (werte, quellen) oder wirft ProxyError. quellen[feld] ist 'dienst', 'global' oder 'standard'.""" eff, src = {}, {} def take(key, default=""): if svc[key]: eff[key], src[key] = svc[key], "dienst" elif glob[key]: eff[key], src[key] = glob[key], "global" else: eff[key], src[key] = default, "standard" take("type", "http") take("host") take("port") take("no_proxy") if svc["user"]: eff["user"], eff["password"] = svc["user"], svc["password"] src["user"] = src["password"] = "dienst" elif svc["host"]: eff["user"] = eff["password"] = "" src["user"] = src["password"] = "dienst" else: eff["user"], eff["password"] = glob["user"], glob["password"] src["user"] = src["password"] = "global" eff["mode"] = svc.get("mode") or "env" src["mode"] = "dienst" if svc.get("mode") else "standard" if not eff["host"]: raise ProxyError("kein Proxy-Host (weder beim Dienst noch global gesetzt)") if not eff["port"]: raise ProxyError("kein Proxy-Port (weder beim Dienst noch global gesetzt)") if eff["mode"] == "proxychains": if eff["type"] not in PROXYCHAINS_TYPES: raise ProxyError("proxychains kann keinen '%s'-Proxy (nur http, socks4, socks5)" % eff["type"]) if re.search(r"\s", eff["user"] + eff["password"]): raise ProxyError("proxychains: Benutzer/Kennwort dürfen keine Leerzeichen enthalten") return eff, src def find_proxychains_lib(): override = os.environ.get("SVC_PROXY_PROXYCHAINS_LIB") if override: return override for pattern in PROXYCHAINS_LIB_GLOBS: for path in sorted(glob_paths(pattern)): if os.path.isfile(path): return path raise ProxyError("proxychains ist nicht installiert (Debian/Ubuntu: apt install proxychains4, " "RHEL/Fedora: dnf install proxychains-ng)") def proxy_ipv4(host): """proxychains akzeptiert in der ProxyList nur IPv4-Adressen.""" try: addr = ipaddress.ip_address(host) except ValueError: addr = None if addr is not None: if addr.version != 4: raise ProxyError("proxychains braucht eine IPv4-Adresse für den Proxy, nicht %s" % host) return str(addr) try: infos = socket.getaddrinfo(host, None, socket.AF_INET, socket.SOCK_STREAM) except socket.gaierror as e: raise ProxyError("Proxy-Host %s nicht auflösbar (%s) – proxychains braucht eine IPv4-Adresse" % (host, e.strerror)) return infos[0][4][0] def proxychains_localnets(no_proxy): """Standard-Ausnahmen plus alle IP-/Netz-Einträge aus der No-Proxy-Liste.""" nets = list(PROXYCHAINS_LOCALNETS) for entry in no_proxy.split(","): entry = entry.strip().strip("[]") try: net = ipaddress.ip_network(entry, strict=False) except ValueError: continue # Hostnamen/Domains kann proxychains nicht ausnehmen if net.version == 4: item = "%s/%s" % (net.network_address, net.netmask) else: item = "%s/%d" % (net.network_address, net.prefixlen) if item not in nets: nets.append(item) return nets def check_mode(eff): """Prüft vor dem Speichern, ob der gewählte Modus nutzbar ist.""" if eff["mode"] == "proxychains": find_proxychains_lib() proxy_ipv4(eff["host"]) def service_user(unit): """User= des Dienstes (leer = root).""" rc, out, _ = run_cmd(["systemctl", "show", "-p", "User", "-p", "DynamicUser", unit]) if rc != 0: return "", False vals = dict(line.split("=", 1) for line in out.splitlines() if "=" in line) return vals.get("User", ""), vals.get("DynamicUser", "no") == "yes" def proxy_url(eff, mask=False): auth = "" if eff.get("user"): auth = quote(eff["user"], safe="") if eff.get("password"): auth += ":" + ("****" if mask else quote(eff["password"], safe="")) auth += "@" host = eff["host"] if ":" in host: host = "[%s]" % host return "%s://%s%s:%s" % (eff["type"] or "http", auth, host, eff["port"]) def parse_proxy_url(url): url = url.strip() if "://" not in url: url = "http://" + url try: p = urlsplit(url) port = p.port except ValueError as e: raise ProxyError("Ungültige Proxy-URL %r: %s" % (url, e)) scheme = p.scheme.lower() if scheme not in PROXY_TYPES: raise ProxyError("Unbekannter Proxy-Typ '%s' in URL (erlaubt: %s)" % (scheme, ", ".join(PROXY_TYPES))) if not p.hostname: raise ProxyError("Proxy-URL ohne Host: %r" % url) return { "type": scheme, "host": p.hostname, "port": str(port) if port else "", "user": unquote(p.username or ""), "password": unquote(p.password or ""), } def is_inheriting(svc): return not any(svc[k] for k in FIELDS) def mode_suffix(eff): return " [proxychains]" if eff["mode"] == "proxychains" else "" # -------------------------------------------------------------------------- # Konfiguration und Erzeugen der Dateien # -------------------------------------------------------------------------- class Config: def __init__(self, config_dir, systemd_dir): self.config_dir = config_dir self.config_file = os.path.join(config_dir, "config.json") self.env_dir = os.path.join(config_dir, "env") self.pc_dir = os.path.join(config_dir, "proxychains") self.systemd_dir = systemd_dir self.global_ = empty_settings() self.services = {} # Pfade @staticmethod def safe_name(unit): return re.sub(r"[^A-Za-z0-9:_.@-]", "_", unit) def env_file(self, unit): return os.path.join(self.env_dir, self.safe_name(unit) + ".env") def pc_file(self, unit): return os.path.join(self.pc_dir, self.safe_name(unit) + ".conf") def dropin_dir(self, unit): return os.path.join(self.systemd_dir, unit + ".d") def dropin_file(self, unit): return os.path.join(self.dropin_dir(unit), DROPIN_NAME) # Laden / Speichern def load(self): try: with open(self.config_file, encoding="utf-8") as f: data = json.load(f) except FileNotFoundError: return except PermissionError: raise ProxyError("Keine Berechtigung für %s – bitte als root ausführen" % self.config_file) except (OSError, ValueError) as e: raise ProxyError("Konfiguration %s nicht lesbar: %s" % (self.config_file, e)) def clean(d, keys): s = empty_settings() for k in keys: if isinstance(d, dict) and isinstance(d.get(k), str): s[k] = d[k] return s self.global_ = clean(data.get("global"), FIELDS) self.services = {} for unit, s in (data.get("services") or {}).items(): if UNIT_RE.match(unit): self.services[unit] = clean(s, FIELDS + ["mode"]) def ensure_config_dir(self): # 0711: Dienste mit eigenem Benutzer müssen ihre proxychains-Datei erreichen, # dürfen das Verzeichnis aber nicht auflisten. config.json und env/ bleiben 0600/0700. os.makedirs(self.config_dir, mode=0o711, exist_ok=True) os.chmod(self.config_dir, 0o711) def save(self): self.ensure_config_dir() data = { "version": 1, "global": {k: self.global_[k] for k in FIELDS}, "services": {u: self.services[u] for u in sorted(self.services)}, } atomic_write(self.config_file, json.dumps(data, indent=2, ensure_ascii=False) + "\n", 0o600) # Drop-ins def render_env(self, unit, eff): lines = [ "# Erzeugt von %s fuer %s - Aenderungen werden ueberschrieben." % (PROG, unit), "# Bearbeiten mit: %s set %s ..." % (PROG, unit), ] if eff["mode"] == "proxychains": lines.append('LD_PRELOAD="%s"' % find_proxychains_lib()) lines.append('PROXYCHAINS_CONF_FILE="%s"' % self.pc_file(unit)) lines.append('PROXYCHAINS_QUIET_MODE="1"') return "\n".join(lines) + "\n" url = proxy_url(eff) for var in PROXY_ENV_VARS: lines.append('%s="%s"' % (var, url)) lines.append('%s="%s"' % (var.upper(), url)) if eff["no_proxy"]: lines.append('no_proxy="%s"' % eff["no_proxy"]) lines.append('NO_PROXY="%s"' % eff["no_proxy"]) return "\n".join(lines) + "\n" def render_proxychains(self, unit, eff): entry = "%s %s %s" % (PROXYCHAINS_TYPES[eff["type"]], proxy_ipv4(eff["host"]), eff["port"]) if eff["user"]: entry += " %s %s" % (eff["user"], eff["password"]) lines = [ "# Erzeugt von %s fuer %s - Aenderungen werden ueberschrieben." % (PROG, unit), "# Proxy: %s (%s)" % (proxy_url(eff, mask=True), eff["host"]), "strict_chain", "quiet_mode", "tcp_read_time_out 15000", "tcp_connect_time_out 8000", "# Kein proxy_dns: DNS wird normal aufgeloest (wichtig z.B. fuer MX-Abfragen).", "", "# Diese Ziele werden direkt (ohne Proxy) verbunden:", ] lines += ["localnet " + n for n in proxychains_localnets(eff["no_proxy"])] lines += ["", "[ProxyList]", entry] return "\n".join(lines) + "\n" def write_proxychains(self, unit, eff): """Schreibt die proxychains-Datei. Liefert (geändert, warnung).""" self.ensure_config_dir() os.makedirs(self.pc_dir, mode=0o755, exist_ok=True) os.chmod(self.pc_dir, 0o755) path = self.pc_file(unit) secret = bool(eff["user"]) changed = write_if_different(path, self.render_proxychains(unit, eff), 0o600 if secret else 0o644) os.chmod(path, 0o600 if secret else 0o644) warning = None # Die Bibliothek liest die Datei im Dienst-Prozess, also mit dessen Benutzer. user, dynamic = service_user(unit) if secret and os.geteuid() == 0: uid = gid = 0 if dynamic: warning = ("%s: nutzt DynamicUser – die proxychains-Datei mit Kennwort ist für den " "Dienst nicht lesbar. Proxy ohne Anmeldung verwenden." % unit) elif user and user != "root": try: pw = pwd.getpwnam(user) uid, gid = pw.pw_uid, pw.pw_gid except KeyError: warning = "%s: Benutzer '%s' unbekannt – proxychains-Datei evtl. nicht lesbar" % (unit, user) st = os.stat(path) if (st.st_uid, st.st_gid) != (uid, gid): os.chown(path, uid, gid) return changed, warning def render_dropin(self, unit): return ( "# Erzeugt von %s - nicht manuell bearbeiten.\n" "# Entfernen mit: %s unset %s\n" "[Service]\n" "EnvironmentFile=%s\n" % (PROG, PROG, unit, self.env_file(unit)) ) def remove_unit_files(self, unit): changed = remove_file(self.dropin_file(unit)) changed |= remove_file(self.env_file(unit)) changed |= remove_file(self.pc_file(unit)) try: os.rmdir(self.dropin_dir(unit)) except OSError: pass return changed def apply_unit(self, unit): """Schreibt/entfernt die Dateien eines Dienstes. Liefert (geändert, warnung).""" svc = self.services.get(unit) if svc is None: return self.remove_unit_files(unit), None warning = None try: eff, _ = resolve(svc, self.global_) changed = False if eff["mode"] == "proxychains": changed, warning = self.write_proxychains(unit, eff) else: changed = remove_file(self.pc_file(unit)) env = self.render_env(unit, eff) except ProxyError as e: return self.remove_unit_files(unit), "%s: %s – Proxy für diesen Dienst deaktiviert" % (unit, e) self.ensure_config_dir() os.makedirs(self.env_dir, mode=0o700, exist_ok=True) changed |= write_if_different(self.env_file(unit), env, 0o600) changed |= write_if_different(self.dropin_file(unit), self.render_dropin(unit), 0o644) return changed, warning def managed_dropins(self): """Units, für die ein Drop-in von svc-proxy existiert.""" found = set() try: entries = os.listdir(self.systemd_dir) except OSError: return found for entry in entries: if entry.endswith(".d") and os.path.isfile(os.path.join(self.systemd_dir, entry, DROPIN_NAME)): found.add(entry[:-2]) return found def apply_all(self): changed, warnings = set(), [] for unit in sorted(set(self.services) | self.managed_dropins()): ch, warn = self.apply_unit(unit) if ch: changed.add(unit) if warn: warnings.append(warn) return changed, warnings # -------------------------------------------------------------------------- # systemd # -------------------------------------------------------------------------- def run_cmd(cmd): try: r = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True) except FileNotFoundError: return 127, "", "%s nicht gefunden" % cmd[0] return r.returncode, r.stdout, r.stderr def daemon_reload(): rc, _, err = run_cmd(["systemctl", "daemon-reload"]) if rc != 0: raise ProxyError("systemctl daemon-reload fehlgeschlagen: %s" % err.strip()) def unit_active(unit): rc, out, _ = run_cmd(["systemctl", "is-active", unit]) return out.strip() or "unbekannt" def restart_units(units): """Startet laufende Dienste neu. Liefert Liste von (unit, ok, text).""" results = [] for unit in sorted(units): state = unit_active(unit) if state not in ("active", "reloading", "activating"): results.append((unit, True, "nicht aktiv (%s) – kein Neustart nötig" % state)) continue rc, _, err = run_cmd(["systemctl", "try-restart", unit]) if rc == 0: results.append((unit, True, "neu gestartet")) else: results.append((unit, False, "Neustart fehlgeschlagen: %s" % err.strip())) return results def list_system_services(): """Alle Dienste: {name: {'active':..., 'sub':..., 'file':..., 'desc':...}}""" units = {} rc, out, _ = run_cmd(["systemctl", "list-unit-files", "--type=service", "--no-legend", "--no-pager", "--plain"]) if rc == 0: for line in out.splitlines(): parts = line.split() if not parts or parts[0].endswith("@.service"): continue units[parts[0]] = {"active": "", "sub": "", "file": parts[1] if len(parts) > 1 else "", "desc": ""} rc, out, _ = run_cmd(["systemctl", "list-units", "--type=service", "--all", "--no-legend", "--no-pager", "--plain"]) if rc == 0: for line in out.splitlines(): parts = line.lstrip("●* ").split(None, 4) if len(parts) < 4: continue name, load, active, sub = parts[:4] info = units.setdefault(name, {"active": "", "sub": "", "file": "", "desc": ""}) info["active"], info["sub"] = active, sub if load == "not-found": info["file"] = "not-found" info["desc"] = parts[4] if len(parts) > 4 else "" return units # -------------------------------------------------------------------------- # ncurses-Oberfläche # -------------------------------------------------------------------------- def put(win, y, x, text, attr=0, width=None): h, w = win.getmaxyx() if y < 0 or y >= h or x < 0 or x >= w: return maxw = w - x if width is None else min(width, w - x) if maxw <= 0: return try: win.addnstr(y, x, text, maxw, attr) except curses.error: pass def is_enter(k): return k in ("\n", "\r", curses.KEY_ENTER) def is_backspace(k): return k in ("\x7f", "\x08", curses.KEY_BACKSPACE) class FormField: def __init__(self, key, label, value, kind="text", choices=None, secret=False, placeholder="", display=None): self.key = key self.label = label self.kind = kind # text | choice | button self.value = value self.choices = choices or [] self.display = display or {} self.secret = secret self.placeholder = placeholder self.pos = len(value) self.off = 0 class TUI: def __init__(self, scr, cfg, args): self.scr = scr self.cfg = cfg self.args = args self.changed = set() self.filter = "" self.filter_mode = False self.only_configured = False self.sel = 0 self.top = 0 self.msg = "" self.msg_attr = 0 self.units = {} self.visible = [] self.init_colors() self.refresh_units() # ---------- Grundlagen ---------- def init_colors(self): self.c_title = curses.A_REVERSE | curses.A_BOLD self.c_sel = curses.A_REVERSE self.c_dim = curses.A_DIM self.c_err = curses.A_BOLD self.c_ok = curses.A_BOLD self.c_global = curses.A_NORMAL self.c_own = curses.A_BOLD if curses.has_colors(): try: curses.start_color() curses.use_default_colors() curses.init_pair(1, curses.COLOR_WHITE, curses.COLOR_BLUE) curses.init_pair(2, curses.COLOR_RED, -1) curses.init_pair(3, curses.COLOR_GREEN, -1) curses.init_pair(4, curses.COLOR_CYAN, -1) curses.init_pair(5, curses.COLOR_YELLOW, -1) curses.init_pair(6, curses.COLOR_BLACK, curses.COLOR_CYAN) self.c_title = curses.color_pair(1) | curses.A_BOLD self.c_err = curses.color_pair(2) | curses.A_BOLD self.c_ok = curses.color_pair(3) | curses.A_BOLD self.c_global = curses.color_pair(4) self.c_own = curses.color_pair(5) | curses.A_BOLD self.c_sel = curses.color_pair(6) except curses.error: pass def cursor(self, visible): try: curses.curs_set(1 if visible else 0) except curses.error: pass def set_msg(self, text, attr=None): self.msg = text self.msg_attr = self.c_ok if attr is None else attr def title_bar(self, text): h, w = self.scr.getmaxyx() put(self.scr, 0, 0, (" " + text).ljust(w), self.c_title) def key_bar(self, text): h, w = self.scr.getmaxyx() put(self.scr, h - 1, 0, (" " + text).ljust(w - 1), self.c_title) def refresh_units(self): self.units = list_system_services() for unit in self.cfg.services: self.units.setdefault(unit, {"active": "", "sub": "", "file": "?", "desc": ""}) self.apply_filter() def apply_filter(self): f = self.filter.lower() names = sorted(self.units) if self.only_configured: names = [n for n in names if n in self.cfg.services] if f: names = [n for n in names if f in n.lower() or f in self.units[n]["desc"].lower()] self.visible = names self.sel = max(0, min(self.sel, len(names) - 1)) def global_summary(self): g = self.cfg.global_ if not g["host"]: return "(nicht gesetzt)" text = proxy_url(g, mask=True) if g["no_proxy"]: text += " No-Proxy: " + g["no_proxy"] return text def service_summary(self, unit): svc = self.cfg.services.get(unit) if svc is None: return "", 0 try: eff, _ = resolve(svc, self.cfg.global_) except ProxyError as e: return "FEHLER: " + str(e), self.c_err if is_inheriting(svc): return "global → " + proxy_url(eff, mask=True) + mode_suffix(eff), self.c_global return "eigen → " + proxy_url(eff, mask=True) + mode_suffix(eff), self.c_own # ---------- Hauptliste ---------- def draw_main(self): scr = self.scr scr.erase() h, w = scr.getmaxyx() self.title_bar("%s %s – Proxy für systemd-Dienste" % (PROG, VERSION)) put(scr, 1, 1, "Global: ", curses.A_BOLD) put(scr, 1, 9, self.global_summary(), self.c_global if self.cfg.global_["host"] else self.c_dim) info = "%d Dienste" % len(self.visible) if self.only_configured: info += " (nur konfigurierte)" if self.filter or self.filter_mode: info += " Filter: " + self.filter + ("_" if self.filter_mode else "") put(scr, 2, 1, info, self.c_dim if not self.filter_mode else curses.A_BOLD) name_w = max([len(n) for n in self.visible] + [10]) name_w = min(name_w, max(20, w // 2 - 14)) state_w = 10 put(scr, 3, 0, " %-*s %-*s %s" % (name_w, "Dienst", state_w, "Status", "Proxy"), curses.A_UNDERLINE) list_top, list_h = 4, max(1, h - 6) if self.sel < self.top: self.top = self.sel elif self.sel >= self.top + list_h: self.top = self.sel - list_h + 1 for i in range(list_h): idx = self.top + i if idx >= len(self.visible): break unit = self.visible[idx] info = self.units.get(unit, {}) svc = self.cfg.services.get(unit) mark = " " if svc is not None: mark = "G" if is_inheriting(svc) else "E" star = "*" if unit in self.changed else " " state = info.get("active") or info.get("file") or "" name = unit if len(unit) <= name_w else unit[:name_w - 1] + "…" summary, sattr = self.service_summary(unit) y = list_top + i line = " %s%s %-*s %-*s " % (mark, star, name_w, name, state_w, state[:state_w]) if idx == self.sel: put(scr, y, 0, (line + summary).ljust(w), self.c_sel) else: put(scr, y, 0, line) put(scr, y, len(line), summary, sattr) if not self.visible: put(scr, list_top, 4, "Keine Dienste gefunden.", self.c_dim) put(scr, h - 2, 1, self.msg, self.msg_attr) 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") scr.refresh() def run(self): self.cursor(False) if not self.cfg.global_["host"]: self.set_msg("Tipp: Mit 'g' zuerst einen globalen Proxy festlegen.", self.c_dim) while True: self.draw_main() try: k = self.scr.get_wch() except curses.error: continue if k == curses.KEY_RESIZE: continue if self.filter_mode: self.handle_filter_key(k) continue self.msg = "" n = len(self.visible) h, _ = self.scr.getmaxyx() page = max(1, h - 7) if k in (curses.KEY_UP, "k"): self.sel = max(0, self.sel - 1) elif k in (curses.KEY_DOWN, "j"): self.sel = min(n - 1, self.sel + 1) elif k == curses.KEY_PPAGE: self.sel = max(0, self.sel - page) elif k == curses.KEY_NPAGE: self.sel = min(n - 1, self.sel + page) elif k == curses.KEY_HOME: self.sel = 0 elif k == curses.KEY_END: self.sel = max(0, n - 1) elif is_enter(k) and n: self.edit_service(self.visible[self.sel]) elif k in ("g", "G"): self.edit_global() elif k in (curses.KEY_DC, "d") and n: self.remove_service(self.visible[self.sel]) elif k == "/": self.filter_mode = True elif k in ("c", "C"): self.only_configured = not self.only_configured self.apply_filter() elif k == "r" and n: self.restart_now([self.visible[self.sel]]) elif k == curses.KEY_F5: self.refresh_units() self.set_msg("Dienstliste aktualisiert.") elif k in (curses.KEY_F1, "?", "h"): self.show_help() elif k in ("q", "Q", curses.KEY_F10): return self.ask_restart_on_exit() self.sel = max(0, min(self.sel, len(self.visible) - 1)) def handle_filter_key(self, k): if is_enter(k): self.filter_mode = False elif k == "\x1b": self.filter_mode = False self.filter = "" elif is_backspace(k): self.filter = self.filter[:-1] elif isinstance(k, str) and k.isprintable(): self.filter += k self.sel = 0 self.top = 0 self.apply_filter() # ---------- Aktionen ---------- def after_change(self, changed, warnings=()): self.changed |= changed if changed and not self.args.no_reload: try: daemon_reload() except ProxyError as e: self.message("Fehler", [str(e)], error=True) return if warnings: self.message("Hinweis", list(warnings), error=True) def edit_service(self, unit): svc = self.cfg.services.get(unit) result = self.form(unit, svc or empty_settings(), is_global=False, exists=svc is not None) if result is None: return if result == "remove": self.remove_service(unit, confirm=False) return self.cfg.services[unit] = result self.cfg.save() ch, warn = self.cfg.apply_unit(unit) self.after_change({unit} if ch else set(), [warn] if warn else []) self.set_msg("%s gespeichert%s." % (unit, "" if ch else " (keine Änderung)")) def edit_global(self): g = dict(self.cfg.global_) if not g["host"] and not g["no_proxy"]: g["no_proxy"] = DEFAULT_NO_PROXY result = self.form(None, g, is_global=True, exists=bool(self.cfg.global_["host"])) if result is None: return if result == "remove": if not self.confirm("Globalen Proxy wirklich löschen? Dienste, die ihn erben, verlieren ihren Proxy."): return result = empty_settings() self.cfg.global_ = result self.cfg.save() changed, warnings = self.cfg.apply_all() self.after_change(changed, warnings) self.set_msg("Globaler Proxy gespeichert, %d Dienst(e) angepasst." % len(changed)) def remove_service(self, unit, confirm=True): if unit not in self.cfg.services: self.set_msg("%s hat keinen Proxy." % unit, self.c_dim) return if confirm and not self.confirm("Proxy von %s entfernen?" % unit): return del self.cfg.services[unit] self.cfg.save() ch, _ = self.cfg.apply_unit(unit) self.after_change({unit} if ch else set()) self.apply_filter() self.set_msg("Proxy von %s entfernt." % unit) def restart_now(self, units): if not self.confirm("%s jetzt neu starten?" % ", ".join(units)): return self.set_msg("Starte neu ...", self.c_dim) self.draw_main() results = restart_units(units) self.changed -= {u for u, ok, _ in results if ok} self.refresh_units() self.message("Neustart", ["%s: %s" % (u, t) for u, ok, t in results], error=not all(ok for _, ok, _ in results)) def ask_restart_on_exit(self): """Liefert die Liste der neu zu startenden Dienste.""" changed = sorted(self.changed) if not changed or self.args.restart is False: return [] if self.args.restart is True: return changed chosen = self.checklist( "Geänderte Dienste neu starten?", ["Die neuen Proxy-Einstellungen gelten erst nach einem Neustart.", "Leertaste: an/aus a: alle/keine Enter: neu starten Esc: nicht neu starten"], changed) return chosen or [] # ---------- Dialoge ---------- def box(self, title, lines, height_extra=2): h, w = self.scr.getmaxyx() bw = min(w - 4, max([len(title) + 6] + [len(l) + 4 for l in lines] + [40])) bh = min(h - 2, len(lines) + 2 + height_extra) y, x = max(0, (h - bh) // 2), max(0, (w - bw) // 2) win = curses.newwin(bh, bw, y, x) win.keypad(True) win.erase() try: win.box() except curses.error: pass put(win, 0, 2, " %s " % title, curses.A_BOLD) return win, bh, bw def message(self, title, lines, error=False): win, bh, bw = self.box(title, lines) for i, l in enumerate(lines[:bh - 4]): put(win, 1 + i, 2, l, self.c_err if error else 0, bw - 4) put(win, bh - 2, 2, "[ OK – beliebige Taste ]", curses.A_REVERSE) win.refresh() win.get_wch() del win self.scr.touchwin() def confirm(self, text): win, bh, bw = self.box("Bestätigen", [text]) put(win, 1, 2, text, 0, bw - 4) put(win, bh - 2, 2, "[ j = Ja ] [ n = Nein ]", curses.A_BOLD) win.refresh() while True: k = win.get_wch() if k in ("j", "J", "y", "Y"): res = True break if k in ("n", "N", "\x1b", "q"): res = False break del win self.scr.touchwin() return res def checklist(self, title, header, items): selected = set(items) cur, top = 0, 0 while True: h, w = self.scr.getmaxyx() self.scr.erase() self.title_bar(title) for i, l in enumerate(header): put(self.scr, 2 + i, 2, l) list_top = 3 + len(header) list_h = max(1, h - list_top - 2) if cur < top: top = cur elif cur >= top + list_h: top = cur - list_h + 1 for i in range(list_h): idx = top + i if idx >= len(items): break line = " [%s] %s" % ("x" if items[idx] in selected else " ", items[idx]) put(self.scr, list_top + i, 2, line.ljust(w - 4), self.c_sel if idx == cur else 0) self.key_bar("↑↓ Wählen Leertaste an/aus a alle/keine Enter neu starten Esc abbrechen") self.scr.refresh() k = self.scr.get_wch() if k == curses.KEY_UP: cur = max(0, cur - 1) elif k == curses.KEY_DOWN: cur = min(len(items) - 1, cur + 1) elif k == " ": selected ^= {items[cur]} elif k in ("a", "A"): selected = set() if len(selected) == len(items) else set(items) elif is_enter(k): return [i for i in items if i in selected] elif k in ("\x1b", "q"): return None def show_help(self): lines = [ "Markierungen in der Liste:", " G Dienst nutzt den globalen Proxy (alle Felder leer)", " E Dienst hat eigene Werte (leere Felder erben global)", " * in dieser Sitzung geändert, Neustart noch offen", "", "Vererbung: Leere Felder beim Dienst übernehmen den globalen Wert.", "Benutzer/Kennwort werden nur geerbt, wenn auch der Host global ist.", "Eigener Host ohne Benutzer = Proxy ohne Anmeldung.", "", "Modus (pro Dienst):", " Proxy-Variablen setzt http_proxy & Co. – nur für Programme, die sie beachten", " proxychains leitet ALLE TCP-Verbindungen des Dienstes über den Proxy", " (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.", "Nach jeder Änderung wird 'systemctl daemon-reload' ausgeführt.", "Beim Beenden kann man geänderte Dienste neu starten lassen.", ] self.message("Hilfe", lines) # ---------- Formular ---------- def form(self, unit, settings, is_global, exists): g = self.cfg.global_ def ph(key): if is_global: return "" if key in ("user", "password"): return "(global, falls Host leer)" if g[key] else "" return ("global: " + g[key]) if g[key] else "" if is_global: type_choices = PROXY_TYPES tval = settings["type"] or "http" else: type_choices = [""] + PROXY_TYPES tval = settings["type"] fields = [] if not is_global: fields.append(FormField("mode", "Modus", settings["mode"] or "env", kind="choice", choices=MODES, display=MODE_LABELS)) fields.append(FormField("type", "Proxy-Typ", tval, kind="choice", choices=type_choices)) for key in ("host", "port", "user", "password", "no_proxy"): fields.append(FormField(key, FIELD_LABELS[key], settings[key], secret=(key == "password"), placeholder=ph(key))) buttons = [FormField("save", "Speichern", "", kind="button"), FormField("cancel", "Abbrechen", "", kind="button")] if exists: buttons.append(FormField("remove", "Proxy entfernen" if not is_global else "Global löschen", "", kind="button")) items = fields + buttons cur = [f.key for f in fields].index("host") error = "" def collect(): return {f.key: f.value for f in fields} while True: scr = self.scr scr.erase() h, w = scr.getmaxyx() if is_global: self.title_bar("Globaler Proxy") put(scr, 2, 2, "Gilt für alle Dienste, deren Felder leer sind.", self.c_dim) else: info = self.units.get(unit, {}) self.title_bar("Proxy für %s" % unit) desc = info.get("desc") or "" put(scr, 2, 2, "%s [%s]" % (desc, info.get("active") or info.get("file") or "?"), self.c_dim) label_w = 12 input_x = 2 + label_w + 3 input_w = max(10, min(w - input_x - 2, 60)) cursor_pos = None step = 2 if h >= 4 + len(fields) * 2 + 9 else 1 for i, f in enumerate(fields): y = 4 + i * step focused = (i == cur) put(scr, y, 2, f.label.rjust(label_w), curses.A_BOLD if focused else 0) put(scr, y, 2 + label_w, " : ") if f.kind == "choice": if f.display: shown = f.display.get(f.value, f.value) else: shown = f.value if f.value else "(global: %s)" % (g["type"] or "http") text = "◀ %s ▶" % shown put(scr, y, input_x, text.ljust(input_w), self.c_sel if focused else curses.A_UNDERLINE) else: val = "*" * len(f.value) if f.secret else f.value if f.pos < f.off: f.off = f.pos if f.pos - f.off >= input_w: f.off = f.pos - input_w + 1 vis = val[f.off:f.off + input_w] attr = self.c_sel if focused else curses.A_UNDERLINE put(scr, y, input_x, vis.ljust(input_w), attr) if not f.value and f.placeholder: put(scr, y, input_x, f.placeholder[:input_w], attr | curses.A_DIM) if focused: cursor_pos = (y, input_x + f.pos - f.off) by = 4 + len(fields) * step bx = input_x for j, b in enumerate(buttons): focused = (len(fields) + j == cur) label = "< %s >" % b.label put(scr, by, bx, label, self.c_sel if focused else curses.A_BOLD) bx += len(label) + 2 # Vorschau py = by + 2 try: cand = validate_settings(collect(), is_global) if is_global: preview = proxy_url(cand, mask=True) if cand["host"] else "(kein globaler Proxy)" else: eff, _ = resolve(cand, g) preview = proxy_url(eff, mask=True) + mode_suffix(eff) if eff["no_proxy"]: preview += " No-Proxy: " + eff["no_proxy"] put(scr, py, 2, "Ergebnis: ", curses.A_BOLD) put(scr, py, 12, preview, self.c_ok) except ProxyError as e: put(scr, py, 2, "Ergebnis: ", curses.A_BOLD) put(scr, py, 12, str(e), self.c_err) if error: put(scr, py + 1, 2, error, self.c_err) if not is_global: put(scr, py + 2, 2, "Leere Felder = globaler Wert. Benutzer/Kennwort werden nur geerbt,", self.c_dim) put(scr, py + 3, 2, "wenn auch der Host leer ist; eigener Host ohne Benutzer = ohne Anmeldung.", self.c_dim) if fields[0].value == "proxychains": put(scr, py + 4, 2, "proxychains: alle TCP-Verbindungen über den Proxy; direkt bleiben localhost,", self.c_dim) put(scr, py + 5, 2, "private Netze, DNS (Port 53) und IP-/Netz-Einträge aus No-Proxy.", self.c_dim) 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") if cursor_pos and items[cur].kind == "text": self.cursor(True) try: scr.move(*cursor_pos) except curses.error: pass else: self.cursor(False) scr.refresh() try: k = scr.get_wch() except curses.error: continue item = items[cur] action = None if k == curses.KEY_RESIZE: continue elif k == "\x1b": action = "cancel" elif k == curses.KEY_F2: action = "save" elif k in ("\t", curses.KEY_DOWN): cur = (cur + 1) % len(items) elif k in (curses.KEY_BTAB, curses.KEY_UP): cur = (cur - 1) % len(items) elif is_enter(k): if item.kind == "button": action = item.key else: cur = (cur + 1) % len(items) elif item.kind == "button": if k == curses.KEY_RIGHT: cur = min(len(items) - 1, cur + 1) elif k == curses.KEY_LEFT: cur = max(len(fields), cur - 1) elif k == " ": action = item.key elif item.kind == "choice": idx = item.choices.index(item.value) if item.value in item.choices else 0 if k in (curses.KEY_RIGHT, " ", "+"): item.value = item.choices[(idx + 1) % len(item.choices)] elif k in (curses.KEY_LEFT, "-"): item.value = item.choices[(idx - 1) % len(item.choices)] else: # Textfeld if k == curses.KEY_LEFT: item.pos = max(0, item.pos - 1) elif k == curses.KEY_RIGHT: item.pos = min(len(item.value), item.pos + 1) elif k in (curses.KEY_HOME, "\x01"): item.pos = 0 elif k in (curses.KEY_END, "\x05"): item.pos = len(item.value) elif is_backspace(k): if item.pos > 0: item.value = item.value[:item.pos - 1] + item.value[item.pos:] item.pos -= 1 elif k == curses.KEY_DC: item.value = item.value[:item.pos] + item.value[item.pos + 1:] elif k == "\x15": # Strg+U item.value, item.pos, item.off = "", 0, 0 elif isinstance(k, str) and k.isprintable(): item.value = item.value[:item.pos] + k + item.value[item.pos:] item.pos += len(k) if action == "cancel": self.cursor(False) return None if action == "remove": self.cursor(False) return "remove" if action == "save": try: cand = validate_settings(collect(), is_global) if not is_global: eff, _ = resolve(cand, g) check_mode(eff) except ProxyError as e: error = "Nicht gespeichert: %s" % e continue self.cursor(False) return cand error = "" def run_tui(cfg, args): os.environ.setdefault("ESCDELAY", "25") try: to_restart = curses.wrapper(lambda scr: TUI(scr, cfg, args).run()) except KeyboardInterrupt: to_restart = [] if to_restart: print("Starte geänderte Dienste neu:") return 0 if print_restart_results(restart_units(to_restart)) else 1 return 0 # -------------------------------------------------------------------------- # Kommandozeile # -------------------------------------------------------------------------- def print_restart_results(results): ok_all = True for unit, ok, text in results: print(" %-40s %s" % (unit, text)) ok_all &= ok return ok_all def finish(changed, warnings, args): for w in warnings: print("Warnung: " + w, file=sys.stderr) if not changed: print("Keine Änderungen an Diensten nötig.") return 0 print("Geänderte Dienste: " + ", ".join(sorted(changed))) if not args.no_reload: daemon_reload() print("systemd neu geladen (daemon-reload).") restart = args.restart if restart is None and sys.stdin.isatty() and sys.stdout.isatty(): try: ans = input("Geänderte Dienste jetzt neu starten? [j/N] ").strip().lower() except EOFError: ans = "" restart = ans in ("j", "ja", "y", "yes") if restart: return 0 if print_restart_results(restart_units(changed)) else 1 print("Hinweis: Die Änderungen gelten erst nach einem Neustart, z.B.:") print(" systemctl restart " + " ".join(sorted(changed))) return 0 def fields_from_args(args): """Liefert nur die per Parameter angegebenen Felder.""" vals = {} if args.proxy is not None: vals.update(parse_proxy_url(args.proxy)) for key in ("type", "host", "port", "user", "password", "no_proxy", "mode"): v = getattr(args, key) if v is not None: vals[key] = v if args.password_stdin: line = sys.stdin.readline() vals["password"] = line.rstrip("\r\n") return vals def cmd_global(cfg, args): if args.clear: cfg.global_ = empty_settings() else: vals = fields_from_args(args) if not vals: return cmd_show(cfg, argparse.Namespace(unit=None)) new = dict(cfg.global_) if not new["host"] and not new["no_proxy"] and "no_proxy" not in vals: new["no_proxy"] = DEFAULT_NO_PROXY new.update(vals) cfg.global_ = validate_settings(new, is_global=True) cfg.save() print("Globaler Proxy: " + (proxy_url(cfg.global_, mask=True) if cfg.global_["host"] else "(gelöscht)")) changed, warnings = cfg.apply_all() return finish(changed, warnings, args) def cmd_set(cfg, args): vals = fields_from_args(args) units = [normalize_unit(u) for u in args.units] staged = {} for unit in units: base = empty_settings() if args.reset or unit not in cfg.services else dict(cfg.services[unit]) base.update(vals) svc = validate_settings(base, is_global=False) try: eff, _ = resolve(svc, cfg.global_) check_mode(eff) except ProxyError as e: raise ProxyError("%s: %s" % (unit, e)) staged[unit] = svc print("%s -> %s%s%s" % (unit, proxy_url(eff, mask=True), " (global)" if is_inheriting(svc) else "", mode_suffix(eff))) cfg.services.update(staged) cfg.save() changed, warnings = set(), [] for unit in units: ch, warn = cfg.apply_unit(unit) if ch: changed.add(unit) if warn: warnings.append(warn) return finish(changed, warnings, args) def cmd_unset(cfg, args): if args.all: units = sorted(set(cfg.services) | cfg.managed_dropins()) elif args.units: units = [normalize_unit(u) for u in args.units] else: raise ProxyError("Bitte Dienst(e) angeben oder --all verwenden") for unit in units: if unit not in cfg.services and unit not in cfg.managed_dropins(): print("%s: hat keinen Proxy" % unit) cfg.services.pop(unit, None) cfg.save() changed = set() for unit in units: ch, _ = cfg.apply_unit(unit) if ch: changed.add(unit) print("%s: Proxy entfernt" % unit) return finish(changed, [], args) def cmd_apply(cfg, args): changed, warnings = cfg.apply_all() return finish(changed, warnings, args) def cmd_list(cfg, args): print("Global: " + (proxy_url(cfg.global_, mask=True) if cfg.global_["host"] else "(nicht gesetzt)")) if cfg.global_["no_proxy"]: print(" No-Proxy: " + cfg.global_["no_proxy"]) print() if args.all: units = list_system_services() for u in cfg.services: units.setdefault(u, {"active": "", "file": "?"}) names = sorted(units) else: units = {} names = sorted(cfg.services) if not names: print("Keine Dienste mit Proxy konfiguriert. (--all zeigt alle Dienste)") return 0 width = max([len(n) for n in names] + [6]) print(" %-*s %-10s %s" % (width, "Dienst", "Status", "Proxy")) for name in names: svc = cfg.services.get(name) state = (units.get(name) or {}).get("active") or (unit_active(name) if not args.all else "") if svc is None: mark, text = " ", "" else: mark = "G" if is_inheriting(svc) else "E" try: eff, _ = resolve(svc, cfg.global_) text = proxy_url(eff, mask=True) + mode_suffix(eff) except ProxyError as e: text = "FEHLER: %s" % e print(" %s %-*s %-10s %s" % (mark, width, name, state, text)) print("\nG = nutzt globalen Proxy, E = eigene Werte") return 0 def cmd_show(cfg, args): if not args.unit: g = cfg.global_ print("Globaler Proxy") if not g["host"]: print(" (nicht gesetzt)") for key in FIELDS: val = g[key] if key == "password" and val: val = "****" print(" %-10s %s" % (FIELD_LABELS[key], val or "-")) if g["host"]: print(" %-10s %s" % ("URL", proxy_url(g, mask=True))) return 0 unit = normalize_unit(args.unit) svc = cfg.services.get(unit) print("Dienst: %s" % unit) print("Status: %s" % unit_active(unit)) if svc is None: print("Proxy: keiner") return 0 print("Herkunft: %s" % ("nutzt globalen Proxy" if is_inheriting(svc) else "eigene Werte (leere Felder = global)")) try: eff, src = resolve(svc, cfg.global_) except ProxyError as e: print("FEHLER: %s" % e) return 1 print("Modus: %s" % MODE_LABELS[eff["mode"]]) for key in FIELDS: val = eff[key] if key == "password" and val: val = "****" print(" %-10s %-35s (%s)" % (FIELD_LABELS[key], val or "-", src[key])) print("Proxy-URL: %s" % proxy_url(eff, mask=True)) print("Drop-in: %s" % cfg.dropin_file(unit)) print("Env-Datei: %s" % cfg.env_file(unit)) if eff["mode"] == "proxychains": print("proxychains-Konfiguration: %s" % cfg.pc_file(unit)) return 0 def build_parser(): def common(sub): d = argparse.SUPPRESS if sub else None p = argparse.ArgumentParser(add_help=False) g = p.add_mutually_exclusive_group() g.add_argument("--restart", dest="restart", action="store_const", const=True, default=d) g.add_argument("--no-restart", dest="restart", action="store_const", const=False, default=d) p.add_argument("--no-reload", action="store_true", default=argparse.SUPPRESS if sub else False) p.add_argument("--config-dir", default=d) p.add_argument("--systemd-dir", default=d) return p fields = argparse.ArgumentParser(add_help=False) fields.add_argument("--proxy") fields.add_argument("--type", type=str.lower, choices=PROXY_TYPES) fields.add_argument("--host") fields.add_argument("--port") fields.add_argument("--user") fields.add_argument("--password") fields.add_argument("--password-stdin", action="store_true") fields.add_argument("--no-proxy", dest="no_proxy") fields.add_argument("--mode", type=str.lower, choices=MODES) parser = argparse.ArgumentParser(prog=PROG, add_help=False, parents=[common(False)]) parser.add_argument("-V", "--version", action="version", version="%s %s" % (PROG, VERSION)) sub = parser.add_subparsers(dest="command") c = common(True) sub.add_parser("tui", add_help=False, parents=[c]) sub.add_parser("help", add_help=False) p = sub.add_parser("list", add_help=False, parents=[c]) p.add_argument("--all", action="store_true") p = sub.add_parser("show", add_help=False, parents=[c]) p.add_argument("unit", nargs="?") p = sub.add_parser("global", add_help=False, parents=[c, fields]) p.add_argument("--clear", action="store_true") p = sub.add_parser("set", add_help=False, parents=[c, fields]) p.add_argument("units", nargs="+", metavar="DIENST") p.add_argument("--reset", action="store_true") p = sub.add_parser("unset", add_help=False, parents=[c]) p.add_argument("units", nargs="*", metavar="DIENST") p.add_argument("--all", action="store_true") sub.add_parser("apply", add_help=False, parents=[c]) return parser def main(argv=None): argv = sys.argv[1:] if argv is None else argv if "-h" in argv or "--help" in argv or (argv and argv[0] == "help"): print(HELP_TEXT) return 0 args = build_parser().parse_args(argv) command = args.command or "tui" 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: print("%s: Bitte als root ausführen (z.B. sudo %s)." % (PROG, PROG), file=sys.stderr) return 1 try: locale.setlocale(locale.LC_ALL, "") except locale.Error: pass try: cfg.load() 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, }[command] return handler(cfg, args) except ProxyError as e: print("%s: Fehler: %s" % (PROG, e), file=sys.stderr) return 1 except OSError as e: print("%s: Fehler: %s" % (PROG, e), file=sys.stderr) return 1 if __name__ == "__main__": sys.exit(main())