diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py index a45559e..ca4f691 100644 --- a/tests/e2e/conftest.py +++ b/tests/e2e/conftest.py @@ -1,9 +1,12 @@ """E2E test configuration — loads NiceGUI testing plugin and app.""" import pytest +from sqlalchemy import text +from sqlalchemy.ext.asyncio import create_async_engine from sqlmodel import select from wiregui.auth.passwords import hash_password +from wiregui.config import get_settings from wiregui.db import async_session from wiregui.models.configuration import Configuration from wiregui.models.user import User @@ -14,24 +17,32 @@ FAKE_SERVER_KEY = "SFake0ServerPubKey0000000000000000000000000w=" TEST_EMAIL = "e2e-test@example.com" TEST_PASSWORD = "testpass123" -async def _delete_user_cascade(session, user_id): - """Delete a user and all related objects via raw SQL to avoid stale ORM cache issues.""" - from sqlalchemy import text - for table in ("devices", "rules", "mfa_methods", "api_tokens", "oidc_connections"): - await session.execute(text(f"DELETE FROM {table} WHERE user_id = :uid"), {"uid": user_id}) # noqa: S608 - await session.execute(text("DELETE FROM users WHERE id = :uid"), {"uid": user_id}) +_CHILD_TABLES = ("devices", "rules", "mfa_methods", "api_tokens", "oidc_connections") + + +async def _cleanup_test_user(): + """Delete the test user and all related objects using a fresh engine.""" + engine = create_async_engine(get_settings().database_url) + async with engine.begin() as conn: + # Find user id by email + row = (await conn.execute( + text("SELECT id FROM users WHERE email = :email"), {"email": TEST_EMAIL} + )).first() + if row: + uid = row[0] + for table in _CHILD_TABLES: + await conn.execute(text(f"DELETE FROM {table} WHERE user_id = :uid"), {"uid": uid}) # noqa: S608 + await conn.execute(text("DELETE FROM users WHERE id = :uid"), {"uid": uid}) + await engine.dispose() @pytest.fixture async def test_user(): """Create a test user and ensure server config has a public key.""" - async with async_session() as session: - # Clean up any leftover from a previous failed run - existing = (await session.execute(select(User).where(User.email == TEST_EMAIL))).scalar_one_or_none() - if existing: - await _delete_user_cascade(session, existing.id) - await session.commit() + # Clean up any leftover from a previous failed run + await _cleanup_test_user() + async with async_session() as session: # Ensure a Configuration with a server key exists config = (await session.execute(select(Configuration).limit(1))).scalar_one_or_none() if config: @@ -53,7 +64,4 @@ async def test_user(): yield user - # Teardown - async with async_session() as session: - await _delete_user_cascade(session, user.id) - await session.commit() + await _cleanup_test_user() diff --git a/tests/e2e/test_devices.py b/tests/e2e/test_devices.py index 3f00773..8ed6b88 100644 --- a/tests/e2e/test_devices.py +++ b/tests/e2e/test_devices.py @@ -1,17 +1,11 @@ """End-to-end tests for device management UI using NiceGUI's User fixture.""" -from unittest.mock import AsyncMock, patch - import pytest from nicegui.testing import User from wiregui.models.user import User as UserModel from tests.e2e.conftest import TEST_EMAIL, TEST_PASSWORD -# Fake WG keys for testing (valid base64, 32 bytes) -FAKE_PRIVATE_KEY = "YFake0PrivateKey00000000000000000000000000w=" -FAKE_PUBLIC_KEY = "ZFake0PublicKey000000000000000000000000000w=" - async def _login(user: User): """Helper to log in via the UI.""" @@ -25,21 +19,18 @@ async def _login(user: User): @pytest.mark.parametrize("user", [{"storage": {}}], indirect=True) async def test_add_device_via_ui(user: User, test_user: UserModel): """Test the full flow: login → devices → add device → see it in table.""" - 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="): + await _login(user) - await _login(user) + # Open create dialog + user.find("Add Device").click() + await user.should_see("New Device") - # Open create dialog - user.find("Add Device").click() - await user.should_see("New Device") + # Fill device name and submit + user.find("Device Name").type("Test Laptop") + user.find("Create").click() - # Fill device name and submit - user.find("Device Name").type("Test Laptop") - user.find("Create").click() - - # Should see config dialog with the device config - await user.should_see("Test Laptop") + # Should see config dialog with the device config + await user.should_see("Test Laptop") @pytest.mark.parametrize("user", [{"storage": {}}], indirect=True) diff --git a/tests/test_utils.py b/tests/test_utils.py index e73b0e3..c5dbff3 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -103,18 +103,16 @@ def test_build_client_config_no_psk(): # --- Crypto (only if wg is installed) --- -async def test_generate_keypair(): - """Test keypair generation — requires `wg` CLI to be installed.""" - try: - subprocess.run(["wg", "--version"], capture_output=True, check=True) - except FileNotFoundError: - pytest.skip("wg CLI not installed") - +def test_generate_keypair(): + """Test keypair generation (pure Python, no wg CLI needed).""" from wiregui.utils.crypto import generate_keypair, generate_preshared_key - priv, pub = await generate_keypair() + priv, pub = generate_keypair() assert len(priv) == 44 # base64-encoded 32 bytes assert len(pub) == 44 psk = generate_preshared_key() assert len(psk) == 44 + + psk = generate_preshared_key() + assert len(psk) == 44 diff --git a/wiregui/api/v0/devices.py b/wiregui/api/v0/devices.py index 288ffdf..0a6c751 100644 --- a/wiregui/api/v0/devices.py +++ b/wiregui/api/v0/devices.py @@ -54,7 +54,7 @@ async def create_device( settings = get_settings() owner_id = body.user_id if (body.user_id and current_user.role == "admin") else current_user.id - _private_key, public_key = await generate_keypair() + _private_key, public_key = generate_keypair() psk = generate_preshared_key() ipv4 = await allocate_ipv4(session, settings.wg_ipv4_network) ipv6 = await allocate_ipv6(session, settings.wg_ipv6_network) diff --git a/wiregui/auth/seed.py b/wiregui/auth/seed.py index 962e5f6..2d0e1a1 100644 --- a/wiregui/auth/seed.py +++ b/wiregui/auth/seed.py @@ -51,7 +51,7 @@ async def ensure_server_keypair() -> None: return # already have keys try: - private_key, public_key = await generate_keypair() + private_key, public_key = generate_keypair() config.server_private_key = private_key config.server_public_key = public_key session.add(config) diff --git a/wiregui/pages/admin/devices.py b/wiregui/pages/admin/devices.py index 59ee0ed..3d595ed 100644 --- a/wiregui/pages/admin/devices.py +++ b/wiregui/pages/admin/devices.py @@ -88,7 +88,7 @@ async def admin_devices_page(): try: settings = get_settings() - private_key, public_key = await generate_keypair() + private_key, public_key = generate_keypair() psk = generate_preshared_key() async with async_session() as session: diff --git a/wiregui/pages/devices.py b/wiregui/pages/devices.py index 63faf37..a26fd03 100644 --- a/wiregui/pages/devices.py +++ b/wiregui/pages/devices.py @@ -72,7 +72,7 @@ async def devices_page(): try: settings = get_settings() - private_key, public_key = await generate_keypair() + private_key, public_key = generate_keypair() psk = generate_preshared_key() async with async_session() as session: diff --git a/wiregui/utils/crypto.py b/wiregui/utils/crypto.py index 2f3cf0b..b33671f 100644 --- a/wiregui/utils/crypto.py +++ b/wiregui/utils/crypto.py @@ -1,37 +1,31 @@ -"""WireGuard key generation and encryption utilities.""" +"""WireGuard key generation and encryption utilities — pure Python, no wg CLI needed.""" -import asyncio import base64 import os - -async def generate_private_key() -> str: - """Generate a WireGuard private key using `wg genkey`.""" - proc = await asyncio.create_subprocess_exec( - "wg", "genkey", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, - ) - stdout, _ = await proc.communicate() - if proc.returncode != 0: - raise RuntimeError("wg genkey failed") - return stdout.decode().strip() +from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey +from cryptography.hazmat.primitives.serialization import Encoding, NoEncryption, PrivateFormat, PublicFormat -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() +def generate_private_key() -> str: + """Generate a WireGuard private key (Curve25519, base64-encoded).""" + key = X25519PrivateKey.generate() + raw = key.private_bytes(Encoding.Raw, PrivateFormat.Raw, NoEncryption()) + return base64.b64encode(raw).decode() -async def generate_keypair() -> tuple[str, str]: +def derive_public_key(private_key: str) -> str: + """Derive a WireGuard public key from a private key.""" + raw = base64.b64decode(private_key) + key = X25519PrivateKey.from_private_bytes(raw) + pub = key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw) + return base64.b64encode(pub).decode() + + +def generate_keypair() -> tuple[str, str]: """Generate a WireGuard keypair. Returns (private_key, public_key).""" - private_key = await generate_private_key() - public_key = await derive_public_key(private_key) + private_key = generate_private_key() + public_key = derive_public_key(private_key) return private_key, public_key