diff --git a/ampbrowser/cli.py b/ampbrowser/cli.py index cb6051d..50f35b0 100644 --- a/ampbrowser/cli.py +++ b/ampbrowser/cli.py @@ -190,6 +190,11 @@ def _cmd_fixture(args) -> int: f"actual_transport={check.actual_transport} " f"expected_profile={check.expected_profile} " f"actual_profile={check.actual_profile} " + f"tier={check.tier} " + f"identity={check.identity} " + f"payments={check.payments} " + f"realtime={str(check.realtime).lower()} " + f"public_allowed={str(check.public_allowed).lower()} " f"status={check.status} " f"message=\"{_safe(check.message)}\"" ) diff --git a/ampbrowser/fixtures.py b/ampbrowser/fixtures.py index 7c51799..4cc2637 100644 --- a/ampbrowser/fixtures.py +++ b/ampbrowser/fixtures.py @@ -5,6 +5,7 @@ import json from pathlib import Path from typing import Any +from .interaction_policy import interaction_failures from .routing import route_url @@ -20,6 +21,11 @@ class FixtureCheck: actual_transport: str expected_profile: str actual_profile: str + tier: str + identity: str + payments: str + realtime: bool + public_allowed: bool status: str message: str @@ -70,6 +76,7 @@ def _check_fixture(site_id: str, fixture: Any, path: Path) -> FixtureCheck: raise ValueError(f"{path}: {protocol}: missing checks table") expected_transport = _string(checks, "transport") expected_profile = _string(checks, "profile") + interaction = _interaction_policy(fixture) route = route_url(url) failures: list[str] = [] @@ -77,6 +84,16 @@ def _check_fixture(site_id: str, fixture: Any, path: Path) -> FixtureCheck: failures.append(f"transport expected {expected_transport} got {route.transport}") if route.profile != expected_profile: failures.append(f"profile expected {expected_profile} got {route.profile}") + failures.extend( + interaction_failures( + expected_transport, + tier=interaction["tier"], + identity=interaction["identity"], + payments=interaction["payments"], + realtime=interaction["realtime"], + public_allowed=interaction["public_allowed"], + ) + ) return FixtureCheck( site_id=site_id, @@ -86,6 +103,11 @@ def _check_fixture(site_id: str, fixture: Any, path: Path) -> FixtureCheck: actual_transport=route.transport, expected_profile=expected_profile, actual_profile=route.profile, + tier=interaction["tier"], + identity=interaction["identity"], + payments=interaction["payments"], + realtime=interaction["realtime"], + public_allowed=interaction["public_allowed"], status="fail" if failures else "ok", message="; ".join(failures) if failures else "route matches", ) @@ -103,3 +125,30 @@ def _string(data: dict[str, Any], key: str) -> str: if not isinstance(value, str) or not value: raise ValueError(f"missing required string field {key!r}") return value + + +def _interaction_policy(fixture: dict[str, Any]) -> dict[str, Any]: + raw = fixture.get("interaction", {}) + if not isinstance(raw, dict): + raise ValueError(f"{fixture.get('protocol', 'fixture')}: interaction must be a table") + return { + "tier": _optional_string(raw, "tier", "static"), + "identity": _optional_string(raw, "identity", "none"), + "payments": _optional_string(raw, "payments", "none"), + "realtime": _optional_bool(raw, "realtime", False), + "public_allowed": _optional_bool(raw, "public_allowed", True), + } + + +def _optional_string(data: dict[str, Any], key: str, default: str) -> str: + value = data.get(key, default) + if not isinstance(value, str) or not value: + raise ValueError(f"interaction.{key} must be a non-empty string") + return value + + +def _optional_bool(data: dict[str, Any], key: str, default: bool) -> bool: + value = data.get(key, default) + if not isinstance(value, bool): + raise ValueError(f"interaction.{key} must be true or false") + return value diff --git a/ampbrowser/interaction_policy.py b/ampbrowser/interaction_policy.py new file mode 100644 index 0000000..caa61af --- /dev/null +++ b/ampbrowser/interaction_policy.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +from dataclasses import dataclass + + +TIER_ORDER = { + "static": 0, + "interactive-lite": 1, + "identity": 2, + "transactional": 3, + "realtime": 4, + "internal": 5, +} + + +@dataclass(frozen=True) +class TransportInteractionPolicy: + transport: str + max_tier: str + identities: tuple[str, ...] + payments: tuple[str, ...] + realtime: str + + +TRANSPORT_POLICIES = { + "clearnet": TransportInteractionPolicy( + "clearnet", + "realtime", + ("none", "http-session", "siwe"), + ("none", "server-invoice", "wallet-signature"), + "yes", + ), + "tor": TransportInteractionPolicy( + "tor", + "transactional", + ("none", "http-session", "siwe"), + ("none", "server-invoice", "wallet-signature"), + "private-only", + ), + "i2p": TransportInteractionPolicy( + "i2p", + "transactional", + ("none", "http-session", "siwe"), + ("none", "server-invoice", "wallet-signature"), + "private-only", + ), + "gemini": TransportInteractionPolicy( + "gemini", + "interactive-lite", + ("none", "signed-link"), + ("none", "static-instructions"), + "no", + ), + "ipfs": TransportInteractionPolicy( + "ipfs", + "static", + ("none",), + ("none", "static-instructions"), + "no", + ), + "reticulum": TransportInteractionPolicy( + "reticulum", + "interactive-lite", + ("none", "signed-link"), + ("none", "static-instructions"), + "research", + ), +} + + +def interaction_failures( + transport: str, + *, + tier: str, + identity: str, + payments: str, + realtime: bool, + public_allowed: bool, +) -> list[str]: + failures: list[str] = [] + policy = TRANSPORT_POLICIES.get(transport) + if not policy: + return [f"no interaction policy for transport {transport}"] + if not public_allowed: + failures.append("fixture is not public_allowed") + if tier not in TIER_ORDER: + failures.append(f"unknown tier {tier}") + elif TIER_ORDER[tier] > TIER_ORDER[policy.max_tier]: + failures.append(f"tier {tier} exceeds {transport} max {policy.max_tier}") + if identity not in policy.identities: + failures.append(f"identity {identity} not allowed on {transport}") + if payments not in policy.payments: + failures.append(f"payments {payments} not allowed on {transport}") + if realtime and policy.realtime not in {"yes", "private-only", "research"}: + failures.append(f"realtime not allowed on {transport}") + return failures diff --git a/docs/fixture-manifests.md b/docs/fixture-manifests.md index e238c69..b3e4bc3 100644 --- a/docs/fixture-manifests.md +++ b/docs/fixture-manifests.md @@ -3,7 +3,8 @@ > Status: draft | Updated 2026-07-06 | Applies to: AMPB and AMPG compatibility checks AMPB can check AMPG fixture manifests without contacting any network. The check confirms -that each generated fixture URL routes to the expected AMPB transport and profile. +that each generated fixture URL routes to the expected AMPB transport and profile, and +that its declared interaction policy fits that transport. ## Command @@ -11,12 +12,13 @@ that each generated fixture URL routes to the expected AMPB transport and profil python3 -m ampbrowser fixture check ../adaptive-multi-protocol-gateway/dist/wownero/ampg-fixture-manifest.json ``` -The command exits non-zero when any fixture route does not match the manifest. +The command exits non-zero when any fixture route or interaction policy does not match the +manifest. ## Scope -Fixture checks validate route selection only. They do not verify daemon health, hidden -service reachability, IPFS pinning, TLS, or content freshness. +Fixture checks validate route selection and declared interaction policy only. They do not +verify daemon health, hidden service reachability, IPFS pinning, TLS, or content freshness. The manifest must not contain private deployment notes, service keys, credentials, or host inventory. diff --git a/tests/test_fixtures.py b/tests/test_fixtures.py index cdf432b..f147a74 100644 --- a/tests/test_fixtures.py +++ b/tests/test_fixtures.py @@ -32,6 +32,7 @@ class FixtureCheckTest(unittest.TestCase): "protocol": "ipfs", "url": "ipfs://bafyexample", "checks": {"transport": "ipfs", "profile": "ipfs"}, + "interaction": {"tier": "static"}, }, ], } @@ -70,6 +71,58 @@ class FixtureCheckTest(unittest.TestCase): self.assertEqual("fail", result.checks[0].status) self.assertIn("transport expected tor got clearnet", result.checks[0].message) + def test_rejects_transactional_ipfs_fixture(self) -> None: + result = _check_one( + "ipfs", + "ipfs://bafyexample", + {"tier": "transactional", "payments": "server-invoice"}, + ) + + self.assertFalse(result.ok) + self.assertIn("tier transactional exceeds ipfs max static", result.checks[0].message) + self.assertIn("payments server-invoice not allowed on ipfs", result.checks[0].message) + + def test_allows_transactional_tor_fixture(self) -> None: + result = _check_one( + "tor", + "http://example.onion/", + {"tier": "transactional", "identity": "http-session", "payments": "server-invoice"}, + ) + + self.assertTrue(result.ok) + + def test_allows_interactive_lite_gemini_but_rejects_realtime(self) -> None: + ok = _check_one("gemini", "gemini://example.test/", {"tier": "interactive-lite"}) + bad = _check_one("gemini", "gemini://example.test/", {"tier": "realtime", "realtime": True}) + + self.assertTrue(ok.ok) + self.assertFalse(bad.ok) + self.assertIn("tier realtime exceeds gemini max interactive-lite", bad.checks[0].message) + self.assertIn("realtime not allowed on gemini", bad.checks[0].message) + + +def _check_one(transport: str, url: str, interaction: dict) -> object: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "manifest.json" + path.write_text( + json.dumps( + { + "schema": "ampg.fixture-manifest.v1", + "site": {"id": "example", "domain": "example.test"}, + "fixtures": [ + { + "protocol": transport, + "url": url, + "checks": {"transport": transport, "profile": transport}, + "interaction": interaction, + } + ], + } + ), + encoding="utf-8", + ) + return check_fixture_manifest(path) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_interaction_policy.py b/tests/test_interaction_policy.py new file mode 100644 index 0000000..e2ae898 --- /dev/null +++ b/tests/test_interaction_policy.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +import unittest + +from ampbrowser.interaction_policy import interaction_failures + + +class InteractionPolicyTest(unittest.TestCase): + def test_static_ipfs_is_allowed(self) -> None: + self.assertEqual( + [], + interaction_failures( + "ipfs", + tier="static", + identity="none", + payments="none", + realtime=False, + public_allowed=True, + ), + ) + + def test_public_allowed_false_fails(self) -> None: + failures = interaction_failures( + "tor", + tier="static", + identity="none", + payments="none", + realtime=False, + public_allowed=False, + ) + + self.assertEqual(["fixture is not public_allowed"], failures) + + +if __name__ == "__main__": + unittest.main()