fix: make keypair generation async to avoid blocking the event loop
All checks were successful
CI / test (push) Successful in 1m58s
CI / release (push) Successful in 35s
CI / docker (push) Successful in 35s

generate_keypair() used synchronous subprocess.run() which blocked
the NiceGUI event loop during wg genkey/pubkey calls. This caused
WebSocket disconnects, page reloads, and the config dialog never
appearing after device creation.

Switched to asyncio.create_subprocess_exec so the event loop stays
responsive while waiting for the wg CLI.
This commit is contained in:
Stefano Bertelli 2026-03-30 22:57:00 -05:00
parent e51c53f247
commit 92554d4089
7 changed files with 31 additions and 22 deletions

View file

@ -1,6 +1,6 @@
"""End-to-end tests for device management UI using NiceGUI's User fixture.""" """End-to-end tests for device management UI using NiceGUI's User fixture."""
from unittest.mock import patch from unittest.mock import AsyncMock, patch
import pytest import pytest
from nicegui.testing import User from nicegui.testing import User
@ -25,7 +25,7 @@ async def _login(user: User):
@pytest.mark.parametrize("user", [{"storage": {}}], indirect=True) @pytest.mark.parametrize("user", [{"storage": {}}], indirect=True)
async def test_add_device_via_ui(user: User, test_user: UserModel): async def test_add_device_via_ui(user: User, test_user: UserModel):
"""Test the full flow: login → devices → add device → see it in table.""" """Test the full flow: login → devices → add device → see it in table."""
with patch("wiregui.pages.devices.generate_keypair", return_value=(FAKE_PRIVATE_KEY, FAKE_PUBLIC_KEY)), \ with patch("wiregui.pages.devices.generate_keypair", new_callable=AsyncMock, return_value=(FAKE_PRIVATE_KEY, FAKE_PUBLIC_KEY)), \
patch("wiregui.pages.devices.generate_preshared_key", return_value="cHJlc2hhcmVkMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDA="): patch("wiregui.pages.devices.generate_preshared_key", return_value="cHJlc2hhcmVkMDAwMDAwMDAwMDAwMDAwMDAwMDAwMDA="):
await _login(user) await _login(user)

View file

@ -103,7 +103,7 @@ def test_build_client_config_no_psk():
# --- Crypto (only if wg is installed) --- # --- Crypto (only if wg is installed) ---
def test_generate_keypair(): async def test_generate_keypair():
"""Test keypair generation — requires `wg` CLI to be installed.""" """Test keypair generation — requires `wg` CLI to be installed."""
try: try:
subprocess.run(["wg", "--version"], capture_output=True, check=True) subprocess.run(["wg", "--version"], capture_output=True, check=True)
@ -112,7 +112,7 @@ def test_generate_keypair():
from wiregui.utils.crypto import generate_keypair, generate_preshared_key from wiregui.utils.crypto import generate_keypair, generate_preshared_key
priv, pub = generate_keypair() priv, pub = await generate_keypair()
assert len(priv) == 44 # base64-encoded 32 bytes assert len(priv) == 44 # base64-encoded 32 bytes
assert len(pub) == 44 assert len(pub) == 44

View file

@ -54,7 +54,7 @@ async def create_device(
settings = get_settings() settings = get_settings()
owner_id = body.user_id if (body.user_id and current_user.role == "admin") else current_user.id owner_id = body.user_id if (body.user_id and current_user.role == "admin") else current_user.id
_private_key, public_key = generate_keypair() _private_key, public_key = await generate_keypair()
psk = generate_preshared_key() psk = generate_preshared_key()
ipv4 = await allocate_ipv4(session, settings.wg_ipv4_network) ipv4 = await allocate_ipv4(session, settings.wg_ipv4_network)
ipv6 = await allocate_ipv6(session, settings.wg_ipv6_network) ipv6 = await allocate_ipv6(session, settings.wg_ipv6_network)

View file

@ -51,7 +51,7 @@ async def ensure_server_keypair() -> None:
return # already have keys return # already have keys
try: try:
private_key, public_key = generate_keypair() private_key, public_key = await generate_keypair()
config.server_private_key = private_key config.server_private_key = private_key
config.server_public_key = public_key config.server_public_key = public_key
session.add(config) session.add(config)

View file

@ -88,7 +88,7 @@ async def admin_devices_page():
try: try:
settings = get_settings() settings = get_settings()
private_key, public_key = generate_keypair() private_key, public_key = await generate_keypair()
psk = generate_preshared_key() psk = generate_preshared_key()
async with async_session() as session: async with async_session() as session:

View file

@ -72,7 +72,7 @@ async def devices_page():
try: try:
settings = get_settings() settings = get_settings()
private_key, public_key = generate_keypair() private_key, public_key = await generate_keypair()
psk = generate_preshared_key() psk = generate_preshared_key()
async with async_session() as session: async with async_session() as session:

View file

@ -1,28 +1,37 @@
"""WireGuard key generation and encryption utilities.""" """WireGuard key generation and encryption utilities."""
import asyncio
import base64 import base64
import os import os
import subprocess
def generate_private_key() -> str: async def generate_private_key() -> str:
"""Generate a WireGuard private key using `wg genkey`.""" """Generate a WireGuard private key using `wg genkey`."""
result = subprocess.run(["wg", "genkey"], capture_output=True, text=True, check=True) proc = await asyncio.create_subprocess_exec(
return result.stdout.strip() "wg", "genkey", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
def derive_public_key(private_key: str) -> str:
"""Derive a WireGuard public key from a private key using `wg pubkey`."""
result = subprocess.run(
["wg", "pubkey"], input=private_key, capture_output=True, text=True, check=True
) )
return result.stdout.strip() stdout, _ = await proc.communicate()
if proc.returncode != 0:
raise RuntimeError("wg genkey failed")
return stdout.decode().strip()
def generate_keypair() -> tuple[str, str]: async def derive_public_key(private_key: str) -> str:
"""Derive a WireGuard public key from a private key using `wg pubkey`."""
proc = await asyncio.create_subprocess_exec(
"wg", "pubkey", stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
)
stdout, _ = await proc.communicate(input=private_key.encode())
if proc.returncode != 0:
raise RuntimeError("wg pubkey failed")
return stdout.decode().strip()
async def generate_keypair() -> tuple[str, str]:
"""Generate a WireGuard keypair. Returns (private_key, public_key).""" """Generate a WireGuard keypair. Returns (private_key, public_key)."""
private_key = generate_private_key() private_key = await generate_private_key()
public_key = derive_public_key(private_key) public_key = await derive_public_key(private_key)
return private_key, public_key return private_key, public_key