Initial AMPB scaffold

ampbrowser/__init__.py

@@ -0,0 +1,5 @@
+"""Adaptive Multi-Protocol Browser."""
+
+__all__ = ["__version__"]
+
+__version__ = "0.1.0"

ampbrowser/__main__.py

@@ -0,0 +1,5 @@
+from .cli import main
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

ampbrowser/cli.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+import sys
+
+from .docsgen import generate_docs
+from .plan import plan_url
+from .routing import route_url
+from .transports import inspect_transports
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(prog="ampbrowser")
+    subcommands = parser.add_subparsers(dest="command", required=True)
+
+    route_parser = subcommands.add_parser("route", help="Route a URL to a transport.")
+    route_parser.add_argument("url")
+
+    plan_parser = subcommands.add_parser("plan", help="Plan how a URL would be opened.")
+    plan_parser.add_argument("url")
+
+    subcommands.add_parser("inspect", help="Inspect local transport readiness.")
+
+    docs_parser = subcommands.add_parser("docs", help="Generate or check generated docs.")
+    docs_subcommands = docs_parser.add_subparsers(dest="docs_command", required=True)
+    docs_generate = docs_subcommands.add_parser("generate", help="Generate docs from code.")
+    docs_generate.add_argument("--check", action="store_true")
+
+    args = parser.parse_args(argv)
+    try:
+        if args.command == "route":
+            return _cmd_route(args.url)
+        if args.command == "plan":
+            return _cmd_plan(args.url)
+        if args.command == "inspect":
+            return _cmd_inspect()
+        if args.command == "docs":
+            return _cmd_docs(args)
+    except Exception as exc:  # noqa: BLE001
+        print(f"AMPBROWSER status=error message={exc}", file=sys.stderr)
+        return 1
+    return 1
+
+
+def _cmd_route(url: str) -> int:
+    route = route_url(url)
+    print(
+        "AMPBROWSER_ROUTE "
+        f"url={route.normalized} "
+        f"transport={route.transport} "
+        f"profile={route.profile} "
+        f"reason=\"{route.reason}\""
+    )
+    return 0
+
+
+def _cmd_plan(url: str) -> int:
+    plan = plan_url(url)
+    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"
+    prompt = plan.prompt.replace('"', "'") if plan.prompt else "-"
+    print(
+        "AMPBROWSER_PLAN "
+        f"url={plan.route.normalized} "
+        f"transport={plan.route.transport} "
+        f"profile={plan.route.profile} "
+        f"installed={installed} "
+        f"running={running} "
+        f"endpoint={endpoint} "
+        f"requires_consent={str(plan.requires_consent).lower()} "
+        f"action=\"{plan.action}\" "
+        f"prompt=\"{prompt}\""
+    )
+    return 0
+
+
+def _cmd_inspect() -> int:
+    for status in inspect_transports():
+        print(
+            "AMPBROWSER_INSPECT "
+            f"transport={status.transport} "
+            f"installed={str(status.installed).lower()} "
+            f"running={str(status.running).lower()} "
+            f"adoptable={str(status.adoptable).lower()} "
+            f"manage_supported={str(status.manage_supported).lower()} "
+            f"endpoint={status.endpoint} "
+            f"note=\"{status.note}\""
+        )
+    return 0
+
+
+def _cmd_docs(args) -> int:
+    if args.docs_command == "generate":
+        changed = generate_docs(Path.cwd(), check=args.check)
+        changed_text = ",".join(str(path) for path in changed) if changed else "-"
+        mode = "check" if args.check else "write"
+        print(f"AMPBROWSER_DOCS status=ok mode={mode} changed={changed_text}")
+        return 0
+    return 1

ampbrowser/docsgen.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from .metadata import ROUTE_RULES, TRANSPORT_DEFINITIONS
+
+
+GENERATED_DIR = Path("docs/generated")
+
+
+def generate_docs(root: Path, *, check: bool = False) -> list[Path]:
+    docs = {
+        GENERATED_DIR / "route-rules.md": _route_rules_doc(),
+        GENERATED_DIR / "transports.md": _transports_doc(),
+    }
+    changed: list[Path] = []
+    for rel_path, content in docs.items():
+        path = root / rel_path
+        if path.exists() and path.read_text(encoding="utf-8") == content:
+            continue
+        changed.append(rel_path)
+        if not check:
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(content, encoding="utf-8")
+    if check and changed:
+        names = ", ".join(str(path) for path in changed)
+        raise RuntimeError(f"generated docs are stale: {names}")
+    return changed
+
+
+def _header(title: str) -> str:
+    return (
+        f"# {title}\n\n"
+        "> Status: generated | Updated by `python3 -m ampbrowser docs generate` | Applies to: AMPB\n\n"
+        "This file is generated from code. Do not edit it by hand.\n\n"
+    )
+
+
+def _route_rules_doc() -> str:
+    rows = [
+        "| Match | Transport | Profile | Note |",
+        "| --- | --- | --- | --- |",
+    ]
+    for rule in ROUTE_RULES:
+        rows.append(f"| `{rule.match}` | `{rule.transport}` | `{rule.profile}` | {rule.note} |")
+    return _header("Generated Route Rules") + "\n".join(rows) + "\n"
+
+
+def _transports_doc() -> str:
+    rows = [
+        "| Transport | Adopt Check | Managed State | Profile | Note |",
+        "| --- | --- | --- | --- | --- |",
+    ]
+    for transport in TRANSPORT_DEFINITIONS:
+        rows.append(
+            f"| `{transport.name}` | {transport.adopt_check} | `{transport.managed_state}` | "
+            f"`{transport.profile}` | {transport.note} |"
+        )
+    return _header("Generated Transports") + "\n".join(rows) + "\n"

ampbrowser/metadata.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class RouteRule:
+    match: str
+    transport: str
+    profile: str
+    note: str
+
+
+@dataclass(frozen=True)
+class TransportDefinition:
+    name: str
+    adopt_check: str
+    managed_state: str
+    profile: str
+    note: str
+
+
+ROUTE_RULES = (
+    RouteRule("*.onion", "tor", "tor", "route through Tor SOCKS"),
+    RouteRule("*.i2p", "i2p", "i2p", "route through I2P HTTP proxy"),
+    RouteRule("gemini://*", "gemini", "gemini", "fetch and render Gemtext"),
+    RouteRule("rns://*, lxmf://*, nomad://*", "reticulum", "reticulum", "route through Reticulum adapter"),
+    RouteRule("http://*, https://*", "clearnet", "clearnet", "ordinary web profile"),
+)
+
+TRANSPORT_DEFINITIONS = (
+    TransportDefinition(
+        "tor",
+        "SOCKS on 127.0.0.1:9050",
+        ".ampb/transports/tor",
+        "tor",
+        "Adopt existing Tor when healthy; otherwise prompt before managed setup.",
+    ),
+    TransportDefinition(
+        "i2p",
+        "HTTP proxy on 127.0.0.1:4444",
+        ".ampb/transports/i2p",
+        "i2p",
+        "Adopt existing i2pd/I2P router when healthy; otherwise prompt before managed setup.",
+    ),
+    TransportDefinition(
+        "gemini",
+        "built-in renderer available",
+        ".ampb/transports/gemini",
+        "gemini",
+        "No daemon required for static Gemtext browsing.",
+    ),
+    TransportDefinition(
+        "reticulum",
+        "RNS tools and configured interfaces",
+        ".ampb/transports/reticulum",
+        "reticulum",
+        "Adapter planned; physical/link-layer setup may require operator config.",
+    ),
+)

ampbrowser/plan.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .routing import Route, route_url
+from .transports import TransportStatus, inspect_transport
+
+
+@dataclass(frozen=True)
+class BrowsePlan:
+    route: Route
+    status: TransportStatus | None
+    action: str
+    requires_consent: bool = False
+    prompt: str = ""
+
+
+def plan_url(raw_url: str) -> BrowsePlan:
+    route = route_url(raw_url)
+    status = inspect_transport(route.transport)
+    if route.transport in {"clearnet", "unknown"}:
+        return BrowsePlan(route=route, status=status, action="open with clearnet profile")
+    if status and status.adoptable:
+        return BrowsePlan(route=route, status=status, action="adopt existing transport")
+    if status and status.manage_supported:
+        action = "prompt to start managed transport"
+        prompt = (
+            f"{route.transport} is not running. AMPB can start a managed {route.transport} "
+            f"transport for {route.normalized}."
+        )
+        if not status.installed:
+            action = "prompt to install and start managed transport"
+            prompt = (
+                f"{route.transport} is not installed or running. AMPB can install and start "
+                f"a managed {route.transport} transport for {route.normalized}."
+            )
+        return BrowsePlan(
+            route=route,
+            status=status,
+            action=action,
+            requires_consent=True,
+            prompt=prompt,
+        )
+    return BrowsePlan(route=route, status=status, action="blocked until adapter is configured")

ampbrowser/routing.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from urllib.parse import urlparse
+
+
+@dataclass(frozen=True)
+class Route:
+    original: str
+    normalized: str
+    transport: str
+    profile: str
+    reason: str
+
+
+def route_url(raw_url: str) -> Route:
+    normalized = _normalize_url(raw_url)
+    parsed = urlparse(normalized)
+    scheme = parsed.scheme.lower()
+    host = (parsed.hostname or "").lower()
+
+    if scheme == "gemini":
+        return Route(raw_url, normalized, "gemini", "gemini", "gemini scheme")
+    if scheme in {"rns", "lxmf", "nomad"}:
+        return Route(raw_url, normalized, "reticulum", "reticulum", f"{scheme} scheme")
+    if host.endswith(".onion"):
+        return Route(raw_url, normalized, "tor", "tor", ".onion host")
+    if host.endswith(".i2p"):
+        return Route(raw_url, normalized, "i2p", "i2p", ".i2p host")
+    if scheme in {"http", "https"}:
+        return Route(raw_url, normalized, "clearnet", "clearnet", "http(s) URL")
+    return Route(raw_url, normalized, "unknown", "unknown", "no route rule matched")
+
+
+def _normalize_url(raw_url: str) -> str:
+    value = raw_url.strip()
+    if "://" not in value and not value.startswith(("rns:", "lxmf:")):
+        value = "https://" + value
+    return value

ampbrowser/transports.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import shutil
+import socket
+
+
+@dataclass(frozen=True)
+class TransportStatus:
+    transport: str
+    installed: bool
+    running: bool
+    endpoint: str
+    adoptable: bool
+    manage_supported: bool
+    note: str
+
+
+TRANSPORTS = ("tor", "i2p", "reticulum", "gemini")
+
+
+def inspect_transports() -> list[TransportStatus]:
+    return [
+        _inspect_tor(),
+        _inspect_i2p(),
+        _inspect_reticulum(),
+        _inspect_gemini(),
+    ]
+
+
+def inspect_transport(name: str) -> TransportStatus | None:
+    for status in inspect_transports():
+        if status.transport == name:
+            return status
+    return None
+
+
+def _inspect_tor() -> TransportStatus:
+    installed = shutil.which("tor") is not None
+    running = _can_connect("127.0.0.1", 9050)
+    return TransportStatus(
+        transport="tor",
+        installed=installed,
+        running=running,
+        endpoint="socks5://127.0.0.1:9050",
+        adoptable=running,
+        manage_supported=True,
+        note="Tor SOCKS proxy",
+    )
+
+
+def _inspect_i2p() -> TransportStatus:
+    installed = shutil.which("i2pd") is not None or shutil.which("i2prouter") is not None
+    running = _can_connect("127.0.0.1", 4444)
+    return TransportStatus(
+        transport="i2p",
+        installed=installed,
+        running=running,
+        endpoint="http://127.0.0.1:4444",
+        adoptable=running,
+        manage_supported=True,
+        note="I2P HTTP proxy",
+    )
+
+
+def _inspect_reticulum() -> TransportStatus:
+    installed = any(shutil.which(command) for command in ("rnsd", "rnstatus", "rnodeconf"))
+    return TransportStatus(
+        transport="reticulum",
+        installed=installed,
+        running=False,
+        endpoint="rns://local",
+        adoptable=False,
+        manage_supported=False,
+        note="Reticulum adapter is planned; interface readiness needs explicit config",
+    )
+
+
+def _inspect_gemini() -> TransportStatus:
+    return TransportStatus(
+        transport="gemini",
+        installed=True,
+        running=True,
+        endpoint="builtin://gemtext-renderer",
+        adoptable=True,
+        manage_supported=True,
+        note="Built-in Gemtext fetch/render path",
+    )
+
+
+def _can_connect(host: str, port: int) -> bool:
+    try:
+        with socket.create_connection((host, port), timeout=0.2):
+            return True
+    except OSError:
+        return False