initial
This commit is contained in:
marcus hinz
Executable
+575
@@ -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())
|
||||
Reference in New Issue
Block a user