1452 lines
54 KiB
Python
Executable File
1452 lines
54 KiB
Python
Executable File
#!/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 json
|
||
import locale
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
from urllib.parse import quote, unquote, urlsplit
|
||
|
||
VERSION = "1.0.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"]
|
||
|
||
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 die Umgebungsvariablen
|
||
http_proxy, https_proxy, ftp_proxy, all_proxy und no_proxy (jeweils auch in
|
||
GROSSBUCHSTABEN) setzt. Alle Programme, die diese Variablen beachten (curl,
|
||
wget, apt, Python, Go, Node, ...), laufen dann über den angegebenen Proxy.
|
||
|
||
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
|
||
|
||
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
|
||
|
||
# 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)
|
||
{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():
|
||
return {k: "" for k in FIELDS}
|
||
|
||
|
||
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"])
|
||
|
||
if is_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"
|
||
|
||
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)")
|
||
return eff, src
|
||
|
||
|
||
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)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 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.systemd_dir = systemd_dir
|
||
self.global_ = empty_settings()
|
||
self.services = {}
|
||
|
||
# Pfade
|
||
def env_file(self, unit):
|
||
return os.path.join(self.env_dir, re.sub(r"[^A-Za-z0-9:_.@-]", "_", unit) + ".env")
|
||
|
||
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):
|
||
s = empty_settings()
|
||
for k in FIELDS:
|
||
if isinstance(d, dict) and isinstance(d.get(k), str):
|
||
s[k] = d[k]
|
||
return s
|
||
|
||
self.global_ = clean(data.get("global"))
|
||
self.services = {}
|
||
for unit, s in (data.get("services") or {}).items():
|
||
if UNIT_RE.match(unit):
|
||
self.services[unit] = clean(s)
|
||
|
||
def save(self):
|
||
os.makedirs(self.config_dir, mode=0o700, exist_ok=True)
|
||
data = {
|
||
"version": 1,
|
||
"global": self.global_,
|
||
"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):
|
||
url = proxy_url(eff)
|
||
lines = [
|
||
"# Erzeugt von %s fuer %s - Aenderungen werden ueberschrieben." % (PROG, unit),
|
||
"# Bearbeiten mit: %s set %s ..." % (PROG, unit),
|
||
]
|
||
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_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))
|
||
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
|
||
try:
|
||
eff, _ = resolve(svc, self.global_)
|
||
except ProxyError as e:
|
||
return self.remove_unit_files(unit), "%s: %s – Proxy für diesen Dienst deaktiviert" % (unit, e)
|
||
os.makedirs(self.env_dir, mode=0o700, exist_ok=True)
|
||
changed = write_if_different(self.env_file(unit), self.render_env(unit, eff), 0o600)
|
||
changed |= write_if_different(self.dropin_file(unit), self.render_dropin(unit), 0o644)
|
||
return changed, None
|
||
|
||
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=""):
|
||
self.key = key
|
||
self.label = label
|
||
self.kind = kind # text | choice | button
|
||
self.value = value
|
||
self.choices = choices 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), self.c_global
|
||
return "eigen → " + proxy_url(eff, mask=True), 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.",
|
||
"",
|
||
"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 = [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 = 1 # 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
|
||
for i, f in enumerate(fields):
|
||
y = 4 + i * 2
|
||
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":
|
||
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) * 2
|
||
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)
|
||
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 + 3, 2, "Leere Felder = globaler Wert. Benutzer/Kennwort werden nur geerbt,", self.c_dim)
|
||
put(scr, py + 4, 2, "wenn auch der Host leer ist; eigener Host ohne Benutzer = ohne Anmeldung.", self.c_dim)
|
||
else:
|
||
put(scr, py + 3, 2, "Benutzer/Kennwort leer = Proxy ohne Anmeldung.", self.c_dim)
|
||
|
||
self.key_bar("Tab/↑↓ Feld ←→ Typ 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:
|
||
resolve(cand, g)
|
||
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"):
|
||
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_)
|
||
except ProxyError as e:
|
||
raise ProxyError("%s: %s" % (unit, e))
|
||
staged[unit] = svc
|
||
print("%s -> %s%s" % (unit, proxy_url(eff, mask=True),
|
||
" (global)" if is_inheriting(svc) else ""))
|
||
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)
|
||
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("Modus: %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
|
||
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))
|
||
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")
|
||
|
||
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())
|