347 lines
14 KiB
Python
347 lines
14 KiB
Python
"""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()
|