"""First-run configuration wizard — interactive and non-interactive modes. Runs three-layer diagnostics and produces a router config: Layer 1: Local-only probes (CPU, RAM, disk) Layer 2: Passive network probes (STUN, DNS, RTT, bandwidth) Layer 3: Post-bootstrap (deferred to router runtime) """ from __future__ import annotations import json import logging import os from pathlib import Path from i2p_apps.setup.config_advisor import advise from i2p_apps.setup.local_probe import run_local_probes from i2p_apps.setup.network_probe import run_network_probes from i2p_apps.setup.stun_probe import detect_nat_type log = logging.getLogger(__name__) CONFIG_FILENAME = "router.config.json" class SetupWizard: """First-run configuration wizard.""" def __init__(self, data_dir: str = "~/.i2p-python"): self.data_dir = os.path.expanduser(data_dir) os.makedirs(self.data_dir, exist_ok=True) @property def config_path(self) -> str: return os.path.join(self.data_dir, CONFIG_FILENAME) def is_first_run(self) -> bool: """Check if router.config.json exists.""" return not os.path.exists(self.config_path) def save_config(self, config: dict) -> None: """Save config to data_dir/router.config.json.""" with open(self.config_path, "w") as f: json.dump(config, f, indent=2) log.info("Config saved to %s", self.config_path) def load_config(self) -> dict | None: """Load saved config, or None if first run.""" if not os.path.exists(self.config_path): return None with open(self.config_path) as f: return json.load(f) def run_auto(self, mode: str = "normal") -> dict: """Run non-interactive wizard. Modes: - normal: all probes including network - paranoid: local probes only, no network traffic - performance: all probes, maximize resource allocation """ log.info("Running setup wizard in %s mode", mode) # Layer 1: Local probes (always run) local = run_local_probes(data_dir=self.data_dir) log.info( "Local probes: %d crypto ops/s, %d MB RAM, %.0f MB/s disk", local.crypto_ops_per_sec, local.available_ram_bytes // (1024 * 1024), local.disk_write_mbps, ) nat = None network = None if mode != "paranoid": # Layer 2a: STUN NAT detection nat = detect_nat_type() log.info("NAT probe: type=%s, external=%s:%s", nat.nat_type, nat.external_ip, nat.external_port) # Layer 2b+2c: Network probes network = run_network_probes( include_bandwidth=(mode != "paranoid"), ) log.info("Network probe: DNS=%s, RTT=%.0fms", network.dns.can_resolve_neutral, network.latency.rtt_ms) # Generate recommendation recommendation = advise(local, nat=nat, network=network) config = recommendation.to_router_config_overrides() config["mode"] = mode config["version"] = 1 if recommendation.warnings: for w in recommendation.warnings: log.warning("Setup: %s", w) self.save_config(config) return config def run_interactive(self) -> dict: """Run interactive CLI wizard. Returns config dict. For non-TTY environments, falls back to run_auto("normal"). """ if not os.isatty(0): return self.run_auto("normal") print("\nWelcome to I2P Python Router\n") print("Checking your system...\n") # Layer 1 local = run_local_probes(data_dir=self.data_dir) print(f" CPU: {local.crypto_ops_per_sec:,} crypto ops/sec") print(f" RAM: {local.available_ram_bytes // (1024**3)} GB available") print(f" Disk: {local.disk_write_mbps:.0f} MB/s, " f"{local.disk_free_bytes // (1024**3)} GB free\n") # Ask about network tests print("Would you like to test your network connection?") print("This uses standard protocols (STUN, HTTPS) that look like") print("normal video call and web browsing traffic.\n") print("[Y]es [S]kip [P]aranoid (no network tests at all)") choice = input("> ").strip().lower() nat = None network = None if choice in ("y", "yes", ""): nat = detect_nat_type() network = run_network_probes() print(f"\n NAT: {nat.nat_type}") if nat.external_ip: print(f" External IP: {nat.external_ip}") print(f" DNS: {'OK' if network.dns.can_resolve_neutral else 'FAILED'}") print(f" Latency: {network.latency.rtt_ms:.0f}ms median") if network.bandwidth: print(f" Bandwidth: ~{network.bandwidth.download_mbps:.0f} Mbps") recommendation = advise(local, nat=nat, network=network) config = recommendation.to_router_config_overrides() print(f"\nRecommended configuration:") print(f" Bandwidth share: {config['share_percentage']}%") print(f" Transit tunnels: {config['inbound_tunnel_count']}") print(f" Floodfill: {'yes' if config['floodfill'] else 'no'}") print(f" Transport port: {config['listen_port']}") print(f" NAT traversal: {'UPnP enabled' if config['upnp_enabled'] else 'disabled'}") for w in recommendation.warnings: print(f" WARNING: {w}") print("\n[A]ccept [Q]uit") accept = input("> ").strip().lower() if accept in ("a", "accept", ""): config["version"] = 1 self.save_config(config) return config else: raise SystemExit("Setup cancelled.")