vpn/wuerth_vpn

#!/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",
            "--no-xmlpost",
            "--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())