From c1d8337365336dadc132a3eedd72620cbc1c04bd Mon Sep 17 00:00:00 2001 From: jwinterm Date: Mon, 6 Jul 2026 03:58:47 -0400 Subject: [PATCH] Explain route exposure decisions --- README.md | 2 + ampg/cli.py | 63 +++++++++++++ ampg/config.py | 15 ++- ampg/manifest.py | 53 ++--------- ampg/route_policy.py | 185 +++++++++++++++++++++++++++++++++++++ docs/fixture-manifests.md | 4 +- docs/interactive-apps.md | 16 ++++ tests/test_route_policy.py | 169 +++++++++++++++++++++++++++++++++ 8 files changed, 455 insertions(+), 52 deletions(-) create mode 100644 ampg/route_policy.py create mode 100644 tests/test_route_policy.py diff --git a/README.md b/README.md index 95cc281..e78cd0e 100644 --- a/README.md +++ b/README.md @@ -103,6 +103,8 @@ python3 -m ampg --config examples/wownero.gateway.toml plan python3 -m ampg --config examples/wownero.gateway.toml plan --write-artifacts python3 -m ampg --config examples/wownero.gateway.toml build python3 -m ampg --config examples/wownero.gateway.toml manifest +python3 -m ampg --config examples/wownero.gateway.toml routes explain +python3 -m ampg --config examples/wownero.gateway.toml routes validate python3 -m ampg --config examples/wownero.gateway.toml audit python3 -m ampg --config examples/i2p-only.gateway.toml build python3 -m ampg docs generate diff --git a/ampg/cli.py b/ampg/cli.py index 07ee88e..6565c56 100644 --- a/ampg/cli.py +++ b/ampg/cli.py @@ -10,6 +10,7 @@ from .config import load_config from .docsgen import generate_docs from .manifest import write_fixture_manifests from .plan import plan_gateway, write_plan_artifacts +from .route_policy import RouteExposure, RouteIssue, route_exposures, route_issues def main(argv: list[str] | None = None) -> int: @@ -29,6 +30,13 @@ def main(argv: list[str] | None = None) -> int: ) subcommands.add_parser("build", help="Build enabled protocol outputs.") subcommands.add_parser("manifest", help="Write AMPB fixture manifests.") + routes_parser = subcommands.add_parser("routes", help="Explain route exposure policy.") + routes_subcommands = routes_parser.add_subparsers(dest="routes_command", required=True) + routes_subcommands.add_parser("explain", help="Print per-protocol route decisions.") + routes_subcommands.add_parser( + "validate", + help="Fail when a public route has no compatible enabled protocol.", + ) audit_parser = subcommands.add_parser("audit", help="Audit source HTML quality.") audit_parser.add_argument( "--fail-on-warn", @@ -56,6 +64,8 @@ def main(argv: list[str] | None = None) -> int: return _cmd_build(config) if args.command == "manifest": return _cmd_manifest(config) + if args.command == "routes": + return _cmd_routes(config, args) if args.command == "audit": return _cmd_audit(config, fail_on_warn=args.fail_on_warn) except Exception as exc: # noqa: BLE001 - CLI should print concise failures. @@ -127,6 +137,59 @@ def _cmd_manifest(config) -> int: return 0 +def _cmd_routes(config, args) -> int: + exposures = route_exposures(config) + issues = route_issues(config) + for exposure in exposures: + _print_route_exposure(exposure) + for issue in issues: + _print_route_issue(issue) + print( + "AMPG_ROUTE_SUMMARY " + f"sites={len(config.sites)} " + f"routes={sum(len(site.interactions.routes) for site in config.sites)} " + f"decisions={len(exposures)} " + f"exposed={sum(1 for exposure in exposures if exposure.status == 'exposed')} " + f"skipped={sum(1 for exposure in exposures if exposure.status == 'skipped')} " + f"issues={len(issues)}" + ) + if args.routes_command == "validate" and issues: + return 1 + return 0 + + +def _print_route_exposure(exposure: RouteExposure) -> None: + print( + "AMPG_ROUTE " + f"site={exposure.site_id} " + f"protocol={exposure.protocol} " + f"route_index={exposure.route_index} " + f"route=\"{exposure.match}\" " + f"source={exposure.source} " + f"tier={exposure.tier} " + f"identity={exposure.identity} " + f"payments={exposure.payments} " + f"realtime={str(exposure.realtime).lower()} " + f"public_allowed={str(exposure.public_allowed).lower()} " + f"max_tier={exposure.max_tier} " + f"status={exposure.status} " + f"reason=\"{exposure.reason}\"" + ) + + +def _print_route_issue(issue: RouteIssue) -> None: + print( + "AMPG_ROUTE_ISSUE " + f"site={issue.site_id} " + f"route_index={issue.route_index} " + f"route=\"{issue.match}\" " + f"source={issue.source} " + f"tier={issue.tier} " + f"code={issue.code} " + f"message=\"{issue.message}\"" + ) + + def _cmd_audit(config, *, fail_on_warn: bool) -> int: issues = audit_gateway(config) for issue in issues: diff --git a/ampg/config.py b/ampg/config.py index 2c8264b..0a69d81 100644 --- a/ampg/config.py +++ b/ampg/config.py @@ -59,6 +59,7 @@ class RoutePolicyConfig: payments: str = "none" realtime: bool = False public_allowed: bool = True + source: str = "config" @dataclass(frozen=True) @@ -165,11 +166,11 @@ def _parse_interactions( deny_routes = _string_array(manifest_data.get("deny_routes", ()), "route manifest deny_routes") deny_routes += _string_array(raw_interactions.get("deny_routes", ()), "site.interactions.deny_routes") routes = tuple( - _parse_route_policy(raw_route, default_tier) + _parse_route_policy(raw_route, default_tier, source="route-manifest") for raw_route in _route_array(manifest_data.get("routes", ()), "route manifest routes") ) routes += tuple( - _parse_route_policy(raw_route, default_tier) + _parse_route_policy(raw_route, default_tier, source="config") for raw_route in _route_array(raw_interactions.get("route", ()), "site.interactions.route") ) return InteractionConfig( @@ -180,7 +181,12 @@ def _parse_interactions( ) -def _parse_route_policy(raw_route: dict[str, Any], default_tier: str) -> RoutePolicyConfig: +def _parse_route_policy( + raw_route: dict[str, Any], + default_tier: str, + *, + source: str, +) -> RoutePolicyConfig: if not isinstance(raw_route, dict): raise ValueError("route policy entries must be objects") return RoutePolicyConfig( @@ -190,6 +196,7 @@ def _parse_route_policy(raw_route: dict[str, Any], default_tier: str) -> RoutePo payments=str(raw_route.get("payments", "none")), realtime=bool(raw_route.get("realtime", False)), public_allowed=bool(raw_route.get("public_allowed", True)), + source=source, ) @@ -215,6 +222,8 @@ def _load_route_manifest(path: Path, site_id: str) -> dict[str, Any]: raise ValueError( f"{site_id}: route manifest schema must be {ROUTE_MANIFEST_SCHEMA!r}: {path}" ) + if "routes" not in data: + raise ValueError(f"{site_id}: route manifest missing required routes array: {path}") return data diff --git a/ampg/manifest.py b/ampg/manifest.py index 1c07f0a..916442a 100644 --- a/ampg/manifest.py +++ b/ampg/manifest.py @@ -6,25 +6,10 @@ from pathlib import Path from typing import Any from .config import GatewayConfig, ProtocolConfig, RoutePolicyConfig, SiteConfig +from .route_policy import fixture_path, route_exposed SCHEMA = "ampg.fixture-manifest.v1" -TIER_ORDER = { - "static": 0, - "interactive-lite": 1, - "identity": 2, - "transactional": 3, - "realtime": 4, - "internal": 5, -} -DEFAULT_MAX_TIERS = { - "clearnet": "realtime", - "tor": "transactional", - "i2p": "transactional", - "gemini": "interactive-lite", - "ipfs": "static", - "reticulum": "interactive-lite", -} @dataclass(frozen=True) @@ -43,7 +28,7 @@ def fixture_manifest(site: SiteConfig) -> dict[str, Any]: fixtures.extend( _fixture_entry(site, protocol, route_policy) for route_policy in site.interactions.routes - if _route_is_public(route_policy) and _route_supported(protocol, route_policy) + if route_exposed(site, protocol, route_policy) ) return { "schema": SCHEMA, @@ -102,7 +87,7 @@ def _fixture_entry( if route_policy: entry["route"] = { "match": route_policy.match, - "fixture_path": _fixture_path(route_policy.match), + "fixture_path": fixture_path(route_policy.match), } return entry @@ -113,22 +98,6 @@ def _transport_for(protocol_name: str) -> str: return protocol_name -def _route_is_public(route_policy: RoutePolicyConfig) -> bool: - return route_policy.public_allowed and route_policy.tier != "internal" - - -def _route_supported(protocol: ProtocolConfig, route_policy: RoutePolicyConfig) -> bool: - max_tier = str(protocol.options.get("max_tier", DEFAULT_MAX_TIERS.get(protocol.name, "static"))) - return _tier_rank(route_policy.tier) <= _tier_rank(max_tier) - - -def _tier_rank(tier: str) -> int: - try: - return TIER_ORDER[tier] - except KeyError as exc: - raise ValueError(f"unsupported interaction tier {tier!r}") from exc - - def _interaction_policy( protocol: ProtocolConfig, route_policy: RoutePolicyConfig | None, @@ -153,26 +122,16 @@ def _interaction_policy( def _url_for_route(base_url: str, route_policy: RoutePolicyConfig | None) -> str: if not route_policy: return base_url - fixture_path = _fixture_path(route_policy.match) + route_fixture_path = fixture_path(route_policy.match) if base_url.startswith(("rns://", "lxmf://", "nomad://", "ipfs://", "ipns://")): return base_url - return base_url.rstrip("/") + fixture_path + return base_url.rstrip("/") + route_fixture_path def _output_path_for_route(protocol_name: str, route_policy: RoutePolicyConfig | None) -> str: if not route_policy: return protocol_name - return protocol_name + _fixture_path(route_policy.match) - - -def _fixture_path(match: str) -> str: - value = match.strip() - if not value.startswith("/"): - value = "/" + value - value = value.replace("*", "") - if not value or value == "/": - return "/" - return value if value.endswith("/") else value + "/" + return protocol_name + fixture_path(route_policy.match) def _fixture_url(site: SiteConfig, protocol: ProtocolConfig) -> tuple[str, str]: diff --git a/ampg/route_policy.py b/ampg/route_policy.py new file mode 100644 index 0000000..07f7f54 --- /dev/null +++ b/ampg/route_policy.py @@ -0,0 +1,185 @@ +from __future__ import annotations + +from dataclasses import dataclass +from fnmatch import fnmatchcase + +from .config import GatewayConfig, ProtocolConfig, RoutePolicyConfig, SiteConfig + + +TIER_ORDER = { + "static": 0, + "interactive-lite": 1, + "identity": 2, + "transactional": 3, + "realtime": 4, + "internal": 5, +} +DEFAULT_MAX_TIERS = { + "clearnet": "realtime", + "tor": "transactional", + "i2p": "transactional", + "gemini": "interactive-lite", + "ipfs": "static", + "reticulum": "interactive-lite", +} + + +@dataclass(frozen=True) +class RouteExposure: + site_id: str + protocol: str + route_index: int + match: str + source: str + tier: str + identity: str + payments: str + realtime: bool + public_allowed: bool + max_tier: str + status: str + reason: str + + +@dataclass(frozen=True) +class RouteIssue: + site_id: str + route_index: int + match: str + source: str + tier: str + code: str + message: str + + +def route_exposures(config: GatewayConfig) -> list[RouteExposure]: + exposures: list[RouteExposure] = [] + for site in config.sites: + enabled_protocols = [protocol for protocol in site.protocols.values() if protocol.enabled] + for route_index, route_policy in enumerate(site.interactions.routes): + for protocol in enabled_protocols: + status, reason = route_decision(site, protocol, route_policy) + exposures.append( + RouteExposure( + site_id=site.id, + protocol=protocol.name, + route_index=route_index, + match=route_policy.match, + source=route_policy.source, + tier=route_policy.tier, + identity=route_policy.identity, + payments=route_policy.payments, + realtime=route_policy.realtime, + public_allowed=route_policy.public_allowed, + max_tier=protocol_max_tier(protocol), + status=status, + reason=reason, + ) + ) + return exposures + + +def route_issues(config: GatewayConfig) -> list[RouteIssue]: + issues: list[RouteIssue] = [] + exposures = route_exposures(config) + by_site_route_index = { + (exposure.site_id, exposure.route_index): [] + for exposure in exposures + } + for exposure in exposures: + by_site_route_index[(exposure.site_id, exposure.route_index)].append(exposure) + + for site in config.sites: + for route_index, route_policy in enumerate(site.interactions.routes): + if not route_has_public_intent(site, route_policy): + continue + decisions = by_site_route_index.get((site.id, route_index), []) + if any(decision.status == "exposed" for decision in decisions): + continue + issues.append( + RouteIssue( + site_id=site.id, + route_index=route_index, + match=route_policy.match, + source=route_policy.source, + tier=route_policy.tier, + code="no-compatible-protocol", + message="public route is not exposed by any enabled protocol", + ) + ) + return issues + + +def route_exposed(site: SiteConfig, protocol: ProtocolConfig, route_policy: RoutePolicyConfig) -> bool: + status, _reason = route_decision(site, protocol, route_policy) + return status == "exposed" + + +def route_decision( + site: SiteConfig, + protocol: ProtocolConfig, + route_policy: RoutePolicyConfig, +) -> tuple[str, str]: + if route_policy.tier == "internal": + return "skipped", "tier is internal" + if not route_policy.public_allowed: + return "skipped", "public_allowed=false" + + deny_route = denied_by(site, route_policy) + if deny_route: + return "skipped", f"matched deny_route {deny_route}" + + max_tier = protocol_max_tier(protocol) + if tier_rank(route_policy.tier) > tier_rank(max_tier): + return "skipped", f"tier {route_policy.tier} exceeds protocol max_tier {max_tier}" + + return "exposed", "public route within protocol max_tier" + + +def route_has_public_intent(site: SiteConfig, route_policy: RoutePolicyConfig) -> bool: + return ( + route_policy.tier != "internal" + and route_policy.public_allowed + and denied_by(site, route_policy) is None + ) + + +def denied_by(site: SiteConfig, route_policy: RoutePolicyConfig) -> str | None: + route_variants = _pattern_variants(route_policy.match) + for deny_route in site.interactions.deny_routes: + deny_pattern = _normalize_pattern(deny_route) + if any(fnmatchcase(route_variant, deny_pattern) for route_variant in route_variants): + return deny_route + return None + + +def protocol_max_tier(protocol: ProtocolConfig) -> str: + return str(protocol.options.get("max_tier", DEFAULT_MAX_TIERS.get(protocol.name, "static"))) + + +def tier_rank(tier: str) -> int: + try: + return TIER_ORDER[tier] + except KeyError as exc: + raise ValueError(f"unsupported interaction tier {tier!r}") from exc + + +def fixture_path(match: str) -> str: + value = _normalize_pattern(match).replace("*", "") + if not value or value == "/": + return "/" + return value if value.endswith("/") else value + "/" + + +def _pattern_variants(pattern: str) -> tuple[str, ...]: + normalized = _normalize_pattern(pattern) + fixture = fixture_path(pattern) + trimmed = fixture.rstrip("/") or "/" + return tuple(dict.fromkeys((normalized, fixture, trimmed))) + + +def _normalize_pattern(pattern: str) -> str: + value = pattern.strip() + if not value.startswith("/"): + value = "/" + value + return value diff --git a/docs/fixture-manifests.md b/docs/fixture-manifests.md index 5ba43a5..c6b0e65 100644 --- a/docs/fixture-manifests.md +++ b/docs/fixture-manifests.md @@ -60,8 +60,8 @@ Fixture interaction policy defaults to: Protocol-level options can declare the base fixture policy that AMPB should check. When `[[site.interactions.route]]` entries are configured, AMPG emits additional fixtures -for public route groups. Routes with `tier = "internal"` or `public_allowed = false` are -not emitted. +for public route groups. Routes with `tier = "internal"`, `public_allowed = false`, or a +match covered by `deny_routes` are not emitted. Routes imported from `site.interactions.route_manifest` are treated the same way as inline TOML routes. diff --git a/docs/interactive-apps.md b/docs/interactive-apps.md index eb95217..e66b4d8 100644 --- a/docs/interactive-apps.md +++ b/docs/interactive-apps.md @@ -66,6 +66,22 @@ Optional top-level fields: - `default_tier`: inherited by route entries without a `tier`. - `deny_routes`: route patterns that must not be published. +## Route Exposure Checks + +Use route checks before publishing an application target: + +```sh +python3 -m ampg --config gateway.toml routes explain +python3 -m ampg --config gateway.toml routes validate +``` + +`routes explain` prints one decision per route and enabled protocol. Decisions include the +route source, tier, protocol `max_tier`, status, and reason. + +`routes validate` exits nonzero when a public, non-denied route has no compatible enabled +protocol. Routes marked `internal`, `public_allowed = false`, or matched by +`deny_routes` are treated as intentionally unpublished. + ## Protocol Mapping | Tier | Clearnet | Tor/I2P | Gemini | IPFS | Reticulum | diff --git a/tests/test_route_policy.py b/tests/test_route_policy.py new file mode 100644 index 0000000..bcd70d5 --- /dev/null +++ b/tests/test_route_policy.py @@ -0,0 +1,169 @@ +import contextlib +import io +import tempfile +import unittest +from pathlib import Path + +from ampg.cli import main +from ampg.config import load_config +from ampg.route_policy import route_exposures, route_issues + + +class RoutePolicyTest(unittest.TestCase): + def test_route_exposure_explains_exposed_denied_and_tier_skips(self): + with tempfile.TemporaryDirectory() as tmp: + config = load_config(_write_app_config(Path(tmp))) + + exposures = route_exposures(config) + by_protocol_route = { + (exposure.protocol, exposure.match): exposure + for exposure in exposures + } + + self.assertEqual("route-manifest", by_protocol_route[("tor", "/play/*")].source) + self.assertEqual("exposed", by_protocol_route[("tor", "/play/*")].status) + self.assertEqual("exposed", by_protocol_route[("gemini", "/play/*")].status) + self.assertEqual("skipped", by_protocol_route[("ipfs", "/play/*")].status) + self.assertEqual( + "tier interactive-lite exceeds protocol max_tier static", + by_protocol_route[("ipfs", "/play/*")].reason, + ) + + self.assertEqual("skipped", by_protocol_route[("tor", "/checkout/*")].status) + self.assertEqual( + "tier transactional exceeds protocol max_tier interactive-lite", + by_protocol_route[("tor", "/checkout/*")].reason, + ) + + self.assertEqual("skipped", by_protocol_route[("tor", "/admin/*")].status) + self.assertEqual( + "matched deny_route /admin/*", + by_protocol_route[("tor", "/admin/*")].reason, + ) + + issues = route_issues(config) + self.assertEqual(["/checkout/*"], [issue.match for issue in issues]) + self.assertEqual("no-compatible-protocol", issues[0].code) + + def test_route_validate_exits_nonzero_for_unsupported_public_route(self): + with tempfile.TemporaryDirectory() as tmp: + config_path = _write_app_config(Path(tmp)) + stdout = io.StringIO() + + with contextlib.redirect_stdout(stdout): + status = main(["--config", str(config_path), "routes", "validate"]) + + output = stdout.getvalue() + self.assertEqual(1, status) + self.assertIn("AMPG_ROUTE site=app protocol=tor route_index=0 route=\"/play/*\"", output) + self.assertIn("AMPG_ROUTE_ISSUE site=app route_index=1 route=\"/checkout/*\"", output) + self.assertIn("AMPG_ROUTE_SUMMARY sites=1 routes=3 decisions=9 exposed=2 skipped=7 issues=1", output) + + def test_route_explain_exits_zero_with_issues(self): + with tempfile.TemporaryDirectory() as tmp: + config_path = _write_app_config(Path(tmp)) + stdout = io.StringIO() + + with contextlib.redirect_stdout(stdout): + status = main(["--config", str(config_path), "routes", "explain"]) + + self.assertEqual(0, status) + self.assertIn("issues=1", stdout.getvalue()) + + def test_route_manifest_requires_routes_array(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + source = root / "site" + source.mkdir() + (source / "index.html").write_text("

App

", encoding="utf-8") + (root / "routes.json").write_text( + '{"schema": "ampg.route-manifest.v1"}', + encoding="utf-8", + ) + config_path = root / "gateway.toml" + config_path.write_text( + """ +[[site]] +id = "app" +domain = "app.test" + +[site.source] +kind = "static-html" +path = "./site" + +[site.outputs] +root = "./out" +plan_root = "./plan" + +[site.interactions] +route_manifest = "./routes.json" + +[site.protocols.tor] +enabled = true +renderer = "privacy-html" +""", + encoding="utf-8", + ) + + with self.assertRaisesRegex(ValueError, "missing required routes array"): + load_config(config_path) + + +def _write_app_config(root: Path) -> Path: + source = root / "site" + source.mkdir() + (source / "index.html").write_text("

App

", encoding="utf-8") + (root / "routes.json").write_text( + """ +{ + "schema": "ampg.route-manifest.v1", + "default_tier": "interactive-lite", + "deny_routes": ["/admin/*"], + "routes": [ + {"match": "/play/*"}, + {"match": "/checkout/*", "tier": "transactional"}, + {"match": "/admin/*", "tier": "static"} + ] +} +""", + encoding="utf-8", + ) + config_path = root / "gateway.toml" + config_path.write_text( + """ +[[site]] +id = "app" +domain = "app.test" + +[site.source] +kind = "static-html" +path = "./site" + +[site.outputs] +root = "./out" +plan_root = "./plan" + +[site.interactions] +route_manifest = "./routes.json" + +[site.protocols.tor] +enabled = true +renderer = "privacy-html" +max_tier = "interactive-lite" + +[site.protocols.gemini] +enabled = true +renderer = "gemtext" + +[site.protocols.ipfs] +enabled = true +renderer = "clearnet" +cid = "bafyexample" +""", + encoding="utf-8", + ) + return config_path + + +if __name__ == "__main__": + unittest.main()