"""Signed-device FastAPI boundary tests.""" from __future__ import annotations import base64 import asyncio from dataclasses import replace from datetime import datetime, timedelta, timezone import hashlib import hmac import json from pathlib import Path from typing import Any import unittest from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey import fastapi.routing import fastapi.dependencies.utils import httpx2 from sqlalchemy import create_engine, func, select from sqlalchemy.orm import Session from sqlalchemy.pool import StaticPool from licensing_shared.device_proof import ( DeviceActionProof, device_action_payload_digest, device_proof_signing_bytes, ) from licensing_shared.canonical_json import canonicalize_json from licensing_shared.catalog import LicensingCatalog from licensing_shared.constants import TRIAL_MAXIMUM_TOTAL_DURATION_HOURS from licensing_shared.offline import device_public_key_thumbprint from licensing_shared.offline import ( DeviceSignedOfflineRequest, OfflineRequestPayload, OfflineRequestType, offline_request_signing_bytes, ) from licensing_server.app.api import create_app from licensing_server.app.auth import AuthenticatedPrincipal from licensing_server.app.billing import ProviderBillingSession from licensing_server.app.config import ServerSettings from licensing_server.app.constants import MAX_API_REQUEST_BODY_BYTES from licensing_server.app.database import Base from licensing_server.app.models import ( Activation, AuditEvent, CatalogRelease, Device, DeviceInstallation, Grant, IdempotencyRecord, License, OfflineCertificate, OfflineRequest, OrganizationInvitation, OrganizationSeat, OutboxEvent, Serial, SigningKeyMetadata, Subscription, Trial, User, WebhookEvent, ) from licensing_server.app.security import Ed25519SnapshotSigner from licensing_server.app.subscriptions import process_subscription_outbox TEST_SERIAL_PEPPER = b"serial-test-pepper-material-32-bytes-minimum" TEST_FINGERPRINT_PEPPER = b"fingerprint-test-pepper-material-32-bytes" TEST_STRIPE_WEBHOOK_SECRET = "whsec_test_licensing_subscription" TEST_STRIPE_PRICE_ID = "price_test_ai_assist" class FakeBillingProvider: def __init__(self) -> None: self.checkout_requests: list[dict[str, object]] = [] self.portal_requests: list[dict[str, object]] = [] def create_checkout_session(self, **kwargs: object) -> ProviderBillingSession: self.checkout_requests.append(dict(kwargs)) return ProviderBillingSession( session_id="cs_test_owned_checkout", url="https://checkout.stripe.test/session/owned", expires_at=datetime.now(timezone.utc) + timedelta(minutes=30), ) def create_portal_session(self, **kwargs: object) -> ProviderBillingSession: self.portal_requests.append(dict(kwargs)) return ProviderBillingSession( session_id="bps_test_owned_portal", url="https://billing.stripe.test/session/owned", ) class StaticTokenValidator: def validate(self, token: str) -> AuthenticatedPrincipal: now = datetime.now(timezone.utc) if token == "admin-token": return AuthenticatedPrincipal( issuer="https://identity.test", subject="admin-subject", scopes=frozenset( ( "licensing:serials:generate", "licensing:support:manage", "licensing:audit:read", "licensing:catalog:publish", "licensing:products:configure", "licensing:offline:approve", "licensing:privacy:manage", ) ), verified_email="admin@example.test", authentication_time=now, ) if token == "stale-admin-token": return AuthenticatedPrincipal( issuer="https://identity.test", subject="admin-subject", scopes=frozenset( ( "licensing:serials:generate", "licensing:support:manage", "licensing:offline:approve", ) ), verified_email="admin@example.test", authentication_time=now - timedelta(minutes=20), ) if token == "support-customer-token": return AuthenticatedPrincipal( issuer="https://identity.test", subject="customer-subject", scopes=frozenset(("licensing:support:manage",)), verified_email="customer@example.test", authentication_time=now, ) if token == "customer-token": return AuthenticatedPrincipal( issuer="https://identity.test", subject="customer-subject", scopes=frozenset(), verified_email="customer@example.test", authentication_time=now, ) if token == "stale-customer-token": return AuthenticatedPrincipal( issuer="https://identity.test", subject="customer-subject", scopes=frozenset(), verified_email="customer@example.test", authentication_time=now - timedelta(minutes=20), ) if token == "member-token": return AuthenticatedPrincipal( issuer="https://identity.test", subject="member-subject", scopes=frozenset(), verified_email="member@example.test", authentication_time=now, ) if token == "wrong-member-token": return AuthenticatedPrincipal( issuer="https://identity.test", subject="wrong-member-subject", scopes=frozenset(), verified_email="other@example.test", authentication_time=now, ) raise RuntimeError("unexpected test token") class LicensingApiTests(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: self.original_run_in_threadpool = fastapi.routing.run_in_threadpool self.original_dependency_run_in_threadpool = ( fastapi.dependencies.utils.run_in_threadpool ) async def direct_test_call(function, *args, **kwargs): return function(*args, **kwargs) # This WSL host drops worker-thread event-loop callbacks. The API keeps # normal FastAPI dispatch; only this in-process ASGI harness runs route # functions inline so request/response and domain behavior stay testable. fastapi.routing.run_in_threadpool = direct_test_call fastapi.dependencies.utils.run_in_threadpool = direct_test_call self.engine = create_engine( "sqlite+pysqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) Base.metadata.create_all(self.engine) self.signer = Ed25519SnapshotSigner( "license.test.api", Ed25519PrivateKey.generate(), ) settings = ServerSettings( database_url="sqlite+pysqlite:///:memory:", environment_name="test", signing_key_id=self.signer.key_id, signing_private_key_path=Path("unused-test-signing-key"), serial_pepper=TEST_SERIAL_PEPPER, fingerprint_pepper=TEST_FINGERPRINT_PEPPER, oidc_issuer="https://identity.test", oidc_audience="licensing-api", oidc_jwks_url="https://identity.test/.well-known/jwks.json", allow_development_signer=False, stripe_webhook_secret=TEST_STRIPE_WEBHOOK_SECRET, stripe_price_sku_map={ TEST_STRIPE_PRICE_ID: "subscription.ai_assist" }, oidc_authorization_endpoint="https://identity.test/authorize", oidc_token_endpoint="https://identity.test/oauth/token", oidc_native_client_id="apolon-desktop-test", ) self.app = create_app( settings, engine=self.engine, signer=self.signer, token_validator=StaticTokenValidator(), ) with Session(self.engine) as session, session.begin(): now = datetime.now(timezone.utc) catalog = self.app.state.catalog session.add_all( ( User( id="user.test.admin", external_issuer="https://identity.test", external_subject="admin-subject", verified_email="admin@example.test", status="active", is_server_admin=True, ), CatalogRelease( id="catalog_release.test.api", product_id=catalog.product_id, revision=catalog.revision, document_json=catalog.to_mapping(), document_digest=bytes.fromhex(catalog.sha256()), status="active", staged_by_user_id="user.test.admin", staged_at=now - timedelta(days=1), published_by_user_id="user.test.admin", published_at=now - timedelta(days=1), retired_at=None, ), SigningKeyMetadata( id="signing_key.test.api", key_id=self.signer.key_id, purpose="license_snapshot", public_key_bytes=self.signer.public_key_bytes(), status="active", not_before=now - timedelta(days=1), expires_at=None, activated_at=now - timedelta(days=1), retired_at=None, ), ) ) self.device_private_key = ec.generate_private_key(ec.SECP256R1()) self.device_public_key_der = self.device_private_key.public_key().public_bytes( encoding=serialization.Encoding.DER, format=serialization.PublicFormat.SubjectPublicKeyInfo, ) self.device_public_key_base64 = base64.b64encode( self.device_public_key_der ).decode("ascii") self.device_thumbprint = device_public_key_thumbprint(self.device_public_key_der) def tearDown(self) -> None: fastapi.routing.run_in_threadpool = self.original_run_in_threadpool fastapi.dependencies.utils.run_in_threadpool = ( self.original_dependency_run_in_threadpool ) self.engine.dispose() async def _request(self, method: str, url: str, **kwargs: Any) -> httpx2.Response: transport = httpx2.ASGITransport(app=self.app) async with httpx2.AsyncClient( transport=transport, base_url="https://licensing.test", ) as client: return await asyncio.wait_for( client.request(method, url, **kwargs), timeout=5, ) def _device(self) -> dict[str, Any]: return { "installationId": "installation.test.api", "devicePublicKey": self.device_public_key_base64, "deviceKeyThumbprint": self.device_thumbprint, "keyProvider": "software.test", "friendlyName": "API test computer", "evidence": { "system_uuid": "api-device-uuid", "baseboard_serial": "api-device-board", }, } def _proof( self, action: str, payload: dict[str, Any], *, private_key: ec.EllipticCurvePrivateKey | None = None, public_key_base64: str | None = None, key_thumbprint: str | None = None, ) -> dict[str, Any]: custom_key_values = (private_key, public_key_base64, key_thumbprint) if any(value is not None for value in custom_key_values) and not all( value is not None for value in custom_key_values ): raise ValueError("a complete replacement device key is required") effective_private_key = private_key or self.device_private_key effective_public_key_base64 = ( public_key_base64 or self.device_public_key_base64 ) effective_key_thumbprint = key_thumbprint or self.device_thumbprint now = datetime.now(timezone.utc) placeholder = DeviceActionProof( action=action, installation_id=payload.get("installationId") or payload["device"]["installationId"], device_public_key_base64=effective_public_key_base64, device_key_thumbprint=effective_key_thumbprint, payload_digest=device_action_payload_digest(payload), created_at=now, expires_at=now + timedelta(minutes=5), nonce="api-device-proof-nonce", signature_base64=base64.b64encode(b"placeholder").decode("ascii"), ) signature = effective_private_key.sign( device_proof_signing_bytes(placeholder), ec.ECDSA(hashes.SHA256()), ) result = placeholder.to_mapping() result["signature"] = base64.b64encode(signature).decode("ascii") return result async def _admin_serial( self, sku_id: str = "edition.professional", *, surface_profile_id: str | None = None, trial_duration_hours: int | None = None, ) -> str: body: dict[str, object] = { "skuId": sku_id, "quantity": 1, "reason": "API test serial", "correlationId": "correlation.test.api_generation", } if surface_profile_id is not None: body["surfaceProfileId"] = surface_profile_id if trial_duration_hours is not None: body["trialDurationHours"] = trial_duration_hours response = await self._request( "POST", "/v1/admin/serial-batches", headers={"Authorization": "Bearer admin-token"}, json=body, ) self.assertEqual(response.status_code, 200, response.text) return response.json()["serials"][0] def _offline_trial_request(self, suffix: str) -> DeviceSignedOfflineRequest: if not isinstance(suffix, str) or not suffix.strip(): raise ValueError("suffix must be a non-empty string") normalized_suffix = suffix.strip() now = datetime.now(timezone.utc) payload = OfflineRequestPayload( request_type=OfflineRequestType.TRIAL, request_id=f"offline_trial.test.{normalized_suffix}", product_id="apolon.precision_lab_analyzer", application_major_version=1, installation_id=f"installation.test.offline_trial.{normalized_suffix}", device_id=None, device_public_key_base64=self.device_public_key_base64, device_key_thumbprint=self.device_thumbprint, evidence_digests=( ("system_uuid", hashlib.sha256(suffix.encode()).hexdigest()), ), key_provider="software.test", friendly_name=f"Offline trial {normalized_suffix}", requested_sku_id=None, created_at=now, expires_at=now + timedelta(days=1), nonce=f"offline-trial-nonce-{normalized_suffix}", ) signature = self.device_private_key.sign( offline_request_signing_bytes(payload), ec.ECDSA(hashes.SHA256()), ) return DeviceSignedOfflineRequest( payload=payload, signature_base64=base64.b64encode(signature).decode("ascii"), ) async def test_health_catalog_and_admin_authorization(self) -> None: health = await self._request("GET", "/v1/health") self.assertEqual(health.status_code, 200) readiness = await self._request("GET", "/v1/readiness") self.assertEqual(readiness.status_code, 200) self.assertEqual(readiness.json()["signingKeyId"], self.signer.key_id) catalog = await self._request("GET", "/v1/catalog/public") self.assertEqual(catalog.status_code, 200) self.assertIn("edition.professional", catalog.json()["skus"]) denied = await self._request( "POST", "/v1/admin/serial-batches", headers={"Authorization": "Bearer customer-token"}, json={ "skuId": "edition.personal", "quantity": 1, "reason": "Must be denied", "correlationId": "correlation.test.api_denied", }, ) self.assertEqual(denied.status_code, 403) self.assertEqual(denied.json()["code"], "authorization_denied") async def test_admin_surface_profiles_and_bounded_trial_serial_api(self) -> None: denied = await self._request( "GET", "/v1/admin/product-surfaces/inventory", headers={"Authorization": "Bearer customer-token"}, ) self.assertEqual(denied.status_code, 403, denied.text) inventory = await self._request( "GET", "/v1/admin/product-surfaces/inventory", headers={"Authorization": "Bearer admin-token"}, ) self.assertEqual(inventory.status_code, 200, inventory.text) surfaces = inventory.json()["surfaces"] self.assertGreater(len(surfaces), 90) self.assertRegex(inventory.json()["sha256"], r"^[0-9a-f]{64}$") created = await self._request( "POST", "/v1/admin/product-surface-profiles", headers={"Authorization": "Bearer admin-token"}, json={ "name": "API focused profile", "description": "Acquisition and spectrum only", "analysisIds": ["fft_spectrum", "welch_psd"], "mainTabIds": ["acquisition", "analysis"], "reason": "API profile regression", "correlationId": "correlation.test.api_surface_create", }, ) self.assertEqual(created.status_code, 200, created.text) profile_id = created.json()["profileId"] self.assertEqual( created.json()["policy"]["analysisIds"], ["fft_spectrum", "welch_psd"], ) listed = await self._request( "GET", "/v1/admin/product-surface-profiles", headers={"Authorization": "Bearer admin-token"}, ) self.assertEqual(listed.status_code, 200, listed.text) self.assertEqual( [value["profileId"] for value in listed.json()["profiles"]], [profile_id], ) trial = await self._request( "POST", "/v1/admin/serial-batches", headers={"Authorization": "Bearer admin-token"}, json={ "skuId": "trial.full", "quantity": 1, "reason": "One-hour API trial", "correlationId": "correlation.test.api_trial_serial", "surfaceProfileId": profile_id, "trialDurationHours": 1, }, ) self.assertEqual(trial.status_code, 200, trial.text) self.assertEqual(trial.json()["surfaceProfileId"], profile_id) self.assertEqual(trial.json()["trialDurationHours"], 1) self.assertEqual(len(trial.json()["serials"]), 1) overlong = await self._request( "POST", "/v1/admin/serial-batches", headers={"Authorization": "Bearer admin-token"}, json={ "skuId": "trial.full", "quantity": 1, "reason": "Overlong API trial", "correlationId": "correlation.test.api_trial_overlong", "trialDurationHours": TRIAL_MAXIMUM_TOTAL_DURATION_HOURS + 1, }, ) self.assertEqual(overlong.status_code, 422, overlong.text) archived = await self._request( "POST", f"/v1/admin/product-surface-profiles/{profile_id}/archive", headers={"Authorization": "Bearer admin-token"}, json={ "reason": "API profile archive regression", "correlationId": "correlation.test.api_surface_archive", }, ) self.assertEqual(archived.status_code, 200, archived.text) self.assertEqual(archived.json()["status"], "archived") async def test_readiness_fails_closed_without_active_registered_signer(self) -> None: with Session(self.engine) as session, session.begin(): session.query(SigningKeyMetadata).delete() response = await self._request("GET", "/v1/readiness") self.assertEqual(response.status_code, 503) self.assertEqual(response.json()["code"], "signing_unavailable") self.assertNotIn("publicKey", response.text) trial_unsigned = { "applicationMajorVersion": 1, "idempotencyKey": "idempotency-api-unregistered-signer", "correlationId": "correlation.test.api_unregistered_signer", "device": self._device(), } trial_body = dict(trial_unsigned) trial_body["deviceProof"] = self._proof("trial.start", trial_unsigned) signing_attempt = await self._request( "POST", "/v1/trials/start", headers={"Authorization": "Bearer customer-token"}, json=trial_body, ) self.assertEqual(signing_attempt.status_code, 503, signing_attempt.text) self.assertEqual(signing_attempt.json()["code"], "signing_unavailable") with Session(self.engine) as session: self.assertEqual( session.scalar(select(func.count()).select_from(License)), 0, ) self.assertEqual( session.scalar(select(func.count()).select_from(Trial)), 0, ) self.assertEqual( session.scalar(select(func.count()).select_from(Activation)), 0, ) async def test_catalog_publication_is_reviewed_and_runtime_signing_is_atomic( self, ) -> None: with Session(self.engine) as session, session.begin(): session.query(CatalogRelease).delete() readiness = await self._request("GET", "/v1/readiness") self.assertEqual(readiness.status_code, 503) self.assertEqual(readiness.json()["code"], "catalog_unavailable") unpublished_catalog = await self._request("GET", "/v1/catalog/public") self.assertEqual(unpublished_catalog.status_code, 503) trial_unsigned = { "applicationMajorVersion": 1, "idempotencyKey": "idempotency-api-unpublished-catalog", "correlationId": "correlation.test.api_unpublished_catalog", "device": self._device(), } trial_body = dict(trial_unsigned) trial_body["deviceProof"] = self._proof("trial.start", trial_unsigned) signing_attempt = await self._request( "POST", "/v1/trials/start", headers={"Authorization": "Bearer customer-token"}, json=trial_body, ) self.assertEqual(signing_attempt.status_code, 503, signing_attempt.text) self.assertEqual(signing_attempt.json()["code"], "catalog_unavailable") serial_attempt = await self._request( "POST", "/v1/admin/serial-batches", headers={"Authorization": "Bearer admin-token"}, json={ "skuId": "edition.personal", "quantity": 1, "reason": "Must fail before catalog publication", "correlationId": "correlation.test.api_unpublished_serial", }, ) self.assertEqual(serial_attempt.status_code, 503, serial_attempt.text) self.assertEqual(serial_attempt.json()["code"], "catalog_unavailable") with Session(self.engine) as session: self.assertEqual( session.scalar(select(func.count()).select_from(License)), 0, ) self.assertEqual( session.scalar(select(func.count()).select_from(Trial)), 0, ) self.assertEqual( session.scalar(select(func.count()).select_from(Activation)), 0, ) self.assertEqual( session.scalar(select(func.count()).select_from(Serial)), 0, ) document = self.app.state.catalog.to_mapping() denied = await self._request( "POST", "/v1/admin/catalog/releases/stage", headers={"Authorization": "Bearer customer-token"}, json={ "document": document, "reasonCode": "initial_provisioning", "idempotencyKey": "idempotency-catalog-stage-denied", "correlationId": "correlation.test.catalog_stage_denied", }, ) self.assertEqual(denied.status_code, 403) staged = await self._request( "POST", "/v1/admin/catalog/releases/stage", headers={"Authorization": "Bearer admin-token"}, json={ "document": document, "reasonCode": "initial_provisioning", "idempotencyKey": "idempotency-catalog-stage-api", "correlationId": "correlation.test.catalog_stage_api", }, ) self.assertEqual(staged.status_code, 200, staged.text) self.assertTrue(staged.json()["registered"]) revision = self.app.state.catalog.revision preview = await self._request( "POST", f"/v1/admin/catalog/releases/{revision}/publication-preview", headers={"Authorization": "Bearer admin-token"}, json={ "reasonCode": "initial_provisioning", "correlationId": "correlation.test.catalog_publish_api", }, ) self.assertEqual(preview.status_code, 200, preview.text) self.assertTrue(preview.json()["canPublish"]) published = await self._request( "POST", f"/v1/admin/catalog/releases/{revision}/publish", headers={"Authorization": "Bearer admin-token"}, json={ "reasonCode": "initial_provisioning", "expectedStateDigest": preview.json()["stateDigest"], "idempotencyKey": "idempotency-catalog-publish-api", "correlationId": "correlation.test.catalog_publish_api", }, ) self.assertEqual(published.status_code, 200, published.text) self.assertTrue(published.json()["catalogReady"]) ready = await self._request("GET", "/v1/readiness") self.assertEqual(ready.status_code, 200, ready.text) self.assertEqual(ready.json()["catalogSha256"], self.app.state.catalog.sha256()) async def test_serial_batch_revocation_requires_review_and_preserves_no_plaintext( self, ) -> None: generated = await self._request( "POST", "/v1/admin/serial-batches", headers={"Authorization": "Bearer admin-token"}, json={ "skuId": "edition.professional", "quantity": 2, "reason": "API revocation test", "correlationId": "correlation.test.api_batch_generation", }, ) self.assertEqual(generated.status_code, 200, generated.text) batch_id = generated.json()["batchId"] serials = generated.json()["serials"] preview_body = { "reasonCode": "serial_exposure", "note": "Undelivered export was exposed", "correlationId": "correlation.test.api_batch_preview", } denied = await self._request( "POST", f"/v1/admin/serial-batches/{batch_id}/revocation-preview", headers={"Authorization": "Bearer customer-token"}, json=preview_body, ) self.assertEqual(denied.status_code, 403, denied.text) stale = await self._request( "POST", f"/v1/admin/serial-batches/{batch_id}/revocation-preview", headers={"Authorization": "Bearer stale-admin-token"}, json=preview_body, ) self.assertEqual(stale.status_code, 403, stale.text) self.assertEqual(stale.json()["code"], "reauthentication_required") preview = await self._request( "POST", f"/v1/admin/serial-batches/{batch_id}/revocation-preview", headers={"Authorization": "Bearer admin-token"}, json=preview_body, ) self.assertEqual(preview.status_code, 200, preview.text) self.assertEqual(preview.json()["wouldRevokeSerialCount"], 2) state_digest = preview.json()["stateDigest"] execute_body = { **preview_body, "expectedStateDigest": "0" * 64, "idempotencyKey": "idempotency-api-batch-revocation", "correlationId": "correlation.test.api_batch_execute", } wrong_state = await self._request( "POST", f"/v1/admin/serial-batches/{batch_id}/revoke", headers={"Authorization": "Bearer admin-token"}, json=execute_body, ) self.assertEqual(wrong_state.status_code, 409, wrong_state.text) self.assertEqual(wrong_state.json()["code"], "conflict") execute_body["expectedStateDigest"] = state_digest executed = await self._request( "POST", f"/v1/admin/serial-batches/{batch_id}/revoke", headers={"Authorization": "Bearer admin-token"}, json=execute_body, ) self.assertEqual(executed.status_code, 200, executed.text) self.assertEqual(executed.json()["newlyRevokedSerialCount"], 2) self.assertFalse(executed.json()["idempotentReplay"]) replay = await self._request( "POST", f"/v1/admin/serial-batches/{batch_id}/revoke", headers={"Authorization": "Bearer admin-token"}, json=execute_body, ) self.assertEqual(replay.status_code, 200, replay.text) self.assertTrue(replay.json()["idempotentReplay"]) redemption_body = { "serial": serials[0], "applicationMajorVersion": 1, "idempotencyKey": "idempotency-api-revoked-serial", "correlationId": "correlation.test.api_revoked_serial", "targetLicenseId": None, "device": self._device(), } redemption_body["deviceProof"] = self._proof( "serial.redeem", redemption_body, ) redemption = await self._request( "POST", "/v1/serials/redeem", json=redemption_body, ) self.assertEqual(redemption.status_code, 409, redemption.text) self.assertEqual(redemption.json()["code"], "serial_invalid") audited = await self._request( "POST", "/v1/admin/audit-events/search", headers={"Authorization": "Bearer admin-token"}, json={"action": "serial.batch_revoked", "limit": 10}, ) self.assertEqual(audited.status_code, 200, audited.text) self.assertEqual(len(audited.json()["events"]), 1) self.assertEqual( set(audited.json()["events"][0]["metadata"]), { "activeSerialCount", "reasonCode", "redeemedLicenseCount", "redeemedSerialCount", "revokedSerialCount", "skuId", "stateDigest", "wouldRevokeSerialCount", }, ) with Session(self.engine) as session: event = session.scalar( select(AuditEvent).where(AuditEvent.action == "serial.batch_revoked") ) self.assertIsNotNone(event) assert event is not None self.assertNotIn(serials[0], json.dumps(event.metadata_json)) async def test_offline_activation_page_is_self_contained_and_hardened(self) -> None: response = await self._request("GET", "/offline-activation") self.assertEqual(response.status_code, 200, response.text) self.assertIn("/v1/offline/requests", response.text) self.assertIn("/v1/offline/trial-requests", response.text) self.assertIn("/v1/activations/release-file", response.text) self.assertNotIn("https://", response.text) policy = response.headers["content-security-policy"] self.assertIn("frame-ancestors 'none'", policy) self.assertIn("script-src 'nonce-", policy) self.assertNotIn("unsafe-inline", policy) self.assertEqual(response.headers["cache-control"], "no-store") async def test_request_body_limit_rejects_oversized_payload(self) -> None: response = await self._request( "POST", "/v1/trials/start", content=b"x" * (MAX_API_REQUEST_BODY_BYTES + 1), headers={"content-type": "application/json"}, ) self.assertEqual(response.status_code, 413, response.text) self.assertEqual(response.json()["code"], "invalid_request") async def test_native_authorization_configuration_is_public_and_fail_closed(self) -> None: configured = await self._request("GET", "/v1/auth/native/config") self.assertEqual(configured.status_code, 200, configured.text) self.assertEqual( configured.json(), { "authorizationEndpoint": "https://identity.test/authorize", "tokenEndpoint": "https://identity.test/oauth/token", "clientId": "apolon-desktop-test", "scopes": ["openid", "profile", "email", "offline_access"], "audience": "licensing-api", }, ) unavailable_app = create_app( replace( self.app.state.settings, oidc_authorization_endpoint=None, oidc_token_endpoint=None, oidc_native_client_id=None, ), engine=self.engine, signer=self.signer, token_validator=StaticTokenValidator(), ) transport = httpx2.ASGITransport(app=unavailable_app) async with httpx2.AsyncClient( transport=transport, base_url="https://licensing.test", ) as client: unavailable = await client.get("/v1/auth/native/config") self.assertEqual(unavailable.status_code, 503, unavailable.text) self.assertEqual(unavailable.json()["code"], "authentication_required") async def test_serial_redemption_refresh_deactivation_and_account_listing(self) -> None: serial_value = await self._admin_serial() unsigned = { "serial": serial_value, "applicationMajorVersion": 1, "idempotencyKey": "idempotency-api-redemption", "correlationId": "correlation.test.api_redemption", "targetLicenseId": None, "device": self._device(), } request_body = dict(unsigned) request_body["deviceProof"] = self._proof("serial.redeem", unsigned) response = await self._request( "POST", "/v1/serials/redeem", headers={"Authorization": "Bearer customer-token"}, json=request_body, ) self.assertEqual(response.status_code, 200, response.text) redeemed = response.json() self.assertFalse(redeemed["idempotentReplay"]) self.assertIn("licenseDocument", redeemed) listed = await self._request( "GET", "/v1/me/licenses", headers={"Authorization": "Bearer customer-token"}, ) self.assertEqual(listed.status_code, 200, listed.text) self.assertEqual(listed.json()["licenses"][0]["licenseId"], redeemed["licenseId"]) refresh_unsigned = { "licenseId": redeemed["licenseId"], "installationId": self._device()["installationId"], "applicationMajorVersion": 1, "correlationId": "correlation.test.api_refresh", } refresh_body = dict(refresh_unsigned) refresh_body["deviceProof"] = self._proof( "activation.refresh", refresh_unsigned, ) refreshed = await self._request( "POST", "/v1/activations/refresh", json=refresh_body, ) self.assertEqual(refreshed.status_code, 200, refreshed.text) self.assertNotEqual( refreshed.json()["licenseDocument"]["payload"]["snapshotId"], redeemed["licenseDocument"]["payload"]["snapshotId"], ) deactivate_unsigned = { "licenseId": redeemed["licenseId"], "installationId": self._device()["installationId"], "correlationId": "correlation.test.api_deactivate", "reason": "local_device_release", } deactivate_body = dict(deactivate_unsigned) deactivate_body["deviceProof"] = self._proof( "activation.deactivate", deactivate_unsigned, ) deactivated = await self._request( "POST", "/v1/activations/deactivate", json=deactivate_body, ) self.assertEqual(deactivated.status_code, 200, deactivated.text) self.assertEqual(deactivated.json()["status"], "deactivated") activation_proof_payload = { "licenseId": redeemed["licenseId"], "applicationMajorVersion": 1, "idempotencyKey": "idempotency-api-account-activation", "correlationId": "correlation.test.api_account_activation", "device": self._device(), } activation_body = { key: value for key, value in activation_proof_payload.items() if key != "licenseId" } activation_body["deviceProof"] = self._proof( "account.license.activate", activation_proof_payload, ) activated = await self._request( "POST", f"/v1/me/licenses/{redeemed['licenseId']}/activate", headers={"Authorization": "Bearer customer-token"}, json=activation_body, ) self.assertEqual(activated.status_code, 200, activated.text) self.assertFalse(activated.json()["idempotentReplay"]) replayed = await self._request( "POST", f"/v1/me/licenses/{redeemed['licenseId']}/activate", headers={"Authorization": "Bearer customer-token"}, json=activation_body, ) self.assertEqual(replayed.status_code, 200, replayed.text) self.assertTrue(replayed.json()["idempotentReplay"]) self.assertEqual( replayed.json()["licenseDocument"], activated.json()["licenseDocument"], ) denied = await self._request( "POST", f"/v1/me/licenses/{redeemed['licenseId']}/activate", headers={"Authorization": "Bearer admin-token"}, json=activation_body, ) self.assertEqual(denied.status_code, 403, denied.text) self.assertEqual(denied.json()["code"], "authorization_denied") async def test_payload_tampering_is_rejected_before_serial_consumption(self) -> None: serial_value = await self._admin_serial("edition.personal") unsigned = { "serial": serial_value, "applicationMajorVersion": 1, "idempotencyKey": "idempotency-api-tamper", "correlationId": "correlation.test.api_tamper", "targetLicenseId": None, "device": self._device(), } request_body = dict(unsigned) request_body["deviceProof"] = self._proof("serial.redeem", unsigned) request_body["applicationMajorVersion"] = 2 response = await self._request( "POST", "/v1/serials/redeem", json=request_body, ) self.assertEqual(response.status_code, 400, response.text) self.assertEqual(response.json()["code"], "invalid_signature") with Session(self.engine) as session: active = session.scalar( select(func.count()).select_from(Serial).where(Serial.status == "active") ) self.assertEqual(active, 1) async def test_unsigned_trial_request_is_rejected_by_schema(self) -> None: response = await self._request( "POST", "/v1/trials/start", json={ "applicationMajorVersion": 1, "idempotencyKey": "idempotency-api-unsigned", "correlationId": "correlation.test.api_unsigned", "device": self._device(), }, ) self.assertEqual(response.status_code, 422) async def test_offline_serial_activation_binds_request_nonce_and_expiry(self) -> None: serial_value = await self._admin_serial("edition.personal") now = datetime.now(timezone.utc) payload = OfflineRequestPayload( request_type=OfflineRequestType.ACTIVATE, request_id="request.test.api_offline", product_id="apolon.precision_lab_analyzer", application_major_version=1, installation_id=self._device()["installationId"], device_id=None, device_public_key_base64=self.device_public_key_base64, device_key_thumbprint=self.device_thumbprint, evidence_digests=(("system_uuid", "ab" * 32),), key_provider="software.test", friendly_name="Offline API test computer", requested_sku_id="edition.personal", created_at=now, expires_at=now + timedelta(days=1), nonce="offline-api-request-nonce", ) signature = self.device_private_key.sign( offline_request_signing_bytes(payload), ec.ECDSA(hashes.SHA256()), ) offline_request = DeviceSignedOfflineRequest( payload=payload, signature_base64=base64.b64encode(signature).decode("ascii"), ) target_mismatch = await self._request( "POST", "/v1/offline/requests", json={ "offlineRequest": offline_request.to_mapping(), "serial": serial_value, "idempotencyKey": "idempotency-api-offline-mismatch", "correlationId": "correlation.test.api_offline_mismatch", "targetLicenseId": "license.tampered.target", }, ) self.assertEqual(target_mismatch.status_code, 403, target_mismatch.text) self.assertEqual(target_mismatch.json()["code"], "authorization_denied") response = await self._request( "POST", "/v1/offline/requests", json={ "offlineRequest": offline_request.to_mapping(), "serial": serial_value, "idempotencyKey": "idempotency-api-offline", "correlationId": "correlation.test.api_offline", }, ) self.assertEqual(response.status_code, 200, response.text) result = response.json() snapshot = result["licenseDocument"]["payload"] self.assertEqual(result["offlineRequestId"], payload.request_id) self.assertEqual(snapshot["nonce"], payload.nonce) self.assertIsNotNone(snapshot["offlineExpiresAt"]) self.assertEqual(snapshot["leaseExpiresAt"], snapshot["offlineExpiresAt"]) async def test_offline_trial_approval_is_scoped_replay_safe_and_retrievable( self, ) -> None: offline_request = self._offline_trial_request("approved") submitted = await self._request( "POST", "/v1/offline/trial-requests", json={"offlineRequest": offline_request.to_mapping()}, ) self.assertEqual(submitted.status_code, 200, submitted.text) self.assertEqual(submitted.json()["status"], "pending") self.assertNotIn("licenseDocument", submitted.json()) altered_payload = replace( offline_request.payload, friendly_name="Different signed request", ) altered_signature = self.device_private_key.sign( offline_request_signing_bytes(altered_payload), ec.ECDSA(hashes.SHA256()), ) conflicting = await self._request( "POST", "/v1/offline/trial-requests", json={ "offlineRequest": DeviceSignedOfflineRequest( payload=altered_payload, signature_base64=base64.b64encode(altered_signature).decode("ascii"), ).to_mapping() }, ) self.assertEqual(conflicting.status_code, 409, conflicting.text) wrong_proof = await self._request( "POST", "/v1/offline/trial-requests/status", json={ "requestId": offline_request.payload.request_id, "nonce": "wrong-offline-trial-proof", }, ) self.assertEqual(wrong_proof.status_code, 403, wrong_proof.text) denied = await self._request( "GET", "/v1/admin/offline-requests?status=pending", headers={"Authorization": "Bearer support-customer-token"}, ) self.assertEqual(denied.status_code, 403, denied.text) stale = await self._request( "GET", "/v1/admin/offline-requests?status=pending", headers={"Authorization": "Bearer stale-admin-token"}, ) self.assertEqual(stale.status_code, 403, stale.text) self.assertEqual(stale.json()["code"], "reauthentication_required") listed = await self._request( "GET", "/v1/admin/offline-requests?status=pending", headers={"Authorization": "Bearer admin-token"}, ) self.assertEqual(listed.status_code, 200, listed.text) request_summary = listed.json()["requests"][0] self.assertEqual(request_summary["requestId"], offline_request.payload.request_id) self.assertEqual(request_summary["evidenceComponents"], ["system_uuid"]) self.assertNotIn("evidenceDigests", request_summary) decision = { "reasonCode": "evaluation_request", "note": "Verified laboratory evaluation request", "expectedDecisionGeneration": 0, "idempotencyKey": "idempotency-offline-trial-approval", "correlationId": "correlation.test.offline_trial_approval", } approved = await self._request( "POST", f"/v1/admin/offline-requests/{request_summary['id']}/approve", headers={"Authorization": "Bearer admin-token"}, json=decision, ) self.assertEqual(approved.status_code, 200, approved.text) approved_mapping = approved.json() self.assertEqual(approved_mapping["status"], "approved") self.assertEqual( approved_mapping["offlineRequestNonce"], offline_request.payload.nonce, ) snapshot = approved_mapping["licenseDocument"]["payload"] self.assertEqual(snapshot["nonce"], offline_request.payload.nonce) self.assertEqual(snapshot["leaseExpiresAt"], snapshot["offlineExpiresAt"]) replayed = await self._request( "POST", f"/v1/admin/offline-requests/{request_summary['id']}/approve", headers={"Authorization": "Bearer admin-token"}, json=decision, ) self.assertEqual(replayed.status_code, 200, replayed.text) self.assertEqual( replayed.json()["licenseDocument"], approved_mapping["licenseDocument"], ) retrieved = await self._request( "POST", "/v1/offline/trial-requests/status", json={ "requestId": offline_request.payload.request_id, "nonce": offline_request.payload.nonce, }, ) self.assertEqual(retrieved.status_code, 200, retrieved.text) self.assertEqual(retrieved.json()["status"], "approved") self.assertEqual( retrieved.json()["licenseDocument"], approved_mapping["licenseDocument"], ) rejected_request = self._offline_trial_request("rejected") rejected_submission = await self._request( "POST", "/v1/offline/trial-requests", json={"offlineRequest": rejected_request.to_mapping()}, ) self.assertEqual(rejected_submission.status_code, 200, rejected_submission.text) rejected_list = await self._request( "GET", "/v1/admin/offline-requests?status=pending", headers={"Authorization": "Bearer admin-token"}, ) rejected_summary = next( value for value in rejected_list.json()["requests"] if value["requestId"] == rejected_request.payload.request_id ) rejected = await self._request( "POST", f"/v1/admin/offline-requests/{rejected_summary['id']}/reject", headers={"Authorization": "Bearer admin-token"}, json={ "reasonCode": "trial_policy_ineligible", "expectedDecisionGeneration": 0, "idempotencyKey": "idempotency-offline-trial-rejection", "correlationId": "correlation.test.offline_trial_rejection", }, ) self.assertEqual(rejected.status_code, 200, rejected.text) self.assertEqual(rejected.json()["status"], "rejected") self.assertNotIn("licenseDocument", rejected.json()) cannot_approve = await self._request( "POST", f"/v1/admin/offline-requests/{rejected_summary['id']}/approve", headers={"Authorization": "Bearer admin-token"}, json={ "reasonCode": "evaluation_request", "expectedDecisionGeneration": 1, "idempotencyKey": "idempotency-rejected-trial-approval", "correlationId": "correlation.test.rejected_trial_approval", }, ) self.assertEqual(cannot_approve.status_code, 409, cannot_approve.text) with Session(self.engine) as session: self.assertEqual( session.scalar(select(func.count()).select_from(OfflineRequest)), 2, ) self.assertEqual( session.scalar(select(func.count()).select_from(OfflineCertificate)), 1, ) async def test_account_offline_renewal_needs_ownership_and_active_device(self) -> None: serial_value = await self._admin_serial("edition.personal") unsigned = { "serial": serial_value, "applicationMajorVersion": 1, "idempotencyKey": "idempotency-api-renewal-base", "correlationId": "correlation.test.api_renewal_base", "targetLicenseId": None, "device": self._device(), } body = dict(unsigned) body["deviceProof"] = self._proof("serial.redeem", unsigned) activated = await self._request( "POST", "/v1/serials/redeem", headers={"Authorization": "Bearer customer-token"}, json=body, ) self.assertEqual(activated.status_code, 200, activated.text) activation = activated.json() now = datetime.now(timezone.utc) renewal_payload = OfflineRequestPayload( request_type=OfflineRequestType.RENEW, request_id="offline_renewal.test.api", product_id="apolon.precision_lab_analyzer", application_major_version=1, installation_id=self._device()["installationId"], device_id=activation["deviceId"], device_public_key_base64=self.device_public_key_base64, device_key_thumbprint=self.device_thumbprint, evidence_digests=(("system_uuid", "ab" * 32),), key_provider="software.test", friendly_name="API renewal computer", requested_sku_id=None, created_at=now, expires_at=now + timedelta(days=1), nonce="offline-api-renewal-nonce", target_license_id=activation["licenseId"], ) renewal_signature = self.device_private_key.sign( offline_request_signing_bytes(renewal_payload), ec.ECDSA(hashes.SHA256()), ) renewal_request = DeviceSignedOfflineRequest( payload=renewal_payload, signature_base64=base64.b64encode(renewal_signature).decode("ascii"), ).to_mapping() renewal_body = { "offlineRequest": renewal_request, "idempotencyKey": "idempotency-api-offline-renewal", "correlationId": "correlation.test.api_offline_renewal", } wrong_account = await self._request( "POST", f"/v1/me/licenses/{activation['licenseId']}/offline-renewal", headers={"Authorization": "Bearer member-token"}, json=renewal_body, ) self.assertEqual(wrong_account.status_code, 403, wrong_account.text) renewed = await self._request( "POST", f"/v1/me/licenses/{activation['licenseId']}/offline-renewal", headers={"Authorization": "Bearer customer-token"}, json=renewal_body, ) self.assertEqual(renewed.status_code, 200, renewed.text) renewed_mapping = renewed.json() self.assertFalse(renewed_mapping["idempotentReplay"]) self.assertEqual( renewed_mapping["offlineRequestId"], renewal_payload.request_id, ) snapshot = renewed_mapping["licenseDocument"]["payload"] self.assertEqual(snapshot["nonce"], renewal_payload.nonce) self.assertEqual(snapshot["deviceId"], activation["deviceId"]) self.assertEqual(snapshot["leaseExpiresAt"], snapshot["offlineExpiresAt"]) self.assertEqual( datetime.fromisoformat(snapshot["offlineExpiresAt"]) - datetime.fromisoformat(snapshot["issuedAt"]), timedelta(days=1095), ) replayed = await self._request( "POST", f"/v1/me/licenses/{activation['licenseId']}/offline-renewal", headers={"Authorization": "Bearer customer-token"}, json=renewal_body, ) self.assertEqual(replayed.status_code, 200, replayed.text) self.assertTrue(replayed.json()["idempotentReplay"]) self.assertEqual( replayed.json()["licenseDocument"], renewed_mapping["licenseDocument"], ) with Session(self.engine) as session: self.assertEqual( session.scalar(select(func.count()).select_from(Grant)), 1, ) idempotency = session.scalar( select(IdempotencyRecord).where( IdempotencyRecord.scope == "account.license.offline_renew" ) ) self.assertIsNotNone(idempotency) deactivate_unsigned = { "licenseId": activation["licenseId"], "installationId": self._device()["installationId"], "correlationId": "correlation.test.api_renewal_deactivate", "reason": "local_device_release", } deactivate_body = dict(deactivate_unsigned) deactivate_body["deviceProof"] = self._proof( "activation.deactivate", deactivate_unsigned, ) deactivated = await self._request( "POST", "/v1/activations/deactivate", json=deactivate_body, ) self.assertEqual(deactivated.status_code, 200, deactivated.text) second_payload = replace( renewal_payload, request_id="offline_renewal.test.released", nonce="offline-api-renewal-released-nonce", ) second_signature = self.device_private_key.sign( offline_request_signing_bytes(second_payload), ec.ECDSA(hashes.SHA256()), ) released_device = await self._request( "POST", f"/v1/me/licenses/{activation['licenseId']}/offline-renewal", headers={"Authorization": "Bearer customer-token"}, json={ "offlineRequest": DeviceSignedOfflineRequest( payload=second_payload, signature_base64=base64.b64encode(second_signature).decode("ascii"), ).to_mapping(), "idempotencyKey": "idempotency-api-offline-renewal-released", "correlationId": "correlation.test.api_offline_renewal_released", }, ) self.assertEqual(released_device.status_code, 403, released_device.text) self.assertEqual(released_device.json()["code"], "authorization_denied") async def test_signed_offline_release_is_bound_and_idempotent(self) -> None: serial_value = await self._admin_serial("edition.personal") unsigned = { "serial": serial_value, "applicationMajorVersion": 1, "idempotencyKey": "idempotency-api-offline-release-base", "correlationId": "correlation.test.api_offline_release_base", "targetLicenseId": None, "device": self._device(), } request_body = dict(unsigned) request_body["deviceProof"] = self._proof("serial.redeem", unsigned) activated = await self._request( "POST", "/v1/serials/redeem", json=request_body, ) self.assertEqual(activated.status_code, 200, activated.text) activated_body = activated.json() now = datetime.now(timezone.utc) release_payload = OfflineRequestPayload( request_type=OfflineRequestType.RELEASE, request_id="offline_release.test.api", product_id="apolon.precision_lab_analyzer", application_major_version=1, installation_id=self._device()["installationId"], device_id=activated_body["deviceId"], device_public_key_base64=self.device_public_key_base64, device_key_thumbprint=self.device_thumbprint, evidence_digests=(("system_uuid", "ab" * 32),), key_provider="software.test", friendly_name="Offline release API test computer", requested_sku_id=None, created_at=now, expires_at=now + timedelta(days=1), nonce="offline-release-api-nonce", target_license_id=activated_body["licenseId"], ) release_signature = self.device_private_key.sign( offline_request_signing_bytes(release_payload), ec.ECDSA(hashes.SHA256()), ) release_document = DeviceSignedOfflineRequest( payload=release_payload, signature_base64=base64.b64encode(release_signature).decode("ascii"), ).to_mapping() released = await self._request( "POST", "/v1/activations/release-file", json={"offlineRequest": release_document}, ) self.assertEqual(released.status_code, 200, released.text) self.assertEqual(released.json()["status"], "deactivated") replayed = await self._request( "POST", "/v1/activations/release-file", json={"offlineRequest": release_document}, ) self.assertEqual(replayed.status_code, 200, replayed.text) self.assertEqual( replayed.json()["activationId"], released.json()["activationId"], ) with Session(self.engine) as session: active_allocations = session.scalar( select(func.count()) .select_from(Activation) .where(Activation.status == "active") ) self.assertEqual(active_allocations, 0) async def test_anonymous_serial_claim_and_remote_device_release(self) -> None: anonymous_serial = await self._admin_serial("edition.personal") anonymous_unsigned = { "serial": anonymous_serial, "applicationMajorVersion": 1, "idempotencyKey": "idempotency-api-anonymous-claim", "correlationId": "correlation.test.api_anonymous_claim", "targetLicenseId": None, "device": self._device(), } anonymous_body = dict(anonymous_unsigned) anonymous_body["deviceProof"] = self._proof( "serial.redeem", anonymous_unsigned, ) anonymous_response = await self._request( "POST", "/v1/serials/redeem", json=anonymous_body, ) self.assertEqual(anonymous_response.status_code, 200, anonymous_response.text) claimed = await self._request( "POST", "/v1/me/licenses/claim-serial", headers={"Authorization": "Bearer customer-token"}, json={ "serial": anonymous_serial, "correlationId": "correlation.test.api_claim", }, ) self.assertEqual(claimed.status_code, 200, claimed.text) self.assertEqual( claimed.json()["licenseId"], anonymous_response.json()["licenseId"], ) listed = await self._request( "GET", "/v1/me/licenses", headers={"Authorization": "Bearer customer-token"}, ) self.assertEqual(listed.status_code, 200, listed.text) license_entry = listed.json()["licenses"][0] device_id = license_entry["devices"][0]["deviceId"] rename_body = { "friendlyName": "Renamed API laboratory computer", "idempotencyKey": "idempotency-api-device-rename", "correlationId": "correlation.test.api_device_rename", } renamed = await self._request( "PATCH", f"/v1/me/licenses/{license_entry['licenseId']}/devices/{device_id}", headers={"Authorization": "Bearer customer-token"}, json=rename_body, ) self.assertEqual(renamed.status_code, 200, renamed.text) self.assertEqual( renamed.json()["friendlyName"], "Renamed API laboratory computer", ) replayed_rename = await self._request( "PATCH", f"/v1/me/licenses/{license_entry['licenseId']}/devices/{device_id}", headers={"Authorization": "Bearer customer-token"}, json=rename_body, ) self.assertEqual(replayed_rename.status_code, 200, replayed_rename.text) self.assertEqual(replayed_rename.json(), renamed.json()) unauthorized_rename = await self._request( "PATCH", f"/v1/me/licenses/{license_entry['licenseId']}/devices/{device_id}", headers={"Authorization": "Bearer member-token"}, json={ "friendlyName": "Unauthorized rename", "idempotencyKey": "idempotency-api-device-rename-denied", "correlationId": "correlation.test.api_device_rename_denied", }, ) self.assertEqual(unauthorized_rename.status_code, 403, unauthorized_rename.text) self.assertEqual(unauthorized_rename.json()["code"], "authorization_denied") renamed_listing = await self._request( "GET", "/v1/me/licenses", headers={"Authorization": "Bearer customer-token"}, ) self.assertEqual( renamed_listing.json()["licenses"][0]["devices"][0]["friendlyName"], "Renamed API laboratory computer", ) released = await self._request( "POST", f"/v1/me/licenses/{license_entry['licenseId']}/devices/{device_id}/release-unavailable", headers={"Authorization": "Bearer customer-token"}, json={ "correlationId": "correlation.test.api_remote_release", "reason": "marked_lost", }, ) self.assertEqual(released.status_code, 200, released.text) self.assertEqual(released.json()["remainingAutomaticReleases"], 2) after_release = await self._request( "GET", "/v1/me/licenses", headers={"Authorization": "Bearer customer-token"}, ) self.assertEqual(after_release.status_code, 200, after_release.text) self.assertEqual(after_release.json()["licenses"][0]["devices"], []) with Session(self.engine) as session: device = session.get(Device, device_id) installation = session.scalar( select(DeviceInstallation).where( DeviceInstallation.device_id == device_id ) ) self.assertIsNotNone(device) self.assertIsNotNone(installation) assert device is not None assert installation is not None owner = session.scalar( select(User).where(User.external_subject == "customer-subject") ) self.assertIsNotNone(owner) assert owner is not None self.assertEqual(device.owner_user_id, owner.id) self.assertEqual(device.status, "lost") self.assertEqual(installation.status, "revoked") reactivation_proof_payload = { "licenseId": license_entry["licenseId"], "applicationMajorVersion": 1, "idempotencyKey": "idempotency-api-lost-reactivation", "correlationId": "correlation.test.api_lost_reactivation", "device": self._device(), } reactivation_body = dict(reactivation_proof_payload) reactivation_body.pop("licenseId") reactivation_body["deviceProof"] = self._proof( "account.license.activate", reactivation_proof_payload, ) reactivation = await self._request( "POST", f"/v1/me/licenses/{license_entry['licenseId']}/activate", headers={"Authorization": "Bearer customer-token"}, json=reactivation_body, ) self.assertEqual(reactivation.status_code, 409, reactivation.text) self.assertEqual(reactivation.json()["code"], "device_recovery_required") replacement_private_key = ec.generate_private_key(ec.SECP256R1()) replacement_public_key_der = replacement_private_key.public_key().public_bytes( serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo, ) replacement_public_key_base64 = base64.b64encode( replacement_public_key_der ).decode("ascii") replacement_thumbprint = device_public_key_thumbprint( replacement_public_key_der ) replacement_device = self._device() replacement_device.update( { "installationId": "installation.test.api.recovered", "devicePublicKey": replacement_public_key_base64, "deviceKeyThumbprint": replacement_thumbprint, } ) replacement_proof_payload = { "licenseId": license_entry["licenseId"], "applicationMajorVersion": 1, "idempotencyKey": "idempotency-api-lost-replacement-key", "correlationId": "correlation.test.api_lost_replacement_key", "device": replacement_device, } replacement_body = dict(replacement_proof_payload) replacement_body.pop("licenseId") replacement_body["deviceProof"] = self._proof( "account.license.activate", replacement_proof_payload, private_key=replacement_private_key, public_key_base64=replacement_public_key_base64, key_thumbprint=replacement_thumbprint, ) replacement_while_lost = await self._request( "POST", f"/v1/me/licenses/{license_entry['licenseId']}/activate", headers={"Authorization": "Bearer customer-token"}, json=replacement_body, ) self.assertEqual( replacement_while_lost.status_code, 409, replacement_while_lost.text, ) self.assertEqual( replacement_while_lost.json()["code"], "device_recovery_required", ) with Session(self.engine) as session: self.assertEqual( session.scalar(select(func.count()).select_from(Device)), 1, ) self.assertEqual( session.scalar(select(func.count()).select_from(DeviceInstallation)), 1, ) missing_scope = await self._request( "POST", f"/v1/admin/devices/{device_id}/recovery-preview", headers={"Authorization": "Bearer customer-token"}, json={ "reasonCode": "customer_recovered_device", "correlationId": "correlation.test.api_device_recovery_missing_scope", }, ) self.assertEqual(missing_scope.status_code, 403, missing_scope.text) scoped_non_admin = await self._request( "POST", f"/v1/admin/devices/{device_id}/recovery-preview", headers={"Authorization": "Bearer support-customer-token"}, json={ "reasonCode": "customer_recovered_device", "correlationId": "correlation.test.api_device_recovery_non_admin", }, ) self.assertEqual(scoped_non_admin.status_code, 403, scoped_non_admin.text) stale_admin = await self._request( "POST", f"/v1/admin/devices/{device_id}/recovery-preview", headers={"Authorization": "Bearer stale-admin-token"}, json={ "reasonCode": "customer_recovered_device", "correlationId": "correlation.test.api_device_recovery_stale_admin", }, ) self.assertEqual(stale_admin.status_code, 403, stale_admin.text) self.assertEqual(stale_admin.json()["code"], "reauthentication_required") preview_body = { "reasonCode": "customer_recovered_device", "note": "Customer verified possession of the recovered computer.", "correlationId": "correlation.test.api_device_recovery_preview", } previewed = await self._request( "POST", f"/v1/admin/devices/{device_id}/recovery-preview", headers={"Authorization": "Bearer admin-token"}, json=preview_body, ) self.assertEqual(previewed.status_code, 200, previewed.text) preview = previewed.json() self.assertEqual(preview["currentStatus"], "lost") self.assertEqual(preview["revokedInstallationCount"], 1) self.assertTrue(preview["requiresNewDeviceProofKey"]) stale_execute = await self._request( "POST", f"/v1/admin/devices/{device_id}/recover", headers={"Authorization": "Bearer admin-token"}, json={ **preview_body, "expectedStateDigest": "0" * 64, "idempotencyKey": "idempotency-api-device-recovery-stale", }, ) self.assertEqual(stale_execute.status_code, 409, stale_execute.text) execute_body = { **preview_body, "expectedStateDigest": preview["stateDigest"], "idempotencyKey": "idempotency-api-device-recovery", } restored = await self._request( "POST", f"/v1/admin/devices/{device_id}/recover", headers={"Authorization": "Bearer admin-token"}, json=execute_body, ) self.assertEqual(restored.status_code, 200, restored.text) self.assertEqual(restored.json()["currentStatus"], "active") self.assertFalse(restored.json()["idempotentReplay"]) replayed_restore = await self._request( "POST", f"/v1/admin/devices/{device_id}/recover", headers={"Authorization": "Bearer admin-token"}, json=execute_body, ) self.assertEqual(replayed_restore.status_code, 200, replayed_restore.text) self.assertTrue(replayed_restore.json()["idempotentReplay"]) old_key_after_restore = await self._request( "POST", f"/v1/me/licenses/{license_entry['licenseId']}/activate", headers={"Authorization": "Bearer customer-token"}, json=reactivation_body, ) self.assertEqual(old_key_after_restore.status_code, 409, old_key_after_restore.text) self.assertEqual( old_key_after_restore.json()["code"], "device_identity_replacement_required", ) replacement_after_restore = await self._request( "POST", f"/v1/me/licenses/{license_entry['licenseId']}/activate", headers={"Authorization": "Bearer customer-token"}, json=replacement_body, ) self.assertEqual( replacement_after_restore.status_code, 200, replacement_after_restore.text, ) self.assertEqual(replacement_after_restore.json()["deviceId"], device_id) with Session(self.engine) as session: device = session.get(Device, device_id) installations = session.scalars( select(DeviceInstallation).where( DeviceInstallation.device_id == device_id ) ).all() event = session.scalar( select(AuditEvent).where( AuditEvent.action == "device.recovery_approved" ) ) self.assertIsNotNone(device) self.assertIsNotNone(event) assert device is not None self.assertEqual(device.status, "active") self.assertEqual(len(installations), 2) self.assertEqual( {installation.status for installation in installations}, {"active", "revoked"}, ) async def test_team_organization_invitation_seat_activation_and_release(self) -> None: serial_value = await self._admin_serial("edition.lab_team") unsigned = { "serial": serial_value, "applicationMajorVersion": 1, "idempotencyKey": "idempotency-api-team-redemption", "correlationId": "correlation.test.api_team_redemption", "targetLicenseId": None, "device": self._device(), } redemption_body = dict(unsigned) redemption_body["deviceProof"] = self._proof("serial.redeem", unsigned) redeemed = await self._request( "POST", "/v1/serials/redeem", headers={"Authorization": "Bearer customer-token"}, json=redemption_body, ) self.assertEqual(redeemed.status_code, 200, redeemed.text) license_id = redeemed.json()["licenseId"] organization_request = { "displayName": "Apolon Test Laboratory", "idempotencyKey": "idempotency-api-organization-create", "correlationId": "correlation.test.api_organization_create", } organization = await self._request( "POST", "/v1/me/organizations", headers={"Authorization": "Bearer customer-token"}, json=organization_request, ) self.assertEqual(organization.status_code, 200, organization.text) organization_id = organization.json()["organizationId"] attached = await self._request( "POST", f"/v1/me/organizations/{organization_id}/licenses/{license_id}/attach", headers={"Authorization": "Bearer customer-token"}, json={ "idempotencyKey": "idempotency-api-team-attach", "correlationId": "correlation.test.api_team_attach", }, ) self.assertEqual(attached.status_code, 200, attached.text) self.assertEqual(len(attached.json()["seats"]), 5) available_seat = next( seat for seat in attached.json()["seats"] if seat["status"] == "available" ) invitation_request = { "verifiedEmail": "member@example.test", "role": "member", "idempotencyKey": "idempotency-api-team-invite", "correlationId": "correlation.test.api_team_invite", } invited = await self._request( "POST", f"/v1/me/organizations/{organization_id}/invitations", headers={"Authorization": "Bearer customer-token"}, json=invitation_request, ) self.assertEqual(invited.status_code, 200, invited.text) invitation_token = invited.json()["invitationToken"] replayed_invitation = await self._request( "POST", f"/v1/me/organizations/{organization_id}/invitations", headers={"Authorization": "Bearer customer-token"}, json=invitation_request, ) self.assertEqual(replayed_invitation.status_code, 200, replayed_invitation.text) self.assertEqual( replayed_invitation.json()["invitationToken"], invitation_token, ) malformed_invitation = await self._request( "POST", "/v1/me/organization-invitations/accept", headers={"Authorization": "Bearer member-token"}, json={ "invitationToken": "not-an-organization-invitation", "idempotencyKey": "idempotency-api-team-malformed-accept", "correlationId": "correlation.test.api_team_malformed_accept", }, ) self.assertEqual( malformed_invitation.status_code, 400, malformed_invitation.text, ) self.assertEqual(malformed_invitation.json()["code"], "invalid_request") wrong_account = await self._request( "POST", "/v1/me/organization-invitations/accept", headers={"Authorization": "Bearer wrong-member-token"}, json={ "invitationToken": invitation_token, "idempotencyKey": "idempotency-api-team-wrong-accept", "correlationId": "correlation.test.api_team_wrong_accept", }, ) self.assertEqual(wrong_account.status_code, 403, wrong_account.text) self.assertEqual(wrong_account.json()["code"], "authorization_denied") accepted = await self._request( "POST", "/v1/me/organization-invitations/accept", headers={"Authorization": "Bearer member-token"}, json={ "invitationToken": invitation_token, "idempotencyKey": "idempotency-api-team-accept", "correlationId": "correlation.test.api_team_accept", }, ) self.assertEqual(accepted.status_code, 200, accepted.text) listed = await self._request( "GET", "/v1/me/organizations", headers={"Authorization": "Bearer customer-token"}, ) self.assertEqual(listed.status_code, 200, listed.text) organization_summary = listed.json()["organizations"][0] member = next( value for value in organization_summary["memberships"] if value["verifiedEmail"] == "member@example.test" ) assigned = await self._request( "POST", f"/v1/me/organizations/{organization_id}/seats/{available_seat['seatId']}/assign", headers={"Authorization": "Bearer customer-token"}, json={ "userId": member["userId"], "idempotencyKey": "idempotency-api-team-seat-assign", "correlationId": "correlation.test.api_team_seat_assign", }, ) self.assertEqual(assigned.status_code, 200, assigned.text) self.assertEqual(assigned.json()["assignedUserId"], member["userId"]) member_private_key = ec.generate_private_key(ec.SECP256R1()) member_public_key_der = member_private_key.public_key().public_bytes( encoding=serialization.Encoding.DER, format=serialization.PublicFormat.SubjectPublicKeyInfo, ) member_public_key_base64 = base64.b64encode(member_public_key_der).decode("ascii") member_thumbprint = device_public_key_thumbprint(member_public_key_der) member_device = { "installationId": "installation.test.api.team_member", "devicePublicKey": member_public_key_base64, "deviceKeyThumbprint": member_thumbprint, "keyProvider": "software.test", "friendlyName": "Team member computer", "evidence": {"system_uuid": "team-member-system-uuid"}, } activation_payload = { "licenseId": license_id, "applicationMajorVersion": 1, "idempotencyKey": "idempotency-api-team-member-activate", "correlationId": "correlation.test.api_team_member_activate", "device": member_device, } proof_time = datetime.now(timezone.utc) unsigned_proof = DeviceActionProof( action="account.license.activate", installation_id=member_device["installationId"], device_public_key_base64=member_public_key_base64, device_key_thumbprint=member_thumbprint, payload_digest=device_action_payload_digest(activation_payload), created_at=proof_time, expires_at=proof_time + timedelta(minutes=5), nonce="api-team-member-device-proof", signature_base64=base64.b64encode(b"placeholder").decode("ascii"), ) proof_signature = member_private_key.sign( device_proof_signing_bytes(unsigned_proof), ec.ECDSA(hashes.SHA256()), ) activation_body = { key: value for key, value in activation_payload.items() if key != "licenseId" } activation_body["deviceProof"] = { **unsigned_proof.to_mapping(), "signature": base64.b64encode(proof_signature).decode("ascii"), } activated = await self._request( "POST", f"/v1/me/licenses/{license_id}/activate", headers={"Authorization": "Bearer member-token"}, json=activation_body, ) self.assertEqual(activated.status_code, 200, activated.text) member_listing = await self._request( "GET", "/v1/me/licenses", headers={"Authorization": "Bearer member-token"}, ) self.assertEqual(member_listing.status_code, 200, member_listing.text) member_devices = member_listing.json()["licenses"][0]["devices"] self.assertEqual(len(member_devices), 1) self.assertEqual(member_devices[0]["deviceId"], activated.json()["deviceId"]) member_renamed = await self._request( "PATCH", f"/v1/me/licenses/{license_id}/devices/{activated.json()['deviceId']}", headers={"Authorization": "Bearer member-token"}, json={ "friendlyName": "Member analysis workstation", "idempotencyKey": "idempotency-api-team-member-rename", "correlationId": "correlation.test.api_team_member_rename", }, ) self.assertEqual(member_renamed.status_code, 200, member_renamed.text) another_seat_denied = await self._request( "PATCH", f"/v1/me/licenses/{license_id}/devices/{redeemed.json()['deviceId']}", headers={"Authorization": "Bearer member-token"}, json={ "friendlyName": "Another seat device", "idempotencyKey": "idempotency-api-team-other-seat-rename", "correlationId": "correlation.test.api_team_other_seat_rename", }, ) self.assertEqual( another_seat_denied.status_code, 403, another_seat_denied.text, ) self.assertEqual( another_seat_denied.json()["code"], "authorization_denied", ) released = await self._request( "POST", f"/v1/me/organizations/{organization_id}/seats/{available_seat['seatId']}/release", headers={"Authorization": "Bearer customer-token"}, json={ "idempotencyKey": "idempotency-api-team-seat-release", "correlationId": "correlation.test.api_team_seat_release", }, ) self.assertEqual(released.status_code, 200, released.text) self.assertEqual(released.json()["status"], "available") self.assertEqual(released.json()["releasedActivationCount"], 1) with Session(self.engine) as session: stored_idempotency = session.scalar( select(IdempotencyRecord).where( IdempotencyRecord.scope == "organization.invitation.create" ) ) self.assertNotIn("invitationToken", stored_idempotency.response_json) stored_invitation = session.scalar(select(OrganizationInvitation)) self.assertNotEqual( stored_invitation.token_digest, invitation_token.encode("utf-8"), ) member_seat = session.get(OrganizationSeat, available_seat["seatId"]) self.assertEqual(member_seat.status, "available") active_member_activation = session.scalar( select(func.count()) .select_from(Activation) .where( Activation.seat_id == available_seat["seatId"], Activation.status == "active", ) ) self.assertEqual(active_member_activation, 0) async def test_support_inspection_and_grant_revocation_preserve_other_rights(self) -> None: personal_serial = await self._admin_serial("edition.personal") personal_unsigned = { "serial": personal_serial, "applicationMajorVersion": 1, "idempotencyKey": "idempotency-api-support-personal", "correlationId": "correlation.test.api_support_personal", "targetLicenseId": None, "device": self._device(), } personal_body = dict(personal_unsigned) personal_body["deviceProof"] = self._proof( "serial.redeem", personal_unsigned, ) personal = await self._request( "POST", "/v1/serials/redeem", headers={"Authorization": "Bearer customer-token"}, json=personal_body, ) self.assertEqual(personal.status_code, 200, personal.text) license_id = personal.json()["licenseId"] addon_serial = await self._admin_serial("addon.advanced_analysis") addon_unsigned = { "serial": addon_serial, "applicationMajorVersion": 1, "idempotencyKey": "idempotency-api-support-addon", "correlationId": "correlation.test.api_support_addon", "targetLicenseId": license_id, "device": self._device(), } addon_body = dict(addon_unsigned) addon_body["deviceProof"] = self._proof("serial.redeem", addon_unsigned) addon = await self._request( "POST", "/v1/serials/redeem", headers={"Authorization": "Bearer customer-token"}, json=addon_body, ) self.assertEqual(addon.status_code, 200, addon.text) self.assertIn( "feature.analysis.advanced", addon.json()["licenseDocument"]["payload"]["entitlements"], ) missing_scope = await self._request( "GET", f"/v1/admin/licenses/{license_id}", headers={"Authorization": "Bearer customer-token"}, ) self.assertEqual(missing_scope.status_code, 403, missing_scope.text) scoped_non_admin = await self._request( "GET", f"/v1/admin/licenses/{license_id}", headers={"Authorization": "Bearer support-customer-token"}, ) self.assertEqual(scoped_non_admin.status_code, 403, scoped_non_admin.text) stale_admin = await self._request( "GET", f"/v1/admin/licenses/{license_id}", headers={"Authorization": "Bearer stale-admin-token"}, ) self.assertEqual(stale_admin.status_code, 403, stale_admin.text) self.assertEqual(stale_admin.json()["code"], "reauthentication_required") inspected = await self._request( "GET", f"/v1/admin/licenses/{license_id}", headers={"Authorization": "Bearer admin-token"}, ) self.assertEqual(inspected.status_code, 200, inspected.text) inspection = inspected.json() serialized_inspection = json.dumps(inspection) self.assertNotIn("sourceReference", serialized_inspection) self.assertNotIn("evidence", serialized_inspection) self.assertNotIn("devicePublicKey", serialized_inspection) edition_grant = next( value for value in inspection["grants"] if value["skuId"] == "edition.personal" ) addon_grant = next( value for value in inspection["grants"] if value["skuId"] == "addon.advanced_analysis" ) preview_body = { "reasonCode": "refund", "note": "Customer requested an add-on refund", "correlationId": "correlation.test.api_support_revoke_preview", } previewed = await self._request( "POST", ( f"/v1/admin/licenses/{license_id}/grants/" f"{addon_grant['grantId']}/revocation-preview" ), headers={"Authorization": "Bearer admin-token"}, json=preview_body, ) self.assertEqual(previewed.status_code, 200, previewed.text) preview = previewed.json() self.assertEqual(preview["removedEntitlements"], ["feature.analysis.advanced"]) self.assertIn("feature.capture.core", preview["preservedEntitlements"]) self.assertFalse(preview["wouldDeactivateLicense"]) execute_body = { **preview_body, "idempotencyKey": "idempotency-api-support-revoke", "expectedRevocationGeneration": preview["previewRevocationGeneration"], } stale_execute = await self._request( "POST", ( f"/v1/admin/licenses/{license_id}/grants/" f"{addon_grant['grantId']}/revoke" ), headers={"Authorization": "Bearer admin-token"}, json={ **execute_body, "expectedRevocationGeneration": ( preview["previewRevocationGeneration"] + 1 ), "idempotencyKey": "idempotency-api-support-revoke-stale", }, ) self.assertEqual(stale_execute.status_code, 409, stale_execute.text) revoked = await self._request( "POST", ( f"/v1/admin/licenses/{license_id}/grants/" f"{addon_grant['grantId']}/revoke" ), headers={"Authorization": "Bearer admin-token"}, json=execute_body, ) self.assertEqual(revoked.status_code, 200, revoked.text) self.assertFalse(revoked.json()["idempotentReplay"]) replayed = await self._request( "POST", ( f"/v1/admin/licenses/{license_id}/grants/" f"{addon_grant['grantId']}/revoke" ), headers={"Authorization": "Bearer admin-token"}, json=execute_body, ) self.assertEqual(replayed.status_code, 200, replayed.text) self.assertTrue(replayed.json()["idempotentReplay"]) after = await self._request( "GET", f"/v1/admin/licenses/{license_id}", headers={"Authorization": "Bearer admin-token"}, ) self.assertEqual(after.status_code, 200, after.text) after_mapping = after.json() after_edition = next( value for value in after_mapping["grants"] if value["grantId"] == edition_grant["grantId"] ) after_addon = next( value for value in after_mapping["grants"] if value["grantId"] == addon_grant["grantId"] ) self.assertEqual(after_edition["status"], "active") self.assertEqual(after_addon["status"], "revoked") self.assertIn("feature.capture.core", after_mapping["effectiveEntitlements"]) self.assertNotIn( "feature.analysis.advanced", after_mapping["effectiveEntitlements"], ) refresh_unsigned = { "licenseId": license_id, "installationId": self._device()["installationId"], "applicationMajorVersion": 1, "correlationId": "correlation.test.api_support_refresh", } refresh_body = dict(refresh_unsigned) refresh_body["deviceProof"] = self._proof( "activation.refresh", refresh_unsigned, ) refreshed = await self._request( "POST", "/v1/activations/refresh", json=refresh_body, ) self.assertEqual(refreshed.status_code, 200, refreshed.text) refreshed_entitlements = refreshed.json()["licenseDocument"]["payload"][ "entitlements" ] self.assertIn("feature.capture.core", refreshed_entitlements) self.assertNotIn("feature.analysis.advanced", refreshed_entitlements) with Session(self.engine) as session: audit = session.scalar( select(AuditEvent).where(AuditEvent.action == "grant.revoked") ) self.assertEqual(audit.metadata_json["reasonCode"], "refund") async def test_support_trial_extension_is_bounded_additive_and_idempotent(self) -> None: trial_unsigned = { "applicationMajorVersion": 1, "idempotencyKey": "idempotency-api-support-trial-start", "correlationId": "correlation.test.api_support_trial_start", "device": self._device(), } trial_body = dict(trial_unsigned) trial_body["deviceProof"] = self._proof("trial.start", trial_unsigned) started = await self._request( "POST", "/v1/trials/start", headers={"Authorization": "Bearer customer-token"}, json=trial_body, ) self.assertEqual(started.status_code, 200, started.text) license_id = started.json()["licenseId"] inspected = await self._request( "GET", f"/v1/admin/licenses/{license_id}", headers={"Authorization": "Bearer admin-token"}, ) self.assertEqual(inspected.status_code, 200, inspected.text) initial = inspected.json() trial = initial["trials"][0] original_grant = next( value for value in initial["grants"] if value["grantId"] == trial["originalGrantId"] ) preview_body = { "reasonCode": "evaluation_extension", "note": "Approved evaluation recovery", "correlationId": "correlation.test.api_support_trial_preview", } previewed = await self._request( "POST", f"/v1/admin/trials/{trial['trialId']}/extension-preview", headers={"Authorization": "Bearer admin-token"}, json=preview_body, ) self.assertEqual(previewed.status_code, 200, previewed.text) preview = previewed.json() self.assertEqual(preview["extensionDays"], 7) self.assertEqual(preview["extensionNumber"], 1) extension_start = datetime.fromisoformat(preview["extensionStartsAt"]) extension_end = datetime.fromisoformat(preview["extensionEndsAt"]) self.assertEqual(extension_end - extension_start, timedelta(days=7)) execute_body = { **preview_body, "idempotencyKey": "idempotency-api-support-trial-extend", "expectedRevocationGeneration": preview["previewRevocationGeneration"], } extended = await self._request( "POST", f"/v1/admin/trials/{trial['trialId']}/extend", headers={"Authorization": "Bearer admin-token"}, json=execute_body, ) self.assertEqual(extended.status_code, 200, extended.text) self.assertFalse(extended.json()["idempotentReplay"]) replayed = await self._request( "POST", f"/v1/admin/trials/{trial['trialId']}/extend", headers={"Authorization": "Bearer admin-token"}, json=execute_body, ) self.assertEqual(replayed.status_code, 200, replayed.text) self.assertTrue(replayed.json()["idempotentReplay"]) exhausted = await self._request( "POST", f"/v1/admin/trials/{trial['trialId']}/extension-preview", headers={"Authorization": "Bearer admin-token"}, json={ **preview_body, "correlationId": "correlation.test.api_support_trial_exhausted", }, ) self.assertEqual(exhausted.status_code, 409, exhausted.text) self.assertEqual(exhausted.json()["code"], "trial_not_eligible") after = await self._request( "GET", f"/v1/admin/licenses/{license_id}", headers={"Authorization": "Bearer admin-token"}, ) self.assertEqual(after.status_code, 200, after.text) after_mapping = after.json() after_trial = after_mapping["trials"][0] self.assertEqual(after_trial["extensionCount"], 1) self.assertEqual(len(after_trial["extensionGrantIds"]), 1) unchanged_original = next( value for value in after_mapping["grants"] if value["grantId"] == original_grant["grantId"] ) self.assertEqual(unchanged_original["endsAt"], original_grant["endsAt"]) extension_grant = next( value for value in after_mapping["grants"] if value["grantId"] == after_trial["extensionGrantIds"][0] ) self.assertEqual(extension_grant["sourceType"], "trial_extension") with Session(self.engine) as session: extension_audits = session.scalar( select(func.count()) .select_from(AuditEvent) .where(AuditEvent.action == "trial.extended") ) self.assertEqual(extension_audits, 1) async def test_stripe_webhook_uses_raw_signature_and_durable_idempotent_queue(self) -> None: serial_value = await self._admin_serial("edition.personal") unsigned = { "serial": serial_value, "applicationMajorVersion": 1, "idempotencyKey": "idempotency-api-subscription-base", "correlationId": "correlation.test.api_subscription_base", "targetLicenseId": None, "device": self._device(), } body = dict(unsigned) body["deviceProof"] = self._proof("serial.redeem", unsigned) redeemed = await self._request( "POST", "/v1/serials/redeem", headers={"Authorization": "Bearer customer-token"}, json=body, ) self.assertEqual(redeemed.status_code, 200, redeemed.text) now = datetime.now(timezone.utc) event = { "id": "evt_test_api_subscription", "object": "event", "type": "customer.subscription.updated", "created": int(now.timestamp()), "data": { "object": { "id": "sub_test_api_subscription", "customer": "cus_test_api_subscription", "object": "subscription", "status": "active", "cancel_at_period_end": False, "metadata": {"license_id": redeemed.json()["licenseId"]}, "items": { "data": [ { "id": "si_test_api_subscription", "quantity": 1, "current_period_start": int(now.timestamp()), "current_period_end": int( (now + timedelta(days=30)).timestamp() ), "price": {"id": TEST_STRIPE_PRICE_ID}, } ] }, } }, } raw = json.dumps(event, separators=(",", ":")).encode("utf-8") timestamp = int(now.timestamp()) signature = hmac.new( TEST_STRIPE_WEBHOOK_SECRET.encode("utf-8"), str(timestamp).encode("ascii") + b"." + raw, hashlib.sha256, ).hexdigest() headers = { "Content-Type": "application/json", "Stripe-Signature": f"t={timestamp},v1={signature}", } queued = await self._request( "POST", "/v1/webhooks/stripe", headers=headers, content=raw, ) self.assertEqual(queued.status_code, 200, queued.text) self.assertTrue(queued.json()["queued"]) duplicate = await self._request( "POST", "/v1/webhooks/stripe", headers=headers, content=raw, ) self.assertEqual(duplicate.status_code, 200, duplicate.text) self.assertFalse(duplicate.json()["queued"]) with Session(self.engine) as session: processed = process_subscription_outbox( session, self.app.state.catalog, now=datetime.now(timezone.utc) + timedelta(seconds=1), ) self.assertEqual(processed, 1) with Session(self.engine) as session: self.assertEqual(session.scalar(select(func.count()).select_from(WebhookEvent)), 1) self.assertEqual(session.scalar(select(func.count()).select_from(OutboxEvent)), 1) self.assertEqual(session.scalar(select(Subscription)).status, "active") subscription_grant = session.scalar( select(Grant).where(Grant.source_type == "subscription") ) self.assertEqual(subscription_grant.status, "pending") incident = { "id": "evt_test_api_dispute", "object": "event", "type": "charge.dispute.created", "created": int(now.timestamp()), "data": { "object": { "id": "dp_test_api_subscription", "object": "dispute", "charge": "ch_test_api_subscription", "status": "needs_response", } }, } incident_raw = json.dumps(incident, separators=(",", ":")).encode("utf-8") incident_signature = hmac.new( TEST_STRIPE_WEBHOOK_SECRET.encode("utf-8"), str(timestamp).encode("ascii") + b"." + incident_raw, hashlib.sha256, ).hexdigest() incident_response = await self._request( "POST", "/v1/webhooks/stripe", headers={ "Content-Type": "application/json", "Stripe-Signature": f"t={timestamp},v1={incident_signature}", }, content=incident_raw, ) self.assertEqual(incident_response.status_code, 200, incident_response.text) self.assertTrue(incident_response.json()["queued"]) with Session(self.engine) as session: self.assertEqual(session.scalar(select(func.count()).select_from(WebhookEvent)), 2) self.assertEqual(session.scalar(select(func.count()).select_from(OutboxEvent)), 2) async def test_checkout_and_portal_sessions_are_owned_idempotent_and_non_authoritative( self, ) -> None: serial_value = await self._admin_serial("edition.personal") unsigned = { "serial": serial_value, "applicationMajorVersion": 1, "idempotencyKey": "idempotency-api-billing-base", "correlationId": "correlation.test.api_billing_base", "targetLicenseId": None, "device": self._device(), } body = dict(unsigned) body["deviceProof"] = self._proof("serial.redeem", unsigned) redeemed = await self._request( "POST", "/v1/serials/redeem", headers={"Authorization": "Bearer customer-token"}, json=body, ) self.assertEqual(redeemed.status_code, 200, redeemed.text) license_id = redeemed.json()["licenseId"] provider = FakeBillingProvider() settings = replace( self.app.state.settings, stripe_commerce_enabled=True, stripe_secret_key="sk_test_owned_billing", stripe_checkout_success_url="https://account.example.test/checkout/success", stripe_checkout_cancel_url="https://account.example.test/checkout/cancel", stripe_billing_portal_return_url="https://account.example.test/licenses", ) self.app = create_app( settings, engine=self.engine, signer=self.signer, token_validator=StaticTokenValidator(), billing_provider=provider, ) original_catalog = self.app.state.catalog active_subscription = replace( original_catalog.skus["subscription.ai_assist"], active=True, ) subscription_catalog = LicensingCatalog( product_id=original_catalog.product_id, revision=original_catalog.revision + 1, entitlements=original_catalog.entitlements, device_policies=original_catalog.device_policies, skus={ **original_catalog.skus, active_subscription.sku_id: active_subscription, }, ) self.app.state.catalog = subscription_catalog now = datetime.now(timezone.utc) with Session(self.engine) as session, session.begin(): active_release = session.scalar( select(CatalogRelease).where(CatalogRelease.status == "active") ) assert active_release is not None active_release.status = "retired" active_release.retired_at = now session.flush() session.add( CatalogRelease( id="catalog_release.test.subscription_checkout", product_id=subscription_catalog.product_id, revision=subscription_catalog.revision, document_json=subscription_catalog.to_mapping(), document_digest=bytes.fromhex(subscription_catalog.sha256()), status="active", staged_by_user_id="user.test.admin", staged_at=now, published_by_user_id="user.test.admin", published_at=now, retired_at=None, ) ) checkout_body = { "skuId": "subscription.ai_assist", "idempotencyKey": "idempotency-api-owned-checkout", "correlationId": "correlation.test.api_owned_checkout", } with Session(self.engine) as session: grant_count_before = session.scalar(select(func.count()).select_from(Grant)) checkout = await self._request( "POST", f"/v1/me/licenses/{license_id}/checkout-sessions", headers={"Authorization": "Bearer customer-token"}, json=checkout_body, ) self.assertEqual(checkout.status_code, 200, checkout.text) self.assertFalse(checkout.json()["idempotentReplay"]) self.assertEqual( checkout.json()["url"], "https://checkout.stripe.test/session/owned", ) replay = await self._request( "POST", f"/v1/me/licenses/{license_id}/checkout-sessions", headers={"Authorization": "Bearer customer-token"}, json=checkout_body, ) self.assertEqual(replay.status_code, 200, replay.text) self.assertTrue(replay.json()["idempotentReplay"]) self.assertEqual(len(provider.checkout_requests), 1) self.assertEqual( provider.checkout_requests[0]["license_id"], license_id, ) with Session(self.engine) as session: self.assertEqual( session.scalar(select(func.count()).select_from(Grant)), grant_count_before, ) self.assertEqual(session.scalar(select(func.count()).select_from(Subscription)), 0) denied = await self._request( "POST", f"/v1/me/licenses/{license_id}/checkout-sessions", headers={"Authorization": "Bearer member-token"}, json={ **checkout_body, "idempotencyKey": "idempotency-api-denied-checkout", }, ) self.assertEqual(denied.status_code, 403, denied.text) stale = await self._request( "POST", f"/v1/me/licenses/{license_id}/checkout-sessions", headers={"Authorization": "Bearer stale-customer-token"}, json={ **checkout_body, "idempotencyKey": "idempotency-api-stale-checkout", }, ) self.assertEqual(stale.status_code, 403, stale.text) self.assertEqual(stale.json()["code"], "reauthentication_required") now = datetime.now(timezone.utc) with Session(self.engine) as session, session.begin(): session.add( Subscription( id="subscription.test.billing_portal", license_id=license_id, provider="stripe", provider_subscription_id="sub_test_billing_portal", provider_customer_id="cus_test_billing_portal", status="active", current_period_start=now, current_period_end=now + timedelta(days=30), cancel_at_period_end=False, grace_ends_at=None, provider_event_created_at=now, ) ) portal_body = { "idempotencyKey": "idempotency-api-owned-portal", "correlationId": "correlation.test.api_owned_portal", } portal = await self._request( "POST", f"/v1/me/licenses/{license_id}/billing-portal-sessions", headers={"Authorization": "Bearer customer-token"}, json=portal_body, ) self.assertEqual(portal.status_code, 200, portal.text) self.assertEqual( provider.portal_requests[0]["customer_id"], "cus_test_billing_portal", ) listing = await self._request( "GET", "/v1/me/licenses", headers={"Authorization": "Bearer customer-token"}, ) self.assertEqual(listing.status_code, 200, listing.text) self.assertTrue(listing.json()["commerce"]["checkoutAvailable"]) listed_license = next( value for value in listing.json()["licenses"] if value["licenseId"] == license_id ) self.assertTrue(listed_license["canManageBilling"]) self.assertEqual(listed_license["subscriptions"][0]["status"], "active") self.assertIsNone(listed_license["subscriptions"][0]["billingHold"]) async def test_billing_sessions_remain_disabled_by_default(self) -> None: response = await self._request( "POST", "/v1/me/licenses/license.missing/checkout-sessions", headers={"Authorization": "Bearer customer-token"}, json={ "skuId": "subscription.ai_assist", "idempotencyKey": "idempotency-api-disabled-checkout", "correlationId": "correlation.test.api_disabled_checkout", }, ) self.assertEqual(response.status_code, 503, response.text) self.assertEqual(response.json()["code"], "billing_unavailable") listing = await self._request( "GET", "/v1/me/licenses", headers={"Authorization": "Bearer customer-token"}, ) self.assertEqual(listing.status_code, 200, listing.text) self.assertFalse(listing.json()["commerce"]["checkoutAvailable"]) self.assertFalse(listing.json()["commerce"]["billingPortalAvailable"]) async def test_privacy_export_and_reviewed_deletion_are_recent_and_no_store( self, ) -> None: stale_export = await self._request( "POST", "/v1/me/privacy/export", headers={"Authorization": "Bearer stale-customer-token"}, json={"correlationId": "correlation.test.api_privacy_stale"}, ) self.assertEqual(stale_export.status_code, 403, stale_export.text) self.assertEqual( stale_export.json()["code"], "reauthentication_required", ) exported = await self._request( "POST", "/v1/me/privacy/export", headers={"Authorization": "Bearer customer-token"}, json={"correlationId": "correlation.test.api_privacy_export"}, ) self.assertEqual(exported.status_code, 200, exported.text) self.assertEqual( exported.headers["cache-control"], "no-store, max-age=0", ) self.assertEqual( exported.json()["account"]["verifiedEmail"], "customer@example.test", ) original_user_id = exported.json()["account"]["userId"] missing_acknowledgement = await self._request( "POST", "/v1/me/privacy/deletion-requests", headers={"Authorization": "Bearer customer-token"}, json={ "acknowledgePermanentAccountClosure": False, "idempotencyKey": "idempotency-api-privacy-missing-ack", "correlationId": "correlation.test.api_privacy_missing_ack", }, ) self.assertEqual(missing_acknowledgement.status_code, 422) requested = await self._request( "POST", "/v1/me/privacy/deletion-requests", headers={"Authorization": "Bearer customer-token"}, json={ "acknowledgePermanentAccountClosure": True, "idempotencyKey": "idempotency-api-privacy-request", "correlationId": "correlation.test.api_privacy_request", }, ) self.assertEqual(requested.status_code, 200, requested.text) request_id = requested.json()["privacyRequestId"] self.assertTrue(requested.json()["canComplete"]) admin_listing = await self._request( "GET", "/v1/admin/privacy/deletion-requests", headers={"Authorization": "Bearer admin-token"}, ) self.assertEqual(admin_listing.status_code, 200, admin_listing.text) self.assertEqual( admin_listing.json()["requests"][0]["privacyRequestId"], request_id, ) preview = await self._request( "POST", f"/v1/admin/privacy/deletion-requests/{request_id}/preview", headers={"Authorization": "Bearer admin-token"}, json={ "reasonCode": "customer_request", "note": "Authenticated customer request", "correlationId": "correlation.test.api_privacy_preview", }, ) self.assertEqual(preview.status_code, 200, preview.text) self.assertTrue(preview.json()["canComplete"]) execution_body = { "reasonCode": "customer_request", "note": "Authenticated customer request", "expectedStateDigest": preview.json()["stateDigest"], "idempotencyKey": "idempotency-api-privacy-execute", "correlationId": "correlation.test.api_privacy_execute", } executed = await self._request( "POST", f"/v1/admin/privacy/deletion-requests/{request_id}/execute", headers={"Authorization": "Bearer admin-token"}, json=execution_body, ) self.assertEqual(executed.status_code, 200, executed.text) self.assertTrue(executed.json()["executed"]) self.assertFalse(executed.json()["customerRightsDeleted"]) replay = await self._request( "POST", f"/v1/admin/privacy/deletion-requests/{request_id}/execute", headers={"Authorization": "Bearer admin-token"}, json={ **execution_body, "correlationId": "correlation.test.api_privacy_execute_replay", }, ) self.assertEqual(replay.status_code, 200, replay.text) self.assertTrue(replay.json()["idempotentReplay"]) with Session(self.engine) as session: erased = session.get(User, original_user_id) self.assertEqual(erased.status, "privacy_erased") self.assertIsNone(erased.verified_email) recreated = await self._request( "GET", "/v1/me/privacy/deletion-requests", headers={"Authorization": "Bearer customer-token"}, ) self.assertEqual(recreated.status_code, 200, recreated.text) self.assertEqual(recreated.json(), {"requests": []}) self.assertEqual( recreated.headers["cache-control"], "no-store, max-age=0", ) with Session(self.engine) as session: recreated_user = session.scalar( select(User).where( User.external_issuer == "https://identity.test", User.external_subject == "customer-subject", ) ) self.assertIsNotNone(recreated_user) self.assertNotEqual(recreated_user.id, original_user_id) async def test_admin_audit_search_export_is_scoped_paginated_and_allowlisted( self, ) -> None: await self._admin_serial("edition.personal") await self._admin_serial("addon.advanced_analysis") denied = await self._request( "POST", "/v1/admin/audit-events/search", headers={"Authorization": "Bearer support-customer-token"}, json={"action": "serial.batch_generated", "limit": 1}, ) self.assertEqual(denied.status_code, 403, denied.text) first = await self._request( "POST", "/v1/admin/audit-events/search", headers={"Authorization": "Bearer admin-token"}, json={"action": "serial.batch_generated", "limit": 1}, ) self.assertEqual(first.status_code, 200, first.text) self.assertTrue(first.json()["hasMore"]) self.assertEqual(len(first.json()["events"]), 1) event = first.json()["events"][0] self.assertEqual(event["action"], "serial.batch_generated") self.assertEqual(set(event["metadata"]), {"quantity", "skuId"}) self.assertNotIn("sourceAddressDigest", event) second = await self._request( "POST", "/v1/admin/audit-events/search", headers={"Authorization": "Bearer admin-token"}, json={ "action": "serial.batch_generated", "limit": 1, "cursor": first.json()["nextCursor"], }, ) self.assertEqual(second.status_code, 200, second.text) self.assertNotEqual( second.json()["events"][0]["eventId"], event["eventId"], ) exported = await self._request( "POST", "/v1/admin/audit-events/export", headers={"Authorization": "Bearer admin-token"}, json={ "action": "serial.batch_generated", "limit": 10, "correlationId": "correlation.test.audit_export", }, ) self.assertEqual(exported.status_code, 200, exported.text) document = exported.json() expected_digest = document.pop("sha256") self.assertEqual( hashlib.sha256(canonicalize_json(document)).hexdigest(), expected_digest, ) self.assertEqual(document["schema"], "apolon.licensing.audit-export") self.assertEqual(len(document["events"]), 2) with Session(self.engine) as session: self.assertIsNotNone( session.scalar( select(AuditEvent).where(AuditEvent.action == "audit.exported") ) ) if __name__ == "__main__": unittest.main()