"""Immutable signed-entitlement document models.""" from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timedelta, timezone from enum import Enum from typing import Any, Mapping, Optional from .constants import ( MAX_DISPLAY_TEXT_LENGTH, MAX_ENTITLEMENTS_PER_SNAPSHOT, MAX_GRANT_SUMMARIES_PER_SNAPSHOT, MAX_NONCE_LENGTH, PRODUCT_SURFACE_POLICY_REQUIRED_CATALOG_REVISION, SIGNED_DOCUMENT_ALGORITHM, SIGNED_DOCUMENT_SCHEMA, SIGNED_DOCUMENT_SCHEMA_VERSION, SNAPSHOT_SCHEMA, SNAPSHOT_SCHEMA_VERSION, TRIAL_MAXIMUM_TOTAL_DURATION_HOURS, validate_identifier, ) from .errors import LicenseErrorCode, LicensingError from .product_surfaces import ( PRODUCT_SURFACE_INVENTORY_REVISION, PRODUCT_SURFACE_INVENTORY_SHA256, PRODUCT_SURFACE_KIND_ANALYSIS, PRODUCT_SURFACE_KIND_MAIN_TAB, validate_product_surface_ids, ) class LicenseState(str, Enum): UNINITIALIZED = "uninitialized" TRIAL_PROVISIONAL = "trial_provisional" TRIAL_ACTIVE = "trial_active" LICENSED_ACTIVE = "licensed_active" REFRESH_RECOMMENDED = "refresh_recommended" OFFLINE_GRACE = "offline_grace" SUBSCRIPTION_PAYMENT_GRACE = "subscription_payment_grace" MISSING = "missing" EXPIRED = "expired" SUSPENDED = "suspended" REVOKED = "revoked" REFUNDED = "refunded" WRONG_DEVICE = "wrong_device" SEAT_LIMIT_REACHED = "seat_limit_reached" WRONG_APPLICATION_VERSION = "wrong_application_version" INVALID_DOCUMENT = "invalid_signature_document" SERVICE_UNAVAILABLE = "service_unavailable" CLOCK_ANOMALY = "clock_anomaly" ACCOUNT_ACTION_REQUIRED = "account_action_required" class SubjectType(str, Enum): USER = "user" ORGANIZATION = "organization" DEVICE = "device" LICENSE = "license" def parse_rfc3339(value: object, field_name: str) -> datetime: if not isinstance(field_name, str): raise TypeError("field_name must be a string") label = field_name.strip() if not label: raise ValueError("field_name must not be empty") if not isinstance(value, str): raise TypeError(f"{label} must be an RFC 3339 string") normalized = value.strip() if not normalized: raise ValueError(f"{label} must not be empty") try: parsed = datetime.fromisoformat(normalized.replace("Z", "+00:00")) except ValueError as exc: raise ValueError(f"{label} must be a valid RFC 3339 timestamp") from exc if parsed.tzinfo is None or parsed.utcoffset() is None: raise ValueError(f"{label} must include a timezone") return parsed.astimezone(timezone.utc) def format_rfc3339(value: datetime, field_name: str) -> str: if not isinstance(field_name, str): raise TypeError("field_name must be a string") label = field_name.strip() if not label: raise ValueError("field_name must not be empty") if not isinstance(value, datetime): raise TypeError(f"{label} must be a datetime") if value.tzinfo is None or value.utcoffset() is None: raise ValueError(f"{label} must include a timezone") utc_value = value.astimezone(timezone.utc) if utc_value.microsecond: text = utc_value.isoformat(timespec="microseconds") else: text = utc_value.isoformat(timespec="seconds") return text.replace("+00:00", "Z") def _optional_identifier(value: object, field_name: str) -> Optional[str]: if value is None: return None return validate_identifier(value, field_name) def _display_text(value: object, field_name: str) -> str: if not isinstance(value, str): raise TypeError(f"{field_name} must be a string") normalized = value.strip() if not normalized: raise ValueError(f"{field_name} must not be empty") if len(normalized) > MAX_DISPLAY_TEXT_LENGTH: raise ValueError( f"{field_name} exceeds the maximum length of {MAX_DISPLAY_TEXT_LENGTH}" ) return normalized def _nonnegative_integer(value: object, field_name: str) -> int: if isinstance(value, bool) or not isinstance(value, int): raise TypeError(f"{field_name} must be an integer") if value < 0: raise ValueError(f"{field_name} must not be negative") return value def _positive_integer(value: object, field_name: str) -> int: result = _nonnegative_integer(value, field_name) if result < 1: raise ValueError(f"{field_name} must be at least one") return result @dataclass(frozen=True) class GrantSummary: grant_id: str source: str sku_id: str label: str ends_at: Optional[datetime] = None def __post_init__(self) -> None: object.__setattr__(self, "grant_id", validate_identifier(self.grant_id, "grant_id")) object.__setattr__(self, "source", validate_identifier(self.source, "grant source")) object.__setattr__(self, "sku_id", validate_identifier(self.sku_id, "sku_id")) object.__setattr__(self, "label", _display_text(self.label, "grant label")) if self.ends_at is not None: parse_rfc3339(format_rfc3339(self.ends_at, "ends_at"), "ends_at") def to_mapping(self) -> dict[str, Any]: return { "grantId": self.grant_id, "source": self.source, "skuId": self.sku_id, "label": self.label, "endsAt": ( None if self.ends_at is None else format_rfc3339(self.ends_at, "ends_at") ), } @classmethod def from_mapping(cls, raw: Mapping[str, Any]) -> "GrantSummary": if not isinstance(raw, Mapping): raise TypeError("grant summary must be an object") ends_at = raw.get("endsAt") return cls( grant_id=raw.get("grantId"), source=raw.get("source"), sku_id=raw.get("skuId"), label=raw.get("label"), ends_at=None if ends_at is None else parse_rfc3339(ends_at, "endsAt"), ) @dataclass(frozen=True) class SnapshotDevicePolicy: policy_id: str maximum_devices: int named_seats: int seat_id: Optional[str] = None def __post_init__(self) -> None: object.__setattr__(self, "policy_id", validate_identifier(self.policy_id, "policy_id")) object.__setattr__( self, "maximum_devices", _positive_integer(self.maximum_devices, "maximum_devices"), ) object.__setattr__(self, "named_seats", _positive_integer(self.named_seats, "named_seats")) object.__setattr__(self, "seat_id", _optional_identifier(self.seat_id, "seat_id")) def to_mapping(self) -> dict[str, Any]: return { "policyId": self.policy_id, "maximumDevices": self.maximum_devices, "namedSeats": self.named_seats, "seatId": self.seat_id, } @classmethod def from_mapping(cls, raw: Mapping[str, Any]) -> "SnapshotDevicePolicy": if not isinstance(raw, Mapping): raise TypeError("device policy claim must be an object") return cls( policy_id=raw.get("policyId"), maximum_devices=raw.get("maximumDevices"), named_seats=raw.get("namedSeats"), seat_id=raw.get("seatId"), ) @dataclass(frozen=True) class ProductSurfacePolicy: """Exact signed analysis and primary-workspace inclusion policy.""" analysis_ids: tuple[str, ...] main_tab_ids: tuple[str, ...] profile_id: Optional[str] = None inventory_revision: int = PRODUCT_SURFACE_INVENTORY_REVISION inventory_sha256: str = PRODUCT_SURFACE_INVENTORY_SHA256 def __post_init__(self) -> None: object.__setattr__( self, "analysis_ids", validate_product_surface_ids( PRODUCT_SURFACE_KIND_ANALYSIS, self.analysis_ids, ), ) object.__setattr__( self, "main_tab_ids", validate_product_surface_ids( PRODUCT_SURFACE_KIND_MAIN_TAB, self.main_tab_ids, require_nonempty=True, ), ) object.__setattr__( self, "profile_id", _optional_identifier(self.profile_id, "profile_id"), ) revision = _positive_integer( self.inventory_revision, "inventory_revision", ) if revision != PRODUCT_SURFACE_INVENTORY_REVISION: raise ValueError("product surface inventory revision is not supported") object.__setattr__(self, "inventory_revision", revision) if not isinstance(self.inventory_sha256, str): raise TypeError("inventory_sha256 must be a string") normalized_digest = self.inventory_sha256.strip().lower() if ( len(normalized_digest) != 64 or any(character not in "0123456789abcdef" for character in normalized_digest) ): raise ValueError("inventory_sha256 must be a SHA-256 hexadecimal digest") if normalized_digest != PRODUCT_SURFACE_INVENTORY_SHA256: raise ValueError("product surface inventory digest does not match this app") object.__setattr__(self, "inventory_sha256", normalized_digest) def to_mapping(self) -> dict[str, Any]: return { "profileId": self.profile_id, "inventoryRevision": self.inventory_revision, "inventorySha256": self.inventory_sha256, "analysisIds": list(self.analysis_ids), "mainTabIds": list(self.main_tab_ids), } @classmethod def from_mapping(cls, raw: Mapping[str, Any]) -> "ProductSurfacePolicy": if not isinstance(raw, Mapping): raise TypeError("product surface policy must be an object") analysis_ids = raw.get("analysisIds") main_tab_ids = raw.get("mainTabIds") if not isinstance(analysis_ids, list) or not isinstance(main_tab_ids, list): raise TypeError("product surface policy identifiers must be lists") return cls( profile_id=raw.get("profileId"), inventory_revision=raw.get("inventoryRevision"), inventory_sha256=raw.get("inventorySha256"), analysis_ids=tuple(analysis_ids), main_tab_ids=tuple(main_tab_ids), ) @dataclass(frozen=True) class TrialPolicyClaim: """Signed original trial interval constrained by the shipped client ceiling.""" original_started_at: datetime final_expires_at: datetime def __post_init__(self) -> None: started = parse_rfc3339( format_rfc3339(self.original_started_at, "original_started_at"), "original_started_at", ) expires = parse_rfc3339( format_rfc3339(self.final_expires_at, "final_expires_at"), "final_expires_at", ) if expires <= started: raise ValueError("final_expires_at must be later than original_started_at") maximum_duration = timedelta(hours=TRIAL_MAXIMUM_TOTAL_DURATION_HOURS) if expires - started > maximum_duration: raise ValueError( "trial interval exceeds the maximum duration accepted by this app" ) object.__setattr__(self, "original_started_at", started) object.__setattr__(self, "final_expires_at", expires) def to_mapping(self) -> dict[str, str]: return { "originalStartedAt": format_rfc3339( self.original_started_at, "original_started_at", ), "finalExpiresAt": format_rfc3339( self.final_expires_at, "final_expires_at", ), } @classmethod def from_mapping(cls, raw: Mapping[str, Any]) -> "TrialPolicyClaim": if not isinstance(raw, Mapping): raise TypeError("trial policy claim must be an object") return cls( original_started_at=parse_rfc3339( raw.get("originalStartedAt"), "originalStartedAt", ), final_expires_at=parse_rfc3339( raw.get("finalExpiresAt"), "finalExpiresAt", ), ) @dataclass(frozen=True) class EntitlementSnapshot: product_id: str snapshot_id: str license_id: str signature_key_id: str subject_type: SubjectType subject_id: str license_state: LicenseState application_major_minimum: int application_major_maximum: int entitlements: tuple[str, ...] grant_summaries: tuple[GrantSummary, ...] device_policy: SnapshotDevicePolicy issued_at: datetime not_before: datetime refresh_after: datetime lease_expires_at: datetime offline_expires_at: Optional[datetime] device_id: Optional[str] device_key_thumbprint: Optional[str] catalog_revision: int revocation_generation: int nonce: str surface_policy: Optional[ProductSurfacePolicy] = None trial_policy: Optional[TrialPolicyClaim] = None schema: str = SNAPSHOT_SCHEMA schema_version: int = SNAPSHOT_SCHEMA_VERSION def __post_init__(self) -> None: if self.schema != SNAPSHOT_SCHEMA or self.schema_version != SNAPSHOT_SCHEMA_VERSION: raise ValueError("unsupported entitlement snapshot schema") object.__setattr__(self, "product_id", validate_identifier(self.product_id, "product_id")) object.__setattr__(self, "snapshot_id", validate_identifier(self.snapshot_id, "snapshot_id")) object.__setattr__(self, "license_id", validate_identifier(self.license_id, "license_id")) object.__setattr__( self, "signature_key_id", validate_identifier(self.signature_key_id, "signature_key_id"), ) if not isinstance(self.subject_type, SubjectType): raise TypeError("subject_type must be a SubjectType") object.__setattr__(self, "subject_id", validate_identifier(self.subject_id, "subject_id")) if not isinstance(self.license_state, LicenseState): raise TypeError("license_state must be a LicenseState") minimum = _positive_integer(self.application_major_minimum, "application_major_minimum") maximum = _positive_integer(self.application_major_maximum, "application_major_maximum") if maximum < minimum: raise ValueError("application_major_maximum must not be below the minimum") object.__setattr__(self, "application_major_minimum", minimum) object.__setattr__(self, "application_major_maximum", maximum) if not isinstance(self.entitlements, tuple): raise TypeError("entitlements must be a tuple") if len(self.entitlements) > MAX_ENTITLEMENTS_PER_SNAPSHOT: raise ValueError("snapshot contains too many entitlements") normalized_entitlements = tuple( validate_identifier(value, "entitlement") for value in self.entitlements ) if tuple(sorted(set(normalized_entitlements))) != normalized_entitlements: raise ValueError("entitlements must be sorted and unique") if not isinstance(self.grant_summaries, tuple): raise TypeError("grant_summaries must be a tuple") if len(self.grant_summaries) > MAX_GRANT_SUMMARIES_PER_SNAPSHOT: raise ValueError("snapshot contains too many grant summaries") if not all(isinstance(value, GrantSummary) for value in self.grant_summaries): raise TypeError("grant_summaries must contain GrantSummary values") if not isinstance(self.device_policy, SnapshotDevicePolicy): raise TypeError("device_policy must be a SnapshotDevicePolicy") issued = parse_rfc3339(format_rfc3339(self.issued_at, "issued_at"), "issued_at") not_before = parse_rfc3339( format_rfc3339(self.not_before, "not_before"), "not_before", ) refresh_after = parse_rfc3339( format_rfc3339(self.refresh_after, "refresh_after"), "refresh_after", ) expires = parse_rfc3339( format_rfc3339(self.lease_expires_at, "lease_expires_at"), "lease_expires_at", ) if not_before > expires: raise ValueError("not_before must not be later than lease_expires_at") if refresh_after < not_before or refresh_after > expires: raise ValueError("refresh_after must be inside the lease validity interval") if issued > expires: raise ValueError("issued_at must not be later than lease_expires_at") object.__setattr__(self, "issued_at", issued) object.__setattr__(self, "not_before", not_before) object.__setattr__(self, "refresh_after", refresh_after) object.__setattr__(self, "lease_expires_at", expires) if self.offline_expires_at is not None: offline_expiry = parse_rfc3339( format_rfc3339(self.offline_expires_at, "offline_expires_at"), "offline_expires_at", ) if offline_expiry < not_before: raise ValueError("offline_expires_at must not precede not_before") object.__setattr__(self, "offline_expires_at", offline_expiry) object.__setattr__(self, "device_id", _optional_identifier(self.device_id, "device_id")) object.__setattr__( self, "device_key_thumbprint", _optional_identifier(self.device_key_thumbprint, "device_key_thumbprint"), ) object.__setattr__( self, "catalog_revision", _positive_integer(self.catalog_revision, "catalog_revision"), ) if self.surface_policy is not None and not isinstance( self.surface_policy, ProductSurfacePolicy, ): raise TypeError("surface_policy must be a ProductSurfacePolicy or None") if ( self.catalog_revision >= PRODUCT_SURFACE_POLICY_REQUIRED_CATALOG_REVISION and self.surface_policy is None ): raise ValueError( "current catalog snapshots require an explicit product surface policy" ) trial_state = self.license_state in ( LicenseState.TRIAL_ACTIVE, LicenseState.TRIAL_PROVISIONAL, ) if self.trial_policy is not None and not isinstance( self.trial_policy, TrialPolicyClaim, ): raise TypeError("trial_policy must be a TrialPolicyClaim or None") if not trial_state and self.trial_policy is not None: raise ValueError("only trial snapshots may contain a trial policy claim") if ( trial_state and self.catalog_revision >= PRODUCT_SURFACE_POLICY_REQUIRED_CATALOG_REVISION and self.trial_policy is None ): raise ValueError("current catalog trial snapshots require a trial policy claim") if self.trial_policy is not None: if issued < self.trial_policy.original_started_at: raise ValueError("trial original start must not be later than issued_at") if expires > self.trial_policy.final_expires_at: raise ValueError("trial lease must not outlive the final trial expiry") if any( summary.ends_at is not None and summary.ends_at > self.trial_policy.final_expires_at for summary in self.grant_summaries ): raise ValueError("trial grant summary outlives the final trial expiry") object.__setattr__( self, "revocation_generation", _nonnegative_integer(self.revocation_generation, "revocation_generation"), ) if not isinstance(self.nonce, str): raise TypeError("nonce must be a string") normalized_nonce = self.nonce.strip() if not normalized_nonce or len(normalized_nonce) > MAX_NONCE_LENGTH: raise ValueError("nonce is empty or exceeds the maximum length") object.__setattr__(self, "nonce", normalized_nonce) def to_payload_mapping(self) -> dict[str, Any]: result = { "schema": self.schema, "schemaVersion": self.schema_version, "productId": self.product_id, "snapshotId": self.snapshot_id, "licenseId": self.license_id, "signatureKeyId": self.signature_key_id, "subjectType": self.subject_type.value, "subjectId": self.subject_id, "licenseState": self.license_state.value, "applicationMajorMinimum": self.application_major_minimum, "applicationMajorMaximum": self.application_major_maximum, "entitlements": list(self.entitlements), "grantSummaries": [value.to_mapping() for value in self.grant_summaries], "devicePolicy": self.device_policy.to_mapping(), "issuedAt": format_rfc3339(self.issued_at, "issued_at"), "notBefore": format_rfc3339(self.not_before, "not_before"), "refreshAfter": format_rfc3339(self.refresh_after, "refresh_after"), "leaseExpiresAt": format_rfc3339(self.lease_expires_at, "lease_expires_at"), "offlineExpiresAt": ( None if self.offline_expires_at is None else format_rfc3339(self.offline_expires_at, "offline_expires_at") ), "deviceId": self.device_id, "deviceKeyThumbprint": self.device_key_thumbprint, "catalogRevision": self.catalog_revision, "revocationGeneration": self.revocation_generation, "nonce": self.nonce, } if self.surface_policy is not None: result["surfacePolicy"] = self.surface_policy.to_mapping() if self.trial_policy is not None: result["trialPolicy"] = self.trial_policy.to_mapping() return result @classmethod def from_payload_mapping(cls, raw: Mapping[str, Any]) -> "EntitlementSnapshot": if not isinstance(raw, Mapping): raise TypeError("snapshot payload must be an object") if raw.get("schema") != SNAPSHOT_SCHEMA or raw.get("schemaVersion") != SNAPSHOT_SCHEMA_VERSION: raise LicensingError( LicenseErrorCode.UNSUPPORTED_SCHEMA, "unsupported entitlement snapshot schema", ) entitlements = raw.get("entitlements") grant_summaries = raw.get("grantSummaries") if not isinstance(entitlements, list): raise TypeError("snapshot entitlements must be a list") if not isinstance(grant_summaries, list): raise TypeError("snapshot grantSummaries must be a list") try: return cls( product_id=raw.get("productId"), snapshot_id=raw.get("snapshotId"), license_id=raw.get("licenseId"), signature_key_id=raw.get("signatureKeyId"), subject_type=SubjectType(raw.get("subjectType")), subject_id=raw.get("subjectId"), license_state=LicenseState(raw.get("licenseState")), application_major_minimum=raw.get("applicationMajorMinimum"), application_major_maximum=raw.get("applicationMajorMaximum"), entitlements=tuple(entitlements), grant_summaries=tuple( GrantSummary.from_mapping(value) for value in grant_summaries ), device_policy=SnapshotDevicePolicy.from_mapping(raw.get("devicePolicy")), issued_at=parse_rfc3339(raw.get("issuedAt"), "issuedAt"), not_before=parse_rfc3339(raw.get("notBefore"), "notBefore"), refresh_after=parse_rfc3339(raw.get("refreshAfter"), "refreshAfter"), lease_expires_at=parse_rfc3339(raw.get("leaseExpiresAt"), "leaseExpiresAt"), offline_expires_at=( None if raw.get("offlineExpiresAt") is None else parse_rfc3339(raw.get("offlineExpiresAt"), "offlineExpiresAt") ), device_id=raw.get("deviceId"), device_key_thumbprint=raw.get("deviceKeyThumbprint"), catalog_revision=raw.get("catalogRevision"), revocation_generation=raw.get("revocationGeneration"), nonce=raw.get("nonce"), surface_policy=( None if raw.get("surfacePolicy") is None else ProductSurfacePolicy.from_mapping(raw.get("surfacePolicy")) ), trial_policy=( None if raw.get("trialPolicy") is None else TrialPolicyClaim.from_mapping(raw.get("trialPolicy")) ), schema=raw.get("schema"), schema_version=raw.get("schemaVersion"), ) except LicensingError: raise except (TypeError, ValueError) as exc: raise LicensingError( LicenseErrorCode.INVALID_DOCUMENT, f"invalid entitlement snapshot: {exc}", ) from exc @dataclass(frozen=True) class SignedLicenseDocument: key_id: str payload: Mapping[str, Any] signature_base64: str algorithm: str = SIGNED_DOCUMENT_ALGORITHM schema: str = SIGNED_DOCUMENT_SCHEMA schema_version: int = SIGNED_DOCUMENT_SCHEMA_VERSION def __post_init__(self) -> None: if self.schema != SIGNED_DOCUMENT_SCHEMA or self.schema_version != SIGNED_DOCUMENT_SCHEMA_VERSION: raise ValueError("unsupported signed-document schema") if self.algorithm != SIGNED_DOCUMENT_ALGORITHM: raise ValueError("unsupported signed-document algorithm") object.__setattr__(self, "key_id", validate_identifier(self.key_id, "key_id")) if not isinstance(self.payload, Mapping): raise TypeError("payload must be an object") object.__setattr__(self, "payload", dict(self.payload)) if not isinstance(self.signature_base64, str): raise TypeError("signature_base64 must be a string") signature = self.signature_base64.strip() if not signature or len(signature) > 256: raise ValueError("signature_base64 is empty or too long") object.__setattr__(self, "signature_base64", signature) def to_mapping(self) -> dict[str, Any]: return { "schema": self.schema, "schemaVersion": self.schema_version, "algorithm": self.algorithm, "keyId": self.key_id, "payload": dict(self.payload), "signature": self.signature_base64, } @classmethod def from_mapping(cls, raw: Mapping[str, Any]) -> "SignedLicenseDocument": if not isinstance(raw, Mapping): raise TypeError("signed document must be an object") if ( raw.get("schema") != SIGNED_DOCUMENT_SCHEMA or raw.get("schemaVersion") != SIGNED_DOCUMENT_SCHEMA_VERSION ): raise LicensingError( LicenseErrorCode.UNSUPPORTED_SCHEMA, "unsupported signed-license document schema", ) try: return cls( key_id=raw.get("keyId"), payload=raw.get("payload"), signature_base64=raw.get("signature"), algorithm=raw.get("algorithm"), schema=raw.get("schema"), schema_version=raw.get("schemaVersion"), ) except LicensingError: raise except (TypeError, ValueError) as exc: raise LicensingError( LicenseErrorCode.INVALID_DOCUMENT, f"invalid signed-license document: {exc}", ) from exc @dataclass(frozen=True) class VerifiedEntitlementSnapshot: snapshot: EntitlementSnapshot verified_at: datetime key_id: str def __post_init__(self) -> None: if not isinstance(self.snapshot, EntitlementSnapshot): raise TypeError("snapshot must be an EntitlementSnapshot") verified = parse_rfc3339( format_rfc3339(self.verified_at, "verified_at"), "verified_at", ) object.__setattr__(self, "verified_at", verified) object.__setattr__(self, "key_id", validate_identifier(self.key_id, "key_id")) __all__ = [ "EntitlementSnapshot", "GrantSummary", "LicenseState", "ProductSurfacePolicy", "SignedLicenseDocument", "SnapshotDevicePolicy", "SubjectType", "TrialPolicyClaim", "VerifiedEntitlementSnapshot", "format_rfc3339", "parse_rfc3339", ]