openvpn-2fa-autoconnect/launch_vpn_gui.py

"""launch_vpn.py — VPN с TOTP-аутентификацией и индикатором в трее.
auth = static+TOTP, конкатенация без разделителя."""
import sys, ctypes
import subprocess
import time
import threading
import tempfile
import os
import pyotp
import keyring
import pystray
from PIL import Image, ImageDraw

USERNAME = "boyko_vn"
OVPN_CONFIG = r"C:\Users\User\app\vpn\ovpn-config-ano.ovpn"
OPENVPN_BIN = r"C:\Program Files\OpenVPN\bin\openvpn.exe"
DEV_NODE = "OpenVPN Data Channel Offload"
SERVICE = "openvpn-totp"
LOG_PATH = r"ovpn.log"

proc = None          # текущий процесс openvpn
proc_lock = threading.Lock()
icon = None


def make_icon(connected: bool) -> Image.Image:
    """Кружок: зелёный = подключено, серый = отключено."""
    img = Image.new("RGBA", (64, 64), (0, 0, 0, 0))
    d = ImageDraw.Draw(img)
    color = (40, 200, 80, 255) if connected else (130, 130, 130, 255)
    d.ellipse((12, 12, 52, 52), fill=color)
    return img


def update_icon():
    if icon is None:
        return
    connected = is_running()
    icon.icon = make_icon(connected)
    icon.title = "VPN: подключено" if connected else "VPN: отключено"
    icon.update_menu()          # перерисовать тексты/доступность пунктов


def is_running() -> bool:
    with proc_lock:
        return proc is not None and proc.poll() is None


def build_auth_file() -> str:
    secret = keyring.get_password(SERVICE, "secret")
    static = keyring.get_password(SERVICE, "static")
    code = pyotp.TOTP(secret).now()
    fd, path = tempfile.mkstemp(suffix=".txt")
    with os.fdopen(fd, "w", newline="\n") as f:
        f.write(f"{USERNAME}\n{static}{code}\n")
    return path


def run_vpn():
    """Запуск openvpn в отдельном потоке; чистит auth-файл и обновляет иконку."""
    global proc
    auth_path = build_auth_file()
    try:
        logf = open(rf"ovpn-wrapper.log", "a", encoding="utf-8")
        p = subprocess.Popen(
            [OPENVPN_BIN, "--config", OVPN_CONFIG,
             "--auth-user-pass", auth_path,
             "--dev-node", DEV_NODE,
             "--log", LOG_PATH,
             "--verb", "4"],
            stdout=logf, stderr=logf,
            creationflags=subprocess.CREATE_NO_WINDOW,
        )
        with proc_lock:
            proc = p
        update_icon()
        p.wait()                      # блокируется до завершения процесса
    finally:
        # auth-файл больше не нужен сразу после старта; openvpn уже прочитал его,
        # но удаляем после завершения на всякий случай
        try:
            os.remove(auth_path)
        except OSError:
            pass
        with proc_lock:
            proc = None
        update_icon()


def on_connect(_icon, _item):
    if is_running():
        return
    threading.Thread(target=run_vpn, daemon=True).start()


def on_disconnect(_icon, _item):
    with proc_lock:
        if proc is not None and proc.poll() is None:
            proc.terminate()


def on_open_log(_icon, _item):
    """Открыть файл лога в редакторе по умолчанию; иначе — в блокноте."""
    log_abs = os.path.abspath(LOG_PATH)
    # создаём пустой файл, если лога ещё нет, чтобы было что открывать
    if not os.path.exists(log_abs):
        try:
            open(log_abs, "a", encoding="utf-8").close()
        except OSError:
            pass
    try:
        # глагол "edit" открывает в редакторе по умолчанию для типа файла
        os.startfile(log_abs, "edit")
    except OSError:
        try:
            # запасной вариант — обычное открытие ассоциированной программой
            os.startfile(log_abs)
        except OSError:
            # последний резерв — блокнот Windows
            subprocess.Popen(["notepad.exe", log_abs])


def on_exit(_icon, _item):
    on_disconnect(_icon, _item)
    _icon.stop()


def menu():
    return pystray.Menu(
        pystray.MenuItem(
            lambda item: "Отключено" if not is_running() else "Подключено",
            None, enabled=False),
        pystray.MenuItem("Открыть лог", on_open_log),
        pystray.Menu.SEPARATOR,
        pystray.MenuItem("Подключить", on_connect,
                         enabled=lambda item: not is_running()),
        pystray.MenuItem("Отключить", on_disconnect,
                         enabled=lambda item: is_running()),
        pystray.Menu.SEPARATOR,
        pystray.MenuItem("Выход", on_exit),
    )



def watcher():
    last = None
    while True:
        cur = is_running()
        if cur != last:
            update_icon()
            last = cur
        time.sleep(1)


def ensure_admin_old():
    try:
        if ctypes.windll.shell32.IsUserAnAdmin():
            return
    except Exception:
        pass
    params = " ".join(f'"{a}"' for a in sys.argv)
    ctypes.windll.shell32.ShellExecuteW(None, "runas", sys.executable, params, None, 1)
    sys.exit(0)


def ensure_admin():
    try:
        if ctypes.windll.shell32.IsUserAnAdmin():
            return
    except Exception:
        pass
    # запускать elevated-копию через pythonw.exe (без консоли)
    exe = sys.executable
    pyw = exe.replace("python.exe", "pythonw.exe")
    if not os.path.exists(pyw):
        pyw = exe  # fallback, если pythonw не нашёлся
    params = " ".join(f'"{a}"' for a in sys.argv)
    ctypes.windll.shell32.ShellExecuteW(None, "runas", pyw, params, None, 1)
    sys.exit(0)


def main():
    ensure_admin()
    global icon
    icon = pystray.Icon("vpn-totp", make_icon(False), "VPN: отключено", menu())
    threading.Thread(target=watcher, daemon=True).start()
    icon.run()


if __name__ == "__main__":
    main()