From dad491928e877a35addc52f2b5be99223aaa5f2c Mon Sep 17 00:00:00 2001 From: jwinterm Date: Wed, 8 Jul 2026 10:53:15 -0400 Subject: [PATCH] Add route-aware desktop proxy profile --- README.md | 1 + ampbrowser/cli.py | 20 +++++++- ampbrowser/launch.py | 102 ++++++++++++++++++++++++++++++++++++++--- tests/test_cli_open.py | 10 ++++ tests/test_launch.py | 34 ++++++++++++++ 5 files changed, 159 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index 3a07808..ee2a7e0 100644 --- a/README.md +++ b/README.md @@ -39,6 +39,7 @@ python3 -m ampbrowser route ipfs://bafyexample/ python3 -m ampbrowser inspect python3 -m ampbrowser plan gemini://wownero.org/ python3 -m ampbrowser open https://wownero.org/ +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/ampbrowser/cli.py b/ampbrowser/cli.py index 66efbda..1a7f039 100644 --- a/ampbrowser/cli.py +++ b/ampbrowser/cli.py @@ -39,6 +39,11 @@ def main(argv: list[str] | None = None) -> int: action="store_true", help="Approve first-use setup in the dry-run plan.", ) + open_parser.add_argument( + "--route-aware", + action="store_true", + help="Launch one desktop profile with PAC routing for clearnet, Tor, and I2P.", + ) open_mode = open_parser.add_mutually_exclusive_group() open_mode.add_argument( "--dry-run", @@ -99,6 +104,7 @@ def main(argv: list[str] | None = None) -> int: launch=args.launch, config_path=args.config, platform=args.platform, + route_aware=args.route_aware, ) if args.command == "shell": return _cmd_shell(args.url, approve=args.yes, config_path=args.config) @@ -162,9 +168,17 @@ def _cmd_open( launch: bool, config_path: Path | None, platform: str | None, + route_aware: bool, ) -> int: config = load_config(Path.cwd(), config_path) - open_plan = prepare_open(url, consent=consent, dry_run=dry_run, config=config, platform=platform) + open_plan = prepare_open( + url, + consent=consent, + dry_run=dry_run, + config=config, + platform=platform, + route_aware=route_aware, + ) if launch: open_plan = launch_open_plan(open_plan, config=config, root=Path.cwd()) _print_open_plan("AMPBROWSER_OPEN", open_plan) @@ -192,6 +206,8 @@ def _print_open_plan(prefix: str, open_plan, extra_fields: tuple[str, ...] = ()) launch_command = shlex.join(launch_spec.command) if launch_spec else "-" runtime_path = launch_spec.runtime_path if launch_spec else "-" user_js_path = launch_spec.user_js_path if launch_spec else "-" + route_aware = str(bool(launch_spec and launch_spec.route_aware)).lower() + pac_path = launch_spec.pac_path if launch_spec else "-" message = _safe(open_plan.message) extra = (" " + " ".join(extra_fields)) if extra_fields else "" print( @@ -210,8 +226,10 @@ def _print_open_plan(prefix: str, open_plan, extra_fields: tuple[str, ...] = ()) f"browser_pid={open_plan.browser_pid} " f"profile_path={open_plan.profile_path} " f"proxy={open_plan.proxy} " + f"route_aware={route_aware} " f"runtime_path={runtime_path} " f"user_js_path={user_js_path} " + f"pac_path={pac_path} " f"setup_prompt_title=\"{_safe(open_plan.setup_prompt_title)}\" " f"setup_prompt_body=\"{_safe(open_plan.setup_prompt_body)}\" " f"setup_prompt_approve_label=\"{_safe(open_plan.setup_prompt_approve_label)}\" " diff --git a/ampbrowser/launch.py b/ampbrowser/launch.py index b256c00..f873395 100644 --- a/ampbrowser/launch.py +++ b/ampbrowser/launch.py @@ -14,6 +14,11 @@ from .config import AppConfig, default_config from .plan import BrowsePlan, plan_url +ROUTE_AWARE_PROFILE = "route-aware" +ROUTE_AWARE_PAC_NAME = "ampb-proxy.pac" +PAC_URL_PLACEHOLDER = "__AMPB_ROUTE_AWARE_PAC_URL__" + + @dataclass(frozen=True) class BrowserLaunchSpec: runtime_path: str @@ -21,6 +26,9 @@ class BrowserLaunchSpec: user_js_path: str command: tuple[str, ...] prefs: tuple[str, ...] + route_aware: bool = False + pac_path: str = "-" + pac_content: str = "" @dataclass(frozen=True) @@ -55,12 +63,18 @@ def prepare_open( dry_run: bool = True, config: AppConfig | None = None, platform: str | None = None, + route_aware: bool = False, ) -> OpenPlan: config = config or default_config() browse_plan = plan_url(raw_url, config=config, platform=platform) - profile_path = config.profile_path(browse_plan.route.profile) + profile_name = ( + ROUTE_AWARE_PROFILE + if route_aware and browse_plan.route.transport in {"clearnet", "tor", "i2p"} + else browse_plan.route.profile + ) + profile_path = config.profile_path(profile_name) proxy = _proxy_for(browse_plan) - launch_spec = _launch_spec_for(browse_plan, config=config, profile_path=profile_path) + launch_spec = _launch_spec_for(browse_plan, config=config, profile_path=profile_path, route_aware=route_aware) if browse_plan.action.startswith("blocked"): return OpenPlan( @@ -77,7 +91,7 @@ def prepare_open( if browse_plan.requires_consent: setup_steps = _setup_steps(browse_plan) - setup_prompt = _setup_prompt(browse_plan, setup_steps) + setup_prompt = _setup_prompt(browse_plan, setup_steps, route_aware=route_aware) if not consent: return OpenPlan( browse_plan=browse_plan, @@ -192,7 +206,7 @@ class _SetupPrompt: approval_command: str -def _setup_prompt(browse_plan: BrowsePlan, setup_steps: tuple[str, ...]) -> _SetupPrompt: +def _setup_prompt(browse_plan: BrowsePlan, setup_steps: tuple[str, ...], *, route_aware: bool = False) -> _SetupPrompt: transport = browse_plan.route.transport label = _transport_label(transport) platform = browse_plan.platform_capability.platform @@ -214,7 +228,10 @@ def _setup_prompt(browse_plan: BrowsePlan, setup_steps: tuple[str, ...]) -> _Set f"{availability} {management} The browser will use {endpoint} for " f"{browse_plan.route.normalized}. Setup plan: {setup_summary}." ) - approval_command = shlex.join(("ampbrowser", "open", browse_plan.route.normalized, "--yes", "--launch")) + approval_args = ["ampbrowser", "open", browse_plan.route.normalized, "--yes", "--launch"] + if route_aware: + approval_args.append("--route-aware") + approval_command = shlex.join(approval_args) return _SetupPrompt( title=f"Set up {label}?", body=body, @@ -239,19 +256,25 @@ def _launch_spec_for( *, config: AppConfig, profile_path: str, + route_aware: bool = False, ) -> BrowserLaunchSpec | None: if browse_plan.route.transport not in {"clearnet", "tor", "i2p"}: return None runtime_path = _runtime_path(config) user_js_path = str(Path(profile_path) / "user.js") command = (runtime_path, "-no-remote", "-profile", profile_path, browse_plan.route.normalized) - prefs = _prefs_for(browse_plan) + 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 "" return BrowserLaunchSpec( runtime_path=runtime_path, profile_path=profile_path, user_js_path=user_js_path, command=command, prefs=prefs, + route_aware=route_aware, + pac_path=pac_path, + pac_content=pac_content, ) @@ -277,7 +300,7 @@ def _runtime_candidates(build_root: Path) -> tuple[Path, ...]: ) -def _prefs_for(browse_plan: BrowsePlan) -> tuple[str, ...]: +def _prefs_for(browse_plan: BrowsePlan, *, route_aware: bool = False) -> tuple[str, ...]: prefs: list[tuple[str, str | int | bool]] = [ ("browser.shell.checkDefaultBrowser", False), ("datareporting.policy.dataSubmissionEnabled", False), @@ -286,6 +309,18 @@ def _prefs_for(browse_plan: BrowsePlan) -> tuple[str, ...]: ("network.predictor.enabled", False), ] + if route_aware: + prefs.extend( + [ + ("network.proxy.type", 2), + ("network.proxy.autoconfig_url", PAC_URL_PLACEHOLDER), + ("network.proxy.failover_direct", False), + ("network.proxy.socks_remote_dns", True), + ("network.proxy.no_proxies_on", ""), + ] + ) + return tuple(_format_pref(name, value) for name, value in prefs) + transport = browse_plan.route.transport if transport == "tor" and browse_plan.status: host, port = _endpoint_host_port(browse_plan.status.endpoint) @@ -317,6 +352,52 @@ def _prefs_for(browse_plan: BrowsePlan) -> tuple[str, ...]: return tuple(_format_pref(name, value) for name, value in prefs) +def _route_aware_pac() -> str: + tor_proxy = _pac_proxy_for_endpoint( + "SOCKS5", + _endpoint_for_transport("tor", "socks5://127.0.0.1:9050"), + ) + i2p_proxy = _pac_proxy_for_endpoint( + "PROXY", + _endpoint_for_transport("i2p", "http://127.0.0.1:4444"), + ) + return "\n".join( + ( + "// Managed by AMPB. Route-aware proxy policy.", + "function hasSuffix(value, suffix) {", + " return value.length >= suffix.length && value.substring(value.length - suffix.length) === suffix;", + "}", + "", + "function FindProxyForURL(url, host) {", + ' var h = (host || "").toLowerCase();', + ' if (h.length > 0 && h.charAt(h.length - 1) === ".") {', + " h = h.substring(0, h.length - 1);", + " }", + ' if (hasSuffix(h, ".onion")) {', + f' return "{tor_proxy}";', + " }", + ' if (hasSuffix(h, ".i2p")) {', + f' return "{i2p_proxy}";', + " }", + ' return "DIRECT";', + "}", + "", + ) + ) + + +def _endpoint_for_transport(transport: str, fallback: str) -> str: + adapter = adapter_for(transport) + if adapter: + return adapter.endpoint + return fallback + + +def _pac_proxy_for_endpoint(proxy_type: str, endpoint: str) -> str: + host, port = _endpoint_host_port(endpoint) + return f"{proxy_type} {host}:{port}" + + def _endpoint_host_port(endpoint: str) -> tuple[str, int]: parsed = urlparse(endpoint) return parsed.hostname or "127.0.0.1", parsed.port or 0 @@ -335,8 +416,15 @@ def _format_pref(name: str, value: str | int | bool) -> str: def _write_profile(root: Path, spec: BrowserLaunchSpec) -> None: profile_path = _path_from(root, spec.profile_path) profile_path.mkdir(parents=True, exist_ok=True) + pac_url = "-" + if spec.route_aware and spec.pac_content: + pac_path = _path_from(root, spec.pac_path) + 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() 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) user_js_path.write_text(text, encoding="utf-8") diff --git a/tests/test_cli_open.py b/tests/test_cli_open.py index 0a80e1e..ed759eb 100644 --- a/tests/test_cli_open.py +++ b/tests/test_cli_open.py @@ -108,6 +108,16 @@ class CliOpenTest(unittest.TestCase): self.assertIn("prompt_shown=false", output) self.assertIn("prompt_approved=false", output) + def test_route_aware_open_prints_profile_and_pac_path(self) -> None: + with patch.object(sys, "stdout", new_callable=io.StringIO) as stdout: + code = main(["open", "https://ampgateway.site/", "--route-aware"]) + + self.assertEqual(0, code) + output = stdout.getvalue() + self.assertIn("route_aware=true", output) + self.assertIn("profile_path=.ampb/profiles/route-aware", output) + self.assertIn("pac_path=.ampb/profiles/route-aware/ampb-proxy.pac", output) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_launch.py b/tests/test_launch.py index add355c..48103f9 100644 --- a/tests/test_launch.py +++ b/tests/test_launch.py @@ -123,6 +123,40 @@ class PrepareOpenTest(unittest.TestCase): self.assertTrue(user_js.exists()) self.assertIn('user_pref("network.proxy.type", 0);', user_js.read_text(encoding="utf-8")) + def test_route_aware_open_writes_pac_profile(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + runtime = root / "firefox" + runtime.write_text("#!/bin/sh\n", encoding="utf-8") + config = AppConfig(runtime_path=str(runtime), transport_modes={}) + plan = prepare_open("https://ampgateway.site/", config=config, route_aware=True) + + self.assertEqual(".ampb/profiles/route-aware", plan.profile_path) + self.assertIsNotNone(plan.launch_spec) + self.assertTrue(plan.launch_spec.route_aware) + self.assertEqual(".ampb/profiles/route-aware/ampb-proxy.pac", plan.launch_spec.pac_path) + 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) + + self.assertEqual("launched", launched.status) + popen.assert_called_once_with( + (str(runtime), "-no-remote", "-profile", ".ampb/profiles/route-aware", "https://ampgateway.site/"), + cwd=str(root), + ) + profile = root / ".ampb/profiles/route-aware" + user_js = (profile / "user.js").read_text(encoding="utf-8") + pac = (profile / "ampb-proxy.pac").read_text(encoding="utf-8") + self.assertIn((profile / "ampb-proxy.pac").resolve().as_uri(), user_js) + self.assertNotIn("__AMPB_ROUTE_AWARE_PAC_URL__", user_js) + self.assertIn('hasSuffix(h, ".onion")', pac) + self.assertIn('return "SOCKS5 127.0.0.1:9050";', pac) + self.assertIn('hasSuffix(h, ".i2p")', pac) + self.assertIn('return "PROXY 127.0.0.1:4444";', pac) + self.assertIn('return "DIRECT";', pac) + def test_execute_open_blocks_when_runtime_is_missing(self) -> None: config = AppConfig(runtime_path="/missing/ampb/firefox", transport_modes={}) plan = prepare_open("wownero.org", config=config)