"""Public-key verification for server-issued entitlement snapshots.""" from __future__ import annotations import base64 from dataclasses import dataclass from datetime import datetime, timezone from types import MappingProxyType from typing import Any, Mapping, Optional from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from .canonical_json import canonicalize_json, parse_bounded_json from .constants import ( MAX_SIGNED_DOCUMENT_BYTES, PRODUCT_ID, SIGNED_DOCUMENT_DOMAIN, validate_identifier, ) from .errors import LicenseErrorCode, LicensingError from .models import ( EntitlementSnapshot, SignedLicenseDocument, VerifiedEntitlementSnapshot, format_rfc3339, parse_rfc3339, ) ED25519_PUBLIC_KEY_BYTES = 32 ED25519_SIGNATURE_BYTES = 64 @dataclass(frozen=True) class PublicKeyRecord: key_id: str public_key_bytes: bytes not_before: Optional[datetime] = None expires_at: Optional[datetime] = None def __post_init__(self) -> None: object.__setattr__(self, "key_id", validate_identifier(self.key_id, "key_id")) if not isinstance(self.public_key_bytes, bytes): raise TypeError("public_key_bytes must be bytes") if len(self.public_key_bytes) != ED25519_PUBLIC_KEY_BYTES: raise ValueError( f"Ed25519 public key must contain {ED25519_PUBLIC_KEY_BYTES} bytes" ) if self.not_before is not None: object.__setattr__( self, "not_before", parse_rfc3339(format_rfc3339(self.not_before, "not_before"), "not_before"), ) if self.expires_at is not None: object.__setattr__( self, "expires_at", parse_rfc3339(format_rfc3339(self.expires_at, "expires_at"), "expires_at"), ) if ( self.not_before is not None and self.expires_at is not None and self.expires_at <= self.not_before ): raise ValueError("expires_at must be later than not_before") @classmethod def from_base64( cls, key_id: str, public_key_base64: str, *, not_before: Optional[datetime] = None, expires_at: Optional[datetime] = None, ) -> "PublicKeyRecord": if not isinstance(public_key_base64, str): raise TypeError("public_key_base64 must be a string") try: key_bytes = base64.b64decode(public_key_base64.strip(), validate=True) except (ValueError, TypeError) as exc: raise ValueError("public_key_base64 is not valid Base64") from exc return cls( key_id=key_id, public_key_bytes=key_bytes, not_before=not_before, expires_at=expires_at, ) def validate_issuance_time(self, issued_at: datetime) -> None: if not isinstance(issued_at, datetime): raise TypeError("issued_at must be a datetime") normalized = parse_rfc3339( format_rfc3339(issued_at, "issued_at"), "issued_at", ) if self.not_before is not None and normalized < self.not_before: raise LicensingError( LicenseErrorCode.KEY_NOT_YET_VALID, "license was issued before the signing key became active", ) if self.expires_at is not None and normalized >= self.expires_at: raise LicensingError( LicenseErrorCode.KEY_EXPIRED, "license was issued after the signing key retired", ) class PublicKeyRing: def __init__(self, records: tuple[PublicKeyRecord, ...] | list[PublicKeyRecord]) -> None: if not isinstance(records, (tuple, list)): raise TypeError("records must be a tuple or list") normalized: dict[str, PublicKeyRecord] = {} for record in records: if not isinstance(record, PublicKeyRecord): raise TypeError("records must contain PublicKeyRecord values") if record.key_id in normalized: raise ValueError(f"duplicate public signing key ID: {record.key_id}") normalized[record.key_id] = record if not normalized: raise ValueError("public key ring must contain at least one key") self.records = MappingProxyType(normalized) def require(self, key_id: str) -> PublicKeyRecord: normalized_id = validate_identifier(key_id, "key_id") record = self.records.get(normalized_id) if record is None: raise LicensingError( LicenseErrorCode.UNKNOWN_SIGNING_KEY, f"signed license uses unknown key ID: {normalized_id}", ) return record class LicenseVerifier: """Verify signatures and immutable structural claims with public keys only.""" def __init__( self, key_ring: PublicKeyRing, *, expected_product_id: str = PRODUCT_ID, ) -> None: if not isinstance(key_ring, PublicKeyRing): raise TypeError("key_ring must be a PublicKeyRing") self.key_ring = key_ring self.expected_product_id = validate_identifier( expected_product_id, "expected_product_id", ) @staticmethod def signing_bytes(payload: Mapping[str, Any]) -> bytes: if not isinstance(payload, Mapping): raise TypeError("payload must be an object") return SIGNED_DOCUMENT_DOMAIN + canonicalize_json(payload) def parse_document( self, document: str | bytes | bytearray | Mapping[str, Any], ) -> SignedLicenseDocument: if isinstance(document, Mapping): raw = document else: raw = parse_bounded_json(document, maximum_bytes=MAX_SIGNED_DOCUMENT_BYTES) if not isinstance(raw, Mapping): raise LicensingError( LicenseErrorCode.INVALID_DOCUMENT, "signed-license document root must be an object", ) return SignedLicenseDocument.from_mapping(raw) def verify( self, document: str | bytes | bytearray | Mapping[str, Any], *, at: Optional[datetime] = None, ) -> VerifiedEntitlementSnapshot: if at is None: verified_at = datetime.now(timezone.utc) else: if not isinstance(at, datetime): raise TypeError("at must be a datetime or None") verified_at = parse_rfc3339(format_rfc3339(at, "at"), "at") signed = self.parse_document(document) record = self.key_ring.require(signed.key_id) try: signature = base64.b64decode(signed.signature_base64, validate=True) except (TypeError, ValueError) as exc: raise LicensingError( LicenseErrorCode.INVALID_DOCUMENT, "license signature is not valid Base64", ) from exc if len(signature) != ED25519_SIGNATURE_BYTES: raise LicensingError( LicenseErrorCode.INVALID_DOCUMENT, f"Ed25519 signature must contain {ED25519_SIGNATURE_BYTES} bytes", ) try: Ed25519PublicKey.from_public_bytes(record.public_key_bytes).verify( signature, self.signing_bytes(signed.payload), ) except InvalidSignature as exc: raise LicensingError( LicenseErrorCode.INVALID_SIGNATURE, "license signature verification failed", ) from exc snapshot = EntitlementSnapshot.from_payload_mapping(signed.payload) record.validate_issuance_time(snapshot.issued_at) if snapshot.signature_key_id != signed.key_id: raise LicensingError( LicenseErrorCode.INVALID_DOCUMENT, "payload signing key ID does not match the signed envelope", ) if snapshot.product_id != self.expected_product_id: raise LicensingError( LicenseErrorCode.WRONG_PRODUCT, "license is for a different product", ) if verified_at < snapshot.not_before: raise LicensingError( LicenseErrorCode.CLAIM_NOT_YET_VALID, "license is not valid yet", ) return VerifiedEntitlementSnapshot( snapshot=snapshot, verified_at=verified_at, key_id=signed.key_id, ) __all__ = [ "ED25519_PUBLIC_KEY_BYTES", "ED25519_SIGNATURE_BYTES", "LicenseVerifier", "PublicKeyRecord", "PublicKeyRing", ]