#!/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. Auth: GlobalProtect-SAML (Duo). Der SAML-Login laeuft als eigener Cocoa/pyobjc- Prozess (heron_gp_saml.py --auth-only, via CLT-Python) und liefert nur den prelogin-cookie; diese App startet damit openconnect (+HIP-Report) ueber ocproxy. Abhaengigkeiten: openconnect + ocproxy (macOS: brew install openconnect ocproxy), CLT-Python mit pyobjc (pyobjc-framework-WebKit) fuer heron_gp_saml.py. """ from __future__ import annotations import glob as _glob 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 QProcess, QProcessEnvironment, QTimer, Qt from PyQt6.QtGui import ( QAction, QColor, QIcon, QImage, QPainter, QPixmap, qAlpha, qRgba, ) from PyQt6.QtWidgets import ( QApplication, QSystemTrayIcon, QMenu, QMessageBox, ) # ===== CONFIG ================================================================= APP_NAME = "Heron VPN" PROTOCOL = "gp" # GlobalProtect (Palo Alto) SERVER = "ssl.heron.at" PORT = 11090 # SOCKS5 (ocproxy) – andere Ports als Wuerth PACPORT = 11091 # lokaler PAC-HTTP-Server # Hosts, die ueber das VPN laufen (Domain-Globs und/oder IP-Globs). INTERNAL_GLOBS = ["*.heron.at", "172.16.*"] PROFILE = os.path.expanduser("~/.heron-edge") START_URL = "https://robotunits.com" LOGFILE = os.path.expanduser("~/.heron-vpn.log") PAC_NAME = "heron.pac" CLT_PYTHON = "/Library/Developer/CommandLineTools/usr/bin/python3" # hat pyobjc # Lesezeichen, die ins frische Profil gesaet werden (Name, URL). BOOKMARKS: list[tuple[str, str]] = [ ("Storybook", "http://172.16.80.30:30060/"), ("Heron", "https://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 _saml_script() -> str: """Pfad zu heron_gp_saml.py (im Bundle unter Resources, sonst neben uns).""" base = os.environ.get( "RESOURCEPATH", os.path.dirname(os.path.abspath(__file__))) return os.path.join(base, "heron_gp_saml.py") def _hipreport() -> str | None: for p in ["/opt/homebrew/opt/openconnect/libexec/openconnect/hipreport.sh", *_glob.glob("/opt/homebrew/Cellar/openconnect/*/libexec/openconnect/hipreport.sh"), "/usr/local/opt/openconnect/libexec/openconnect/hipreport.sh"]: if os.path.exists(p): return p return None def _resolver_rules() -> str: return " , ".join(f"MAP {g} ~NOTFOUND" for g in INTERNAL_GLOBS) # ---------- PAC-Server (in-process) ------------------------------------------ _PAC_COND = " || ".join(f'shExpMatch(host, "{g}")' for g in INTERNAL_GLOBS) PAC_BODY = ( "function FindProxyForURL(url, host) {\n" f" if ({_PAC_COND}) {{\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 (Reiher-Silhouette in Zustandsfarbe) ------------------------ _ICON_CACHE: dict[str, QIcon] = {} _MASK: QImage | None = None def _heron_mask() -> QImage: global _MASK if _MASK is None: base = os.environ.get( "RESOURCEPATH", os.path.dirname(os.path.abspath(__file__))) img = QImage(os.path.join(base, "heron_mask.png")) _MASK = (img.scaled(44, 44, Qt.AspectRatioMode.KeepAspectRatio, Qt.TransformationMode.SmoothTransformation) .convertToFormat(QImage.Format.Format_ARGB32) if not img.isNull() else QImage()) return _MASK def heron_icon(color: str) -> QIcon: if color in _ICON_CACHE: return _ICON_CACHE[color] m = _heron_mask() if m.isNull(): # Fallback: einfacher Punkt pm = QPixmap(22, 22) pm.fill(Qt.GlobalColor.transparent) p = QPainter(pm) p.setRenderHint(QPainter.RenderHint.Antialiasing) p.setBrush(QColor(color)) p.setPen(Qt.PenStyle.NoPen) p.drawEllipse(4, 4, 14, 14) p.end() return QIcon(pm) tgt = QColor(color) o = QImage(m) for y in range(o.height()): for x in range(o.width()): a = qAlpha(o.pixel(x, y)) if a: o.setPixel(x, y, qRgba(tgt.red(), tgt.green(), tgt.blue(), a)) icon = QIcon(QPixmap.fromImage(o)) _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.auth_proc: QProcess | None = None self.state = "disconnected" self.pac = start_pac_server() self.tray = QSystemTrayIcon() self.tray.setIcon(heron_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(heron_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): if self.auth_proc is not None: return self._set_state("connecting") # SAML-Login als eigener pyobjc/WKWebView-Prozess -> liefert Cookie als JSON. self.auth_proc = QProcess() # Bereinigte Umgebung: sonst erbt das CLT-Python die PYTHONHOME/PATH des # py2app-Bundles und findet sein pyobjc nicht (Subprozess stirbt sofort). env = QProcessEnvironment.systemEnvironment() for v in ("PYTHONHOME", "PYTHONPATH", "PYTHONEXECUTABLE", "PYTHONNOUSERSITE", "PYTHONDONTWRITEBYTECODE"): env.remove(v) self.auth_proc.setProcessEnvironment(env) self.auth_proc.finished.connect(self._on_auth_done) self.auth_proc.start(CLT_PYTHON, [_saml_script(), "--auth-only"]) if not self.auth_proc.waitForStarted(3000): self.auth_proc = None self._set_state("error") QMessageBox.critical(None, APP_NAME, "SAML-Login (CLT-Python/pyobjc) nicht startbar.") def _on_auth_done(self, code, status): out = bytes(self.auth_proc.readAllStandardOutput()).decode("utf-8", "replace") self.auth_proc = None auth = None for line in out.strip().splitlines(): line = line.strip() if line.startswith("{"): try: auth = json.loads(line) except Exception: pass if not auth or not auth.get("prelogin-cookie"): self._set_state("error") return self._start_openconnect(auth.get("saml-username", ""), auth["prelogin-cookie"]) def _start_openconnect(self, user: str, cookie: str): # Reste eines frueheren Laufs wegraeumen, sonst Portkonflikt auf SOCKS. if _system() in ("Darwin", "Linux"): subprocess.run(["pkill", "-f", f"ocproxy -D {PORT}"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) cmd = [ _tool("openconnect"), f"--protocol={PROTOCOL}", "--user", user, "--usergroup=gateway:prelogin-cookie", "--passwd-on-stdin", "--script", f"{_tool('ocproxy')} -D {PORT}", "--script-tun", ] hip = _hipreport() if hip: cmd += ["--csd-wrapper", hip] # HIP-Report, sonst trennt das Gateway cmd.append(SERVER) try: self._log = open(LOGFILE, "wb") self.oc = subprocess.Popen( cmd, stdin=subprocess.PIPE, stdout=self._log, stderr=subprocess.STDOUT, ) self.oc.stdin.write((cookie + "\n").encode()) self.oc.stdin.flush() except FileNotFoundError: self._set_state("error") QMessageBox.critical(None, APP_NAME, "openconnect nicht gefunden.") return # bleibt "connecting"; _tick erkennt den offenen SOCKS-Port def disconnect(self): if self.auth_proc is not None: try: self.auth_proc.kill() except Exception: pass self.auth_proc = None 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={_resolver_rules()}", "--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()