From d8321749820f6f14bc51c1777a5cf6fedc38ab40 Mon Sep 17 00:00:00 2001 From: jwinterm Date: Tue, 7 Jul 2026 20:44:20 -0400 Subject: [PATCH] Add AMPB-owned transport controls --- README.md | 3 + ampbrowser/cli.py | 44 +++++- ampbrowser/launch.py | 5 +- ampbrowser/transport_manager.py | 262 +++++++++++++++++++++++++++++++- docs/transport-management.md | 5 + tests/test_launch.py | 1 + tests/test_transport_manager.py | 98 ++++++++++-- 7 files changed, 405 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 8170461..aa28354 100644 --- a/README.md +++ b/README.md @@ -40,6 +40,9 @@ python3 -m ampbrowser inspect python3 -m ampbrowser plan gemini://wownero.org/ python3 -m ampbrowser open https://wownero.org/ python3 -m ampbrowser open http://example.b32.i2p/ --yes +python3 -m ampbrowser transport start tor +python3 -m ampbrowser transport status tor +python3 -m ampbrowser transport stop tor python3 -m ampbrowser open http://example.b32.i2p/ --config examples/config.toml python3 -m ampbrowser open http://example.b32.i2p/ --platform android python3 -m ampbrowser fixture check ../adaptive-multi-protocol-gateway/dist/wownero/ampg-fixture-manifest.json diff --git a/ampbrowser/cli.py b/ampbrowser/cli.py index 0285c55..89ae75d 100644 --- a/ampbrowser/cli.py +++ b/ampbrowser/cli.py @@ -13,7 +13,7 @@ from .launch import execute_open, prepare_open from .plan import plan_url from .platforms import PLATFORM_CHOICES from .routing import route_url -from .transport_manager import ensure_transport_ready +from .transport_manager import ensure_transport_ready, stop_managed_transport, transport_status from .transports import inspect_transports @@ -53,6 +53,18 @@ def main(argv: list[str] | None = None) -> int: subcommands.add_parser("inspect", help="Inspect local transport readiness.") + transport_parser = subcommands.add_parser("transport", help="Manage AMPB-owned transports.") + transport_subcommands = transport_parser.add_subparsers(dest="transport_command", required=True) + transport_status_parser = transport_subcommands.add_parser("status", help="Show AMPB transport ownership state.") + transport_status_parser.add_argument("transport") + transport_status_parser.add_argument("--config", type=Path, help="Path to an AMPB config file.") + transport_start_parser = transport_subcommands.add_parser("start", help="Start an AMPB-owned transport.") + transport_start_parser.add_argument("transport") + transport_start_parser.add_argument("--config", type=Path, help="Path to an AMPB config file.") + transport_stop_parser = transport_subcommands.add_parser("stop", help="Stop an AMPB-owned transport.") + transport_stop_parser.add_argument("transport") + transport_stop_parser.add_argument("--config", type=Path, help="Path to an AMPB config file.") + fixture_parser = subcommands.add_parser("fixture", help="Check AMPG fixture manifests.") fixture_subcommands = fixture_parser.add_subparsers(dest="fixture_command", required=True) fixture_check = fixture_subcommands.add_parser("check", help="Validate route expectations.") @@ -80,6 +92,8 @@ def main(argv: list[str] | None = None) -> int: ) if args.command == "inspect": return _cmd_inspect() + if args.command == "transport": + return _cmd_transport(args) if args.command == "fixture": return _cmd_fixture(args) if args.command == "docs": @@ -177,6 +191,7 @@ def _cmd_open( f"policy={browse_plan.policy_mode} " f"requires_consent={str(browse_plan.requires_consent).lower()} " f"consent_granted={str(open_plan.consent_granted).lower()} " + f"browser_pid={open_plan.browser_pid} " f"profile_path={open_plan.profile_path} " f"proxy={open_plan.proxy} " f"runtime_path={runtime_path} " @@ -206,6 +221,33 @@ def _cmd_inspect() -> int: return 0 +def _cmd_transport(args) -> int: + config = load_config(Path.cwd(), args.config) + if args.transport_command == "status": + result = transport_status(args.transport, config=config, root=Path.cwd()) + elif args.transport_command == "start": + result = ensure_transport_ready(args.transport, config=config, root=Path.cwd()) + elif args.transport_command == "stop": + result = stop_managed_transport(args.transport, config=config, root=Path.cwd()) + else: + return 1 + command = shlex.join(result.command) if result.command else "-" + print( + "AMPBROWSER_TRANSPORT " + f"transport={result.transport} " + f"provider={result.provider} " + f"status={result.status} " + f"ready={str(result.ready).lower()} " + f"owned={str(result.owned).lower()} " + f"pid={result.pid} " + f"endpoint={result.endpoint} " + f"state_dir={result.state_dir} " + f"command=\"{_safe(command)}\" " + f"message=\"{_safe(result.message)}\"" + ) + return 0 + + def _cmd_docs(args) -> int: if args.docs_command == "generate": changed = generate_docs(Path.cwd(), check=args.check) diff --git a/ampbrowser/launch.py b/ampbrowser/launch.py index e1f3063..f288a16 100644 --- a/ampbrowser/launch.py +++ b/ampbrowser/launch.py @@ -33,6 +33,7 @@ class OpenPlan: setup_steps: tuple[str, ...] message: str launch_spec: BrowserLaunchSpec | None = None + browser_pid: int = 0 transport_setup_status: str = "-" transport_setup_provider: str = "-" transport_setup_message: str = "-" @@ -131,8 +132,8 @@ def execute_open(open_plan: OpenPlan, *, root: Path | None = None) -> OpenPlan: ) _write_profile(root, spec) - subprocess.Popen(spec.command, cwd=str(root)) # noqa: S603 - return replace(open_plan, dry_run=False, status="launched", message="launched bundled browser") + process = subprocess.Popen(spec.command, cwd=str(root)) # noqa: S603 + return replace(open_plan, dry_run=False, status="launched", browser_pid=process.pid, message="launched bundled browser") def _proxy_for(browse_plan: BrowsePlan) -> str: diff --git a/ampbrowser/transport_manager.py b/ampbrowser/transport_manager.py index 9972715..adb3b21 100644 --- a/ampbrowser/transport_manager.py +++ b/ampbrowser/transport_manager.py @@ -4,6 +4,7 @@ from dataclasses import dataclass import json from pathlib import Path import os +import signal import shutil import socket import subprocess @@ -15,6 +16,7 @@ from .config import AppConfig ARTI_RELATIVE_PATH = Path("providers/arti/bin/arti") +OWNED_STATE_FILE = "ampb-owned.json" @dataclass(frozen=True) @@ -52,6 +54,10 @@ def ensure_transport_ready( if adapter is None: return ManagedTransportResult(transport, "-", "unsupported", "-", False, 0, "-", (), "transport adapter not found") + owned = _owned_transport_result(transport, config=config, root=root, wait_seconds=wait_seconds) + if owned: + return owned + status = status or adapter.inspect() if status.adoptable: return ManagedTransportResult( @@ -82,6 +88,134 @@ def ensure_transport_ready( return _start_tor(config=config, root=root, endpoint=status.endpoint, wait_seconds=wait_seconds) +def transport_status( + transport: str, + *, + config: AppConfig, + root: Path, +) -> ManagedTransportResult: + adapter = adapter_for(transport) + if adapter is None: + return ManagedTransportResult(transport, "-", "unsupported", "-", False, 0, "-", (), "transport adapter not found") + + owned = _owned_transport_result(transport, config=config, root=root, wait_seconds=0.1) + if owned: + return owned + + status = adapter.inspect() + if status.adoptable: + return ManagedTransportResult( + transport, + "-", + "ready", + status.endpoint, + False, + 0, + _managed_state_dir(root, config, transport), + (), + f"{transport} transport is running but not AMPB-owned", + ) + + return ManagedTransportResult( + transport, + "-", + "stopped", + status.endpoint, + False, + 0, + _managed_state_dir(root, config, transport), + (), + f"{transport} transport is not running", + ) + + +def stop_managed_transport( + transport: str, + *, + config: AppConfig, + root: Path, + wait_seconds: float = 5.0, +) -> ManagedTransportResult: + state_path = _owned_state_path(root, config, transport) + state = _read_owned_state(state_path) + endpoint = str(state.get("endpoint") or _endpoint_for(transport)) + state_dir = _managed_state_dir(root, config, transport) + provider = str(state.get("provider") or "-") + command = tuple(str(part) for part in state.get("command", ())) + pid = _state_pid(state) + + if not state or not pid: + return ManagedTransportResult( + transport, + "-", + "not-owned", + endpoint, + False, + 0, + state_dir, + (), + f"no AMPB-owned {transport} process is recorded", + ) + + if not _pid_alive(pid): + _unlink_state(state_path) + return ManagedTransportResult( + transport, + provider, + "stale", + endpoint, + True, + pid, + state_dir, + command, + f"removed stale AMPB-owned {transport} state", + ) + + try: + os.kill(pid, signal.SIGTERM) + except OSError as exc: + return ManagedTransportResult( + transport, + provider, + "stop-failed", + endpoint, + True, + pid, + state_dir, + command, + f"could not stop AMPB-owned {transport}: {exc}", + ) + + deadline = time.monotonic() + wait_seconds + while time.monotonic() <= deadline: + if not _pid_alive(pid): + _unlink_state(state_path) + return ManagedTransportResult( + transport, + provider, + "stopped", + endpoint, + True, + pid, + state_dir, + command, + f"stopped AMPB-owned {transport}", + ) + time.sleep(0.1) + + return ManagedTransportResult( + transport, + provider, + "stop-timeout", + endpoint, + True, + pid, + state_dir, + command, + f"AMPB-owned {transport} did not stop before timeout", + ) + + def _start_tor( *, config: AppConfig, @@ -126,7 +260,7 @@ def _start_tor( ) if _wait_for_endpoint(endpoint, timeout_seconds=wait_seconds): - return ManagedTransportResult( + result = ManagedTransportResult( "tor", provider.kind, "started", @@ -137,6 +271,8 @@ def _start_tor( command, f"started managed {provider.kind} transport", ) + _write_owned_state(root, config, result) + return result process.terminate() return ManagedTransportResult( @@ -228,6 +364,130 @@ def _managed_state_dir(root: Path, config: AppConfig, transport: str) -> str: return str(root / config.state_dir / "transports" / transport) +def _owned_state_path(root: Path, config: AppConfig, transport: str) -> Path: + return Path(_managed_state_dir(root, config, transport)) / OWNED_STATE_FILE + + +def _write_owned_state(root: Path, config: AppConfig, result: ManagedTransportResult) -> None: + state_path = _owned_state_path(root, config, result.transport) + state_path.parent.mkdir(parents=True, exist_ok=True) + _chmod_private(state_path.parent) + data = { + "transport": result.transport, + "provider": result.provider, + "endpoint": result.endpoint, + "pid": result.pid, + "state_dir": result.state_dir, + "command": list(result.command), + "started_at": int(time.time()), + } + state_path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8") + _chmod_private(state_path) + + +def _owned_transport_result( + transport: str, + *, + config: AppConfig, + root: Path, + wait_seconds: float, +) -> ManagedTransportResult | None: + state_path = _owned_state_path(root, config, transport) + state = _read_owned_state(state_path) + if not state: + return None + + pid = _state_pid(state) + provider = str(state.get("provider") or "-") + endpoint = str(state.get("endpoint") or _endpoint_for(transport)) + state_dir = str(state.get("state_dir") or _managed_state_dir(root, config, transport)) + command = tuple(str(part) for part in state.get("command", ())) + + if not pid or not _pid_alive(pid): + _unlink_state(state_path) + return ManagedTransportResult( + transport, + provider, + "stale", + endpoint, + True, + pid, + state_dir, + command, + f"removed stale AMPB-owned {transport} state", + ) + + if _wait_for_endpoint(endpoint, timeout_seconds=wait_seconds): + return ManagedTransportResult( + transport, + provider, + "ready", + endpoint, + True, + pid, + state_dir, + command, + f"using running AMPB-owned {transport}", + ) + + return ManagedTransportResult( + transport, + provider, + "running", + endpoint, + True, + pid, + state_dir, + command, + f"AMPB-owned {transport} process is running but endpoint is not ready", + ) + + +def _read_owned_state(path: Path) -> dict[str, object]: + if not path.exists(): + return {} + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {} + if not isinstance(data, dict): + return {} + return data + + +def _state_pid(state: dict[str, object]) -> int: + value = state.get("pid") + if isinstance(value, int): + return value + if isinstance(value, str) and value.isdigit(): + return int(value) + return 0 + + +def _pid_alive(pid: int) -> bool: + try: + os.kill(pid, 0) + return True + except OSError: + return False + + +def _unlink_state(path: Path) -> None: + try: + path.unlink() + except FileNotFoundError: + pass + except OSError: + pass + + +def _endpoint_for(transport: str) -> str: + adapter = adapter_for(transport) + if not adapter: + return "-" + return adapter.endpoint + + def _chmod_private(path: Path) -> None: try: path.chmod(0o700) diff --git a/docs/transport-management.md b/docs/transport-management.md index 0f47d33..feac63d 100644 --- a/docs/transport-management.md +++ b/docs/transport-management.md @@ -13,6 +13,10 @@ With `--launch --yes`, AMPB can start a configured Tor provider with AMPB-owned before launching. Transport policy comes from `.ampb/config.toml` or an explicit `--config` path. +`ampbrowser transport start|status|stop ` exposes the same ownership model without +launching a browser. It records only AMPB-owned processes and refuses to stop adopted +user or system daemons. + ## Lifecycle 1. Inspect known local endpoints. @@ -31,6 +35,7 @@ system service, or package manager that launched them. Managed transports store runtime state under `.ampb/transports/` by default. Keys, leases, tunnels, and daemon logs in that tree are local state and must not be committed. +AMPB records owned process metadata in `.ampb/transports//ampb-owned.json`. ## Current Adapters diff --git a/tests/test_launch.py b/tests/test_launch.py index 4cbef2c..7b548fe 100644 --- a/tests/test_launch.py +++ b/tests/test_launch.py @@ -103,6 +103,7 @@ class PrepareOpenTest(unittest.TestCase): self.assertEqual("launched", launched.status) self.assertFalse(launched.dry_run) + self.assertEqual(popen.return_value.pid, launched.browser_pid) popen.assert_called_once_with( (str(runtime), "-no-remote", "-profile", ".ampb/profiles/clearnet", "https://wownero.org"), cwd=str(root), diff --git a/tests/test_transport_manager.py b/tests/test_transport_manager.py index c27328d..f17a051 100644 --- a/tests/test_transport_manager.py +++ b/tests/test_transport_manager.py @@ -6,7 +6,7 @@ import unittest from unittest.mock import patch from ampbrowser.config import AppConfig -from ampbrowser.transport_manager import ensure_transport_ready +from ampbrowser.transport_manager import ensure_transport_ready, stop_managed_transport, transport_status from ampbrowser.transports import TransportStatus @@ -71,17 +71,97 @@ class TransportManagerTest(unittest.TestCase): popen.return_value.pid = 1234 result = ensure_transport_ready("tor", config=AppConfig(transport_modes={}), root=root, status=status) - self.assertEqual("started", result.status) + self.assertEqual("started", result.status) + self.assertEqual("arti", result.provider) + self.assertTrue(result.owned) + self.assertEqual(1234, result.pid) + self.assertEqual(str(root / ".ampb/transports/tor"), result.state_dir) + self.assertEqual(str(arti), result.command[0]) + self.assertIn("proxy", result.command) + self.assertIn("-p", result.command) + self.assertIn('storage.state_dir="' + str(root / ".ampb/transports/tor/arti-state") + '"', result.command) + self.assertIn('storage.cache_dir="' + str(root / ".ampb/transports/tor/arti-cache") + '"', result.command) + state_path = root / ".ampb/transports/tor/ampb-owned.json" + self.assertTrue(state_path.exists()) + self.assertIn('"provider": "arti"', state_path.read_text(encoding="utf-8")) + popen.assert_called_once() + + def test_reuses_recorded_ampb_owned_transport(self) -> None: + status = TransportStatus( + transport="tor", + installed=True, + running=False, + endpoint="socks5://127.0.0.1:9050", + adoptable=False, + manage_supported=True, + note="Tor SOCKS proxy", + ) + + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + state_dir = root / ".ampb/transports/tor" + state_dir.mkdir(parents=True) + (state_dir / "ampb-owned.json").write_text( + """ +{ + "command": ["/tmp/arti", "proxy"], + "endpoint": "socks5://127.0.0.1:9050", + "pid": 1234, + "provider": "arti", + "state_dir": "%s", + "transport": "tor" +} +""".strip() + % state_dir, + encoding="utf-8", + ) + with patch("ampbrowser.transport_manager._pid_alive", return_value=True): + with patch("ampbrowser.transport_manager._wait_for_endpoint", return_value=True): + with patch("ampbrowser.transport_manager.subprocess.Popen") as popen: + result = ensure_transport_ready("tor", config=AppConfig(transport_modes={}), root=root, status=status) + + self.assertEqual("ready", result.status) self.assertEqual("arti", result.provider) self.assertTrue(result.owned) self.assertEqual(1234, result.pid) - self.assertEqual(str(root / ".ampb/transports/tor"), result.state_dir) - self.assertEqual(str(arti), result.command[0]) - self.assertIn("proxy", result.command) - self.assertIn("-p", result.command) - self.assertIn('storage.state_dir="' + str(root / ".ampb/transports/tor/arti-state") + '"', result.command) - self.assertIn('storage.cache_dir="' + str(root / ".ampb/transports/tor/arti-cache") + '"', result.command) - popen.assert_called_once() + popen.assert_not_called() + + def test_transport_status_removes_stale_owned_state(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + state_dir = root / ".ampb/transports/tor" + state_dir.mkdir(parents=True) + state_path = state_dir / "ampb-owned.json" + state_path.write_text('{"pid": 1234, "provider": "arti", "endpoint": "socks5://127.0.0.1:9050"}\n', encoding="utf-8") + with patch("ampbrowser.transport_manager._pid_alive", return_value=False): + result = transport_status("tor", config=AppConfig(transport_modes={}), root=root) + + self.assertEqual("stale", result.status) + self.assertFalse(state_path.exists()) + + def test_stop_managed_transport_terminates_recorded_pid(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + state_dir = root / ".ampb/transports/tor" + state_dir.mkdir(parents=True) + state_path = state_dir / "ampb-owned.json" + state_path.write_text('{"pid": 1234, "provider": "arti", "endpoint": "socks5://127.0.0.1:9050"}\n', encoding="utf-8") + + live = [True] + + def fake_alive(pid: int) -> bool: + return live[0] + + def fake_kill(pid: int, sig: int) -> None: + live[0] = False + + with patch("ampbrowser.transport_manager._pid_alive", side_effect=fake_alive): + with patch("ampbrowser.transport_manager.os.kill", side_effect=fake_kill): + result = stop_managed_transport("tor", config=AppConfig(transport_modes={}), root=root) + + self.assertEqual("stopped", result.status) + self.assertEqual("arti", result.provider) + self.assertFalse(state_path.exists()) def test_starts_configured_classic_tor_provider_with_ampb_state(self) -> None: status = TransportStatus(