app.py

import json
import os
import subprocess
from pathlib import Path

import rumps


CONFIG_PATH = Path(__file__).with_name("config.json")
ANYCONNECT_BIN = "/opt/cisco/anyconnect/bin/vpn"


def read_config():
    if not CONFIG_PATH.exists():
        # Create a starter config to guide the user
        example = {
            "server": "vpn.com",
            "username": "user",
            "password": "password"
        }
        CONFIG_PATH.write_text(json.dumps(example, indent=2), encoding="utf-8")
        return example
    try:
        return json.loads(CONFIG_PATH.read_text(encoding="utf-8"))
    except Exception:
        return {"server": "vpn.com", "username": "user", "password": "password"}


def run_shell(command: str, timeout: int | None = 60) -> tuple[int, str, str]:
    process = subprocess.Popen(
        ["/bin/bash", "-lc", command],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
    )
    try:
        stdout, stderr = process.communicate(timeout=timeout)
    except subprocess.TimeoutExpired:
        process.kill()
        stdout, stderr = process.communicate()
    return process.returncode or 0, stdout.strip(), stderr.strip()


def parse_status_output(output: str) -> str:
    text = output.lower()
    if "state: connected" in text or "notice: connected" in text:
        return "connected"
    if "state: disconnected" in text:
        return "disconnected"
    return "unknown"


def get_vpn_status() -> str:
    if not Path(ANYCONNECT_BIN).exists():
        return "unknown"
    code, out, err = run_shell(f"{ANYCONNECT_BIN} status")
    status = parse_status_output(out or err)
    return status


def connect_vpn(username: str, password: str, server: str) -> tuple[bool, str]:
    # Follow user's provided shell pipeline for AnyConnect scripted connect
    payload = f"{username}\\n{password}\\ny"
    cmd = f"printf '{payload}' | {ANYCONNECT_BIN} -s connect {server}"
    code, out, err = run_shell(cmd, timeout=120)
    output = "\n".join([o for o in [out, err] if o])
    success = "notice: Connected" in output or parse_status_output(output) == "connected"
    return success, output


def disconnect_vpn() -> tuple[bool, str]:
    cmd = f"{ANYCONNECT_BIN} disconnect"
    code, out, err = run_shell(cmd, timeout=60)
    output = "\n".join([o for o in [out, err] if o])
    success = "state: Disconnected" in output or parse_status_output(output) == "disconnected"
    return success, output


class VPNTrayApp(rumps.App):
    def __init__(self):
        super().__init__("VPN ✧", quit_button=None)
        self.config = read_config()

        self.menu = [
            rumps.MenuItem("Connect VPN", callback=self.on_connect),
            rumps.MenuItem("Disconnect VPN", callback=self.on_disconnect),
            None,
            rumps.MenuItem("Check status", callback=self.on_check_status),
            None,
            rumps.MenuItem("Open config.json", callback=self.on_open_config),
            rumps.MenuItem("Quit", callback=rumps.quit_application),
        ]

        # Busy flag to avoid concurrent actions
        self._working = False

        # Poll status on main thread via timer
        self._timer = rumps.Timer(self._on_timer, 10)
        self._timer.start()

        # Initial status
        self._update_title(get_vpn_status())

    def on_open_config(self, _):
        # Open config in default editor
        subprocess.Popen(["open", str(CONFIG_PATH)])

    def on_check_status(self, _):
        status = get_vpn_status()
        self._update_title(status)
        message = {
            "connected": "Connected",
            "disconnected": "Disconnected",
        }.get(status, "Unknown")
        rumps.notification("VPN status", "", message)

    def on_connect(self, _):
        if self._working:
            return
        self._set_working(True, "Connecting…")
        ok, output = connect_vpn(
            self.config.get("username", "user"),
            self.config.get("password", "password"),
            self.config.get("server", "vpn.com"),
        )
        status = "connected" if ok else get_vpn_status()
        self._update_title(status)
        self._set_working(False)
        if ok:
            rumps.notification("VPN", "", "Connected successfully")
        else:
            rumps.alert(title="Connection error", message=output or "Unknown error")

    def on_disconnect(self, _):
        if self._working:
            return
        self._set_working(True, "Disconnecting…")
        ok, output = disconnect_vpn()
        status = "disconnected" if ok else get_vpn_status()
        self._update_title(status)
        self._set_working(False)
        if ok:
            rumps.notification("VPN", "", "Disconnected")
        else:
            rumps.alert(title="Disconnect error", message=output or "Unknown error")

    def _update_title(self, status: str):
        # Use emoji as lightweight status indicators in the menu bar title
        if status == "connected":
            self.title = "🟢"
        elif status == "disconnected":
            self.title = "🔘"
        else:
            self.title = "⚪︎"

    def _set_working(self, working: bool, message: str | None = None):
        self._working = working
        if working:
            self.title = f"⏳ {message or ''}".strip()
        else:
            self._update_title(get_vpn_status())
    def _on_timer(self, _):
        status = get_vpn_status()
        self._update_title(status)

if __name__ == "__main__":
    app = VPNTrayApp()
    app.run()