feat(wireguard): support WireGuard as an Xray inbound

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
 

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]

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)