commit f19d52bbe5d440db4c105fef066b463cc9ef8b32 Author: marcus hinz Date: Sun Jun 21 10:48:13 2026 +0200 initial diff --git a/wuerth_vpn b/wuerth_vpn new file mode 100755 index 0000000..25ab4ab --- /dev/null +++ b/wuerth_vpn @@ -0,0 +1,575 @@ +#!/usr/bin/env python3 +""" +Wuerth VPN – PyQt6 macOS client +Requires: pip3 install PyQt6 --break-system-packages + +Uses vpn-slice for split tunneling — no manual routing needed. +""" + +import os +import subprocess +import time +from typing import Optional + +from PyQt6.QtCore import QThread, QObject, pyqtSignal +from PyQt6.QtGui import QTextCursor +from PyQt6.QtWidgets import ( + QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, + QLabel, QLineEdit, QPushButton, QTextEdit, QFrame, QSizePolicy, + QInputDialog, QMessageBox, +) + +# ===== CONFIG ================================================================= +DEFAULT_SERVER = "loki.witglobal.net" +DEFAULT_USER = "ex08802409" +PIDFILE = "/tmp/openconnect-anyconnect.pid" +WINDOW_TITLE = "WuerthVPN" + +DEFAULT_HOSTS = """\ +gitops.apps.ocp-dev01.wgn.wuerth.com +console-openshift-console.apps.ocp-dev01.wgn.wuerth.com +oauth-openshift.apps.ocp-dev01.wgn.wuerth.com +apm-dev.wgn.wuerth.com +odatang9.wgn.wuerth.com +fs.wgn.wuerth.com +fahrzeugeinrichtung.apps.ocp-dev01.wgn.wuerth.com +api.ocp-dev01.wgn.wuerth.com +console-openshift-console.apps.ocp-01.wgn.wuerth.com +gitops.apps.ocp-01.wgn.wuerth.com +oauth-openshift.apps.ocp-01.wgn.wuerth.com +api.ocp-01.wgn.wuerth.com +proxy.wgs.wuerth.com +odatang8.wgn.wuerth.com +fahrzeugeinrichtung.apps.ocp-01.wgn.wuerth.com""" +# ============================================================================= + +_password_cache: Optional[str] = None +_sudo_pw_cache: Optional[str] = None + + +def pw_load() -> Optional[str]: + return _password_cache + + +def pw_save(pw: str): + global _password_cache + _password_cache = pw + + +def sudo_pw_load() -> Optional[str]: + return _sudo_pw_cache + + +def sudo_pw_save(pw: str): + global _sudo_pw_cache + _sudo_pw_cache = pw + + +# ---------- PID / state ------------------------------------------------------- +def _read_pidfile() -> Optional[int]: + try: + with open(PIDFILE) as f: + return int(f.read().strip()) + except Exception: + pass + + try: + out = subprocess.check_output(["pgrep", "-x", "openconnect"], stderr=subprocess.DEVNULL) + return int(out.decode().strip().splitlines()[0]) + except Exception: + return None + + +def _pid_is_openconnect(pid: int) -> bool: + try: + out = subprocess.check_output( + ["ps", "-p", str(pid), "-o", "comm="], + stderr=subprocess.DEVNULL, + ) + return b"openconnect" in out + except Exception: + return False + + +def is_connected() -> bool: + pid = _read_pidfile() + return bool(pid and _pid_is_openconnect(pid)) + + +# ---------- Worker thread ----------------------------------------------------- +class VpnWorker(QObject): + output = pyqtSignal(str) + done = pyqtSignal(bool, str) + + def __init__(self, action: str, server: str, user: str, vpn_pw: str, hosts: list[str], sudo_pw: str = ""): + super().__init__() + self.action = action + self.server = server + self.user = user + self.vpn_pw = vpn_pw + self.hosts = hosts + self.sudo_pw = sudo_pw + + def log(self, text: str): + self.output.emit(text) + + def run(self): + if self.action == "connect": + self._connect() + else: + self._disconnect() + + def _connect(self): + subprocess.run( + ["sudo", "-S", "rm", "-f", PIDFILE], + input=(self.sudo_pw + "\n").encode(), + stderr=subprocess.DEVNULL, + ) + + vpn_slice_arg = "vpn-slice " + " ".join(self.hosts) + + server = self.server + if not server.startswith("https://"): + server = "https://" + server + + cmd_str = ( + f"printf '%s\\n%s\\n' '{self.sudo_pw}' '{self.vpn_pw}'" + f" | sudo -kS openconnect" + f" --protocol=anyconnect" + f" --user {self.user}" + f" --passwd-on-stdin" + f" --pid-file {PIDFILE}" + f" --force-dpd 0" + f" --script '{vpn_slice_arg}'" + f" {server}" + ) + + self.log(f"$ openconnect --protocol=anyconnect --user {self.user} --script '{vpn_slice_arg}' {server}\n\n") + + p = subprocess.Popen( + cmd_str, + shell=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + bufsize=0, + ) + + start = time.time() + results_seen = 0 + expected = len(self.hosts) + + while True: + chunk = p.stdout.read(256) + if chunk: + text = chunk.decode(errors="replace") + self.log(text) + results_seen += text.count("Got results:") + + if results_seen >= expected: + time.sleep(0.5) + self.done.emit(True, "Connected") + return + + if p.poll() is not None: + self.done.emit(False, "Connect failed") + return + + if time.time() - start > 60: + self.done.emit(False, "Timeout") + return + + time.sleep(0.05) + + def _disconnect(self): + pid = _read_pidfile() + if not pid: + self.log("Not connected.\n") + self.done.emit(False, "") + return + + self.log(f"Sending SIGINT to pid {pid}...\n") + subprocess.run( + f"echo '{self.sudo_pw}' | sudo -S kill -INT {pid}", + shell=True, + stderr=subprocess.DEVNULL, + ) + + start = time.time() + while time.time() - start < 12: + if not _pid_is_openconnect(pid): + subprocess.run( + f"echo '{self.sudo_pw}' | sudo -S rm -f {PIDFILE}", + shell=True, + stderr=subprocess.DEVNULL, + ) + self.log("Disconnected.\n") + self.done.emit(False, "Disconnected") + return + + elapsed = time.time() - start + if 3.0 < elapsed < 3.5: + self.log("Escalating: SIGTERM\n") + subprocess.run( + f"echo '{self.sudo_pw}' | sudo -S kill -TERM {pid}", + shell=True, + stderr=subprocess.DEVNULL, + ) + if 8.0 < elapsed < 8.5: + self.log("Escalating: SIGKILL\n") + subprocess.run( + f"echo '{self.sudo_pw}' | sudo -S kill -KILL {pid}", + shell=True, + stderr=subprocess.DEVNULL, + ) + time.sleep(0.2) + + self.done.emit(False, "Disconnect timeout") + + +# ---------- Stylesheet -------------------------------------------------------- +STYLE = """ +QMainWindow, QWidget#root { background: #1a1a1f; } + +QFrame#card { + background: #22222a; + border: 1px solid #2e2e38; + border-radius: 10px; +} +QLineEdit { + background: #16161c; + color: #e8e0d4; + border: 1px solid #2e2e38; + border-radius: 6px; + padding: 6px 10px; + font-size: 13px; +} +QLineEdit:focus { border-color: #c8a96e; } +QLineEdit:disabled { color: #555560; } + +QTextEdit#hosts { + background: #16161c; + color: #e8e0d4; + border: 1px solid #2e2e38; + border-radius: 6px; + font-family: "Menlo", "Monaco", monospace; + font-size: 12px; + padding: 6px; +} +QTextEdit#hosts:focus { border-color: #c8a96e; } + +QPushButton#btn_connect { + background: #c8a96e; color: #1a1a1f; + border: none; border-radius: 7px; + padding: 8px 24px; font-size: 13px; font-weight: 600; +} +QPushButton#btn_connect:hover { background: #d9bb82; } +QPushButton#btn_connect:pressed { background: #b8994e; } +QPushButton#btn_connect:disabled { background: #3a3a44; color: #666; } + +QPushButton#btn_disconnect { + background: #3a2020; color: #e07070; + border: 1px solid #5a3030; border-radius: 7px; + padding: 8px 24px; font-size: 13px; font-weight: 600; +} +QPushButton#btn_disconnect:hover { background: #4a2828; } +QPushButton#btn_disconnect:disabled { background: #2a1818; color: #555; } + +QTextEdit#console { + background: #0f0f14; color: #a8c0a0; + border: 1px solid #2e2e38; border-radius: 8px; + font-family: "Menlo", "Monaco", monospace; + font-size: 11px; padding: 8px; +} +""" + + +# ---------- Main Window ------------------------------------------------------- +class MainWindow(QMainWindow): + def __init__(self): + super().__init__() + self.setWindowTitle(WINDOW_TITLE) + self.resize(820, 640) + self.setStyleSheet(STYLE) + + self._connected = False + self._thread = None + self._worker = None + self._pending_close = False + + root = QWidget() + root.setObjectName("root") + self.setCentralWidget(root) + + layout = QVBoxLayout(root) + layout.setContentsMargins(20, 20, 20, 20) + layout.setSpacing(14) + + hdr = QHBoxLayout() + self._dot = QLabel("●") + self._dot.setStyleSheet("color: #555560; font-size: 18px;") + hdr.addWidget(self._dot) + + lbl = QLabel(WINDOW_TITLE) + lbl.setStyleSheet("color: #e8e0d4; font-size: 16px; font-weight: 600;") + hdr.addWidget(lbl) + + hdr.addStretch() + + self._status_lbl = QLabel("") + self._status_lbl.setStyleSheet("color: #8a8a9a; font-size: 12px;") + hdr.addWidget(self._status_lbl) + + layout.addLayout(hdr) + + card = QFrame() + card.setObjectName("card") + cl = QVBoxLayout(card) + cl.setContentsMargins(16, 14, 16, 14) + cl.setSpacing(10) + + def field_row(label, widget): + h = QHBoxLayout() + lb = QLabel(label) + lb.setStyleSheet("color: #8a8a9a; font-size: 12px;") + lb.setFixedWidth(110) + h.addWidget(lb) + h.addWidget(widget) + return h + + self.server_edit = QLineEdit(DEFAULT_SERVER) + self.user_edit = QLineEdit(DEFAULT_USER) + + self.pw_edit = QLineEdit() + self.pw_edit.setEchoMode(QLineEdit.EchoMode.Password) + self.pw_edit.setPlaceholderText("VPN password") + + self.sudo_pw_edit = QLineEdit() + self.sudo_pw_edit.setEchoMode(QLineEdit.EchoMode.Password) + self.sudo_pw_edit.setPlaceholderText("sudo password") + + self.hosts_edit = QTextEdit() + self.hosts_edit.setObjectName("hosts") + self.hosts_edit.setPlainText(DEFAULT_HOSTS) + self.hosts_edit.setFixedHeight(80) + self.hosts_edit.setPlaceholderText("one hostname per line") + + cl.addLayout(field_row("Server", self.server_edit)) + cl.addLayout(field_row("Username", self.user_edit)) + cl.addLayout(field_row("VPN Password", self.pw_edit)) + cl.addLayout(field_row("Sudo Password", self.sudo_pw_edit)) + + from PyQt6.QtCore import Qt + + hosts_row = QHBoxLayout() + hosts_lbl = QLabel("Hosts") + hosts_lbl.setStyleSheet("color: #8a8a9a; font-size: 12px;") + hosts_lbl.setFixedWidth(110) + hosts_lbl.setAlignment(Qt.AlignmentFlag.AlignTop) + hosts_row.addWidget(hosts_lbl) + hosts_row.addWidget(self.hosts_edit) + cl.addLayout(hosts_row) + + layout.addWidget(card) + + btn_row = QHBoxLayout() + btn_row.addStretch() + + self.btn_connect = QPushButton("Connect") + self.btn_connect.setObjectName("btn_connect") + self.btn_connect.clicked.connect(self.on_connect) + + self.btn_disconnect = QPushButton("Disconnect") + self.btn_disconnect.setObjectName("btn_disconnect") + self.btn_disconnect.clicked.connect(self.on_disconnect) + self.btn_disconnect.setEnabled(False) + + btn_row.addWidget(self.btn_connect) + btn_row.addWidget(self.btn_disconnect) + layout.addLayout(btn_row) + + self.console = QTextEdit() + self.console.setObjectName("console") + self.console.setReadOnly(True) + self.console.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Expanding) + layout.addWidget(self.console) + + if is_connected(): + self._set_connected(True) + self._set_status("Connected", "#6db87a") + self.log("Detected existing VPN connection.\n") + + def log(self, text: str): + self.console.moveCursor(QTextCursor.MoveOperation.End) + self.console.insertPlainText(text) + self.console.moveCursor(QTextCursor.MoveOperation.End) + + def _set_status(self, text: str, color: str = "#8a8a9a"): + self._status_lbl.setText(text) + self._status_lbl.setStyleSheet(f"color: {color}; font-size: 12px;") + + def _set_connected(self, connected: bool): + self._connected = connected + self._dot.setStyleSheet( + f"color: {'#6db87a' if connected else '#555560'}; font-size: 18px;" + ) + self.btn_connect.setEnabled(not connected) + self.btn_disconnect.setEnabled(connected) + self.server_edit.setEnabled(not connected) + self.user_edit.setEnabled(not connected) + self.pw_edit.setEnabled(not connected) + self.sudo_pw_edit.setEnabled(not connected) + self.hosts_edit.setEnabled(not connected) + + def _get_password(self) -> Optional[str]: + pw = self.pw_edit.text().strip() + if pw: + return pw + + pw = pw_load() + if pw: + self.pw_edit.setText(pw) + return pw + + pw, ok = QInputDialog.getText( + self, + "VPN Password", + "Enter VPN password:", + QLineEdit.EchoMode.Password, + ) + if ok and pw: + pw_save(pw) + self.pw_edit.setText(pw) + return pw + + return None + + def _get_sudo_password(self) -> Optional[str]: + pw = self.sudo_pw_edit.text().strip() + if pw: + sudo_pw_save(pw) + return pw + + pw = sudo_pw_load() + if pw: + self.sudo_pw_edit.setText(pw) + return pw + + pw, ok = QInputDialog.getText( + self, + "Sudo Password", + "Enter sudo password:", + QLineEdit.EchoMode.Password, + ) + if ok and pw: + sudo_pw_save(pw) + self.sudo_pw_edit.setText(pw) + return pw + + return None + + def _get_hosts(self) -> list[str]: + lines = self.hosts_edit.toPlainText().strip().splitlines() + return [line.strip() for line in lines if line.strip()] + + def on_connect(self): + server = self.server_edit.text().strip() + user = self.user_edit.text().strip() + pw = self._get_password() + sudo_pw = self._get_sudo_password() + hosts = self._get_hosts() + + if not (server and user and pw): + self._set_status("Missing VPN credentials", "#e07070") + return + + if not sudo_pw: + self._set_status("Missing sudo password", "#e07070") + return + + if not hosts: + self._set_status("No hosts defined", "#e07070") + return + + pw_save(pw) + sudo_pw_save(sudo_pw) + + self.btn_connect.setEnabled(False) + self._set_status("Connecting…", "#c8a96e") + self._start_worker("connect", server, user, pw, hosts, sudo_pw) + + def on_disconnect(self): + sudo_pw = self._get_sudo_password() or "" + self.btn_disconnect.setEnabled(False) + self._set_status("Disconnecting…", "#c8a96e") + self._start_worker("disconnect", "", "", "", [], sudo_pw) + + def _start_worker(self, action, server, user, pw, hosts, sudo_pw=""): + self._thread = QThread() + self._worker = VpnWorker(action, server, user, pw, hosts, sudo_pw) + self._worker.moveToThread(self._thread) + self._thread.started.connect(self._worker.run) + self._worker.output.connect(self.log) + self._worker.done.connect(self._on_done) + self._worker.done.connect(self._thread.quit) + self._thread.start() + + def _on_done(self, connected: bool, message: str): + self._set_connected(connected) + + if connected: + self._set_status("Connected", "#6db87a") + elif message == "Disconnected": + self._set_status("Disconnected", "#8a8a9a") + elif message: + self._set_status(message, "#e07070") + else: + self._set_status("", "#8a8a9a") + + if self._pending_close and not connected: + self._pending_close = False + self.close() + + def closeEvent(self, event): + if not (self._connected and is_connected()): + event.accept() + return + + resp = QMessageBox.question( + self, + "VPN aktiv", + "Die VPN-Verbindung ist aktiv. Vor dem Beenden trennen?", + QMessageBox.StandardButton.Yes + | QMessageBox.StandardButton.No + | QMessageBox.StandardButton.Cancel, + QMessageBox.StandardButton.Yes, + ) + + if resp == QMessageBox.StandardButton.Cancel: + event.ignore() + return + if resp == QMessageBox.StandardButton.No: + event.accept() + return + + sudo_pw = self._get_sudo_password() + if not sudo_pw: + event.ignore() + return + + self._pending_close = True + event.ignore() + self.btn_disconnect.setEnabled(False) + self._set_status("Disconnecting…", "#c8a96e") + self._start_worker("disconnect", "", "", "", [], sudo_pw) + + +if __name__ == "__main__": + import sys + + app = QApplication(sys.argv) + app.setApplicationName(WINDOW_TITLE) + app.setDesktopFileName("WuerthVPN") + w = MainWindow() + w.show() + + sys.exit(app.exec())