420dced9ff
- 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>
61 lines
2.2 KiB
Python
61 lines
2.2 KiB
Python
import os
|
|
import subprocess
|
|
import secrets
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
|
|
class WGInterface:
|
|
def __init__(self, config_path: str, iface_name: str):
|
|
self.config_path = config_path
|
|
self.iface_name = iface_name
|
|
self.up = False
|
|
|
|
def bring_up(self, timeout=30):
|
|
subprocess.run(['sudo', 'wg-quick', 'up', self.config_path],
|
|
check=True, timeout=timeout, capture_output=True, text=True)
|
|
self.up = True
|
|
|
|
def bring_down(self):
|
|
if self.up:
|
|
subprocess.run(['sudo', 'wg-quick', 'down', self.config_path],
|
|
check=False, timeout=15, capture_output=True)
|
|
self.up = False
|
|
|
|
def is_connected(self, server_ip='10.0.0.1', timeout=5) -> bool:
|
|
result = subprocess.run(
|
|
['ping', '-c', '1', '-W', str(timeout), server_ip],
|
|
capture_output=True, timeout=timeout + 2
|
|
)
|
|
return result.returncode == 0
|
|
|
|
|
|
def build_wg_config(private_key: str, peer_ip: str, server_pubkey: str,
|
|
server_endpoint: str, server_port: int = 51820,
|
|
allowed_ips: str = '10.0.0.0/24',
|
|
dns: str = None) -> str:
|
|
# Omit DNS line by default — wg-quick would try to call resolvconf/systemd-resolved
|
|
# to set system DNS, which is not installed in all test environments.
|
|
# DNS tests reach 10.0.0.1 directly via `dig @10.0.0.1` once the tunnel is up.
|
|
dns_line = f"DNS = {dns}\n" if dns else ""
|
|
return (
|
|
f"[Interface]\n"
|
|
f"PrivateKey = {private_key}\n"
|
|
f"Address = {peer_ip}/32\n"
|
|
f"{dns_line}\n"
|
|
f"[Peer]\n"
|
|
f"PublicKey = {server_pubkey}\n"
|
|
f"Endpoint = {server_endpoint}:{server_port}\n"
|
|
f"AllowedIPs = {allowed_ips}\n"
|
|
f"PersistentKeepalive = 25\n"
|
|
)
|
|
|
|
|
|
def cleanup_stale_e2e_interfaces():
|
|
"""Remove any leftover pic-e2e-* interfaces from previous failed runs."""
|
|
result = subprocess.run(['ip', 'link', 'show'], capture_output=True, text=True)
|
|
for line in result.stdout.splitlines():
|
|
if 'pic-e2e-' in line:
|
|
iface = line.split(':')[1].strip().split('@')[0]
|
|
subprocess.run(['sudo', 'ip', 'link', 'delete', iface], capture_output=True)
|