mirror of
https://github.com/Such-Software/adaptive-multi-protocol-gateway.git
synced 2026-10-17 15:41:41 +02:00
250 lines
8.1 KiB
Python
250 lines
8.1 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .capabilities import IDENTITY_ADAPTERS, INTERACTION_TIERS, PAYMENT_ADAPTERS
|
|
|
|
|
|
ROUTE_MANIFEST_SCHEMA = "ampg.route-manifest.v1"
|
|
ROUTE_MANIFEST_SCHEMA_PATH = Path("schemas/ampg.route-manifest.v1.schema.json")
|
|
|
|
TOP_LEVEL_FIELDS = {"schema", "default_tier", "deny_routes", "routes"}
|
|
ROUTE_FIELDS = {"match", "tier", "identity", "payments", "realtime", "public_allowed"}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RouteManifestIssue:
|
|
path: str
|
|
code: str
|
|
message: str
|
|
|
|
|
|
def load_route_manifest(path: Path, *, site_id: str | None = None) -> dict[str, Any]:
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
except FileNotFoundError as exc:
|
|
label = f"{site_id}: " if site_id else ""
|
|
raise ValueError(f"{label}route manifest not found: {path}") from exc
|
|
except json.JSONDecodeError as exc:
|
|
label = f"{site_id}: " if site_id else ""
|
|
raise ValueError(f"{label}invalid route manifest JSON: {path}: {exc}") from exc
|
|
|
|
issues = validate_route_manifest(data)
|
|
if issues:
|
|
label = f"{site_id}: " if site_id else ""
|
|
joined = "; ".join(f"{issue.path}: {issue.message}" for issue in issues)
|
|
raise ValueError(f"{label}invalid route manifest: {path}: {joined}")
|
|
return data
|
|
|
|
|
|
def validate_route_manifest(data: Any) -> list[RouteManifestIssue]:
|
|
issues: list[RouteManifestIssue] = []
|
|
if not isinstance(data, dict):
|
|
return [
|
|
RouteManifestIssue(
|
|
path="$",
|
|
code="type",
|
|
message="route manifest must be a JSON object",
|
|
)
|
|
]
|
|
|
|
_unknown_fields("$", data, TOP_LEVEL_FIELDS, issues)
|
|
schema = data.get("schema")
|
|
if schema != ROUTE_MANIFEST_SCHEMA:
|
|
issues.append(
|
|
RouteManifestIssue(
|
|
path="$.schema",
|
|
code="schema",
|
|
message=f"must be {ROUTE_MANIFEST_SCHEMA!r}",
|
|
)
|
|
)
|
|
|
|
if "default_tier" in data:
|
|
_enum_field("$.default_tier", data["default_tier"], INTERACTION_TIERS, issues)
|
|
|
|
if "deny_routes" in data:
|
|
_string_array("$.deny_routes", data["deny_routes"], issues, require_slash=True)
|
|
|
|
routes = data.get("routes")
|
|
if routes is None:
|
|
issues.append(
|
|
RouteManifestIssue(
|
|
path="$.routes",
|
|
code="required",
|
|
message="missing required routes array",
|
|
)
|
|
)
|
|
return issues
|
|
if not isinstance(routes, list):
|
|
issues.append(
|
|
RouteManifestIssue(
|
|
path="$.routes",
|
|
code="type",
|
|
message="must be an array",
|
|
)
|
|
)
|
|
return issues
|
|
|
|
for index, route in enumerate(routes):
|
|
_route_entry(f"$.routes[{index}]", route, issues)
|
|
return issues
|
|
|
|
|
|
def route_manifest_json_schema() -> dict[str, Any]:
|
|
return {
|
|
"$schema": "https://json-schema.org/draft/2020-12/schema",
|
|
"$id": "urn:ampg:schema:route-manifest:v1",
|
|
"title": "AMPG Route Manifest",
|
|
"description": "Route policy contract emitted by applications for AMPG.",
|
|
"type": "object",
|
|
"additionalProperties": False,
|
|
"required": ["schema", "routes"],
|
|
"properties": {
|
|
"schema": {"const": ROUTE_MANIFEST_SCHEMA},
|
|
"default_tier": {"$ref": "#/$defs/interactionTier"},
|
|
"deny_routes": {
|
|
"type": "array",
|
|
"items": {"$ref": "#/$defs/routePattern"},
|
|
"default": [],
|
|
},
|
|
"routes": {
|
|
"type": "array",
|
|
"items": {"$ref": "#/$defs/routePolicy"},
|
|
},
|
|
},
|
|
"$defs": {
|
|
"interactionTier": {"enum": list(INTERACTION_TIERS)},
|
|
"identityAdapter": {"enum": list(IDENTITY_ADAPTERS)},
|
|
"paymentAdapter": {"enum": list(PAYMENT_ADAPTERS)},
|
|
"routePattern": {
|
|
"type": "string",
|
|
"minLength": 1,
|
|
"pattern": "^/",
|
|
},
|
|
"routePolicy": {
|
|
"type": "object",
|
|
"additionalProperties": False,
|
|
"required": ["match"],
|
|
"properties": {
|
|
"match": {"$ref": "#/$defs/routePattern"},
|
|
"tier": {"$ref": "#/$defs/interactionTier"},
|
|
"identity": {"$ref": "#/$defs/identityAdapter", "default": "none"},
|
|
"payments": {"$ref": "#/$defs/paymentAdapter", "default": "none"},
|
|
"realtime": {"type": "boolean", "default": False},
|
|
"public_allowed": {"type": "boolean", "default": True},
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def route_manifest_schema_json() -> str:
|
|
return json.dumps(route_manifest_json_schema(), indent=2, sort_keys=True) + "\n"
|
|
|
|
|
|
def _route_entry(path: str, value: Any, issues: list[RouteManifestIssue]) -> None:
|
|
if not isinstance(value, dict):
|
|
issues.append(
|
|
RouteManifestIssue(path=path, code="type", message="route entry must be an object")
|
|
)
|
|
return
|
|
_unknown_fields(path, value, ROUTE_FIELDS, issues)
|
|
if "match" not in value:
|
|
issues.append(
|
|
RouteManifestIssue(
|
|
path=f"{path}.match",
|
|
code="required",
|
|
message="missing required route match",
|
|
)
|
|
)
|
|
else:
|
|
_route_pattern(f"{path}.match", value["match"], issues)
|
|
|
|
if "tier" in value:
|
|
_enum_field(f"{path}.tier", value["tier"], INTERACTION_TIERS, issues)
|
|
if "identity" in value:
|
|
_enum_field(f"{path}.identity", value["identity"], IDENTITY_ADAPTERS, issues)
|
|
if "payments" in value:
|
|
_enum_field(f"{path}.payments", value["payments"], PAYMENT_ADAPTERS, issues)
|
|
if "realtime" in value:
|
|
_bool_field(f"{path}.realtime", value["realtime"], issues)
|
|
if "public_allowed" in value:
|
|
_bool_field(f"{path}.public_allowed", value["public_allowed"], issues)
|
|
|
|
|
|
def _unknown_fields(
|
|
path: str,
|
|
data: dict[str, Any],
|
|
allowed: set[str],
|
|
issues: list[RouteManifestIssue],
|
|
) -> None:
|
|
for field in sorted(set(data) - allowed):
|
|
issues.append(
|
|
RouteManifestIssue(
|
|
path=f"{path}.{field}",
|
|
code="unknown-field",
|
|
message="unknown field",
|
|
)
|
|
)
|
|
|
|
|
|
def _enum_field(
|
|
path: str,
|
|
value: Any,
|
|
allowed: tuple[str, ...],
|
|
issues: list[RouteManifestIssue],
|
|
) -> None:
|
|
if value not in allowed:
|
|
joined = ", ".join(allowed)
|
|
issues.append(
|
|
RouteManifestIssue(
|
|
path=path,
|
|
code="enum",
|
|
message=f"must be one of: {joined}",
|
|
)
|
|
)
|
|
|
|
|
|
def _string_array(
|
|
path: str,
|
|
value: Any,
|
|
issues: list[RouteManifestIssue],
|
|
*,
|
|
require_slash: bool = False,
|
|
) -> None:
|
|
if not isinstance(value, list):
|
|
issues.append(RouteManifestIssue(path=path, code="type", message="must be an array"))
|
|
return
|
|
for index, item in enumerate(value):
|
|
item_path = f"{path}[{index}]"
|
|
if require_slash:
|
|
_route_pattern(item_path, item, issues)
|
|
elif not isinstance(item, str) or not item:
|
|
issues.append(
|
|
RouteManifestIssue(
|
|
path=item_path,
|
|
code="type",
|
|
message="must be a non-empty string",
|
|
)
|
|
)
|
|
|
|
|
|
def _route_pattern(path: str, value: Any, issues: list[RouteManifestIssue]) -> None:
|
|
if not isinstance(value, str) or not value:
|
|
issues.append(
|
|
RouteManifestIssue(path=path, code="type", message="must be a non-empty string")
|
|
)
|
|
return
|
|
if not value.startswith("/"):
|
|
issues.append(
|
|
RouteManifestIssue(path=path, code="pattern", message="must start with /")
|
|
)
|
|
|
|
|
|
def _bool_field(path: str, value: Any, issues: list[RouteManifestIssue]) -> None:
|
|
if not isinstance(value, bool):
|
|
issues.append(RouteManifestIssue(path=path, code="type", message="must be boolean"))
|