fix(api): keep blocking SSH work out of the event loop

One hung server (15s connect x retries) or the auto-backup sweep used to
freeze the whole panel for minutes, local endpoints included.

- 66 endpoints that only do blocking work converted from async def to
  def so Starlette runs them in the threadpool
- mixed endpoints (exit-link awaits, file uploads, DATA_LOCK sections)
  stay async; their blocking SSH calls now go through asyncio.to_thread
- get_ssh() (which blocks in ensure_connected) is no longer called from
  async code in app.py, ExitLinkService or ConnectionService
- background tasks were already threaded/to_thread; DATA_LOCK and
  TUNNEL_LOCK semantics untouched

app.py

@@ -1657,7 +1657,7 @@ def verify_password(password: str, password_hash: str) -> bool:
         return False
 
 
-async def perform_delete_user(data: dict, user_id: str):
+def perform_delete_user(data: dict, user_id: str):
     user = next((u for u in data['users'] if u['id'] == user_id), None)
     if not user:
         return False
@@ -1695,7 +1695,7 @@ async def perform_toggle_user(data: dict, user_id: str, enable: bool) -> bool:
             if sid >= len(data['servers']):
                 continue
             server = data['servers'][sid]
-            ssh = get_ssh(server)
+            ssh = await asyncio.to_thread(get_ssh, server)
             await asyncio.to_thread(ssh.connect)
             manager = get_protocol_manager(ssh, uc['protocol'])
             await asyncio.to_thread(
@@ -1742,7 +1742,7 @@ async def perform_mass_operations(delete_uids: List[str] = None, toggle_uids: Li
         srv = current_data['servers'][srv_id]
         
         try:
-            ssh = get_ssh(srv)
+            ssh = await asyncio.to_thread(get_ssh, srv)
             await asyncio.to_thread(ssh.connect)
             
             # 1. Deletes
@@ -1996,7 +1996,7 @@ def tpl(request, template, **kwargs):
 
 
 @app.get('/manifest.webmanifest')
-async def web_manifest(request: Request):
+def web_manifest(request: Request):
     """Installable PWA manifest — public, no auth (browsers fetch without credentials)."""
     data = load_data()
     lang = request.cookies.get('lang', 'en')
@@ -2008,7 +2008,7 @@ async def web_manifest(request: Request):
 
 
 @app.get('/sw.js')
-async def service_worker():
+def service_worker():
     """Root-scoped service worker. Must not live under /static/ or scope is confined."""
     path = os.path.join(application_path, 'static', 'sw.js')
     return FileResponse(
@@ -2756,14 +2756,14 @@ async def periodic_background_tasks():
 # ======================== PAGE ROUTES ========================
 
 @app.get('/login', response_class=HTMLResponse, tags=["System Templates"])
-async def login_page(request: Request):
+def login_page(request: Request):
     if get_current_user(request):
         return RedirectResponse(url='/', status_code=302)
     return tpl(request, 'login.html')
 
 
 @app.get("/set_lang/{lang}", tags=["System Templates"])
-async def set_lang(lang: str, request: Request):
+def set_lang(lang: str, request: Request):
     ref = request.headers.get("referer", "/")
     response = RedirectResponse(url=ref)
     response.set_cookie(key="lang", value=lang, max_age=31536000)
@@ -2771,13 +2771,13 @@ async def set_lang(lang: str, request: Request):
 
 
 @app.get('/logout', tags=["System Templates"])
-async def logout(request: Request):
+def logout(request: Request):
     request.session.clear()
     return RedirectResponse(url='/login', status_code=302)
 
 
 @app.get('/', response_class=HTMLResponse, tags=["System Templates"])
-async def index(request: Request):
+def index(request: Request):
     user = get_current_user(request)
     if not user:
         return RedirectResponse(url='/login', status_code=302)
@@ -2788,7 +2788,7 @@ async def index(request: Request):
 
 
 @app.get('/server/{server_id}', response_class=HTMLResponse, tags=["System Templates"])
-async def server_detail(request: Request, server_id: int):
+def server_detail(request: Request, server_id: int):
     user = get_current_user(request)
     if not user:
         return RedirectResponse(url='/login', status_code=302)
@@ -2803,7 +2803,7 @@ async def server_detail(request: Request, server_id: int):
 
 
 @app.get('/users', response_class=HTMLResponse, tags=["System Templates"])
-async def users_page(request: Request):
+def users_page(request: Request):
     user = get_current_user(request)
     if not user:
         return RedirectResponse(url='/login', status_code=302)
@@ -2820,7 +2820,7 @@ async def users_page(request: Request):
 
 
 @app.get('/my', response_class=HTMLResponse, tags=["System Templates"])
-async def my_connections_page(request: Request):
+def my_connections_page(request: Request):
     user = get_current_user(request)
     if not user:
         return RedirectResponse(url='/login', status_code=302)
@@ -2839,7 +2839,7 @@ async def my_connections_page(request: Request):
 # ======================== AUTH API ========================
 
 @app.get('/api/auth/captcha', tags=["Authentication"])
-async def api_captcha(request: Request):
+def api_captcha(request: Request):
     if not CaptchaGenerator:
         return JSONResponse({"error": "multicolorcaptcha is not installed"}, status_code=500)
     
@@ -2856,7 +2856,7 @@ async def api_captcha(request: Request):
 
 
 @app.post('/api/auth/login', tags=["Authentication"])
-async def api_login(request: Request, req: LoginRequest):
+def api_login(request: Request, req: LoginRequest):
     data = load_data()
     captcha_settings = data.get('settings', {}).get('captcha', {})
     if captcha_settings.get('enabled') is True:
@@ -2915,7 +2915,7 @@ def _check_admin(request):
 
 
 @app.post('/api/servers/add', tags=["Servers"])
-async def api_add_server(request: Request, req: AddServerRequest):
+def api_add_server(request: Request, req: AddServerRequest):
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
     try:
@@ -2952,7 +2952,7 @@ async def api_add_server(request: Request, req: AddServerRequest):
 
 
 @app.post('/api/servers/{server_id}/edit', tags=["Servers"])
-async def api_edit_server(request: Request, server_id: int, req: EditServerRequest):
+def api_edit_server(request: Request, server_id: int, req: EditServerRequest):
     """Update connection details for an existing server entry. Verifies the new
     credentials by SSH-connecting before persisting, so a typo can't lock us out.
     """
@@ -3097,7 +3097,7 @@ async def api_delete_server(request: Request, server_id: int):
         if server_id >= len(data['servers']):
             return JSONResponse({'error': 'Server not found'}, status_code=404)
         server = data['servers'][server_id]
-        drop_ssh(server)
+        await asyncio.to_thread(drop_ssh, server)
         data['servers'].pop(server_id)
         # Clean up connections for this server
         data['user_connections'] = [c for c in data.get('user_connections', []) if c.get('server_id') != server_id]
@@ -3112,7 +3112,7 @@ async def api_delete_server(request: Request, server_id: int):
 
 
 @app.post('/api/servers/{server_id}/reboot', tags=["Servers"])
-async def api_reboot_server(request: Request, server_id: int):
+def api_reboot_server(request: Request, server_id: int):
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
     try:
@@ -3153,8 +3153,8 @@ async def api_clear_server(request: Request, server_id: int):
             logger.warning(f"exit-link cleanup before clear failed: {e}")
         data = load_data()
         server = data['servers'][server_id]
-        ssh = get_ssh(server)
-        ssh.connect()
+        ssh = await asyncio.to_thread(get_ssh, server)
+        await asyncio.to_thread(ssh.connect)
         # Match every Amnezia container by name prefix (catches awg/awg2/awg-legacy,
         # wireguard, xray/ssxray, openvpn, dns, and any future amnezia-* protocol)
         # plus the telemt container which doesn't share that prefix.
@@ -3173,7 +3173,7 @@ done
 docker network rm amnezia-dns-net >/dev/null 2>&1 || true
 rm -rf /opt/amnezia
 """
-        ssh.run_sudo_script(cleanup_script, timeout=120)
+        await asyncio.to_thread(ssh.run_sudo_script, cleanup_script, timeout=120)
 
         server['protocols'] = {}
         save_data(data)
@@ -3185,7 +3185,7 @@ rm -rf /opt/amnezia
 
 
 @app.post('/api/servers/{server_id}/stats', tags=["Servers"])
-async def api_server_stats(request: Request, server_id: int):
+def api_server_stats(request: Request, server_id: int):
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
     try:
@@ -3238,7 +3238,7 @@ async def api_server_stats(request: Request, server_id: int):
 
 
 @app.post('/api/servers/{server_id}/check', tags=["Servers"])
-async def api_check_server(request: Request, server_id: int):
+def api_check_server(request: Request, server_id: int):
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
     try:
@@ -3435,7 +3435,7 @@ INSTALL_PORT_TRANSPORT = {
 
 
 @app.get('/api/servers/{server_id}/used-ports', tags=["Servers"])
-async def api_used_ports(request: Request, server_id: int):
+def api_used_ports(request: Request, server_id: int):
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
     try:
@@ -3488,14 +3488,14 @@ async def api_install_protocol(request: Request, server_id: int, req: InstallPro
             except ValueError as e:
                 return JSONResponse({'error': str(e)}, status_code=400)
 
-        ssh = get_ssh(server)
-        ssh.connect()
+        ssh = await asyncio.to_thread(get_ssh, server)
+        await asyncio.to_thread(ssh.connect)
 
         # Validate the requested port before touching Docker: a busy port
         # otherwise fails mid-install with a half-created broken container.
         transport = INSTALL_PORT_TRANSPORT.get(install_base)
         if transport and req.port and str(req.port).isdigit():
-            used = get_used_ports(ssh)
+            used = await asyncio.to_thread(get_used_ports, ssh)
             owner = used.get(transport, {}).get(str(req.port))
             if owner:
                 return JSONResponse(
@@ -3503,12 +3503,14 @@ async def api_install_protocol(request: Request, server_id: int, req: InstallPro
                               f'Choose another port.'},
                     status_code=400)
 
-        docker_install_log = ensure_docker_installed(ssh)
+        docker_install_log = await asyncio.to_thread(ensure_docker_installed, ssh)
         manager = get_protocol_manager(ssh, install_protocol)
 
         # Pass parameters to installer
+        install_kwargs = None
         if install_base == 'telemt':
-            result = manager.install_protocol(
+            install_args = ()
+            install_kwargs = dict(
                 protocol_type=install_protocol,
                 port=req.port,
                 tls_emulation=req.tls_emulation if req.tls_emulation is not None else True,
@@ -3516,18 +3518,22 @@ async def api_install_protocol(request: Request, server_id: int, req: InstallPro
                 max_connections=req.max_connections if req.max_connections is not None else 0
             )
         elif install_base == 'xray':
-            result = manager.install_protocol(port=req.port)
+            install_args = ()
+            install_kwargs = dict(port=req.port)
         elif install_base == 'wireguard':
-            result = manager.install_protocol(port=req.port)
+            install_args = ()
+            install_kwargs = dict(port=req.port)
         elif install_base == 'socks5':
-            result = manager.install_protocol(
+            install_args = ()
+            install_kwargs = dict(
                 protocol_type=install_protocol,
                 port=req.port,
                 username=req.socks5_username,
                 password=req.socks5_password,
             )
         elif install_base == 'adguard':
-            result = manager.install_protocol(
+            install_args = ()
+            install_kwargs = dict(
                 protocol_type='adguard',
                 mode=req.adguard_mode or 'sidebyside',
                 web_port=req.adguard_web_port,
@@ -3540,29 +3546,33 @@ async def api_install_protocol(request: Request, server_id: int, req: InstallPro
                 expose_doh=bool(req.adguard_expose_doh),
             )
         elif install_base == 'nginx':
-            result = manager.install_protocol(
+            install_args = ()
+            install_kwargs = dict(
                 protocol_type='nginx',
                 port=req.port,
                 domain=req.nginx_domain,
                 email=req.nginx_email,
             )
         elif install_base == 'exit':
-            result = manager.install_protocol(
+            install_args = ()
+            install_kwargs = dict(
                 protocol_type='exit',
                 port=req.port,
                 subnet=req.exit_subnet,
                 obfuscation=bool(req.exit_obfuscation),
             )
         elif install_base in AWG_PROTOCOLS:
-            result = manager.install_protocol(
-                install_protocol,
+            install_args = (install_protocol,)
+            install_kwargs = dict(
                 port=req.port,
                 mtu=req.awg_mtu,
                 dns=join_dns(req.awg_dns1, req.awg_dns2),
                 special_junk=awg_special_junk,
             )
         else:
-            result = manager.install_protocol(install_protocol, port=req.port)
+            install_args = (install_protocol,)
+            install_kwargs = dict(port=req.port)
+        result = await asyncio.to_thread(manager.install_protocol, *install_args, **install_kwargs)
 
         if not isinstance(result, dict):
             result = {'status': 'success', 'message': str(result)}
@@ -3650,7 +3660,7 @@ async def api_install_protocol(request: Request, server_id: int, req: InstallPro
 
 
 @app.get('/api/servers/{server_id}/socks5/credentials', tags=["Protocols"])
-async def api_socks5_get_credentials(request: Request, server_id: int, protocol: str = 'socks5'):
+def api_socks5_get_credentials(request: Request, server_id: int, protocol: str = 'socks5'):
     """Return the current SOCKS5 port/username/password for the panel UI."""
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -3672,7 +3682,7 @@ async def api_socks5_get_credentials(request: Request, server_id: int, protocol:
 
 
 @app.post('/api/servers/{server_id}/socks5/credentials', tags=["Protocols"])
-async def api_socks5_update_credentials(request: Request, server_id: int, req: Socks5SettingsRequest):
+def api_socks5_update_credentials(request: Request, server_id: int, req: Socks5SettingsRequest):
     """Apply new SOCKS5 connection settings — regenerates the 3proxy config and
     reconciles the container (recreating it if the listening port changed)."""
     if not _check_admin(request):
@@ -3713,7 +3723,7 @@ def _exit_manager_for(ssh):
 
 
 @app.post('/api/servers/{server_id}/exit/peers', tags=["Protocols"])
-async def api_exit_peers(request: Request, server_id: int, req: ProtocolRequest):
+def api_exit_peers(request: Request, server_id: int, req: ProtocolRequest):
     """Exit node endpoint data (public key, port, transit subnet, obfuscation)
     and its peers - the entry nodes linked to it - with live handshake and
     transfer counters."""
@@ -3738,7 +3748,7 @@ async def api_exit_peers(request: Request, server_id: int, req: ProtocolRequest)
 
 
 @app.post('/api/servers/{server_id}/exit/peers/add', tags=["Protocols"])
-async def api_exit_peer_add(request: Request, server_id: int, req: ExitPeerAddRequest):
+def api_exit_peer_add(request: Request, server_id: int, req: ExitPeerAddRequest):
     """Register a peer on the exit node by hand (a node not managed by this
     panel). Upserts by `peer_id`; returns the transit address, a fresh PSK and
     the exit's endpoint data for the peer's own WireGuard config."""
@@ -3767,7 +3777,7 @@ async def api_exit_peer_add(request: Request, server_id: int, req: ExitPeerAddRe
 
 
 @app.post('/api/servers/{server_id}/exit/peers/remove', tags=["Protocols"])
-async def api_exit_peer_remove(request: Request, server_id: int, req: ExitPeerRemoveRequest):
+def api_exit_peer_remove(request: Request, server_id: int, req: ExitPeerRemoveRequest):
     """Drop a peer from the exit node by public key or by peer id."""
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -3796,7 +3806,7 @@ def _exit_link_error(e: ExitLinkError):
 
 
 @app.get('/api/exit-nodes', tags=["Servers"])
-async def api_exit_nodes(request: Request):
+def api_exit_nodes(request: Request):
     """Servers with an installed exit node, from data.json (no SSH)."""
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -3892,7 +3902,7 @@ async def api_exit_link_check_egress(request: Request, server_id: int, req: Prot
 
 
 @app.post('/api/servers/{server_id}/awg/settings', tags=["Protocols"])
-async def api_awg_settings_get(request: Request, server_id: int, req: ProtocolRequest):
+def api_awg_settings_get(request: Request, server_id: int, req: ProtocolRequest):
     """Return MTU, DNS and the special junk packets I1-I5 of an AWG server."""
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -3916,7 +3926,7 @@ async def api_awg_settings_get(request: Request, server_id: int, req: ProtocolRe
 
 
 @app.post('/api/servers/{server_id}/awg/settings/save', tags=["Protocols"])
-async def api_awg_settings_save(request: Request, server_id: int, req: AwgSettingsRequest):
+def api_awg_settings_save(request: Request, server_id: int, req: AwgSettingsRequest):
     """Update MTU, DNS and I1-I5 and apply them to the running interface."""
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -3983,13 +3993,13 @@ async def api_uninstall_protocol(request: Request, server_id: int, req: Protocol
                 logger.warning(f"unlink before uninstall failed: {e}")
             data = load_data()
             server = data['servers'][server_id]
-        ssh = get_ssh(server)
-        ssh.connect()
+        ssh = await asyncio.to_thread(get_ssh, server)
+        await asyncio.to_thread(ssh.connect)
         manager = get_protocol_manager(ssh, req.protocol)
         if base in ('xray', 'wireguard'):
-            manager.remove_container()
+            await asyncio.to_thread(manager.remove_container)
         else:
-            manager.remove_container(req.protocol)
+            await asyncio.to_thread(manager.remove_container, req.protocol)
         if req.protocol in server.get('protocols', {}):
             del server['protocols'][req.protocol]
             save_data(data)
@@ -4025,7 +4035,7 @@ CONTAINER_NAMES = {
 
 
 @app.post('/api/servers/{server_id}/backups', tags=["Protocols"])
-async def api_protocol_backups_list(request: Request, server_id: int, req: ProtocolRequest):
+def api_protocol_backups_list(request: Request, server_id: int, req: ProtocolRequest):
     """List backups created on the remote server for one protocol."""
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -4052,7 +4062,7 @@ async def api_protocol_backups_list(request: Request, server_id: int, req: Proto
 
 
 @app.post('/api/servers/{server_id}/backups/create', tags=["Protocols"])
-async def api_protocol_backup_create(request: Request, server_id: int, req: ProtocolRequest):
+def api_protocol_backup_create(request: Request, server_id: int, req: ProtocolRequest):
     """Create a protocol backup archive on the remote server."""
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -4082,7 +4092,7 @@ async def api_protocol_backup_create(request: Request, server_id: int, req: Prot
 
 
 @app.post('/api/servers/{server_id}/backups/download', tags=["Protocols"])
-async def api_protocol_backup_download(request: Request, server_id: int, req: BackupDownloadRequest):
+def api_protocol_backup_download(request: Request, server_id: int, req: BackupDownloadRequest):
     """Download one remote protocol backup archive through the panel."""
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -4169,9 +4179,9 @@ async def api_protocol_backup_upload(
         os.close(fd)
         fd = None
         server = data['servers'][server_id]
-        ssh = get_ssh(server)
-        ssh.connect()
-        result = BackupManager(ssh).upload_backup(protocol, file.filename, tmp_path)
+        ssh = await asyncio.to_thread(get_ssh, server)
+        await asyncio.to_thread(ssh.connect)
+        result = await asyncio.to_thread(BackupManager(ssh).upload_backup, protocol, file.filename, tmp_path)
         if result.get('status') == 'error':
             return JSONResponse({'error': result.get('message', 'Failed to upload backup')}, status_code=400)
         return result
@@ -4207,9 +4217,9 @@ async def api_protocol_backup_restore(request: Request, server_id: int, req: Bac
         container = protocol_container_name(req.protocol)
         if not container:
             return JSONResponse({'error': 'Unknown protocol'}, status_code=400)
-        ssh = get_ssh(server)
-        ssh.connect()
-        result = BackupManager(ssh).restore_backup(req.protocol, container, filename)
+        ssh = await asyncio.to_thread(get_ssh, server)
+        await asyncio.to_thread(ssh.connect)
+        result = await asyncio.to_thread(BackupManager(ssh).restore_backup, req.protocol, container, filename)
         if result.get('status') == 'error':
             return JSONResponse({'error': result.get('message', 'Failed to restore backup')}, status_code=500)
         ssh.disconnect()
@@ -4230,7 +4240,7 @@ async def api_protocol_backup_restore(request: Request, server_id: int, req: Bac
 
 
 @app.post('/api/servers/{server_id}/backups/delete', tags=["Protocols"])
-async def api_protocol_backup_delete(request: Request, server_id: int, req: BackupDownloadRequest):
+def api_protocol_backup_delete(request: Request, server_id: int, req: BackupDownloadRequest):
     """Delete one protocol backup archive on the remote server."""
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -4260,7 +4270,7 @@ async def api_protocol_backup_delete(request: Request, server_id: int, req: Back
 
 
 @app.post('/api/servers/{server_id}/container/toggle', tags=["Protocols"])
-async def api_container_toggle(request: Request, server_id: int, req: ContainerToggleRequest):
+def api_container_toggle(request: Request, server_id: int, req: ContainerToggleRequest):
     """Start or stop a protocol Docker container. Stopping an exit node that
     entries route through is refused (409 `exit_in_use`) unless `force` is
     set - their clients would silently land on the kill-switch."""
@@ -4300,7 +4310,7 @@ async def api_container_toggle(request: Request, server_id: int, req: ContainerT
 
 
 @app.post('/api/servers/{server_id}/server_config', tags=["Protocols"])
-async def api_server_config(request: Request, server_id: int, req: ProtocolRequest):
+def api_server_config(request: Request, server_id: int, req: ProtocolRequest):
     """Get the raw server-side WireGuard/Xray configuration."""
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -4340,7 +4350,7 @@ async def api_server_config(request: Request, server_id: int, req: ProtocolReque
 
 
 @app.post('/api/servers/{server_id}/host_tuning', tags=["Protocols"])
-async def api_host_tuning(request: Request, server_id: int):
+def api_host_tuning(request: Request, server_id: int):
     """Server-level network tuning summary (host sysctls + AWG containers)."""
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -4361,7 +4371,7 @@ async def api_host_tuning(request: Request, server_id: int):
 
 
 @app.post('/api/servers/{server_id}/wgeasy/preview', tags=["Protocols"])
-async def api_wgeasy_preview(request: Request, server_id: int, req: WgEasyPreviewRequest):
+def api_wgeasy_preview(request: Request, server_id: int, req: WgEasyPreviewRequest):
     """Fetch the client list from a wg-easy / amnezia-wg-easy panel running on
     this server (via its local web API over SSH). No secrets are returned."""
     if not _check_admin(request):
@@ -4405,7 +4415,7 @@ async def api_wgeasy_preview(request: Request, server_id: int, req: WgEasyPrevie
 
 
 @app.post('/api/servers/{server_id}/protocol/rename', tags=["Protocols"])
-async def api_rename_protocol(request: Request, server_id: int, req: RenameProtocolRequest):
+def api_rename_protocol(request: Request, server_id: int, req: RenameProtocolRequest):
     """Set or clear a custom display name for an installed protocol instance."""
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -4432,7 +4442,7 @@ async def api_rename_protocol(request: Request, server_id: int, req: RenameProto
 
 
 @app.post('/api/servers/{server_id}/wgeasy/import', tags=["Protocols"])
-async def api_wgeasy_import(request: Request, server_id: int, req: WgEasyImportRequest):
+def api_wgeasy_import(request: Request, server_id: int, req: WgEasyImportRequest):
     """Migrate clients from a wg-easy panel on this server into a panel-managed
     WireGuard instance, preserving keys/IPs/port so client configs keep working."""
     if not _check_admin(request):
@@ -4490,7 +4500,7 @@ async def api_wgeasy_import(request: Request, server_id: int, req: WgEasyImportR
 
 
 @app.post('/api/servers/{server_id}/server_config/save', tags=["Protocols"])
-async def api_server_config_save(request: Request, server_id: int, req: ServerConfigSaveRequest):
+def api_server_config_save(request: Request, server_id: int, req: ServerConfigSaveRequest):
     """Save the raw server-side WireGuard/Xray configuration and apply changes."""
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -4537,7 +4547,7 @@ async def api_server_config_save(request: Request, server_id: int, req: ServerCo
 
 
 @app.post('/api/servers/{server_id}/nginx/site', tags=["Protocols"])
-async def api_nginx_site_get(request: Request, server_id: int, req: ProtocolRequest):
+def api_nginx_site_get(request: Request, server_id: int, req: ProtocolRequest):
     """Return editable NGINX site index.html."""
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -4561,7 +4571,7 @@ async def api_nginx_site_get(request: Request, server_id: int, req: ProtocolRequ
 
 
 @app.post('/api/servers/{server_id}/nginx/site/save', tags=["Protocols"])
-async def api_nginx_site_save(request: Request, server_id: int, req: NginxSiteSaveRequest):
+def api_nginx_site_save(request: Request, server_id: int, req: NginxSiteSaveRequest):
     """Save editable NGINX site index.html."""
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -4584,7 +4594,7 @@ async def api_nginx_site_save(request: Request, server_id: int, req: NginxSiteSa
         return JSONResponse({'error': str(e)}, status_code=500)
 
 @app.get('/api/servers/{server_id}/connections', tags=["Connections"])
-async def api_get_connections(request: Request, server_id: int, protocol: str = Query(default='awg')):
+def api_get_connections(request: Request, server_id: int, protocol: str = Query(default='awg')):
     if not protocol:
         protocol = 'awg'
     if not _check_admin(request):
@@ -4621,7 +4631,7 @@ async def api_get_connections(request: Request, server_id: int, protocol: str =
 
 
 @app.post('/api/servers/{server_id}/connections/add', tags=["Connections"])
-async def api_add_connection(request: Request, server_id: int, req: AddConnectionRequest):
+def api_add_connection(request: Request, server_id: int, req: AddConnectionRequest):
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
     try:
@@ -4675,7 +4685,7 @@ async def api_add_connection(request: Request, server_id: int, req: AddConnectio
 
 
 @app.post('/api/servers/{server_id}/connections/remove', tags=["Connections"])
-async def api_remove_connection(request: Request, server_id: int, req: ConnectionActionRequest):
+def api_remove_connection(request: Request, server_id: int, req: ConnectionActionRequest):
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
     try:
@@ -4703,7 +4713,7 @@ async def api_remove_connection(request: Request, server_id: int, req: Connectio
 
 
 @app.post('/api/servers/{server_id}/connections/edit', tags=["Connections"])
-async def api_edit_connection(request: Request, server_id: int, req: EditConnectionRequest):
+def api_edit_connection(request: Request, server_id: int, req: EditConnectionRequest):
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
     try:
@@ -4734,7 +4744,7 @@ async def api_edit_connection(request: Request, server_id: int, req: EditConnect
 
 
 @app.post('/api/servers/{server_id}/connections/clear_warnings', tags=["Connections"])
-async def api_clear_conn_warnings(request: Request, server_id: int, req: ConnectionActionRequest):
+def api_clear_conn_warnings(request: Request, server_id: int, req: ConnectionActionRequest):
     """Clear recorded connection-flood (torrent) warnings for one peer."""
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -4755,7 +4765,7 @@ async def api_clear_conn_warnings(request: Request, server_id: int, req: Connect
 
 
 @app.post('/api/servers/{server_id}/connections/rename', tags=["Connections"])
-async def api_rename_connection(request: Request, server_id: int, req: RenameConnectionRequest):
+def api_rename_connection(request: Request, server_id: int, req: RenameConnectionRequest):
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
     try:
@@ -4802,7 +4812,7 @@ async def api_rename_connection(request: Request, server_id: int, req: RenameCon
 
 
 @app.post('/api/servers/{server_id}/connections/config/save', tags=["Connections"])
-async def api_save_connection_config(request: Request, server_id: int, req: SaveConnectionConfigRequest):
+def api_save_connection_config(request: Request, server_id: int, req: SaveConnectionConfigRequest):
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
     try:
@@ -4827,7 +4837,7 @@ async def api_save_connection_config(request: Request, server_id: int, req: Save
 
 
 @app.post('/api/servers/{server_id}/connections/config', tags=["Connections"])
-async def api_get_connection_config(request: Request, server_id: int, req: ConnectionActionRequest):
+def api_get_connection_config(request: Request, server_id: int, req: ConnectionActionRequest):
     user = get_current_user(request)
     if not user:
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -4860,7 +4870,7 @@ async def api_get_connection_config(request: Request, server_id: int, req: Conne
 
 
 @app.post('/api/servers/{server_id}/connections/toggle', tags=["Connections"])
-async def api_toggle_connection(request: Request, server_id: int, req: ToggleConnectionRequest):
+def api_toggle_connection(request: Request, server_id: int, req: ToggleConnectionRequest):
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
     try:
@@ -4885,7 +4895,7 @@ async def api_toggle_connection(request: Request, server_id: int, req: ToggleCon
 # ======================== USER API (admin only) ========================
 
 @app.get('/api/users', tags=["Users"])
-async def api_list_users(request: Request, search: str = '', page: int = 1, size: int = 10):
+def api_list_users(request: Request, search: str = '', page: int = 1, size: int = 10):
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
     data = load_data()
@@ -4940,7 +4950,7 @@ async def api_list_users(request: Request, search: str = '', page: int = 1, size
 
 
 @app.post('/api/users/add', tags=["Users"])
-async def api_add_user(request: Request, req: AddUserRequest):
+def api_add_user(request: Request, req: AddUserRequest):
     cur = get_current_user(request)
     if not cur or cur['role'] != 'admin':
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -5095,7 +5105,7 @@ async def api_delete_user(request: Request, user_id: str):
         return JSONResponse({'error': _t('cannot_delete_self', lang)}, status_code=400)
     try:
         data = load_data()
-        success = await perform_delete_user(data, user_id)
+        success = await asyncio.to_thread(perform_delete_user, data, user_id)
         if not success:
             return JSONResponse({'error': 'User not found'}, status_code=404)
         save_data(data)
@@ -5123,7 +5133,7 @@ async def api_toggle_user(request: Request, user_id: str, req: ToggleUserRequest
 
 
 @app.post('/api/users/{user_id}/traffic/reset', tags=["Users"])
-async def api_reset_user_traffic(request: Request, user_id: str):
+def api_reset_user_traffic(request: Request, user_id: str):
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
     try:
@@ -5159,7 +5169,7 @@ async def api_add_user_connection(request: Request, user_id: str, req: AddUserCo
         server = data['servers'][req.server_id]
         proto_info = server.get('protocols', {}).get(req.protocol, {})
         port = proto_info.get('port', '55424')
-        ssh = get_ssh(server)
+        ssh = await asyncio.to_thread(get_ssh, server)
         await asyncio.to_thread(ssh.connect)
         manager = get_protocol_manager(ssh, req.protocol)
         
@@ -5211,7 +5221,7 @@ async def api_add_user_connection(request: Request, user_id: str, req: AddUserCo
 
 
 @app.get('/api/users/{user_id}/connections', tags=["Users"])
-async def api_get_user_connections(request: Request, user_id: str):
+def api_get_user_connections(request: Request, user_id: str):
     user = get_current_user(request)
     if not user:
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -5230,7 +5240,7 @@ async def api_get_user_connections(request: Request, user_id: str):
 # ======================== MY CONNECTIONS API (for user role) ========================
 
 @app.get('/api/my/connections', tags=["Self-service"])
-async def api_my_connections(request: Request):
+def api_my_connections(request: Request):
     user = get_current_user(request)
     if not user:
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -5282,7 +5292,7 @@ async def api_my_connection_delete(request: Request, connection_id: str):
 
 
 @app.post('/api/users/{user_id}/share/setup', tags=["Users"])
-async def api_user_share_setup(user_id: str, req: ShareSetupRequest, request: Request):
+def api_user_share_setup(user_id: str, req: ShareSetupRequest, request: Request):
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
     data = load_data()
@@ -5303,7 +5313,7 @@ async def api_user_share_setup(user_id: str, req: ShareSetupRequest, request: Re
 
 
 @app.get('/share/{token}', response_class=HTMLResponse, tags=["System Templates"])
-async def share_page(token: str, request: Request):
+def share_page(token: str, request: Request):
     data = load_data()
     user = next((u for u in data['users'] if u.get('share_token') == token), None)
     if not user or not user.get('share_enabled'):
@@ -5320,7 +5330,7 @@ async def share_page(token: str, request: Request):
 
 
 @app.post('/api/share/{token}/auth', tags=["Sharing"])
-async def api_share_auth(token: str, req: ShareAuthRequest, request: Request):
+def api_share_auth(token: str, req: ShareAuthRequest, request: Request):
     data = load_data()
     user = next((u for u in data['users'] if u.get('share_token') == token), None)
     if not user or not user.get('share_enabled'):
@@ -5335,7 +5345,7 @@ async def api_share_auth(token: str, req: ShareAuthRequest, request: Request):
 
 
 @app.get('/api/share/{token}/connections', tags=["Sharing"])
-async def api_share_connections(token: str, request: Request):
+def api_share_connections(token: str, request: Request):
     data = load_data()
     user = next((u for u in data['users'] if u.get('share_token') == token), None)
     if not user or not user.get('share_enabled'):
@@ -5357,7 +5367,7 @@ async def api_share_connections(token: str, request: Request):
 
 
 @app.post('/api/share/{token}/config/{connection_id}', tags=["Sharing"])
-async def api_share_config(token: str, connection_id: str, request: Request):
+def api_share_config(token: str, connection_id: str, request: Request):
     data = load_data()
     user = next((u for u in data['users'] if u.get('share_token') == token), None)
     if not user or not user.get('share_enabled'):
@@ -5389,7 +5399,7 @@ async def api_share_config(token: str, connection_id: str, request: Request):
 
 
 @app.post('/api/my/connections/{connection_id}/config', tags=["Self-service"])
-async def api_my_connection_config(request: Request, connection_id: str):
+def api_my_connection_config(request: Request, connection_id: str):
     user = get_current_user(request)
     if not user:
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
@@ -5420,7 +5430,7 @@ async def api_my_connection_config(request: Request, connection_id: str):
 
 
 @app.get('/settings', tags=["System Templates"])
-async def settings_page(request: Request):
+def settings_page(request: Request):
     user = _check_admin(request)
     if not user:
         return RedirectResponse('/login')
@@ -5436,7 +5446,7 @@ async def settings_page(request: Request):
 
 
 @app.get('/api/settings', tags=["Settings"])
-async def api_get_settings(request: Request):
+def api_get_settings(request: Request):
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
     data = load_data()
@@ -5444,7 +5454,7 @@ async def api_get_settings(request: Request):
 
 
 @app.get('/api/settings/tunnels/status', tags=["Settings"])
-async def api_tunnels_status(request: Request):
+def api_tunnels_status(request: Request):
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
     return {
@@ -5552,7 +5562,7 @@ async def api_warp_disconnect(request: Request):
 #     return {'status': 'success'}
 
 @app.post('/api/settings/save', tags=["Settings"])
-async def save_settings(request: Request, payload: SaveSettingsRequest):
+def save_settings(request: Request, payload: SaveSettingsRequest):
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
     data = load_data()
@@ -5647,7 +5657,7 @@ async def api_sync_delete(request: Request):
 
 
 @app.get('/api/servers/{server_id}/{protocol}/clients', tags=["Connections"])
-async def api_get_server_clients(request: Request, server_id: int, protocol: str):
+def api_get_server_clients(request: Request, server_id: int, protocol: str):
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
     try:
@@ -5679,7 +5689,7 @@ async def api_get_server_clients(request: Request, server_id: int, protocol: str
 
 
 @app.get('/api/settings/tokens', tags=["API Tokens"])
-async def api_list_tokens(request: Request):
+def api_list_tokens(request: Request):
     """List metadata for every API token. The raw token value is never
     returned by this endpoint — only its prefix and timestamps are visible
     after creation, by design."""
@@ -5764,7 +5774,7 @@ async def api_revoke_token(request: Request, token_id: str):
 
 
 @app.get('/api/settings/backup/download', tags=["Settings"])
-async def api_backup_download(request: Request):
+def api_backup_download(request: Request):
     if not _check_admin(request):
         return JSONResponse({'error': 'Forbidden'}, status_code=403)
     if not os.path.exists(DATA_FILE):

connection_service.py

@@ -166,7 +166,8 @@ class ConnectionService:
                 server = dict(data['servers'][server_id])
                 port = server.get('protocols', {}).get(protocol, {}).get('port', '55424')
 
-            ssh = self.get_ssh(server)
+            # get_ssh() blocks on ensure_connected — keep it off the event loop.
+            ssh = await asyncio.to_thread(self.get_ssh, server)
             remote_client_id = None
             manager = None
             try:
@@ -247,7 +248,8 @@ class ConnectionService:
                     raise SelfServiceError('Server not found', status_code=404)
                 server = dict(data['servers'][server_id])
 
-            ssh = self.get_ssh(server)
+            # get_ssh() blocks on ensure_connected — keep it off the event loop.
+            ssh = await asyncio.to_thread(self.get_ssh, server)
             try:
                 await asyncio.to_thread(ssh.connect)
                 manager = self.get_protocol_manager(ssh, protocol)

exit_link_service.py

@@ -119,11 +119,13 @@ class ExitLinkService:
             raise ExitLinkError('exit_not_found', 'Exit node not found or not installed', 404)
         return server, rec
 
-    def _awg(self, server):
-        return self.awg_manager_factory(self.get_ssh(server))
+    async def _awg(self, server):
+        # get_ssh() blocks on ensure_connected (TCP+SSH handshake, up to the
+        # connect timeout) — never run it inside the event loop.
+        return self.awg_manager_factory(await asyncio.to_thread(self.get_ssh, server))
 
-    def _exit_manager(self, server):
-        return self.exit_manager_factory(self.get_ssh(server))
+    async def _exit_manager(self, server):
+        return self.exit_manager_factory(await asyncio.to_thread(self.get_ssh, server))
 
     # ----- persistence under the lock, resolved by uid (indices may shift) -----
 
@@ -181,7 +183,7 @@ class ExitLinkService:
             exit_srv, _ = self._exit(data, link.get('exit_uid'))
             self._require_exit_dns(exit_srv)
         async with self._locks[(entry['uid'], protocol)]:
-            awg = self._awg(entry)
+            awg = await self._awg(entry)
             try:
                 log = await asyncio.to_thread(awg.exit_set_dns_via_exit, protocol, enabled)
             except Exception as err:
@@ -236,7 +238,7 @@ class ExitLinkService:
                 warnings.append('exit_dns_disabled_no_dns')
 
         async with self._locks[(entry['uid'], protocol)]:
-            awg = self._awg(entry)
+            awg = await self._awg(entry)
             entry_pub = await asyncio.to_thread(awg.exit_prepare_keys, protocol)
             try:
                 mtu = int(await asyncio.to_thread(awg._get_mtu, protocol) or 0)
@@ -262,7 +264,7 @@ class ExitLinkService:
                     await self._update_record(entry['uid'], protocol,
                                               lambda r: (r.get('exit_link') or {}).__setitem__('stale', reason))
 
-            exit_manager = self._exit_manager(exit_srv)
+            exit_manager = await self._exit_manager(exit_srv)
             name = f"{entry.get('name') or entry.get('host', '')} / {self.protocol_display_name(protocol)}"
             peer = await asyncio.to_thread(exit_manager.add_peer, peer_id, name, entry_pub)
             link = {
@@ -298,7 +300,8 @@ class ExitLinkService:
                 # New link confirmed: drop our peer on the previous exit, best effort.
                 try:
                     old_exit, _ = self._exit(data, previous['exit_uid'])
-                    await asyncio.to_thread(self._exit_manager(old_exit).remove_peer, None, peer_id)
+                    old_exit_manager = await self._exit_manager(old_exit)
+                    await asyncio.to_thread(old_exit_manager.remove_peer, None, peer_id)
                 except Exception as err:
                     logger.warning("previous exit %s unreachable, peer left behind: %s", previous['exit_uid'], err)
                     warnings.append('old_exit_unreachable_orphan_peer')
@@ -328,10 +331,12 @@ class ExitLinkService:
         warnings = []
         async with self._locks[(entry['uid'], protocol)]:
             # Entry first: direct egress must come back even when the exit is dead.
-            await asyncio.to_thread(self._awg(entry).exit_unlink, protocol)
+            awg = await self._awg(entry)
+            await asyncio.to_thread(awg.exit_unlink, protocol)
             try:
                 exit_srv, _ = self._exit(data, link.get('exit_uid'))
-                await asyncio.to_thread(self._exit_manager(exit_srv).remove_peer, None,
+                exit_manager = await self._exit_manager(exit_srv)
+                await asyncio.to_thread(exit_manager.remove_peer, None,
                                         peer_id_for(entry['uid'], protocol))
             except Exception as err:
                 logger.warning("exit %s unreachable on unlink, peer left behind: %s", link.get('exit_uid'), err)
@@ -399,14 +404,15 @@ class ExitLinkService:
         # The panel knows of no link, so a link file coming out of the archive
         # would silently route clients through an exit nobody is tracking.
         try:
-            info = await asyncio.to_thread(self._awg(entry).exit_link_info, protocol)
+            awg = await self._awg(entry)
+            info = await asyncio.to_thread(awg.exit_link_info, protocol)
         except Exception as err:
             logger.warning("could not read the link state after restore: %s", err)
             return {}
         if not info:
             return {}
         try:
-            await asyncio.to_thread(self._awg(entry).exit_unlink, protocol)
+            await asyncio.to_thread(awg.exit_unlink, protocol)
             return {'exit_link_restored': 'removed', 'exit_name': info.get('exit_name', '')}
         except Exception as err:
             logger.warning("could not remove the restored link file: %s", err)
@@ -422,7 +428,8 @@ class ExitLinkService:
         for item in self.linked_entries(data, exit_uid):
             entry = data['servers'][item['server_id']]
             try:
-                await asyncio.to_thread(self._awg(entry).exit_unlink, item['protocol'])
+                awg = await self._awg(entry)
+                await asyncio.to_thread(awg.exit_unlink, item['protocol'])
                 await self._update_record(item['server_uid'], item['protocol'], lambda r: r.pop('exit_link', None))
                 results.append({**item, 'status': 'success', 'error': None})
             except Exception as err:
@@ -442,7 +449,8 @@ class ExitLinkService:
                 continue
             try:
                 exit_srv, _ = self._exit(data, link['exit_uid'])
-                await asyncio.to_thread(self._exit_manager(exit_srv).remove_peer, None,
+                exit_manager = await self._exit_manager(exit_srv)
+                await asyncio.to_thread(exit_manager.remove_peer, None,
                                         peer_id_for(server['uid'], proto))
                 results.append({'protocol': proto, 'exit_uid': link['exit_uid'], 'status': 'success'})
             except Exception as err:
@@ -467,7 +475,7 @@ class ExitLinkService:
         link = rec.get('exit_link')
         if not link:
             raise ExitLinkError('not_linked', 'This instance is not linked to an exit node')
-        awg = self._awg(entry)
+        awg = await self._awg(entry)
         try:
             probe = await asyncio.to_thread(awg.exit_check_egress, protocol)
         except RuntimeError as err:
@@ -496,7 +504,7 @@ class ExitLinkService:
     async def status(self, entry_server_id, protocol):
         data = self.load_data()
         entry, rec = self._entry(data, entry_server_id, protocol)
-        awg = self._awg(entry)
+        awg = await self._awg(entry)
         link_status = await asyncio.to_thread(awg.exit_link_status, protocol)
         try:
             mtu = int(await asyncio.to_thread(awg._get_mtu, protocol) or 0)