Add dry-run open consent flow

This commit is contained in:
jwinterm
2026-07-05 20:23:12 -04:00
parent ff287b8b4b
commit 237d1e4fcb
7 changed files with 233 additions and 0 deletions
+43
View File
@@ -5,6 +5,7 @@ from pathlib import Path
import sys
from .docsgen import generate_docs
from .launch import prepare_open
from .plan import plan_url
from .routing import route_url
from .transports import inspect_transports
@@ -20,6 +21,20 @@ 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")
open_parser = subcommands.add_parser("open", help="Prepare a transport-aware open plan.")
open_parser.add_argument("url")
open_parser.add_argument(
"--yes",
action="store_true",
help="Approve first-use setup in the dry-run plan.",
)
open_parser.add_argument(
"--dry-run",
action="store_true",
default=True,
help="Print the open plan without side effects. This is the current default.",
)
subcommands.add_parser("inspect", help="Inspect local transport readiness.")
docs_parser = subcommands.add_parser("docs", help="Generate or check generated docs.")
@@ -33,6 +48,8 @@ def main(argv: list[str] | None = None) -> int:
return _cmd_route(args.url)
if args.command == "plan":
return _cmd_plan(args.url)
if args.command == "open":
return _cmd_open(args.url, consent=args.yes, dry_run=args.dry_run)
if args.command == "inspect":
return _cmd_inspect()
if args.command == "docs":
@@ -76,6 +93,28 @@ 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)
browse_plan = open_plan.browse_plan
setup_steps = "|".join(open_plan.setup_steps) if open_plan.setup_steps else "-"
message = _safe(open_plan.message)
print(
"AMPBROWSER_OPEN "
f"url={browse_plan.route.normalized} "
f"transport={browse_plan.route.transport} "
f"profile={browse_plan.route.profile} "
f"status={open_plan.status} "
f"dry_run={str(open_plan.dry_run).lower()} "
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} "
f"proxy={open_plan.proxy} "
f"setup_steps=\"{_safe(setup_steps)}\" "
f"message=\"{message}\""
)
return 0
def _cmd_inspect() -> int:
for status in inspect_transports():
print(
@@ -99,3 +138,7 @@ def _cmd_docs(args) -> int:
print(f"AMPBROWSER_DOCS status=ok mode={mode} changed={changed_text}")
return 0
return 1
def _safe(value: str) -> str:
return value.replace('"', "'")
+95
View File
@@ -0,0 +1,95 @@
from __future__ import annotations
from dataclasses import dataclass
from .plan import BrowsePlan, plan_url
PROFILE_ROOT = ".ampb/profiles"
@dataclass(frozen=True)
class OpenPlan:
browse_plan: BrowsePlan
status: str
dry_run: bool
consent_granted: bool
profile_path: str
proxy: str
setup_steps: tuple[str, ...]
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}"
proxy = _proxy_for(browse_plan)
if browse_plan.action.startswith("blocked"):
return OpenPlan(
browse_plan=browse_plan,
status="blocked",
dry_run=dry_run,
consent_granted=consent,
profile_path=profile_path,
proxy=proxy,
setup_steps=(),
message=browse_plan.action,
)
if browse_plan.requires_consent:
setup_steps = _setup_steps(browse_plan)
if not consent:
return OpenPlan(
browse_plan=browse_plan,
status="consent-required",
dry_run=dry_run,
consent_granted=False,
profile_path=profile_path,
proxy=proxy,
setup_steps=setup_steps,
message=browse_plan.prompt,
)
return OpenPlan(
browse_plan=browse_plan,
status="setup-approved",
dry_run=dry_run,
consent_granted=True,
profile_path=profile_path,
proxy=proxy,
setup_steps=setup_steps,
message="first-use setup approved; execution is not implemented yet",
)
return OpenPlan(
browse_plan=browse_plan,
status="ready",
dry_run=dry_run,
consent_granted=consent,
profile_path=profile_path,
proxy=proxy,
setup_steps=(),
message="ready to open with isolated profile",
)
def _proxy_for(browse_plan: BrowsePlan) -> str:
if browse_plan.route.transport in {"tor", "i2p"} and browse_plan.status:
return browse_plan.status.endpoint
return "-"
def _setup_steps(browse_plan: BrowsePlan) -> tuple[str, ...]:
if not browse_plan.status:
return ()
steps: list[str] = []
if not browse_plan.status.installed:
steps.append(f"install {browse_plan.route.transport}")
steps.extend(
[
f"start managed {browse_plan.route.transport}",
f"wait for {browse_plan.status.endpoint}",
]
)
return tuple(steps)