Implement Linux proxy controller, subscription state, and CLI/TUI

4 files changed, 688 insertions, 0 deletions

diff --git a/tests/cli_acceptance.rs b/tests/cli_acceptance.rs
new file mode 100644
index 0000000..97def0c
--- /dev/null
+++ b/tests/cli_acceptance.rs
@@ -0,0 +1,31 @@
+use std::process::Command;
+
+#[test]
+fn help_describes_the_controller_commands_without_contacting_services() {
+    let output = Command::new(binary()).arg("--help").output().unwrap();
+    assert!(output.status.success());
+    let text = String::from_utf8(output.stdout).unwrap();
+    for command in [
+        "subscription",
+        "configuration",
+        "enable",
+        "disable",
+        "status",
+        "recover",
+    ] {
+        assert!(text.contains(command), "Missing command {command}: {text}");
+    }
+}
+
+#[test]
+fn subscription_url_cannot_be_passed_on_the_command_line() {
+    let output = Command::new(binary())
+        .args(["subscription", "add", "https://secret.example/token"])
+        .output()
+        .unwrap();
+    assert!(!output.status.success());
+}
+
+fn binary() -> String {
+    std::env::var("ROENTGEN_TEST_BINARY").unwrap_or_else(|_| env!("CARGO_BIN_EXE_roentgen").into())
+}
diff --git a/tests/controller_acceptance.rs b/tests/controller_acceptance.rs
new file mode 100644
index 0000000..0d83cd1
--- /dev/null
+++ b/tests/controller_acceptance.rs
@@ -0,0 +1,355 @@
+use async_trait::async_trait;
+use roentgen::controller::{Caller, Controller, Operation};
+use roentgen::runtime::OfflineRuntime;
+use roentgen::subscription::SubscriptionSource;
+
+struct FixtureSource(String);
+impl FixtureSource {
+    fn new(value: &str) -> Self {
+        Self(value.into())
+    }
+}
+#[async_trait]
+impl SubscriptionSource for FixtureSource {
+    async fn fetch(&self, _: &str) -> anyhow::Result<Vec<u8>> {
+        Ok(self.0.as_bytes().to_vec())
+    }
+}
+use tempfile::TempDir;
+
+#[tokio::test]
+async fn importing_requires_explicit_selection_and_enabling() {
+    let home = TempDir::new().unwrap();
+    let mut controller = Controller::open(
+        home.path(),
+        Box::new(OfflineRuntime),
+        Box::new(FixtureSource::new(
+            "trojan://secret@example.com:443?security=tls&type=tcp#First\n",
+        )),
+    )
+    .unwrap();
+    let root = Caller::root();
+    controller
+        .execute(
+            &root,
+            Operation::AddSubscription {
+                name: "Personal".into(),
+                url: "https://provider.example/token".into(),
+            },
+        )
+        .await
+        .unwrap();
+    assert!(
+        controller
+            .status()
+            .unwrap()
+            .selected_configuration
+            .is_none()
+    );
+    assert!(!controller.status().unwrap().requested_enabled);
+    let entries = controller.configurations(None).unwrap();
+    assert_eq!(entries.len(), 1);
+    assert_eq!(entries[0].name, "First");
+    assert_eq!(entries[0].status, "supported");
+    assert!(controller.execute(&root, Operation::Enable).await.is_err());
+    assert!(!controller.status().unwrap().requested_enabled);
+}
+
+#[derive(Clone)]
+struct MutableSource(std::sync::Arc<std::sync::Mutex<Option<String>>>);
+#[async_trait]
+impl SubscriptionSource for MutableSource {
+    async fn fetch(&self, _: &str) -> anyhow::Result<Vec<u8>> {
+        self.0
+            .lock()
+            .unwrap()
+            .as_ref()
+            .map(|v| v.as_bytes().to_vec())
+            .ok_or_else(|| anyhow::anyhow!("Fixture fetch failed"))
+    }
+}
+
+struct GuardedRuntime {
+    armed: bool,
+    routing: &'static str,
+}
+#[async_trait]
+impl roentgen::runtime::Runtime for GuardedRuntime {
+    async fn arm(&mut self) -> anyhow::Result<()> {
+        self.armed = true;
+        self.routing = "blocked";
+        Ok(())
+    }
+    async fn apply(
+        &mut self,
+        connection: Option<&roentgen::model::Trojan>,
+        _: u64,
+    ) -> anyhow::Result<()> {
+        anyhow::ensure!(self.armed, "Attempted unprotected activation");
+        self.routing = if connection.is_some() {
+            "tun"
+        } else {
+            "blocked"
+        };
+        Ok(())
+    }
+    async fn disable(&mut self) -> anyhow::Result<()> {
+        self.armed = false;
+        self.routing = "direct";
+        Ok(())
+    }
+    fn routing(&self) -> &str {
+        self.routing
+    }
+}
+
+async fn fixture() -> (TempDir, Controller, MutableSource) {
+    let home = TempDir::new().unwrap();
+    let source = MutableSource(std::sync::Arc::new(std::sync::Mutex::new(Some(
+        "trojan://secret@example.com:443?security=tls#First\n".into(),
+    ))));
+    let mut controller = Controller::open(
+        home.path(),
+        Box::new(GuardedRuntime {
+            armed: false,
+            routing: "direct",
+        }),
+        Box::new(source.clone()),
+    )
+    .unwrap();
+    controller
+        .execute(
+            &Caller::root(),
+            Operation::AddSubscription {
+                name: "Personal".into(),
+                url: "https://provider.example/token".into(),
+            },
+        )
+        .await
+        .unwrap();
+    (home, controller, source)
+}
+
+#[tokio::test]
+async fn losing_a_selection_keeps_enabled_intent_and_blocks_without_replacement() {
+    let (_home, mut controller, source) = fixture().await;
+    let id = controller.configurations(None).unwrap()[0].id.clone();
+    controller
+        .execute(&Caller::root(), Operation::Select { id })
+        .await
+        .unwrap();
+    controller
+        .execute(&Caller::root(), Operation::Enable)
+        .await
+        .unwrap();
+    *source.0.lock().unwrap() = Some("trojan://another@other.example:443#Replacement\n".into());
+    controller
+        .execute(&Caller::root(), Operation::Refresh { name: None })
+        .await
+        .unwrap();
+    let status = controller.status().unwrap();
+    assert!(status.requested_enabled);
+    assert!(status.selected_configuration.is_none());
+    assert_eq!(status.routing, "blocked");
+}
+
+#[tokio::test]
+async fn failed_fetch_preserves_selection_but_an_empty_received_list_clears_it() {
+    let (_home, mut controller, source) = fixture().await;
+    let id = controller.configurations(None).unwrap()[0].id.clone();
+    controller
+        .execute(&Caller::root(), Operation::Select { id: id.clone() })
+        .await
+        .unwrap();
+    *source.0.lock().unwrap() = None;
+    assert!(
+        controller
+            .execute(&Caller::root(), Operation::Refresh { name: None })
+            .await
+            .is_err()
+    );
+    assert_eq!(
+        controller
+            .status()
+            .unwrap()
+            .selected_configuration
+            .as_deref(),
+        Some(id.as_str())
+    );
+    *source.0.lock().unwrap() = Some(String::new());
+    controller
+        .execute(&Caller::root(), Operation::Refresh { name: None })
+        .await
+        .unwrap();
+    assert!(
+        controller
+            .status()
+            .unwrap()
+            .selected_configuration
+            .is_none()
+    );
+    assert!(controller.configurations(None).unwrap().is_empty());
+}
+
+#[tokio::test]
+async fn mixed_base64_import_keeps_all_rows_and_prohibits_invalid_selection() {
+    use base64::Engine;
+    let (_home, mut controller, source) = fixture().await;
+    let body = "trojan://secret@example.com:443#Good\nvless://other@example.com:443#Unsupported\ntrojan://missing-port@host#Broken\n";
+    *source.0.lock().unwrap() = Some(base64::engine::general_purpose::STANDARD.encode(body));
+    controller
+        .execute(&Caller::root(), Operation::Refresh { name: None })
+        .await
+        .unwrap();
+    let entries = controller.configurations(None).unwrap();
+    assert_eq!(
+        entries
+            .iter()
+            .map(|e| e.status.as_str())
+            .collect::<Vec<_>>(),
+        ["supported", "unsupported", "invalid"]
+    );
+    for entry in entries.iter().skip(1) {
+        assert!(
+            controller
+                .execute(
+                    &Caller::root(),
+                    Operation::Select {
+                        id: entry.id.clone()
+                    }
+                )
+                .await
+                .is_err()
+        );
+    }
+    let public = serde_json::to_string(&entries).unwrap();
+    assert!(!public.contains("secret"));
+    assert!(!public.contains("missing-port"));
+}
+
+#[tokio::test]
+async fn denied_operations_leave_durable_state_unchanged() {
+    let (_home, mut controller, _source) = fixture().await;
+    let before = serde_json::to_string(&controller.status().unwrap()).unwrap();
+    assert!(
+        controller
+            .execute(
+                &Caller::denied(),
+                Operation::DeleteSubscription {
+                    name: "Personal".into()
+                }
+            )
+            .await
+            .is_err()
+    );
+    assert_eq!(
+        serde_json::to_string(&controller.status().unwrap()).unwrap(),
+        before
+    );
+    assert_eq!(controller.subscriptions().len(), 1);
+}
+
+#[tokio::test]
+async fn explicit_unset_and_deletion_never_disable_or_select_another_entry() {
+    let (_home, mut controller, _source) = fixture().await;
+    let id = controller.configurations(None).unwrap()[0].id.clone();
+    controller
+        .execute(&Caller::root(), Operation::Select { id: id.clone() })
+        .await
+        .unwrap();
+    controller
+        .execute(&Caller::root(), Operation::Enable)
+        .await
+        .unwrap();
+    controller
+        .execute(&Caller::root(), Operation::Unset)
+        .await
+        .unwrap();
+    assert!(controller.status().unwrap().requested_enabled);
+    assert_eq!(controller.status().unwrap().routing, "blocked");
+    assert_eq!(controller.configurations(None).unwrap().len(), 1);
+    controller
+        .execute(&Caller::root(), Operation::Select { id })
+        .await
+        .unwrap();
+    controller
+        .execute(
+            &Caller::root(),
+            Operation::DeleteSubscription {
+                name: "Personal".into(),
+            },
+        )
+        .await
+        .unwrap();
+    assert!(controller.status().unwrap().requested_enabled);
+    assert!(
+        controller
+            .status()
+            .unwrap()
+            .selected_configuration
+            .is_none()
+    );
+    assert_eq!(controller.status().unwrap().routing, "blocked");
+}
+
+#[tokio::test]
+async fn persisted_intent_restores_under_protection_after_controller_restart() {
+    let (home, mut controller, source) = fixture().await;
+    let id = controller.configurations(None).unwrap()[0].id.clone();
+    controller
+        .execute(&Caller::root(), Operation::Select { id: id.clone() })
+        .await
+        .unwrap();
+    controller
+        .execute(&Caller::root(), Operation::Enable)
+        .await
+        .unwrap();
+    drop(controller);
+    let mut restored = Controller::open(
+        home.path(),
+        Box::new(GuardedRuntime {
+            armed: false,
+            routing: "unknown",
+        }),
+        Box::new(source),
+    )
+    .unwrap();
+    restored.restore().await.unwrap();
+    assert_eq!(restored.status().unwrap().selected_configuration, Some(id));
+    assert!(restored.status().unwrap().requested_enabled);
+    assert_eq!(restored.status().unwrap().routing, "tun");
+}
+
+#[tokio::test]
+async fn label_changes_keep_identity_but_duplicate_candidates_clear_selection() {
+    let (_home, mut controller, source) = fixture().await;
+    let id = controller.configurations(None).unwrap()[0].id.clone();
+    controller
+        .execute(&Caller::root(), Operation::Select { id: id.clone() })
+        .await
+        .unwrap();
+    *source.0.lock().unwrap() = Some("trojan://secret@example.com:443#Renamed\n".into());
+    controller
+        .execute(&Caller::root(), Operation::Refresh { name: None })
+        .await
+        .unwrap();
+    assert_eq!(
+        controller.status().unwrap().selected_configuration,
+        Some(id)
+    );
+    *source.0.lock().unwrap() = Some(
+        "trojan://secret@example.com:443#Duplicate1\ntrojan://secret@example.com:443#Duplicate2\n"
+            .into(),
+    );
+    controller
+        .execute(&Caller::root(), Operation::Refresh { name: None })
+        .await
+        .unwrap();
+    assert!(
+        controller
+            .status()
+            .unwrap()
+            .selected_configuration
+            .is_none()
+    );
+}
diff --git a/tests/vm_controller_acceptance.py b/tests/vm_controller_acceptance.py
new file mode 100644
index 0000000..d39b16f
--- /dev/null
+++ b/tests/vm_controller_acceptance.py
@@ -0,0 +1,144 @@
+#!/usr/bin/env python3
+"""Exercise the actual CLI/system-bus/polkit boundary on the supplied ALT VM."""
+import contextlib
+import http.server
+import json
+import os
+from pathlib import Path
+import socket
+import subprocess
+import threading
+import time
+
+ROOT = Path('/var/tmp/roentgen-implementation')
+BIN = str(ROOT / 'roentgen')
+BODY = b'trojan://never-print-password@example.com:443?security=tls#First\n'
+
+
+def command(*args, input=None, ok=True, user=None, env=None):
+    argv = (["runuser", "-u", user, "--"] if user else []) + list(args)
+    result = subprocess.run(argv, input=input, text=True, capture_output=True, timeout=40, env=env)
+    if ok and result.returncode:
+        raise AssertionError(f'{args[0:2]} failed: {result.stderr}')
+    if not ok and not result.returncode:
+        raise AssertionError(f'{args[0:2]} unexpectedly succeeded')
+    return result
+
+
+def cli(*args, **kwargs): return command(BIN, *args, **kwargs)
+
+
+def status(): return json.loads(cli('status', '--json').stdout)
+
+
+class Handler(http.server.BaseHTTPRequestHandler):
+    def do_GET(self):
+        self.send_response(200)
+        self.send_header('Content-Type', 'text/plain')
+        self.end_headers()
+        self.wfile.write(BODY)
+    def log_message(self, *args): pass
+
+
+def main():
+    global BODY
+    assert os.geteuid() == 0 and socket.gethostname() == 'altctlrs'
+    assert (ROOT / 'allow-network-tests').exists()
+    # Test setup installs integration fixtures directly. This is deliberately not
+    # an assertion that product install succeeds with the VM's unsupported core.
+    Path('/etc/roentgen').mkdir(mode=0o700, exist_ok=True)
+    Path('/var/lib/roentgen').mkdir(mode=0o700, exist_ok=True)
+    assets = ROOT / 'assets' 
+    destinations = {
+        'roentgen.service': '/etc/systemd/system/roentgen.service',
+        'roentgen-guard.service': '/etc/systemd/system/roentgen-guard.service',
+        'roentgen-xray.service': '/etc/systemd/system/roentgen-xray.service',
+        'org.roentgen.Controller1.conf': '/etc/dbus-1/system.d/org.roentgen.Controller1.conf',
+        'org.roentgen.policy': '/usr/share/polkit-1/actions/org.roentgen.policy',
+    }
+    for source, destination in destinations.items():
+        Path(destination).write_bytes((assets / source).read_bytes())
+        Path(destination).chmod(0o644)
+    target = Path('/usr/local/bin/roentgen')
+    target.write_bytes(Path(BIN).read_bytes()); target.chmod(0o755)
+    command('systemctl', 'daemon-reload')
+    command('busctl', 'call', 'org.freedesktop.DBus', '/org/freedesktop/DBus', 'org.freedesktop.DBus', 'ReloadConfig')
+    command('systemctl', 'start', 'roentgen-guard.service')
+    command('systemctl', 'start', 'roentgen.service')
+    server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), Handler)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    token_url = f'http://127.0.0.1:{server.server_port}/never-print-subscription-token'
+    if 'Personal' in cli('subscription', 'list').stdout: cli('subscription', 'delete', 'Personal')
+    results = []
+    policy = Path('/etc/polkit-1/rules.d/00-roentgen-vm-test.rules')
+    try:
+        cli('subscription', 'add', '--name', 'Personal', input=token_url + '\n')
+        before = status()
+        assert before['selected_configuration'] is None and not before['requested_enabled']
+        public = cli('subscription', 'list').stdout + cli('configuration', 'list').stdout + cli('status', '--json').stdout
+        assert 'never-print-password' not in public and 'never-print-subscription-token' not in public
+        results.append('real HTTP import remains disabled/unselected and public CLI output omits credentials')
+        cli('enable', ok=False)
+        assert status()['generation'] == before['generation']
+        rows = command('busctl', 'call', 'org.roentgen.Controller1', '/org/roentgen/Controller1', 'org.roentgen.Controller1', 'Configurations', 's', '').stdout
+        # busctl prints a quoted JSON string. Decode that string before its JSON payload.
+        entries = json.loads(json.loads(rows[2:].strip()))
+        selected = entries[0]['id']
+        cli('configuration', 'select', selected)
+        before = status()
+        denied = cli('configuration', 'unset', ok=False, user='rustacean')
+        assert status()['selected_configuration'] == selected and status()['generation'] == before['generation']
+        assert 'authorization' in denied.stderr.lower() or 'accessdenied' in denied.stderr.lower()
+        results.append('noninteractive non-root mutation is denied with durable state unchanged')
+        policy.write_text('polkit.addRule(function(action, subject) { if (action.id == "org.roentgen.control" && subject.user == "rustacean") return polkit.Result.YES; });\n')
+        policy.chmod(0o644)
+        time.sleep(1)
+        cli('configuration', 'unset', user='rustacean')
+        assert status()['selected_configuration'] is None
+        results.append('polkit site allow policy authorizes the actual ordinary bus caller')
+        policy.unlink()
+        time.sleep(1)
+        cli('configuration', 'select', selected)
+        BODY = b'trojan://never-print-password@example.com:443#Renamed\nvless://id@example.com:443#Unsupported\ntrojan://broken@host#Invalid\n'
+        cli('subscription', 'update', 'Personal')
+        listing = cli('configuration', 'list').stdout
+        assert 'Renamed' in listing and 'unsupported' in listing and 'invalid' in listing
+        assert status()['selected_configuration'] == selected
+        results.append('mixed update retains ordered invalid/unsupported rows and normalized identity')
+        BODY = b'<html>provider login page</html>'
+        cli('subscription', 'update', 'Personal', ok=False)
+        assert status()['selected_configuration'] == selected
+        BODY = b''
+        cli('subscription', 'update', 'Personal')
+        assert status()['selected_configuration'] is None
+        results.append('whole-document failure preserves selection while received empty list clears it')
+        BODY = b'trojan://never-print-password@example.com:443#First\n'
+        cli('subscription', 'update', '--all')
+        output = command('busctl', 'call', 'org.roentgen.Controller1', '/org/roentgen/Controller1', 'org.roentgen.Controller1', 'Configurations', 's', '').stdout
+        selected = json.loads(json.loads(output[2:].strip()))[0]['id']
+        cli('configuration', 'select', selected)
+        error = cli('enable', ok=False).stderr
+        assert 'Unsupported Xray release' in error
+        assert not status()['requested_enabled']
+        results.append('unsupported packaged Xray activation is rejected before enabled intent changes')
+        command('systemctl', 'restart', 'roentgen.service')
+        assert status()['selected_configuration'] == selected
+        results.append('service restart retains explicit selection and disabled intent')
+        editor = ROOT / 'check-editor.py'
+        editor.write_text('import os,pathlib,sys\np=pathlib.Path(sys.argv[-1])\nassert p.suffix==".txt" and (p.stat().st_mode & 0o777)==0o600\nassert p.stat().st_uid==os.getuid()\nassert sys.argv[1:3]==["--literal","two words"]\npathlib.Path("/var/tmp/roentgen-editor-path").write_text(str(p))\n')
+        env = dict(os.environ, EDITOR=f'python3 {editor} --literal "two words"')
+        cli('logs', '--editor', user='rustacean', env=env)
+        snapshot = Path('/var/tmp/roentgen-editor-path').read_text()
+        assert not Path(snapshot).exists()
+        results.append('editor receives literal arguments and a user-owned private snapshot that is removed afterward')
+        cli('subscription', 'delete', 'Personal')
+        assert status()['selected_configuration'] is None
+        results.append('subscription deletion clears selection without replacement')
+        print(json.dumps({'passed': results}, indent=2))
+    finally:
+        server.shutdown()
+        if policy.exists(): policy.unlink()
+        command('systemctl', 'stop', 'roentgen.service', ok=True)
+
+
+if __name__ == '__main__': main()
diff --git a/tests/vm_guard_acceptance.py b/tests/vm_guard_acceptance.py
new file mode 100644
index 0000000..ff99c33
--- /dev/null
+++ b/tests/vm_guard_acceptance.py
@@ -0,0 +1,158 @@
+#!/usr/bin/env python3
+"""CLI acceptance checks. Run as root inside the authorized disposable ALT VM."""
+import contextlib
+import json
+import os
+from pathlib import Path
+import socket
+import subprocess
+import threading
+import time
+
+ROOT = Path('/var/tmp/roentgen-implementation')
+BINARY = ROOT / 'roentgen'
+
+
+def run(*args, **kwargs):
+    return subprocess.run(args, check=True, capture_output=True, text=True, timeout=20, **kwargs)
+
+
+def guard(enabled):
+    temporary = Path('/etc/roentgen/guard.toml.tmp')
+    temporary.write_text(f'version = 1\nenabled = {str(enabled).lower()}\n')
+    temporary.chmod(0o600)
+    temporary.replace('/etc/roentgen/guard.toml')
+    run(str(BINARY), 'guard')
+
+
+def echo(listener):
+    while True:
+        try:
+            connection, _ = listener.accept()
+        except OSError:
+            return
+        threading.Thread(target=stream, args=(connection,), daemon=True).start()
+
+
+def stream(connection):
+    with connection:
+        while data := connection.recv(1024):
+            connection.sendall(data)
+
+
+def exchange(connection):
+    connection.sendall(b'roentgen-acceptance')
+    assert connection.recv(1024) == b'roentgen-acceptance'
+
+
+def connect(host, port):
+    return socket.create_connection((host, port), timeout=1)
+
+
+def main():
+    assert os.geteuid() == 0
+    assert socket.gethostname() == 'altctlrs', 'This test is restricted to the supplied VM'
+    assert (ROOT / 'allow-network-tests').exists(), 'Missing VM-only authorization marker'
+    Path('/etc/roentgen').mkdir(mode=0o700, exist_ok=True)
+    run('id', 'roentgen-xray')
+    run('ip', 'netns', 'add', 'roentgen-peer')
+    children = []
+    listeners = []
+    results = []
+    try:
+        run('ip', 'link', 'add', 'rgtest0', 'type', 'veth', 'peer', 'name', 'rgpeer0', 'netns', 'roentgen-peer')
+        for address in ['198.51.100.1/24', '10.254.0.1/24', '2001:db8:726f::1/64', 'fded:726f::1/64']:
+            run('ip', 'address', 'add', address, 'dev', 'rgtest0', *(['nodad'] if ':' in address else []))
+        run('ip', 'link', 'set', 'rgtest0', 'up')
+        for address in ['198.51.100.2/24', '10.254.0.2/24', '2001:db8:726f::2/64', 'fded:726f::2/64']:
+            run('ip', 'netns', 'exec', 'roentgen-peer', 'ip', 'address', 'add', address, 'dev', 'rgpeer0', *(['nodad'] if ':' in address else []))
+        run('ip', 'netns', 'exec', 'roentgen-peer', 'ip', 'link', 'set', 'rgpeer0', 'up')
+        run('ip', 'netns', 'exec', 'roentgen-peer', 'ip', 'link', 'set', 'lo', 'up')
+        server = '''import socket,threading,time
+
+def serve(host,port):
+ s=socket.socket(socket.AF_INET6 if ':' in host else socket.AF_INET);s.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1);s.bind((host,port));s.listen()
+ def handle(c):
+  with c:
+   while True:
+    data=c.recv(1024)
+    if not data:return
+    c.sendall(data)
+ while True:
+  c,_=s.accept();threading.Thread(target=handle,args=(c,),daemon=True).start()
+for host in ['198.51.100.2','10.254.0.2','2001:db8:726f::2','fded:726f::2']:
+ for port in [53,18080]:threading.Thread(target=serve,args=(host,port),daemon=True).start()
+time.sleep(120)
+'''
+        children.append(subprocess.Popen(['ip', 'netns', 'exec', 'roentgen-peer', 'python3', '-u', '-c', server], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE))
+        time.sleep(2)
+        guard(False)
+        outgoing = connect('198.51.100.2', 18080)
+        exchange(outgoing)
+        listener = socket.socket()
+        listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        listener.bind(('198.51.100.1', 18081))
+        listener.listen()
+        listeners.append(listener)
+        threading.Thread(target=echo, args=(listener,), daemon=True).start()
+        client = '''import socket,sys
+s=socket.create_connection(('198.51.100.1',18081),timeout=3)
+print('ready',flush=True)
+for line in sys.stdin:
+ s.sendall(b'keep-incoming');assert s.recv(1024)==b'keep-incoming';print('ok',flush=True)
+'''
+        incoming = subprocess.Popen(['ip', 'netns', 'exec', 'roentgen-peer', 'python3', '-u', '-c', client], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+        children.append(incoming)
+        assert incoming.stdout.readline().strip() == 'ready'
+        incoming.stdin.write('before\n'); incoming.stdin.flush()
+        assert incoming.stdout.readline().strip() == 'ok'
+        guard(True)
+        for host in ['10.254.0.2', 'fded:726f::2']:
+            for port in [53,18080]:
+                try:
+                    with connect(host, port) as connection: exchange(connection)
+                except Exception as error:
+                    raise AssertionError(f'Private direct connection failed: {host}:{port}') from error
+        results.append('private IPv4/IPv6 destinations stay direct on ports 53 and 18080')
+        for host in ['198.51.100.2', '2001:db8:726f::2']:
+            for port in [53,18080]:
+                try:
+                    with connect(host, port) as connection: exchange(connection)
+                except (TimeoutError, OSError):
+                    pass
+                else:
+                    raise AssertionError(f'Public direct fallback leaked to {host}:{port}')
+        results.append('public IPv4/IPv6 destinations are blocked on ports 53 and 18080')
+        incoming.stdin.write('enabled\n'); incoming.stdin.flush()
+        assert incoming.stdout.readline().strip() == 'ok'
+        guard(False)
+        guard(True)
+        incoming.stdin.write('after-toggle\n'); incoming.stdin.flush()
+        assert incoming.stdout.readline().strip() == 'ok'
+        results.append('classified incoming TCP connection survives guard replacement and toggles')
+        try:
+            exchange(outgoing)
+        except (TimeoutError, OSError):
+            pass
+        else:
+            raise AssertionError('A preexisting direct outgoing connection bypassed the guard')
+        outgoing.close()
+        results.append('preexisting outgoing TCP connection receives no established-flow exemption')
+        guard(False)
+        with connect('198.51.100.2', 18080) as connection: exchange(connection)
+        results.append('explicit disabled guard restores ordinary direct routing')
+        print(json.dumps({'passed': results}, indent=2))
+    finally:
+        import sys
+        if sys.exc_info()[0]: print(run('nft', 'list', 'table', 'inet', 'roentgen').stdout, flush=True)
+        with contextlib.suppress(Exception): guard(False)
+        for child in children:
+            child.terminate()
+            with contextlib.suppress(Exception): child.wait(timeout=3)
+            if child.poll() is None: child.kill()
+        for listener in listeners: listener.close()
+        subprocess.run(['ip', 'link', 'del', 'rgtest0'], capture_output=True)
+        subprocess.run(['ip', 'netns', 'del', 'roentgen-peer'], capture_output=True)
+
+
+if __name__ == '__main__': main()