"""Manual, replay-safe approval for device-signed offline trial requests.""" from __future__ import annotations import base64 from dataclasses import dataclass from datetime import datetime, timedelta, timezone import hashlib import hmac from typing import Callable, Mapping, Optional from sqlalchemy import select from sqlalchemy.orm import Session, sessionmaker from licensing_shared.canonical_json import canonicalize_json from licensing_shared.catalog import LicensingCatalog from licensing_shared.constants import ( TRIAL_MAXIMUM_TOTAL_DURATION_HOURS, validate_identifier, ) from licensing_shared.errors import LicensingError from licensing_shared.models import ( EntitlementSnapshot, LicenseState, parse_rfc3339, ) from licensing_shared.offline import ( DeviceSignedOfflineRequest, OfflineRequestType, parse_and_verify_offline_request, ) from .constants import ( MAX_IDEMPOTENCY_KEY_CHARACTERS, MAX_REASON_CHARACTERS, OFFLINE_TRIAL_APPROVAL_REASONS, OFFLINE_TRIAL_REJECTION_REASONS, ) from .errors import ServerErrorCode, ServerLicensingError from .models import ( AuditEvent, Lease, OfflineCertificate, OfflineRequest, User, ) from .security import SnapshotSigner, anonymous_subject_digest, new_identifier from .signing_key_administration import RegistryBoundSnapshotSigner from .catalog_administration import CatalogBoundSnapshotSigner from .services import DeviceEnrollment, LicensingService MAXIMUM_OFFLINE_REQUEST_RESULTS = 100 OFFLINE_REQUEST_NONCE_MAXIMUM_CHARACTERS = 256 @dataclass(frozen=True) class _PreparedOfflineTrial: row_id: str request: DeviceSignedOfflineRequest def _utc(value: datetime) -> datetime: if not isinstance(value, datetime): raise TypeError("value must be a datetime") if value.tzinfo is None or value.utcoffset() is None: return value.replace(tzinfo=timezone.utc) return value.astimezone(timezone.utc) def _bounded_text(value: str, field_name: str, maximum: int) -> str: if not isinstance(value, str): raise TypeError(f"{field_name} must be a string") normalized = value.strip() if not normalized or len(normalized) > maximum: raise ValueError(f"{field_name} is empty or too long") return normalized def _reason( reason_code: str, note: Optional[str], allowed: frozenset[str], ) -> tuple[str, Optional[str]]: normalized_reason = validate_identifier(reason_code, "reason_code") if normalized_reason not in allowed: raise ValueError("reason_code is not allowed for this decision") normalized_note = ( None if note is None else _bounded_text(note, "note", MAX_REASON_CHARACTERS) ) return normalized_reason, normalized_note class OfflineTrialApprovalService: def __init__( self, session_factory: sessionmaker[Session], catalog: LicensingCatalog, signer: SnapshotSigner, serial_pepper: bytes, fingerprint_pepper: bytes, *, now_factory: Callable[[], datetime] = lambda: datetime.now(timezone.utc), ) -> None: if not callable(session_factory): raise TypeError("session_factory must be callable") if not isinstance(catalog, LicensingCatalog): raise TypeError("catalog must be a LicensingCatalog") if not isinstance(serial_pepper, bytes) or not serial_pepper: raise TypeError("serial_pepper must be non-empty bytes") if not isinstance(fingerprint_pepper, bytes) or not fingerprint_pepper: raise TypeError("fingerprint_pepper must be non-empty bytes") if not callable(now_factory): raise TypeError("now_factory must be callable") self._session_factory = session_factory self._catalog = catalog self._signer = signer self._serial_pepper = serial_pepper self._fingerprint_pepper = fingerprint_pepper self._now_factory = now_factory def submit(self, request: DeviceSignedOfflineRequest) -> dict[str, object]: if not isinstance(request, DeviceSignedOfflineRequest): raise TypeError("request must be a DeviceSignedOfflineRequest") payload = request.payload if payload.request_type is not OfflineRequestType.TRIAL: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, "offline request is not a trial request", ) now = self._now() document = request.to_mapping() request_digest = hashlib.sha256(canonicalize_json(document)).digest() retrieval_digest = self._retrieval_digest(payload.nonce) with self._session_factory() as session, session.begin(): existing = session.scalar( select(OfflineRequest) .where(OfflineRequest.client_request_id == payload.request_id) .with_for_update() ) if existing is not None: if not hmac.compare_digest(existing.request_digest, request_digest): raise ServerLicensingError( ServerErrorCode.CONFLICT, "offline request ID is already bound to another signed request", status_code=409, ) if not hmac.compare_digest(existing.retrieval_digest, retrieval_digest): raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "offline request retrieval proof does not match", status_code=403, ) return self._public_mapping(session, existing, now) row = OfflineRequest( id=new_identifier("offline_request"), client_request_id=payload.request_id, request_type=payload.request_type.value, status="pending", request_digest=request_digest, request_document_json=document, retrieval_digest=retrieval_digest, installation_id=payload.installation_id, device_key_thumbprint=payload.device_key_thumbprint, requested_at=now, expires_at=payload.expires_at, decision_generation=0, ) session.add(row) self._audit( session, "offline_trial.requested", row.id, actor_id=None, correlation_id=payload.request_id, metadata={"requestType": payload.request_type.value}, ) session.flush() return self._public_mapping(session, row, now) def status(self, request_id: str, nonce: str) -> dict[str, object]: normalized_request = validate_identifier(request_id, "request_id") retrieval_digest = self._retrieval_digest(nonce) now = self._now() with self._session_factory() as session: row = session.scalar( select(OfflineRequest).where( OfflineRequest.client_request_id == normalized_request ) ) if row is None: raise ServerLicensingError( ServerErrorCode.OFFLINE_REQUEST_NOT_FOUND, "offline trial request was not found", status_code=404, ) if not hmac.compare_digest(row.retrieval_digest, retrieval_digest): raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "offline request retrieval proof does not match", status_code=403, ) return self._public_mapping(session, row, now) def list_requests( self, actor_user_id: str, *, status: Optional[str] = None, limit: int = 50, ) -> dict[str, object]: normalized_status = None if status is not None: normalized_status = validate_identifier(status, "status") if normalized_status not in ("pending", "approved", "rejected", "expired"): raise ValueError("status is not supported") if isinstance(limit, bool) or not isinstance(limit, int): raise TypeError("limit must be an integer") if limit < 1 or limit > MAXIMUM_OFFLINE_REQUEST_RESULTS: raise ValueError( f"limit must be between one and {MAXIMUM_OFFLINE_REQUEST_RESULTS}" ) now = self._now() with self._session_factory() as session: self._require_admin(session, actor_user_id) query = select(OfflineRequest) if normalized_status == "expired": query = query.where( OfflineRequest.status == "pending", OfflineRequest.expires_at <= now, ) elif normalized_status == "pending": query = query.where( OfflineRequest.status == "pending", OfflineRequest.expires_at > now, ) elif normalized_status is not None: query = query.where(OfflineRequest.status == normalized_status) rows = session.scalars( query.order_by( OfflineRequest.requested_at.desc(), OfflineRequest.id.desc(), ).limit(limit) ).all() return { "requests": [self._admin_mapping(row, now) for row in rows], "limit": limit, } def inspect(self, request_id: str, actor_user_id: str) -> dict[str, object]: normalized_request = validate_identifier(request_id, "request_id") now = self._now() with self._session_factory() as session: self._require_admin(session, actor_user_id) row = session.get(OfflineRequest, normalized_request) if row is None: raise ServerLicensingError( ServerErrorCode.OFFLINE_REQUEST_NOT_FOUND, "offline trial request was not found", status_code=404, ) return self._admin_mapping(row, now) def approve( self, request_id: str, actor_user_id: str, reason_code: str, expected_decision_generation: int, idempotency_key: str, correlation_id: str, *, note: Optional[str] = None, ) -> dict[str, object]: normalized_idempotency = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_reason, normalized_note = _reason( reason_code, note, OFFLINE_TRIAL_APPROVAL_REASONS, ) prepared, replay = self._prepare_decision( request_id, actor_user_id, "approved", normalized_reason, normalized_note, expected_decision_generation, normalized_idempotency, correlation_id, ) if replay is not None: return replay assert prepared is not None payload = prepared.request.payload enrollment = DeviceEnrollment( installation_id=payload.installation_id, public_key_der=base64.b64decode( payload.device_public_key_base64, validate=True, ), key_thumbprint=payload.device_key_thumbprint, key_provider=payload.key_provider, friendly_name=payload.friendly_name, evidence=dict(payload.evidence_digests), ) with self._session_factory() as session: result = LicensingService( session, self._catalog, CatalogBoundSnapshotSigner( session, self._catalog, RegistryBoundSnapshotSigner( session, self._signer, now_factory=self._now_factory, ), ), self._serial_pepper, self._fingerprint_pepper, now_factory=self._now_factory, ).start_trial( enrollment, payload.application_major_version, payload.request_id, correlation_id, offline_certificate=True, offline_request_nonce=payload.nonce, ) return self._complete_approval( prepared, result.to_response_mapping(), actor_user_id, normalized_reason, normalized_note, expected_decision_generation, normalized_idempotency, correlation_id, ) def reject( self, request_id: str, actor_user_id: str, reason_code: str, expected_decision_generation: int, idempotency_key: str, correlation_id: str, *, note: Optional[str] = None, ) -> dict[str, object]: normalized_idempotency = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_reason, normalized_note = _reason( reason_code, note, OFFLINE_TRIAL_REJECTION_REASONS, ) prepared, replay = self._prepare_decision( request_id, actor_user_id, "rejected", normalized_reason, normalized_note, expected_decision_generation, normalized_idempotency, correlation_id, ) if replay is not None: return replay assert prepared is not None decision_digest = self._decision_digest( "rejected", normalized_reason, normalized_note, expected_decision_generation, ) now = self._now() with self._session_factory() as session, session.begin(): actor = self._require_admin(session, actor_user_id) row = session.scalar( select(OfflineRequest) .where(OfflineRequest.id == prepared.row_id) .with_for_update() ) if row is None: raise ServerLicensingError( ServerErrorCode.OFFLINE_REQUEST_NOT_FOUND, "offline trial request was not found", status_code=404, ) replay = self._decision_replay( session, row, normalized_idempotency, decision_digest, now, ) if replay is not None: return replay self._require_pending_generation(row, expected_decision_generation, now) row.status = "rejected" row.decision_generation += 1 row.decided_by_user_id = actor.id row.decision_reason = normalized_reason row.decision_note = normalized_note row.decided_at = now row.decision_idempotency_key = normalized_idempotency row.decision_request_digest = decision_digest row.decision_response_json = { "status": "rejected", "reasonCode": normalized_reason, } self._audit( session, "offline_trial.rejected", row.id, actor_id=actor.id, correlation_id=correlation_id, reason=normalized_note, metadata={"reasonCode": normalized_reason}, ) session.flush() return self._public_mapping(session, row, now) def _prepare_decision( self, request_id: str, actor_user_id: str, decision: str, reason_code: str, note: Optional[str], expected_decision_generation: int, idempotency_key: str, correlation_id: str, ) -> tuple[Optional[_PreparedOfflineTrial], Optional[dict[str, object]]]: normalized_request = validate_identifier(request_id, "request_id") validate_identifier(actor_user_id, "actor_user_id") normalized_key = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) validate_identifier(correlation_id, "correlation_id") if isinstance(expected_decision_generation, bool) or not isinstance( expected_decision_generation, int, ): raise TypeError("expected_decision_generation must be an integer") if expected_decision_generation < 0: raise ValueError("expected_decision_generation must not be negative") decision_digest = self._decision_digest( decision, reason_code, note, expected_decision_generation, ) now = self._now() with self._session_factory() as session: self._require_admin(session, actor_user_id) row = session.scalar( select(OfflineRequest) .where(OfflineRequest.id == normalized_request) .with_for_update() ) if row is None: raise ServerLicensingError( ServerErrorCode.OFFLINE_REQUEST_NOT_FOUND, "offline trial request was not found", status_code=404, ) replay = self._decision_replay( session, row, normalized_key, decision_digest, now, ) if replay is not None: return None, replay self._require_pending_generation(row, expected_decision_generation, now) parsed = parse_and_verify_offline_request( row.request_document_json, at=now, ) if parsed.payload.request_type is not OfflineRequestType.TRIAL: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, "stored offline request is not a trial request", ) return _PreparedOfflineTrial(row.id, parsed), None def _complete_approval( self, prepared: _PreparedOfflineTrial, mutation: Mapping[str, object], actor_user_id: str, reason_code: str, note: Optional[str], expected_decision_generation: int, idempotency_key: str, correlation_id: str, ) -> dict[str, object]: decision_digest = self._decision_digest( "approved", reason_code, note, expected_decision_generation, ) now = self._now() document = mutation.get("licenseDocument") if not isinstance(document, Mapping): raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "trial service returned no signed license document", status_code=500, ) snapshot_id = validate_identifier( document.get("payload", {}).get("snapshotId") if isinstance(document.get("payload"), Mapping) else None, "snapshot_id", ) document_payload = document.get("payload") if not isinstance(document_payload, Mapping): raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "trial service returned an invalid signed payload", status_code=500, ) try: snapshot = EntitlementSnapshot.from_payload_mapping(document_payload) except (LicensingError, TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "trial service returned an invalid signed entitlement payload", status_code=500, ) from exc if ( snapshot.license_state not in ( LicenseState.TRIAL_ACTIVE, LicenseState.TRIAL_PROVISIONAL, ) or snapshot.trial_policy is None or snapshot.trial_policy.final_expires_at - snapshot.trial_policy.original_started_at > timedelta(hours=TRIAL_MAXIMUM_TOTAL_DURATION_HOURS) ): raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "offline trial response exceeds the duration accepted by shipped clients", status_code=500, ) if document_payload.get("nonce") != prepared.request.payload.nonce: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "offline trial response nonce does not match its request", status_code=500, ) try: offline_expiry = parse_rfc3339( document_payload.get("offlineExpiresAt"), "offlineExpiresAt", ) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "trial service returned no valid offline expiry", status_code=500, ) from exc response = dict(mutation) response["offlineRequestId"] = prepared.request.payload.request_id response["offlineRequestNonce"] = prepared.request.payload.nonce response_digest = hashlib.sha256(canonicalize_json(response)).digest() with self._session_factory() as session, session.begin(): actor = self._require_admin(session, actor_user_id) row = session.scalar( select(OfflineRequest) .where(OfflineRequest.id == prepared.row_id) .with_for_update() ) if row is None: raise ServerLicensingError( ServerErrorCode.OFFLINE_REQUEST_NOT_FOUND, "offline trial request was not found", status_code=404, ) replay = self._decision_replay( session, row, idempotency_key, decision_digest, now, ) if replay is not None: return replay self._require_pending_generation(row, expected_decision_generation, now) lease = session.scalar(select(Lease).where(Lease.snapshot_id == snapshot_id)) if lease is None: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "offline trial lease record is missing", status_code=500, ) certificate = OfflineCertificate( id=new_identifier("offline_certificate"), offline_request_id=row.id, license_id=validate_identifier(mutation.get("licenseId"), "licenseId"), activation_id=validate_identifier( mutation.get("activationId"), "activationId", ), lease_id=lease.id, file_digest=response_digest, issued_at=_utc(lease.issued_at), expires_at=offline_expiry, ) session.add(certificate) row.status = "approved" row.decision_generation += 1 row.decided_by_user_id = actor.id row.decision_reason = reason_code row.decision_note = note row.decided_at = now row.decision_idempotency_key = idempotency_key row.decision_request_digest = decision_digest row.decision_response_json = { "status": "approved", "licenseId": certificate.license_id, "activationId": certificate.activation_id, "recoveredExistingDevice": bool( mutation.get("recoveredExistingDevice") ), } self._audit( session, "offline_trial.approved", row.id, actor_id=actor.id, correlation_id=correlation_id, reason=note, metadata={ "reasonCode": reason_code, "licenseId": certificate.license_id, "activationId": certificate.activation_id, "offlineCertificate": True, }, ) session.flush() return self._public_mapping(session, row, now) def _decision_replay( self, session: Session, row: OfflineRequest, idempotency_key: str, request_digest: bytes, now: datetime, ) -> Optional[dict[str, object]]: if row.decision_idempotency_key is None: if row.status != "pending": raise ServerLicensingError( ServerErrorCode.CONFLICT, "offline trial request already has a decision", status_code=409, ) return None if row.decision_idempotency_key != idempotency_key: raise ServerLicensingError( ServerErrorCode.CONFLICT, "offline trial request already has a decision", status_code=409, ) if row.decision_request_digest is None or not hmac.compare_digest( row.decision_request_digest, request_digest, ): raise ServerLicensingError( ServerErrorCode.IDEMPOTENCY_CONFLICT, "idempotency key was already used for another offline decision", status_code=409, ) return self._public_mapping(session, row, now) def _require_pending_generation( self, row: OfflineRequest, expected_generation: int, now: datetime, ) -> None: if row.status != "pending": raise ServerLicensingError( ServerErrorCode.CONFLICT, "offline trial request already has a decision", status_code=409, ) if row.decision_generation != expected_generation: raise ServerLicensingError( ServerErrorCode.CONFLICT, "offline trial request changed after preview; preview again", status_code=409, ) if _utc(row.expires_at) <= now: raise ServerLicensingError( ServerErrorCode.OFFLINE_REQUEST_EXPIRED, "offline trial request has expired; create a new request", status_code=409, ) def _public_mapping( self, session: Session, row: OfflineRequest, now: datetime, ) -> dict[str, object]: effective_status = ( "expired" if row.status == "pending" and _utc(row.expires_at) <= now else row.status ) result: dict[str, object] = { "requestId": row.client_request_id, "status": effective_status, "decisionGeneration": row.decision_generation, "requestExpiresAt": _utc(row.expires_at).isoformat(), } if effective_status == "rejected": result["reasonCode"] = row.decision_reason if effective_status != "approved": return result certificate = session.scalar( select(OfflineCertificate).where( OfflineCertificate.offline_request_id == row.id ) ) if certificate is None: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "approved offline request has no certificate", status_code=500, ) lease = session.get(Lease, certificate.lease_id) if lease is None: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "approved offline request lease is missing", status_code=500, ) stored_request = DeviceSignedOfflineRequest.from_mapping(row.request_document_json) result.update( { "licenseId": certificate.license_id, "activationId": certificate.activation_id, "deviceId": lease.document_json["payload"]["deviceId"], "recoveredExistingDevice": bool( (row.decision_response_json or {}).get( "recoveredExistingDevice", False, ) ), "idempotentReplay": True, "licenseDocument": lease.document_json, "offlineRequestId": row.client_request_id, "offlineRequestNonce": stored_request.payload.nonce, } ) return result def _admin_mapping(self, row: OfflineRequest, now: datetime) -> dict[str, object]: request = DeviceSignedOfflineRequest.from_mapping(row.request_document_json) payload = request.payload return { "id": row.id, "requestId": row.client_request_id, "requestType": row.request_type, "status": ( "expired" if row.status == "pending" and _utc(row.expires_at) <= now else row.status ), "decisionGeneration": row.decision_generation, "applicationMajorVersion": payload.application_major_version, "friendlyName": payload.friendly_name, "keyProvider": payload.key_provider, "evidenceComponents": [key for key, _value in payload.evidence_digests], "requestedAt": _utc(row.requested_at).isoformat(), "expiresAt": _utc(row.expires_at).isoformat(), "decisionReason": row.decision_reason, "decidedAt": ( None if row.decided_at is None else _utc(row.decided_at).isoformat() ), } def _require_admin(self, session: Session, actor_user_id: str) -> User: normalized_actor = validate_identifier(actor_user_id, "actor_user_id") actor = session.get(User, normalized_actor) if actor is None or actor.status != "active" or not actor.is_server_admin: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "offline approval requires an active server administrator", status_code=403, ) return actor def _retrieval_digest(self, nonce: str) -> bytes: normalized_nonce = _bounded_text( nonce, "nonce", OFFLINE_REQUEST_NONCE_MAXIMUM_CHARACTERS, ) return anonymous_subject_digest( self._fingerprint_pepper, f"offline-trial-retrieval:{normalized_nonce}", ) def _decision_digest( self, decision: str, reason_code: str, note: Optional[str], expected_generation: int, ) -> bytes: return hashlib.sha256( canonicalize_json( { "decision": validate_identifier(decision, "decision"), "reasonCode": validate_identifier(reason_code, "reason_code"), "note": note, "expectedDecisionGeneration": expected_generation, } ) ).digest() def _audit( self, session: Session, action: str, target_id: str, *, actor_id: Optional[str], correlation_id: str, reason: Optional[str] = None, metadata: Optional[dict[str, object]] = None, ) -> None: session.add( AuditEvent( id=new_identifier("audit"), actor_type="user" if actor_id is not None else "anonymous", actor_id=actor_id, action=validate_identifier(action, "action"), target_type="offline_request", target_id=validate_identifier(target_id, "target_id"), reason=reason, correlation_id=validate_identifier(correlation_id, "correlation_id"), source_address_digest=None, metadata_json={} if metadata is None else metadata, ) ) def _now(self) -> datetime: return _utc(self._now_factory()) __all__ = ["OfflineTrialApprovalService"]