"""FastAPI composition and versioned licensing routes.""" from __future__ import annotations import base64 from datetime import datetime, timedelta, timezone from typing import Optional from fastapi import APIRouter, Depends, FastAPI, Query, Request from fastapi.responses import JSONResponse from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from sqlalchemy import Engine, or_, select, text from sqlalchemy.orm import Session, sessionmaker import stripe from licensing_shared.catalog import load_builtin_catalog from licensing_shared.device_proof import DeviceActionProof, verify_device_action_proof from licensing_shared.errors import LicensingError from licensing_shared.models import parse_rfc3339 from licensing_shared.offline import ( OfflineRequestType, parse_and_verify_offline_request, ) from licensing_shared.verifier import ED25519_PUBLIC_KEY_BYTES from .auth import AuthenticatedPrincipal, OIDCTokenValidator, TokenValidator from .admin_audit import AuditAdministrationService from .billing import ( AccountBillingService, BillingProvider, StripeBillingProvider, commerce_capabilities, subscription_statuses, ) from .catalog_administration import ( CatalogAdministrationService, CatalogBoundSnapshotSigner, inspect_catalog_registry, ) from .config import ServerSettings from .constants import ( API_V1_PREFIX, MAX_API_REQUEST_BODY_BYTES, MAX_WEBHOOK_BODY_BYTES, SERVER_NAME, ) from .database import create_database_engine, create_session_factory from .errors import ServerErrorCode, ServerLicensingError from .models import ( Activation, Device, Grant, License, Membership, OrganizationSeat, User, ) from .middleware import RequestBodyLimitMiddleware from .offline_approvals import OfflineTrialApprovalService from .offline_portal import OFFLINE_ACTIVATION_PATH, offline_activation_response from .rate_limits import ( DatabaseRateLimiter, RequestRateLimitContextMiddleware, request_rate_limit_context, ) from .schemas import ( AccountLicenseActivationRequest, ActivationDeactivateRequest, ActivationRefreshRequest, AdminCatalogPublicationExecuteRequest, AdminCatalogPublicationPreviewRequest, AdminCatalogStageRequest, AdminDeviceRecoveryExecuteRequest, AdminDeviceRecoveryPreviewRequest, AdminGrantRevocationExecuteRequest, AdminGrantRevocationPreviewRequest, AdminAuditExportRequest, AdminAuditSearchRequest, AdminOfflineApprovalRequest, AdminOfflineRejectionRequest, AdminPrivacyDeletionExecuteRequest, AdminPrivacyDeletionPreviewRequest, AdminSerialBatchRevocationExecuteRequest, AdminSerialBatchRevocationPreviewRequest, AdminTrialExtensionExecuteRequest, AdminTrialExtensionPreviewRequest, BillingPortalSessionRequest, CheckoutSessionRequest, DeviceRenameRequest, LicenseSurfaceProfileAssignmentRequest, OfflineActivationRequest, OfflineRenewalRequest, OfflineReleaseRequest, OfflineTrialStatusRequest, OfflineTrialSubmissionRequest, OrganizationCreateRequest, OrganizationInvitationAcceptRequest, OrganizationInvitationCreateRequest, OrganizationLicenseAttachRequest, OrganizationSeatAssignRequest, OrganizationSeatReleaseRequest, PrivacyDeletionCancelRequest, PrivacyDeletionRequest, PrivacyExportRequest, ProductSurfaceProfileArchiveRequest, ProductSurfaceProfileCreateRequest, RemoteDeviceReleaseRequest, SerialBatchRequest, SerialClaimRequest, SerialRedemptionRequest, TrialStartRequest, ) from .security import Ed25519SnapshotSigner, SnapshotSigner, new_identifier from .privacy import AccountPrivacyService from .serial_administration import SerialAdministrationService from .signing_key_administration import ( RegistryBoundSnapshotSigner, inspect_signing_key_registry, ) from .services import DeviceEnrollment, LicenseMutationResult, LicensingService from .subscriptions import ( STRIPE_BILLING_INCIDENT_EVENT_TYPES, STRIPE_SUBSCRIPTION_EVENT_TYPES, enqueue_stripe_billing_incident_event, enqueue_stripe_subscription_event, normalize_stripe_billing_incident_event, normalize_stripe_subscription_event, ) AUTHENTICATION_SCHEME = HTTPBearer(auto_error=False) ADMIN_SERIAL_SCOPE = "licensing:serials:generate" ADMIN_CATALOG_SCOPE = "licensing:catalog:publish" ADMIN_SUPPORT_SCOPE = "licensing:support:manage" ADMIN_AUDIT_SCOPE = "licensing:audit:read" ADMIN_OFFLINE_SCOPE = "licensing:offline:approve" ADMIN_PRIVACY_SCOPE = "licensing:privacy:manage" ADMIN_PRODUCT_SCOPE = "licensing:products:configure" ADMIN_REAUTHENTICATION_MAX_AGE = timedelta(minutes=10) CUSTOMER_REAUTHENTICATION_MAX_AGE = timedelta(minutes=10) def _error_mapping(error: ServerLicensingError) -> dict[str, object]: return { "code": error.code.value, "message": str(error), "retryable": error.retryable, "existingResourceId": error.existing_resource_id, } def _mutation_mapping(result: LicenseMutationResult) -> dict[str, object]: if not isinstance(result, LicenseMutationResult): raise TypeError("result must be a LicenseMutationResult") response = result.to_response_mapping() response["idempotentReplay"] = result.idempotent_replay return response def _privacy_response(document: dict[str, object]) -> JSONResponse: if not isinstance(document, dict): raise TypeError("document must be a dictionary") return JSONResponse( content=document, headers={ "Cache-Control": "no-store, max-age=0", "Pragma": "no-cache", }, ) def _enforce_request_rate_limits(request: Request) -> None: if not isinstance(request, Request): raise TypeError("request must be a Request") limiter: DatabaseRateLimiter = request.app.state.rate_limiter limiter.enforce_request(request_rate_limit_context(request.state)) def _principal_from_credentials( request: Request, credentials: Optional[HTTPAuthorizationCredentials], *, required: bool, ) -> Optional[AuthenticatedPrincipal]: if not isinstance(request, Request): raise TypeError("request must be a Request") if not isinstance(required, bool): raise TypeError("required must be a Boolean") if credentials is None: if required: raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "sign in is required", status_code=401, ) return None if credentials.scheme.lower() != "bearer": raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "authorization scheme must be Bearer", status_code=401, ) validator: TokenValidator = request.app.state.token_validator principal = validator.validate(credentials.credentials) if not isinstance(principal, AuthenticatedPrincipal): raise TypeError("token validator must return an AuthenticatedPrincipal") limiter: DatabaseRateLimiter = request.app.state.rate_limiter limiter.enforce_principal( request_rate_limit_context(request.state), principal.issuer, principal.subject, ) return principal def _recent_admin_principal( request: Request, credentials: Optional[HTTPAuthorizationCredentials], required_scope: str, operation: str, ) -> AuthenticatedPrincipal: if not isinstance(required_scope, str) or not required_scope.strip(): raise ValueError("required_scope must be a non-empty string") if not isinstance(operation, str) or not operation.strip(): raise ValueError("operation must be a non-empty string") principal = _principal_from_credentials(request, credentials, required=True) assert principal is not None principal.require_scope(required_scope) if ( principal.authentication_time is None or datetime.now(timezone.utc) - principal.authentication_time > ADMIN_REAUTHENTICATION_MAX_AGE ): raise ServerLicensingError( ServerErrorCode.REAUTHENTICATION_REQUIRED, f"recent reauthentication is required for {operation.strip()}", status_code=403, ) return principal def _recent_customer_principal( request: Request, credentials: Optional[HTTPAuthorizationCredentials], operation: str, ) -> AuthenticatedPrincipal: if not isinstance(operation, str) or not operation.strip(): raise ValueError("operation must be a non-empty string") principal = _principal_from_credentials(request, credentials, required=True) assert principal is not None if ( principal.authentication_time is None or datetime.now(timezone.utc) - principal.authentication_time > CUSTOMER_REAUTHENTICATION_MAX_AGE ): raise ServerLicensingError( ServerErrorCode.REAUTHENTICATION_REQUIRED, f"recent reauthentication is required for {operation.strip()}", status_code=403, ) return principal def _resolve_user(session: Session, principal: AuthenticatedPrincipal) -> User: if not isinstance(session, Session): raise TypeError("session must be a Session") if not isinstance(principal, AuthenticatedPrincipal): raise TypeError("principal must be an AuthenticatedPrincipal") user = session.scalar( select(User).where( User.external_issuer == principal.issuer, User.external_subject == principal.subject, ) ) if user is None: user = User( id=new_identifier("user"), external_issuer=principal.issuer, external_subject=principal.subject, verified_email=principal.verified_email, status="active", is_server_admin=False, ) session.add(user) elif principal.verified_email is not None: user.verified_email = principal.verified_email session.commit() if user.status != "active": raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "account is not active", status_code=403, ) return user def _enrollment_from_request(request_device) -> DeviceEnrollment: return DeviceEnrollment( installation_id=request_device.installation_id, public_key_der=request_device.public_key_der(), key_thumbprint=request_device.device_key_thumbprint, key_provider=request_device.key_provider, friendly_name=request_device.friendly_name, evidence=request_device.evidence, ) def _verified_proof( proof_request, action: str, payload: dict[str, object], *, installation_id: str, request: Optional[Request] = None, ) -> DeviceActionProof: proof = verify_device_action_proof( proof_request.model_dump(by_alias=True), action, payload, ) if proof.installation_id != installation_id: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "device proof installation does not match the request", status_code=403, ) if request is not None: limiter: DatabaseRateLimiter = request.app.state.rate_limiter limiter.enforce_verified_device( request_rate_limit_context(request.state), proof.device_key_thumbprint, ) return proof def _service(request: Request, session: Session) -> LicensingService: if not isinstance(request, Request): raise TypeError("request must be a Request") if not isinstance(session, Session): raise TypeError("session must be a Session") settings: ServerSettings = request.app.state.settings signer = CatalogBoundSnapshotSigner( session, request.app.state.catalog, RegistryBoundSnapshotSigner( session, request.app.state.signer, ), ) return LicensingService( session, request.app.state.catalog, signer, settings.serial_pepper, settings.fingerprint_pepper, ) def _require_published_runtime_catalog(request: Request, session: Session) -> None: if not isinstance(request, Request): raise TypeError("request must be a Request") if not isinstance(session, Session): raise TypeError("session must be a Session") status = CatalogAdministrationService( session, request.app.state.catalog, ).status() if status["catalogReady"] is not True: raise ServerLicensingError( ServerErrorCode.CATALOG_UNAVAILABLE, "runtime catalog is not the active reviewed catalog publication", status_code=503, retryable=True, ) def _billing_service(request: Request, session: Session) -> AccountBillingService: if not isinstance(request, Request): raise TypeError("request must be a Request") if not isinstance(session, Session): raise TypeError("session must be a Session") _require_published_runtime_catalog(request, session) settings: ServerSettings = request.app.state.settings return AccountBillingService( session, request.app.state.catalog, request.app.state.billing_provider, settings.stripe_price_sku_map, settings.stripe_commerce_enabled, settings.stripe_checkout_success_url, settings.stripe_checkout_cancel_url, settings.stripe_billing_portal_return_url, ) def _offline_trial_service(request: Request) -> OfflineTrialApprovalService: if not isinstance(request, Request): raise TypeError("request must be a Request") settings: ServerSettings = request.app.state.settings factory: sessionmaker[Session] = request.app.state.session_factory return OfflineTrialApprovalService( factory, request.app.state.catalog, request.app.state.signer, settings.serial_pepper, settings.fingerprint_pepper, ) def _privacy_service(request: Request, session: Session) -> AccountPrivacyService: if not isinstance(request, Request): raise TypeError("request must be a Request") if not isinstance(session, Session): raise TypeError("session must be a Session") settings: ServerSettings = request.app.state.settings return AccountPrivacyService(session, settings.fingerprint_pepper) def create_api_router() -> APIRouter: router = APIRouter( prefix=API_V1_PREFIX, dependencies=[Depends(_enforce_request_rate_limits)], ) @router.get("/health") def health(request: Request) -> dict[str, object]: factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: session.execute(text("SELECT 1")) return {"status": "ok", "service": SERVER_NAME} @router.get("/readiness") def readiness(request: Request) -> dict[str, object]: factory: sessionmaker[Session] = request.app.state.session_factory try: with factory() as session: session.execute(text("SELECT 1")) catalog_registry = inspect_catalog_registry( session, request.app.state.catalog, ) except Exception as exc: raise ServerLicensingError( ServerErrorCode.CATALOG_UNAVAILABLE, "catalog registry is unavailable", status_code=503, retryable=True, ) from exc if catalog_registry.get("catalogReady") is not True: raise ServerLicensingError( ServerErrorCode.CATALOG_UNAVAILABLE, "runtime catalog is not the active reviewed catalog publication", status_code=503, retryable=True, ) try: with factory() as session: registry = inspect_signing_key_registry( session, request.app.state.signer, datetime.now(timezone.utc), ) except Exception as exc: raise ServerLicensingError( ServerErrorCode.SIGNING_UNAVAILABLE, "signing-key registry is unavailable", status_code=503, retryable=True, ) from exc signer: SnapshotSigner = request.app.state.signer public_key = signer.public_key_bytes() if ( not isinstance(public_key, bytes) or len(public_key) != ED25519_PUBLIC_KEY_BYTES ): raise ServerLicensingError( ServerErrorCode.SIGNING_UNAVAILABLE, "licensing signer is not ready", status_code=503, retryable=True, ) if registry.get("issuanceReady") is not True: raise ServerLicensingError( ServerErrorCode.SIGNING_UNAVAILABLE, "configured signing key is not active in the public metadata registry", status_code=503, retryable=True, ) return { "status": "ready", "service": SERVER_NAME, "catalogRevision": request.app.state.catalog.revision, "catalogSha256": request.app.state.catalog.sha256(), "catalogStateDigest": catalog_registry["stateDigest"], "signingKeyId": signer.key_id, "signingKeyStateDigest": registry["stateDigest"], } @router.get("/catalog/public") def public_catalog(request: Request) -> dict[str, object]: factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: status = inspect_catalog_registry( session, request.app.state.catalog, ) if status.get("catalogReady") is not True: raise ServerLicensingError( ServerErrorCode.CATALOG_UNAVAILABLE, "public catalog is not the active reviewed publication", status_code=503, retryable=True, ) return request.app.state.catalog.to_mapping() @router.get("/auth/native/config") def native_authorization_config(request: Request) -> dict[str, object]: settings: ServerSettings = request.app.state.settings if ( settings.oidc_authorization_endpoint is None or settings.oidc_token_endpoint is None or settings.oidc_native_client_id is None ): raise ServerLicensingError( ServerErrorCode.AUTHENTICATION_REQUIRED, "native account sign-in is not configured", status_code=503, ) return { "authorizationEndpoint": settings.oidc_authorization_endpoint, "tokenEndpoint": settings.oidc_token_endpoint, "clientId": settings.oidc_native_client_id, "scopes": list(settings.oidc_native_scopes), "audience": settings.oidc_audience, } @router.post("/serials/redeem") def redeem_serial( body: SerialRedemptionRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _principal_from_credentials(request, credentials, required=False) proof = _verified_proof( body.device_proof, "serial.redeem", body.proof_payload(), installation_id=body.device.installation_id, request=request, ) if ( proof.device_public_key_base64 != body.device.device_public_key or proof.device_key_thumbprint != body.device.device_key_thumbprint ): raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "device proof key does not match the enrollment", status_code=403, ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: actor_id = None if principal is not None: actor_id = _resolve_user(session, principal).id result = _service(request, session).redeem_serial( body.serial, _enrollment_from_request(body.device), body.application_major_version, body.idempotency_key, body.correlation_id, actor_user_id=actor_id, target_license_id=body.target_license_id, ) return _mutation_mapping(result) @router.post("/trials/start") def start_trial( body: TrialStartRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _principal_from_credentials(request, credentials, required=False) proof = _verified_proof( body.device_proof, "trial.start", body.proof_payload(), installation_id=body.device.installation_id, request=request, ) if ( proof.device_public_key_base64 != body.device.device_public_key or proof.device_key_thumbprint != body.device.device_key_thumbprint ): raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "device proof key does not match the enrollment", status_code=403, ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: actor_id = None if principal is not None: actor_id = _resolve_user(session, principal).id result = _service(request, session).start_trial( _enrollment_from_request(body.device), body.application_major_version, body.idempotency_key, body.correlation_id, actor_user_id=actor_id, ) return _mutation_mapping(result) @router.post("/offline/requests") def activate_offline_request( body: OfflineActivationRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _principal_from_credentials(request, credentials, required=False) offline_request = parse_and_verify_offline_request(body.offline_request) payload = offline_request.payload request.app.state.rate_limiter.enforce_verified_device( request_rate_limit_context(request.state), payload.device_key_thumbprint, ) if payload.request_type is not OfflineRequestType.ACTIVATE: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, "offline request is not an activation request", ) if payload.requested_sku_id is None: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, "offline activation request must bind an intended SKU", ) if body.target_license_id != payload.target_license_id: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "target license does not match the signed offline request", status_code=403, ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: actor_id = None if principal is not None: actor_id = _resolve_user(session, principal).id result = _service(request, session).redeem_serial( body.serial, DeviceEnrollment( installation_id=payload.installation_id, public_key_der=base64.b64decode( payload.device_public_key_base64, validate=True, ), key_thumbprint=payload.device_key_thumbprint, key_provider=payload.key_provider, friendly_name=payload.friendly_name, evidence=dict(payload.evidence_digests), ), payload.application_major_version, body.idempotency_key, body.correlation_id, actor_user_id=actor_id, target_license_id=payload.target_license_id, offline_certificate=True, offline_request_nonce=payload.nonce, expected_sku_id=payload.requested_sku_id, ) response = _mutation_mapping(result) response["offlineRequestId"] = payload.request_id response["offlineRequestNonce"] = payload.nonce return response @router.post("/activations/refresh") def refresh_activation( body: ActivationRefreshRequest, request: Request, ) -> dict[str, object]: proof = _verified_proof( body.device_proof, "activation.refresh", body.proof_payload(), installation_id=body.installation_id, request=request, ) public_key_der = base64.b64decode(proof.device_public_key_base64, validate=True) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: result = _service(request, session).refresh_activation( body.license_id, body.installation_id, public_key_der, proof.device_key_thumbprint, body.application_major_version, body.correlation_id, evidence=body.evidence, ) return _mutation_mapping(result) @router.post("/activations/deactivate") def deactivate_activation( body: ActivationDeactivateRequest, request: Request, ) -> dict[str, object]: proof = _verified_proof( body.device_proof, "activation.deactivate", body.proof_payload(), installation_id=body.installation_id, request=request, ) public_key_der = base64.b64decode(proof.device_public_key_base64, validate=True) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: activation_id = _service(request, session).deactivate_device( body.license_id, body.installation_id, public_key_der, proof.device_key_thumbprint, body.correlation_id, reason=body.reason, idempotency_key=body.idempotency_key or proof.nonce, evidence=body.evidence, ) return {"activationId": activation_id, "status": "deactivated"} @router.post("/activations/release-file") def release_offline_activation( body: OfflineReleaseRequest, request: Request, ) -> dict[str, object]: offline_request = parse_and_verify_offline_request(body.offline_request) payload = offline_request.payload request.app.state.rate_limiter.enforce_verified_device( request_rate_limit_context(request.state), payload.device_key_thumbprint, ) if payload.request_type is not OfflineRequestType.RELEASE: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, "offline request is not a release request", ) if payload.requested_sku_id is not None: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, "offline release request must not contain a purchase SKU", ) if payload.target_license_id is None or payload.device_id is None: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, "offline release request must bind a license and device", ) public_key_der = base64.b64decode( payload.device_public_key_base64, validate=True, ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: activation_id = _service(request, session).deactivate_device( payload.target_license_id, payload.installation_id, public_key_der, payload.device_key_thumbprint, payload.request_id, reason="offline_device_release", idempotency_key=payload.request_id, expected_device_id=payload.device_id, ) return { "activationId": activation_id, "status": "deactivated", "offlineRequestId": payload.request_id, } @router.get("/me/licenses") def my_licenses( request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _principal_from_credentials(request, credentials, required=True) assert principal is not None factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) organization_ids = select(Membership.organization_id).where( Membership.user_id == user.id, Membership.status == "active", or_( Membership.valid_until.is_(None), Membership.valid_until > datetime.now(timezone.utc), ), ) licenses = tuple( session.scalars( select(License).where( or_( License.owner_user_id == user.id, License.owner_organization_id.in_(organization_ids), ) ) ) ) values = [] for license_row in licenses: assigned_seat = None can_manage_billing = license_row.owner_user_id == user.id if license_row.owner_organization_id is not None: assigned_seat = session.scalar( select(OrganizationSeat).where( OrganizationSeat.license_id == license_row.id, OrganizationSeat.assigned_user_id == user.id, OrganizationSeat.status == "assigned", ) ) grants = tuple( session.scalars( select(Grant).where(Grant.license_id == license_row.id) ) ) device_query = ( select(Device) .join(Activation, Activation.device_id == Device.id) .where( Activation.license_id == license_row.id, Activation.status == "active", ) ) if license_row.owner_organization_id is not None: can_manage_organization = session.scalar( select(Membership.id).where( Membership.organization_id == license_row.owner_organization_id, Membership.user_id == user.id, Membership.status == "active", Membership.role.in_(("owner", "admin")), or_( Membership.valid_until.is_(None), Membership.valid_until > datetime.now(timezone.utc), ), ) ) is not None can_manage_billing = can_manage_organization if not can_manage_organization: if assigned_seat is None: active_devices = [] else: active_devices = session.scalars( device_query.where( Activation.seat_id == assigned_seat.id ) ).all() else: active_devices = session.scalars(device_query).all() else: active_devices = session.scalars(device_query).all() values.append( { "licenseId": license_row.id, "status": license_row.status, "devicePolicyId": license_row.device_policy_id, "organizationId": license_row.owner_organization_id, "seatId": None if assigned_seat is None else assigned_seat.id, "canActivate": ( license_row.owner_user_id == user.id or assigned_seat is not None ), "canManageBilling": can_manage_billing, "subscriptions": subscription_statuses(session, license_row.id), "grants": [ { "grantId": grant.id, "skuId": grant.sku_id, "source": grant.source_type, "status": grant.status, } for grant in grants ], "devices": [ { "deviceId": device.id, "friendlyName": device.friendly_name, "status": device.status, } for device in active_devices ], } ) settings: ServerSettings = request.app.state.settings commerce = commerce_capabilities( request.app.state.catalog, settings.stripe_commerce_enabled, settings.stripe_price_sku_map, ) return {"licenses": values, "commerce": commerce} @router.post("/me/licenses/{license_id}/checkout-sessions") def create_checkout_session( license_id: str, body: CheckoutSessionRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_customer_principal( request, credentials, "subscription checkout", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _billing_service(request, session).create_checkout_session( license_id, body.sku_id, user.id, body.idempotency_key, body.correlation_id, ) @router.post("/me/licenses/{license_id}/billing-portal-sessions") def create_billing_portal_session( license_id: str, body: BillingPortalSessionRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_customer_principal( request, credentials, "billing management", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _billing_service(request, session).create_portal_session( license_id, user.id, body.idempotency_key, body.correlation_id, ) @router.post("/me/licenses/{license_id}/activate") def activate_account_license( license_id: str, body: AccountLicenseActivationRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _principal_from_credentials(request, credentials, required=True) assert principal is not None proof_payload = body.proof_payload() proof_payload["licenseId"] = license_id proof = _verified_proof( body.device_proof, "account.license.activate", proof_payload, installation_id=body.device.installation_id, request=request, ) if ( proof.device_public_key_base64 != body.device.device_public_key or proof.device_key_thumbprint != body.device.device_key_thumbprint ): raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "device proof key does not match the enrollment", status_code=403, ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) result = _service(request, session).activate_owned_license( license_id, _enrollment_from_request(body.device), body.application_major_version, body.idempotency_key, body.correlation_id, user.id, ) return _mutation_mapping(result) @router.post("/me/licenses/{license_id}/offline-renewal") def renew_offline_license( license_id: str, body: OfflineRenewalRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _principal_from_credentials(request, credentials, required=True) assert principal is not None if ( principal.authentication_time is None or datetime.now(timezone.utc) - principal.authentication_time > CUSTOMER_REAUTHENTICATION_MAX_AGE ): raise ServerLicensingError( ServerErrorCode.REAUTHENTICATION_REQUIRED, "recent reauthentication is required to renew an offline license", status_code=403, ) offline_request = parse_and_verify_offline_request(body.offline_request) payload = offline_request.payload request.app.state.rate_limiter.enforce_verified_device( request_rate_limit_context(request.state), payload.device_key_thumbprint, ) if payload.request_type is not OfflineRequestType.RENEW: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, "offline request is not a renewal request", ) if payload.target_license_id != license_id or payload.device_id is None: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "offline renewal does not match the requested license and device", status_code=403, ) if payload.requested_sku_id is not None: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, "offline renewal must not contain a purchase SKU", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) result = _service(request, session).activate_owned_license( license_id, DeviceEnrollment( installation_id=payload.installation_id, public_key_der=base64.b64decode( payload.device_public_key_base64, validate=True, ), key_thumbprint=payload.device_key_thumbprint, key_provider=payload.key_provider, friendly_name=payload.friendly_name, evidence={}, ), payload.application_major_version, body.idempotency_key, body.correlation_id, user.id, offline_certificate=True, offline_request_nonce=payload.nonce, expected_device_id=payload.device_id, ) response = _mutation_mapping(result) response["offlineRequestId"] = payload.request_id response["offlineRequestNonce"] = payload.nonce return response @router.post("/offline/trial-requests") def submit_offline_trial_request( body: OfflineTrialSubmissionRequest, request: Request, ) -> dict[str, object]: offline_request = parse_and_verify_offline_request(body.offline_request) request.app.state.rate_limiter.enforce_verified_device( request_rate_limit_context(request.state), offline_request.payload.device_key_thumbprint, ) return _offline_trial_service(request).submit(offline_request) @router.post("/offline/trial-requests/status") def offline_trial_request_status( body: OfflineTrialStatusRequest, request: Request, ) -> dict[str, object]: try: return _offline_trial_service(request).status(body.request_id, body.nonce) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, f"offline trial status request is invalid: {exc}", ) from exc @router.post("/me/licenses/claim-serial") def claim_redeemed_serial( body: SerialClaimRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _principal_from_credentials(request, credentials, required=True) assert principal is not None factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) license_id = _service(request, session).claim_redeemed_serial( body.serial, user.id, body.correlation_id, ) return {"licenseId": license_id, "status": "claimed"} @router.patch("/me/licenses/{license_id}/devices/{device_id}") def rename_account_device( license_id: str, device_id: str, body: DeviceRenameRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _principal_from_credentials(request, credentials, required=True) assert principal is not None factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _service(request, session).rename_account_device( license_id, device_id, body.friendly_name, user.id, body.idempotency_key, body.correlation_id, ) @router.post("/me/licenses/{license_id}/devices/{device_id}/release-unavailable") def release_unavailable_device( license_id: str, device_id: str, body: RemoteDeviceReleaseRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _principal_from_credentials(request, credentials, required=True) assert principal is not None if ( principal.authentication_time is None or datetime.now(timezone.utc) - principal.authentication_time > CUSTOMER_REAUTHENTICATION_MAX_AGE ): raise ServerLicensingError( ServerErrorCode.REAUTHENTICATION_REQUIRED, "recent reauthentication is required to release an unavailable device", status_code=403, ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) activation_id, remaining = _service( request, session, ).release_unavailable_device( license_id, device_id, user.id, body.correlation_id, reason=body.reason, ) return { "activationId": activation_id, "status": "deactivated", "remainingAutomaticReleases": remaining, } @router.post("/me/organizations") def create_organization( body: OrganizationCreateRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _principal_from_credentials(request, credentials, required=True) assert principal is not None factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _service(request, session).create_organization( body.display_name, user.id, body.idempotency_key, body.correlation_id, ) @router.get("/me/organizations") def list_organizations( request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _principal_from_credentials(request, credentials, required=True) assert principal is not None factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) organizations = _service(request, session).list_organizations(user.id) return {"organizations": list(organizations)} @router.post("/me/organizations/{organization_id}/invitations") def invite_organization_member( organization_id: str, body: OrganizationInvitationCreateRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _principal_from_credentials(request, credentials, required=True) assert principal is not None if ( principal.authentication_time is None or datetime.now(timezone.utc) - principal.authentication_time > CUSTOMER_REAUTHENTICATION_MAX_AGE ): raise ServerLicensingError( ServerErrorCode.REAUTHENTICATION_REQUIRED, "recent reauthentication is required to invite an organization member", status_code=403, ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _service(request, session).invite_organization_member( organization_id, body.verified_email, body.role, user.id, body.idempotency_key, body.correlation_id, ) @router.post("/me/organization-invitations/accept") def accept_organization_invitation( body: OrganizationInvitationAcceptRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _principal_from_credentials(request, credentials, required=True) assert principal is not None factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _service(request, session).accept_organization_invitation( body.invitation_token, user.id, body.idempotency_key, body.correlation_id, ) @router.post("/me/organizations/{organization_id}/licenses/{license_id}/attach") def attach_team_license( organization_id: str, license_id: str, body: OrganizationLicenseAttachRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _principal_from_credentials(request, credentials, required=True) assert principal is not None if ( principal.authentication_time is None or datetime.now(timezone.utc) - principal.authentication_time > CUSTOMER_REAUTHENTICATION_MAX_AGE ): raise ServerLicensingError( ServerErrorCode.REAUTHENTICATION_REQUIRED, "recent reauthentication is required to transfer a team license", status_code=403, ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _service(request, session).attach_team_license( organization_id, license_id, user.id, body.idempotency_key, body.correlation_id, ) @router.post("/me/organizations/{organization_id}/seats/{seat_id}/assign") def assign_organization_seat( organization_id: str, seat_id: str, body: OrganizationSeatAssignRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _principal_from_credentials(request, credentials, required=True) assert principal is not None if ( principal.authentication_time is None or datetime.now(timezone.utc) - principal.authentication_time > CUSTOMER_REAUTHENTICATION_MAX_AGE ): raise ServerLicensingError( ServerErrorCode.REAUTHENTICATION_REQUIRED, "recent reauthentication is required to assign a named seat", status_code=403, ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _service(request, session).assign_organization_seat( organization_id, seat_id, body.user_id, user.id, body.idempotency_key, body.correlation_id, ) @router.post("/me/organizations/{organization_id}/seats/{seat_id}/release") def release_organization_seat( organization_id: str, seat_id: str, body: OrganizationSeatReleaseRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _principal_from_credentials(request, credentials, required=True) assert principal is not None if ( principal.authentication_time is None or datetime.now(timezone.utc) - principal.authentication_time > CUSTOMER_REAUTHENTICATION_MAX_AGE ): raise ServerLicensingError( ServerErrorCode.REAUTHENTICATION_REQUIRED, "recent reauthentication is required to release a named seat", status_code=403, ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _service(request, session).release_organization_seat( organization_id, seat_id, user.id, body.idempotency_key, body.correlation_id, ) @router.post("/me/privacy/export") def export_account_privacy_data( body: PrivacyExportRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> JSONResponse: principal = _recent_customer_principal( request, credentials, "account privacy export", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) try: document = _privacy_service(request, session).export_account( user.id, body.correlation_id, ) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, f"account privacy export is invalid: {exc}", ) from exc return _privacy_response(document) @router.get("/me/privacy/deletion-requests") def list_account_deletion_requests( request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> JSONResponse: principal = _principal_from_credentials(request, credentials, required=True) assert principal is not None factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) document = _privacy_service(request, session).list_deletion_requests( user.id ) return _privacy_response(document) @router.post("/me/privacy/deletion-requests") def request_account_deletion( body: PrivacyDeletionRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> JSONResponse: principal = _recent_customer_principal( request, credentials, "account deletion request", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) try: document = _privacy_service(request, session).request_deletion( user.id, body.acknowledge_permanent_account_closure, body.idempotency_key, body.correlation_id, ) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, f"account deletion request is invalid: {exc}", ) from exc return _privacy_response(document) @router.post("/me/privacy/deletion-requests/{privacy_request_id}/cancel") def cancel_account_deletion( privacy_request_id: str, body: PrivacyDeletionCancelRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> JSONResponse: principal = _recent_customer_principal( request, credentials, "account deletion cancellation", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) try: document = _privacy_service( request, session, ).cancel_deletion_request( privacy_request_id, user.id, body.correlation_id, ) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, f"account deletion cancellation is invalid: {exc}", ) from exc return _privacy_response(document) @router.get("/admin/privacy/deletion-requests") def list_privacy_deletion_requests( request: Request, status: Optional[str] = "pending", limit: int = 50, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> JSONResponse: principal = _recent_admin_principal( request, credentials, ADMIN_PRIVACY_SCOPE, "privacy request inspection", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) try: document = _privacy_service( request, session, ).list_pending_for_administration( user.id, status=status, limit=limit, ) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, f"privacy request search is invalid: {exc}", ) from exc return _privacy_response(document) @router.post( "/admin/privacy/deletion-requests/{privacy_request_id}/preview" ) def preview_privacy_deletion( privacy_request_id: str, body: AdminPrivacyDeletionPreviewRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> JSONResponse: principal = _recent_admin_principal( request, credentials, ADMIN_PRIVACY_SCOPE, "account deletion preview", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) try: document = _privacy_service(request, session).preview_deletion( privacy_request_id, user.id, body.reason_code, body.correlation_id, note=body.note, ) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, f"account deletion preview is invalid: {exc}", ) from exc return _privacy_response(document) @router.post( "/admin/privacy/deletion-requests/{privacy_request_id}/execute" ) def execute_privacy_deletion( privacy_request_id: str, body: AdminPrivacyDeletionExecuteRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> JSONResponse: principal = _recent_admin_principal( request, credentials, ADMIN_PRIVACY_SCOPE, "account deletion execution", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) try: document = _privacy_service(request, session).execute_deletion( privacy_request_id, user.id, body.reason_code, body.expected_state_digest, body.idempotency_key, body.correlation_id, note=body.note, ) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, f"account deletion execution is invalid: {exc}", ) from exc return _privacy_response(document) @router.get("/admin/offline-requests") def list_offline_requests( request: Request, status: Optional[str] = None, limit: int = 50, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_OFFLINE_SCOPE, "offline request inspection", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) try: return _offline_trial_service(request).list_requests( user.id, status=status, limit=limit, ) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, f"offline request search is invalid: {exc}", ) from exc @router.get("/admin/offline-requests/{offline_request_id}") def inspect_offline_request( offline_request_id: str, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_OFFLINE_SCOPE, "offline request inspection", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) try: return _offline_trial_service(request).inspect(offline_request_id, user.id) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, f"offline request inspection is invalid: {exc}", ) from exc @router.post("/admin/offline-requests/{offline_request_id}/approve") def approve_offline_request( offline_request_id: str, body: AdminOfflineApprovalRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_OFFLINE_SCOPE, "offline trial approval", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) try: return _offline_trial_service(request).approve( offline_request_id, user.id, body.reason_code, body.expected_decision_generation, body.idempotency_key, body.correlation_id, note=body.note, ) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, f"offline trial approval is invalid: {exc}", ) from exc @router.post("/admin/offline-requests/{offline_request_id}/reject") def reject_offline_request( offline_request_id: str, body: AdminOfflineRejectionRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_OFFLINE_SCOPE, "offline trial rejection", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) try: return _offline_trial_service(request).reject( offline_request_id, user.id, body.reason_code, body.expected_decision_generation, body.idempotency_key, body.correlation_id, note=body.note, ) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, f"offline trial rejection is invalid: {exc}", ) from exc @router.get("/admin/licenses/{license_id}") def inspect_license( license_id: str, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_SUPPORT_SCOPE, "license inspection", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _service(request, session).inspect_license(license_id, user.id) @router.post("/admin/devices/{device_id}/recovery-preview") def preview_device_recovery( device_id: str, body: AdminDeviceRecoveryPreviewRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_SUPPORT_SCOPE, "lost device recovery preview", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _service(request, session).preview_device_recovery( device_id, user.id, body.reason_code, body.correlation_id, note=body.note, ) @router.post("/admin/devices/{device_id}/recover") def recover_device( device_id: str, body: AdminDeviceRecoveryExecuteRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_SUPPORT_SCOPE, "lost device recovery", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _service(request, session).recover_device( device_id, user.id, body.reason_code, body.expected_state_digest, body.idempotency_key, body.correlation_id, note=body.note, ) @router.post("/admin/audit-events/search") def search_audit_events( body: AdminAuditSearchRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_AUDIT_SCOPE, "audit search", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) try: return AuditAdministrationService(session).search( user.id, action=body.action, target_type=body.target_type, target_id=body.target_id, event_actor_id=body.actor_id, occurred_from=( None if body.occurred_from is None else parse_rfc3339(body.occurred_from, "occurredFrom") ), occurred_to=( None if body.occurred_to is None else parse_rfc3339(body.occurred_to, "occurredTo") ), cursor=body.cursor, limit=body.limit, ) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, f"audit search request is invalid: {exc}", ) from exc @router.post("/admin/audit-events/export") def export_audit_events( body: AdminAuditExportRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_AUDIT_SCOPE, "audit export", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) try: return AuditAdministrationService(session).export( user.id, body.correlation_id, action=body.action, target_type=body.target_type, target_id=body.target_id, event_actor_id=body.actor_id, occurred_from=( None if body.occurred_from is None else parse_rfc3339(body.occurred_from, "occurredFrom") ), occurred_to=( None if body.occurred_to is None else parse_rfc3339(body.occurred_to, "occurredTo") ), limit=body.limit, ) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, f"audit export request is invalid: {exc}", ) from exc @router.post("/admin/trials/{trial_id}/extension-preview") def preview_trial_extension( trial_id: str, body: AdminTrialExtensionPreviewRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_SUPPORT_SCOPE, "trial extension preview", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _service(request, session).preview_trial_extension( trial_id, user.id, body.reason_code, body.correlation_id, note=body.note, ) @router.post("/admin/trials/{trial_id}/extend") def extend_trial( trial_id: str, body: AdminTrialExtensionExecuteRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_SUPPORT_SCOPE, "trial extension", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _service(request, session).extend_trial( trial_id, user.id, body.reason_code, body.expected_revocation_generation, body.idempotency_key, body.correlation_id, note=body.note, ) @router.post( "/admin/licenses/{license_id}/grants/{grant_id}/revocation-preview" ) def preview_grant_revocation( license_id: str, grant_id: str, body: AdminGrantRevocationPreviewRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_SUPPORT_SCOPE, "grant revocation preview", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _service(request, session).preview_grant_revocation( license_id, grant_id, user.id, body.reason_code, body.correlation_id, note=body.note, ) @router.post("/admin/licenses/{license_id}/grants/{grant_id}/revoke") def revoke_grant( license_id: str, grant_id: str, body: AdminGrantRevocationExecuteRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_SUPPORT_SCOPE, "grant revocation", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _service(request, session).revoke_grant( license_id, grant_id, user.id, body.reason_code, body.expected_revocation_generation, body.idempotency_key, body.correlation_id, note=body.note, ) @router.get("/admin/catalog/releases") def catalog_release_status( request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_CATALOG_SCOPE, "catalog release status", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: _resolve_user(session, principal) return CatalogAdministrationService( session, request.app.state.catalog, ).status() @router.post("/admin/catalog/releases/stage") def stage_catalog_release( body: AdminCatalogStageRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_CATALOG_SCOPE, "catalog candidate staging", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return CatalogAdministrationService( session, request.app.state.catalog, ).stage_candidate( body.document, user.id, body.reason_code, body.idempotency_key, body.correlation_id, note=body.note, ) @router.post( "/admin/catalog/releases/{revision}/publication-preview" ) def preview_catalog_publication( revision: int, body: AdminCatalogPublicationPreviewRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_CATALOG_SCOPE, "catalog publication preview", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return CatalogAdministrationService( session, request.app.state.catalog, ).preview_publication( revision, user.id, body.reason_code, body.correlation_id, note=body.note, ) @router.post("/admin/catalog/releases/{revision}/publish") def publish_catalog_release( revision: int, body: AdminCatalogPublicationExecuteRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_CATALOG_SCOPE, "catalog publication", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return CatalogAdministrationService( session, request.app.state.catalog, ).publish_candidate( revision, user.id, body.reason_code, body.expected_state_digest, body.idempotency_key, body.correlation_id, note=body.note, ) @router.get("/admin/product-surfaces/inventory") def product_surface_inventory( request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_PRODUCT_SCOPE, "product surface inventory access", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _service(request, session).product_surface_inventory(user.id) @router.get("/admin/product-surface-profiles") def list_product_surface_profiles( request: Request, include_archived: bool = Query(default=False, alias="includeArchived"), credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_PRODUCT_SCOPE, "product surface profile access", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _service(request, session).list_surface_profiles( user.id, include_archived=include_archived, ) @router.post("/admin/product-surface-profiles") def create_product_surface_profile( body: ProductSurfaceProfileCreateRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_PRODUCT_SCOPE, "product surface profile creation", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _service(request, session).create_surface_profile( body.name, body.description, body.analysis_ids, body.main_tab_ids, user.id, body.reason, body.correlation_id, ) @router.post("/admin/product-surface-profiles/{profile_id}/archive") def archive_product_surface_profile( profile_id: str, body: ProductSurfaceProfileArchiveRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_PRODUCT_SCOPE, "product surface profile archival", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _service(request, session).archive_surface_profile( profile_id, user.id, body.reason, body.correlation_id, ) @router.post("/admin/licenses/{license_id}/surface-profile") def assign_license_surface_profile( license_id: str, body: LicenseSurfaceProfileAssignmentRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_PRODUCT_SCOPE, "license product surface assignment", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return _service(request, session).assign_license_surface_profile( license_id, body.surface_profile_id, user.id, body.reason, body.correlation_id, ) @router.post("/admin/serial-batches/{batch_id}/revocation-preview") def preview_serial_batch_revocation( batch_id: str, body: AdminSerialBatchRevocationPreviewRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_SERIAL_SCOPE, "serial batch revocation preview", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return SerialAdministrationService(session).preview_batch_revocation( batch_id, user.id, body.reason_code, body.correlation_id, note=body.note, ) @router.post("/admin/serial-batches/{batch_id}/revoke") def revoke_serial_batch( batch_id: str, body: AdminSerialBatchRevocationExecuteRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_SERIAL_SCOPE, "serial batch revocation", ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) return SerialAdministrationService(session).revoke_batch( batch_id, user.id, body.reason_code, body.expected_state_digest, body.idempotency_key, body.correlation_id, note=body.note, ) @router.post("/admin/serial-batches") def generate_serial_batch( body: SerialBatchRequest, request: Request, credentials: Optional[HTTPAuthorizationCredentials] = Depends(AUTHENTICATION_SCHEME), ) -> dict[str, object]: principal = _recent_admin_principal( request, credentials, ADMIN_SERIAL_SCOPE, "serial generation", ) deadline = ( None if body.redemption_deadline is None else parse_rfc3339(body.redemption_deadline, "redemptionDeadline") ) factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session: user = _resolve_user(session, principal) _require_published_runtime_catalog(request, session) result = _service(request, session).generate_serial_batch( body.sku_id, body.quantity, user.id, body.reason, body.correlation_id, campaign=body.campaign, redemption_deadline=deadline, surface_profile_id=body.surface_profile_id, trial_duration_hours=body.trial_duration_hours, ) return { "batchId": result.batch_id, "skuId": result.sku_id, "serials": list(result.serials), "surfaceProfileId": result.surface_profile_id, "trialDurationHours": result.trial_duration_hours, "warning": "Serials are shown once. Store and deliver them securely.", } @router.post("/webhooks/stripe") async def stripe_webhook(request: Request) -> dict[str, object]: settings: ServerSettings = request.app.state.settings if settings.stripe_webhook_secret is None: raise ServerLicensingError( ServerErrorCode.SUBSCRIPTION_INACTIVE, "Stripe subscription webhooks are not configured", status_code=503, ) signature = request.headers.get("stripe-signature") if not signature: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, "Stripe-Signature header is required", ) chunks = [] total = 0 async for chunk in request.stream(): total += len(chunk) if total > MAX_WEBHOOK_BODY_BYTES: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, "webhook request body is too large", status_code=413, ) chunks.append(chunk) raw_payload = b"".join(chunks) if not raw_payload: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, "webhook request body is empty", ) try: stripe_event = stripe.Webhook.construct_event( raw_payload, signature, settings.stripe_webhook_secret, ) except (ValueError, stripe.error.SignatureVerificationError) as exc: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "Stripe webhook signature or payload is invalid", status_code=400, ) from exc event = dict(stripe_event) event_type = str(event.get("type") or "") if event_type not in ( STRIPE_SUBSCRIPTION_EVENT_TYPES | STRIPE_BILLING_INCIDENT_EVENT_TYPES ): return {"received": True, "queued": False, "ignored": True} try: if event_type in STRIPE_SUBSCRIPTION_EVENT_TYPES: normalized = normalize_stripe_subscription_event( event, settings.stripe_price_sku_map, ) else: normalized = normalize_stripe_billing_incident_event(event) except (TypeError, ValueError) as exc: raise ServerLicensingError( ServerErrorCode.INVALID_REQUEST, f"Stripe billing event cannot be projected: {exc}", ) from exc factory: sessionmaker[Session] = request.app.state.session_factory with factory() as session, session.begin(): if event_type in STRIPE_SUBSCRIPTION_EVENT_TYPES: queued = enqueue_stripe_subscription_event( session, normalized, raw_payload, now=datetime.now(timezone.utc), ) else: queued = enqueue_stripe_billing_incident_event( session, normalized, raw_payload, now=datetime.now(timezone.utc), ) return { "received": True, "queued": queued, "ignored": False, } return router def create_app( settings: Optional[ServerSettings] = None, *, engine: Optional[Engine] = None, signer: Optional[SnapshotSigner] = None, token_validator: Optional[TokenValidator] = None, billing_provider: Optional[BillingProvider] = None, rate_limiter: Optional[DatabaseRateLimiter] = None, ) -> FastAPI: resolved_settings = ServerSettings.from_environment() if settings is None else settings if not isinstance(resolved_settings, ServerSettings): raise TypeError("settings must be ServerSettings or None") resolved_engine = create_database_engine(resolved_settings) if engine is None else engine if not isinstance(resolved_engine, Engine): raise TypeError("engine must be a SQLAlchemy Engine or None") if signer is None: if not ( resolved_settings.allow_development_signer or resolved_settings.allow_production_file_signer ): raise RuntimeError( "a signer adapter must be injected; local file signing is disabled" ) signer = Ed25519SnapshotSigner.from_private_key_file( resolved_settings.signing_key_id, resolved_settings.signing_private_key_path, require_owner_only=resolved_settings.allow_production_file_signer, ) resolved_validator = ( OIDCTokenValidator(resolved_settings) if token_validator is None else token_validator ) resolved_billing_provider = billing_provider if resolved_billing_provider is None and resolved_settings.stripe_commerce_enabled: assert resolved_settings.stripe_secret_key is not None resolved_billing_provider = StripeBillingProvider( resolved_settings.stripe_secret_key ) if resolved_billing_provider is not None and not all( callable(getattr(resolved_billing_provider, name, None)) for name in ("create_checkout_session", "create_portal_session") ): raise TypeError("billing_provider must expose Checkout and portal session creation") resolved_session_factory = create_session_factory(resolved_engine) resolved_rate_limiter = ( DatabaseRateLimiter( resolved_session_factory, resolved_settings.fingerprint_pepper, ) if rate_limiter is None else rate_limiter ) if not isinstance(resolved_rate_limiter, DatabaseRateLimiter): raise TypeError("rate_limiter must be a DatabaseRateLimiter or None") app = FastAPI(title=SERVER_NAME, version="1.0.0") app.add_middleware( RequestRateLimitContextMiddleware, trusted_proxy_networks=resolved_settings.trusted_proxy_ips, ) app.add_middleware( RequestBodyLimitMiddleware, maximum_bytes=MAX_API_REQUEST_BODY_BYTES, ) app.state.settings = resolved_settings app.state.engine = resolved_engine app.state.session_factory = resolved_session_factory app.state.signer = signer app.state.token_validator = resolved_validator app.state.billing_provider = resolved_billing_provider app.state.rate_limiter = resolved_rate_limiter app.state.catalog = load_builtin_catalog() @app.exception_handler(ServerLicensingError) async def server_error_handler( _request: Request, error: ServerLicensingError, ) -> JSONResponse: headers = ( None if error.retry_after_seconds is None else {"Retry-After": str(error.retry_after_seconds)} ) return JSONResponse( status_code=error.status_code, content=_error_mapping(error), headers=headers, ) @app.exception_handler(LicensingError) async def shared_error_handler( _request: Request, error: LicensingError, ) -> JSONResponse: return JSONResponse( status_code=400, content={ "code": error.code.value, "message": str(error), "retryable": False, "existingResourceId": None, }, ) app.add_api_route( OFFLINE_ACTIVATION_PATH, offline_activation_response, methods=["GET"], dependencies=[Depends(_enforce_request_rate_limits)], include_in_schema=False, ) app.include_router(create_api_router()) return app __all__ = ["create_api_router", "create_app"]