Add transport policy config

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} "

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

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"):

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,
+    )