"""Immutable, fail-closed contracts for the enterprise edge boundary.""" from __future__ import annotations import hashlib import json import math import re from collections.abc import Mapping from dataclasses import dataclass from dataclasses import field as dataclass_field from datetime import UTC, datetime, timedelta from decimal import Decimal, InvalidOperation from types import MappingProxyType from typing import ClassVar from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey class EdgeContractError(ValueError): """Raised when an edge contract is malformed or ambiguous.""" TASK_OPERATIONS = frozenset( {"collect", "profile", "quality", "lineage", "controlled_query"} ) TASK_CLASSIFICATIONS = frozenset( { "raw", "recent_detail", "restricted", "desensitized_metadata", "statistics", "lineage", "evidence", } ) EVENT_CLASSIFICATIONS = frozenset( { "desensitized_metadata", "statistics", "lineage", "evidence", "health_summary", "diagnostic_summary", } ) CONTRACT_VERSION = 1 MAX_ATTEMPT = 5 PREFLIGHT_MAX_DEPTH = 64 PREFLIGHT_MAX_NODES = 10_000 PREFLIGHT_MAX_STRING_BYTES = 131_072 PREFLIGHT_MAX_TOTAL_BYTES = 1_048_576 _IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,254}$") _SHA256 = re.compile(r"^[0-9a-f]{64}$") _ED25519_SIGNATURE = re.compile(r"^[0-9a-f]{128}$") _RFC3339 = re.compile( r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?(?:Z|[+-]\d{2}:\d{2})$" ) def preflight_json( value: object, *, max_depth: int = PREFLIGHT_MAX_DEPTH, max_nodes: int = PREFLIGHT_MAX_NODES, max_string_bytes: int = PREFLIGHT_MAX_STRING_BYTES, max_total_bytes: int = PREFLIGHT_MAX_TOTAL_BYTES, ) -> None: """Bound a JSON-like graph iteratively before copying or encoding it.""" nodes = 0 estimated_bytes = 0 active: set[int] = set() stack: list[tuple[bool, object, int]] = [(False, value, 0)] try: while stack: exiting, current, depth = stack.pop() if exiting: active.remove(id(current)) continue if depth > max_depth: raise EdgeContractError("contract exceeds the maximum depth") nodes += 1 if nodes > max_nodes: raise EdgeContractError("contract exceeds the maximum node count") if isinstance(current, str): size = len(current.encode("utf-8")) estimated_bytes += size + 16 if estimated_bytes > max_total_bytes: raise EdgeContractError( "contract exceeds the conservative byte limit" ) if size > max_string_bytes: raise EdgeContractError("contract string exceeds the byte limit") elif current is None or isinstance(current, bool): estimated_bytes += 8 elif isinstance(current, int): if current.bit_length() > 4096: raise EdgeContractError("contract integer exceeds the byte limit") estimated_bytes += max(8, current.bit_length() // 3 + 4) elif isinstance(current, float): if not math.isfinite(current): raise EdgeContractError("contract numbers must be finite") estimated_bytes += 32 elif isinstance(current, Mapping): identity = id(current) if identity in active: raise EdgeContractError("contract contains a cycle") active.add(identity) stack.append((True, current, depth)) estimated_bytes += 16 children: list[object] = [] for key, item in current.items(): if not isinstance(key, str): raise EdgeContractError("contract property names must be strings") key_size = len(key.encode("utf-8")) if key_size > max_string_bytes: raise EdgeContractError("contract key exceeds the byte limit") estimated_bytes += key_size + 8 children.append(item) stack.extend((False, item, depth + 1) for item in reversed(children)) elif isinstance(current, (list, tuple)): identity = id(current) if identity in active: raise EdgeContractError("contract contains a cycle") active.add(identity) stack.append((True, current, depth)) estimated_bytes += 16 stack.extend((False, item, depth + 1) for item in reversed(current)) else: raise EdgeContractError("contract contains a non-JSON value") if estimated_bytes > max_total_bytes: raise EdgeContractError("contract exceeds the conservative byte limit") except (MemoryError, RecursionError) as exc: raise EdgeContractError("contract exceeds safe structural limits") from exc def _length_prefixed(tag: bytes, content: bytes) -> bytes: return tag + str(len(content)).encode("ascii") + b":" + content def _number_text(value: int | float) -> str: try: number = Decimal(value if isinstance(value, int) else str(value)) except (InvalidOperation, ValueError) as exc: raise EdgeContractError("contract number is invalid") from exc if not number.is_finite(): raise EdgeContractError("contract numbers must be finite") if number == 0: return "0" normalized = format(number.normalize(), "f") if "." in normalized: normalized = normalized.rstrip("0").rstrip(".") return normalized def _canonical_encode(value: object) -> bytes: if value is None: return b"n" if isinstance(value, bool): return b"b1" if value else b"b0" if isinstance(value, (int, float)): return _length_prefixed(b"d", _number_text(value).encode("ascii")) if isinstance(value, str): return _length_prefixed(b"s", value.encode("utf-8")) if isinstance(value, Mapping): encoded = bytearray(b"m" + str(len(value)).encode("ascii") + b":") for key in sorted(value, key=lambda item: item.encode("utf-8")): key_bytes = key.encode("utf-8") encoded.extend(_length_prefixed(b"k", key_bytes)) encoded.extend(_canonical_encode(value[key])) return bytes(encoded) if isinstance(value, (list, tuple)): encoded = bytearray(b"l" + str(len(value)).encode("ascii") + b":") for item in value: encoded.extend(_canonical_encode(item)) return bytes(encoded) raise EdgeContractError("contract contains a non-JSON value") def canonical_json_bytes(value: object) -> bytes: """Encode the explicit type-tagged, length-prefixed canonical protocol.""" preflight_json(value) try: return _canonical_encode(value) except (MemoryError, RecursionError) as exc: raise EdgeContractError("contract exceeds safe encoding limits") from exc def canonical_sha256(value: object) -> str: return hashlib.sha256(canonical_json_bytes(value)).hexdigest() def strict_json_bytes(value: object) -> bytes: """Serialize validated data as strict transport/storage JSON.""" preflight_json(value) try: return json.dumps( value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False, ).encode("utf-8") except (MemoryError, RecursionError, TypeError, ValueError) as exc: raise EdgeContractError("contract cannot be serialized as strict JSON") from exc def stable_event_id(identity: Mapping[str, object]) -> str: if not isinstance(identity, Mapping): raise EdgeContractError("event identity must be a mapping") normalized = dict(identity) if "occurred_at" in normalized: normalized["occurred_at"] = _timestamp( normalized["occurred_at"], "occurred_at" ) return f"evt_{canonical_sha256(normalized)}" def _strict_mapping(value: object, *, label: str) -> dict[str, object]: if not isinstance(value, Mapping): raise EdgeContractError(f"{label} must be a mapping") preflight_json(value) return dict(value) def _require_exact_properties( value: Mapping[str, object], *, required: frozenset[str], optional: frozenset[str] = frozenset(), ) -> None: keys = frozenset(value) unknown = keys - required - optional missing = required - keys if unknown: raise EdgeContractError( f"contract has unknown properties: {', '.join(sorted(unknown))}" ) if missing: raise EdgeContractError( f"contract is missing properties: {', '.join(sorted(missing))}" ) def _text(value: object, label: str, *, maximum: int = 255) -> str: if ( not isinstance(value, str) or not value or value.strip() != value or len(value.encode("utf-8")) > maximum or "\x00" in value ): raise EdgeContractError(f"{label} is invalid") return value def _identifier(value: object, label: str) -> str: candidate = _text(value, label) if not _IDENTIFIER.fullmatch(candidate): raise EdgeContractError(f"{label} is invalid") return candidate def _attempt(value: object) -> int: if isinstance(value, bool) or not isinstance(value, int) or not 1 <= value <= 5: raise EdgeContractError("attempt must be between 1 and 5") return value def _version(value: object) -> int: if isinstance(value, bool) or value != CONTRACT_VERSION: raise EdgeContractError("contract_version is not supported") return CONTRACT_VERSION def _timestamp(value: object, label: str) -> str: candidate = _text(value, label, maximum=64) if not _RFC3339.fullmatch(candidate): raise EdgeContractError(f"{label} must be an RFC3339 timestamp") try: parsed = datetime.fromisoformat(candidate.replace("Z", "+00:00")) except ValueError as exc: raise EdgeContractError(f"{label} must be an RFC3339 timestamp") from exc if parsed.tzinfo is None or parsed.utcoffset() is None: raise EdgeContractError(f"{label} must be an RFC3339 timestamp") utc = parsed.astimezone(UTC) canonical = utc.strftime("%Y-%m-%dT%H:%M:%S") if utc.microsecond: canonical += f".{utc.microsecond:06d}".rstrip("0") return f"{canonical}Z" def canonical_timestamp(value: object, label: str = "timestamp") -> str: return _timestamp(value, label) def _digest(value: object, label: str) -> str: candidate = _text(value, label, maximum=64) if not _SHA256.fullmatch(candidate): raise EdgeContractError(f"{label} must be a lowercase SHA256 digest") return candidate def _json_copy(value: object) -> object: try: encoded = strict_json_bytes(value) return json.loads(encoded.decode("utf-8")) except (MemoryError, RecursionError, TypeError, ValueError) as exc: raise EdgeContractError("contract cannot be copied as strict JSON") from exc def _deep_freeze(value: object) -> object: if isinstance(value, dict): return MappingProxyType({key: _deep_freeze(item) for key, item in value.items()}) if isinstance(value, list): return tuple(_deep_freeze(item) for item in value) return value def _deep_thaw(value: object) -> object: if isinstance(value, Mapping): return {key: _deep_thaw(item) for key, item in value.items()} if isinstance(value, tuple): return [_deep_thaw(item) for item in value] return value @dataclass(frozen=True, slots=True) class EdgeTaskContract: task_id: str gateway_id: str environment: str network_zone: str purpose: str classification: str operation: str contract_version: int deadline_at: str attempt: int idempotency_key: str policy_digest: str def __post_init__(self) -> None: values = { "task_id": _identifier(self.task_id, "task_id"), "gateway_id": _identifier(self.gateway_id, "gateway_id"), "environment": _identifier(self.environment, "environment"), "network_zone": _identifier(self.network_zone, "network_zone"), "purpose": _identifier(self.purpose, "purpose"), "contract_version": _version(self.contract_version), "deadline_at": _timestamp(self.deadline_at, "deadline_at"), "attempt": _attempt(self.attempt), "idempotency_key": _identifier(self.idempotency_key, "idempotency_key"), "policy_digest": _digest(self.policy_digest, "policy_digest"), } if self.operation not in TASK_OPERATIONS: raise EdgeContractError("task operation is not approved") if self.classification not in TASK_CLASSIFICATIONS: raise EdgeContractError("task classification is invalid") for field, value in values.items(): object.__setattr__(self, field, value) @property def task_type(self) -> str: return self.operation @classmethod def from_mapping(cls, value: Mapping[str, object]) -> EdgeTaskContract: data = _strict_mapping(value, label="task contract") required = frozenset( { "task_id", "gateway_id", "environment", "network_zone", "purpose", "classification", "contract_version", "deadline_at", "attempt", "idempotency_key", "policy_digest", } ) _require_exact_properties( data, required=required, optional=frozenset({"task_type", "operation"}), ) if ("task_type" in data) == ("operation" in data): raise EdgeContractError( "task contract requires exactly one operation property" ) return cls( task_id=data["task_id"], # type: ignore[arg-type] gateway_id=data["gateway_id"], # type: ignore[arg-type] environment=data["environment"], # type: ignore[arg-type] network_zone=data["network_zone"], # type: ignore[arg-type] purpose=data["purpose"], # type: ignore[arg-type] classification=data["classification"], # type: ignore[arg-type] operation=data.get("task_type", data.get("operation")), # type: ignore[arg-type] contract_version=data["contract_version"], # type: ignore[arg-type] deadline_at=data["deadline_at"], # type: ignore[arg-type] attempt=data["attempt"], # type: ignore[arg-type] idempotency_key=data["idempotency_key"], # type: ignore[arg-type] policy_digest=data["policy_digest"], # type: ignore[arg-type] ) def to_mapping(self) -> dict[str, object]: return { "task_id": self.task_id, "gateway_id": self.gateway_id, "environment": self.environment, "network_zone": self.network_zone, "purpose": self.purpose, "classification": self.classification, "task_type": self.operation, "contract_version": self.contract_version, "deadline_at": self.deadline_at, "attempt": self.attempt, "idempotency_key": self.idempotency_key, "policy_digest": self.policy_digest, } @property def digest(self) -> str: return canonical_sha256(self.to_mapping()) @dataclass(frozen=True, slots=True) class SignedTaskEnvelope: """Control-plane authority over an exact, context-bound task contract.""" task: EdgeTaskContract authority_key_id: str signature_algorithm: str contract_digest: str gateway_id: str environment: str network_zone: str policy_digest: str purpose: str issued_at: str expires_at: str signature: str _verified: bool = dataclass_field(default=False, init=False, repr=False, compare=False) _UNSIGNED_FIELDS: ClassVar[frozenset[str]] = frozenset( { "task", "authority_key_id", "signature_algorithm", "contract_digest", "gateway_id", "environment", "network_zone", "policy_digest", "purpose", "issued_at", "expires_at", } ) def __post_init__(self) -> None: if not isinstance(self.task, EdgeTaskContract): raise EdgeContractError("signed task envelope requires a task contract") normalized_task = EdgeTaskContract.from_mapping(self.task.to_mapping()) unsigned = { "task": normalized_task.to_mapping(), "authority_key_id": self.authority_key_id, "signature_algorithm": self.signature_algorithm, "contract_digest": self.contract_digest, "gateway_id": self.gateway_id, "environment": self.environment, "network_zone": self.network_zone, "policy_digest": self.policy_digest, "purpose": self.purpose, "issued_at": self.issued_at, "expires_at": self.expires_at, } self.canonical_unsigned_bytes(unsigned) if not isinstance(self.signature, str) or not _ED25519_SIGNATURE.fullmatch(self.signature): raise EdgeContractError("signature must be a lowercase Ed25519 signature") object.__setattr__(self, "task", normalized_task) @classmethod def canonical_unsigned_bytes(cls, value: Mapping[str, object]) -> bytes: data = _strict_mapping(value, label="signed task envelope") _require_exact_properties(data, required=cls._UNSIGNED_FIELDS) task_raw = data["task"] if isinstance(task_raw, EdgeTaskContract): task_raw = task_raw.to_mapping() task = EdgeTaskContract.from_mapping(task_raw) # type: ignore[arg-type] normalized = { "task": task.to_mapping(), "authority_key_id": _identifier(data["authority_key_id"], "authority_key_id"), "signature_algorithm": data["signature_algorithm"], "contract_digest": _digest(data["contract_digest"], "contract_digest"), "gateway_id": _identifier(data["gateway_id"], "gateway_id"), "environment": _identifier(data["environment"], "environment"), "network_zone": _identifier(data["network_zone"], "network_zone"), "policy_digest": _digest(data["policy_digest"], "policy_digest"), "purpose": _identifier(data["purpose"], "purpose"), "issued_at": _timestamp(data["issued_at"], "issued_at"), "expires_at": _timestamp(data["expires_at"], "expires_at"), } if normalized["signature_algorithm"] != "Ed25519": raise EdgeContractError("signature_algorithm is not supported") for field in ("gateway_id", "environment", "network_zone", "policy_digest", "purpose"): if normalized[field] != task.to_mapping()[field]: raise EdgeContractError(f"signed authority {field} does not bind the task") if normalized["contract_digest"] != task.digest: raise EdgeContractError("contract_digest does not bind the task") issued = datetime.fromisoformat(str(normalized["issued_at"]).replace("Z", "+00:00")) expires = datetime.fromisoformat(str(normalized["expires_at"]).replace("Z", "+00:00")) deadline = datetime.fromisoformat(task.deadline_at.replace("Z", "+00:00")) if expires <= issued or expires > deadline: raise EdgeContractError("signed authority validity window is invalid") return canonical_json_bytes(normalized) @classmethod def verify_mapping( cls, value: Mapping[str, object], *, authority_keys: Mapping[str, bytes | Ed25519PublicKey], now: datetime, allowed_future_skew_seconds: int = 0, ) -> SignedTaskEnvelope: data = _strict_mapping(value, label="signed task envelope") _require_exact_properties(data, required=cls._UNSIGNED_FIELDS | {"signature"}) if not isinstance(now, datetime) or now.tzinfo is None or now.utcoffset() is None: raise EdgeContractError("verification time must be timezone-aware") if ( isinstance(allowed_future_skew_seconds, bool) or not isinstance(allowed_future_skew_seconds, int) or not 0 <= allowed_future_skew_seconds <= 300 ): raise EdgeContractError("task authority clock skew is invalid") signature = _text(data["signature"], "signature", maximum=128) if not _ED25519_SIGNATURE.fullmatch(signature): raise EdgeContractError("signature must be a lowercase Ed25519 signature") unsigned = {key: data[key] for key in cls._UNSIGNED_FIELDS} signed_bytes = cls.canonical_unsigned_bytes(unsigned) key_id = _identifier(data["authority_key_id"], "authority_key_id") key = authority_keys.get(key_id) if key is None: raise EdgeContractError("task authority key is not trusted") try: public_key = ( key if isinstance(key, Ed25519PublicKey) else Ed25519PublicKey.from_public_bytes(bytes(key)) ) public_key.verify(bytes.fromhex(signature), signed_bytes) except (InvalidSignature, TypeError, ValueError) as exc: raise EdgeContractError("task authority signature is invalid") from exc issued_at = _timestamp(data["issued_at"], "issued_at") expires_at = _timestamp(data["expires_at"], "expires_at") instant = now.astimezone(UTC) issued = datetime.fromisoformat(issued_at.replace("Z", "+00:00")) expires = datetime.fromisoformat(expires_at.replace("Z", "+00:00")) if instant + timedelta(seconds=allowed_future_skew_seconds) < issued: raise EdgeContractError("signed task is not yet valid") if instant >= expires: raise EdgeContractError("signed task authority has expired") task = EdgeTaskContract.from_mapping(data["task"]) # type: ignore[arg-type] envelope = cls( task=task, authority_key_id=key_id, signature_algorithm="Ed25519", contract_digest=task.digest, gateway_id=task.gateway_id, environment=task.environment, network_zone=task.network_zone, policy_digest=task.policy_digest, purpose=task.purpose, issued_at=issued_at, expires_at=expires_at, signature=signature, ) object.__setattr__(envelope, "_verified", True) return envelope @property def is_verified(self) -> bool: return self._verified def to_mapping(self) -> dict[str, object]: return { "task": self.task.to_mapping(), "authority_key_id": self.authority_key_id, "signature_algorithm": self.signature_algorithm, "contract_digest": self.contract_digest, "gateway_id": self.gateway_id, "environment": self.environment, "network_zone": self.network_zone, "policy_digest": self.policy_digest, "purpose": self.purpose, "issued_at": self.issued_at, "expires_at": self.expires_at, "signature": self.signature, } @property def digest(self) -> str: return canonical_sha256(self.to_mapping()) @dataclass(frozen=True, slots=True) class EdgeEventContract: event_id: str task_id: str gateway_id: str environment: str network_zone: str purpose: str classification: str contract_version: int occurred_at: str attempt: int idempotency_key: str policy_digest: str payload: Mapping[str, object] def __post_init__(self) -> None: payload = _strict_mapping(self.payload, label="event payload") payload_copy = _json_copy(payload) if not isinstance(payload_copy, dict): raise EdgeContractError("event payload must be a mapping") values = { "task_id": _identifier(self.task_id, "task_id"), "gateway_id": _identifier(self.gateway_id, "gateway_id"), "environment": _identifier(self.environment, "environment"), "network_zone": _identifier(self.network_zone, "network_zone"), "purpose": _identifier(self.purpose, "purpose"), "contract_version": _version(self.contract_version), "occurred_at": _timestamp(self.occurred_at, "occurred_at"), "attempt": _attempt(self.attempt), "idempotency_key": _identifier(self.idempotency_key, "idempotency_key"), "policy_digest": _digest(self.policy_digest, "policy_digest"), } if self.classification not in EVENT_CLASSIFICATIONS: raise EdgeContractError("event classification is not approved") identity = { "task_id": values["task_id"], "gateway_id": values["gateway_id"], "environment": values["environment"], "network_zone": values["network_zone"], "purpose": values["purpose"], "classification": self.classification, "contract_version": values["contract_version"], "occurred_at": values["occurred_at"], "attempt": values["attempt"], "idempotency_key": values["idempotency_key"], "policy_digest": values["policy_digest"], "payload": payload_copy, } expected_id = stable_event_id(identity) if self.event_id != expected_id: raise EdgeContractError("event_id is not the stable event identifier") object.__setattr__(self, "event_id", expected_id) object.__setattr__(self, "payload", _deep_freeze(payload_copy)) for field, value in values.items(): object.__setattr__(self, field, value) @classmethod def from_mapping(cls, value: Mapping[str, object]) -> EdgeEventContract: data = _strict_mapping(value, label="event contract") required = frozenset( { "event_id", "task_id", "gateway_id", "environment", "network_zone", "purpose", "classification", "contract_version", "occurred_at", "attempt", "idempotency_key", "policy_digest", "payload", } ) _require_exact_properties(data, required=required) return cls(**data) # type: ignore[arg-type] def to_mapping(self) -> dict[str, object]: return { "event_id": self.event_id, "task_id": self.task_id, "gateway_id": self.gateway_id, "environment": self.environment, "network_zone": self.network_zone, "purpose": self.purpose, "classification": self.classification, "contract_version": self.contract_version, "occurred_at": self.occurred_at, "attempt": self.attempt, "idempotency_key": self.idempotency_key, "policy_digest": self.policy_digest, "payload": _deep_thaw(self.payload), } @property def digest(self) -> str: return canonical_sha256(self.to_mapping())