"""launch_vpn.py — VPN с TOTP-аутентификацией и индикатором в трее. auth = static+TOTP, конкатенация без разделителя.""" import sys, ctypes import subprocess import time import threading import tempfile import json import os import base64 import pyotp import keyring import pystray from PIL import Image, ImageDraw # --- Настройки ----------------------------------------------------------------- # Все параметры лежат в JSON-файле рядом со скриптом. Правится вручную (пункт # меню "Открыть настройки"); перечитывается автоматически при изменении. # Секреты (TOTP secret и статический пароль) здесь НЕ хранятся — они лежат в # keyring (Windows Credential Manager) под именем сервиса keyring_service; # кладёт их init_vpn.py. CONFIG_PATH = os.path.join( os.path.dirname(os.path.abspath(__file__)), "vpn_config.json") DEFAULT_CONFIG = { # --- параметры подключения / авторизации --- # имя пользователя VPN. "username": "boyko_vn", # путь к .ovpn-профилю. "ovpn_config": r"C:\Users\User\app\vpn\ovpn-config-ano.ovpn", # путь к openvpn.exe. "openvpn_bin": r"C:\Program Files\OpenVPN\bin\openvpn.exe", # имя dev-node адаптера OpenVPN. "dev_node": "OpenVPN Data Channel Offload", # имя сервиса в keyring, откуда берутся secret (TOTP) и static (пароль). "keyring_service": "openvpn-totp", # формат передачи OTP серверу. # False -> пароль = static+code одной строкой (сервер сам отрезает OTP). # True -> формат SCRV1 (static и OTP кодируются раздельно в base64); нужен # для серверов со static-challenge. Саму строку static-challenge при # этом надо УБРАТЬ из .ovpn, иначе openvpn спросит код интерактивно. "static_challenge": False, # путь к лог-файлу openvpn. "log_path": "ovpn.log", # --- проверка связи --- # включить проверку доступности сервера пингами. # False -> работает как раньше: серый (выкл) / зелёный (вкл), без жёлтого. "ping_check_enabled": True, # IP или хост, доступность которого проверяем (обычно шлюз VPN-сети). "ping_host": "10.8.0.1", # интервал между проверками, сек. "ping_interval": 5, # таймаут одного пинга, мс. "ping_timeout_ms": 1000, } config = dict(DEFAULT_CONFIG) _config_mtime = None # Три состояния индикатора в трее. STATE_OFF = "off" # серый — процесс openvpn не запущен STATE_WAIT = "wait" # жёлтый — процесс запущен, но пинги до сервера не доходят STATE_ON = "on" # зелёный — процесс запущен, пинги есть STATE_COLORS = { STATE_OFF: (130, 130, 130, 255), STATE_WAIT: (230, 200, 40, 255), STATE_ON: (40, 200, 80, 255), } STATE_TITLES = { STATE_OFF: "VPN: отключено", STATE_WAIT: "VPN: запущен, нет связи с сервером", STATE_ON: "VPN: подключено", } proc = None # текущий процесс openvpn proc_lock = threading.Lock() icon = None ping_alive = False # результат последней проверки пингов def load_config(): """Загрузить настройки из JSON, дополнив дефолтами. Молча создаёт файл, если его нет, и падает обратно на дефолты при ошибке чтения.""" global _config_mtime try: mtime = os.path.getmtime(CONFIG_PATH) except OSError: # файла нет — создаём с дефолтами try: with open(CONFIG_PATH, "w", encoding="utf-8") as f: json.dump(DEFAULT_CONFIG, f, ensure_ascii=False, indent=2) _config_mtime = os.path.getmtime(CONFIG_PATH) except OSError: _config_mtime = None config.clear() config.update(DEFAULT_CONFIG) return if mtime == _config_mtime: return # файл не менялся — ничего не делаем try: with open(CONFIG_PATH, "r", encoding="utf-8") as f: data = json.load(f) except (OSError, ValueError): data = {} merged = dict(DEFAULT_CONFIG) if isinstance(data, dict): merged.update(data) config.clear() config.update(merged) _config_mtime = mtime def ping_ok(host: str) -> bool: """Один ICMP-пинг до host. Возвращает True только при реальном ответе (наличие 'TTL=' в выводе), чтобы не считать 'Destination unreachable' успехом — Windows-ping в этом случае тоже отдаёт код возврата 0.""" if not host: return False try: r = subprocess.run( ["ping", "-n", "1", "-w", str(config["ping_timeout_ms"]), host], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, creationflags=subprocess.CREATE_NO_WINDOW, ) return b"TTL=" in r.stdout.upper() except Exception: return False def current_state() -> str: """Текущее состояние индикатора с учётом настроек.""" if not is_running(): return STATE_OFF if not config.get("ping_check_enabled", True): return STATE_ON # проверки выключены — старое поведение return STATE_ON if ping_alive else STATE_WAIT def make_icon(state: str) -> Image.Image: """Кружок: серый = выкл, жёлтый = нет связи, зелёный = связь есть.""" img = Image.new("RGBA", (64, 64), (0, 0, 0, 0)) d = ImageDraw.Draw(img) color = STATE_COLORS.get(state, STATE_COLORS[STATE_OFF]) d.ellipse((12, 12, 52, 52), fill=color) return img def update_icon(): if icon is None: return state = current_state() icon.icon = make_icon(state) icon.title = STATE_TITLES[state] icon.update_menu() # перерисовать тексты/доступность пунктов def is_running() -> bool: with proc_lock: return proc is not None and proc.poll() is None def auth_password(static: str, code: str) -> str: """Собрать поле пароля для auth-user-pass с учётом режима static_challenge.""" if config.get("static_challenge"): s = base64.b64encode(static.encode()).decode() r = base64.b64encode(code.encode()).decode() return f"SCRV1:{s}:{r}" return f"{static}{code}" def build_auth_file() -> str: service = config.get("keyring_service", "openvpn-totp") secret = keyring.get_password(service, "secret") # static — необязателен: если его нет в keyring, пароль = только TOTP-код. static = keyring.get_password(service, "static") or "" code = pyotp.TOTP(secret).now() fd, path = tempfile.mkstemp(suffix=".txt") with os.fdopen(fd, "w", newline="\n") as f: f.write(f"{config['username']}\n{auth_password(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( [config["openvpn_bin"], "--config", config["ovpn_config"], "--auth-user-pass", auth_path, "--dev-node", config["dev_node"], "--log", config["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(config.get("log_path", "ovpn.log")) # создаём пустой файл, если лога ещё нет, чтобы было что открывать 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_open_settings(_icon, _item): """Открыть файл настроек в редакторе; создаёт его при отсутствии.""" load_config() # гарантирует, что файл существует try: os.startfile(CONFIG_PATH, "edit") except OSError: try: os.startfile(CONFIG_PATH) except OSError: subprocess.Popen(["notepad.exe", CONFIG_PATH]) def _status_text(_item): return STATE_TITLES[current_state()] def on_exit(_icon, _item): on_disconnect(_icon, _item) _icon.stop() def menu(): return pystray.Menu( pystray.MenuItem(_status_text, None, enabled=False), pystray.MenuItem("Открыть лог", on_open_log), pystray.MenuItem("Открыть настройки", on_open_settings), 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(): """Следит за состоянием процесса и связью с сервером, обновляя иконку только при смене итогового состояния (серый/жёлтый/зелёный).""" global ping_alive last = None while True: load_config() # подхватываем правки настроек на лету if is_running() and config.get("ping_check_enabled", True): ping_alive = ping_ok(config.get("ping_host", "")) else: ping_alive = False state = current_state() if state != last: update_icon() last = state # пока идёт активная проверка пингов — ждём интервал, иначе опрашиваем # процесс раз в секунду для отзывчивого серо/зелёного переключения if is_running() and config.get("ping_check_enabled", True): time.sleep(max(1, int(config.get("ping_interval", 5)))) else: 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() load_config() global icon icon = pystray.Icon( "vpn-totp", make_icon(STATE_OFF), STATE_TITLES[STATE_OFF], menu()) threading.Thread(target=watcher, daemon=True).start() icon.run() if __name__ == "__main__": main()