"""Transactional licensing authority regression tests.""" from __future__ import annotations from dataclasses import replace from datetime import datetime, timedelta, timezone import unittest from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from sqlalchemy import create_engine, func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from sqlalchemy.pool import StaticPool from licensing_shared.catalog import LicensingCatalog, load_builtin_catalog from licensing_shared.constants import TRIAL_MAXIMUM_TOTAL_DURATION_HOURS from licensing_shared.models import EntitlementSnapshot from licensing_shared.offline import device_public_key_thumbprint from licensing_shared.verifier import LicenseVerifier, PublicKeyRecord, PublicKeyRing from licensing_server.app.database import Base from licensing_server.app.errors import ServerErrorCode, ServerLicensingError from licensing_server.app.models import ( Activation, AuditEvent, Device, DeviceInstallation, Grant, Lease, License, Membership, Organization, OrganizationSeat, Serial, SerialBatch, Trial, User, ) from licensing_server.app.security import Ed25519SnapshotSigner from licensing_server.app.serial_administration import SerialAdministrationService from licensing_server.app.services import DeviceEnrollment, LicensingService TEST_NOW = datetime(2026, 8, 15, 12, 0, tzinfo=timezone.utc) TEST_SERIAL_PEPPER = b"serial-test-pepper-material-32-bytes-minimum" TEST_FINGERPRINT_PEPPER = b"fingerprint-test-pepper-material-32-bytes" TEST_ADMIN_ID = "user.test.admin" TEST_CUSTOMER_ID = "user.test.customer" class LicensingServerServiceTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine( "sqlite+pysqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) Base.metadata.create_all(self.engine) self.catalog = load_builtin_catalog() self.private_key = Ed25519PrivateKey.generate() self.signer = Ed25519SnapshotSigner("license.test.server", self.private_key) self.now = TEST_NOW with Session(self.engine) as session, session.begin(): session.add_all( ( User( id=TEST_ADMIN_ID, external_issuer="https://identity.test", external_subject="admin-subject", verified_email="admin@example.test", status="active", is_server_admin=True, ), User( id=TEST_CUSTOMER_ID, external_issuer="https://identity.test", external_subject="customer-subject", verified_email="customer@example.test", status="active", is_server_admin=False, ), ) ) def tearDown(self) -> None: self.engine.dispose() def _service(self, session: Session) -> LicensingService: return LicensingService( session, self.catalog, self.signer, TEST_SERIAL_PEPPER, TEST_FINGERPRINT_PEPPER, now_factory=lambda: self.now, ) def _enrollment( self, suffix: str, *, system_uuid: str, baseboard_serial: str | None = None, ) -> DeviceEnrollment: private_key = ec.generate_private_key(ec.SECP256R1()) public_key_der = private_key.public_key().public_bytes( encoding=serialization.Encoding.DER, format=serialization.PublicFormat.SubjectPublicKeyInfo, ) return DeviceEnrollment( installation_id=f"installation.test.{suffix}", public_key_der=public_key_der, key_thumbprint=device_public_key_thumbprint(public_key_der), key_provider="software.test", friendly_name=f"Test computer {suffix}", evidence={ "system_uuid": system_uuid, "baseboard_serial": ( f"board-{system_uuid}" if baseboard_serial is None else baseboard_serial ), }, ) def _generate_serials(self, sku_id: str, quantity: int) -> tuple[str, ...]: with Session(self.engine) as session: result = self._service(session).generate_serial_batch( sku_id, quantity, TEST_ADMIN_ID, "Automated test issuance", "correlation.test.generation", ) return result.serials def test_serial_generation_stores_only_keyed_digest_and_requires_admin(self) -> None: serials = self._generate_serials("edition.professional", 3) self.assertEqual(len(serials), 3) self.assertEqual(len(set(serials)), 3) with Session(self.engine) as session: rows = tuple(session.scalars(select(Serial))) self.assertEqual(len(rows), 3) for row, plaintext in zip(rows, serials): self.assertEqual(len(row.secret_digest), 32) self.assertNotIn(plaintext.encode("ascii"), row.secret_digest) self.assertFalse(hasattr(row, "plaintext_serial")) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as context: self._service(session).generate_serial_batch( "edition.professional", 1, TEST_CUSTOMER_ID, "Unauthorized issuance", "correlation.test.denied", ) self.assertEqual(context.exception.code, ServerErrorCode.AUTHORIZATION_DENIED) def test_serial_export_failure_rolls_back_the_batch(self) -> None: def fail_export(_result) -> None: raise OSError("simulated secure delivery failure") with Session(self.engine) as session: with self.assertRaises(OSError): self._service(session).generate_serial_batch( "edition.personal", 1, TEST_ADMIN_ID, "Export failure test", "correlation.test.export_failure", plaintext_exporter=fail_export, ) with Session(self.engine) as session: self.assertEqual(session.scalar(select(func.count()).select_from(Serial)), 0) def test_surface_profile_is_signed_into_license_and_survives_archival(self) -> None: selected_analyses = ("fft_spectrum", "welch_psd") selected_tabs = ("acquisition", "analysis") with Session(self.engine) as session: profile = self._service(session).create_surface_profile( "Focused spectrum", "Spectrum-only launch profile", selected_analyses, selected_tabs, TEST_ADMIN_ID, "Focused launch cohort", "correlation.test.surface_profile_create", ) profile_id = str(profile["profileId"]) with Session(self.engine) as session: generated = self._service(session).generate_serial_batch( "edition.personal", 1, TEST_ADMIN_ID, "Focused launch serial", "correlation.test.surface_serial", surface_profile_id=profile_id, ) enrollment = self._enrollment( "surface_profile", system_uuid="surface-profile-device", ) with Session(self.engine) as session: redeemed = self._service(session).redeem_serial( generated.serials[0], enrollment, 1, "idempotency-surface-profile", "correlation.test.surface_redeem", actor_user_id=TEST_CUSTOMER_ID, ) snapshot = EntitlementSnapshot.from_payload_mapping(redeemed.document.payload) self.assertIsNotNone(snapshot.surface_policy) assert snapshot.surface_policy is not None self.assertEqual(snapshot.surface_policy.profile_id, profile_id) self.assertEqual(snapshot.surface_policy.analysis_ids, selected_analyses) self.assertEqual(snapshot.surface_policy.main_tab_ids, selected_tabs) with Session(self.engine) as session: archived = self._service(session).archive_surface_profile( profile_id, TEST_ADMIN_ID, "Profile retired after issuance", "correlation.test.surface_profile_archive", ) self.assertEqual(archived["status"], "archived") with Session(self.engine) as session: refreshed = self._service(session).refresh_activation( redeemed.license_id, enrollment.installation_id, enrollment.public_key_der, enrollment.key_thumbprint, 1, "correlation.test.surface_refresh", evidence=enrollment.evidence, ) refreshed_snapshot = EntitlementSnapshot.from_payload_mapping( refreshed.document.payload ) self.assertEqual(refreshed_snapshot.surface_policy, snapshot.surface_policy) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as context: self._service(session).generate_serial_batch( "edition.personal", 1, TEST_ADMIN_ID, "Archived profile must not issue", "correlation.test.archived_surface_serial", surface_profile_id=profile_id, ) self.assertEqual( context.exception.code, ServerErrorCode.SURFACE_PROFILE_NOT_FOUND, ) def test_trial_serials_are_hour_bounded_and_cannot_restart_or_extend_past_ceiling( self, ) -> None: with Session(self.engine) as session: one_hour = self._service(session).generate_serial_batch( "trial.full", 2, TEST_ADMIN_ID, "One-hour QA trial", "correlation.test.trial_serial_one_hour", trial_duration_hours=1, ) enrollment = self._enrollment( "trial_serial", system_uuid="trial-serial-device", ) with Session(self.engine) as session: redeemed = self._service(session).redeem_serial( one_hour.serials[0], enrollment, 1, "idempotency-trial-serial-first", "correlation.test.trial_serial_first", actor_user_id=TEST_CUSTOMER_ID, ) snapshot = EntitlementSnapshot.from_payload_mapping(redeemed.document.payload) self.assertEqual(snapshot.lease_expires_at, self.now + timedelta(hours=1)) self.assertIsNotNone(snapshot.trial_policy) assert snapshot.trial_policy is not None self.assertEqual(snapshot.trial_policy.original_started_at, self.now) self.assertEqual(snapshot.trial_policy.final_expires_at, self.now + timedelta(hours=1)) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as restarted: self._service(session).redeem_serial( one_hour.serials[1], self._enrollment( "trial_serial_second", system_uuid="trial-serial-second-device", ), 1, "idempotency-trial-serial-second", "correlation.test.trial_serial_second", actor_user_id=TEST_CUSTOMER_ID, ) self.assertEqual(restarted.exception.code, ServerErrorCode.TRIAL_NOT_ELIGIBLE) with Session(self.engine) as session: with self.assertRaises(ValueError): self._service(session).generate_serial_batch( "trial.full", 1, TEST_ADMIN_ID, "Invalid overlong trial", "correlation.test.trial_serial_overlong", trial_duration_hours=TRIAL_MAXIMUM_TOTAL_DURATION_HOURS + 1, ) with Session(self.engine) as session: maximum = self._service(session).generate_serial_batch( "trial.full", 1, TEST_ADMIN_ID, "Maximum bounded trial", "correlation.test.trial_serial_maximum", trial_duration_hours=TRIAL_MAXIMUM_TOTAL_DURATION_HOURS, ) maximum_enrollment = self._enrollment( "trial_serial_maximum", system_uuid="trial-serial-maximum-device", ) with Session(self.engine) as session: maximum_result = self._service(session).redeem_serial( maximum.serials[0], maximum_enrollment, 1, "idempotency-trial-serial-maximum", "correlation.test.trial_serial_maximum_redeem", ) maximum_snapshot = EntitlementSnapshot.from_payload_mapping( maximum_result.document.payload ) self.assertEqual(maximum_snapshot.lease_expires_at, self.now + timedelta(days=14)) with Session(self.engine) as session: trial_id = session.scalar( select(Trial.id).where(Trial.license_id == maximum_result.license_id) ) self.assertIsNotNone(trial_id) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as extension: self._service(session).preview_trial_extension( str(trial_id), TEST_ADMIN_ID, "evaluation_extension", "correlation.test.trial_serial_extension_denied", ) self.assertEqual(extension.exception.code, ServerErrorCode.TRIAL_NOT_ELIGIBLE) def test_serial_batch_revocation_is_guarded_idempotent_and_preserves_redemptions( self, ) -> None: with Session(self.engine) as session: generation = self._service(session).generate_serial_batch( "edition.professional", 3, TEST_ADMIN_ID, "Revocation regression batch", "correlation.test.batch_revocation_generation", ) first_redemption = self._enrollment("batch-first", system_uuid="batch-device-a") with Session(self.engine) as session: first_result = self._service(session).redeem_serial( generation.serials[0], first_redemption, 1, "idempotency-batch-first-redemption", "correlation.test.batch_first_redemption", ) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as denied: SerialAdministrationService( session, now_factory=lambda: self.now, ).preview_batch_revocation( generation.batch_id, TEST_CUSTOMER_ID, "serial_exposure", "correlation.test.batch_revocation_denied", ) self.assertEqual(denied.exception.code, ServerErrorCode.AUTHORIZATION_DENIED) with Session(self.engine) as session: preview = SerialAdministrationService( session, now_factory=lambda: self.now, ).preview_batch_revocation( generation.batch_id, TEST_ADMIN_ID, "serial_exposure", "correlation.test.batch_revocation_preview", note="Leaked fulfillment export", ) self.assertEqual(preview["activeSerialCount"], 2) self.assertEqual(preview["redeemedSerialCount"], 1) self.assertEqual(preview["wouldRevokeSerialCount"], 2) self.assertTrue(preview["canExecute"]) second_redemption = self._enrollment("batch-second", system_uuid="batch-device-b") with Session(self.engine) as session: second_result = self._service(session).redeem_serial( generation.serials[1], second_redemption, 1, "idempotency-batch-second-redemption", "correlation.test.batch_second_redemption", ) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as stale: SerialAdministrationService( session, now_factory=lambda: self.now, ).revoke_batch( generation.batch_id, TEST_ADMIN_ID, "serial_exposure", str(preview["stateDigest"]), "idempotency-batch-revocation", "correlation.test.batch_revocation_execute", note="Leaked fulfillment export", ) self.assertEqual(stale.exception.code, ServerErrorCode.CONFLICT) with Session(self.engine) as session: current = SerialAdministrationService( session, now_factory=lambda: self.now, ).preview_batch_revocation( generation.batch_id, TEST_ADMIN_ID, "serial_exposure", "correlation.test.batch_revocation_preview_current", note="Leaked fulfillment export", ) self.assertEqual(current["activeSerialCount"], 1) self.assertEqual(current["redeemedSerialCount"], 2) with Session(self.engine) as session: executed = SerialAdministrationService( session, now_factory=lambda: self.now, ).revoke_batch( generation.batch_id, TEST_ADMIN_ID, "serial_exposure", str(current["stateDigest"]), "idempotency-batch-revocation", "correlation.test.batch_revocation_execute", note="Leaked fulfillment export", ) self.assertTrue(executed["executed"]) self.assertFalse(executed["idempotentReplay"]) self.assertEqual(executed["newlyRevokedSerialCount"], 1) self.assertEqual(executed["preservedRedeemedLicenseCount"], 2) with Session(self.engine) as session: replay = SerialAdministrationService( session, now_factory=lambda: self.now, ).revoke_batch( generation.batch_id, TEST_ADMIN_ID, "serial_exposure", str(current["stateDigest"]), "idempotency-batch-revocation", "correlation.test.batch_revocation_execute", note="Leaked fulfillment export", ) self.assertTrue(replay["idempotentReplay"]) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as reused: SerialAdministrationService( session, now_factory=lambda: self.now, ).revoke_batch( generation.batch_id, TEST_ADMIN_ID, "support_correction", str(current["stateDigest"]), "idempotency-batch-revocation", "correlation.test.batch_revocation_execute", note="Changed request", ) self.assertEqual(reused.exception.code, ServerErrorCode.IDEMPOTENCY_CONFLICT) with Session(self.engine) as session: claimed_license_id = self._service(session).claim_redeemed_serial( generation.serials[0], TEST_CUSTOMER_ID, "correlation.test.batch_redeemed_recovery", ) self.assertEqual(claimed_license_id, first_result.license_id) with Session(self.engine) as session: batch = session.get(SerialBatch, generation.batch_id) self.assertIsNotNone(batch) assert batch is not None self.assertEqual(batch.status, "revoked") self.assertEqual(batch.revoked_by_user_id, TEST_ADMIN_ID) self.assertEqual(batch.revocation_reason, "serial_exposure") serial_statuses = tuple( session.scalars( select(Serial.status) .where(Serial.batch_id == generation.batch_id) .order_by(Serial.status) ) ) self.assertEqual(serial_statuses, ("redeemed", "redeemed", "revoked")) self.assertEqual( session.scalar( select(func.count()) .select_from(Grant) .where(Grant.status == "active") ), 2, ) self.assertEqual(session.get(License, first_result.license_id).status, "active") self.assertEqual(session.get(License, second_result.license_id).status, "active") audit = session.scalar( select(AuditEvent).where(AuditEvent.action == "serial.batch_revoked") ) self.assertIsNotNone(audit) assert audit is not None self.assertNotIn(generation.serials[2], str(audit.metadata_json)) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as revoked: self._service(session).redeem_serial( generation.serials[2], self._enrollment("batch-revoked", system_uuid="batch-device-c"), 1, "idempotency-batch-revoked-redemption", "correlation.test.batch_revoked_redemption", ) self.assertEqual(revoked.exception.code, ServerErrorCode.SERIAL_INVALID) def test_serial_redemption_is_signed_atomic_and_idempotent(self) -> None: serial_value = self._generate_serials("edition.professional", 1)[0] enrollment = self._enrollment("primary", system_uuid="device-a") with Session(self.engine) as session: result = self._service(session).redeem_serial( serial_value, enrollment, 1, "idempotency-test-redemption", "correlation.test.redemption", ) public_key = self.private_key.public_key().public_bytes( encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw, ) verifier = LicenseVerifier( PublicKeyRing( [PublicKeyRecord(key_id=self.signer.key_id, public_key_bytes=public_key)] ) ) verified = verifier.verify(result.document.to_mapping(), at=self.now) self.assertIn("right.commercial_use", verified.snapshot.entitlements) self.assertEqual(verified.snapshot.device_id, result.device_id) self.assertEqual( verified.snapshot.refresh_after, self.now + timedelta(days=30), ) self.assertEqual( verified.snapshot.lease_expires_at, self.now + timedelta(days=365), ) with Session(self.engine) as session: replay = self._service(session).redeem_serial( serial_value, enrollment, 1, "idempotency-test-redemption", "correlation.test.redemption_retry", ) self.assertTrue(replay.idempotent_replay) self.assertEqual(replay.document.to_mapping(), result.document.to_mapping()) with Session(self.engine) as session: self.assertEqual(session.scalar(select(func.count()).select_from(License)), 1) self.assertEqual(session.scalar(select(func.count()).select_from(Grant)), 1) self.assertEqual(session.scalar(select(func.count()).select_from(Activation)), 1) self.assertEqual(session.scalar(select(func.count()).select_from(Lease)), 1) serial_row = session.scalar(select(Serial)) self.assertEqual(serial_row.status, "redeemed") def test_higher_edition_serial_upgrades_the_existing_license_additively(self) -> None: personal_serial = self._generate_serials("edition.personal", 1)[0] professional_serial = self._generate_serials("edition.professional", 1)[0] second_personal_serial = self._generate_serials("edition.personal", 1)[0] enrollment = self._enrollment("upgrade", system_uuid="device-upgrade") with Session(self.engine) as session: personal = self._service(session).redeem_serial( personal_serial, enrollment, 1, "idempotency-upgrade-personal", "correlation.test.upgrade_personal", ) with Session(self.engine) as session: upgraded = self._service(session).redeem_serial( professional_serial, enrollment, 1, "idempotency-upgrade-professional", "correlation.test.upgrade_professional", target_license_id=personal.license_id, ) self.assertEqual(upgraded.license_id, personal.license_id) entitlements = upgraded.document.payload["entitlements"] self.assertIn("right.personal_use", entitlements) self.assertIn("right.commercial_use", entitlements) self.assertIn("feature.analysis.advanced", entitlements) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as context: self._service(session).redeem_serial( second_personal_serial, enrollment, 1, "idempotency-upgrade-downgrade", "correlation.test.upgrade_downgrade", target_license_id=personal.license_id, ) self.assertEqual(context.exception.code, ServerErrorCode.SKU_NOT_AVAILABLE) with Session(self.engine) as session: self.assertEqual( session.scalar(select(func.count()).select_from(Grant)), 2, ) active_serials = session.scalar( select(func.count()).select_from(Serial).where(Serial.status == "active") ) self.assertEqual(active_serials, 1) def test_idempotency_key_cannot_be_reused_for_another_serial(self) -> None: first_serial, second_serial = self._generate_serials("edition.personal", 2) enrollment = self._enrollment("primary", system_uuid="device-a") with Session(self.engine) as session: self._service(session).redeem_serial( first_serial, enrollment, 1, "idempotency-test-conflict", "correlation.test.first", ) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as context: self._service(session).redeem_serial( second_serial, enrollment, 1, "idempotency-test-conflict", "correlation.test.second", ) self.assertEqual(context.exception.code, ServerErrorCode.IDEMPOTENCY_CONFLICT) with Session(self.engine) as session: second_row = session.scalar( select(Serial).where(Serial.status == "active") ) self.assertIsNotNone(second_row) def test_trial_recovers_same_device_and_enforces_two_device_capacity(self) -> None: first = self._enrollment("trial_a", system_uuid="device-a") recovered_install = self._enrollment("trial_a_reinstall", system_uuid="device-a") second = self._enrollment("trial_b", system_uuid="device-b") third = self._enrollment("trial_c", system_uuid="device-c") with Session(self.engine) as session: first_result = self._service(session).start_trial( first, 1, "idempotency-trial-a", "correlation.test.trial_a", actor_user_id=TEST_CUSTOMER_ID, ) first_snapshot = EntitlementSnapshot.from_payload_mapping( first_result.document.payload ) self.assertEqual(first_snapshot.refresh_after, self.now + timedelta(days=1)) self.assertEqual( first_snapshot.lease_expires_at, self.now + timedelta(days=7), ) with Session(self.engine) as session: recovered = self._service(session).start_trial( recovered_install, 1, "idempotency-trial-a-reinstall", "correlation.test.trial_a_reinstall", actor_user_id=TEST_CUSTOMER_ID, ) self.assertTrue(recovered.recovered_existing_device) self.assertEqual(recovered.device_id, first_result.device_id) with Session(self.engine) as session: self._service(session).start_trial( second, 1, "idempotency-trial-b", "correlation.test.trial_b", actor_user_id=TEST_CUSTOMER_ID, ) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as context: self._service(session).start_trial( third, 1, "idempotency-trial-c", "correlation.test.trial_c", actor_user_id=TEST_CUSTOMER_ID, ) self.assertEqual(context.exception.code, ServerErrorCode.DEVICE_LIMIT_REACHED) with Session(self.engine) as session: self.assertEqual(session.scalar(select(func.count()).select_from(Trial)), 1) self.assertEqual( session.scalar( select(func.count()).select_from(Activation).where( Activation.status == "active" ) ), 2, ) self.assertEqual(session.scalar(select(func.count()).select_from(Device)), 2) def test_offline_refresh_reminders_are_catalog_bounded(self) -> None: serial_value = self._generate_serials("edition.personal", 1)[0] paid_enrollment = self._enrollment( "offline_paid", system_uuid="offline-paid-device", ) with Session(self.engine) as session: paid = self._service(session).redeem_serial( serial_value, paid_enrollment, 1, "idempotency-offline-paid", "correlation.test.offline_paid", actor_user_id=TEST_CUSTOMER_ID, offline_certificate=True, offline_request_nonce="offline-paid-request-nonce", ) paid_snapshot = EntitlementSnapshot.from_payload_mapping(paid.document.payload) self.assertEqual( paid_snapshot.lease_expires_at, self.now + timedelta(days=1095), ) self.assertEqual( paid_snapshot.refresh_after, paid_snapshot.lease_expires_at - timedelta(days=90), ) trial_enrollment = self._enrollment( "offline_trial", system_uuid="offline-trial-device", ) with Session(self.engine) as session: trial = self._service(session).start_trial( trial_enrollment, 1, "idempotency-offline-trial", "correlation.test.offline_trial", offline_certificate=True, offline_request_nonce="offline-trial-request-nonce", ) trial_snapshot = EntitlementSnapshot.from_payload_mapping(trial.document.payload) self.assertEqual( trial_snapshot.lease_expires_at, self.now + timedelta(days=7), ) self.assertEqual( trial_snapshot.refresh_after, trial_snapshot.lease_expires_at - timedelta(days=1), ) with Session(self.engine) as session, session.begin(): session.add( Grant( id="grant.test.offline_paid_trial", license_id=paid.license_id, sku_id="trial.full", source_type="support", source_reference="support.test.offline_paid_trial", catalog_revision=self.catalog.revision, starts_at=self.now, ends_at=self.now + timedelta(days=14), status="active", metadata_json={}, ) ) with Session(self.engine) as session: combined = self._service(session).activate_owned_license( paid.license_id, paid_enrollment, 1, "idempotency-offline-paid-trial-renewal", "correlation.test.offline_paid_trial_renewal", TEST_CUSTOMER_ID, offline_certificate=True, offline_request_nonce="offline-paid-trial-request-nonce", expected_device_id=paid.device_id, ) combined_snapshot = EntitlementSnapshot.from_payload_mapping( combined.document.payload ) self.assertEqual( combined_snapshot.lease_expires_at, self.now + timedelta(days=14), ) self.assertEqual( combined_snapshot.refresh_after, self.now + timedelta(days=13), ) def test_additive_subscription_uses_most_restrictive_refresh_cadence( self, ) -> None: subscription_sku = replace( self.catalog.skus["subscription.ai_assist"], active=True, ) active_catalog = LicensingCatalog( product_id=self.catalog.product_id, revision=self.catalog.revision, entitlements=self.catalog.entitlements, device_policies=self.catalog.device_policies, skus={ **self.catalog.skus, subscription_sku.sku_id: subscription_sku, }, schema_version=self.catalog.schema_version, ) serial_value = self._generate_serials("edition.professional", 1)[0] enrollment = self._enrollment( "subscription_refresh", system_uuid="subscription-refresh-device", ) with Session(self.engine) as session: service = LicensingService( session, active_catalog, self.signer, TEST_SERIAL_PEPPER, TEST_FINGERPRINT_PEPPER, now_factory=lambda: self.now, ) redeemed = service.redeem_serial( serial_value, enrollment, 1, "idempotency-subscription-refresh-base", "correlation.test.subscription_refresh_base", actor_user_id=TEST_CUSTOMER_ID, ) with Session(self.engine) as session, session.begin(): session.add( Grant( id="grant.test.subscription_refresh", license_id=redeemed.license_id, sku_id=subscription_sku.sku_id, source_type="subscription", source_reference="subscription.test.refresh", catalog_revision=active_catalog.revision, starts_at=self.now, ends_at=self.now + timedelta(days=30), status="active", metadata_json={"subscriptionState": "active"}, ) ) with Session(self.engine) as session: refreshed = LicensingService( session, active_catalog, self.signer, TEST_SERIAL_PEPPER, TEST_FINGERPRINT_PEPPER, now_factory=lambda: self.now, ).refresh_activation( redeemed.license_id, enrollment.installation_id, enrollment.public_key_der, enrollment.key_thumbprint, 1, "correlation.test.subscription_refresh", evidence=enrollment.evidence, ) snapshot = EntitlementSnapshot.from_payload_mapping(refreshed.document.payload) self.assertIn("service.ai.assist", snapshot.entitlements) self.assertEqual(snapshot.refresh_after, self.now + timedelta(hours=24)) self.assertEqual(snapshot.lease_expires_at, self.now + timedelta(days=14)) def test_anonymous_reinstall_does_not_restart_trial(self) -> None: first = self._enrollment("anonymous_a", system_uuid="anonymous-device") reinstall = self._enrollment( "anonymous_reinstall", system_uuid="anonymous-device", baseboard_serial="replacement-board-anonymous-device", ) with Session(self.engine) as session: original = self._service(session).start_trial( first, 1, "idempotency-anonymous-a", "correlation.test.anonymous_a", ) self.now += timedelta(days=15) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as context: self._service(session).start_trial( reinstall, 1, "idempotency-anonymous-reinstall", "correlation.test.anonymous_reinstall", ) self.assertEqual(context.exception.code, ServerErrorCode.TRIAL_NOT_ELIGIBLE) with Session(self.engine) as session: self.assertEqual(session.scalar(select(func.count()).select_from(Trial)), 1) self.assertEqual(session.scalar(select(func.count()).select_from(License)), 1) self.assertIsNotNone(original.license_id) def test_anonymous_active_trial_reinstall_recovers_canonical_device(self) -> None: first = self._enrollment("anonymous_active_a", system_uuid="anonymous-active") reinstall = self._enrollment( "anonymous_active_reinstall", system_uuid="anonymous-active", baseboard_serial="replacement-board-anonymous-active", ) with Session(self.engine) as session: original = self._service(session).start_trial( first, 1, "idempotency-anonymous-active-a", "correlation.test.anonymous_active_a", ) with Session(self.engine) as session: recovered = self._service(session).start_trial( reinstall, 1, "idempotency-anonymous-active-reinstall", "correlation.test.anonymous_active_reinstall", ) self.assertTrue(recovered.recovered_existing_device) self.assertEqual(recovered.device_id, original.device_id) 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)), 2, ) self.assertEqual( session.scalar(select(func.count()).select_from(Activation)), 1, ) def test_existing_installation_rejects_materially_different_hardware(self) -> None: original = self._enrollment("clone_source", system_uuid="clone-source") with Session(self.engine) as session: started = self._service(session).start_trial( original, 1, "idempotency-clone-source", "correlation.test.clone_source", ) cloned = DeviceEnrollment( installation_id=original.installation_id, public_key_der=original.public_key_der, key_thumbprint=original.key_thumbprint, key_provider=original.key_provider, friendly_name=original.friendly_name, evidence={ "system_uuid": "clone-destination", "baseboard_serial": "board-clone-destination", }, ) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as refresh_context: self._service(session).refresh_activation( started.license_id, original.installation_id, original.public_key_der, original.key_thumbprint, 1, "correlation.test.clone_refresh", evidence=cloned.evidence, ) self.assertEqual(refresh_context.exception.code, ServerErrorCode.CONFLICT) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as context: self._service(session).start_trial( cloned, 1, "idempotency-clone-destination", "correlation.test.clone_destination", ) self.assertEqual(context.exception.code, ServerErrorCode.CONFLICT) def test_redeemed_anonymous_serial_can_be_claimed_once_by_account(self) -> None: serial_value = self._generate_serials("edition.personal", 1)[0] enrollment = self._enrollment("claim", system_uuid="claim-device") with Session(self.engine) as session: redeemed = self._service(session).redeem_serial( serial_value, enrollment, 1, "idempotency-claim-redemption", "correlation.test.claim_redemption", ) with Session(self.engine) as session: license_id = self._service(session).claim_redeemed_serial( serial_value, TEST_CUSTOMER_ID, "correlation.test.claim", ) self.assertEqual(license_id, redeemed.license_id) with Session(self.engine) as session: license_row = session.get(License, license_id) device = session.get(Device, redeemed.device_id) self.assertEqual(license_row.owner_user_id, TEST_CUSTOMER_ID) self.assertIsNone(license_row.anonymous_subject_digest) self.assertEqual(device.owner_user_id, TEST_CUSTOMER_ID) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as context: self._service(session).claim_redeemed_serial( serial_value, TEST_ADMIN_ID, "correlation.test.claim_denied", ) self.assertEqual(context.exception.code, ServerErrorCode.AUTHORIZATION_DENIED) def test_owned_license_activation_is_authorized_signed_and_idempotent(self) -> None: serial_value = self._generate_serials("edition.professional", 1)[0] first_enrollment = self._enrollment("account_first", system_uuid="account-device-a") second_enrollment = self._enrollment("account_second", system_uuid="account-device-b") denied_enrollment = self._enrollment("account_denied", system_uuid="account-device-c") with Session(self.engine) as session: redeemed = self._service(session).redeem_serial( serial_value, first_enrollment, 1, "idempotency-account-redemption", "correlation.test.account_redemption", actor_user_id=TEST_CUSTOMER_ID, ) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as context: self._service(session).activate_owned_license( redeemed.license_id, denied_enrollment, 1, "idempotency-account-denied", "correlation.test.account_denied", TEST_ADMIN_ID, ) self.assertEqual(context.exception.code, ServerErrorCode.AUTHORIZATION_DENIED) with Session(self.engine) as session: activated = self._service(session).activate_owned_license( redeemed.license_id, second_enrollment, 1, "idempotency-account-activation", "correlation.test.account_activation", TEST_CUSTOMER_ID, ) with Session(self.engine) as session: replay = self._service(session).activate_owned_license( redeemed.license_id, second_enrollment, 1, "idempotency-account-activation", "correlation.test.account_activation_retry", TEST_CUSTOMER_ID, ) self.assertTrue(replay.idempotent_replay) self.assertEqual(replay.document.to_mapping(), activated.document.to_mapping()) self.assertEqual( activated.document.payload["devicePolicy"]["seatId"], TEST_CUSTOMER_ID, ) with Session(self.engine) as session: self.assertEqual(session.scalar(select(func.count()).select_from(Activation)), 2) self.assertEqual(session.scalar(select(func.count()).select_from(Lease)), 2) def test_rapid_device_churn_requires_review_but_allows_known_device(self) -> None: serial_value = self._generate_serials("edition.professional", 1)[0] first = self._enrollment("churn_1", system_uuid="churn-device-1") with Session(self.engine) as session: redeemed = self._service(session).redeem_serial( serial_value, first, 1, "idempotency-churn-redemption", "correlation.test.churn_redemption", actor_user_id=TEST_CUSTOMER_ID, ) for device_index in range(2, 6): with Session(self.engine) as session, session.begin(): active = tuple( session.scalars( select(Activation).where( Activation.license_id == redeemed.license_id, Activation.status == "active", ) ) ) for activation in active: activation.status = "deactivated" activation.deactivated_at = self.now activation.release_reason = "test_device_replacement" self.now += timedelta(hours=1) enrollment = self._enrollment( f"churn_{device_index}", system_uuid=f"churn-device-{device_index}", ) with Session(self.engine) as session: self._service(session).activate_owned_license( redeemed.license_id, enrollment, 1, f"idempotency-churn-{device_index}", f"correlation.test.churn_{device_index}", TEST_CUSTOMER_ID, ) with Session(self.engine) as session, session.begin(): active = tuple( session.scalars( select(Activation).where( Activation.license_id == redeemed.license_id, Activation.status == "active", ) ) ) for activation in active: activation.status = "deactivated" activation.deactivated_at = self.now activation.release_reason = "test_device_replacement" self.now += timedelta(hours=1) sixth = self._enrollment("churn_6", system_uuid="churn-device-6") with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as context: self._service(session).activate_owned_license( redeemed.license_id, sixth, 1, "idempotency-churn-6", "correlation.test.churn_6", TEST_CUSTOMER_ID, ) self.assertEqual(context.exception.code, ServerErrorCode.DEVICE_CHURN_REVIEW) known_reinstall = self._enrollment( "churn_1_reinstall", system_uuid="churn-device-1", ) with Session(self.engine) as session: recovered = self._service(session).activate_owned_license( redeemed.license_id, known_reinstall, 1, "idempotency-churn-known", "correlation.test.churn_known", TEST_CUSTOMER_ID, ) self.assertTrue(recovered.recovered_existing_device) self.assertEqual(recovered.device_id, redeemed.device_id) def test_team_named_seats_each_receive_their_device_allowance(self) -> None: serial_value = self._generate_serials("edition.lab_team", 1)[0] owner_enrollment = self._enrollment("team_owner", system_uuid="team-owner-device") with Session(self.engine) as session: redeemed = self._service(session).redeem_serial( serial_value, owner_enrollment, 1, "idempotency-team-redemption", "correlation.test.team_redemption", actor_user_id=TEST_CUSTOMER_ID, ) member_ids = tuple(f"user.test.team_member_{index}" for index in range(1, 6)) organization_id = "organization.test.team" with Session(self.engine) as session, session.begin(): session.add(Organization(id=organization_id, display_name="Test Lab", status="active")) session.add( Membership( id="membership.test.team_owner", organization_id=organization_id, user_id=TEST_CUSTOMER_ID, role="owner", status="active", valid_until=None, ) ) for index, member_id in enumerate(member_ids, start=1): session.add( User( id=member_id, external_issuer="https://identity.test", external_subject=f"team-member-subject-{index}", verified_email=f"team-member-{index}@example.test", status="active", is_server_admin=False, ) ) session.add( Membership( id=f"membership.test.team_member_{index}", organization_id=organization_id, user_id=member_id, role="member", status="active", valid_until=None, ) ) license_row = session.get(License, redeemed.license_id) license_row.owner_user_id = None license_row.owner_organization_id = organization_id license_row.subject_type = "organization" owner_seat_id = "organization_seat.test.team_owner" session.add( OrganizationSeat( id=owner_seat_id, organization_id=organization_id, license_id=redeemed.license_id, assigned_user_id=TEST_CUSTOMER_ID, status="assigned", assigned_at=self.now, released_at=None, last_reassigned_at=self.now, assignment_generation=1, ) ) session.get(Activation, redeemed.activation_id).seat_id = owner_seat_id for index, member_id in enumerate(member_ids[:4], start=1): session.add( OrganizationSeat( id=f"organization_seat.test.team_member_{index}", organization_id=organization_id, license_id=redeemed.license_id, assigned_user_id=member_id, status="assigned", assigned_at=self.now, released_at=None, last_reassigned_at=self.now, assignment_generation=1, ) ) first_member_enrollment = self._enrollment( "team_member_1a", system_uuid="team-member-1-device-a", ) with Session(self.engine) as session: first_member_result = self._service(session).activate_owned_license( redeemed.license_id, first_member_enrollment, 1, "idempotency-team-member-1a", "correlation.test.team_member_1a", member_ids[0], ) second_member_enrollment = self._enrollment( "team_member_1b", system_uuid="team-member-1-device-b", ) with Session(self.engine) as session: second_member_result = self._service(session).activate_owned_license( redeemed.license_id, second_member_enrollment, 1, "idempotency-team-member-1b", "correlation.test.team_member_1b", member_ids[0], ) self.assertNotEqual(first_member_result.device_id, second_member_result.device_id) for index, member_id in enumerate(member_ids[1:4], start=2): with Session(self.engine) as session: self._service(session).activate_owned_license( redeemed.license_id, self._enrollment( f"team_member_{index}", system_uuid=f"team-member-{index}-device", ), 1, f"idempotency-team-member-{index}", f"correlation.test.team_member_{index}", member_id, ) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as context: self._service(session).activate_owned_license( redeemed.license_id, self._enrollment( "team_member_5", system_uuid="team-member-5-device", ), 1, "idempotency-team-member-5", "correlation.test.team_member_5", member_ids[4], ) self.assertEqual(context.exception.code, ServerErrorCode.SEAT_NOT_ASSIGNED) with Session(self.engine) as session: active_seats = session.scalar( select(func.count(func.distinct(Activation.seat_id))).where( Activation.license_id == redeemed.license_id, Activation.status == "active", ) ) self.assertEqual(active_seats, 5) def test_database_rejects_two_assigned_seats_for_one_license_member(self) -> None: organization_id = "organization.test.unique_seat" license_id = "license.test.unique_seat" with Session(self.engine) as session: session.add( Organization( id=organization_id, display_name="Unique Seat Laboratory", status="active", ) ) session.add( License( id=license_id, product_id="apolon.measurement-studio", owner_user_id=None, owner_organization_id=organization_id, anonymous_subject_digest=None, subject_type="organization", status="active", device_policy_id="device_policy.lab_team", revocation_generation=0, ) ) for index in range(2): session.add( OrganizationSeat( id=f"organization_seat.test.unique_{index}", organization_id=organization_id, license_id=license_id, assigned_user_id=TEST_CUSTOMER_ID, status="assigned", assigned_at=self.now, released_at=None, last_reassigned_at=self.now, assignment_generation=1, ) ) with self.assertRaises(IntegrityError): session.commit() session.rollback() def test_remote_release_is_audited_bounded_and_increments_generation(self) -> None: serial_value = self._generate_serials("edition.professional", 1)[0] enrollment = self._enrollment("remote", system_uuid="remote-device") with Session(self.engine) as session: redeemed = self._service(session).redeem_serial( serial_value, enrollment, 1, "idempotency-remote-redemption", "correlation.test.remote_redemption", actor_user_id=TEST_CUSTOMER_ID, ) for release_index in range(3): with Session(self.engine) as session: activation_id, remaining = self._service( session ).release_unavailable_device( redeemed.license_id, redeemed.device_id, TEST_CUSTOMER_ID, f"correlation.test.remote_{release_index}", ) self.assertEqual(activation_id, redeemed.activation_id) self.assertEqual(remaining, 2 - release_index) if release_index < 2: with Session(self.engine) as session, session.begin(): activation = session.get(Activation, redeemed.activation_id) activation.status = "active" activation.deactivated_at = None activation.release_reason = None with Session(self.engine) as session, session.begin(): activation = session.get(Activation, redeemed.activation_id) activation.status = "active" activation.deactivated_at = None activation.release_reason = None with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as context: self._service(session).release_unavailable_device( redeemed.license_id, redeemed.device_id, TEST_CUSTOMER_ID, "correlation.test.remote_exhausted", ) self.assertEqual( context.exception.code, ServerErrorCode.FORCED_RELEASE_ALLOWANCE_EXHAUSTED, ) with Session(self.engine) as session: license_row = session.get(License, redeemed.license_id) activation = session.get(Activation, redeemed.activation_id) self.assertEqual(license_row.revocation_generation, 3) self.assertEqual(activation.status, "active") def test_mark_lost_revokes_every_enrolled_installation_key(self) -> None: serial_value = self._generate_serials("edition.professional", 1)[0] second_serial_value = self._generate_serials("edition.professional", 1)[0] original = self._enrollment("lost_original", system_uuid="lost-device") recovered = self._enrollment("lost_recovered", system_uuid="lost-device") replacement = self._enrollment("lost_replacement", system_uuid="lost-device") with Session(self.engine) as session: redeemed = self._service(session).redeem_serial( serial_value, original, 1, "idempotency-lost-redemption", "correlation.test.lost_redemption", actor_user_id=TEST_CUSTOMER_ID, ) with Session(self.engine) as session: recovered_result = self._service(session).activate_owned_license( redeemed.license_id, recovered, 1, "idempotency-lost-recovery", "correlation.test.lost_recovery", TEST_CUSTOMER_ID, ) self.assertTrue(recovered_result.recovered_existing_device) self.assertEqual(recovered_result.device_id, redeemed.device_id) with Session(self.engine) as session: second_license = self._service(session).redeem_serial( second_serial_value, recovered, 1, "idempotency-lost-second-license", "correlation.test.lost_second_license", actor_user_id=TEST_CUSTOMER_ID, ) self.assertEqual(second_license.device_id, redeemed.device_id) with Session(self.engine) as session: _activation_id, remaining = self._service( session ).release_unavailable_device( redeemed.license_id, redeemed.device_id, TEST_CUSTOMER_ID, "correlation.test.marked_lost", reason="marked_lost", ) self.assertEqual(remaining, 2) with Session(self.engine) as session: device = session.get(Device, redeemed.device_id) activation = session.get(Activation, redeemed.activation_id) installations = tuple( session.scalars( select(DeviceInstallation).where( DeviceInstallation.device_id == redeemed.device_id ) ) ) event = session.scalar( select(AuditEvent).where(AuditEvent.action == "device.marked_lost") ) self.assertIsNotNone(device) self.assertIsNotNone(activation) self.assertIsNotNone(event) assert device is not None assert activation is not None assert event is not None self.assertEqual(device.status, "lost") self.assertEqual(activation.status, "deactivated") second_activation = session.get(Activation, second_license.activation_id) self.assertIsNotNone(second_activation) assert second_activation is not None self.assertEqual(second_activation.status, "active") self.assertEqual(len(installations), 2) self.assertEqual({value.status for value in installations}, {"revoked"}) self.assertEqual(event.metadata_json["revokedInstallationCount"], 2) for enrollment, suffix in ( (original, "original"), (recovered, "recovered"), ): with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as context: self._service(session).activate_owned_license( redeemed.license_id, enrollment, 1, f"idempotency-lost-reactivate-{suffix}", f"correlation.test.lost_reactivate_{suffix}", TEST_CUSTOMER_ID, ) self.assertEqual( context.exception.code, ServerErrorCode.DEVICE_RECOVERY_REQUIRED, ) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as context: self._service(session).activate_owned_license( redeemed.license_id, replacement, 1, "idempotency-lost-replacement-blocked", "correlation.test.lost_replacement_blocked", TEST_CUSTOMER_ID, ) self.assertEqual(context.exception.code, ServerErrorCode.DEVICE_RECOVERY_REQUIRED) self.assertEqual(context.exception.existing_resource_id, redeemed.device_id) 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)), 2, ) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as context: self._service(session).preview_device_recovery( redeemed.device_id, TEST_CUSTOMER_ID, "customer_recovered_device", "correlation.test.lost_preview_denied", ) self.assertEqual(context.exception.code, ServerErrorCode.AUTHORIZATION_DENIED) with Session(self.engine) as session: preview = self._service(session).preview_device_recovery( redeemed.device_id, TEST_ADMIN_ID, "customer_recovered_device", "correlation.test.lost_preview", note="Ownership and physical recovery were verified.", ) self.assertEqual(preview["revokedInstallationCount"], 2) self.assertEqual(preview["activeActivationCount"], 1) self.assertEqual(preview["activeLicenseIds"], [second_license.license_id]) self.assertTrue(preview["requiresNewDeviceProofKey"]) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as context: self._service(session).recover_device( redeemed.device_id, TEST_ADMIN_ID, "customer_recovered_device", "0" * 64, "idempotency-lost-recover-stale", "correlation.test.lost_recover_stale", ) self.assertEqual(context.exception.code, ServerErrorCode.CONFLICT) with Session(self.engine) as session: restored = self._service(session).recover_device( redeemed.device_id, TEST_ADMIN_ID, "customer_recovered_device", preview["stateDigest"], "idempotency-lost-recover", "correlation.test.lost_recover", note="Ownership and physical recovery were verified.", ) self.assertTrue(restored["executed"]) self.assertFalse(restored["idempotentReplay"]) self.assertEqual(restored["currentStatus"], "active") with Session(self.engine) as session: replayed = self._service(session).recover_device( redeemed.device_id, TEST_ADMIN_ID, "customer_recovered_device", preview["stateDigest"], "idempotency-lost-recover", "correlation.test.lost_recover", note="Ownership and physical recovery were verified.", ) self.assertTrue(replayed["idempotentReplay"]) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as context: self._service(session).activate_owned_license( redeemed.license_id, original, 1, "idempotency-lost-old-key-after-restore", "correlation.test.lost_old_key_after_restore", TEST_CUSTOMER_ID, ) self.assertEqual( context.exception.code, ServerErrorCode.DEVICE_IDENTITY_REPLACEMENT_REQUIRED, ) with Session(self.engine) as session: replacement_result = self._service(session).activate_owned_license( redeemed.license_id, replacement, 1, "idempotency-lost-replacement", "correlation.test.lost_replacement", TEST_CUSTOMER_ID, ) self.assertTrue(replacement_result.recovered_existing_device) self.assertEqual(replacement_result.device_id, redeemed.device_id) with Session(self.engine) as session: second_refresh = self._service(session).refresh_activation( second_license.license_id, replacement.installation_id, replacement.public_key_der, replacement.key_thumbprint, 1, "correlation.test.lost_second_license_refresh", evidence=replacement.evidence, ) self.assertEqual(second_refresh.device_id, redeemed.device_id) with Session(self.engine) as session: self.assertEqual(session.scalar(select(func.count()).select_from(Device)), 1) event = session.scalar( select(AuditEvent).where( AuditEvent.action == "device.recovery_approved" ) ) self.assertIsNotNone(event) assert event is not None self.assertTrue(event.metadata_json["requiresNewDeviceProofKey"]) if __name__ == "__main__": unittest.main()