roof
0d32038150
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>
80 lines
2.8 KiB
Python
80 lines
2.8 KiB
Python
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"
|