Files
pic/tests/e2e/wg/test_wg_acl.py
roof 420dced9ff fix: WireGuard peer sync, privileged mode, E2E and integration test correctness
- api/app.py: sync WireGuard server config on peer add/remove (non-fatal)
- docker-compose.yml: add privileged:true to wireguard service
- E2E tests: fix logout selector, DNS IP lookup, wg config DNS line, VIP skip guards,
  badge text selectors, heading .first, async logout wait
- Integration tests: fix 4 tests that sent unauthenticated requests expecting 400
  (now use authenticated session helpers); accept 401 as valid in webui proxy test;
  add password field to service_access validation test
- Remove stale tracked config templates (config/api/api/*, config/api/cell.env, etc.)
  that no longer exist on disk after config layout was reorganised

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-26 06:04:40 -04:00

102 lines
3.6 KiB
Python

import pytest
import subprocess
import time
pytestmark = pytest.mark.wg
def _vip_reachable(ip: str, port: int, timeout: int = 2) -> bool:
result = subprocess.run(
['nc', '-z', '-w', str(timeout), ip, str(port)],
capture_output=True, timeout=timeout + 1
)
return result.returncode == 0
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 if VIPs are live."""
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)
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 if VIP actually has a service behind it before asserting
if not _vip_reachable(cal_vip, 5232):
pytest.skip(
f"Calendar VIP {cal_vip}:5232 not reachable — "
"requires routing infrastructure (DNAT/VIP not configured in this environment)"
)
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"
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 if VIPs are live."""
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")
any_vip_reachable = False
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
if not _vip_reachable(vip, port):
continue
any_vip_reachable = True
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"
if not any_vip_reachable:
pytest.skip(
"No service VIPs reachable — requires routing infrastructure "
"(DNAT/VIP rules not configured in this environment)"
)