#!/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 = [ "sudo", "-kS", "openconnect", "--protocol=anyconnect", "--user", self.user, "--passwd-on-stdin", "--pid-file", PIDFILE, "--force-dpd", "0", "--script", vpn_slice_arg, server, ] self.log(f"$ openconnect --protocol=anyconnect --user {self.user} --script '{vpn_slice_arg}' {server}\n\n") p = subprocess.Popen( cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=0, ) # wie die funktionierende printf-Pipe: sudo liest Zeile 1, openconnect # Zeile 2 (--passwd-on-stdin); stdin schließen, damit openconnect EOF sieht p.stdin.write(f"{self.sudo_pw}\n{self.vpn_pw}\n".encode()) p.stdin.flush() p.stdin.close() 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 sudo_in = (self.sudo_pw + "\n").encode() self.log(f"Sending SIGINT to pid {pid}...\n") subprocess.run( ["sudo", "-S", "kill", "-INT", str(pid)], input=sudo_in, stderr=subprocess.DEVNULL, ) start = time.time() while time.time() - start < 12: if not _pid_is_openconnect(pid): subprocess.run( ["sudo", "-S", "rm", "-f", PIDFILE], input=sudo_in, 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( ["sudo", "-S", "kill", "-TERM", str(pid)], input=sudo_in, stderr=subprocess.DEVNULL, ) if 8.0 < elapsed < 8.5: self.log("Escalating: SIGKILL\n") subprocess.run( ["sudo", "-S", "kill", "-KILL", str(pid)], input=sudo_in, 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())