Merge remote-tracking branch 'origin/main' into codex/mcp-federation-runtime-20260715

This commit is contained in:
ogt
2026-07-15 08:37:37 +08:00
14 changed files with 565 additions and 32 deletions

View File

@@ -13,6 +13,7 @@ import json
import re
from collections.abc import Mapping
from dataclasses import dataclass
from datetime import UTC, datetime
from enum import Enum
from pathlib import Path
from typing import Any
@@ -337,6 +338,92 @@ def load_canonical_telegram_routing_registry(
return payload
def build_canonical_telegram_routing_runtime_readback(
security_dir: Path | None = None,
) -> dict[str, Any]:
"""Project the committed registry through the live fail-closed resolver.
The source snapshot deliberately records the no-send boundaries of the run
that created it. Once the gateway guard is deployed, that historical
``source_status`` must not be presented as the current runtime state. This
readback resolves every route with the same code path used before Telegram
provider calls and returns symbolic aliases only.
"""
registry = load_canonical_telegram_routing_registry(security_dir)
resolved_routes: list[dict[str, Any]] = []
for route in registry["routes"]:
decision = resolve_canonical_telegram_route(
route["product_id"],
route["signal_family"],
route["severity"][0],
requested_destination=route["allowed_destination"],
registry=registry,
)
resolved_routes.append(
{
**route,
"decision": decision["decision"],
"decision_reason": decision["reason"],
}
)
allowed_route_count = sum(
route["decision"] == "allow" for route in resolved_routes
)
blocked_route_count = len(resolved_routes) - allowed_route_count
unknown_probe = resolve_canonical_telegram_route(
"unknown-product",
"unknown-signal",
"P1",
registry=registry,
)
return {
"schema_version": "telegram_canonical_routing_runtime_readback_v1",
"status": "runtime_policy_enforced_product_delivery_receipts_partial",
"source_status": registry["status"],
"source_mode": registry["mode"],
"source_generated_at": registry["generated_at"],
"runtime_generated_at": datetime.now(UTC).isoformat(),
"runtime_enforcement": {
"canonical_resolver_active": True,
"gateway_pre_send_gate_active": True,
"trusted_ingress_context_required": True,
"unknown_route_decision": unknown_probe["decision"],
"unknown_route_reason": unknown_probe["reason"],
"raw_numeric_destination_count": registry["rollups"][
"raw_numeric_destination_count"
],
"destination_identity_exposed": False,
"all_product_delivery_receipts_ready": blocked_route_count == 0,
},
"rollups": {
**registry["rollups"],
"runtime_allowed_route_count": allowed_route_count,
"runtime_blocked_route_count": blocked_route_count,
},
"policy": registry["policy"],
"destination_roles": registry["destination_roles"],
"products": registry["products"],
"routes": resolved_routes,
"operation_boundaries": {
"read_only": True,
"telegram_send_performed": False,
"bot_api_call_performed": False,
"runtime_route_changed": False,
"secret_value_read": False,
"raw_identifier_included": False,
},
"source_refs": {
"registry": "docs/security/telegram-canonical-routing-registry.snapshot.json",
"resolver": "apps/api/src/services/notification_matrix.py",
"gateway_gate": "apps/api/src/services/telegram_gateway.py",
"verifier": "apps/api/tests/test_telegram_canonical_routing_registry.py",
},
}
def resolve_canonical_telegram_route(
product_id: str,
signal_family: str,