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 = '10.0.0.1') -> str: return ( f"[Interface]\n" f"PrivateKey = {private_key}\n" f"Address = {peer_ip}/32\n" f"DNS = {dns}\n\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)