Spaces:
Sleeping
Sleeping
File size: 2,265 Bytes
e61e751 8511813 e61e751 8511813 e61e751 8511813 e61e751 8511813 e61e751 8511813 e61e751 8511813 e61e751 8511813 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | """
Federation Node — Trust Resolution (Phase 2 Vault)
==================================================
Contract: C-FED-NODE-001 v0.1.1
Trust tier resolution backed by Cloudflare D1.
"""
from __future__ import annotations
from datetime import datetime, timezone
from typing import Optional
from models import TrustTier, RefusalReason, FederationRefusal
import d1_client
def make_refusal(reason: RefusalReason, message: str, node_seal: str) -> dict:
return FederationRefusal(
refusal=True,
reason=reason,
message=message,
node_seal=node_seal,
timestamp=datetime.now(timezone.utc).isoformat(),
).model_dump()
async def resolve_trust_tier(sender_seal: str) -> TrustTier:
"""Resolve the trust tier for a sender based on D1 registry."""
rs = await d1_client.execute_sql("SELECT tier FROM trust_registry WHERE seal = ?", [sender_seal])
if rs:
tier_str = rs[0]["tier"]
try:
return TrustTier(tier_str)
except ValueError:
pass
return TrustTier.UNKNOWN
async def promote_to_trusted(seal: str, alias: str = ""):
"""Promote a seal to TRUSTED in the D1 registry."""
await d1_client.execute_sql(
'''INSERT INTO trust_registry (seal, alias, tier)
VALUES (?, ?, 'TRUSTED')
ON CONFLICT(seal) DO UPDATE
SET tier='TRUSTED', alias=excluded.alias, last_seen_at=CURRENT_TIMESTAMP''',
[seal, alias]
)
def apply_trust_policy(
tier: TrustTier,
default_action: str,
blocked_behavior: str,
node_seal: str,
) -> Optional[dict]:
if tier == TrustTier.BLOCKED:
if blocked_behavior == "SILENT_DROP":
return {"action": "SILENT_DROP"}
return make_refusal(
RefusalReason.SENDER_BLOCKED,
"Sender is on the blocklist. Refusal Protocol Active.",
node_seal,
)
if tier == TrustTier.TRUSTED:
return None
if default_action == "REFUSE":
return make_refusal(
RefusalReason.SENDER_NOT_TRUSTED,
"Sender not in trusted list. Default policy: REFUSE.",
node_seal,
)
if default_action == "QUARANTINE":
return {"action": "QUARANTINE"}
return None
|