feat: add comprehensive E2E test suite (Playwright + WireGuard + API)

Adds tests/e2e/ with three layers of E2E coverage:
- API layer (tests/e2e/api/): unauthenticated access, admin endpoints,
  peer endpoints, access control enforcement — 24 tests
- Playwright UI (tests/e2e/ui/): login flows, admin navigation, peer
  dashboard/services, role-based ACL, password change — 60+ tests
- WireGuard connectivity (tests/e2e/wg/): tunnel up/down, DNS resolution
  through VPN, service ACL enforcement via iptables, full-tunnel routing
Shared helpers: PicAPIClient, WGInterface, playwright_login, cleanup.
Makefile targets: test-e2e-api, test-e2e-ui, test-e2e-wg, test-e2e.
Adds scripts/reset_admin_password.py for test bootstrap.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-04-25 16:41:13 -04:00
parent 1e81b3b618
commit 0d32038150
34 changed files with 2122 additions and 15 deletions
View File
+105
View File
@@ -0,0 +1,105 @@
import os
import pytest
import tempfile
import secrets
from helpers.wg_runner import WGInterface, build_wg_config, cleanup_stale_e2e_interfaces
@pytest.fixture(scope='session', autouse=True)
def cleanup_stale_wg_interfaces():
cleanup_stale_e2e_interfaces()
yield
cleanup_stale_e2e_interfaces()
@pytest.fixture(scope='session')
def wg_server_info(admin_client, pic_host):
"""Get server public key and endpoint from the running API."""
r = admin_client.get('/api/wireguard/status')
data = r.json()
# status might be nested — check common shapes
server_pubkey = (
data.get('public_key') or
data.get('server_public_key') or
data.get('status', {}).get('public_key', '')
)
port = data.get('port') or data.get('listen_port') or 51820
return {
'public_key': server_pubkey,
'endpoint': pic_host,
'port': int(port),
}
@pytest.fixture
def connected_peer(make_peer, wg_server_info, tmp_path):
"""
Creates a peer, builds its WireGuard config, brings the tunnel up, yields,
then tears everything down.
Requires: sudo wg-quick available on the test runner.
"""
peer = make_peer('e2etest-wg-basic', service_access=['calendar', 'files', 'mail', 'webdav'])
iface_name = f"pic-e2e-{secrets.token_hex(3)}"
conf_path = str(tmp_path / f"{iface_name}.conf")
config_text = build_wg_config(
private_key=peer['private_key'],
peer_ip=peer['ip'],
server_pubkey=wg_server_info['public_key'],
server_endpoint=wg_server_info['endpoint'],
server_port=wg_server_info['port'],
allowed_ips='10.0.0.0/24',
)
# Write config with restricted permissions
with open(conf_path, 'w') as f:
f.write(config_text)
os.chmod(conf_path, 0o600)
iface = WGInterface(conf_path, iface_name)
try:
iface.bring_up()
peer['iface'] = iface
peer['conf_path'] = conf_path
yield peer
finally:
iface.bring_down()
try:
os.unlink(conf_path)
except Exception:
pass
@pytest.fixture
def full_tunnel_peer(make_peer, wg_server_info, tmp_path):
"""Like connected_peer but with AllowedIPs=0.0.0.0/0 (full tunnel)."""
peer = make_peer('e2etest-wg-fulltunnel')
iface_name = f"pic-e2e-{secrets.token_hex(3)}"
conf_path = str(tmp_path / f"{iface_name}.conf")
config_text = build_wg_config(
private_key=peer['private_key'],
peer_ip=peer['ip'],
server_pubkey=wg_server_info['public_key'],
server_endpoint=wg_server_info['endpoint'],
server_port=wg_server_info['port'],
allowed_ips='0.0.0.0/0',
)
with open(conf_path, 'w') as f:
f.write(config_text)
os.chmod(conf_path, 0o600)
iface = WGInterface(conf_path, iface_name)
try:
iface.bring_up()
peer['iface'] = iface
peer['conf_path'] = conf_path
yield peer
finally:
iface.bring_down()
try:
os.unlink(conf_path)
except Exception:
pass
+79
View File
@@ -0,0 +1,79 @@
import pytest
import subprocess
import time
pytestmark = pytest.mark.wg
def test_restricted_peer_can_reach_allowed_service(make_peer, wg_server_info, tmp_path, admin_client):
"""Peer with service_access=['calendar'] can reach calendar VIP."""
from helpers.wg_runner import WGInterface, build_wg_config
import os
import secrets
peer = make_peer('e2etest-wg-restricted', service_access=['calendar'])
iface_name = f"pic-e2e-{secrets.token_hex(3)}"
conf_path = str(tmp_path / f"{iface_name}.conf")
config_text = build_wg_config(
private_key=peer['private_key'],
peer_ip=peer['ip'],
server_pubkey=wg_server_info['public_key'],
server_endpoint=wg_server_info['endpoint'],
server_port=wg_server_info['port'],
)
with open(conf_path, 'w') as f:
f.write(config_text)
os.chmod(conf_path, 0o600)
iface = WGInterface(conf_path, iface_name)
try:
iface.bring_up()
time.sleep(2)
# Get service VIPs
r = admin_client.get('/api/config')
sips = r.json().get('service_ips', {}) if r.status_code == 200 else {}
cal_vip = sips.get('vip_calendar', '')
files_vip = sips.get('vip_files', '')
if not cal_vip:
pytest.skip("service_ips not in config response — check /api/config shape")
# Calendar VIP should be reachable (TCP port 5232)
result = subprocess.run(
['nc', '-z', '-w', '3', cal_vip, '5232'],
capture_output=True, timeout=5
)
assert result.returncode == 0, f"Calendar VIP {cal_vip}:5232 should be reachable for restricted peer"
# Files VIP should be blocked
if files_vip:
result = subprocess.run(
['nc', '-z', '-w', '3', files_vip, '80'],
capture_output=True, timeout=5
)
assert result.returncode != 0, f"Files VIP should be blocked for calendar-only peer"
finally:
iface.bring_down()
try:
os.unlink(conf_path)
except Exception:
pass
def test_full_access_peer_can_reach_all_services(connected_peer, admin_client):
"""Peer with full service_access can reach all service VIPs."""
r = admin_client.get('/api/config')
sips = r.json().get('service_ips', {}) if r.status_code == 200 else {}
if not sips:
pytest.skip("service_ips not available in config")
for service, vip_key in [('calendar', 'vip_calendar'), ('files', 'vip_files')]:
vip = sips.get(vip_key, '')
if not vip:
continue
port = 5232 if service == 'calendar' else 80
result = subprocess.run(
['nc', '-z', '-w', '3', vip, str(port)],
capture_output=True, timeout=5
)
assert result.returncode == 0, f"{service} VIP {vip}:{port} should be reachable for full-access peer"
+28
View File
@@ -0,0 +1,28 @@
import pytest
import subprocess
pytestmark = pytest.mark.wg
def test_wg_connect_and_ping_server(connected_peer):
"""Scenario 25+26: create peer, connect, ping server VPN IP."""
iface = connected_peer['iface']
assert iface.up, "WireGuard interface should be up"
assert iface.is_connected('10.0.0.1'), "Server VPN IP 10.0.0.1 should be reachable via WireGuard"
def test_wg_peer_has_assigned_ip(connected_peer):
"""Verify the assigned peer IP is routed correctly."""
peer_ip = connected_peer['ip']
result = subprocess.run(['ip', 'addr', 'show'], capture_output=True, text=True)
assert peer_ip in result.stdout, f"Peer IP {peer_ip} should be assigned to the WG interface"
def test_wg_disconnect_removes_route(connected_peer):
"""Scenario 29: after disconnect, VPN IP is not reachable."""
iface = connected_peer['iface']
iface.bring_down()
result = subprocess.run(['ping', '-c', '1', '-W', '2', '10.0.0.1'],
capture_output=True, timeout=5)
# After disconnect, ping should fail
assert result.returncode != 0, "VPN IP should not be reachable after disconnect"
+29
View File
@@ -0,0 +1,29 @@
import pytest
import subprocess
pytestmark = pytest.mark.wg
def test_dns_resolves_via_vpn(connected_peer, admin_client):
"""Scenario 27: DNS queries for cell domain resolve via 10.0.0.1 (CoreDNS)."""
# Get the configured domain
r = admin_client.get('/api/config')
domain = r.json().get('domain', 'cell') if r.status_code == 200 else 'cell'
# Query CoreDNS at the server VPN IP
result = subprocess.run(
['dig', f'@10.0.0.1', f'mail.{domain}', '+short', '+time=5'],
capture_output=True, text=True, timeout=10
)
# CoreDNS should respond (not necessarily with an IP — just not SERVFAIL)
assert result.returncode == 0, f"DNS query failed: {result.stderr}"
def test_dns_server_reachable_via_vpn(connected_peer):
"""CoreDNS port 53 is reachable from within the VPN."""
result = subprocess.run(
['dig', '@10.0.0.1', 'health.check', '+time=2'],
capture_output=True, text=True, timeout=5
)
# Even a NXDOMAIN response means DNS is up
assert 'SERVFAIL' not in result.stdout or result.returncode == 0 or 'status:' in result.stdout
+31
View File
@@ -0,0 +1,31 @@
import pytest
import subprocess
pytestmark = [pytest.mark.wg, pytest.mark.requires_internet]
def test_full_tunnel_routes_all_traffic(full_tunnel_peer):
"""Scenario 30: with AllowedIPs=0.0.0.0/0, external traffic routes through VPN."""
# Check routing table — 0.0.0.0/0 should be via the WG interface
result = subprocess.run(['ip', 'route', 'show'], capture_output=True, text=True)
iface_name = full_tunnel_peer['iface'].iface_name
# In full tunnel mode, the default route or the 0.0.0.0/1 + 128.0.0.0/1 split routes
# point to the WG interface
assert (iface_name in result.stdout or
'0.0.0.0/1' in result.stdout or
'128.0.0.0/1' in result.stdout), "Full tunnel routes not found"
@pytest.mark.requires_internet
def test_full_tunnel_changes_apparent_ip(full_tunnel_peer, pic_host):
"""External IP check via a local echo service — skip if no internet."""
result = subprocess.run(
['curl', '-s', '--max-time', '5', 'https://ifconfig.me'],
capture_output=True, text=True, timeout=10
)
if result.returncode != 0:
pytest.skip("No internet access from test runner")
apparent_ip = result.stdout.strip()
# The apparent IP should NOT be the test runner's local IP
# (it should be pic0's external IP if full tunnel is working)
assert apparent_ip != '', "Could not determine apparent IP"