diff --git a/ampbrowser/cli.py b/ampbrowser/cli.py
index 1a7f039..e489e68 100644
--- a/ampbrowser/cli.py
+++ b/ampbrowser/cli.py
@@ -13,6 +13,7 @@ from .launch import prepare_open
from .open_runner import launch_open_plan
from .plan import plan_url
from .platforms import PLATFORM_CHOICES
+from .route_helper import serve_route_helper
from .routing import route_url
from .transport_manager import ensure_transport_ready, stop_managed_transport, transport_status
from .transports import inspect_transports
@@ -90,6 +91,13 @@ def main(argv: list[str] | None = None) -> int:
docs_generate = docs_subcommands.add_parser("generate", help="Generate docs from code.")
docs_generate.add_argument("--check", action="store_true")
+ helper_parser = subcommands.add_parser("helper", help=argparse.SUPPRESS)
+ helper_parser.add_argument("--host", default="127.0.0.1")
+ helper_parser.add_argument("--port", type=int, required=True)
+ helper_parser.add_argument("--token", required=True)
+ helper_parser.add_argument("--root", type=Path, default=Path.cwd())
+ helper_parser.add_argument("--config", type=Path, help="Path to an AMPB config file.")
+
args = parser.parse_args(argv)
try:
if args.command == "route":
@@ -116,6 +124,8 @@ def main(argv: list[str] | None = None) -> int:
return _cmd_fixture(args)
if args.command == "docs":
return _cmd_docs(args)
+ if args.command == "helper":
+ return _cmd_helper(args)
except Exception as exc: # noqa: BLE001
print(f"AMPBROWSER status=error message={exc}", file=sys.stderr)
return 1
@@ -241,6 +251,10 @@ def _print_open_plan(prefix: str, open_plan, extra_fields: tuple[str, ...] = ())
f"transport_setup_pid={open_plan.transport_setup_pid} "
f"transport_setup_endpoint={open_plan.transport_setup_endpoint} "
f"transport_setup_message=\"{_safe(open_plan.transport_setup_message)}\" "
+ f"route_helper_status={open_plan.route_helper_status} "
+ f"route_helper_endpoint={open_plan.route_helper_endpoint} "
+ f"route_helper_pid={open_plan.route_helper_pid} "
+ f"route_helper_message=\"{_safe(open_plan.route_helper_message)}\" "
f"launch_command=\"{_safe(launch_command)}\" "
f"setup_steps=\"{_safe(setup_steps)}\" "
f"message=\"{message}\""
@@ -301,6 +315,17 @@ def _cmd_docs(args) -> int:
return 1
+def _cmd_helper(args) -> int:
+ serve_route_helper(
+ host=args.host,
+ port=args.port,
+ token=args.token,
+ root=args.root,
+ config_path=args.config,
+ )
+ return 0
+
+
def _cmd_fixture(args) -> int:
if args.fixture_command == "check":
result = check_fixture_manifest(args.manifest)
diff --git a/ampbrowser/launch.py b/ampbrowser/launch.py
index f873395..4c50671 100644
--- a/ampbrowser/launch.py
+++ b/ampbrowser/launch.py
@@ -5,8 +5,11 @@ from dataclasses import replace
import json
import os
from pathlib import Path
+import secrets
import shlex
+import socket
import subprocess
+import sys
from urllib.parse import urlparse
from .adapters import adapter_for
@@ -16,7 +19,26 @@ from .plan import BrowsePlan, plan_url
ROUTE_AWARE_PROFILE = "route-aware"
ROUTE_AWARE_PAC_NAME = "ampb-proxy.pac"
+ROUTE_HELPER_EXTENSION_ID = "ampb-route-helper@such.software"
PAC_URL_PLACEHOLDER = "__AMPB_ROUTE_AWARE_PAC_URL__"
+HELPER_URL_PLACEHOLDER = "__AMPB_ROUTE_HELPER_URL__"
+HELPER_TOKEN_PLACEHOLDER = "__AMPB_ROUTE_HELPER_TOKEN__"
+
+
+@dataclass(frozen=True)
+class ProfileAsset:
+ path: str
+ content: str
+
+
+@dataclass(frozen=True)
+class RouteHelperLaunch:
+ status: str
+ endpoint: str = "-"
+ token: str = "-"
+ pid: int = 0
+ command: tuple[str, ...] = ()
+ message: str = "-"
@dataclass(frozen=True)
@@ -29,6 +51,8 @@ class BrowserLaunchSpec:
route_aware: bool = False
pac_path: str = "-"
pac_content: str = ""
+ extension_path: str = "-"
+ extension_assets: tuple[ProfileAsset, ...] = ()
@dataclass(frozen=True)
@@ -54,6 +78,10 @@ class OpenPlan:
transport_setup_pid: int = 0
transport_setup_endpoint: str = "-"
transport_setup_message: str = "-"
+ route_helper_status: str = "-"
+ route_helper_endpoint: str = "-"
+ route_helper_pid: int = 0
+ route_helper_message: str = "-"
def prepare_open(
@@ -163,9 +191,23 @@ def execute_open(open_plan: OpenPlan, *, root: Path | None = None) -> OpenPlan:
message=f"browser runtime not found: {spec.runtime_path}",
)
- _write_profile(root, spec)
+ route_helper = _start_route_helper(root) if spec.route_aware else RouteHelperLaunch("disabled")
+ _write_profile(root, spec, route_helper=route_helper)
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")
+ message = "launched bundled browser"
+ if spec.route_aware:
+ message = f"launched bundled browser with route helper {route_helper.status}"
+ return replace(
+ open_plan,
+ dry_run=False,
+ status="launched",
+ browser_pid=process.pid,
+ message=message,
+ route_helper_status=route_helper.status,
+ route_helper_endpoint=route_helper.endpoint,
+ route_helper_pid=route_helper.pid,
+ route_helper_message=route_helper.message,
+ )
def _proxy_for(browse_plan: BrowsePlan) -> str:
@@ -266,6 +308,8 @@ def _launch_spec_for(
prefs = _prefs_for(browse_plan, route_aware=route_aware)
pac_path = str(Path(profile_path) / ROUTE_AWARE_PAC_NAME) if route_aware else "-"
pac_content = _route_aware_pac() if route_aware else ""
+ extension_path = str(Path(profile_path) / "extensions" / ROUTE_HELPER_EXTENSION_ID) if route_aware else "-"
+ extension_assets = _route_helper_extension_assets(extension_path) if route_aware else ()
return BrowserLaunchSpec(
runtime_path=runtime_path,
profile_path=profile_path,
@@ -275,6 +319,8 @@ def _launch_spec_for(
route_aware=route_aware,
pac_path=pac_path,
pac_content=pac_content,
+ extension_path=extension_path,
+ extension_assets=extension_assets,
)
@@ -317,6 +363,9 @@ def _prefs_for(browse_plan: BrowsePlan, *, route_aware: bool = False) -> tuple[s
("network.proxy.failover_direct", False),
("network.proxy.socks_remote_dns", True),
("network.proxy.no_proxies_on", ""),
+ ("extensions.autoDisableScopes", 0),
+ ("extensions.enabledScopes", 5),
+ ("xpinstall.signatures.required", False),
]
)
return tuple(_format_pref(name, value) for name, value in prefs)
@@ -398,6 +447,177 @@ def _pac_proxy_for_endpoint(proxy_type: str, endpoint: str) -> str:
return f"{proxy_type} {host}:{port}"
+def _route_helper_extension_assets(extension_path: str) -> tuple[ProfileAsset, ...]:
+ base = Path(extension_path)
+ return (
+ ProfileAsset(str(base / "manifest.json"), _route_helper_manifest()),
+ ProfileAsset(str(base / "background.js"), _route_helper_background_js()),
+ ProfileAsset(str(base / "setup.html"), _route_helper_setup_html()),
+ ProfileAsset(str(base / "setup.js"), _route_helper_setup_js()),
+ ProfileAsset(str(base / "setup.css"), _route_helper_setup_css()),
+ )
+
+
+def _route_helper_manifest() -> str:
+ return json.dumps(
+ {
+ "manifest_version": 2,
+ "name": "AMPB Route Helper",
+ "version": "0.1.0",
+ "description": "Starts AMPB-managed transports when alternate-network links are opened.",
+ "applications": {"gecko": {"id": ROUTE_HELPER_EXTENSION_ID}},
+ "permissions": [
+ "tabs",
+ "webNavigation",
+ "http://127.0.0.1/*",
+ ],
+ "background": {"scripts": ["background.js"]},
+ },
+ indent=2,
+ sort_keys=True,
+ ) + "\n"
+
+
+def _route_helper_background_js() -> str:
+ return "\n".join(
+ (
+ '"use strict";',
+ "",
+ f'const HELPER_URL = "{HELPER_URL_PLACEHOLDER}";',
+ f'const HELPER_TOKEN = "{HELPER_TOKEN_PLACEHOLDER}";',
+ "const pendingByTab = new Map();",
+ "",
+ "function transportForUrl(rawUrl) {",
+ " try {",
+ " const parsed = new URL(rawUrl);",
+ " const host = parsed.hostname.toLowerCase().replace(/\\.$/, '');",
+ " if (host.endsWith('.onion')) return 'tor';",
+ " if (host.endsWith('.i2p')) return 'i2p';",
+ " } catch (_error) {}",
+ " return '';",
+ "}",
+ "",
+ "async function helper(action, transport, url) {",
+ " const response = await fetch(HELPER_URL, {",
+ " method: 'POST',",
+ " headers: {",
+ " 'Content-Type': 'application/json',",
+ " 'X-AMPB-Token': HELPER_TOKEN,",
+ " },",
+ " body: JSON.stringify({ action, transport, url }),",
+ " });",
+ " return response.json();",
+ "}",
+ "",
+ "async function handleNavigate(details) {",
+ " if (details.frameId !== 0) return;",
+ " const transport = transportForUrl(details.url);",
+ " if (!transport) return;",
+ " const key = `${details.tabId}:${details.url}`;",
+ " if (pendingByTab.get(details.tabId) === details.url) return;",
+ " let status;",
+ " try {",
+ " status = await helper('status', transport, details.url);",
+ " } catch (error) {",
+ " status = { ok: false, ready: false, message: String(error) };",
+ " }",
+ " if (status.ready) return;",
+ " pendingByTab.set(details.tabId, details.url);",
+ " const setupUrl = browser.runtime.getURL(",
+ " `setup.html?transport=${encodeURIComponent(transport)}&url=${encodeURIComponent(details.url)}&message=${encodeURIComponent(status.message || '')}`",
+ " );",
+ " await browser.tabs.update(details.tabId, { url: setupUrl });",
+ "}",
+ "",
+ "browser.webNavigation.onBeforeNavigate.addListener(handleNavigate);",
+ "",
+ "browser.runtime.onMessage.addListener((message) => {",
+ " if (!message || message.type !== 'ampb-ensure') return false;",
+ " return helper('ensure', message.transport, message.url);",
+ "});",
+ "",
+ )
+ )
+
+
+def _route_helper_setup_html() -> str:
+ return "\n".join(
+ (
+ "",
+ '',
+ "",
+ ' ',
+ ' ',
+ " Set Up Transport",
+ ' ',
+ "",
+ "",
+ ' ',
+ ' AMPB Route Helper
',
+ ' Set Up Transport
',
+ ' AMPB needs to start a local transport before opening this route.
',
+ ' ',
+ ' ',
+ ' ',
+ ' ',
+ " ",
+ ' ',
+ "",
+ "",
+ "",
+ )
+ )
+
+
+def _route_helper_setup_js() -> str:
+ return "\n".join(
+ (
+ '"use strict";',
+ "",
+ "const params = new URLSearchParams(window.location.search);",
+ "const transport = params.get('transport') || '';",
+ "const url = params.get('url') || '';",
+ "const message = params.get('message') || '';",
+ "const label = transport === 'tor' ? 'Tor' : transport === 'i2p' ? 'I2P' : transport;",
+ "",
+ "document.getElementById('title').textContent = `Set up ${label}?`;",
+ "document.getElementById('body').textContent = `AMPB can start a managed ${label} session and keep its state isolated.`;",
+ "document.getElementById('url').textContent = url;",
+ "document.getElementById('message').textContent = message;",
+ "document.getElementById('approve').textContent = `Start ${label} and open`;",
+ "",
+ "document.getElementById('approve').addEventListener('click', async () => {",
+ " const status = document.getElementById('status');",
+ " status.textContent = `Starting ${label}...`;",
+ " const result = await browser.runtime.sendMessage({ type: 'ampb-ensure', transport, url });",
+ " if (result && result.ready) {",
+ " status.textContent = `${label} is ready. Opening route...`;",
+ " window.location.href = url;",
+ " return;",
+ " }",
+ " status.textContent = (result && result.message) || `${label} did not become ready.`;",
+ "});",
+ "",
+ )
+ )
+
+
+def _route_helper_setup_css() -> str:
+ return "\n".join(
+ (
+ ":root { color-scheme: light dark; font-family: system-ui, sans-serif; }",
+ "body { margin: 0; min-height: 100vh; display: grid; place-items: center; background: #111; color: #f7f1e8; }",
+ ".panel { width: min(680px, calc(100vw - 32px)); }",
+ ".eyebrow { color: #36d1c4; font-weight: 700; letter-spacing: .08em; text-transform: uppercase; }",
+ "h1 { font-size: 2rem; margin: 0 0 12px; }",
+ "pre { white-space: pre-wrap; overflow-wrap: anywhere; padding: 12px; border: 1px solid #444; }",
+ "button { font: inherit; font-weight: 700; padding: 12px 16px; border: 0; background: #36d1c4; color: #111; }",
+ ".message, .status { color: #cfc7bb; }",
+ "",
+ )
+ )
+
+
def _endpoint_host_port(endpoint: str) -> tuple[str, int]:
parsed = urlparse(endpoint)
return parsed.hostname or "127.0.0.1", parsed.port or 0
@@ -413,7 +633,38 @@ def _format_pref(name: str, value: str | int | bool) -> str:
return f"user_pref({json.dumps(name)}, {literal});"
-def _write_profile(root: Path, spec: BrowserLaunchSpec) -> None:
+def _start_route_helper(root: Path) -> RouteHelperLaunch:
+ port = _free_loopback_port()
+ token = secrets.token_urlsafe(24)
+ endpoint = f"http://127.0.0.1:{port}/"
+ command = (
+ sys.executable,
+ "-m",
+ "ampbrowser",
+ "helper",
+ "--host",
+ "127.0.0.1",
+ "--port",
+ str(port),
+ "--token",
+ token,
+ "--root",
+ str(root),
+ )
+ try:
+ process = subprocess.Popen(command, cwd=str(root), start_new_session=True) # noqa: S603
+ except OSError as exc:
+ return RouteHelperLaunch("start-failed", endpoint=endpoint, token=token, command=command, message=str(exc))
+ return RouteHelperLaunch("started", endpoint=endpoint, token=token, pid=process.pid, command=command)
+
+
+def _free_loopback_port() -> int:
+ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+ sock.bind(("127.0.0.1", 0))
+ return int(sock.getsockname()[1])
+
+
+def _write_profile(root: Path, spec: BrowserLaunchSpec, *, route_helper: RouteHelperLaunch | None = None) -> None:
profile_path = _path_from(root, spec.profile_path)
profile_path.mkdir(parents=True, exist_ok=True)
pac_url = "-"
@@ -422,6 +673,14 @@ def _write_profile(root: Path, spec: BrowserLaunchSpec) -> None:
pac_path.parent.mkdir(parents=True, exist_ok=True)
pac_path.write_text(spec.pac_content, encoding="utf-8")
pac_url = pac_path.resolve().as_uri()
+ if spec.route_aware:
+ helper = route_helper or RouteHelperLaunch("disabled")
+ for asset in spec.extension_assets:
+ asset_path = _path_from(root, asset.path)
+ asset_path.parent.mkdir(parents=True, exist_ok=True)
+ content = asset.content.replace(HELPER_URL_PLACEHOLDER, helper.endpoint)
+ content = content.replace(HELPER_TOKEN_PLACEHOLDER, helper.token)
+ asset_path.write_text(content, encoding="utf-8")
user_js_path = _path_from(root, spec.user_js_path)
text = "// Managed by AMPB. Local browser profile policy.\n" + "\n".join(spec.prefs) + "\n"
text = text.replace(PAC_URL_PLACEHOLDER, pac_url)
diff --git a/ampbrowser/route_helper.py b/ampbrowser/route_helper.py
new file mode 100644
index 0000000..895ed9c
--- /dev/null
+++ b/ampbrowser/route_helper.py
@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+from pathlib import Path
+from typing import Any
+
+from .config import AppConfig, load_config
+from .routing import route_url
+from .transport_manager import ManagedTransportResult, ensure_transport_ready, transport_status
+
+
+SUPPORTED_HELPER_TRANSPORTS = {"tor", "i2p"}
+
+
+def handle_helper_message(
+ message: dict[str, Any],
+ *,
+ root: Path,
+ config: AppConfig,
+) -> dict[str, Any]:
+ action = _string(message.get("action"))
+ transport = _message_transport(message)
+ if not transport:
+ return _error_response("unsupported-route", "URL does not route to a managed browser transport")
+ if transport not in SUPPORTED_HELPER_TRANSPORTS:
+ return _error_response("unsupported-transport", f"{transport} is not supported by the route helper yet")
+
+ if action == "status":
+ return _result_response(transport_status(transport, config=config, root=root))
+ if action == "ensure":
+ return _result_response(ensure_transport_ready(transport, config=config, root=root))
+ return _error_response("unsupported-action", f"unsupported helper action: {action or '-'}")
+
+
+def serve_route_helper(
+ *,
+ host: str,
+ port: int,
+ token: str,
+ root: Path,
+ config_path: Path | None = None,
+) -> None:
+ config = load_config(root, config_path)
+ server = _RouteHelperServer((host, port), _RouteHelperHandler)
+ server.token = token
+ server.root_path = root
+ server.config = config
+ server.serve_forever()
+
+
+class _RouteHelperServer(ThreadingHTTPServer):
+ token: str
+ root_path: Path
+ config: AppConfig
+
+
+class _RouteHelperHandler(BaseHTTPRequestHandler):
+ server: _RouteHelperServer
+
+ def do_OPTIONS(self) -> None:
+ self._write_json(200, {"ok": True})
+
+ def do_POST(self) -> None:
+ if self.headers.get("X-AMPB-Token") != self.server.token:
+ self._write_json(403, _error_response("forbidden", "invalid helper token"))
+ return
+ try:
+ length = int(self.headers.get("Content-Length", "0"))
+ except ValueError:
+ length = 0
+ try:
+ payload = json.loads(self.rfile.read(length).decode("utf-8") or "{}")
+ except json.JSONDecodeError:
+ self._write_json(400, _error_response("bad-json", "request body is not valid JSON"))
+ return
+ if not isinstance(payload, dict):
+ self._write_json(400, _error_response("bad-request", "request body must be a JSON object"))
+ return
+ response = handle_helper_message(payload, root=self.server.root_path, config=self.server.config)
+ self._write_json(200 if response.get("ok") else 400, response)
+
+ def log_message(self, _format: str, *_args: object) -> None:
+ return
+
+ def _write_json(self, status: int, payload: dict[str, Any]) -> None:
+ data = json.dumps(payload, sort_keys=True).encode("utf-8")
+ self.send_response(status)
+ self.send_header("Access-Control-Allow-Origin", "*")
+ self.send_header("Access-Control-Allow-Headers", "Content-Type, X-AMPB-Token")
+ self.send_header("Access-Control-Allow-Methods", "POST, OPTIONS")
+ self.send_header("Content-Type", "application/json")
+ self.send_header("Content-Length", str(len(data)))
+ self.end_headers()
+ self.wfile.write(data)
+
+
+def _message_transport(message: dict[str, Any]) -> str:
+ raw_transport = _string(message.get("transport"))
+ if raw_transport:
+ return raw_transport
+ raw_url = _string(message.get("url"))
+ if not raw_url:
+ return ""
+ return route_url(raw_url).transport
+
+
+def _result_response(result: ManagedTransportResult) -> dict[str, Any]:
+ return {
+ "ok": True,
+ "transport": result.transport,
+ "provider": result.provider,
+ "provider_source": result.provider_source,
+ "status": result.status,
+ "ready": result.ready,
+ "owned": result.owned,
+ "pid": result.pid,
+ "endpoint": result.endpoint,
+ "state_dir": result.state_dir,
+ "message": result.message,
+ "command": list(result.command),
+ }
+
+
+def _error_response(status: str, message: str) -> dict[str, Any]:
+ return {
+ "ok": False,
+ "status": status,
+ "ready": False,
+ "message": message,
+ }
+
+
+def _string(value: object) -> str:
+ return value if isinstance(value, str) else ""
diff --git a/docs/open-flow.md b/docs/open-flow.md
index 14c6cf3..17ef3a6 100644
--- a/docs/open-flow.md
+++ b/docs/open-flow.md
@@ -34,6 +34,12 @@ adopted daemon from an AMPB-owned managed process.
through a desktop dialog when possible, falls back to a terminal prompt when needed, and
launches through the same transport runner after approval.
+`ampbrowser open --route-aware --launch` creates one desktop profile for clearnet,
+Tor, and I2P. The profile uses a PAC file for `.onion` and `.i2p` routing, installs the
+AMPB route-helper extension, and starts a loopback AMPB helper. If a user clicks an
+alternate-network link before its local transport is ready, the extension shows a setup
+page and asks the helper to start the AMPB-managed transport after approval.
+
## Examples
```sh
@@ -43,6 +49,7 @@ python3 -m ampbrowser open http://example.b32.i2p/ --yes
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 open https://wownero.org/ --launch
+python3 -m ampbrowser open https://ampgateway.site/ --route-aware --launch
python3 -m ampbrowser open http://example.onion/ --launch
python3 -m ampbrowser open http://example.onion/ --yes --launch
python3 -m ampbrowser shell http://example.onion/
diff --git a/docs/transport-management.md b/docs/transport-management.md
index 780449a..98f606a 100644
--- a/docs/transport-management.md
+++ b/docs/transport-management.md
@@ -17,6 +17,11 @@ state before launching. Transport policy comes from `.ampb/config.toml` or an ex
launching a browser. It records only AMPB-owned processes and refuses to stop adopted
user or system daemons.
+Route-aware desktop launches add a local helper process for in-browser setup prompts. The
+helper accepts token-gated loopback requests from the AMPB profile extension, reports
+transport status, and starts only supported AMPB-managed transports. It does not expose a
+public network listener or manage adopted daemons.
+
## Lifecycle
1. Inspect known local endpoints.
diff --git a/tests/test_launch.py b/tests/test_launch.py
index 48103f9..6e981d0 100644
--- a/tests/test_launch.py
+++ b/tests/test_launch.py
@@ -6,7 +6,7 @@ import tempfile
from pathlib import Path
from ampbrowser.config import AppConfig
-from ampbrowser.launch import execute_open, prepare_open
+from ampbrowser.launch import RouteHelperLaunch, execute_open, prepare_open
from ampbrowser.transports import TransportStatus
@@ -138,10 +138,15 @@ class PrepareOpenTest(unittest.TestCase):
self.assertIn('user_pref("network.proxy.type", 2);', plan.launch_spec.prefs)
self.assertIn("__AMPB_ROUTE_AWARE_PAC_URL__", "\n".join(plan.launch_spec.prefs))
- with patch("ampbrowser.launch.subprocess.Popen") as popen:
- launched = execute_open(plan, root=root)
+ helper = RouteHelperLaunch("started", endpoint="http://127.0.0.1:44001/", token="test-token", pid=4321)
+ with patch("ampbrowser.launch._start_route_helper", return_value=helper):
+ with patch("ampbrowser.launch.subprocess.Popen") as popen:
+ launched = execute_open(plan, root=root)
self.assertEqual("launched", launched.status)
+ self.assertEqual("started", launched.route_helper_status)
+ self.assertEqual("http://127.0.0.1:44001/", launched.route_helper_endpoint)
+ self.assertEqual(4321, launched.route_helper_pid)
popen.assert_called_once_with(
(str(runtime), "-no-remote", "-profile", ".ampb/profiles/route-aware", "https://ampgateway.site/"),
cwd=str(root),
@@ -156,6 +161,14 @@ class PrepareOpenTest(unittest.TestCase):
self.assertIn('hasSuffix(h, ".i2p")', pac)
self.assertIn('return "PROXY 127.0.0.1:4444";', pac)
self.assertIn('return "DIRECT";', pac)
+ extension = profile / "extensions/ampb-route-helper@such.software"
+ manifest = (extension / "manifest.json").read_text(encoding="utf-8")
+ background = (extension / "background.js").read_text(encoding="utf-8")
+ setup = (extension / "setup.html").read_text(encoding="utf-8")
+ self.assertIn("webNavigation", manifest)
+ self.assertIn("http://127.0.0.1:44001/", background)
+ self.assertIn("test-token", background)
+ self.assertIn("Set Up Transport", setup)
def test_execute_open_blocks_when_runtime_is_missing(self) -> None:
config = AppConfig(runtime_path="/missing/ampb/firefox", transport_modes={})
diff --git a/tests/test_route_helper.py b/tests/test_route_helper.py
new file mode 100644
index 0000000..6d9e7c2
--- /dev/null
+++ b/tests/test_route_helper.py
@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from ampbrowser.config import AppConfig
+from ampbrowser.route_helper import handle_helper_message
+from ampbrowser.transport_manager import ManagedTransportResult
+
+
+class RouteHelperTest(unittest.TestCase):
+ def test_status_routes_url_to_transport(self) -> None:
+ result = ManagedTransportResult(
+ transport="tor",
+ provider="arti",
+ status="ready",
+ endpoint="socks5://127.0.0.1:9050",
+ owned=True,
+ pid=1234,
+ state_dir=".ampb/transports/tor",
+ command=("/tmp/arti", "proxy"),
+ message="using running AMPB-owned tor",
+ provider_source="bundled-sidecar",
+ )
+
+ with tempfile.TemporaryDirectory() as tmp:
+ with patch("ampbrowser.route_helper.transport_status", return_value=result) as status:
+ response = handle_helper_message(
+ {"action": "status", "url": "http://example.onion/"},
+ root=Path(tmp),
+ config=AppConfig(transport_modes={}),
+ )
+
+ self.assertTrue(response["ok"])
+ self.assertTrue(response["ready"])
+ self.assertEqual("tor", response["transport"])
+ self.assertEqual("bundled-sidecar", response["provider_source"])
+ status.assert_called_once()
+
+ def test_ensure_starts_i2p_transport(self) -> None:
+ result = ManagedTransportResult(
+ transport="i2p",
+ provider="i2pd",
+ status="started",
+ endpoint="http://127.0.0.1:4444",
+ owned=True,
+ pid=1234,
+ state_dir=".ampb/transports/i2p",
+ command=("/opt/i2pd",),
+ message="started managed i2pd transport",
+ provider_source="system-package",
+ )
+
+ with tempfile.TemporaryDirectory() as tmp:
+ with patch("ampbrowser.route_helper.ensure_transport_ready", return_value=result) as ensure:
+ response = handle_helper_message(
+ {"action": "ensure", "transport": "i2p", "url": "http://example.b32.i2p/"},
+ root=Path(tmp),
+ config=AppConfig(transport_modes={}),
+ )
+
+ self.assertTrue(response["ok"])
+ self.assertTrue(response["ready"])
+ self.assertEqual("started", response["status"])
+ ensure.assert_called_once()
+
+ def test_blocks_unsupported_routes(self) -> None:
+ with tempfile.TemporaryDirectory() as tmp:
+ response = handle_helper_message(
+ {"action": "ensure", "url": "https://example.com/"},
+ root=Path(tmp),
+ config=AppConfig(transport_modes={}),
+ )
+
+ self.assertFalse(response["ok"])
+ self.assertEqual("unsupported-transport", response["status"])
+
+
+if __name__ == "__main__":
+ unittest.main()