"""Transactional licensing domain services.""" from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timedelta, timezone import hashlib import hmac from typing import Callable, Iterable, Mapping, Optional from sqlalchemy import Select, func, select from sqlalchemy.orm import Session from licensing_shared.canonical_json import canonicalize_json from licensing_shared.catalog import HOURS_PER_DAY, LicensingCatalog, SkuDefinition, SkuKind from licensing_shared.constants import ( PRODUCT_ID, TRIAL_MAXIMUM_TOTAL_DURATION_HOURS, TRIAL_MINIMUM_AUTHORITY_DURATION_HOURS, validate_identifier, ) from licensing_shared.models import ( EntitlementSnapshot, GrantSummary, LicenseState, ProductSurfacePolicy, SignedLicenseDocument, SnapshotDevicePolicy, SubjectType, TrialPolicyClaim, ) from licensing_shared.offline import device_public_key_thumbprint from licensing_shared.serials import ( SERIAL_ALPHABET, SERIAL_LOOKUP_CHARACTERS, SERIAL_PRODUCT_PREFIX, SERIAL_SECRET_CHARACTERS, ParsedSerial, parse_serial, serial_checksum, ) from licensing_shared.product_surfaces import ( PRODUCT_SURFACE_INVENTORY_REVISION, PRODUCT_SURFACE_INVENTORY_SHA256, PRODUCT_SURFACE_KIND_ANALYSIS, PRODUCT_SURFACE_KIND_MAIN_TAB, product_surface_ids, product_surface_inventory_mapping, validate_product_surface_ids, ) from .constants import ( DEFAULT_IDEMPOTENCY_DAYS, DEFAULT_NOT_BEFORE_SKEW_SECONDS, LEGACY_CONNECTED_REFRESH_TARGET_HOURS, LEGACY_OFFLINE_REFRESH_REMINDER_DAYS, MAX_FRIENDLY_DEVICE_NAME_CHARACTERS, MAX_IDEMPOTENCY_KEY_CHARACTERS, MAX_REASON_CHARACTERS, MAX_SERIAL_BATCH_QUANTITY, MAX_SUPPORT_TIMELINE_EVENTS, ORGANIZATION_INVITATION_DAYS, ORGANIZATION_SEAT_REASSIGNMENT_REVIEW_DAYS, SUPPORT_DEVICE_RECOVERY_REASONS, SUPPORT_GRANT_REVOCATION_REASONS, SUPPORT_TRIAL_EXTENSION_REASONS, ) from .errors import ServerErrorCode, ServerLicensingError from .models import ( Activation, AuditEvent, Device, DeviceInstallation, Grant, IdempotencyRecord, Lease, License, Membership, Organization, OrganizationInvitation, OrganizationSeat, ProductSurfaceProfile, Serial, SerialBatch, SerialRedemption, Trial, User, ) from .security import ( SnapshotSigner, anonymous_subject_digest, digest_device_evidence, new_identifier, random_token, serial_secret_digest, ) UNIQUE_DEVICE_EVIDENCE_COMPONENTS = frozenset( ("system_uuid", "baseboard_serial") ) @dataclass(frozen=True) class DeviceEnrollment: installation_id: str public_key_der: bytes key_thumbprint: str key_provider: str friendly_name: str evidence: Mapping[str, str] def __post_init__(self) -> None: object.__setattr__( self, "installation_id", validate_identifier(self.installation_id, "installation_id"), ) if not isinstance(self.public_key_der, bytes): raise TypeError("public_key_der must be bytes") expected_thumbprint = device_public_key_thumbprint(self.public_key_der) normalized_thumbprint = validate_identifier(self.key_thumbprint, "key_thumbprint") if normalized_thumbprint != expected_thumbprint: raise ValueError("key_thumbprint does not match public_key_der") object.__setattr__(self, "key_thumbprint", normalized_thumbprint) object.__setattr__( self, "key_provider", validate_identifier(self.key_provider, "key_provider"), ) if not isinstance(self.friendly_name, str): raise TypeError("friendly_name must be a string") friendly_name = self.friendly_name.strip() if not friendly_name or len(friendly_name) > MAX_FRIENDLY_DEVICE_NAME_CHARACTERS: raise ValueError("friendly_name is empty or too long") object.__setattr__(self, "friendly_name", friendly_name) if not isinstance(self.evidence, Mapping): raise TypeError("evidence must be a mapping") @dataclass(frozen=True) class SerialGenerationResult: batch_id: str sku_id: str serials: tuple[str, ...] surface_profile_id: Optional[str] = None trial_duration_hours: Optional[int] = None @dataclass(frozen=True) class LicenseMutationResult: license_id: str activation_id: str device_id: str document: SignedLicenseDocument recovered_existing_device: bool idempotent_replay: bool = False def to_response_mapping(self) -> dict[str, object]: return { "licenseId": self.license_id, "activationId": self.activation_id, "deviceId": self.device_id, "recoveredExistingDevice": self.recovered_existing_device, "licenseDocument": self.document.to_mapping(), } 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 _normalized_surface_selection( kind: str, values: Iterable[str], *, require_nonempty: bool = False, ) -> tuple[str, ...]: if isinstance(values, (str, bytes, bytearray)): raise TypeError("surface selection must be an iterable of identifiers") normalized = tuple(validate_identifier(value, "surface_id") for value in values) if len(set(normalized)) != len(normalized): raise ValueError("surface selection must not contain duplicate identifiers") return validate_product_surface_ids( kind, tuple(sorted(normalized)), require_nonempty=require_nonempty, ) def _support_reason( reason_code: str, note: Optional[str], allowed_reasons: frozenset[str], ) -> tuple[str, Optional[str]]: normalized_code = validate_identifier(reason_code, "reason_code") if not isinstance(allowed_reasons, frozenset) or not all( isinstance(value, str) and value for value in allowed_reasons ): raise TypeError("allowed_reasons must be a frozenset of non-empty strings") if normalized_code not in allowed_reasons: raise ValueError("reason_code is not allowed for this support action") normalized_note = ( None if note is None else _bounded_text(note, "note", MAX_REASON_CHARACTERS) ) return normalized_code, normalized_note class LicensingService: def __init__( self, session: Session, catalog: LicensingCatalog, signer: SnapshotSigner, serial_pepper: bytes, fingerprint_pepper: bytes, *, now_factory: Callable[[], datetime] = lambda: datetime.now(timezone.utc), ) -> None: if not isinstance(session, Session): raise TypeError("session must be a SQLAlchemy Session") if not isinstance(catalog, LicensingCatalog): raise TypeError("catalog must be a LicensingCatalog") if not isinstance(serial_pepper, bytes): raise TypeError("serial_pepper must be bytes") if not isinstance(fingerprint_pepper, bytes): raise TypeError("fingerprint_pepper must be bytes") if not callable(now_factory): raise TypeError("now_factory must be callable") self._session = session self._catalog = catalog self._signer = signer self._serial_pepper = serial_pepper self._fingerprint_pepper = fingerprint_pepper self._now_factory = now_factory def _now(self) -> datetime: return _utc(self._now_factory()) def _audit( self, action: str, target_type: str, target_id: Optional[str], *, actor_id: Optional[str], correlation_id: str, reason: Optional[str] = None, metadata: Optional[dict[str, object]] = None, ) -> None: normalized_action = validate_identifier(action, "audit action") normalized_target_type = validate_identifier(target_type, "audit target type") normalized_correlation = validate_identifier(correlation_id, "correlation_id") if reason is not None: reason = _bounded_text(reason, "reason", MAX_REASON_CHARACTERS) self._session.add( AuditEvent( id=new_identifier("audit"), actor_type="user" if actor_id is not None else "anonymous", actor_id=actor_id, action=normalized_action, target_type=normalized_target_type, target_id=target_id, reason=reason, correlation_id=normalized_correlation, source_address_digest=None, metadata_json={} if metadata is None else metadata, ) ) def _require_server_admin(self, actor_user_id: str) -> User: normalized_actor = validate_identifier(actor_user_id, "actor_user_id") actor = self._session.get(User, normalized_actor) if actor is None or actor.status != "active" or not actor.is_server_admin: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "support administration requires an active administrator", status_code=403, ) return actor @staticmethod def _surface_profile_mapping( profile: ProductSurfaceProfile, ) -> dict[str, object]: if not isinstance(profile, ProductSurfaceProfile): raise TypeError("profile must be a ProductSurfaceProfile") policy = ProductSurfacePolicy( profile_id=profile.id, inventory_revision=profile.inventory_revision, inventory_sha256=profile.inventory_sha256, analysis_ids=tuple(profile.analysis_ids_json), main_tab_ids=tuple(profile.main_tab_ids_json), ) return { "profileId": profile.id, "productId": profile.product_id, "name": profile.name, "description": profile.description, "status": profile.status, "createdByUserId": profile.created_by_user_id, "createdAt": _utc(profile.created_at).isoformat(), "updatedAt": _utc(profile.updated_at).isoformat(), "policy": policy.to_mapping(), } def product_surface_inventory( self, actor_user_id: str, ) -> dict[str, object]: normalized_actor = validate_identifier(actor_user_id, "actor_user_id") with self._session.begin(): self._require_server_admin(normalized_actor) result = product_surface_inventory_mapping() result["sha256"] = PRODUCT_SURFACE_INVENTORY_SHA256 return result def list_surface_profiles( self, actor_user_id: str, *, include_archived: bool = False, ) -> dict[str, object]: normalized_actor = validate_identifier(actor_user_id, "actor_user_id") if not isinstance(include_archived, bool): raise TypeError("include_archived must be a Boolean") with self._session.begin(): self._require_server_admin(normalized_actor) query = select(ProductSurfaceProfile).where( ProductSurfaceProfile.product_id == PRODUCT_ID ) if not include_archived: query = query.where(ProductSurfaceProfile.status == "active") profiles = tuple( self._session.scalars( query.order_by(ProductSurfaceProfile.name, ProductSurfaceProfile.id) ) ) return { "inventoryRevision": PRODUCT_SURFACE_INVENTORY_REVISION, "inventorySha256": PRODUCT_SURFACE_INVENTORY_SHA256, "profiles": [ self._surface_profile_mapping(profile) for profile in profiles ], } def create_surface_profile( self, name: str, description: str, analysis_ids: Iterable[str], main_tab_ids: Iterable[str], actor_user_id: str, reason: str, correlation_id: str, ) -> dict[str, object]: normalized_name = _bounded_text(name, "name", 256) normalized_description = _bounded_text(description, "description", 512) normalized_analysis_ids = _normalized_surface_selection( PRODUCT_SURFACE_KIND_ANALYSIS, analysis_ids, ) normalized_main_tab_ids = _normalized_surface_selection( PRODUCT_SURFACE_KIND_MAIN_TAB, main_tab_ids, require_nonempty=True, ) normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_reason = _bounded_text(reason, "reason", MAX_REASON_CHARACTERS) normalized_correlation = validate_identifier(correlation_id, "correlation_id") with self._session.begin(): actor = self._require_server_admin(normalized_actor) existing = self._session.scalar( select(ProductSurfaceProfile.id).where( ProductSurfaceProfile.product_id == PRODUCT_ID, ProductSurfaceProfile.name == normalized_name, ) ) if existing is not None: raise ServerLicensingError( ServerErrorCode.CONFLICT, "a product surface profile with this name already exists", status_code=409, existing_resource_id=existing, ) profile = ProductSurfaceProfile( id=new_identifier("surface_profile"), product_id=PRODUCT_ID, name=normalized_name, description=normalized_description, status="active", inventory_revision=PRODUCT_SURFACE_INVENTORY_REVISION, inventory_sha256=PRODUCT_SURFACE_INVENTORY_SHA256, analysis_ids_json=list(normalized_analysis_ids), main_tab_ids_json=list(normalized_main_tab_ids), created_by_user_id=actor.id, ) self._session.add(profile) self._session.flush() self._audit( "surface_profile.created", "product_surface_profile", profile.id, actor_id=actor.id, correlation_id=normalized_correlation, reason=normalized_reason, metadata={ "analysisCount": len(normalized_analysis_ids), "mainTabCount": len(normalized_main_tab_ids), "inventoryRevision": PRODUCT_SURFACE_INVENTORY_REVISION, "inventorySha256": PRODUCT_SURFACE_INVENTORY_SHA256, }, ) return self._surface_profile_mapping(profile) def archive_surface_profile( self, profile_id: str, actor_user_id: str, reason: str, correlation_id: str, ) -> dict[str, object]: normalized_profile = validate_identifier(profile_id, "profile_id") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_reason = _bounded_text(reason, "reason", MAX_REASON_CHARACTERS) normalized_correlation = validate_identifier(correlation_id, "correlation_id") with self._session.begin(): actor = self._require_server_admin(normalized_actor) profile = self._session.scalar( select(ProductSurfaceProfile) .where(ProductSurfaceProfile.id == normalized_profile) .with_for_update() ) if profile is None or profile.product_id != PRODUCT_ID: raise ServerLicensingError( ServerErrorCode.SURFACE_PROFILE_NOT_FOUND, "product surface profile was not found", status_code=404, ) profile.status = "archived" self._audit( "surface_profile.archived", "product_surface_profile", profile.id, actor_id=actor.id, correlation_id=normalized_correlation, reason=normalized_reason, metadata={}, ) self._session.flush() return self._surface_profile_mapping(profile) def assign_license_surface_profile( self, license_id: str, profile_id: Optional[str], actor_user_id: str, reason: str, correlation_id: str, ) -> dict[str, object]: normalized_license = validate_identifier(license_id, "license_id") normalized_profile = ( None if profile_id is None else validate_identifier(profile_id, "profile_id") ) normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_reason = _bounded_text(reason, "reason", MAX_REASON_CHARACTERS) normalized_correlation = validate_identifier(correlation_id, "correlation_id") with self._session.begin(): actor = self._require_server_admin(normalized_actor) license_row = self._session.scalar( select(License) .where(License.id == normalized_license) .with_for_update() ) if license_row is None or license_row.product_id != PRODUCT_ID: raise ServerLicensingError( ServerErrorCode.LICENSE_NOT_FOUND, "license was not found", status_code=404, ) profile = None if normalized_profile is not None: profile = self._session.get(ProductSurfaceProfile, normalized_profile) if ( profile is None or profile.product_id != PRODUCT_ID or profile.status != "active" ): raise ServerLicensingError( ServerErrorCode.SURFACE_PROFILE_NOT_FOUND, "active product surface profile was not found", status_code=404, ) self._surface_profile_mapping(profile) changed = license_row.surface_profile_id != normalized_profile previous_profile_id = license_row.surface_profile_id if changed: license_row.surface_profile_id = normalized_profile license_row.revocation_generation += 1 self._audit( "license.surface_profile_assigned", "license", license_row.id, actor_id=actor.id, correlation_id=normalized_correlation, reason=normalized_reason, metadata={ "changed": changed, "previousProfileId": previous_profile_id, "surfaceProfileId": normalized_profile, "revocationGeneration": license_row.revocation_generation, }, ) return { "licenseId": license_row.id, "surfaceProfileId": license_row.surface_profile_id, "surfaceProfile": ( None if profile is None else self._surface_profile_mapping(profile) ), "changed": changed, "revocationGeneration": license_row.revocation_generation, } def generate_serial_batch( self, sku_id: str, quantity: int, actor_user_id: str, reason: str, correlation_id: str, *, campaign: Optional[str] = None, redemption_deadline: Optional[datetime] = None, plaintext_exporter: Optional[Callable[[SerialGenerationResult], None]] = None, surface_profile_id: Optional[str] = None, trial_duration_hours: Optional[int] = None, ) -> SerialGenerationResult: sku = self._catalog.require_sku(sku_id) if sku.kind not in (SkuKind.EDITION, SkuKind.ADDON, SkuKind.TRIAL): raise ServerLicensingError( ServerErrorCode.SKU_NOT_AVAILABLE, "serials can be generated only for editions, add-ons, and trials", ) if isinstance(quantity, bool) or not isinstance(quantity, int): raise TypeError("quantity must be an integer") if quantity < 1 or quantity > MAX_SERIAL_BATCH_QUANTITY: raise ValueError( f"quantity must be between 1 and {MAX_SERIAL_BATCH_QUANTITY}" ) normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_reason = _bounded_text(reason, "reason", MAX_REASON_CHARACTERS) normalized_correlation = validate_identifier(correlation_id, "correlation_id") normalized_campaign = None if campaign is not None: normalized_campaign = _bounded_text(campaign, "campaign", 256) deadline = None if redemption_deadline is None else _utc(redemption_deadline) if deadline is not None and deadline <= self._now(): raise ValueError("redemption_deadline must be in the future") if plaintext_exporter is not None and not callable(plaintext_exporter): raise TypeError("plaintext_exporter must be callable or None") normalized_surface_profile = ( None if surface_profile_id is None else validate_identifier(surface_profile_id, "surface_profile_id") ) if trial_duration_hours is not None and ( isinstance(trial_duration_hours, bool) or not isinstance(trial_duration_hours, int) ): raise TypeError("trial_duration_hours must be an integer or None") normalized_trial_duration = trial_duration_hours if sku.kind is SkuKind.TRIAL: if normalized_trial_duration is None: raise ValueError("trial serial generation requires trial_duration_hours") if not ( TRIAL_MINIMUM_AUTHORITY_DURATION_HOURS <= normalized_trial_duration <= TRIAL_MAXIMUM_TOTAL_DURATION_HOURS ): raise ValueError( "trial_duration_hours must be within the shipped trial ceiling" ) elif normalized_trial_duration is not None: raise ValueError("trial_duration_hours is valid only for a trial SKU") if sku.kind is SkuKind.ADDON and normalized_surface_profile is not None: raise ValueError("add-on serial batches cannot replace a license surface profile") plaintext_serials: list[str] = [] result: Optional[SerialGenerationResult] = None with self._session.begin(): actor = self._session.get(User, normalized_actor) if actor is None or actor.status != "active" or not actor.is_server_admin: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "serial generation requires an active administrator", status_code=403, ) profile = None if normalized_surface_profile is not None: profile = self._session.get( ProductSurfaceProfile, normalized_surface_profile, ) if ( profile is None or profile.product_id != PRODUCT_ID or profile.status != "active" ): raise ServerLicensingError( ServerErrorCode.SURFACE_PROFILE_NOT_FOUND, "active product surface profile was not found", status_code=404, ) self._surface_profile_mapping(profile) batch = SerialBatch( id=new_identifier("batch"), sku_id=sku.sku_id, quantity=quantity, campaign=normalized_campaign, reason=normalized_reason, created_by_user_id=actor.id, surface_profile_id=normalized_surface_profile, trial_duration_hours=normalized_trial_duration, status="active", redemption_deadline=deadline, ) self._session.add(batch) for _index in range(quantity): parsed = self._create_unique_serial() self._session.add( Serial( id=new_identifier("serial"), batch_id=batch.id, lookup_id=parsed.lookup_id, secret_digest=serial_secret_digest( self._serial_pepper, parsed.compact, ), status="active", ) ) plaintext_serials.append(parsed.formatted) self._audit( "serial.batch_generated", "serial_batch", batch.id, actor_id=actor.id, correlation_id=normalized_correlation, reason=normalized_reason, metadata={ "quantity": quantity, "skuId": sku.sku_id, **( {"surfaceProfileId": normalized_surface_profile} if normalized_surface_profile is not None else {} ), **( {"trialDurationHours": normalized_trial_duration} if normalized_trial_duration is not None else {} ), }, ) result = SerialGenerationResult( batch_id=batch.id, sku_id=sku.sku_id, serials=tuple(plaintext_serials), surface_profile_id=normalized_surface_profile, trial_duration_hours=normalized_trial_duration, ) if plaintext_exporter is not None: plaintext_exporter(result) if result is None: raise RuntimeError("serial generation completed without a result") return result def _create_unique_serial(self) -> ParsedSerial: for _attempt in range(100): lookup_id = random_token(SERIAL_LOOKUP_CHARACTERS, SERIAL_ALPHABET) exists = self._session.scalar( select(func.count()).select_from(Serial).where(Serial.lookup_id == lookup_id) ) if exists: continue secret = random_token(SERIAL_SECRET_CHARACTERS, SERIAL_ALPHABET) checksum = serial_checksum(SERIAL_PRODUCT_PREFIX, lookup_id, secret) return ParsedSerial(SERIAL_PRODUCT_PREFIX, lookup_id, secret, checksum) raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "could not allocate a unique serial lookup identifier", status_code=500, retryable=True, ) def redeem_serial( self, serial_value: str, enrollment: DeviceEnrollment, application_major_version: int, idempotency_key: str, correlation_id: str, *, actor_user_id: Optional[str] = None, target_license_id: Optional[str] = None, offline_certificate: bool = False, offline_request_nonce: Optional[str] = None, expected_sku_id: Optional[str] = None, ) -> LicenseMutationResult: try: parsed = parse_serial(serial_value) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.SERIAL_INVALID, "serial format or checksum is invalid", ) from exc if not isinstance(enrollment, DeviceEnrollment): raise TypeError("enrollment must be a DeviceEnrollment") 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_idempotency = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_correlation = validate_identifier(correlation_id, "correlation_id") normalized_actor = ( None if actor_user_id is None else validate_identifier(actor_user_id, "actor_user_id") ) normalized_target = ( None if target_license_id is None else validate_identifier(target_license_id, "target_license_id") ) if not isinstance(offline_certificate, bool): raise TypeError("offline_certificate must be a Boolean") normalized_offline_nonce = None if offline_request_nonce is not None: normalized_offline_nonce = _bounded_text( offline_request_nonce, "offline_request_nonce", 256, ) if offline_certificate != (normalized_offline_nonce is not None): raise ValueError( "offline certificates require exactly one offline request nonce" ) normalized_expected_sku = ( None if expected_sku_id is None else validate_identifier(expected_sku_id, "expected_sku_id") ) subject_key = normalized_actor or enrollment.installation_id request_digest = self._request_digest( { "serialDigest": serial_secret_digest( self._serial_pepper, parsed.compact, ).hex(), "installationId": enrollment.installation_id, "applicationMajorVersion": application_major_version, "targetLicenseId": normalized_target, "offlineCertificate": offline_certificate, "offlineRequestNonce": normalized_offline_nonce, "expectedSkuId": normalized_expected_sku, } ) now = self._now() with self._session.begin(): replay = self._load_idempotent_response( "serial.redeem", subject_key, normalized_idempotency, request_digest, now, ) if replay is not None: return self._mutation_from_mapping(replay, idempotent_replay=True) serial_row = self._session.scalar( select(Serial) .where(Serial.lookup_id == parsed.lookup_id) .with_for_update() ) if serial_row is None or not hmac.compare_digest( serial_row.secret_digest, serial_secret_digest(self._serial_pepper, parsed.compact), ): raise ServerLicensingError( ServerErrorCode.SERIAL_NOT_FOUND, "serial was not found", status_code=404, ) if serial_row.status == "redeemed": raise ServerLicensingError( ServerErrorCode.SERIAL_ALREADY_REDEEMED, "serial has already been redeemed; sign in to recover the license", status_code=409, existing_resource_id=serial_row.redeemed_license_id, ) if serial_row.status != "active": raise ServerLicensingError( ServerErrorCode.SERIAL_INVALID, "serial is not active", status_code=409, ) batch = self._session.get(SerialBatch, serial_row.batch_id) if batch is None or batch.status != "active": raise ServerLicensingError( ServerErrorCode.SERIAL_INVALID, "serial batch is not active", status_code=409, ) if batch.redemption_deadline is not None and _utc(batch.redemption_deadline) <= now: raise ServerLicensingError( ServerErrorCode.SERIAL_EXPIRED, "serial redemption deadline has passed", status_code=409, ) sku = self._catalog.require_sku(batch.sku_id) if normalized_expected_sku is not None and sku.sku_id != normalized_expected_sku: raise ServerLicensingError( ServerErrorCode.SKU_NOT_AVAILABLE, "serial SKU does not match the signed offline request", status_code=409, ) if not ( sku.application_major_minimum <= application_major_version <= sku.application_major_maximum ): raise ServerLicensingError( ServerErrorCode.SKU_NOT_AVAILABLE, "serial does not cover this application major version", status_code=409, ) actor = None if normalized_actor is None else self._session.get(User, normalized_actor) if sku.account_required and (actor is None or actor.status != "active"): raise ServerLicensingError( ServerErrorCode.ACCOUNT_REQUIRED, "this SKU must be claimed by a signed-in account", status_code=401, ) if serial_row.reserved_user_id is not None: if actor is None or actor.id != serial_row.reserved_user_id: raise ServerLicensingError( ServerErrorCode.SERIAL_RESERVED, "serial is reserved for another account", status_code=403, ) installation, recovered = self._enroll_device(enrollment, actor, now) if sku.kind is SkuKind.TRIAL: if normalized_target is not None: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, "trial serial redemption cannot target an existing license", ) license_row, grant = self._create_trial_from_serial( sku, batch, serial_row, actor, enrollment, installation, now, ) else: license_row = self._resolve_license_for_sku( sku, actor, installation, normalized_target, surface_profile_id=batch.surface_profile_id, ) grant = Grant( id=new_identifier("grant"), license_id=license_row.id, sku_id=sku.sku_id, source_type="serial_redemption", source_reference=serial_row.id, catalog_revision=self._catalog.revision, starts_at=now, ends_at=None, status="active", metadata_json={}, ) self._session.add(grant) self._session.flush() activation = self._ensure_activation(license_row, installation, now) document = self._issue_snapshot( license_row, activation, installation, application_major_version, now, offline_certificate=offline_certificate, request_nonce=normalized_offline_nonce, ) serial_row.status = "redeemed" serial_row.redeemed_at = now serial_row.redeemed_by_user_id = None if actor is None else actor.id serial_row.redeemed_license_id = license_row.id redemption = SerialRedemption( id=new_identifier("redemption"), serial_id=serial_row.id, license_id=license_row.id, grant_id=grant.id, activation_id=activation.id, actor_user_id=None if actor is None else actor.id, idempotency_key=normalized_idempotency, snapshot_id=document.payload["snapshotId"], ) self._session.add(redemption) result = LicenseMutationResult( license_id=license_row.id, activation_id=activation.id, device_id=installation.device_id, document=document, recovered_existing_device=recovered, ) self._store_idempotent_response( "serial.redeem", subject_key, normalized_idempotency, request_digest, result.to_response_mapping(), now, ) self._audit( "serial.redeemed", "license", license_row.id, actor_id=None if actor is None else actor.id, correlation_id=normalized_correlation, metadata={ "skuId": sku.sku_id, "serialLookupId": parsed.lookup_id, "deviceId": installation.device_id, }, ) return result def start_trial( self, enrollment: DeviceEnrollment, application_major_version: int, idempotency_key: str, correlation_id: str, *, actor_user_id: Optional[str] = None, offline_certificate: bool = False, offline_request_nonce: Optional[str] = None, ) -> LicenseMutationResult: if not isinstance(enrollment, DeviceEnrollment): raise TypeError("enrollment must be a DeviceEnrollment") 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_key = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_correlation = validate_identifier(correlation_id, "correlation_id") normalized_actor = ( None if actor_user_id is None else validate_identifier(actor_user_id, "actor_user_id") ) if not isinstance(offline_certificate, bool): raise TypeError("offline_certificate must be a Boolean") normalized_offline_nonce = None if offline_request_nonce is None else _bounded_text( offline_request_nonce, "offline_request_nonce", 256 ) if offline_certificate != (normalized_offline_nonce is not None): raise ValueError("offline certificates require exactly one offline request nonce") subject_text = self._trial_subject_key(normalized_actor, enrollment) subject_digest = anonymous_subject_digest(self._fingerprint_pepper, subject_text) request_digest = self._request_digest( { "subjectDigest": subject_digest.hex(), "installationId": enrollment.installation_id, "applicationMajorVersion": application_major_version, **({"offlineCertificate": True, "offlineRequestNonce": normalized_offline_nonce} if offline_certificate else {}), } ) now = self._now() trial_sku = self._catalog.require_sku("trial.full") if not ( trial_sku.application_major_minimum <= application_major_version <= trial_sku.application_major_maximum ): raise ServerLicensingError( ServerErrorCode.SKU_NOT_AVAILABLE, "trial does not cover this application major version", status_code=409, ) with self._session.begin(): replay = self._load_idempotent_response( "trial.start", subject_text, normalized_key, request_digest, now, ) if replay is not None: return self._mutation_from_mapping(replay, idempotent_replay=True) actor = None if normalized_actor is None else self._session.get(User, normalized_actor) if normalized_actor is not None and (actor is None or actor.status != "active"): raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "trial account is not active", status_code=401, ) installation, recovered = self._enroll_device(enrollment, actor, now) trial = self._session.scalar( select(Trial).where(Trial.subject_digest == subject_digest).with_for_update() ) if trial is None and actor is None: trial = self._session.scalar( select(Trial) .join(License, License.id == Trial.license_id) .join(Activation, Activation.license_id == Trial.license_id) .where( Activation.device_id == installation.device_id, License.owner_user_id.is_(None), License.owner_organization_id.is_(None), ) .order_by(Trial.starts_at, Trial.id) .with_for_update() ) if trial is None: policy = self._catalog.device_policies[trial_sku.device_policy_id] trial_duration_hours = policy.trial_days * HOURS_PER_DAY if not ( TRIAL_MINIMUM_AUTHORITY_DURATION_HOURS <= trial_duration_hours <= TRIAL_MAXIMUM_TOTAL_DURATION_HOURS ): raise ServerLicensingError( ServerErrorCode.CATALOG_UNAVAILABLE, "trial policy exceeds the duration accepted by shipped clients", status_code=503, ) license_row = License( id=new_identifier("license"), product_id=PRODUCT_ID, owner_user_id=None if actor is None else actor.id, owner_organization_id=None, anonymous_subject_digest=None if actor is not None else subject_digest, subject_type="user" if actor is not None else "license", status="active", device_policy_id=policy.policy_id, surface_profile_id=None, revocation_generation=0, ) self._session.add(license_row) grant = Grant( id=new_identifier("grant"), license_id=license_row.id, sku_id=trial_sku.sku_id, source_type="trial", source_reference=new_identifier("trial_source"), catalog_revision=self._catalog.revision, starts_at=now, ends_at=now + timedelta(hours=trial_duration_hours), status="active", metadata_json={}, ) self._session.add(grant) trial = Trial( id=new_identifier("trial"), license_id=license_row.id, grant_id=grant.id, subject_digest=subject_digest, starts_at=now, ends_at=grant.ends_at, status="active", extension_count=0, ) self._session.add(trial) self._session.flush() else: license_row = self._session.get(License, trial.license_id) if license_row is None: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "trial license record is missing", status_code=500, ) if trial.status != "active" or _utc(trial.ends_at) <= now: raise ServerLicensingError( ServerErrorCode.TRIAL_NOT_ELIGIBLE, "the original trial period has ended", status_code=409, ) activation = self._ensure_activation(license_row, installation, now) document = self._issue_snapshot( license_row, activation, installation, application_major_version, now, offline_certificate=offline_certificate, request_nonce=normalized_offline_nonce, ) result = LicenseMutationResult( license_id=license_row.id, activation_id=activation.id, device_id=installation.device_id, document=document, recovered_existing_device=recovered, ) self._store_idempotent_response( "trial.start", subject_text, normalized_key, request_digest, result.to_response_mapping(), now, ) self._audit( "trial.started_or_recovered", "trial", trial.id, actor_id=None if actor is None else actor.id, correlation_id=normalized_correlation, metadata={"deviceId": installation.device_id}, ) return result def inspect_license( self, license_id: str, actor_user_id: str, ) -> dict[str, object]: normalized_license = validate_identifier(license_id, "license_id") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") now = self._now() with self._session.begin(): self._require_server_admin(normalized_actor) license_row = self._session.get(License, normalized_license) if license_row is None: raise ServerLicensingError( ServerErrorCode.LICENSE_NOT_FOUND, "license was not found", status_code=404, ) return self._support_license_mapping(license_row, now) def preview_device_recovery( self, device_id: str, actor_user_id: str, reason_code: str, correlation_id: str, *, note: Optional[str] = None, ) -> dict[str, object]: normalized_device = validate_identifier(device_id, "device_id") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_reason, _normalized_note = _support_reason( reason_code, note, SUPPORT_DEVICE_RECOVERY_REASONS, ) normalized_correlation = validate_identifier(correlation_id, "correlation_id") with self._session.begin(): self._require_server_admin(normalized_actor) device = self._session.get(Device, normalized_device) if device is None: raise ServerLicensingError( ServerErrorCode.DEVICE_NOT_FOUND, "device was not found", status_code=404, ) result = self._device_recovery_preview_mapping(device) result["reasonCode"] = normalized_reason result["correlationId"] = normalized_correlation return result def recover_device( self, device_id: str, actor_user_id: str, reason_code: str, expected_state_digest: str, idempotency_key: str, correlation_id: str, *, note: Optional[str] = None, ) -> dict[str, object]: normalized_device = validate_identifier(device_id, "device_id") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_reason, normalized_note = _support_reason( reason_code, note, SUPPORT_DEVICE_RECOVERY_REASONS, ) if ( not isinstance(expected_state_digest, str) or len(expected_state_digest) != 64 or any(character not in "0123456789abcdef" for character in expected_state_digest) ): raise ValueError("expected_state_digest must be a lowercase SHA-256 digest") normalized_key = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_correlation = validate_identifier(correlation_id, "correlation_id") request_digest = self._request_digest( { "deviceId": normalized_device, "actorUserId": normalized_actor, "reasonCode": normalized_reason, "note": normalized_note, "expectedStateDigest": expected_state_digest, } ) now = self._now() with self._session.begin(): actor = self._require_server_admin(normalized_actor) replay = self._load_idempotent_response( "admin.device.recover", normalized_device, normalized_key, request_digest, now, ) if replay is not None: result = dict(replay) result["idempotentReplay"] = True return result device = self._session.scalar( select(Device).where(Device.id == normalized_device).with_for_update() ) if device is None: raise ServerLicensingError( ServerErrorCode.DEVICE_NOT_FOUND, "device was not found", status_code=404, ) tuple( self._session.scalars( select(DeviceInstallation) .where(DeviceInstallation.device_id == device.id) .with_for_update() ) ) tuple( self._session.scalars( select(Activation) .where(Activation.device_id == device.id) .with_for_update() ) ) preview = self._device_recovery_preview_mapping(device) if preview["stateDigest"] != expected_state_digest: raise ServerLicensingError( ServerErrorCode.CONFLICT, "device state changed after the recovery preview; preview again", status_code=409, ) device.status = "active" result = { **preview, "previousStatus": "lost", "currentStatus": device.status, "reasonCode": normalized_reason, "correlationId": normalized_correlation, "recoveredAt": now.isoformat(), "executed": True, "idempotentReplay": False, } self._store_idempotent_response( "admin.device.recover", device.id, normalized_key, request_digest, result, now, ) self._audit( "device.recovery_approved", "device", device.id, actor_id=actor.id, correlation_id=normalized_correlation, reason=normalized_note or normalized_reason, metadata={ "reasonCode": normalized_reason, "revokedInstallationCount": preview["revokedInstallationCount"], "activeActivationCount": preview["activeActivationCount"], "activeLicenseIds": preview["activeLicenseIds"], "previewStateDigest": expected_state_digest, "requiresNewDeviceProofKey": True, }, ) return result def preview_trial_extension( self, trial_id: str, actor_user_id: str, reason_code: str, correlation_id: str, *, note: Optional[str] = None, ) -> dict[str, object]: normalized_trial = validate_identifier(trial_id, "trial_id") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_reason, _normalized_note = _support_reason( reason_code, note, SUPPORT_TRIAL_EXTENSION_REASONS, ) normalized_correlation = validate_identifier(correlation_id, "correlation_id") now = self._now() with self._session.begin(): self._require_server_admin(normalized_actor) trial = self._session.get(Trial, normalized_trial) if trial is None: raise ServerLicensingError( ServerErrorCode.TRIAL_NOT_ELIGIBLE, "trial was not found", status_code=404, ) license_row = self._session.get(License, trial.license_id) if license_row is None: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "trial license record is missing", status_code=500, ) result = self._trial_extension_preview_mapping(trial, license_row, now) result["reasonCode"] = normalized_reason result["correlationId"] = normalized_correlation return result def extend_trial( self, trial_id: str, actor_user_id: str, reason_code: str, expected_revocation_generation: int, idempotency_key: str, correlation_id: str, *, note: Optional[str] = None, ) -> dict[str, object]: normalized_trial = validate_identifier(trial_id, "trial_id") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_reason, normalized_note = _support_reason( reason_code, note, SUPPORT_TRIAL_EXTENSION_REASONS, ) if isinstance(expected_revocation_generation, bool) or not isinstance( expected_revocation_generation, int, ): raise TypeError("expected_revocation_generation must be an integer") if expected_revocation_generation < 0: raise ValueError("expected_revocation_generation must not be negative") normalized_key = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_correlation = validate_identifier(correlation_id, "correlation_id") request_digest = self._request_digest( { "trialId": normalized_trial, "actorUserId": normalized_actor, "reasonCode": normalized_reason, "note": normalized_note, "expectedRevocationGeneration": expected_revocation_generation, } ) now = self._now() with self._session.begin(): actor = self._require_server_admin(normalized_actor) replay = self._load_idempotent_response( "admin.trial.extend", normalized_trial, normalized_key, request_digest, now, ) if replay is not None: result = dict(replay) result["idempotentReplay"] = True return result trial = self._session.scalar( select(Trial).where(Trial.id == normalized_trial).with_for_update() ) if trial is None: raise ServerLicensingError( ServerErrorCode.TRIAL_NOT_ELIGIBLE, "trial was not found", status_code=404, ) license_row = self._session.scalar( select(License) .where(License.id == trial.license_id) .with_for_update() ) if license_row is None: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "trial license record is missing", status_code=500, ) if license_row.revocation_generation != expected_revocation_generation: raise ServerLicensingError( ServerErrorCode.CONFLICT, "license grants changed after the extension preview; preview again", status_code=409, ) preview = self._trial_extension_preview_mapping(trial, license_row, now) extension_grant = Grant( id=new_identifier("grant"), license_id=license_row.id, sku_id=preview["skuId"], source_type="trial_extension", source_reference=( f"support-trial-extension:{trial.id}:{preview['extensionNumber']}" ), catalog_revision=self._catalog.revision, starts_at=datetime.fromisoformat( str(preview["extensionStartsAt"]).replace("Z", "+00:00") ), ends_at=datetime.fromisoformat( str(preview["extensionEndsAt"]).replace("Z", "+00:00") ), status="active", metadata_json={ "trialId": trial.id, "extensionNumber": preview["extensionNumber"], "reasonCode": normalized_reason, "grantedByUserId": actor.id, }, ) self._session.add(extension_grant) trial.extension_count += 1 trial.ends_at = extension_grant.ends_at trial.status = "active" if license_row.status == "revoked": license_row.status = "active" license_row.revocation_generation += 1 result = { **preview, "extensionGrantId": extension_grant.id, "reasonCode": normalized_reason, "correlationId": normalized_correlation, "revocationGeneration": license_row.revocation_generation, "executed": True, "idempotentReplay": False, } self._store_idempotent_response( "admin.trial.extend", normalized_trial, normalized_key, request_digest, result, now, ) self._audit( "trial.extended", "trial", trial.id, actor_id=actor.id, correlation_id=normalized_correlation, reason=normalized_note or normalized_reason, metadata={ "reasonCode": normalized_reason, "extensionGrantId": extension_grant.id, "extensionNumber": trial.extension_count, "extensionDays": preview["extensionDays"], "revocationGeneration": license_row.revocation_generation, }, ) return result def preview_grant_revocation( self, license_id: str, grant_id: str, actor_user_id: str, reason_code: str, correlation_id: str, *, note: Optional[str] = None, ) -> dict[str, object]: normalized_license = validate_identifier(license_id, "license_id") normalized_grant = validate_identifier(grant_id, "grant_id") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_reason, _normalized_note = _support_reason( reason_code, note, SUPPORT_GRANT_REVOCATION_REASONS, ) normalized_correlation = validate_identifier(correlation_id, "correlation_id") now = self._now() with self._session.begin(): self._require_server_admin(normalized_actor) license_row = self._session.get(License, normalized_license) if license_row is None: raise ServerLicensingError( ServerErrorCode.LICENSE_NOT_FOUND, "license was not found", status_code=404, ) grant = self._session.get(Grant, normalized_grant) if grant is None or grant.license_id != license_row.id: raise ServerLicensingError( ServerErrorCode.GRANT_NOT_ACTIVE, "grant was not found on this license", status_code=404, ) result = self._grant_revocation_preview_mapping(license_row, grant, now) result["reasonCode"] = normalized_reason result["correlationId"] = normalized_correlation return result def revoke_grant( self, license_id: str, grant_id: str, actor_user_id: str, reason_code: str, expected_revocation_generation: int, idempotency_key: str, correlation_id: str, *, note: Optional[str] = None, ) -> dict[str, object]: normalized_license = validate_identifier(license_id, "license_id") normalized_grant = validate_identifier(grant_id, "grant_id") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_reason, normalized_note = _support_reason( reason_code, note, SUPPORT_GRANT_REVOCATION_REASONS, ) if isinstance(expected_revocation_generation, bool) or not isinstance( expected_revocation_generation, int, ): raise TypeError("expected_revocation_generation must be an integer") if expected_revocation_generation < 0: raise ValueError("expected_revocation_generation must not be negative") normalized_key = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_correlation = validate_identifier(correlation_id, "correlation_id") request_digest = self._request_digest( { "licenseId": normalized_license, "grantId": normalized_grant, "actorUserId": normalized_actor, "reasonCode": normalized_reason, "note": normalized_note, "expectedRevocationGeneration": expected_revocation_generation, } ) now = self._now() with self._session.begin(): actor = self._require_server_admin(normalized_actor) replay = self._load_idempotent_response( "admin.grant.revoke", normalized_license, normalized_key, request_digest, now, ) if replay is not None: result = dict(replay) result["idempotentReplay"] = True return result license_row = self._session.scalar( select(License) .where(License.id == normalized_license) .with_for_update() ) if license_row is None: raise ServerLicensingError( ServerErrorCode.LICENSE_NOT_FOUND, "license was not found", status_code=404, ) grant = self._session.scalar( select(Grant) .where( Grant.id == normalized_grant, Grant.license_id == normalized_license, ) .with_for_update() ) if grant is None: raise ServerLicensingError( ServerErrorCode.GRANT_NOT_ACTIVE, "grant was not found on this license", status_code=404, ) if license_row.revocation_generation != expected_revocation_generation: raise ServerLicensingError( ServerErrorCode.CONFLICT, "license grants changed after the revocation preview; preview again", status_code=409, ) preview = self._grant_revocation_preview_mapping(license_row, grant, now) grant.status = "revoked" grant.metadata_json = { **grant.metadata_json, "supportRevocation": { "reasonCode": normalized_reason, "revokedAt": now.isoformat(), "revokedByUserId": actor.id, }, } self._synchronize_trial_after_grant_revocation( license_row.id, grant.id, now, ) if preview["wouldDeactivateLicense"]: license_row.status = "revoked" license_row.revocation_generation += 1 result = { **preview, "reasonCode": normalized_reason, "correlationId": normalized_correlation, "revocationGeneration": license_row.revocation_generation, "executed": True, "idempotentReplay": False, } self._store_idempotent_response( "admin.grant.revoke", normalized_license, normalized_key, request_digest, result, now, ) self._audit( "grant.revoked", "grant", grant.id, actor_id=actor.id, correlation_id=normalized_correlation, reason=normalized_note or normalized_reason, metadata={ "reasonCode": normalized_reason, "licenseId": license_row.id, "skuId": grant.sku_id, "removedEntitlements": preview["removedEntitlements"], "preservedEntitlements": preview["preservedEntitlements"], "revocationGeneration": license_row.revocation_generation, }, ) return result def refresh_activation( self, license_id: str, installation_id: str, public_key_der: bytes, key_thumbprint: str, application_major_version: int, correlation_id: str, evidence: Optional[Mapping[str, str]] = None, ) -> LicenseMutationResult: normalized_license = validate_identifier(license_id, "license_id") normalized_installation = validate_identifier(installation_id, "installation_id") if not isinstance(public_key_der, bytes): raise TypeError("public_key_der must be bytes") normalized_thumbprint = validate_identifier(key_thumbprint, "key_thumbprint") if device_public_key_thumbprint(public_key_der) != normalized_thumbprint: raise ValueError("key_thumbprint does not match public_key_der") 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_correlation = validate_identifier(correlation_id, "correlation_id") now = self._now() with self._session.begin(): installation = self._require_installation_key( normalized_installation, public_key_der, normalized_thumbprint, now, evidence=evidence, ) license_row = self._session.scalar( select(License) .where(License.id == normalized_license) .with_for_update() ) if license_row is None or license_row.status != "active": raise ServerLicensingError( ServerErrorCode.LICENSE_NOT_FOUND, "license was not found or is not active", status_code=404, ) activation = self._session.scalar( select(Activation) .where( Activation.license_id == license_row.id, Activation.device_id == installation.device_id, Activation.status == "active", ) .with_for_update() ) if activation is None: raise ServerLicensingError( ServerErrorCode.DEVICE_NOT_FOUND, "device does not have an active allocation for this license", status_code=404, ) document = self._issue_snapshot( license_row, activation, installation, application_major_version, now, ) self._audit( "activation.refreshed", "activation", activation.id, actor_id=license_row.owner_user_id, correlation_id=normalized_correlation, metadata={"snapshotId": document.payload["snapshotId"]}, ) return LicenseMutationResult( license_id=license_row.id, activation_id=activation.id, device_id=installation.device_id, document=document, recovered_existing_device=True, ) def activate_owned_license( self, license_id: str, enrollment: DeviceEnrollment, application_major_version: int, idempotency_key: str, correlation_id: str, actor_user_id: str, *, offline_certificate: bool = False, offline_request_nonce: Optional[str] = None, expected_device_id: Optional[str] = None, ) -> LicenseMutationResult: normalized_license = validate_identifier(license_id, "license_id") if not isinstance(enrollment, DeviceEnrollment): raise TypeError("enrollment must be a DeviceEnrollment") 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_idempotency = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_correlation = validate_identifier(correlation_id, "correlation_id") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") if not isinstance(offline_certificate, bool): raise TypeError("offline_certificate must be a Boolean") normalized_offline_nonce = ( None if offline_request_nonce is None else _bounded_text(offline_request_nonce, "offline_request_nonce", 256) ) normalized_expected_device = ( None if expected_device_id is None else validate_identifier(expected_device_id, "expected_device_id") ) if offline_certificate != ( normalized_offline_nonce is not None and normalized_expected_device is not None ): raise ValueError( "offline renewal requires a request nonce and existing device ID" ) request_digest = self._request_digest( { "licenseId": normalized_license, "installationId": enrollment.installation_id, "applicationMajorVersion": application_major_version, "deviceKeyThumbprint": enrollment.key_thumbprint, "offlineCertificate": offline_certificate, "offlineRequestNonce": normalized_offline_nonce, "expectedDeviceId": normalized_expected_device, } ) idempotency_scope = ( "account.license.offline_renew" if offline_certificate else "account.license.activate" ) now = self._now() with self._session.begin(): replay = self._load_idempotent_response( idempotency_scope, normalized_actor, normalized_idempotency, request_digest, now, ) if replay is not None: return self._mutation_from_mapping(replay, idempotent_replay=True) actor = self._session.get(User, normalized_actor) if actor is None or actor.status != "active": raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "an active account is required", status_code=403, ) license_row = self._session.scalar( select(License) .where(License.id == normalized_license) .with_for_update() ) if license_row is None or license_row.status != "active": raise ServerLicensingError( ServerErrorCode.LICENSE_NOT_FOUND, "license was not found or is not active", status_code=404, ) authorized = license_row.owner_user_id == actor.id activation_seat_id = actor.id if license_row.owner_organization_id is not None: membership = self._session.scalar( select(Membership).where( Membership.organization_id == license_row.owner_organization_id, Membership.user_id == actor.id, Membership.status == "active", (Membership.valid_until.is_(None) | (Membership.valid_until > now)), ) ) authorized = membership is not None if membership is not None: seats = self._ensure_organization_seat_records( license_row, license_row.owner_organization_id, actor.id, now, assign_if_available=False, ) assigned_seat = next( ( seat for seat in seats if seat.status == "assigned" and seat.assigned_user_id == actor.id ), None, ) if assigned_seat is None: raise ServerLicensingError( ServerErrorCode.SEAT_NOT_ASSIGNED, "an organization administrator must assign a named seat first", status_code=409, ) activation_seat_id = assigned_seat.id if not authorized: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "the authenticated account does not own or hold a seat on this license", status_code=403, ) installation, recovered = self._enroll_device(enrollment, actor, now) if ( normalized_expected_device is not None and installation.device_id != normalized_expected_device ): raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "offline renewal device does not match the enrolled device", status_code=403, ) if offline_certificate: existing_activation = self._session.scalar( select(Activation) .where( Activation.license_id == license_row.id, Activation.device_id == installation.device_id, Activation.status == "active", ) .with_for_update() ) if existing_activation is None: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "offline renewal requires an active existing device allocation", status_code=403, ) if license_row.owner_organization_id is not None: if existing_activation.seat_id != activation_seat_id: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "offline renewal device belongs to another named seat", status_code=403, ) elif existing_activation.seat_id not in (None, activation_seat_id): raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "offline renewal device belongs to another account seat", status_code=403, ) elif existing_activation.seat_id is None: existing_activation.seat_id = activation_seat_id activation = self._ensure_activation( license_row, installation, now, seat_id=activation_seat_id, ) document = self._issue_snapshot( license_row, activation, installation, application_major_version, now, offline_certificate=offline_certificate, request_nonce=normalized_offline_nonce, ) result = LicenseMutationResult( license_id=license_row.id, activation_id=activation.id, device_id=installation.device_id, document=document, recovered_existing_device=recovered, ) self._store_idempotent_response( idempotency_scope, normalized_actor, normalized_idempotency, request_digest, result.to_response_mapping(), now, ) self._audit( ( "account.offline_license_renewed" if offline_certificate else "account.license_activated" ), "activation", activation.id, actor_id=actor.id, correlation_id=normalized_correlation, metadata={ "licenseId": license_row.id, "deviceId": installation.device_id, "offlineCertificate": offline_certificate, }, ) return result def deactivate_device( self, license_id: str, installation_id: str, public_key_der: bytes, key_thumbprint: str, correlation_id: str, *, reason: str = "local_device_release", idempotency_key: Optional[str] = None, expected_device_id: Optional[str] = None, evidence: Optional[Mapping[str, str]] = None, ) -> str: normalized_license = validate_identifier(license_id, "license_id") normalized_installation = validate_identifier(installation_id, "installation_id") if not isinstance(public_key_der, bytes): raise TypeError("public_key_der must be bytes") normalized_thumbprint = validate_identifier(key_thumbprint, "key_thumbprint") if device_public_key_thumbprint(public_key_der) != normalized_thumbprint: raise ValueError("key_thumbprint does not match public_key_der") normalized_correlation = validate_identifier(correlation_id, "correlation_id") normalized_reason = _bounded_text(reason, "reason", MAX_REASON_CHARACTERS) normalized_idempotency = ( None if idempotency_key is None else _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) ) normalized_expected_device = ( None if expected_device_id is None else validate_identifier(expected_device_id, "expected_device_id") ) request_digest = self._request_digest( { "licenseId": normalized_license, "installationId": normalized_installation, "deviceKeyThumbprint": normalized_thumbprint, "reason": normalized_reason, "expectedDeviceId": normalized_expected_device, "evidenceDigests": ( {} if evidence is None else digest_device_evidence(self._fingerprint_pepper, evidence) ), } ) now = self._now() with self._session.begin(): if normalized_idempotency is not None: replay = self._load_idempotent_response( "activation.deactivate", normalized_installation, normalized_idempotency, request_digest, now, ) if replay is not None: return validate_identifier( replay.get("activationId"), "activationId", ) installation = self._require_installation_key( normalized_installation, public_key_der, normalized_thumbprint, now, evidence=evidence, ) if ( normalized_expected_device is not None and installation.device_id != normalized_expected_device ): raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "offline release device does not match the enrolled installation", status_code=403, ) activation = self._session.scalar( select(Activation) .where( Activation.license_id == normalized_license, Activation.device_id == installation.device_id, Activation.status == "active", ) .with_for_update() ) if activation is None: activation = self._session.scalar( select(Activation) .where( Activation.license_id == normalized_license, Activation.device_id == installation.device_id, Activation.status == "deactivated", ) .order_by(Activation.deactivated_at.desc(), Activation.id.desc()) .with_for_update() ) if activation is None: raise ServerLicensingError( ServerErrorCode.DEVICE_NOT_FOUND, "device does not have an allocation for this license", status_code=404, ) else: activation.status = "deactivated" activation.deactivated_at = now activation.release_reason = normalized_reason license_row = self._session.get(License, normalized_license) self._audit( "activation.deactivated", "activation", activation.id, actor_id=None if license_row is None else license_row.owner_user_id, correlation_id=normalized_correlation, reason=normalized_reason, metadata={"deviceId": installation.device_id}, ) if normalized_idempotency is not None: self._store_idempotent_response( "activation.deactivate", normalized_installation, normalized_idempotency, request_digest, {"activationId": activation.id}, now, ) return activation.id def claim_redeemed_serial( self, serial_value: str, actor_user_id: str, correlation_id: str, ) -> str: try: parsed = parse_serial(serial_value) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.SERIAL_INVALID, "serial format or checksum is invalid", ) from exc normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_correlation = validate_identifier(correlation_id, "correlation_id") now = self._now() with self._session.begin(): actor = self._session.get(User, normalized_actor) if actor is None or actor.status != "active": raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "an active account is required to claim a license", status_code=401, ) serial_row = self._session.scalar( select(Serial) .where(Serial.lookup_id == parsed.lookup_id) .with_for_update() ) if serial_row is None or not hmac.compare_digest( serial_row.secret_digest, serial_secret_digest(self._serial_pepper, parsed.compact), ): raise ServerLicensingError( ServerErrorCode.SERIAL_NOT_FOUND, "serial was not found", status_code=404, ) if serial_row.status != "redeemed" or serial_row.redeemed_license_id is None: raise ServerLicensingError( ServerErrorCode.CONFLICT, "serial has not been redeemed and cannot be claimed", status_code=409, ) license_row = self._session.scalar( select(License) .where(License.id == serial_row.redeemed_license_id) .with_for_update() ) if license_row is None: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "redeemed serial license record is missing", status_code=500, ) if license_row.owner_organization_id is not None: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "organization licenses must be managed by an organization administrator", status_code=403, ) if license_row.owner_user_id is not None and license_row.owner_user_id != actor.id: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "this license is already claimed by another account", status_code=403, ) if license_row.owner_user_id is None: license_row.owner_user_id = actor.id license_row.anonymous_subject_digest = None license_row.subject_type = "user" device_ids = self._session.scalars( select(Activation.device_id).where( Activation.license_id == license_row.id ) ).all() for device in self._session.scalars( select(Device).where(Device.id.in_(device_ids)) ): if device.owner_user_id is None and device.owner_organization_id is None: device.owner_user_id = actor.id serial_row.redeemed_by_user_id = actor.id self._audit( "serial.license_claimed", "license", license_row.id, actor_id=actor.id, correlation_id=normalized_correlation, metadata={"serialLookupId": parsed.lookup_id, "claimedAt": now.isoformat()}, ) return license_row.id def rename_account_device( self, license_id: str, device_id: str, friendly_name: str, actor_user_id: str, idempotency_key: str, correlation_id: str, ) -> dict[str, object]: normalized_license = validate_identifier(license_id, "license_id") normalized_device = validate_identifier(device_id, "device_id") normalized_name = _bounded_text( friendly_name, "friendly_name", MAX_FRIENDLY_DEVICE_NAME_CHARACTERS, ) normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_idempotency = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_correlation = validate_identifier(correlation_id, "correlation_id") request_digest = self._request_digest( { "licenseId": normalized_license, "deviceId": normalized_device, "friendlyName": normalized_name, } ) now = self._now() with self._session.begin(): replay = self._load_idempotent_response( "device.rename", f"{normalized_actor}:{normalized_license}", normalized_idempotency, request_digest, now, ) if replay is not None: return dict(replay) actor = self._session.get(User, normalized_actor) if actor is None or actor.status != "active": raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "an active account is required to rename a device", status_code=401, ) license_row = self._session.scalar( select(License) .where(License.id == normalized_license) .with_for_update() ) if license_row is None or license_row.status != "active": raise ServerLicensingError( ServerErrorCode.LICENSE_NOT_FOUND, "license was not found or is not active", status_code=404, ) activation = self._session.scalar( select(Activation) .where( Activation.license_id == license_row.id, Activation.device_id == normalized_device, Activation.status == "active", ) .with_for_update() ) if activation is None: raise ServerLicensingError( ServerErrorCode.DEVICE_NOT_FOUND, "device does not have an active allocation for this license", status_code=404, ) if not self._user_can_manage_activation( actor.id, license_row, activation, now, ): raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "account is not authorized to rename this licensed device", status_code=403, ) device = self._session.scalar( select(Device) .where(Device.id == normalized_device) .with_for_update() ) if device is None: raise ServerLicensingError( ServerErrorCode.DEVICE_NOT_FOUND, "licensed device was not found", status_code=404, ) device.friendly_name = normalized_name response = { "deviceId": device.id, "friendlyName": device.friendly_name, "status": device.status, } self._store_idempotent_response( "device.rename", f"{normalized_actor}:{normalized_license}", normalized_idempotency, request_digest, response, now, ) self._audit( "device.renamed", "device", device.id, actor_id=actor.id, correlation_id=normalized_correlation, metadata={"licenseId": license_row.id}, ) return response def release_unavailable_device( self, license_id: str, device_id: str, actor_user_id: str, correlation_id: str, *, reason: str = "remote_unavailable_release", ) -> tuple[str, int]: normalized_license = validate_identifier(license_id, "license_id") normalized_device = validate_identifier(device_id, "device_id") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_correlation = validate_identifier(correlation_id, "correlation_id") if reason not in ("remote_unavailable_release", "marked_lost"): raise ValueError("reason is not an allowed remote-release reason") now = self._now() with self._session.begin(): actor = self._session.get(User, normalized_actor) if actor is None or actor.status != "active": raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "an active account is required to release an unavailable device", status_code=401, ) license_row = self._session.scalar( select(License) .where(License.id == normalized_license) .with_for_update() ) if license_row is None or license_row.status != "active": raise ServerLicensingError( ServerErrorCode.LICENSE_NOT_FOUND, "license was not found or is not active", status_code=404, ) activation = self._session.scalar( select(Activation) .where( Activation.license_id == license_row.id, Activation.device_id == normalized_device, Activation.status == "active", ) .with_for_update() ) if activation is None: raise ServerLicensingError( ServerErrorCode.DEVICE_NOT_FOUND, "device does not have an active allocation for this license", status_code=404, ) if not self._user_can_manage_activation( actor.id, license_row, activation, now, ): raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "account is not authorized to release this licensed device", status_code=403, ) release_target_type = "license" release_target_id = license_row.id if ( license_row.owner_organization_id is not None and activation.seat_id is not None and self._session.get(OrganizationSeat, activation.seat_id) is not None ): release_target_type = "organization_seat" release_target_id = activation.seat_id policy = self._catalog.device_policies.get(license_row.device_policy_id) if policy is None: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "license references an unknown device policy", status_code=500, ) window_start = now - timedelta(days=policy.forced_release_window_days) used = self._session.scalar( select(func.count()) .select_from(AuditEvent) .where( AuditEvent.target_type == release_target_type, AuditEvent.target_id == release_target_id, AuditEvent.action.in_( ( "device.remote_unavailable_release", "device.marked_lost", ) ), AuditEvent.occurred_at >= window_start, ) ) used_count = int(used or 0) if used_count >= policy.forced_release_limit: raise ServerLicensingError( ServerErrorCode.FORCED_RELEASE_ALLOWANCE_EXHAUSTED, "automatic unavailable-device releases are exhausted; contact support for fast recovery", status_code=409, ) activation.status = "deactivated" activation.deactivated_at = now activation.release_reason = reason device = self._session.scalar( select(Device) .where(Device.id == normalized_device) .with_for_update() ) if device is None: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "licensed device record is missing", status_code=500, ) revoked_installation_count = 0 if reason == "marked_lost": device.status = "lost" installations = tuple( self._session.scalars( select(DeviceInstallation) .where(DeviceInstallation.device_id == device.id) .with_for_update() ) ) for installation in installations: if installation.status == "active": installation.status = "revoked" revoked_installation_count += 1 license_row.revocation_generation += 1 action = ( "device.marked_lost" if reason == "marked_lost" else "device.remote_unavailable_release" ) remaining = policy.forced_release_limit - used_count - 1 self._audit( action, release_target_type, release_target_id, actor_id=actor.id, correlation_id=normalized_correlation, reason=reason, metadata={ "activationId": activation.id, "deviceId": normalized_device, "seatId": activation.seat_id, "remainingAutomaticReleases": remaining, "revokedInstallationCount": revoked_installation_count, "revocationGeneration": license_row.revocation_generation, }, ) return activation.id, remaining def create_organization( self, display_name: str, actor_user_id: str, idempotency_key: str, correlation_id: str, ) -> dict[str, object]: normalized_name = _bounded_text(display_name, "display_name", 256) normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_idempotency = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_correlation = validate_identifier(correlation_id, "correlation_id") request_digest = self._request_digest({"displayName": normalized_name}) now = self._now() with self._session.begin(): replay = self._load_idempotent_response( "organization.create", normalized_actor, normalized_idempotency, request_digest, now, ) if replay is not None: return dict(replay) actor = self._session.get(User, normalized_actor) if actor is None or actor.status != "active": raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "an active account is required to create an organization", status_code=401, ) organization = Organization( id=new_identifier("organization"), display_name=normalized_name, status="active", ) membership = Membership( id=new_identifier("membership"), organization_id=organization.id, user_id=actor.id, role="owner", status="active", valid_until=None, ) self._session.add_all((organization, membership)) response = { "organizationId": organization.id, "displayName": organization.display_name, "membershipId": membership.id, "role": membership.role, } self._store_idempotent_response( "organization.create", normalized_actor, normalized_idempotency, request_digest, response, now, ) self._audit( "organization.created", "organization", organization.id, actor_id=actor.id, correlation_id=normalized_correlation, metadata={"membershipId": membership.id}, ) return response def invite_organization_member( self, organization_id: str, verified_email: str, role: str, actor_user_id: str, idempotency_key: str, correlation_id: str, ) -> dict[str, object]: normalized_organization = validate_identifier( organization_id, "organization_id", ) try: normalized_email = self._normalize_verified_email(verified_email) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, "verified organization invitation email is invalid", ) from exc if role not in ("member", "admin"): raise ValueError("role must be member or admin") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_idempotency = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_correlation = validate_identifier(correlation_id, "correlation_id") email_digest = self._organization_email_digest(normalized_email) request_digest = self._request_digest( { "organizationId": normalized_organization, "verifiedEmailDigest": email_digest.hex(), "role": role, } ) now = self._now() with self._session.begin(): replay = self._load_idempotent_response( "organization.invitation.create", f"{normalized_actor}:{normalized_organization}", normalized_idempotency, request_digest, now, ) if replay is not None: invitation_id = validate_identifier( replay.get("invitationId"), "invitationId", ) invitation = self._session.get(OrganizationInvitation, invitation_id) if invitation is None: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "stored organization invitation is missing", status_code=500, ) response = dict(replay) response["invitationToken"] = self._organization_invitation_token( invitation.id, invitation.invited_email_digest, ) return response actor = self._session.get(User, normalized_actor) if actor is None or actor.status != "active": raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "an active account is required to invite an organization member", status_code=401, ) manager = self._require_organization_manager( normalized_organization, actor.id, now, ) if role == "admin" and manager.role != "owner": raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "only an organization owner may invite another administrator", status_code=403, ) invitation = self._session.scalar( select(OrganizationInvitation) .where( OrganizationInvitation.organization_id == normalized_organization, OrganizationInvitation.invited_email_digest == email_digest, OrganizationInvitation.status == "pending", ) .with_for_update() ) if invitation is not None and _utc(invitation.expires_at) <= now: invitation.status = "expired" invitation = None if invitation is None: invitation_id = new_identifier("organization_invitation") invitation_token = self._organization_invitation_token( invitation_id, email_digest, ) invitation = OrganizationInvitation( id=invitation_id, organization_id=normalized_organization, invited_email_digest=email_digest, token_digest=hashlib.sha256( invitation_token.encode("utf-8") ).digest(), role=role, status="pending", expires_at=now + timedelta(days=ORGANIZATION_INVITATION_DAYS), created_by_user_id=actor.id, accepted_by_user_id=None, accepted_at=None, ) self._session.add(invitation) elif invitation.role != role: raise ServerLicensingError( ServerErrorCode.CONFLICT, "a pending invitation for this account already has another role", status_code=409, existing_resource_id=invitation.id, ) invitation_token = self._organization_invitation_token( invitation.id, invitation.invited_email_digest, ) stored_response = { "organizationId": normalized_organization, "invitationId": invitation.id, "role": invitation.role, "expiresAt": _utc(invitation.expires_at).isoformat(), } self._store_idempotent_response( "organization.invitation.create", f"{normalized_actor}:{normalized_organization}", normalized_idempotency, request_digest, stored_response, now, ) self._audit( "organization.invitation_created", "organization_invitation", invitation.id, actor_id=actor.id, correlation_id=normalized_correlation, metadata={ "organizationId": normalized_organization, "role": invitation.role, }, ) response = dict(stored_response) response["invitationToken"] = invitation_token return response def accept_organization_invitation( self, invitation_token: str, actor_user_id: str, idempotency_key: str, correlation_id: str, ) -> dict[str, object]: try: invitation_id, normalized_token = self._parse_organization_invitation_token( invitation_token ) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, "organization invitation token is invalid", ) from exc normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_idempotency = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_correlation = validate_identifier(correlation_id, "correlation_id") token_digest = hashlib.sha256(normalized_token.encode("utf-8")).digest() request_digest = self._request_digest( { "invitationId": invitation_id, "tokenDigest": token_digest.hex(), } ) now = self._now() with self._session.begin(): replay = self._load_idempotent_response( "organization.invitation.accept", normalized_actor, normalized_idempotency, request_digest, now, ) if replay is not None: return dict(replay) actor = self._session.get(User, normalized_actor) if actor is None or actor.status != "active" or actor.verified_email is None: raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "a signed-in account with a verified email is required", status_code=401, ) invitation = self._session.scalar( select(OrganizationInvitation) .where(OrganizationInvitation.id == invitation_id) .with_for_update() ) if invitation is None or not hmac.compare_digest( invitation.token_digest, token_digest, ): raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "organization invitation is invalid", status_code=403, ) expected_token = self._organization_invitation_token( invitation.id, invitation.invited_email_digest, ) if not hmac.compare_digest(expected_token, normalized_token): raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "organization invitation is invalid", status_code=403, ) actor_email_digest = self._organization_email_digest( self._normalize_verified_email(actor.verified_email) ) if not hmac.compare_digest( actor_email_digest, invitation.invited_email_digest, ): raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "organization invitation belongs to another verified account", status_code=403, ) if invitation.status == "accepted": if invitation.accepted_by_user_id != actor.id: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "organization invitation was accepted by another account", status_code=403, ) elif invitation.status != "pending" or _utc(invitation.expires_at) <= now: raise ServerLicensingError( ServerErrorCode.CONFLICT, "organization invitation is expired or no longer active", status_code=409, ) membership = self._session.scalar( select(Membership) .where( Membership.organization_id == invitation.organization_id, Membership.user_id == actor.id, ) .with_for_update() ) if membership is None: membership = Membership( id=new_identifier("membership"), organization_id=invitation.organization_id, user_id=actor.id, role=invitation.role, status="active", valid_until=None, ) self._session.add(membership) else: membership.status = "active" membership.valid_until = None if membership.role != "owner": membership.role = invitation.role if invitation.status != "accepted": invitation.status = "accepted" invitation.accepted_by_user_id = actor.id invitation.accepted_at = now response = { "organizationId": invitation.organization_id, "membershipId": membership.id, "role": membership.role, "status": membership.status, } self._store_idempotent_response( "organization.invitation.accept", normalized_actor, normalized_idempotency, request_digest, response, now, ) self._audit( "organization.invitation_accepted", "membership", membership.id, actor_id=actor.id, correlation_id=normalized_correlation, metadata={ "organizationId": invitation.organization_id, "invitationId": invitation.id, }, ) return response def attach_team_license( self, organization_id: str, license_id: str, actor_user_id: str, idempotency_key: str, correlation_id: str, ) -> dict[str, object]: normalized_organization = validate_identifier( organization_id, "organization_id", ) normalized_license = validate_identifier(license_id, "license_id") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_idempotency = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_correlation = validate_identifier(correlation_id, "correlation_id") request_digest = self._request_digest( { "organizationId": normalized_organization, "licenseId": normalized_license, } ) now = self._now() with self._session.begin(): replay = self._load_idempotent_response( "organization.license.attach", normalized_actor, normalized_idempotency, request_digest, now, ) if replay is not None: return dict(replay) actor = self._session.get(User, normalized_actor) if actor is None or actor.status != "active": raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "an active account is required to attach a team license", status_code=401, ) manager = self._require_organization_manager( normalized_organization, actor.id, now, ) if manager.role != "owner": raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "only the organization owner may attach a team license", status_code=403, ) license_row = self._session.scalar( select(License) .where(License.id == normalized_license) .with_for_update() ) if license_row is None or license_row.status != "active": raise ServerLicensingError( ServerErrorCode.LICENSE_NOT_FOUND, "team license was not found or is not active", status_code=404, ) if license_row.owner_organization_id not in (None, normalized_organization): raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "license already belongs to another organization", status_code=403, ) if ( license_row.owner_organization_id is None and license_row.owner_user_id != actor.id ): raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "only the current license owner may transfer it", status_code=403, ) active_skus = tuple( self._catalog.require_sku(grant.sku_id, require_active=False) for grant in self._session.scalars( self._active_grants_query(license_row.id, now) ) ) if not any("admin.team_licenses" in sku.entitlements for sku in active_skus): raise ServerLicensingError( ServerErrorCode.SKU_NOT_AVAILABLE, "this license level does not include team administration", status_code=409, ) license_row.owner_user_id = None license_row.anonymous_subject_digest = None license_row.owner_organization_id = normalized_organization license_row.subject_type = "organization" seats = self._ensure_organization_seat_records( license_row, normalized_organization, actor.id, now, ) owner_seat = next( seat for seat in seats if seat.assigned_user_id == actor.id ) active_device_ids = self._session.scalars( select(Activation.device_id).where( Activation.license_id == license_row.id, Activation.status == "active", ) ).all() for activation in self._session.scalars( select(Activation).where( Activation.license_id == license_row.id, Activation.status == "active", ) ): activation.seat_id = owner_seat.id for device in self._session.scalars( select(Device).where(Device.id.in_(active_device_ids)) ): device.owner_user_id = None device.owner_organization_id = normalized_organization response = { "organizationId": normalized_organization, "licenseId": license_row.id, "seats": [self._organization_seat_mapping(seat) for seat in seats], } self._store_idempotent_response( "organization.license.attach", normalized_actor, normalized_idempotency, request_digest, response, now, ) self._audit( "organization.license_attached", "license", license_row.id, actor_id=actor.id, correlation_id=normalized_correlation, metadata={ "organizationId": normalized_organization, "seatCount": len(seats), }, ) return response def assign_organization_seat( self, organization_id: str, seat_id: str, user_id: str, actor_user_id: str, idempotency_key: str, correlation_id: str, ) -> dict[str, object]: normalized_organization = validate_identifier( organization_id, "organization_id", ) normalized_seat = validate_identifier(seat_id, "seat_id") normalized_user = validate_identifier(user_id, "user_id") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_idempotency = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_correlation = validate_identifier(correlation_id, "correlation_id") request_digest = self._request_digest( { "organizationId": normalized_organization, "seatId": normalized_seat, "userId": normalized_user, } ) now = self._now() with self._session.begin(): replay = self._load_idempotent_response( "organization.seat.assign", f"{normalized_actor}:{normalized_organization}", normalized_idempotency, request_digest, now, ) if replay is not None: return dict(replay) actor = self._session.get(User, normalized_actor) if actor is None or actor.status != "active": raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "an active account is required to assign a seat", status_code=401, ) self._require_organization_manager( normalized_organization, actor.id, now, ) member = self._session.scalar( select(Membership).where( Membership.organization_id == normalized_organization, Membership.user_id == normalized_user, Membership.status == "active", ( Membership.valid_until.is_(None) | (Membership.valid_until > now) ), ) ) if member is None: raise ServerLicensingError( ServerErrorCode.SEAT_NOT_ASSIGNED, "the selected account is not an active organization member", status_code=409, ) seat = self._session.scalar( select(OrganizationSeat) .where( OrganizationSeat.id == normalized_seat, OrganizationSeat.organization_id == normalized_organization, ) .with_for_update() ) if seat is None: raise ServerLicensingError( ServerErrorCode.SEAT_NOT_ASSIGNED, "organization seat was not found", status_code=404, ) if seat.status == "suspended": raise ServerLicensingError( ServerErrorCode.SEAT_NOT_ASSIGNED, "organization seat is suspended", status_code=409, ) if seat.status == "assigned" and seat.assigned_user_id != normalized_user: raise ServerLicensingError( ServerErrorCode.SEAT_LIMIT_REACHED, "organization seat is assigned to another account", status_code=409, ) duplicate = self._session.scalar( select(OrganizationSeat).where( OrganizationSeat.license_id == seat.license_id, OrganizationSeat.assigned_user_id == normalized_user, OrganizationSeat.status == "assigned", OrganizationSeat.id != seat.id, ) ) if duplicate is not None: raise ServerLicensingError( ServerErrorCode.CONFLICT, "account already has a named seat on this license", status_code=409, existing_resource_id=duplicate.id, ) if seat.status != "assigned": review_window_start = now - timedelta( days=ORGANIZATION_SEAT_REASSIGNMENT_REVIEW_DAYS ) if ( seat.assignment_generation >= 2 and seat.last_reassigned_at is not None and _utc(seat.last_reassigned_at) >= review_window_start ): raise ServerLicensingError( ServerErrorCode.SEAT_REASSIGNMENT_REVIEW, "rapid repeated seat reassignment requires support review", status_code=409, ) seat.status = "assigned" seat.assigned_user_id = normalized_user seat.assigned_at = now seat.released_at = None seat.last_reassigned_at = now seat.assignment_generation += 1 response = self._organization_seat_mapping(seat) self._store_idempotent_response( "organization.seat.assign", f"{normalized_actor}:{normalized_organization}", normalized_idempotency, request_digest, response, now, ) self._audit( "organization.seat_assigned", "organization_seat", seat.id, actor_id=actor.id, correlation_id=normalized_correlation, metadata={ "organizationId": normalized_organization, "licenseId": seat.license_id, "assignedUserId": normalized_user, "assignmentGeneration": seat.assignment_generation, }, ) return response def release_organization_seat( self, organization_id: str, seat_id: str, actor_user_id: str, idempotency_key: str, correlation_id: str, ) -> dict[str, object]: normalized_organization = validate_identifier( organization_id, "organization_id", ) normalized_seat = validate_identifier(seat_id, "seat_id") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_idempotency = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_correlation = validate_identifier(correlation_id, "correlation_id") request_digest = self._request_digest( { "organizationId": normalized_organization, "seatId": normalized_seat, } ) now = self._now() with self._session.begin(): replay = self._load_idempotent_response( "organization.seat.release", f"{normalized_actor}:{normalized_organization}", normalized_idempotency, request_digest, now, ) if replay is not None: return dict(replay) actor = self._session.get(User, normalized_actor) if actor is None or actor.status != "active": raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "an active account is required to release a seat", status_code=401, ) self._require_organization_manager( normalized_organization, actor.id, now, ) seat = self._session.scalar( select(OrganizationSeat) .where( OrganizationSeat.id == normalized_seat, OrganizationSeat.organization_id == normalized_organization, ) .with_for_update() ) if seat is None: raise ServerLicensingError( ServerErrorCode.SEAT_NOT_ASSIGNED, "organization seat was not found", status_code=404, ) released_activations = 0 if seat.status == "assigned": for activation in self._session.scalars( select(Activation) .where( Activation.license_id == seat.license_id, Activation.seat_id == seat.id, Activation.status == "active", ) .with_for_update() ): activation.status = "deactivated" activation.deactivated_at = now activation.release_reason = "organization_seat_released" released_activations += 1 seat.status = "available" seat.assigned_user_id = None seat.assigned_at = None seat.released_at = now if released_activations: license_row = self._session.get(License, seat.license_id) if license_row is not None: license_row.revocation_generation += 1 response = self._organization_seat_mapping(seat) response["releasedActivationCount"] = released_activations self._store_idempotent_response( "organization.seat.release", f"{normalized_actor}:{normalized_organization}", normalized_idempotency, request_digest, response, now, ) self._audit( "organization.seat_released", "organization_seat", seat.id, actor_id=actor.id, correlation_id=normalized_correlation, metadata={ "organizationId": normalized_organization, "licenseId": seat.license_id, "releasedActivationCount": released_activations, }, ) return response def list_organizations(self, actor_user_id: str) -> tuple[dict[str, object], ...]: normalized_actor = validate_identifier(actor_user_id, "actor_user_id") now = self._now() actor = self._session.get(User, normalized_actor) if actor is None or actor.status != "active": raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "an active account is required to list organizations", status_code=401, ) memberships = tuple( self._session.scalars( select(Membership).where( Membership.user_id == actor.id, Membership.status == "active", ( Membership.valid_until.is_(None) | (Membership.valid_until > now) ), ) ) ) result = [] for own_membership in memberships: organization = self._session.get( Organization, own_membership.organization_id, ) if organization is None or organization.status != "active": continue can_manage = own_membership.role in ("owner", "admin") visible_memberships = ( tuple( self._session.scalars( select(Membership).where( Membership.organization_id == organization.id ) ) ) if can_manage else (own_membership,) ) visible_seats = tuple( self._session.scalars( select(OrganizationSeat).where( OrganizationSeat.organization_id == organization.id, ( True if can_manage else OrganizationSeat.assigned_user_id == actor.id ), ) ) ) member_values = [] for membership in visible_memberships: user = self._session.get(User, membership.user_id) member_values.append( { "membershipId": membership.id, "userId": membership.user_id, "role": membership.role, "status": membership.status, "verifiedEmail": ( None if user is None else user.verified_email ), } ) result.append( { "organizationId": organization.id, "displayName": organization.display_name, "role": own_membership.role, "canManage": can_manage, "memberships": member_values, "seats": [ self._organization_seat_mapping(seat) for seat in visible_seats ], } ) return tuple(result) @staticmethod def _normalize_verified_email(value: str) -> str: if not isinstance(value, str): raise TypeError("verified_email must be a string") normalized = value.strip().casefold() if len(normalized) < 3 or len(normalized) > 256: raise ValueError("verified_email is empty or too long") local, separator, domain = normalized.rpartition("@") if separator != "@" or not local or not domain or "." not in domain: raise ValueError("verified_email is not a valid account email") if any(character.isspace() for character in normalized): raise ValueError("verified_email must not contain whitespace") return normalized def _organization_email_digest(self, normalized_email: str) -> bytes: if not isinstance(normalized_email, str) or not normalized_email: raise ValueError("normalized_email must be a non-empty string") return hmac.new( self._fingerprint_pepper, b"APOLON-ORGANIZATION-INVITATION-EMAIL-V1\x00" + normalized_email.encode("utf-8"), hashlib.sha256, ).digest() def _organization_invitation_token( self, invitation_id: str, email_digest: bytes, ) -> str: normalized_invitation = validate_identifier(invitation_id, "invitation_id") if not isinstance(email_digest, bytes) or len(email_digest) != 32: raise ValueError("email_digest must contain a SHA-256 digest") token_key = hmac.new( self._fingerprint_pepper, b"APOLON-ORGANIZATION-INVITATION-TOKEN-KEY-V1\x00", hashlib.sha256, ).digest() authenticator = hmac.new( token_key, normalized_invitation.encode("utf-8") + b"\x00" + email_digest, hashlib.sha256, ).hexdigest() return f"OINV1~{normalized_invitation}~{authenticator}" @staticmethod def _parse_organization_invitation_token(value: str) -> tuple[str, str]: if not isinstance(value, str): raise TypeError("invitation_token must be a string") normalized = value.strip() if not normalized or len(normalized) > 512: raise ValueError("invitation_token is empty or too long") parts = normalized.split("~") if len(parts) != 3 or parts[0] != "OINV1": raise ValueError("invitation_token format is invalid") invitation_id = validate_identifier(parts[1], "invitation_id") authenticator = parts[2] if len(authenticator) != 64 or any( character not in "0123456789abcdef" for character in authenticator ): raise ValueError("invitation_token authenticator is invalid") return invitation_id, normalized def _require_organization_manager( self, organization_id: str, actor_user_id: str, now: datetime, ) -> Membership: normalized_organization = validate_identifier( organization_id, "organization_id", ) normalized_actor = validate_identifier(actor_user_id, "actor_user_id") if not isinstance(now, datetime) or now.tzinfo is None or now.utcoffset() is None: raise ValueError("now must be a timezone-aware datetime") organization = self._session.scalar( select(Organization) .where( Organization.id == normalized_organization, Organization.status == "active", ) .with_for_update() ) if organization is None: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "organization was not found or is not active", status_code=403, ) membership = self._session.scalar( select(Membership) .where( Membership.organization_id == organization.id, Membership.user_id == normalized_actor, Membership.status == "active", Membership.role.in_(("owner", "admin")), ( Membership.valid_until.is_(None) | (Membership.valid_until > now) ), ) .with_for_update() ) if membership is None: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "account is not an active organization administrator", status_code=403, ) return membership def _ensure_organization_seat_records( self, license_row: License, organization_id: str, preferred_user_id: str, now: datetime, assign_if_available: bool = True, ) -> tuple[OrganizationSeat, ...]: if not isinstance(license_row, License): raise TypeError("license_row must be License") normalized_organization = validate_identifier( organization_id, "organization_id", ) normalized_user = validate_identifier(preferred_user_id, "preferred_user_id") if not isinstance(now, datetime) or now.tzinfo is None or now.utcoffset() is None: raise ValueError("now must be a timezone-aware datetime") if not isinstance(assign_if_available, bool): raise TypeError("assign_if_available must be a Boolean") seats = tuple( self._session.scalars( select(OrganizationSeat) .where(OrganizationSeat.license_id == license_row.id) .order_by(OrganizationSeat.id) .with_for_update() ) ) if seats: if any(seat.organization_id != normalized_organization for seat in seats): raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "organization seat ownership is inconsistent", status_code=500, ) if assign_if_available and not any( seat.assigned_user_id == normalized_user for seat in seats ): available = next( (seat for seat in seats if seat.status == "available"), None, ) if available is None: raise ServerLicensingError( ServerErrorCode.SEAT_LIMIT_REACHED, "no named seat is available for the organization owner", status_code=409, ) available.status = "assigned" available.assigned_user_id = normalized_user available.assigned_at = now available.released_at = None available.last_reassigned_at = now available.assignment_generation += 1 return seats policy = self._catalog.device_policies.get(license_row.device_policy_id) if policy is None: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "license references an unknown device policy", status_code=500, ) created = [] for index in range(policy.named_seats): assigned = index == 0 created.append( OrganizationSeat( id=new_identifier("organization_seat"), organization_id=normalized_organization, license_id=license_row.id, assigned_user_id=normalized_user if assigned else None, status="assigned" if assigned else "available", assigned_at=now if assigned else None, released_at=None, last_reassigned_at=now if assigned else None, assignment_generation=1 if assigned else 0, ) ) self._session.add_all(created) self._session.flush() return tuple(created) @staticmethod def _organization_seat_mapping(seat: OrganizationSeat) -> dict[str, object]: if not isinstance(seat, OrganizationSeat): raise TypeError("seat must be OrganizationSeat") return { "seatId": seat.id, "organizationId": seat.organization_id, "licenseId": seat.license_id, "assignedUserId": seat.assigned_user_id, "status": seat.status, "assignmentGeneration": seat.assignment_generation, } def _user_can_manage_license(self, user_id: str, license_row: License) -> bool: normalized_user = validate_identifier(user_id, "user_id") if not isinstance(license_row, License): raise TypeError("license_row must be License") if license_row.owner_user_id == normalized_user: return True if license_row.owner_organization_id is None: return False now = self._now() membership = self._session.scalar( select(Membership).where( Membership.organization_id == license_row.owner_organization_id, Membership.user_id == normalized_user, Membership.status == "active", Membership.role.in_(("owner", "admin")), ( Membership.valid_until.is_(None) | (Membership.valid_until > now) ), ) ) return membership is not None def _user_can_manage_activation( self, user_id: str, license_row: License, activation: Activation, now: datetime, ) -> bool: normalized_user = validate_identifier(user_id, "user_id") if not isinstance(license_row, License): raise TypeError("license_row must be a License") if not isinstance(activation, Activation): raise TypeError("activation must be an Activation") if not isinstance(now, datetime) or now.tzinfo is None or now.utcoffset() is None: raise ValueError("now must be a timezone-aware datetime") if self._user_can_manage_license(normalized_user, license_row): return True if license_row.owner_organization_id is None or activation.seat_id is None: return False membership = self._session.scalar( select(Membership.id).where( Membership.organization_id == license_row.owner_organization_id, Membership.user_id == normalized_user, Membership.status == "active", ( Membership.valid_until.is_(None) | (Membership.valid_until > now) ), ) ) if membership is None: return False assigned_seat = self._session.scalar( select(OrganizationSeat.id).where( OrganizationSeat.id == activation.seat_id, OrganizationSeat.license_id == license_row.id, OrganizationSeat.assigned_user_id == normalized_user, OrganizationSeat.status == "assigned", ) ) return assigned_seat is not None def _require_installation_key( self, installation_id: str, public_key_der: bytes, key_thumbprint: str, now: datetime, evidence: Optional[Mapping[str, str]] = None, ) -> DeviceInstallation: installation = self._session.scalar( select(DeviceInstallation) .where(DeviceInstallation.installation_id == installation_id) .with_for_update() ) if installation is None: raise ServerLicensingError( ServerErrorCode.DEVICE_NOT_FOUND, "device installation was not found", status_code=404, ) if installation.key_thumbprint != key_thumbprint or not hmac.compare_digest( installation.public_key_der, public_key_der, ): raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "device proof key does not match the enrolled installation", status_code=403, ) device = self._session.get(Device, installation.device_id) if device is None: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "canonical device record is missing", status_code=500, ) self._require_device_available_for_enrollment(device) if installation.status != "active": raise ServerLicensingError( ServerErrorCode.DEVICE_IDENTITY_REPLACEMENT_REQUIRED, "Support approved this recovered computer, but its revoked device proof key " "must be replaced in License Center before activation.", status_code=409, ) installation.last_seen_at = now if evidence is not None: digests = digest_device_evidence(self._fingerprint_pepper, evidence) if self._device_evidence_materially_conflicts( device.evidence_digests, digests, ): raise ServerLicensingError( ServerErrorCode.CONFLICT, "installation evidence identifies different hardware", status_code=409, ) device.evidence_digests = self._merge_device_evidence( device.evidence_digests, digests, ) device.last_seen_at = now return installation @staticmethod def _require_device_available_for_enrollment(device: Device) -> None: if not isinstance(device, Device): raise TypeError("device must be a Device") if device.status == "lost": raise ServerLicensingError( ServerErrorCode.DEVICE_RECOVERY_REQUIRED, "This computer was marked lost or stolen. Contact support to review and " "restore it before replacing its device identity.", status_code=409, existing_resource_id=device.id, ) if device.status != "active": raise ServerLicensingError( ServerErrorCode.DEVICE_NOT_FOUND, "device is no longer active", status_code=409, ) def _trial_subject_key( self, actor_user_id: Optional[str], enrollment: DeviceEnrollment, ) -> str: if actor_user_id is not None: return actor_user_id digests = digest_device_evidence(self._fingerprint_pepper, enrollment.evidence) if "system_uuid" in digests or len(digests) >= 2: digest = hashlib.sha256(canonicalize_json(digests)).hexdigest() return f"device_evidence.{digest}" return enrollment.installation_id def _create_trial_from_serial( self, sku: SkuDefinition, batch: SerialBatch, serial_row: Serial, actor: Optional[User], enrollment: DeviceEnrollment, installation: DeviceInstallation, now: datetime, ) -> tuple[License, Grant]: if not isinstance(sku, SkuDefinition) or sku.kind is not SkuKind.TRIAL: raise TypeError("sku must be a trial SKU definition") if not isinstance(batch, SerialBatch): raise TypeError("batch must be a SerialBatch") if not isinstance(serial_row, Serial): raise TypeError("serial_row must be a Serial") if actor is not None and not isinstance(actor, User): raise TypeError("actor must be a User or None") if not isinstance(enrollment, DeviceEnrollment): raise TypeError("enrollment must be a DeviceEnrollment") if not isinstance(installation, DeviceInstallation): raise TypeError("installation must be a DeviceInstallation") if not isinstance(now, datetime) or now.tzinfo is None or now.utcoffset() is None: raise ValueError("now must be a timezone-aware datetime") duration_hours = batch.trial_duration_hours if ( isinstance(duration_hours, bool) or not isinstance(duration_hours, int) or not ( TRIAL_MINIMUM_AUTHORITY_DURATION_HOURS <= duration_hours <= TRIAL_MAXIMUM_TOTAL_DURATION_HOURS ) ): raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "trial serial batch has an invalid duration", status_code=500, ) subject_text = self._trial_subject_key( None if actor is None else actor.id, enrollment, ) subject_digest = anonymous_subject_digest( self._fingerprint_pepper, subject_text, ) existing = self._session.scalar( select(Trial).where(Trial.subject_digest == subject_digest).with_for_update() ) if existing is None and actor is None: existing = self._session.scalar( select(Trial) .join(License, License.id == Trial.license_id) .join(Activation, Activation.license_id == Trial.license_id) .where( Activation.device_id == installation.device_id, License.owner_user_id.is_(None), License.owner_organization_id.is_(None), ) .order_by(Trial.starts_at, Trial.id) .with_for_update() ) if existing is not None: raise ServerLicensingError( ServerErrorCode.TRIAL_NOT_ELIGIBLE, "this account or computer already has a recorded trial", status_code=409, existing_resource_id=existing.id, ) final_expiry = now + timedelta(hours=duration_hours) license_row = License( id=new_identifier("license"), product_id=PRODUCT_ID, owner_user_id=None if actor is None else actor.id, owner_organization_id=None, anonymous_subject_digest=( subject_digest if actor is None else None ), subject_type="user" if actor is not None else "license", status="active", device_policy_id=sku.device_policy_id, surface_profile_id=batch.surface_profile_id, revocation_generation=0, ) self._session.add(license_row) grant = Grant( id=new_identifier("grant"), license_id=license_row.id, sku_id=sku.sku_id, source_type="trial_serial_redemption", source_reference=serial_row.id, catalog_revision=self._catalog.revision, starts_at=now, ends_at=final_expiry, status="active", metadata_json={"trialDurationHours": duration_hours}, ) self._session.add(grant) trial = Trial( id=new_identifier("trial"), license_id=license_row.id, grant_id=grant.id, subject_digest=subject_digest, starts_at=now, ends_at=final_expiry, status="active", extension_count=0, ) self._session.add(trial) self._session.flush() return license_row, grant def _resolve_license_for_sku( self, sku: SkuDefinition, actor: Optional[User], installation: DeviceInstallation, target_license_id: Optional[str], surface_profile_id: Optional[str] = None, ) -> License: if sku.kind not in (SkuKind.EDITION, SkuKind.ADDON): raise ServerLicensingError( ServerErrorCode.SKU_NOT_AVAILABLE, "serial SKU is not a perpetual edition or add-on", ) normalized_surface_profile = ( None if surface_profile_id is None else validate_identifier(surface_profile_id, "surface_profile_id") ) if sku.kind is SkuKind.ADDON and normalized_surface_profile is not None: raise ValueError("an add-on cannot replace a license surface profile") if target_license_id is None: if sku.kind is SkuKind.ADDON: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, "add-on serial redemption requires a target license", ) subject_digest = anonymous_subject_digest( self._fingerprint_pepper, installation.installation_id, ) license_row = License( id=new_identifier("license"), product_id=PRODUCT_ID, owner_user_id=None if actor is None else actor.id, owner_organization_id=None, anonymous_subject_digest=None if actor is not None else subject_digest, subject_type="user" if actor is not None else "license", status="active", device_policy_id=sku.device_policy_id, surface_profile_id=normalized_surface_profile, revocation_generation=0, ) self._session.add(license_row) self._session.flush() return license_row license_row = self._session.scalar( select(License).where(License.id == target_license_id).with_for_update() ) if license_row is None or license_row.status != "active": raise ServerLicensingError( ServerErrorCode.LICENSE_NOT_FOUND, "target license was not found or is not active", status_code=404, ) if actor is not None: if not self._user_can_manage_license(actor.id, license_row): raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "account cannot manage the target license", status_code=403, ) else: device_owns_activation = self._session.scalar( select(func.count()) .select_from(Activation) .where( Activation.license_id == license_row.id, Activation.device_id == installation.device_id, Activation.status == "active", ) ) if not device_owns_activation: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "the current device is not activated for the target license", status_code=403, ) if sku.kind is SkuKind.EDITION: active_edition_ids = self._active_edition_ids(license_row.id, self._now()) current_level = max( ( self._catalog.require_sku( edition_id, require_active=False, ).edition_level for edition_id in active_edition_ids ), default=0, ) if sku.edition_level <= current_level: raise ServerLicensingError( ServerErrorCode.SKU_NOT_AVAILABLE, "edition serial does not upgrade the target license", status_code=409, ) license_row.device_policy_id = sku.device_policy_id return license_row edition_ids = self._active_edition_ids(license_row.id, self._now()) if sku.compatible_editions and not set(sku.compatible_editions).intersection(edition_ids): raise ServerLicensingError( ServerErrorCode.SKU_NOT_AVAILABLE, "add-on is not compatible with the target license edition", status_code=409, ) return license_row def _enroll_device( self, enrollment: DeviceEnrollment, actor: Optional[User], now: datetime, ) -> tuple[DeviceInstallation, bool]: digests = digest_device_evidence(self._fingerprint_pepper, enrollment.evidence) existing = self._session.scalar( select(DeviceInstallation) .where(DeviceInstallation.installation_id == enrollment.installation_id) .with_for_update() ) if existing is not None: if existing.key_thumbprint != enrollment.key_thumbprint or not hmac.compare_digest( existing.public_key_der, enrollment.public_key_der, ): raise ServerLicensingError( ServerErrorCode.CONFLICT, "installation ID is already bound to another device key", status_code=409, ) device = self._session.get(Device, existing.device_id) if device is None: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "canonical device record is missing", status_code=500, ) self._require_device_available_for_enrollment(device) if existing.status != "active": raise ServerLicensingError( ServerErrorCode.DEVICE_IDENTITY_REPLACEMENT_REQUIRED, "Support approved this recovered computer, but its revoked device proof " "key must be replaced in License Center before activation.", status_code=409, ) existing.last_seen_at = now if self._device_evidence_materially_conflicts( device.evidence_digests, digests, ): raise ServerLicensingError( ServerErrorCode.CONFLICT, "installation evidence identifies different hardware", status_code=409, ) device.evidence_digests = self._merge_device_evidence( device.evidence_digests, digests, ) device.last_seen_at = now return existing, True candidate: Optional[Device] = None if self._unique_device_evidence(digests): owner_conditions = ( ( Device.owner_user_id.is_(None), Device.owner_organization_id.is_(None), ) if actor is None else (Device.owner_user_id == actor.id,) ) candidates = self._session.scalars( select(Device).where( *owner_conditions, ) ).all() for possible in candidates: if self._device_evidence_matches( possible.evidence_digests, digests, ): if candidate is not None and candidate.id != possible.id: raise ServerLicensingError( ServerErrorCode.CONFLICT, "device evidence matches multiple canonical devices", status_code=409, ) candidate = possible if candidate is not None: self._require_device_available_for_enrollment(candidate) recovered = candidate is not None if candidate is None: candidate = Device( id=new_identifier("device"), owner_user_id=None if actor is None else actor.id, owner_organization_id=None, friendly_name=enrollment.friendly_name, device_type="computer", status="active", evidence_schema_version=1, evidence_digests=digests, last_seen_at=now, ) self._session.add(candidate) self._session.flush() else: candidate.evidence_digests = self._merge_device_evidence( candidate.evidence_digests, digests, ) installation = DeviceInstallation( id=new_identifier("installation"), installation_id=enrollment.installation_id, device_id=candidate.id, public_key_der=enrollment.public_key_der, key_thumbprint=enrollment.key_thumbprint, key_provider=enrollment.key_provider, status="active", last_seen_at=now, ) self._session.add(installation) self._session.flush() return installation, recovered @staticmethod def _unique_device_evidence(digests: Mapping[str, str]) -> dict[str, str]: if not isinstance(digests, Mapping): raise TypeError("digests must be a mapping") return { key: value for key, value in digests.items() if key in UNIQUE_DEVICE_EVIDENCE_COMPONENTS } @classmethod def _device_evidence_matches( cls, stored: Mapping[str, str], current: Mapping[str, str], ) -> bool: stored_unique = cls._unique_device_evidence(stored) current_unique = cls._unique_device_evidence(current) return any( current_unique.get(component) == digest for component, digest in stored_unique.items() ) @classmethod def _device_evidence_materially_conflicts( cls, stored: Mapping[str, str], current: Mapping[str, str], ) -> bool: stored_unique = cls._unique_device_evidence(stored) current_unique = cls._unique_device_evidence(current) return bool( stored_unique and current_unique and not cls._device_evidence_matches(stored_unique, current_unique) ) @staticmethod def _merge_device_evidence( stored: Mapping[str, str], current: Mapping[str, str], ) -> dict[str, str]: if not isinstance(stored, Mapping) or not isinstance(current, Mapping): raise TypeError("stored and current evidence must be mappings") merged = dict(stored) for component, digest in current.items(): if component not in merged: merged[component] = digest return merged def _ensure_activation( self, license_row: License, installation: DeviceInstallation, now: datetime, seat_id: Optional[str] = None, ) -> Activation: normalized_seat = ( license_row.owner_user_id if seat_id is None else validate_identifier(seat_id, "seat_id") ) existing = self._session.scalar( select(Activation) .where( Activation.license_id == license_row.id, Activation.device_id == installation.device_id, Activation.status == "active", ) .with_for_update() ) if existing is not None: return existing policy = self._catalog.device_policies.get(license_row.device_policy_id) if policy is None: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "license references an unknown device policy", status_code=500, ) active_device_conditions = [ Activation.license_id == license_row.id, Activation.status == "active", ] if normalized_seat is not None: active_device_conditions.append(Activation.seat_id == normalized_seat) active_count = self._session.scalar( select(func.count(func.distinct(Activation.device_id))).where( *active_device_conditions ) ) if active_count is not None and active_count >= policy.maximum_devices: raise ServerLicensingError( ServerErrorCode.DEVICE_LIMIT_REACHED, "all licensed device slots are in use", status_code=409, ) if normalized_seat is not None: seat_already_active = self._session.scalar( select(func.count()).select_from(Activation).where( Activation.license_id == license_row.id, Activation.status == "active", Activation.seat_id == normalized_seat, ) ) if not seat_already_active: active_seats = self._session.scalar( select(func.count(func.distinct(Activation.seat_id))).where( Activation.license_id == license_row.id, Activation.status == "active", Activation.seat_id.is_not(None), ) ) if active_seats is not None and active_seats >= policy.named_seats: raise ServerLicensingError( ServerErrorCode.SEAT_LIMIT_REACHED, "all named seats on this license are assigned", status_code=409, ) historical_conditions = [ Activation.license_id == license_row.id, Activation.device_id == installation.device_id, ] if normalized_seat is not None: historical_conditions.append(Activation.seat_id == normalized_seat) previously_activated = self._session.scalar( select(Activation.id).where(*historical_conditions).limit(1) ) if previously_activated is None: churn_conditions = [ Activation.license_id == license_row.id, Activation.activated_at >= now - timedelta(days=policy.rapid_churn_window_days), ] if normalized_seat is not None: churn_conditions.append(Activation.seat_id == normalized_seat) recent_device_count = self._session.scalar( select(func.count(func.distinct(Activation.device_id))).where( *churn_conditions ) ) if ( recent_device_count is not None and recent_device_count >= policy.rapid_churn_device_limit ): raise ServerLicensingError( ServerErrorCode.DEVICE_CHURN_REVIEW, "recent device changes require account or support review", status_code=409, ) activation = Activation( id=new_identifier("activation"), license_id=license_row.id, device_id=installation.device_id, installation_id=installation.id, seat_id=normalized_seat, status="active", activated_at=now, deactivated_at=None, release_reason=None, ) self._session.add(activation) self._session.flush() return activation def _active_grants_query(self, license_id: str, now: datetime) -> Select[tuple[Grant]]: return select(Grant).where( Grant.license_id == license_id, Grant.status == "active", Grant.starts_at <= now, (Grant.ends_at.is_(None) | (Grant.ends_at > now)), ) def _support_license_mapping( self, license_row: License, now: datetime, ) -> dict[str, object]: if not isinstance(license_row, License): raise TypeError("license_row must be a License") if not isinstance(now, datetime) or now.tzinfo is None or now.utcoffset() is None: raise ValueError("now must be a timezone-aware datetime") grants = tuple( self._session.scalars( select(Grant) .where(Grant.license_id == license_row.id) .order_by(Grant.starts_at, Grant.id) ) ) trials = tuple( self._session.scalars( select(Trial) .where(Trial.license_id == license_row.id) .order_by(Trial.starts_at, Trial.id) ) ) activation_rows = tuple( self._session.execute( select(Activation, Device) .join(Device, Device.id == Activation.device_id) .where(Activation.license_id == license_row.id) .order_by(Activation.activated_at, Activation.id) ).all() ) active_grants = tuple( grant for grant in grants if self._grant_is_usable_at(grant, now) ) effective_entitlements = self._entitlements_for_grants(active_grants) target_ids = { license_row.id, *(grant.id for grant in grants), *(trial.id for trial in trials), *(activation.id for activation, _device in activation_rows), *(device.id for _activation, device in activation_rows), } timeline = tuple( self._session.scalars( select(AuditEvent) .where(AuditEvent.target_id.in_(tuple(sorted(target_ids)))) .order_by(AuditEvent.occurred_at.desc(), AuditEvent.id.desc()) .limit(MAX_SUPPORT_TIMELINE_EVENTS) ) ) return { "licenseId": license_row.id, "productId": license_row.product_id, "subjectType": license_row.subject_type, "ownerUserId": license_row.owner_user_id, "ownerOrganizationId": license_row.owner_organization_id, "status": license_row.status, "devicePolicyId": license_row.device_policy_id, "surfaceProfileId": license_row.surface_profile_id, "surfacePolicy": self._surface_policy_for_license(license_row).to_mapping(), "revocationGeneration": license_row.revocation_generation, "effectiveEntitlements": list(effective_entitlements), "activeGrantIds": [grant.id for grant in active_grants], "grants": [ { "grantId": grant.id, "skuId": grant.sku_id, "sourceType": grant.source_type, "status": grant.status, "startsAt": _utc(grant.starts_at).isoformat(), "endsAt": ( None if grant.ends_at is None else _utc(grant.ends_at).isoformat() ), "activeNow": self._grant_is_usable_at(grant, now), } for grant in grants ], "trials": [ { "trialId": trial.id, "originalGrantId": trial.grant_id, "extensionGrantIds": [ grant.id for grant in grants if grant.source_type == "trial_extension" and grant.metadata_json.get("trialId") == trial.id ], "status": trial.status, "startsAt": _utc(trial.starts_at).isoformat(), "endsAt": _utc(trial.ends_at).isoformat(), "extensionCount": trial.extension_count, } for trial in trials ], "activations": [ { "activationId": activation.id, "deviceId": device.id, "friendlyName": device.friendly_name, "deviceStatus": device.status, "seatId": activation.seat_id, "status": activation.status, "activatedAt": _utc(activation.activated_at).isoformat(), "deactivatedAt": ( None if activation.deactivated_at is None else _utc(activation.deactivated_at).isoformat() ), "releaseReason": activation.release_reason, } for activation, device in activation_rows ], "timeline": [ { "eventId": event.id, "occurredAt": _utc(event.occurred_at).isoformat(), "actorType": event.actor_type, "actorId": event.actor_id, "action": event.action, "targetType": event.target_type, "targetId": event.target_id, "reason": event.reason, "correlationId": event.correlation_id, } for event in timeline ], } def _device_recovery_preview_mapping( self, device: Device, ) -> dict[str, object]: if not isinstance(device, Device): raise TypeError("device must be a Device") if device.status != "lost": raise ServerLicensingError( ServerErrorCode.CONFLICT, "only a device marked lost can be recovered", status_code=409, ) installations = tuple( self._session.scalars( select(DeviceInstallation) .where(DeviceInstallation.device_id == device.id) .order_by(DeviceInstallation.id) ) ) activations = tuple( self._session.scalars( select(Activation) .where(Activation.device_id == device.id) .order_by(Activation.id) ) ) active_activations = tuple( activation for activation in activations if activation.status == "active" ) active_license_ids = sorted( {activation.license_id for activation in active_activations} ) state_mapping = { "deviceId": device.id, "status": device.status, "ownerUserId": device.owner_user_id, "ownerOrganizationId": device.owner_organization_id, "evidenceDigests": dict(sorted(device.evidence_digests.items())), "installations": [ { "id": installation.id, "installationId": installation.installation_id, "status": installation.status, } for installation in installations ], "activations": [ { "id": activation.id, "licenseId": activation.license_id, "installationId": activation.installation_id, "seatId": activation.seat_id, "status": activation.status, } for activation in activations ], } return { "deviceId": device.id, "friendlyName": device.friendly_name, "currentStatus": device.status, "revokedInstallationCount": sum( installation.status == "revoked" for installation in installations ), "activeActivationCount": len(active_activations), "activeLicenseIds": active_license_ids, "preservesOtherLicenseAllocations": True, "requiresNewDeviceProofKey": True, "stateDigest": hashlib.sha256(canonicalize_json(state_mapping)).hexdigest(), "requiresConfirmation": True, "executed": False, } def _trial_extension_preview_mapping( self, trial: Trial, license_row: License, now: datetime, ) -> dict[str, object]: if not isinstance(trial, Trial): raise TypeError("trial must be a Trial") if not isinstance(license_row, License): raise TypeError("license_row must be a License") if trial.license_id != license_row.id: raise ValueError("trial does not belong to license_row") if not isinstance(now, datetime) or now.tzinfo is None or now.utcoffset() is None: raise ValueError("now must be a timezone-aware datetime") if trial.status != "active": raise ServerLicensingError( ServerErrorCode.TRIAL_NOT_ELIGIBLE, "trial is not eligible for extension", status_code=409, ) original_grant = self._session.get(Grant, trial.grant_id) if original_grant is None or original_grant.license_id != license_row.id: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "trial grant record is missing", status_code=500, ) sku = self._catalog.require_sku(original_grant.sku_id, require_active=False) if sku.kind is not SkuKind.TRIAL: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "trial grant references a non-trial SKU", status_code=500, ) policy = self._catalog.device_policies.get(sku.device_policy_id) if policy is None: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "trial SKU references an unknown device policy", status_code=500, ) if policy.trial_extension_days < 1 or policy.trial_extension_limit < 1: raise ServerLicensingError( ServerErrorCode.TRIAL_NOT_ELIGIBLE, "trial extension is not enabled by the product catalog", status_code=409, ) if trial.extension_count >= policy.trial_extension_limit: raise ServerLicensingError( ServerErrorCode.TRIAL_NOT_ELIGIBLE, "the trial extension allowance is exhausted", status_code=409, ) extension_start = max(now, _utc(trial.ends_at)) absolute_trial_expiry = _utc(trial.starts_at) + timedelta( hours=TRIAL_MAXIMUM_TOTAL_DURATION_HOURS ) if extension_start >= absolute_trial_expiry: raise ServerLicensingError( ServerErrorCode.TRIAL_NOT_ELIGIBLE, "the shipped application trial ceiling has been reached", status_code=409, ) requested_extension_end = extension_start + timedelta( days=policy.trial_extension_days ) extension_end = min(requested_extension_end, absolute_trial_expiry) return { "trialId": trial.id, "licenseId": license_row.id, "skuId": sku.sku_id, "originalGrantId": original_grant.id, "extensionNumber": trial.extension_count + 1, "extensionDays": policy.trial_extension_days, "extensionStartsAt": extension_start.isoformat(), "extensionEndsAt": extension_end.isoformat(), "absoluteTrialExpiresAt": absolute_trial_expiry.isoformat(), "cappedByApplicationMaximum": extension_end < requested_extension_end, "remainingExtensionsAfterExecution": ( policy.trial_extension_limit - trial.extension_count - 1 ), "entitlements": list(sku.entitlements), "previewRevocationGeneration": license_row.revocation_generation, "nextRevocationGeneration": license_row.revocation_generation + 1, "requiresConfirmation": True, "executed": False, } def _grant_revocation_preview_mapping( self, license_row: License, grant: Grant, now: datetime, ) -> dict[str, object]: if not isinstance(license_row, License): raise TypeError("license_row must be a License") if not isinstance(grant, Grant): raise TypeError("grant must be a Grant") if grant.license_id != license_row.id: raise ValueError("grant does not belong to license_row") if not isinstance(now, datetime) or now.tzinfo is None or now.utcoffset() is None: raise ValueError("now must be a timezone-aware datetime") if grant.status != "active": raise ServerLicensingError( ServerErrorCode.GRANT_NOT_ACTIVE, "grant is not active", status_code=409, ) grants = tuple( self._session.scalars( select(Grant) .where(Grant.license_id == license_row.id) .order_by(Grant.starts_at, Grant.id) ) ) active_before = tuple( value for value in grants if self._grant_is_usable_at(value, now) ) active_after = tuple(value for value in active_before if value.id != grant.id) remaining_usable = tuple( value for value in grants if value.id != grant.id and self._grant_is_usable_at(value, now, include_scheduled=True) ) entitlements_before = set(self._entitlements_for_grants(active_before)) entitlements_after = set(self._entitlements_for_grants(active_after)) target_sku = self._catalog.require_sku(grant.sku_id, require_active=False) return { "licenseId": license_row.id, "grantId": grant.id, "skuId": grant.sku_id, "sourceType": grant.source_type, "grantActiveNow": self._grant_is_usable_at(grant, now), "targetEntitlements": list(target_sku.entitlements), "removedEntitlements": sorted(entitlements_before - entitlements_after), "preservedEntitlements": sorted(entitlements_after), "remainingActiveGrantIds": [value.id for value in active_after], "remainingUsableGrantIds": [value.id for value in remaining_usable], "wouldDeactivateLicense": not remaining_usable, "previewRevocationGeneration": license_row.revocation_generation, "nextRevocationGeneration": license_row.revocation_generation + 1, "requiresConfirmation": True, "executed": False, } def _synchronize_trial_after_grant_revocation( self, license_id: str, grant_id: str, now: datetime, ) -> None: normalized_license = validate_identifier(license_id, "license_id") normalized_grant = validate_identifier(grant_id, "grant_id") if not isinstance(now, datetime) or now.tzinfo is None or now.utcoffset() is None: raise ValueError("now must be a timezone-aware datetime") revoked_grant = self._session.get(Grant, normalized_grant) if revoked_grant is None or revoked_grant.license_id != normalized_license: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "revoked grant record is missing", status_code=500, ) trials = tuple( self._session.scalars( select(Trial) .where(Trial.license_id == normalized_license) .with_for_update() ) ) grants = tuple( self._session.scalars( select(Grant).where(Grant.license_id == normalized_license) ) ) for trial in trials: target_matches = ( revoked_grant.id == trial.grant_id or revoked_grant.metadata_json.get("trialId") == trial.id ) if not target_matches: continue remaining = tuple( value for value in grants if value.id != revoked_grant.id and value.status == "active" and value.ends_at is not None and _utc(value.ends_at) > now and ( value.id == trial.grant_id or value.metadata_json.get("trialId") == trial.id ) ) if remaining: trial.status = "active" trial.ends_at = max(_utc(value.ends_at) for value in remaining if value.ends_at) else: trial.status = "revoked" @staticmethod def _grant_is_usable_at( grant: Grant, now: datetime, *, include_scheduled: bool = False, ) -> bool: if not isinstance(grant, Grant): raise TypeError("grant must be a Grant") if not isinstance(now, datetime) or now.tzinfo is None or now.utcoffset() is None: raise ValueError("now must be a timezone-aware datetime") if not isinstance(include_scheduled, bool): raise TypeError("include_scheduled must be a Boolean") starts_in_time = include_scheduled or _utc(grant.starts_at) <= now ends_in_time = grant.ends_at is None or _utc(grant.ends_at) > now return grant.status == "active" and starts_in_time and ends_in_time def _entitlements_for_grants( self, grants: tuple[Grant, ...], ) -> tuple[str, ...]: if not isinstance(grants, tuple) or not all( isinstance(grant, Grant) for grant in grants ): raise TypeError("grants must be a tuple of Grant records") entitlements: set[str] = set() for grant in grants: sku = self._catalog.require_sku(grant.sku_id, require_active=False) entitlements.update(sku.entitlements) return tuple(sorted(entitlements)) def _active_edition_ids(self, license_id: str, now: datetime) -> tuple[str, ...]: values = [] for grant in self._session.scalars(self._active_grants_query(license_id, now)): sku = self._catalog.require_sku(grant.sku_id, require_active=False) if sku.kind is SkuKind.EDITION: values.append(sku.sku_id) return tuple(sorted(set(values))) def _surface_policy_for_license( self, license_row: License, ) -> ProductSurfacePolicy: if not isinstance(license_row, License): raise TypeError("license_row must be a License") if license_row.surface_profile_id is None: return ProductSurfacePolicy( profile_id=None, analysis_ids=tuple( sorted(product_surface_ids(PRODUCT_SURFACE_KIND_ANALYSIS)) ), main_tab_ids=tuple( sorted(product_surface_ids(PRODUCT_SURFACE_KIND_MAIN_TAB)) ), ) profile = self._session.get( ProductSurfaceProfile, license_row.surface_profile_id, ) if profile is None or profile.product_id != license_row.product_id: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "license product surface profile is missing", status_code=500, ) try: return ProductSurfacePolicy( profile_id=profile.id, inventory_revision=profile.inventory_revision, inventory_sha256=profile.inventory_sha256, analysis_ids=tuple(profile.analysis_ids_json), main_tab_ids=tuple(profile.main_tab_ids_json), ) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "license product surface profile is invalid or stale", status_code=500, ) from exc def _trial_policy_for_license( self, license_row: License, ) -> TrialPolicyClaim: if not isinstance(license_row, License): raise TypeError("license_row must be a License") trials = tuple( self._session.scalars( select(Trial).where(Trial.license_id == license_row.id) ) ) if len(trials) != 1: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "trial license must reference exactly one trial history record", status_code=500, ) trial = trials[0] try: return TrialPolicyClaim( original_started_at=_utc(trial.starts_at), final_expires_at=_utc(trial.ends_at), ) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "trial history exceeds the duration accepted by shipped clients", status_code=500, ) from exc def _issue_snapshot( self, license_row: License, activation: Activation, installation: DeviceInstallation, application_major_version: int, now: datetime, *, offline_certificate: bool = False, request_nonce: Optional[str] = None, ) -> SignedLicenseDocument: if not isinstance(offline_certificate, bool): raise TypeError("offline_certificate must be a Boolean") if offline_certificate != (request_nonce is not None): raise ValueError("offline snapshot nonce contract is invalid") grants = tuple(self._session.scalars(self._active_grants_query(license_row.id, now))) if not grants: raise ServerLicensingError( ServerErrorCode.GRANT_NOT_ACTIVE, "license has no active grants", status_code=409, ) entitlements: set[str] = set() summaries: list[GrantSummary] = [] minimums: list[int] = [] maximums: list[int] = [] expirations: list[datetime] = [] lease_day_candidates: list[int] = [] connected_refresh_hour_candidates: list[int] = [] offline_refresh_target_candidates: list[datetime] = [] has_trial = False has_non_trial = False has_subscription_payment_grace = False for grant in grants: sku = self._catalog.require_sku(grant.sku_id, require_active=False) sku_policy = self._catalog.device_policies.get(sku.device_policy_id) if sku_policy is None: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "grant SKU references an unknown device policy", status_code=500, ) candidate_lease_days = ( sku_policy.offline_certificate_days if offline_certificate else sku_policy.connected_lease_days ) if offline_certificate and candidate_lease_days < 1: continue if not ( sku.application_major_minimum <= application_major_version <= sku.application_major_maximum ): continue lease_day_candidates.append(candidate_lease_days) if offline_certificate: reminder_days = ( LEGACY_OFFLINE_REFRESH_REMINDER_DAYS if sku_policy.offline_refresh_reminder_days is None else sku_policy.offline_refresh_reminder_days ) candidate_expiry = now + timedelta(days=candidate_lease_days) if grant.ends_at is not None: candidate_expiry = min( candidate_expiry, _utc(grant.ends_at), ) offline_refresh_target_candidates.append( max( now, candidate_expiry - timedelta(days=reminder_days), ) ) else: connected_refresh_hour_candidates.append( LEGACY_CONNECTED_REFRESH_TARGET_HOURS if sku_policy.connected_refresh_hours is None else sku_policy.connected_refresh_hours ) entitlements.update(sku.entitlements) minimums.append(sku.application_major_minimum) maximums.append(sku.application_major_maximum) if grant.ends_at is not None: expirations.append(_utc(grant.ends_at)) has_trial = has_trial or sku.kind is SkuKind.TRIAL has_non_trial = has_non_trial or sku.kind is not SkuKind.TRIAL has_subscription_payment_grace = ( has_subscription_payment_grace or grant.metadata_json.get("subscriptionState") == "past_due" ) summaries.append( GrantSummary( grant_id=grant.id, source=grant.source_type, sku_id=sku.sku_id, label=sku.label, ends_at=None if grant.ends_at is None else _utc(grant.ends_at), ) ) if not entitlements or not minimums or not maximums: raise ServerLicensingError( ServerErrorCode.GRANT_NOT_ACTIVE, "no active grant covers this application major version", status_code=409, ) policy = self._catalog.device_policies.get(license_row.device_policy_id) if policy is None: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "license references an unknown device policy", status_code=500, ) lease_days = min(lease_day_candidates) if lease_day_candidates else 0 if lease_days < 1: raise ServerLicensingError( ServerErrorCode.SKU_NOT_AVAILABLE, "this license policy does not allow offline certificates", status_code=409, ) expiry = now + timedelta(days=lease_days) if expirations: expiry = min(expiry, min(expirations)) if offline_certificate: refresh = min(expiry, min(offline_refresh_target_candidates)) else: refresh = min( now + timedelta(hours=min(connected_refresh_hour_candidates)), expiry, ) not_before = now - timedelta(seconds=DEFAULT_NOT_BEFORE_SKEW_SECONDS) subject_type = SubjectType.LICENSE subject_id = license_row.id if license_row.owner_user_id is not None: subject_type = SubjectType.USER subject_id = license_row.owner_user_id elif license_row.owner_organization_id is not None: subject_type = SubjectType.ORGANIZATION subject_id = license_row.owner_organization_id state = ( LicenseState.TRIAL_ACTIVE if has_trial and not has_non_trial else LicenseState.SUBSCRIPTION_PAYMENT_GRACE if has_subscription_payment_grace else LicenseState.LICENSED_ACTIVE ) surface_policy = self._surface_policy_for_license(license_row) trial_policy = ( self._trial_policy_for_license(license_row) if state in (LicenseState.TRIAL_ACTIVE, LicenseState.TRIAL_PROVISIONAL) else None ) snapshot = EntitlementSnapshot( product_id=PRODUCT_ID, snapshot_id=new_identifier("snapshot"), license_id=license_row.id, signature_key_id=self._signer.key_id, subject_type=subject_type, subject_id=subject_id, license_state=state, application_major_minimum=max(minimums), application_major_maximum=min(maximums), entitlements=tuple(sorted(entitlements)), grant_summaries=tuple(sorted(summaries, key=lambda value: value.grant_id)), device_policy=SnapshotDevicePolicy( policy_id=policy.policy_id, maximum_devices=policy.maximum_devices, named_seats=policy.named_seats, seat_id=activation.seat_id, ), issued_at=now, not_before=not_before, refresh_after=refresh, lease_expires_at=expiry, offline_expires_at=expiry if offline_certificate else None, device_id=installation.device_id, device_key_thumbprint=installation.key_thumbprint, catalog_revision=self._catalog.revision, revocation_generation=license_row.revocation_generation, nonce=( random_token(26, SERIAL_ALPHABET).lower() if request_nonce is None else request_nonce ), surface_policy=surface_policy, trial_policy=trial_policy, ) document = self._signer.sign_payload(snapshot.to_payload_mapping()) document_mapping = document.to_mapping() self._session.add( Lease( id=new_identifier("lease"), snapshot_id=snapshot.snapshot_id, license_id=license_row.id, activation_id=activation.id, signing_key_id=self._signer.key_id, snapshot_digest=hashlib.sha256(canonicalize_json(document_mapping)).digest(), issued_at=now, expires_at=expiry, revocation_generation=license_row.revocation_generation, document_json=document_mapping, ) ) return document def _request_digest(self, mapping: Mapping[str, object]) -> bytes: if not isinstance(mapping, Mapping): raise TypeError("mapping must be a mapping") return hashlib.sha256(canonicalize_json(mapping)).digest() def _load_idempotent_response( self, scope: str, subject_key: str, idempotency_key: str, request_digest: bytes, now: datetime, ) -> Optional[dict[str, object]]: record = self._session.scalar( select(IdempotencyRecord) .where( IdempotencyRecord.scope == scope, IdempotencyRecord.subject_key == subject_key, IdempotencyRecord.idempotency_key == idempotency_key, ) .with_for_update() ) if record is None: return None if _utc(record.expires_at) <= now: self._session.delete(record) self._session.flush() return None if not hmac.compare_digest(record.request_digest, request_digest): raise ServerLicensingError( ServerErrorCode.IDEMPOTENCY_CONFLICT, "idempotency key was already used for a different request", status_code=409, ) return record.response_json def _store_idempotent_response( self, scope: str, subject_key: str, idempotency_key: str, request_digest: bytes, response: dict[str, object], now: datetime, ) -> None: self._session.add( IdempotencyRecord( id=new_identifier("idempotency"), scope=scope, subject_key=subject_key, idempotency_key=idempotency_key, request_digest=request_digest, response_status=200, response_json=response, expires_at=now + timedelta(days=DEFAULT_IDEMPOTENCY_DAYS), ) ) def _mutation_from_mapping( self, raw: Mapping[str, object], *, idempotent_replay: bool, ) -> LicenseMutationResult: if not isinstance(raw, Mapping): raise TypeError("raw must be a mapping") document_raw = raw.get("licenseDocument") if not isinstance(document_raw, Mapping): raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, "stored idempotency response is invalid", status_code=500, ) return LicenseMutationResult( license_id=validate_identifier(raw.get("licenseId"), "licenseId"), activation_id=validate_identifier(raw.get("activationId"), "activationId"), device_id=validate_identifier(raw.get("deviceId"), "deviceId"), document=SignedLicenseDocument.from_mapping(document_raw), recovered_existing_device=bool(raw.get("recoveredExistingDevice")), idempotent_replay=idempotent_replay, ) __all__ = [ "DeviceEnrollment", "LicenseMutationResult", "LicensingService", "SerialGenerationResult", ]