From f9361d0a7dc252ad6cc13cf5ea4abecc437b9e7a Mon Sep 17 00:00:00 2001 From: Aimili Date: Thu, 27 Aug 2026 01:44:09 +0800 Subject: [PATCH] fix: harden cross-browser dashboard interactions --- proxy_server.py | 3 + tests/test_manager_logic.py | 60 ++++++ vpngate_manager.py | 413 +++++++++++++++++++++++++----------- 3 files changed, 354 insertions(+), 122 deletions(-) diff --git a/proxy_server.py b/proxy_server.py index 00e2a14..7805494 100644 --- a/proxy_server.py +++ b/proxy_server.py @@ -254,6 +254,9 @@ def socks5_client(client: socket.socket, first_byte: bytes) -> None: return client.sendall(b"\x01\x00") else: + if 0 not in methods: + client.sendall(b"\x05\xff") + return client.sendall(b"\x05\x00") version, command, _, address_type = recv_exact(client, 4) if version != 5 or command != 1: diff --git a/tests/test_manager_logic.py b/tests/test_manager_logic.py index 8f0a30e..9279ea2 100644 --- a/tests/test_manager_logic.py +++ b/tests/test_manager_logic.py @@ -98,6 +98,7 @@ class ManagerLogicTests(unittest.TestCase): manager.last_proxy_failure_node_id = "" manager.background_refill_thread = None manager.background_refill_cancel_event.clear() + manager.active_sessions.clear() def tearDown(self) -> None: if manager.connection_attempt_lock.locked(): @@ -407,6 +408,40 @@ class ManagerLogicTests(unittest.TestCase): self.assertIn('typeof document.hidden !== "boolean" || !document.hidden', manager.INDEX_HTML) self.assertEqual(500, manager.WEB_LOG_MAX_ENTRIES) + def test_web_dashboard_has_cross_browser_interaction_safeguards(self) -> None: + self.assertNotIn("fonts.googleapis.com", manager.LOGIN_HTML) + self.assertNotIn("fonts.googleapis.com", manager.INDEX_HTML) + self.assertIn('const pwd = document.getElementById("password").value;', manager.LOGIN_HTML) + self.assertIn('const password = $("cred_password").value;', manager.INDEX_HTML) + self.assertIn("function fetchWithTimeout", manager.LOGIN_HTML) + self.assertIn("function fetchWithTimeout", manager.INDEX_HTML) + self.assertNotIn("await fetch(", manager.INDEX_HTML) + self.assertIn('role="dialog" aria-modal="true"', manager.INDEX_HTML) + self.assertIn('aria-label="关闭网页安全设置"', manager.INDEX_HTML) + self.assertIn('class="option-card active" data-value="auto" aria-pressed="true"', manager.INDEX_HTML) + self.assertIn('class="vps-recommend-tab"', manager.INDEX_HTML) + self.assertIn('position: static;', manager.INDEX_HTML) + self.assertIn('-webkit-overflow-scrolling: touch;', manager.INDEX_HTML) + self.assertIn('formatUrlHost(window.location.hostname)', manager.INDEX_HTML) + self.assertNotIn('id="status" class="status" style="display: none;"', manager.INDEX_HTML) + self.assertIn('${esc(localProxy)}', manager.INDEX_HTML) + self.assertIn('${esc(statusMessage)}', manager.INDEX_HTML) + + def test_random_password_uses_cryptographic_randomness(self) -> None: + with mock.patch.object(manager.secrets, "choice", side_effect=list("aA0aA0aA0aA0")) as choice: + password = manager.generate_random_password() + + self.assertEqual("aA0aA0aA0aA0", password) + self.assertEqual(12, choice.call_count) + + def test_expired_sessions_are_removed(self) -> None: + manager.active_sessions.update({"expired": 99.0, "active": 101.0}) + + removed = manager.purge_expired_sessions(now=100.0) + + self.assertEqual(1, removed) + self.assertEqual({"active": 101.0}, manager.active_sessions) + def test_web_log_reader_only_returns_recent_valid_entries(self) -> None: log_file = manager.DATA_DIR / "logs" / "current.json" log_file.parent.mkdir(parents=True) @@ -597,6 +632,31 @@ class ManagerLogicTests(unittest.TestCase): class ProxyServerConcurrencyTests(unittest.TestCase): + def test_socks5_rejects_client_without_no_auth_method(self) -> None: + class Client: + def __init__(self): + self.incoming = bytearray(b"\x01\x02") + self.sent = bytearray() + self.closed = False + + def recv(self, size): + chunk = self.incoming[:size] + del self.incoming[:size] + return bytes(chunk) + + def sendall(self, data): + self.sent.extend(data) + + def close(self): + self.closed = True + + client = Client() + with mock.patch.object(proxy_server, "proxy_auth_enabled", return_value=False): + proxy_server.socks5_client(client, b"\x05") + + self.assertEqual(b"\x05\xff", bytes(client.sent)) + self.assertTrue(client.closed) + def test_each_proxy_worker_keeps_its_accepted_socket(self) -> None: class Client: def __init__(self, name): diff --git a/vpngate_manager.py b/vpngate_manager.py index c09e8da..9e8968f 100644 --- a/vpngate_manager.py +++ b/vpngate_manager.py @@ -7,6 +7,7 @@ import json import os import queue import re +import secrets import select import shlex import signal @@ -179,6 +180,17 @@ server_start_time = time.time() class ConnectionCancelled(RuntimeError): pass +def purge_expired_sessions(now: float | None = None) -> int: + current_time = time.time() if now is None else now + with lock: + expired_tokens = [ + token for token, expires_at in active_sessions.items() + if expires_at <= current_time + ] + for token in expired_tokens: + active_sessions.pop(token, None) + return len(expired_tokens) + def ensure_dirs() -> None: DATA_DIR.mkdir(exist_ok=True, parents=True) CONFIG_DIR.mkdir(exist_ok=True, parents=True) @@ -219,13 +231,12 @@ def read_json(path: Path, default: Any) -> Any: return default import hashlib -import random def generate_random_password() -> str: import string chars = string.ascii_letters + string.digits while True: - pwd = "".join(random.choices(chars, k=12)) + pwd = "".join(secrets.choice(chars) for _ in range(12)) # Ensure it contains at least one lowercase, one uppercase, and one digit has_lower = any(c.islower() for c in pwd) has_upper = any(c.isupper() for c in pwd) @@ -237,7 +248,7 @@ def generate_random_username() -> str: import string chars = string.ascii_letters + string.digits while True: - uname = "".join(random.choices(chars, k=12)) + uname = "".join(secrets.choice(chars) for _ in range(12)) # Ensure it starts with a letter and contains at least one lowercase, one uppercase, and one digit if uname[0].isalpha(): has_lower = any(c.islower() for c in uname) @@ -2499,7 +2510,6 @@ LOGIN_HTML = r""" AimiliVPN - 安全登录 -