"""Central, typed authorization decisions for every protected boundary.""" from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timezone from typing import Optional from .catalog import LicensingCatalog from .constants import validate_identifier from .models import ( LicenseState, VerifiedEntitlementSnapshot, format_rfc3339, parse_rfc3339, ) @dataclass(frozen=True) class EntitlementRequirement: access_id: str all_of: tuple[str, ...] = () any_of: tuple[str, ...] = () public: bool = False recovery_safe: bool = False def __post_init__(self) -> None: object.__setattr__(self, "access_id", validate_identifier(self.access_id, "access_id")) if not isinstance(self.all_of, tuple): raise TypeError("all_of must be a tuple") if not isinstance(self.any_of, tuple): raise TypeError("any_of must be a tuple") if not isinstance(self.public, bool): raise TypeError("public must be a Boolean") if not isinstance(self.recovery_safe, bool): raise TypeError("recovery_safe must be a Boolean") normalized_all = tuple(validate_identifier(value, "all_of entitlement") for value in self.all_of) normalized_any = tuple(validate_identifier(value, "any_of entitlement") for value in self.any_of) if tuple(sorted(set(normalized_all))) != normalized_all: raise ValueError("all_of entitlements must be sorted and unique") if tuple(sorted(set(normalized_any))) != normalized_any: raise ValueError("any_of entitlements must be sorted and unique") if self.public and (normalized_all or normalized_any): raise ValueError("public requirements cannot declare entitlements") if not self.public and not normalized_all and not normalized_any: raise ValueError("protected requirements must declare at least one entitlement") object.__setattr__(self, "all_of", normalized_all) object.__setattr__(self, "any_of", normalized_any) @dataclass(frozen=True) class AuthorizationDecision: allowed: bool access_id: str state: LicenseState reason_code: str message: str missing_entitlements: tuple[str, ...] = () refresh_recommended: bool = False recovery_actions: tuple[str, ...] = () def __post_init__(self) -> None: if not isinstance(self.allowed, bool): raise TypeError("allowed must be a Boolean") object.__setattr__(self, "access_id", validate_identifier(self.access_id, "access_id")) if not isinstance(self.state, LicenseState): raise TypeError("state must be a LicenseState") object.__setattr__(self, "reason_code", validate_identifier(self.reason_code, "reason_code")) if not isinstance(self.message, str) or not self.message.strip(): raise ValueError("message must be a non-empty string") object.__setattr__(self, "message", self.message.strip()) if not isinstance(self.missing_entitlements, tuple): raise TypeError("missing_entitlements must be a tuple") object.__setattr__( self, "missing_entitlements", tuple( validate_identifier(value, "missing entitlement") for value in self.missing_entitlements ), ) if not isinstance(self.refresh_recommended, bool): raise TypeError("refresh_recommended must be a Boolean") if not isinstance(self.recovery_actions, tuple): raise TypeError("recovery_actions must be a tuple") object.__setattr__( self, "recovery_actions", tuple(validate_identifier(value, "recovery action") for value in self.recovery_actions), ) class LicensePolicy: """Evaluate one immutable verified snapshot at a canonical dispatch point.""" _DENIED_STATES = frozenset( ( LicenseState.EXPIRED, LicenseState.SUSPENDED, LicenseState.REVOKED, LicenseState.REFUNDED, LicenseState.WRONG_DEVICE, LicenseState.SEAT_LIMIT_REACHED, LicenseState.WRONG_APPLICATION_VERSION, LicenseState.INVALID_DOCUMENT, LicenseState.CLOCK_ANOMALY, LicenseState.ACCOUNT_ACTION_REQUIRED, ) ) def __init__(self, catalog: LicensingCatalog) -> None: if not isinstance(catalog, LicensingCatalog): raise TypeError("catalog must be a LicensingCatalog") self.catalog = catalog @staticmethod def _now(value: Optional[datetime]) -> datetime: if value is None: return datetime.now(timezone.utc) if not isinstance(value, datetime): raise TypeError("now must be a datetime or None") return parse_rfc3339(format_rfc3339(value, "now"), "now") @staticmethod def _deny( requirement: EntitlementRequirement, state: LicenseState, reason_code: str, message: str, *, missing: tuple[str, ...] = (), actions: tuple[str, ...] = ("license.activate", "license.sign_in"), ) -> AuthorizationDecision: return AuthorizationDecision( allowed=False, access_id=requirement.access_id, state=state, reason_code=reason_code, message=message, missing_entitlements=missing, recovery_actions=actions, ) def evaluate( self, requirement: EntitlementRequirement, verified: Optional[VerifiedEntitlementSnapshot], *, application_major_version: int, device_id: Optional[str] = None, device_key_thumbprint: Optional[str] = None, now: Optional[datetime] = None, ) -> AuthorizationDecision: if not isinstance(requirement, EntitlementRequirement): raise TypeError("requirement must be an EntitlementRequirement") if verified is not None and not isinstance(verified, VerifiedEntitlementSnapshot): raise TypeError("verified must be a VerifiedEntitlementSnapshot or None") if isinstance(application_major_version, bool) or not isinstance( application_major_version, int, ): raise TypeError("application_major_version must be an integer") if application_major_version < 1: raise ValueError("application_major_version must be at least one") normalized_device_id = ( None if device_id is None else validate_identifier(device_id, "device_id") ) normalized_thumbprint = ( None if device_key_thumbprint is None else validate_identifier(device_key_thumbprint, "device_key_thumbprint") ) evaluated_at = self._now(now) if requirement.public: return AuthorizationDecision( allowed=True, access_id=requirement.access_id, state=LicenseState.LICENSED_ACTIVE if verified else LicenseState.MISSING, reason_code="authorization.public", message="This operation is available without a license.", ) self.catalog.validate_entitlement_ids(list(requirement.all_of + requirement.any_of)) if verified is None: if requirement.recovery_safe: return AuthorizationDecision( allowed=True, access_id=requirement.access_id, state=LicenseState.MISSING, reason_code="authorization.recovery_safe", message="Recovery-safe access remains available without an active license.", ) return self._deny( requirement, LicenseState.MISSING, "authorization.license_missing", "Activate a license, start a trial, or sign in to use this feature.", ) snapshot = verified.snapshot if not ( snapshot.application_major_minimum <= application_major_version <= snapshot.application_major_maximum ): return self._deny( requirement, LicenseState.WRONG_APPLICATION_VERSION, "authorization.application_version", "This license does not cover the installed application major version.", actions=("license.view_versions", "license.upgrade"), ) if snapshot.device_id is not None and snapshot.device_id != normalized_device_id: return self._deny( requirement, LicenseState.WRONG_DEVICE, "authorization.wrong_device", "This signed license belongs to a different activated device.", actions=("license.manage_devices", "license.activate"), ) if ( snapshot.device_key_thumbprint is not None and snapshot.device_key_thumbprint != normalized_thumbprint ): return self._deny( requirement, LicenseState.WRONG_DEVICE, "authorization.wrong_device_key", "The local device credential does not match this signed license.", actions=("license.recover_device", "license.activate"), ) if evaluated_at < snapshot.not_before: return self._deny( requirement, LicenseState.CLOCK_ANOMALY, "authorization.not_yet_valid", "The license is not valid at the current trusted time.", actions=("license.refresh", "license.diagnostics"), ) effective_expiry = snapshot.lease_expires_at if ( snapshot.offline_expires_at is not None and snapshot.offline_expires_at < effective_expiry ): effective_expiry = snapshot.offline_expires_at if evaluated_at >= effective_expiry: if requirement.recovery_safe: return AuthorizationDecision( allowed=True, access_id=requirement.access_id, state=LicenseState.EXPIRED, reason_code="authorization.expired_recovery_safe", message="Recovery-safe access remains available after license expiry.", recovery_actions=("license.refresh", "license.activate"), ) return self._deny( requirement, LicenseState.EXPIRED, "authorization.license_expired", "The signed local authorization has expired and needs renewal.", actions=("license.refresh", "license.offline_activate"), ) if snapshot.license_state in self._DENIED_STATES: if requirement.recovery_safe: return AuthorizationDecision( allowed=True, access_id=requirement.access_id, state=snapshot.license_state, reason_code="authorization.state_recovery_safe", message="Recovery-safe access remains available for this license state.", recovery_actions=("license.center",), ) return self._deny( requirement, snapshot.license_state, f"authorization.{snapshot.license_state.value}", "The current license state does not authorize this operation.", actions=("license.center", "license.diagnostics"), ) available = set(snapshot.entitlements) missing_all = tuple(value for value in requirement.all_of if value not in available) any_satisfied = not requirement.any_of or bool(available.intersection(requirement.any_of)) if missing_all or not any_satisfied: missing = missing_all if not any_satisfied: missing = tuple(sorted(set(missing + requirement.any_of))) return self._deny( requirement, snapshot.license_state, "authorization.entitlement_missing", "This feature is not included in the current edition or add-ons.", missing=missing, actions=("license.view_options", "license.activate"), ) refresh_recommended = evaluated_at >= snapshot.refresh_after decision_state = ( LicenseState.REFRESH_RECOMMENDED if refresh_recommended else snapshot.license_state ) return AuthorizationDecision( allowed=True, access_id=requirement.access_id, state=decision_state, reason_code=( "authorization.allowed_refresh_recommended" if refresh_recommended else "authorization.allowed" ), message="The current signed entitlement authorizes this operation.", refresh_recommended=refresh_recommended, recovery_actions=("license.refresh",) if refresh_recommended else (), ) __all__ = ["AuthorizationDecision", "EntitlementRequirement", "LicensePolicy"]