From 85017ac628319fb8e051a8c815e02686b91d92a1 Mon Sep 17 00:00:00 2001 From: bitwiresys Date: Fri, 3 Jul 2026 15:01:02 +0300 Subject: [PATCH] feat(wireguard): support WireGuard as an Xray inbound MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Recognize a `wireguard` protocol inbound inside an Xray core so its tag is exposed (via /api/inbounds) and users can be assigned to it, and attribute its per-user traffic/online state. Xray now ships a first-class WireGuard inbound with a UserManager (XTLS/Xray-core#6360), so WireGuard can be served through the same Xray core as the other protocols — in addition to the existing standalone WG backend (CoreType.wg). This is purely additive. - app/core/xray.py: `_read_wireguard_inbound` registers a WG-in-Xray inbound (interface/listen_port/address/mtu/public key) alongside the vmess/vless/trojan/shadowsocks/hysteria readers. - app/jobs/record_usages.py: the Xray WG inbound reports per-user stats keyed by the peer's hex public key; map that back to the panel user id so WG traffic and online state are recorded like any other protocol. --- app/core/xray.py | 25 ++++++++++++++- app/jobs/record_usages.py | 63 ++++++++++++++++++++++++++++++++++--- tests/test_record_usages.py | 4 +-- 3 files changed, 84 insertions(+), 8 deletions(-) diff --git a/app/core/xray.py b/app/core/xray.py index ca1986e8a..74c0c2dfa 100644 --- a/app/core/xray.py +++ b/app/core/xray.py @@ -10,7 +10,7 @@ from app.models.core import CoreType from app.models.protocol import ProxyProtocol -from app.utils.crypto import get_cert_SANs, get_x25519_public_key +from app.utils.crypto import get_cert_SANs, get_wireguard_public_key, get_x25519_public_key def _protocols_from_inbounds_by_tag(inbounds_by_tag: dict[str, dict]) -> frozenset[ProxyProtocol]: @@ -412,8 +412,31 @@ def _resolve_inbounds(self): self._read_inbound(inbound) self._protocols = _protocols_from_inbounds_by_tag(self._inbounds_by_tag) + def _read_wireguard_inbound(self, inbound: dict): + """Register a WireGuard-in-Xray inbound so its tag appears in /api/inbounds.""" + wg = inbound.get("settings", {}) + secret_key = wg.get("secretKey", "") + try: + public_key = get_wireguard_public_key(secret_key) if secret_key else "" + except Exception: + public_key = "" + settings = self._create_base_settings(inbound) + settings["listen_port"] = inbound.get("port") + settings["public_key"] = public_key + settings["address"] = wg.get("address", []) + settings["mtu"] = wg.get("mtu", 1420) + settings["network"] = "udp" + tag = inbound["tag"] + if tag not in self._inbounds: + self._inbounds.append(tag) + self._inbounds_by_tag[tag] = settings + def _read_inbound(self, inbound: dict): """Read an inbound and its settings.""" + if inbound["protocol"] == "wireguard": + self._read_wireguard_inbound(inbound) + return + if inbound["protocol"] not in ("vmess", "vless", "trojan", "shadowsocks", "hysteria"): return diff --git a/app/jobs/record_usages.py b/app/jobs/record_usages.py index 76b8cd284..b902e5c18 100644 --- a/app/jobs/record_usages.py +++ b/app/jobs/record_usages.py @@ -1,4 +1,5 @@ import asyncio +import base64 import multiprocessing import random import time @@ -41,6 +42,45 @@ _thread_pool = None _thread_pool_lock = asyncio.Lock() +# WireGuard user stats arrive keyed by the peer's hex public key (xray's +# `user>>>{hex_pubkey}>>>traffic`), not by the panel's numeric `{id}` user id. +# Cache a {hex_public_key -> user_id} map so those stats can be attributed to +# the right user (otherwise WG traffic and online state are silently dropped). +_wg_pubkey_map_cache: dict = {"data": {}, "ts": 0.0} +_WG_PUBKEY_MAP_TTL = 30.0 # seconds + + +async def _get_wg_pubkey_to_uid() -> dict: + """Return {hex_public_key: user_id} for all users that have a WireGuard key.""" + now = time.time() + if _wg_pubkey_map_cache["data"] and (now - _wg_pubkey_map_cache["ts"]) < _WG_PUBKEY_MAP_TTL: + return _wg_pubkey_map_cache["data"] + + mapping: dict[str, int] = {} + try: + async with GetDB() as db: + result = await db.execute(select(User.id, User.proxy_settings)) + for uid, proxy_settings in result.all(): + if not isinstance(proxy_settings, dict): + continue + wg = proxy_settings.get("wireguard") + if not wg: + continue + pub = wg.get("public_key") + if not pub: + continue + try: + mapping[base64.b64decode(pub).hex()] = uid + except Exception: + continue + except Exception as e: + logger.warning("Failed to build WireGuard pubkey->uid map: %s", e) + return _wg_pubkey_map_cache["data"] + + _wg_pubkey_map_cache["data"] = mapping + _wg_pubkey_map_cache["ts"] = now + return mapping + async def _get_thread_pool(): """Get or create the thread pool executor (thread-safe).""" @@ -479,15 +519,23 @@ async def _record_single_node(node_id: int, params: list[dict]): await asyncio.gather(*tasks, return_exceptions=True) -def _process_users_stats_response(stats_response): +def _process_users_stats_response(stats_response, wg_pubkey_map=None): """ Process stats response (CPU-bound operation) - runs in thread pool. Pure function designed for thread-safe execution. Returns tuple: (validated_params, invalid_uids) for logging outside thread. + + wg_pubkey_map maps a WireGuard peer's hex public key to its user id, so WG + stats (keyed by hex pubkey instead of `{id}.{username}`) attribute correctly. """ + wg_pubkey_map = wg_pubkey_map or {} params = defaultdict(int) for stat in filter(attrgetter("value"), stats_response.stats): - params[stat.name] += stat.value + # WireGuard inbound reports stats keyed by the peer's hex public key; + # translate it to the panel user id so traffic/online are recorded. + # (Upstream identifier is plain `{id}`, so no split is needed.) + key = str(wg_pubkey_map.get(stat.name, stat.name)) + params[key] += stat.value validated_params = [] invalid_uids = [] @@ -500,7 +548,7 @@ def _process_users_stats_response(stats_response): return validated_params, invalid_uids -async def get_users_stats(node: PasarGuardNode): +async def get_users_stats(node: PasarGuardNode, wg_pubkey_map=None): """ Get user stats from node using thread pool for CPU-bound processing. This distributes the heavy data processing workload across cores. @@ -514,7 +562,7 @@ async def get_users_stats(node: PasarGuardNode): loop = asyncio.get_running_loop() thread_pool = await _get_thread_pool() validated_params, invalid_uids = await loop.run_in_executor( - thread_pool, _process_users_stats_response, stats_response + thread_pool, _process_users_stats_response, stats_response, wg_pubkey_map ) if invalid_uids: @@ -683,8 +731,13 @@ async def _record_user_usages_impl(): else: usage_coefficient[node_id] = data.get("usage_coefficient", 1) if data else 1.0 + # Build WireGuard hex-pubkey -> user id map so WG stats attribute correctly + wg_pubkey_map = await _get_wg_pubkey_to_uid() + # Gather stats directly - asyncio.gather accepts coroutines, no need for create_task - stats_results = await asyncio.gather(*[get_users_stats(node) for _, node in nodes], return_exceptions=True) + stats_results = await asyncio.gather( + *[get_users_stats(node, wg_pubkey_map) for _, node in nodes], return_exceptions=True + ) api_params = {} for i, result in enumerate(stats_results): node_id = nodes[i][0] diff --git a/tests/test_record_usages.py b/tests/test_record_usages.py index 0ed98880e..3b18e0674 100644 --- a/tests/test_record_usages.py +++ b/tests/test_record_usages.py @@ -150,7 +150,7 @@ async def test_record_user_usages_updates_users_and_admins(monkeypatch: pytest.M node_two_id: [{"uid": str(user_one_id), "value": 75}], } - async def fake_get_users_stats(node: DummyNode): + async def fake_get_users_stats(node: DummyNode, wg_pubkey_map=None): return stats_map[node.node_id] monkeypatch.setattr(record_usages, "get_users_stats", fake_get_users_stats) @@ -280,7 +280,7 @@ async def test_record_user_usages_returns_when_no_usage(monkeypatch: pytest.Monk nodes = [(node_id, DummyNode(node_id))] monkeypatch.setattr(record_usages.node_manager, "get_healthy_nodes", AsyncMock(return_value=nodes)) - async def fake_get_users_stats(_: DummyNode): + async def fake_get_users_stats(_: DummyNode, wg_pubkey_map=None): return [] monkeypatch.setattr(record_usages, "get_users_stats", fake_get_users_stats)