#!/usr/bin/env python3 """ Heron VPN – plattformuebergreifende Tray-App (PyQt6), GlobalProtect. Baut das VPN rootlos auf (openconnect --protocol=gp + ocproxy als SOCKS5) und startet einen dedizierten Browser, der per PAC nur die internen Hosts ueber das VPN schickt und alles andere direkt. Start/Stop ueber das Tray-Icon. Eigene Ports/Profil/Bundle -> laeuft parallel zur Wuerth-App. Abhaengigkeiten: openconnect + ocproxy auf dem PATH (macOS: brew install openconnect ocproxy) ZU FUELLEN: VPN_USER, INTERNAL_GLOB, START_URL, BOOKMARKS. """ from __future__ import annotations import json import os import platform import shutil import signal import socket import subprocess import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from PyQt6.QtCore import QTimer, Qt from PyQt6.QtGui import QAction, QColor, QIcon, QPainter, QPixmap from PyQt6.QtWidgets import ( QApplication, QSystemTrayIcon, QMenu, QInputDialog, QLineEdit, QMessageBox, ) # ===== CONFIG ================================================================= APP_NAME = "Heron VPN" PROTOCOL = "gp" # GlobalProtect (Palo Alto) SERVER = "ssl.heron.at" AUTH = "saml" # "saml" = SSO via externem Browser, "password" VPN_USER = "marcus.hinz@robotunits.com" # nur fuer AUTH="password" noetig PORT = 11090 # SOCKS5 (ocproxy) – andere Ports als Wuerth PACPORT = 11091 # lokaler PAC-HTTP-Server INTERNAL_GLOB = "*.heron.at" # <-- interne Hosts anpassen (ueber VPN) PROFILE = os.path.expanduser("~/.heron-edge") START_URL = "https://robotunits.com" # <-- gewuenschte Startseite LOGFILE = os.path.expanduser("~/.heron-vpn.log") PAC_NAME = "heron.pac" # Lesezeichen, die ins frische Profil gesaet werden (Name, URL). <-- anpassen BOOKMARKS: list[tuple[str, str]] = [ ("Heron", "https://ssl.heron.at"), ] # ============================================================================= # ---------- Plattform-Abstraktion -------------------------------------------- def _system() -> str: return platform.system() # 'Darwin' | 'Linux' | 'Windows' def browser_argv() -> list[str]: s = _system() if s == "Darwin": return ["/Applications/Microsoft Edge.app/Contents/MacOS/Microsoft Edge"] if s == "Windows": for p in ( r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe", r"C:\Program Files\Microsoft\Edge\Application\msedge.exe", ): if os.path.exists(p): return [p] return ["msedge.exe"] for name in ("microsoft-edge", "microsoft-edge-stable", "msedge"): path = shutil.which(name) if path: return [path] return ["microsoft-edge"] def seed_profile(): """Saet beim ersten Start Lesezeichen + sichtbare Leiste ins Profil.""" default = os.path.join(PROFILE, "Default") bm_path = os.path.join(default, "Bookmarks") if not os.path.exists(bm_path): os.makedirs(default, exist_ok=True) children = [ {"date_added": "0", "id": str(100 + i), "name": name, "type": "url", "url": url} for i, (name, url) in enumerate(BOOKMARKS) ] data = { "checksum": "", "roots": { "bookmark_bar": {"children": children, "date_added": "0", "id": "1", "name": "Lesezeichenleiste", "type": "folder"}, "other": {"children": [], "date_added": "0", "id": "2", "name": "Weitere Lesezeichen", "type": "folder"}, "synced": {"children": [], "date_added": "0", "id": "3", "name": "Mobile Lesezeichen", "type": "folder"}, }, "version": 1, } with open(bm_path, "w") as f: json.dump(data, f, indent=2) pref_path = os.path.join(default, "Preferences") if not os.path.exists(pref_path): os.makedirs(default, exist_ok=True) with open(pref_path, "w") as f: json.dump({"bookmark_bar": {"show_on_all_tabs": True}}, f) def _tool(name: str) -> str: path = shutil.which(name) if path: return path if _system() == "Darwin": cand = f"/opt/homebrew/bin/{name}" if os.path.exists(cand): return cand return name def _external_browser() -> str: """Programm, das openconnect fuer das SAML-Login mit der SSO-URL aufruft.""" s = _system() if s == "Darwin": return "/usr/bin/open" # oeffnet Standardbrowser mit der URL if s == "Windows": return "cmd" # TODO: 'cmd /c start' verifizieren return shutil.which("xdg-open") or "xdg-open" # ---------- PAC-Server (in-process) ------------------------------------------ PAC_BODY = ( "function FindProxyForURL(url, host) {\n" f' if (shExpMatch(host, "{INTERNAL_GLOB}")) {{\n' f' return "SOCKS5 127.0.0.1:{PORT}";\n' " }\n" ' return "DIRECT";\n' "}\n" ) class _PacHandler(BaseHTTPRequestHandler): def do_GET(self): body = PAC_BODY.encode() self.send_response(200) self.send_header("Content-Type", "application/x-ns-proxy-autoconfig") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def log_message(self, *_): pass def start_pac_server() -> ThreadingHTTPServer: srv = ThreadingHTTPServer(("127.0.0.1", PACPORT), _PacHandler) threading.Thread(target=srv.serve_forever, daemon=True).start() return srv # ---------- Icon (generiertes "H"-Badge in Zustandsfarbe) -------------------- _ICON_CACHE: dict[str, QIcon] = {} def badge_icon(color: str) -> QIcon: if color in _ICON_CACHE: return _ICON_CACHE[color] pm = QPixmap(44, 44) pm.fill(Qt.GlobalColor.transparent) p = QPainter(pm) p.setRenderHint(QPainter.RenderHint.Antialiasing) p.setBrush(QColor(color)) p.setPen(Qt.PenStyle.NoPen) p.drawRoundedRect(4, 4, 36, 36, 9, 9) p.setPen(QColor("#ffffff")) f = p.font() f.setBold(True) f.setPixelSize(26) p.setFont(f) p.drawText(pm.rect(), Qt.AlignmentFlag.AlignCenter, "H") p.end() icon = QIcon(pm) _ICON_CACHE[color] = icon return icon # ---------- App -------------------------------------------------------------- class VpnTray: def __init__(self, app: QApplication): self.app = app self.oc: subprocess.Popen | None = None self.browser: subprocess.Popen | None = None self.vpn_pw: str | None = None self.state = "disconnected" self.pac = start_pac_server() self.tray = QSystemTrayIcon() self.tray.setIcon(badge_icon("#9a9a9a")) self.tray.setToolTip(f"{APP_NAME} – getrennt") self.menu = QMenu() self.act_status = QAction("Getrennt") self.act_status.setEnabled(False) self.act_connect = QAction("Verbinden") self.act_disconnect = QAction("Trennen") self.act_open = QAction("Edge oeffnen") self.act_quit = QAction("Beenden") self.act_connect.triggered.connect(self.connect) self.act_disconnect.triggered.connect(self.disconnect) self.act_open.triggered.connect(self.launch_browser) self.act_quit.triggered.connect(self.quit) for a in (self.act_status, None, self.act_connect, self.act_disconnect, self.act_open, None, self.act_quit): if a is None: self.menu.addSeparator() else: self.menu.addAction(a) self.tray.setContextMenu(self.menu) self.tray.show() self._set_state("disconnected") self.timer = QTimer() self.timer.timeout.connect(self._tick) self.timer.start(700) def _set_state(self, state: str): self.state = state color, label = { "disconnected": ("#9a9a9a", "Getrennt"), "connecting": ("#e6a817", "Verbinde…"), "connected": ("#2e7d32", "Verbunden"), "error": ("#cc3333", "Fehler"), }[state] self.tray.setIcon(badge_icon(color)) self.tray.setToolTip(f"{APP_NAME} – {label}") self.act_status.setText(label) self.act_connect.setEnabled(state in ("disconnected", "error")) self.act_disconnect.setEnabled(state in ("connecting", "connected")) self.act_open.setEnabled(state == "connected") def connect(self): base = [ _tool("openconnect"), f"--protocol={PROTOCOL}", "--script", f"{_tool('ocproxy')} -D {PORT}", "--script-tun", ] send_pw = None if AUTH == "saml": # SSO: openconnect oeffnet den Standardbrowser fuer das Login. cmd = base + ["--external-browser", _external_browser(), f"https://{SERVER}"] else: if not self.vpn_pw: pw, ok = QInputDialog.getText( None, APP_NAME, f"VPN-Passwort fuer {VPN_USER}:", QLineEdit.EchoMode.Password, ) if not (ok and pw): return self.vpn_pw = pw send_pw = self.vpn_pw cmd = base + ["--user", VPN_USER, "--passwd-on-stdin", f"https://{SERVER}"] try: self._log = open(LOGFILE, "wb") self.oc = subprocess.Popen( cmd, stdin=subprocess.PIPE if send_pw else None, stdout=self._log, stderr=subprocess.STDOUT, ) if send_pw: self.oc.stdin.write((send_pw + "\n").encode()) self.oc.stdin.flush() except FileNotFoundError: self._set_state("error") QMessageBox.critical(None, APP_NAME, "openconnect nicht gefunden.") return self._set_state("connecting") def disconnect(self): self._kill_browser() if self.oc and self.oc.poll() is None: try: self.oc.send_signal(signal.SIGINT) self.oc.wait(timeout=4) except Exception: try: self.oc.terminate() except Exception: pass self.oc = None self._set_state("disconnected") def _kill_browser(self): if _system() in ("Darwin", "Linux"): subprocess.run( ["pkill", "-f", f"user-data-dir={PROFILE}"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) elif self.browser and self.browser.poll() is None: self.browser.terminate() self.browser = None def launch_browser(self): seed_profile() argv = browser_argv() + [ f"--user-data-dir={PROFILE}", f"--proxy-pac-url=http://127.0.0.1:{PACPORT}/{PAC_NAME}", f"--host-resolver-rules=MAP {INTERNAL_GLOB} ~NOTFOUND", "--no-first-run", "--no-default-browser-check", "--disable-features=msSmartScreenProtection", "--test-type", START_URL, ] self.browser = subprocess.Popen(argv) def _tick(self): if self.state == "connecting": if self.oc and self.oc.poll() is not None: self._set_state("error") return if _port_open(PORT): self._set_state("connected") self.launch_browser() elif self.state == "connected": if self.oc and self.oc.poll() is not None: self.disconnect() def quit(self): self.disconnect() self.pac.shutdown() self.app.quit() def _port_open(port: int) -> bool: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.settimeout(0.3) return s.connect_ex(("127.0.0.1", port)) == 0 def main(): app = QApplication([]) app.setApplicationName("HeronVPN") app.setQuitOnLastWindowClosed(False) _tray = VpnTray(app) # noqa: F841 app.exec() if __name__ == "__main__": main()