"""HMAC-authenticated result envelopes for untrusted remote callbacks/final content.""" from __future__ import annotations import hashlib import hmac import json import secrets from dataclasses import dataclass from datetime import datetime, timezone from typing import Any, Callable, Dict, Mapping, Optional, Tuple from arr_ingestion.contracts import OPAQUE_ID_RE, SHA256_RE from arr_processing.contracts import ProcessingResult, canonical_json_bytes from arr_processing.errors import ProcessingError SIGNED_ENVELOPE_VERSION = "arr-processing-signed-1" SIGNATURE_FIELDS = { "algorithm", "key_id", "issued_at", "nonce", "value", } SIGNED_FIELDS = {"signed_envelope_version", "payload", "signature"} MAX_SIGNED_RESULT_BYTES = 512 * 1024 def _duplicate_rejecting_object(pairs: list[Tuple[str, Any]]) -> Dict[str, Any]: value: Dict[str, Any] = {} for key, item in pairs: if key in value: raise ValueError("duplicate JSON key") value[key] = item return value def _reject_constant(_value: str) -> None: raise ValueError("non-finite JSON number") @dataclass(frozen=True) class VerifiedSignedResult: result: ProcessingResult signed_sha256: str key_id: str issued_at: datetime nonce: str class SignedResultCodec: """Sign and verify canonical result payloads without exposing the shared secret.""" def __init__( self, keys: Mapping[str, bytes], *, signing_key_id: Optional[str] = None, max_age_seconds: int = 600, future_skew_seconds: int = 30, clock: Optional[Callable[[], datetime]] = None, ) -> None: normalized: Dict[str, bytes] = {} for key_id, secret in keys.items(): if ( not isinstance(key_id, str) or not OPAQUE_ID_RE.fullmatch(key_id) or not isinstance(secret, bytes) or len(secret) < 32 ): raise ValueError("result signing key configuration is invalid") normalized[key_id] = bytes(secret) if not normalized: raise ValueError("at least one result signing key is required") if signing_key_id is not None and signing_key_id not in normalized: raise ValueError("signing key identifier is unavailable") if max_age_seconds < 1 or future_skew_seconds < 0: raise ValueError("signature age policy is invalid") self._keys = normalized self._signing_key_id = signing_key_id self._max_age_seconds = max_age_seconds self._future_skew_seconds = future_skew_seconds self._clock = clock or (lambda: datetime.now(timezone.utc)) def sign( self, result: ProcessingResult, *, issued_at: Optional[datetime] = None, nonce: Optional[str] = None, ) -> bytes: if self._signing_key_id is None: raise ValueError("codec is verify-only") validated_result = ProcessingResult.from_dict(result.to_dict()) if validated_result != result: raise ValueError("processing result is not canonical") timestamp = self._normalize_time(issued_at or self._clock()) nonce_value = nonce or ("nonce_" + secrets.token_urlsafe(18)) if not OPAQUE_ID_RE.fullmatch(nonce_value): raise ValueError("signature nonce is invalid") issued_text = timestamp.isoformat(timespec="seconds").replace("+00:00", "Z") payload = validated_result.to_dict() value = self._signature_value( self._keys[self._signing_key_id], payload, issued_text, nonce_value, ) return canonical_json_bytes( { "signed_envelope_version": SIGNED_ENVELOPE_VERSION, "payload": payload, "signature": { "algorithm": "hmac-sha256", "key_id": self._signing_key_id, "issued_at": issued_text, "nonce": nonce_value, "value": value, }, } ) def verify(self, raw: bytes) -> VerifiedSignedResult: if not isinstance(raw, bytes) or not raw or len(raw) > MAX_SIGNED_RESULT_BYTES: raise ProcessingError( "PROCESSING_SIGNATURE_INVALID", "signed processing result is invalid" ) try: text = raw.decode("utf-8", errors="strict") parsed = json.loads( text, object_pairs_hook=_duplicate_rejecting_object, parse_constant=_reject_constant, ) except (UnicodeError, ValueError, json.JSONDecodeError): raise ProcessingError( "PROCESSING_SIGNATURE_INVALID", "signed processing result is invalid" ) from None if not isinstance(parsed, Mapping) or set(parsed) != SIGNED_FIELDS: raise ProcessingError( "PROCESSING_SIGNATURE_INVALID", "signed processing result is invalid" ) if parsed.get("signed_envelope_version") != SIGNED_ENVELOPE_VERSION: raise ProcessingError( "PROCESSING_SIGNATURE_INVALID", "signed envelope version is unsupported" ) signature = parsed.get("signature") if not isinstance(signature, Mapping) or set(signature) != SIGNATURE_FIELDS: raise ProcessingError( "PROCESSING_SIGNATURE_INVALID", "processing result signature is invalid" ) algorithm = signature.get("algorithm") key_id = signature.get("key_id") issued_text = signature.get("issued_at") nonce = signature.get("nonce") value = signature.get("value") if ( algorithm != "hmac-sha256" or not isinstance(key_id, str) or key_id not in self._keys or not isinstance(issued_text, str) or not isinstance(nonce, str) or not OPAQUE_ID_RE.fullmatch(nonce) or not isinstance(value, str) or not SHA256_RE.fullmatch(value) ): raise ProcessingError( "PROCESSING_SIGNATURE_INVALID", "processing result signature is invalid" ) issued_at = self._parse_time(issued_text) now = self._normalize_time(self._clock()) age = (now - issued_at).total_seconds() if age > self._max_age_seconds or age < -self._future_skew_seconds: raise ProcessingError( "PROCESSING_SIGNATURE_EXPIRED", "processing result signature is outside its window" ) result = ProcessingResult.from_dict(parsed.get("payload")) expected = self._signature_value( self._keys[key_id], result.to_dict(), issued_text, nonce, ) if not hmac.compare_digest(value, expected): raise ProcessingError( "PROCESSING_SIGNATURE_INVALID", "processing result signature is invalid" ) return VerifiedSignedResult( result=result, signed_sha256=hashlib.sha256(raw).hexdigest(), key_id=key_id, issued_at=issued_at, nonce=nonce, ) @staticmethod def _signature_value( secret: bytes, payload: Mapping[str, Any], issued_at: str, nonce: str, ) -> str: signed = b"\n".join( ( SIGNED_ENVELOPE_VERSION.encode("ascii"), issued_at.encode("ascii"), nonce.encode("ascii"), canonical_json_bytes(payload), ) ) return hmac.new(secret, signed, hashlib.sha256).hexdigest() @staticmethod def _parse_time(value: str) -> datetime: try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError: raise ProcessingError( "PROCESSING_SIGNATURE_INVALID", "processing result signature time is invalid" ) from None if parsed.tzinfo is None or parsed.utcoffset() is None: raise ProcessingError( "PROCESSING_SIGNATURE_INVALID", "processing result signature time is invalid" ) return parsed.astimezone(timezone.utc) @staticmethod def _normalize_time(value: datetime) -> datetime: if value.tzinfo is None or value.utcoffset() is None: raise ValueError("signature clock must return a timezone-aware datetime") return value.astimezone(timezone.utc)