diff --git a/README.md b/README.md index ad20214..97d61d8 100644 --- a/README.md +++ b/README.md @@ -35,6 +35,7 @@ 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 open http://example.b32.i2p/ --config examples/config.toml python3 -m ampbrowser docs generate python3 -m ampbrowser docs generate --check python3 tools/docs_check.py diff --git a/ampbrowser/cli.py b/ampbrowser/cli.py index aef5fe3..b3c9a90 100644 --- a/ampbrowser/cli.py +++ b/ampbrowser/cli.py @@ -4,6 +4,7 @@ import argparse from pathlib import Path import sys +from .config import load_config from .docsgen import generate_docs from .launch import prepare_open from .plan import plan_url @@ -20,9 +21,11 @@ def main(argv: list[str] | None = None) -> int: plan_parser = subcommands.add_parser("plan", help="Plan how a URL would be opened.") plan_parser.add_argument("url") + plan_parser.add_argument("--config", type=Path, help="Path to an AMPB config file.") open_parser = subcommands.add_parser("open", help="Prepare a transport-aware open plan.") open_parser.add_argument("url") + open_parser.add_argument("--config", type=Path, help="Path to an AMPB config file.") open_parser.add_argument( "--yes", action="store_true", @@ -47,9 +50,9 @@ def main(argv: list[str] | None = None) -> int: if args.command == "route": return _cmd_route(args.url) if args.command == "plan": - return _cmd_plan(args.url) + return _cmd_plan(args.url, config_path=args.config) if args.command == "open": - return _cmd_open(args.url, consent=args.yes, dry_run=args.dry_run) + return _cmd_open(args.url, consent=args.yes, dry_run=args.dry_run, config_path=args.config) if args.command == "inspect": return _cmd_inspect() if args.command == "docs": @@ -72,8 +75,9 @@ def _cmd_route(url: str) -> int: return 0 -def _cmd_plan(url: str) -> int: - plan = plan_url(url) +def _cmd_plan(url: str, *, config_path: Path | None) -> int: + config = load_config(Path.cwd(), config_path) + plan = plan_url(url, config=config) endpoint = plan.status.endpoint if plan.status else "-" running = str(plan.status.running).lower() if plan.status else "false" installed = str(plan.status.installed).lower() if plan.status else "false" @@ -86,6 +90,7 @@ def _cmd_plan(url: str) -> int: f"installed={installed} " f"running={running} " f"endpoint={endpoint} " + f"policy={plan.policy_mode} " f"requires_consent={str(plan.requires_consent).lower()} " f"action=\"{plan.action}\" " f"prompt=\"{prompt}\"" @@ -93,8 +98,9 @@ def _cmd_plan(url: str) -> int: return 0 -def _cmd_open(url: str, *, consent: bool, dry_run: bool) -> int: - open_plan = prepare_open(url, consent=consent, dry_run=dry_run) +def _cmd_open(url: str, *, consent: bool, dry_run: bool, config_path: Path | None) -> int: + config = load_config(Path.cwd(), config_path) + open_plan = prepare_open(url, consent=consent, dry_run=dry_run, config=config) browse_plan = open_plan.browse_plan setup_steps = "|".join(open_plan.setup_steps) if open_plan.setup_steps else "-" message = _safe(open_plan.message) @@ -105,6 +111,7 @@ def _cmd_open(url: str, *, consent: bool, dry_run: bool) -> int: f"profile={browse_plan.route.profile} " f"status={open_plan.status} " f"dry_run={str(open_plan.dry_run).lower()} " + 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"profile_path={open_plan.profile_path} " diff --git a/ampbrowser/config.py b/ampbrowser/config.py new file mode 100644 index 0000000..0346486 --- /dev/null +++ b/ampbrowser/config.py @@ -0,0 +1,187 @@ +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +try: + import tomllib +except ModuleNotFoundError: # pragma: no cover - exercised on older Python versions + tomllib = None # type: ignore[assignment] + + +DEFAULT_CONFIG_PATH = Path(".ampb/config.toml") +DEFAULT_TRANSPORT_MODE = "adopt-or-prompt-manage" +TRANSPORT_MODES = {"disabled", "adopt", "adopt-or-prompt-manage"} +MODE_ALIASES = {"adopt-or-manage": "adopt-or-prompt-manage"} + + +class ConfigError(ValueError): + pass + + +@dataclass(frozen=True) +class AppConfig: + state_dir: str = ".ampb" + default_engine: str = "system" + isolate_by_transport: bool = True + transport_modes: dict[str, str] | None = None + + def transport_mode(self, transport: str) -> str: + if not self.transport_modes: + return DEFAULT_TRANSPORT_MODE + return self.transport_modes.get(transport, DEFAULT_TRANSPORT_MODE) + + def profile_path(self, profile: str) -> str: + if self.isolate_by_transport: + return str(Path(self.state_dir) / "profiles" / profile) + return str(Path(self.state_dir) / "profile") + + +def default_config() -> AppConfig: + return AppConfig(transport_modes={}) + + +def load_config(root: Path | None = None, config_path: Path | None = None) -> AppConfig: + root = root or Path.cwd() + path = config_path or root / DEFAULT_CONFIG_PATH + if not path.exists(): + return default_config() + + text = path.read_text(encoding="utf-8") + if tomllib: + try: + data = tomllib.loads(text) + except tomllib.TOMLDecodeError as exc: + raise ConfigError(f"invalid TOML in {path}: {exc}") from exc + else: + data = _loads_simple_toml(text, path) + + if not isinstance(data, dict): + raise ConfigError(f"{path}: top-level config must be a table") + return _parse_config(data, path) + + +def _parse_config(data: dict[str, Any], path: Path) -> AppConfig: + browser = _table(data, "browser", path) + profiles = _table(data, "profiles", path) + transports = _table(data, "transports", path) + + state_dir = _string(browser, "state_dir", ".ampb", path) + default_engine = _string(browser, "default_engine", "system", path) + isolate_by_transport = _bool(profiles, "isolate_by_transport", True, path) + transport_modes = _transport_modes(transports, path) + + return AppConfig( + state_dir=state_dir, + default_engine=default_engine, + isolate_by_transport=isolate_by_transport, + transport_modes=transport_modes, + ) + + +def _transport_modes(transports: dict[str, Any], path: Path) -> dict[str, str]: + modes: dict[str, str] = {} + for name, raw_config in transports.items(): + if not isinstance(raw_config, dict): + raise ConfigError(f"{path}: transports.{name} must be a table") + + enabled = raw_config.get("enabled", True) + if not isinstance(enabled, bool): + raise ConfigError(f"{path}: transports.{name}.enabled must be true or false") + if not enabled: + modes[name] = "disabled" + continue + + raw_mode = raw_config.get("mode", DEFAULT_TRANSPORT_MODE) + if not isinstance(raw_mode, str): + raise ConfigError(f"{path}: transports.{name}.mode must be a string") + mode = MODE_ALIASES.get(raw_mode, raw_mode) + if mode not in TRANSPORT_MODES: + choices = ", ".join(sorted(TRANSPORT_MODES)) + raise ConfigError(f"{path}: transports.{name}.mode must be one of: {choices}") + modes[name] = mode + return modes + + +def _table(data: dict[str, Any], key: str, path: Path) -> dict[str, Any]: + value = data.get(key, {}) + if not isinstance(value, dict): + raise ConfigError(f"{path}: {key} must be a table") + return value + + +def _string(data: dict[str, Any], key: str, default: str, path: Path) -> str: + value = data.get(key, default) + if not isinstance(value, str): + raise ConfigError(f"{path}: {key} must be a string") + return value + + +def _bool(data: dict[str, Any], key: str, default: bool, path: Path) -> bool: + value = data.get(key, default) + if not isinstance(value, bool): + raise ConfigError(f"{path}: {key} must be true or false") + return value + + +def _loads_simple_toml(text: str, path: Path) -> dict[str, Any]: + data: dict[str, Any] = {} + current = data + for lineno, raw_line in enumerate(text.splitlines(), start=1): + line = _strip_comment(raw_line).strip() + if not line: + continue + if line.startswith("[") and line.endswith("]"): + current = _table_for_header(data, line[1:-1].strip(), path, lineno) + continue + if "=" not in line: + raise ConfigError(f"{path}:{lineno}: expected key = value") + key, raw_value = line.split("=", 1) + key = key.strip() + if not key: + raise ConfigError(f"{path}:{lineno}: empty key") + current[key] = _parse_simple_value(raw_value.strip(), path, lineno) + return data + + +def _table_for_header(data: dict[str, Any], header: str, path: Path, lineno: int) -> dict[str, Any]: + if not header: + raise ConfigError(f"{path}:{lineno}: empty table name") + current = data + for part in header.split("."): + if not part: + raise ConfigError(f"{path}:{lineno}: invalid table name") + existing = current.setdefault(part, {}) + if not isinstance(existing, dict): + raise ConfigError(f"{path}:{lineno}: table conflicts with value: {header}") + current = existing + return current + + +def _parse_simple_value(value: str, path: Path, lineno: int) -> str | bool: + if value == "true": + return True + if value == "false": + return False + if len(value) >= 2 and value.startswith('"') and value.endswith('"'): + return value[1:-1] + raise ConfigError(f"{path}:{lineno}: only strings and booleans are supported") + + +def _strip_comment(line: str) -> str: + in_string = False + escaped = False + for index, char in enumerate(line): + if escaped: + escaped = False + continue + if char == "\\": + escaped = True + continue + if char == '"': + in_string = not in_string + continue + if char == "#" and not in_string: + return line[:index] + return line diff --git a/ampbrowser/launch.py b/ampbrowser/launch.py index 1222a89..d75950c 100644 --- a/ampbrowser/launch.py +++ b/ampbrowser/launch.py @@ -2,12 +2,10 @@ from __future__ import annotations from dataclasses import dataclass +from .config import AppConfig, default_config from .plan import BrowsePlan, plan_url -PROFILE_ROOT = ".ampb/profiles" - - @dataclass(frozen=True) class OpenPlan: browse_plan: BrowsePlan @@ -20,9 +18,16 @@ class OpenPlan: message: str -def prepare_open(raw_url: str, *, consent: bool = False, dry_run: bool = True) -> OpenPlan: - browse_plan = plan_url(raw_url) - profile_path = f"{PROFILE_ROOT}/{browse_plan.route.profile}" +def prepare_open( + raw_url: str, + *, + consent: bool = False, + dry_run: bool = True, + config: AppConfig | None = None, +) -> OpenPlan: + config = config or default_config() + browse_plan = plan_url(raw_url, config=config) + profile_path = config.profile_path(browse_plan.route.profile) proxy = _proxy_for(browse_plan) if browse_plan.action.startswith("blocked"): diff --git a/ampbrowser/plan.py b/ampbrowser/plan.py index 41f2e5a..c0607af 100644 --- a/ampbrowser/plan.py +++ b/ampbrowser/plan.py @@ -2,6 +2,7 @@ from __future__ import annotations from dataclasses import dataclass +from .config import AppConfig, default_config from .routing import Route, route_url from .transports import TransportStatus, inspect_transport @@ -13,15 +14,43 @@ class BrowsePlan: action: str requires_consent: bool = False prompt: str = "" + policy_mode: str = "adopt-or-prompt-manage" -def plan_url(raw_url: str) -> BrowsePlan: +def plan_url(raw_url: str, *, config: AppConfig | None = None) -> BrowsePlan: + config = config or default_config() route = route_url(raw_url) status = inspect_transport(route.transport) + policy_mode = config.transport_mode(route.transport) + + if policy_mode == "disabled": + return BrowsePlan( + route=route, + status=status, + action=f"blocked by policy: {route.transport} disabled", + policy_mode=policy_mode, + ) if route.transport in {"clearnet", "unknown"}: - return BrowsePlan(route=route, status=status, action="open with clearnet profile") + return BrowsePlan( + route=route, + status=status, + action="open with clearnet profile", + policy_mode=policy_mode, + ) if status and status.adoptable: - return BrowsePlan(route=route, status=status, action="adopt existing transport") + return BrowsePlan( + route=route, + status=status, + action="adopt existing transport", + policy_mode=policy_mode, + ) + if policy_mode == "adopt": + return BrowsePlan( + route=route, + status=status, + action=f"blocked by policy: {route.transport} mode is adopt", + policy_mode=policy_mode, + ) if status and status.manage_supported: action = "prompt to start managed transport" prompt = ( @@ -40,5 +69,11 @@ def plan_url(raw_url: str) -> BrowsePlan: action=action, requires_consent=True, prompt=prompt, + policy_mode=policy_mode, ) - return BrowsePlan(route=route, status=status, action="blocked until adapter is configured") + return BrowsePlan( + route=route, + status=status, + action="blocked until adapter is configured", + policy_mode=policy_mode, + ) diff --git a/docs/configuration.md b/docs/configuration.md index 1696813..7d613f3 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -17,7 +17,12 @@ user opens a URL that needs it and approves the first-use prompt. - `gemini://` URLs use the Gemini profile. - Reticulum-family URLs are routed to the Reticulum adapter once configured. -## Future Config Shape +## Config File + +By default AMPB reads `.ampb/config.toml` from the current working directory. Commands +that make launch decisions also accept `--config `. + +Example configs live under `examples/`. ```toml [browser] @@ -26,12 +31,12 @@ state_dir = ".ampb" [transports.tor] enabled = true -mode = "adopt-or-manage" +mode = "adopt-or-prompt-manage" socks_endpoint = "127.0.0.1:9050" [transports.i2p] enabled = true -mode = "adopt-or-manage" +mode = "adopt-or-prompt-manage" http_proxy = "127.0.0.1:4444" [transports.reticulum] @@ -42,6 +47,13 @@ mode = "adopt" isolate_by_transport = true ``` +Supported transport modes: + +- `disabled`: do not open URLs that require this transport. +- `adopt`: use the transport only if it is already running. +- `adopt-or-prompt-manage`: adopt a running transport, otherwise show a first-use prompt + before managed setup. + ## Public And Private Files Public docs describe behavior, configuration, and supported fixtures. Private strategy, diff --git a/docs/open-flow.md b/docs/open-flow.md index 8295172..11ea0bc 100644 --- a/docs/open-flow.md +++ b/docs/open-flow.md @@ -26,4 +26,5 @@ the transport adapters and browser launcher are being stabilized. python3 -m ampbrowser open https://wownero.org/ python3 -m ampbrowser open http://example.b32.i2p/ python3 -m ampbrowser open http://example.b32.i2p/ --yes +python3 -m ampbrowser open http://example.b32.i2p/ --config examples/config.toml ``` diff --git a/docs/transport-management.md b/docs/transport-management.md index e0635b1..ab18531 100644 --- a/docs/transport-management.md +++ b/docs/transport-management.md @@ -7,6 +7,7 @@ transports only after the user opens a URL that needs them and approves setup. `ampbrowser open` currently emits a dry-run launch spec. The spec includes the selected profile, proxy endpoint, required setup steps, and whether user consent is still needed. +Transport policy comes from `.ampb/config.toml` or an explicit `--config` path. ## Lifecycle diff --git a/examples/config.toml b/examples/config.toml new file mode 100644 index 0000000..f69e684 --- /dev/null +++ b/examples/config.toml @@ -0,0 +1,17 @@ +[browser] +default_engine = "system" +state_dir = ".ampb" + +[transports.tor] +enabled = true +mode = "adopt-or-prompt-manage" + +[transports.i2p] +enabled = true +mode = "adopt-or-prompt-manage" + +[transports.reticulum] +enabled = false + +[profiles] +isolate_by_transport = true diff --git a/examples/i2p-disabled.toml b/examples/i2p-disabled.toml new file mode 100644 index 0000000..be548f5 --- /dev/null +++ b/examples/i2p-disabled.toml @@ -0,0 +1,9 @@ +[browser] +default_engine = "system" +state_dir = ".ampb" + +[transports.i2p] +enabled = false + +[profiles] +isolate_by_transport = true diff --git a/pyproject.toml b/pyproject.toml index c7818dd..35c849e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -2,7 +2,7 @@ name = "adaptive-multi-protocol-browser" version = "0.1.0" description = "Transport-aware browser orchestration shell." -requires-python = ">=3.11" +requires-python = ">=3.9" license = "Apache-2.0" [project.scripts] diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..5f31e23 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,74 @@ +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from ampbrowser.config import ConfigError, load_config + + +class ConfigTest(unittest.TestCase): + def test_missing_config_uses_defaults(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + config = load_config(Path(tmp)) + + self.assertEqual(".ampb", config.state_dir) + self.assertEqual("adopt-or-prompt-manage", config.transport_mode("tor")) + self.assertEqual(".ampb/profiles/tor", config.profile_path("tor")) + + def test_loads_transport_modes_and_state_dir(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "config.toml" + path.write_text( + """ +[browser] +state_dir = ".local/ampb" + +[transports.tor] +mode = "adopt" + +[transports.i2p] +enabled = false +""".strip(), + encoding="utf-8", + ) + + config = load_config(Path(tmp), path) + + self.assertEqual(".local/ampb", config.state_dir) + self.assertEqual("adopt", config.transport_mode("tor")) + self.assertEqual("disabled", config.transport_mode("i2p")) + self.assertEqual(".local/ampb/profiles/i2p", config.profile_path("i2p")) + + def test_accepts_adopt_or_manage_alias(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "config.toml" + path.write_text( + """ +[transports.tor] +mode = "adopt-or-manage" +""".strip(), + encoding="utf-8", + ) + + config = load_config(Path(tmp), path) + + self.assertEqual("adopt-or-prompt-manage", config.transport_mode("tor")) + + def test_rejects_unknown_mode(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "config.toml" + path.write_text( + """ +[transports.tor] +mode = "surprise" +""".strip(), + encoding="utf-8", + ) + + with self.assertRaises(ConfigError): + load_config(Path(tmp), path) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_launch.py b/tests/test_launch.py index 4692d4e..505bb6e 100644 --- a/tests/test_launch.py +++ b/tests/test_launch.py @@ -3,6 +3,7 @@ from __future__ import annotations import unittest from unittest.mock import patch +from ampbrowser.config import AppConfig from ampbrowser.launch import prepare_open from ampbrowser.transports import TransportStatus @@ -54,6 +55,13 @@ class PrepareOpenTest(unittest.TestCase): self.assertTrue(plan.consent_granted) self.assertEqual(("start managed tor", "wait for socks5://127.0.0.1:9050"), plan.setup_steps) + def test_custom_state_dir_changes_profile_path(self) -> None: + config = AppConfig(state_dir=".local/ampb", transport_modes={}) + + plan = prepare_open("wownero.org", config=config) + + self.assertEqual(".local/ampb/profiles/clearnet", plan.profile_path) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_plan.py b/tests/test_plan.py index 039ab5f..1cddcd8 100644 --- a/tests/test_plan.py +++ b/tests/test_plan.py @@ -3,6 +3,7 @@ from __future__ import annotations import unittest from unittest.mock import patch +from ampbrowser.config import AppConfig from ampbrowser.plan import plan_url from ampbrowser.transports import TransportStatus @@ -60,6 +61,34 @@ class PlanUrlTest(unittest.TestCase): self.assertTrue(plan.requires_consent) self.assertIn("can start", plan.prompt) + def test_disabled_transport_is_blocked_by_policy(self) -> None: + config = AppConfig(transport_modes={"i2p": "disabled"}) + + plan = plan_url("http://example.b32.i2p/", config=config) + + self.assertEqual("disabled", plan.policy_mode) + self.assertEqual("blocked by policy: i2p disabled", plan.action) + self.assertFalse(plan.requires_consent) + + def test_adopt_only_transport_does_not_prompt_to_start(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", + ) + config = AppConfig(transport_modes={"tor": "adopt"}) + + with patch("ampbrowser.plan.inspect_transport", return_value=status): + plan = plan_url("http://example.onion/", config=config) + + self.assertEqual("adopt", plan.policy_mode) + self.assertEqual("blocked by policy: tor mode is adopt", plan.action) + self.assertFalse(plan.requires_consent) + if __name__ == "__main__": unittest.main()