fix(wireguard): get_client_config accepts port; route link flow via _manager_call

Linking an existing WireGuard peer to a user crashed with
'WireGuardManager.get_client_config() takes 3 positional arguments but
5 were given': the user-link endpoint called the manager directly,
bypassing the _manager_call shim, and passed the AWG-style argument
list (protocol, client_id, host, port).

- WireGuardManager.get_client_config now takes an optional port; when
  omitted it falls back to the listen port from the server config, so
  existing direct callers keep working.
- The link endpoint goes through _manager_call like every other call
  site, so each manager gets its own signature.

Adds tests/test_wireguard_get_client_config.py (3 cases).
This commit is contained in:
leonidorlov-hash committed 2026-09-13 22:39:28 +03:00
1 parent dbccc1ec40
commit 37031761a1
3 files changed
+69 -4

No files matched your search

+1 -1
View File
@@ -5167,7 +5167,7 @@ async def api_add_user_connection(request: Request, user_id: str, req: AddUserCo
# Use existing client
target_client_id = req.client_id
# Retrieve config for existing client
config = await asyncio.to_thread(manager.get_client_config, req.protocol, req.client_id, server['host'], port)
config = await asyncio.to_thread(_manager_call, manager, 'get_client_config', req.protocol, req.client_id, server['host'], port)
result = {'client_id': target_client_id, 'config': config}
else:
# Create new client
+8 -3
View File
@@ -855,8 +855,13 @@ PersistentKeepalive = 25
'config': client_config,
}
def get_client_config(self, client_id, server_host):
"""Reconstruct client config from stored data."""
def get_client_config(self, client_id, server_host, port=None):
"""Reconstruct client config from stored data.
`port` is optional: when omitted (direct calls) the listen port is
read from the server config; when passed by the panel (which already
knows the instance port) it is used for the Endpoint.
"""
clients_table = self._get_clients_table()
client = next((c for c in clients_table if c.get('clientId') == client_id), None)
if not client:
@@ -876,7 +881,7 @@ PersistentKeepalive = 25
if not psk:
psk = self._get_server_psk()
port = self._get_listen_port()
port = port if port is not None else self._get_listen_port()
dns = self._get_dns(ud)
+60
View File
@@ -0,0 +1,60 @@
"""WireGuardManager.get_client_config signature compatibility.
The panel calls manager methods through `_manager_call`, which passes
`(client_id, server_host, port)` for WireGuard. The WG manager used to accept
only `(client_id, server_host)`, so linking an existing WG peer to a user
crashed with "takes 3 positional arguments but 5 were given" (protocol id
included). The manager now accepts an optional `port`; when given it is used
for the Endpoint, otherwise the listen port is read from the server config.
"""
import unittest
from unittest import mock
from managers.wireguard_manager import WireGuardManager
CLIENT = {
'clientId': 'pubkey123',
'userData': {
'clientPrivateKey': 'privkey',
'clientIp': '10.8.0.6',
'psk': 'psk123',
},
}
def make_manager():
mgr = WireGuardManager(ssh_manager=mock.Mock())
mgr._get_clients_table = lambda: [CLIENT]
mgr._get_server_public_key = lambda: 'serverpub'
mgr._get_server_psk = lambda: 'serverpsk'
mgr._get_listen_port = lambda: '51820'
mgr._get_dns = lambda ud=None: '1.1.1.1'
return mgr
class GetClientConfigTests(unittest.TestCase):
def test_without_port_reads_listen_port(self):
mgr = make_manager()
config = mgr.get_client_config('pubkey123', 'example.com')
self.assertIn('Endpoint = example.com:51820', config)
self.assertIn('PrivateKey = privkey', config)
def test_explicit_port_overrides_listen_port(self):
mgr = make_manager()
called = []
mgr._get_listen_port = lambda: called.append(1) or '51820'
config = mgr.get_client_config('pubkey123', 'example.com', '55424')
self.assertIn('Endpoint = example.com:55424', config)
self.assertEqual(called, [], 'explicit port must skip _get_listen_port')
def test_manager_call_shim_signature(self):
"""Simulate the shim path: WG gets (client_id, host, port)."""
mgr = make_manager()
config = mgr.get_client_config('pubkey123', 'example.com', '55424')
self.assertIn('[Interface]', config)
if __name__ == '__main__':
unittest.main()