"""launch_vpn.py — VPN с TOTP-аутентификацией и индикатором в трее. auth = static+TOTP, конкатенация без разделителя.""" import sys, ctypes import subprocess import time import threading import tempfile import json import os 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", # путь к лог-файлу 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 build_auth_file() -> str: service = config.get("keyring_service", "openvpn-totp") 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"{config['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( [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()