From dd79fa8f82380eae9dcce4c102ee7afa03b97bf5 Mon Sep 17 00:00:00 2001 From: bitwiresys Date: Fri, 3 Jul 2026 15:01:02 +0300 Subject: [PATCH 1/5] 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 907c0afb0..870186ffa 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 290241c5f..cda66492c 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 @@ -42,6 +43,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).""" @@ -480,15 +520,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 = [] @@ -501,7 +549,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. @@ -515,7 +563,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: @@ -684,8 +732,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 b2151bc43..646148a71 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) @@ -324,7 +324,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) From 46f491a5fc816412a974e435591a9f66e510cd58 Mon Sep 17 00:00:00 2001 From: bitwiresys Date: Sun, 12 Jul 2026 21:57:04 +0300 Subject: [PATCH 2/5] fix(usage): drop the WireGuard hex-pubkey -> uid mapping workaround MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The merged xray-core PR (xtls/Xray-core#6360) matches WireGuard peers on Email like every other inbound, and the node already sends str(user.id) as the Email for WireGuard users (app/node/user.py, serialize_user) same as vmess/vless/trojan/shadowsocks/hysteria. So UsersStat has always arrived keyed by the plain user id for WireGuard too — the pubkey->uid cache in _get_wg_pubkey_to_uid() was solving a problem that didn't exist, added under the assumption that WireGuard's xray-side Email held the peer's public key. Also fixes an unrelated Python 2-style `except ValueError, TypeError:` in _process_users_stats_response that would have crashed the job at runtime, and drops the now-unused base64 import. Verified end-to-end: real WireGuard client -> xray wireguard inbound -> record_user_usages correctly attributes traffic to the user's id with no mapping step. --- app/jobs/record_usages.py | 66 +++++-------------------------------- tests/test_record_usages.py | 4 +-- 2 files changed, 11 insertions(+), 59 deletions(-) diff --git a/app/jobs/record_usages.py b/app/jobs/record_usages.py index cda66492c..fa598a018 100644 --- a/app/jobs/record_usages.py +++ b/app/jobs/record_usages.py @@ -1,5 +1,4 @@ import asyncio -import base64 import multiprocessing import random import time @@ -43,46 +42,6 @@ _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).""" global _thread_pool @@ -520,36 +479,32 @@ 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, wg_pubkey_map=None): +def _process_users_stats_response(stats_response): """ 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. + Every inbound (including WireGuard) reports stats keyed by the user's + email, which the panel sets to the plain `{id}`, so no translation is + needed here. """ - wg_pubkey_map = wg_pubkey_map or {} params = defaultdict(int) for stat in filter(attrgetter("value"), stats_response.stats): - # 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 + params[stat.name] += stat.value validated_params = [] invalid_uids = [] for uid, value in params.items(): try: validated_params.append({"uid": int(uid), "value": value}) - except ValueError, TypeError: + except (ValueError, TypeError): invalid_uids.append(uid) return validated_params, invalid_uids -async def get_users_stats(node: PasarGuardNode, wg_pubkey_map=None): +async def get_users_stats(node: PasarGuardNode): """ Get user stats from node using thread pool for CPU-bound processing. This distributes the heavy data processing workload across cores. @@ -563,7 +518,7 @@ async def get_users_stats(node: PasarGuardNode, wg_pubkey_map=None): 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, wg_pubkey_map + thread_pool, _process_users_stats_response, stats_response ) if invalid_uids: @@ -732,12 +687,9 @@ 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, wg_pubkey_map) for _, node in nodes], return_exceptions=True + *[get_users_stats(node) for _, node in nodes], return_exceptions=True ) api_params = {} for i, result in enumerate(stats_results): diff --git a/tests/test_record_usages.py b/tests/test_record_usages.py index 646148a71..b2151bc43 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, wg_pubkey_map=None): + async def fake_get_users_stats(node: DummyNode): return stats_map[node.node_id] monkeypatch.setattr(record_usages, "get_users_stats", fake_get_users_stats) @@ -324,7 +324,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, wg_pubkey_map=None): + async def fake_get_users_stats(_: DummyNode): return [] monkeypatch.setattr(record_usages, "get_users_stats", fake_get_users_stats) From 1b96c6aa3d95dde1ff1b476be858dbd2719ec2ef Mon Sep 17 00:00:00 2001 From: bitwiresys Date: Wed, 22 Jul 2026 21:52:29 +0300 Subject: [PATCH 3/5] fix wireguard inbound edit form and tag generation --- .../components/xray/xray-inbounds-section.tsx | 15 ++++++++------- .../features/core-editor/kit/sanitize-inbound.ts | 2 ++ 2 files changed, 10 insertions(+), 7 deletions(-) diff --git a/dashboard/src/features/core-editor/components/xray/xray-inbounds-section.tsx b/dashboard/src/features/core-editor/components/xray/xray-inbounds-section.tsx index 5e9c1a196..1fbb16b57 100644 --- a/dashboard/src/features/core-editor/components/xray/xray-inbounds-section.tsx +++ b/dashboard/src/features/core-editor/components/xray/xray-inbounds-section.tsx @@ -1192,9 +1192,10 @@ export function XrayInboundsSection({ headerAddPulse, headerAddEpoch }: XrayInbo let tag = watchedProtocol let network = watchedTransport if (watchedProtocol === 'shadowsocks') network = watchedShadowsocksNetwork + if (isWireguardInboundProtocol(watchedProtocol)) network = 'udp' if (network) tag += `${tagSeparator}${network}` - if (watchedSecurity && watchedSecurity !== 'none') tag += `${tagSeparator}${watchedSecurity}` + if (!isWireguardInboundProtocol(watchedProtocol) && watchedSecurity && watchedSecurity !== 'none') tag += `${tagSeparator}${watchedSecurity}` if (watchedPort) tag += `${tagSeparator}${watchedPort}` tag = tag.toUpperCase() @@ -1622,7 +1623,10 @@ export function XrayInboundsSection({ headerAddPulse, headerAddEpoch }: XrayInbo setDuplicateTagError(tagValue) return } - if (inbound.protocol !== 'unmanaged' && 'transport' in inbound && 'security' in inbound) { + if (isWireguardInboundProtocol(inbound.protocol)) { + const ok = await form.trigger(['protocol', 'tag', 'port', 'wgSecretKey'], { shouldFocus: true }) + if (!ok) return + } else if (inbound.protocol !== 'unmanaged' && 'transport' in inbound && 'security' in inbound) { const fields = ['protocol', 'tag', 'port', ...(form.getValues('security') === 'reality' ? realityInboundZodTriggerFieldNames() : [])] const ok = await form.trigger(fields, { shouldFocus: true }) if (!ok) return @@ -1632,9 +1636,6 @@ export function XrayInboundsSection({ headerAddPulse, headerAddEpoch }: XrayInbo } else if (inbound.protocol === 'tun') { const ok = await form.trigger(['protocol', 'tag', 'tunName', 'tunMtu'], { shouldFocus: true }) if (!ok) return - } else if (inbound.protocol === 'wireguard') { - const ok = await form.trigger(['protocol', 'tag', 'port', 'wgSecretKey'], { shouldFocus: true }) - if (!ok) return } if (!validateWireguardInboundForCommit(inbound)) return let nextInbound: Inbound = inbound @@ -2498,7 +2499,7 @@ export function XrayInboundsSection({ headerAddPulse, headerAddEpoch }: XrayInbo

{t('coreEditor.inbound.unmanagedHint', { defaultValue: 'This inbound is managed as raw JSON in Advanced.' })}

)} - {inbound && inbound.protocol !== 'unmanaged' && 'transport' in inbound && 'security' in inbound && ( + {inbound && inbound.protocol !== 'unmanaged' && !isWireguardInboundProtocol(inbound.protocol) && 'transport' in inbound && 'security' in inbound && (
e.preventDefault()}>
@@ -4509,7 +4510,7 @@ export function XrayInboundsSection({ headerAddPulse, headerAddEpoch }: XrayInbo )} - {inbound && inbound.protocol !== 'unmanaged' && !('transport' in inbound) && !isTunnelInboundProtocol(inbound.protocol) && ( + {inbound && inbound.protocol !== 'unmanaged' && (isWireguardInboundProtocol(inbound.protocol) || !('transport' in inbound)) && !isTunnelInboundProtocol(inbound.protocol) && (
e.preventDefault()}>
diff --git a/dashboard/src/features/core-editor/kit/sanitize-inbound.ts b/dashboard/src/features/core-editor/kit/sanitize-inbound.ts index 99f0860b8..fa6b9b0fe 100644 --- a/dashboard/src/features/core-editor/kit/sanitize-inbound.ts +++ b/dashboard/src/features/core-editor/kit/sanitize-inbound.ts @@ -168,6 +168,8 @@ function sanitizeWireguardEmptyAddress(inbound: Inbound): Inbound { if (Array.isArray(draft.address) && draft.address.length === 0) { delete draft.address } + delete draft.transport + delete draft.security return draft as unknown as Inbound } From 619a0738817cea796c59accc60e0af3e65917481 Mon Sep 17 00:00:00 2001 From: bitwiresys Date: Wed, 22 Jul 2026 22:07:06 +0300 Subject: [PATCH 4/5] fix wireguard host creation failing on missing pre-shared key --- app/core/hosts.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/app/core/hosts.py b/app/core/hosts.py index 8028b426f..b73e4ec3a 100644 --- a/app/core/hosts.py +++ b/app/core/hosts.py @@ -105,7 +105,7 @@ async def _prepare_subscription_inbound_data( transport_config=TCPTransportConfig(path="", host=[]), mux_settings=None, wireguard_public_key=inbound_config.get("public_key", ""), - wireguard_pre_shared_key=inbound_config.get("pre_shared_key", None), + wireguard_pre_shared_key=inbound_config.get("pre_shared_key") or "", wireguard_local_address=inbound_config.get("address", []) or [], wireguard_allowed_ips=allowed_ips, wireguard_keepalive=keepalive, From 2bf15bb5aac00c1391850a3393c3acd951b8c429 Mon Sep 17 00:00:00 2001 From: bitwiresys Date: Mon, 27 Jul 2026 06:05:41 +0300 Subject: [PATCH 5/5] keep record usages formatting consistent with upstream --- app/jobs/record_usages.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/app/jobs/record_usages.py b/app/jobs/record_usages.py index 6c3e9e6dd..71e347fd1 100644 --- a/app/jobs/record_usages.py +++ b/app/jobs/record_usages.py @@ -42,6 +42,7 @@ _thread_pool = None _thread_pool_lock = asyncio.Lock() + async def _get_thread_pool(): """Get or create the thread pool executor (thread-safe).""" global _thread_pool @@ -691,9 +692,7 @@ async def _record_user_usages_impl(): usage_coefficient[node_id] = data.get("usage_coefficient", 1) if data else 1.0 # 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) for _, node in nodes], return_exceptions=True) api_params = {} for i, result in enumerate(stats_results): node_id = nodes[i][0]