"""Account privacy export, reviewed erasure, and retention regressions.""" from __future__ import annotations from datetime import datetime, timedelta, timezone import hashlib import hmac import json import unittest from sqlalchemy import create_engine, func, select from sqlalchemy.orm import Session from sqlalchemy.pool import StaticPool 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, IdempotencyRecord, License, Membership, NotificationDelivery, NotificationFeedbackEvent, NotificationSuppression, Organization, OrganizationInvitation, OrganizationSeat, OutboxEvent, PrivacyRequest, Subscription, SubscriptionItem, User, ) from licensing_server.app.privacy import ( AccountPrivacyService, ORGANIZATION_INVITATION_EMAIL_DOMAIN, PRIVACY_ERASED_ISSUER, PRIVACY_EXPORT_SCHEMA, purge_privacy_requests, ) TEST_FINGERPRINT_PEPPER = b"privacy-test-fingerprint-pepper-material" class AccountPrivacyTests(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) with Session(self.engine) as session, session.begin(): session.add_all( ( User( id="user.test.customer", external_issuer="https://identity.test", external_subject="customer-subject", verified_email="customer@example.test", status="active", is_server_admin=False, ), 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, ), User( id="user.test.owner", external_issuer="https://identity.test", external_subject="owner-subject", verified_email="owner@example.test", status="active", is_server_admin=False, ), ) ) def tearDown(self) -> None: self.engine.dispose() def _service(self, session: Session) -> AccountPrivacyService: return AccountPrivacyService(session, TEST_FINGERPRINT_PEPPER) def _request_deletion(self) -> dict[str, object]: with Session(self.engine) as session: return self._service(session).request_deletion( "user.test.customer", True, "privacy-request-test", "correlation.test.privacy_request", ) def _add_personal_license( self, session: Session, *, subscription_status: str = "canceled", ) -> None: now = datetime.now(timezone.utc) session.add_all( ( License( id="license.test.privacy", product_id="apolon.precision_lab_analyzer", owner_user_id="user.test.customer", owner_organization_id=None, anonymous_subject_digest=None, subject_type="user", status="active", device_policy_id="device_policy.personal", revocation_generation=0, ), Grant( id="grant.test.privacy", license_id="license.test.privacy", sku_id="edition.professional", source_type="purchase", source_reference="order.test.privacy", catalog_revision=1, starts_at=now - timedelta(days=10), ends_at=None, status="active", metadata_json={}, ), Subscription( id="subscription.test.privacy", license_id="license.test.privacy", provider="stripe", provider_subscription_id="sub_test_privacy", provider_customer_id="cus_test_privacy", status=subscription_status, current_period_start=now - timedelta(days=30), current_period_end=now + timedelta(days=1), cancel_at_period_end=True, grace_ends_at=None, provider_event_created_at=now, billing_hold_status=None, billing_hold_provider_id=None, billing_hold_event_created_at=None, ), SubscriptionItem( id="subscription_item.test.privacy", subscription_id="subscription.test.privacy", sku_id="subscription.ai_assist", quantity=1, provider_item_id="si_test_privacy", ), ) ) def test_recent_account_export_contains_owned_data_without_secret_material( self, ) -> None: now = datetime.now(timezone.utc) with Session(self.engine) as session, session.begin(): self._add_personal_license(session) session.add_all( ( Device( id="device.test.privacy", owner_user_id="user.test.customer", owner_organization_id=None, friendly_name="Customer workstation", device_type="computer", status="active", evidence_schema_version=1, evidence_digests={"system_uuid": "ab" * 32}, last_seen_at=now, ), DeviceInstallation( id="device_installation.test.privacy", installation_id="installation.test.privacy", device_id="device.test.privacy", public_key_der=b"public-key-material", key_thumbprint="device_key.test.privacy", key_provider="software.test", status="active", last_seen_at=now, ), ) ) with Session(self.engine) as session: document = self._service(session).export_account( "user.test.customer", "correlation.test.privacy_export", ) self.assertEqual(document["schema"], PRIVACY_EXPORT_SCHEMA) self.assertEqual( document["account"]["externalIdentity"]["subject"], "customer-subject", ) self.assertEqual(document["licenses"][0]["grants"][0]["skuId"], "edition.professional") self.assertEqual( document["licenses"][0]["subscriptions"][0]["providerCustomerId"], "cus_test_privacy", ) self.assertEqual( document["devices"][0]["evidenceDigests"]["system_uuid"], "ab" * 32, ) serialized = json.dumps(document, sort_keys=True) for forbidden in ( "public-key-material", "public_key_der", "secret_digest", "token_digest", "signing_private", ): self.assertNotIn(forbidden, serialized) with Session(self.engine) as session: event = session.scalar( select(AuditEvent).where( AuditEvent.action == "account.privacy_exported" ) ) self.assertIsNotNone(event) self.assertEqual(event.metadata_json, {"requestType": "export"}) def test_deletion_request_is_idempotent_cancelable_and_reopenable(self) -> None: created = self._request_deletion() self.assertEqual(created["status"], "pending") self.assertFalse(created["existingRequest"]) with Session(self.engine) as session: replay = self._service(session).request_deletion( "user.test.customer", True, "privacy-request-test", "correlation.test.privacy_request_replay", ) self.assertTrue(replay["idempotentReplay"]) self.assertEqual(replay["privacyRequestId"], created["privacyRequestId"]) with Session(self.engine) as session: existing = self._service(session).request_deletion( "user.test.customer", True, "privacy-request-another-key", "correlation.test.privacy_request_existing", ) self.assertTrue(existing["existingRequest"]) self.assertFalse(existing["idempotentReplay"]) with Session(self.engine) as session: cancelled = self._service(session).cancel_deletion_request( created["privacyRequestId"], "user.test.customer", "correlation.test.privacy_cancel", ) self.assertEqual(cancelled["status"], "cancelled") with Session(self.engine) as session: replacement = self._service(session).request_deletion( "user.test.customer", True, "privacy-request-replacement", "correlation.test.privacy_replacement", ) self.assertNotEqual( replacement["privacyRequestId"], created["privacyRequestId"], ) self.assertEqual(replacement["status"], "pending") def test_preview_blocks_live_billing_last_owner_and_server_admin(self) -> None: now = datetime.now(timezone.utc) with Session(self.engine) as session, session.begin(): self._add_personal_license(session, subscription_status="active") customer = session.get(User, "user.test.customer") customer.is_server_admin = True session.add( Organization( id="organization.test.privacy_blocked", display_name="Blocked privacy organization", status="active", ) ) session.add( Membership( id="membership.test.privacy_last_owner", organization_id="organization.test.privacy_blocked", user_id="user.test.customer", role="owner", status="active", valid_until=None, ) ) requested = self._request_deletion() with Session(self.engine) as session: preview = self._service(session).preview_deletion( requested["privacyRequestId"], "user.test.admin", "customer_request", "correlation.test.privacy_blocked_preview", ) self.assertFalse(preview["canComplete"]) self.assertEqual( {row["code"] for row in preview["blockingReasons"]}, { "server_administrator_must_be_transferred", "subscription_must_be_cancelled_or_resolved", "organization_ownership_must_be_transferred", }, ) with Session(self.engine) as session, session.begin(): session.get(User, "user.test.customer").is_server_admin = False session.get(Subscription, "subscription.test.privacy").status = "canceled" session.add( Membership( id="membership.test.privacy_other_owner", organization_id="organization.test.privacy_blocked", user_id="user.test.owner", role="owner", status="active", valid_until=now + timedelta(days=365), ) ) with Session(self.engine) as session: resolved = self._service(session).preview_deletion( requested["privacyRequestId"], "user.test.admin", "customer_request", "correlation.test.privacy_resolved_preview", ) self.assertTrue(resolved["canComplete"]) self.assertNotEqual(resolved["stateDigest"], preview["stateDigest"]) def test_reviewed_erasure_preserves_rights_and_redacts_account_links(self) -> None: now = datetime.now(timezone.utc) invitation_email_digest = hmac.new( TEST_FINGERPRINT_PEPPER, ORGANIZATION_INVITATION_EMAIL_DOMAIN + b"customer@example.test", hashlib.sha256, ).digest() with Session(self.engine) as session, session.begin(): self._add_personal_license(session) session.add_all( ( Organization( id="organization.test.privacy", display_name="Privacy team", status="active", ), License( id="license.test.privacy_team", product_id="apolon.precision_lab_analyzer", owner_user_id=None, owner_organization_id="organization.test.privacy", anonymous_subject_digest=None, subject_type="organization", status="active", device_policy_id="device_policy.team", revocation_generation=0, ), Membership( id="membership.test.privacy", organization_id="organization.test.privacy", user_id="user.test.customer", role="member", status="active", valid_until=None, ), Membership( id="membership.test.privacy_owner", organization_id="organization.test.privacy", user_id="user.test.owner", role="owner", status="active", valid_until=None, ), OrganizationSeat( id="organization_seat.test.privacy", organization_id="organization.test.privacy", license_id="license.test.privacy_team", assigned_user_id="user.test.customer", status="assigned", assigned_at=now, released_at=None, last_reassigned_at=now, assignment_generation=1, ), Device( id="device.test.privacy", owner_user_id="user.test.customer", owner_organization_id=None, friendly_name="Personal lab computer", device_type="computer", status="active", evidence_schema_version=1, evidence_digests={"system_uuid": "cd" * 32}, last_seen_at=now, ), DeviceInstallation( id="device_installation.test.privacy", installation_id="installation.test.privacy", device_id="device.test.privacy", public_key_der=b"device-public-key", key_thumbprint="device_key.test.privacy", key_provider="software.test", status="active", last_seen_at=now, ), Activation( id="activation.test.privacy_personal", license_id="license.test.privacy", device_id="device.test.privacy", installation_id="device_installation.test.privacy", seat_id=None, status="active", activated_at=now, deactivated_at=None, release_reason=None, ), Activation( id="activation.test.privacy_team", license_id="license.test.privacy_team", device_id="device.test.privacy", installation_id="device_installation.test.privacy", seat_id="organization_seat.test.privacy", status="active", activated_at=now, deactivated_at=None, release_reason=None, ), OrganizationInvitation( id="organization_invitation.test.privacy", organization_id="organization.test.privacy", invited_email_digest=invitation_email_digest, token_digest=hashlib.sha256(b"invitation-token").digest(), role="member", status="pending", expires_at=now + timedelta(days=7), created_by_user_id="user.test.owner", accepted_by_user_id=None, accepted_at=None, ), OutboxEvent( id="outbox.test.privacy", event_type="customer.notification", aggregate_type="subscription", aggregate_id="subscription.test.privacy", payload_json={"notificationKind": "subscription_cancelled"}, status="processed", available_at=now, attempts=1, processed_at=now, last_error_code=None, ), NotificationDelivery( id="notification_delivery.test.privacy", outbox_event_id="outbox.test.privacy", user_id="user.test.customer", status="processed", attempts=1, provider_message_id="message.test.privacy", recipient_address_digest=hashlib.sha256( b"customer@example.test" ).digest(), last_error_code=None, delivered_at=now, feedback_status="soft_bounce", feedback_at=now, ), NotificationSuppression( id="notification_suppression.test.privacy", user_id="user.test.customer", recipient_address_digest=hashlib.sha256( b"customer@example.test" ).digest(), reason="hard_bounce", source_feedback_id=None, active=True, suppressed_at=now, cleared_at=None, ), IdempotencyRecord( id="idempotency.test.privacy", scope="account.test", subject_key="user.test.customer", idempotency_key="account-privacy-test", request_digest=hashlib.sha256(b"request").digest(), response_status=200, response_json={"deviceName": "Personal lab computer"}, expires_at=now + timedelta(days=7), ), AuditEvent( id="audit.test.privacy_retained", occurred_at=now, actor_type="user", actor_id="user.test.customer", action="account.test_event", target_type="user", target_id="user.test.customer", reason=None, correlation_id="correlation.test.privacy_retained", source_address_digest=None, metadata_json={}, ), ) ) session.flush() session.add( NotificationFeedbackEvent( id="notification_feedback.test.privacy", provider="smtp.test", provider_event_id="feedback.test.privacy", delivery_id="notification_delivery.test.privacy", delivery_reference_id="notification_delivery.test.privacy", provider_message_id="message.test.privacy", feedback_type="soft_bounce", occurred_at=now, payload_digest=hashlib.sha256(b"feedback").digest(), processed_at=now, ) ) requested = self._request_deletion() with Session(self.engine) as session: preview = self._service(session).preview_deletion( requested["privacyRequestId"], "user.test.admin", "customer_request", "correlation.test.privacy_execute_preview", note="Verified customer request", ) self.assertTrue(preview["canComplete"]) with Session(self.engine) as session: completed = self._service(session).execute_deletion( requested["privacyRequestId"], "user.test.admin", "customer_request", preview["stateDigest"], "privacy-execution-test", "correlation.test.privacy_execute", note="Verified customer request", ) self.assertTrue(completed["executed"]) self.assertFalse(completed["customerRightsDeleted"]) self.assertEqual(completed["preservedOwnedLicenseCount"], 1) self.assertEqual(completed["preservedGrantCount"], 1) self.assertEqual(completed["releasedActivationCount"], 2) with Session(self.engine) as session: customer = session.get(User, "user.test.customer") self.assertEqual(customer.status, "privacy_erased") self.assertEqual(customer.external_issuer, PRIVACY_ERASED_ISSUER) self.assertNotEqual(customer.external_subject, "customer-subject") self.assertIsNone(customer.verified_email) self.assertIsNotNone(session.get(License, "license.test.privacy")) self.assertIsNotNone(session.get(Grant, "grant.test.privacy")) self.assertIsNotNone( session.get(Subscription, "subscription.test.privacy") ) device = session.get(Device, "device.test.privacy") self.assertEqual(device.status, "released") self.assertEqual(device.evidence_digests, {}) installation = session.get( DeviceInstallation, "device_installation.test.privacy", ) self.assertEqual(installation.public_key_der, b"\x00") self.assertNotEqual(installation.installation_id, "installation.test.privacy") self.assertEqual( session.get(Activation, "activation.test.privacy_personal").status, "deactivated", ) seat = session.get(OrganizationSeat, "organization_seat.test.privacy") self.assertEqual(seat.status, "available") self.assertIsNone(seat.assigned_user_id) self.assertEqual( session.get(Membership, "membership.test.privacy").status, "removed", ) invitation = session.get( OrganizationInvitation, "organization_invitation.test.privacy", ) self.assertEqual(invitation.status, "cancelled") self.assertNotEqual(invitation.invited_email_digest, invitation_email_digest) self.assertEqual( session.scalar( select(func.count()).select_from(NotificationDelivery) ), 0, ) self.assertEqual( session.scalar( select(func.count()).select_from(NotificationSuppression) ), 0, ) feedback = session.get( NotificationFeedbackEvent, "notification_feedback.test.privacy", ) self.assertIsNone(feedback.delivery_id) self.assertIsNone(feedback.provider_message_id) self.assertIsNone( session.get(IdempotencyRecord, "idempotency.test.privacy") ) self.assertIsNotNone( session.get(AuditEvent, "audit.test.privacy_retained") ) with Session(self.engine) as session: replay = self._service(session).execute_deletion( requested["privacyRequestId"], "user.test.admin", "customer_request", preview["stateDigest"], "privacy-execution-test", "correlation.test.privacy_execute_replay", note="Verified customer request", ) self.assertTrue(replay["idempotentReplay"]) def test_stale_preview_and_changed_execution_idempotency_fail_closed(self) -> None: requested = self._request_deletion() with Session(self.engine) as session: preview = self._service(session).preview_deletion( requested["privacyRequestId"], "user.test.admin", "customer_request", "correlation.test.privacy_stale_preview", ) with Session(self.engine) as session, session.begin(): session.add( License( id="license.test.privacy_state_change", product_id="apolon.precision_lab_analyzer", owner_user_id="user.test.customer", owner_organization_id=None, anonymous_subject_digest=None, subject_type="user", status="active", device_policy_id="device_policy.personal", revocation_generation=0, ) ) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as caught: self._service(session).execute_deletion( requested["privacyRequestId"], "user.test.admin", "customer_request", preview["stateDigest"], "privacy-execution-stale", "correlation.test.privacy_stale_execute", ) self.assertEqual(caught.exception.code, ServerErrorCode.CONFLICT) with Session(self.engine) as session: fresh = self._service(session).preview_deletion( requested["privacyRequestId"], "user.test.admin", "customer_request", "correlation.test.privacy_fresh_preview", ) with Session(self.engine) as session: self._service(session).execute_deletion( requested["privacyRequestId"], "user.test.admin", "customer_request", fresh["stateDigest"], "privacy-execution-fresh", "correlation.test.privacy_fresh_execute", ) with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as changed: self._service(session).execute_deletion( requested["privacyRequestId"], "user.test.admin", "customer_request", fresh["stateDigest"], "privacy-execution-changed", "correlation.test.privacy_changed_replay", ) self.assertEqual(changed.exception.code, ServerErrorCode.IDEMPOTENCY_CONFLICT) def test_retention_purges_only_old_resolved_requests_and_keeps_audit(self) -> None: now = datetime.now(timezone.utc) old = now - timedelta(days=400) with Session(self.engine) as session, session.begin(): session.add_all( ( PrivacyRequest( id="privacy_request.test.old_cancelled", user_id="user.test.customer", request_type="deletion", status="cancelled", requested_at=old, request_digest=hashlib.sha256(b"cancelled").digest(), idempotency_key="privacy-old-cancelled", correlation_id="correlation.test.old_cancelled", state_generation=1, cancelled_at=old, completed_at=None, completed_by_user_id=None, completion_reason=None, completion_idempotency_key=None, completion_request_digest=None, completion_summary_json={}, ), PrivacyRequest( id="privacy_request.test.pending", user_id="user.test.owner", request_type="deletion", status="pending", requested_at=old, request_digest=hashlib.sha256(b"pending").digest(), idempotency_key="privacy-old-pending", correlation_id="correlation.test.old_pending", state_generation=0, cancelled_at=None, completed_at=None, completed_by_user_id=None, completion_reason=None, completion_idempotency_key=None, completion_request_digest=None, completion_summary_json={}, ), AuditEvent( id="audit.test.privacy_retention", occurred_at=old, actor_type="user", actor_id="user.test.customer", action="account.deletion_cancelled", target_type="privacy_request", target_id="privacy_request.test.old_cancelled", reason="customer_request", correlation_id="correlation.test.retention_audit", source_address_digest=None, metadata_json={}, ), ) ) with Session(self.engine) as session: result = purge_privacy_requests( session, now=now, correlation_id="correlation.test.privacy_purge", retention_days=365, maximum_requests=100, ) self.assertEqual(result["purgedPrivacyRequestCount"], 1) with Session(self.engine) as session: self.assertIsNone( session.get(PrivacyRequest, "privacy_request.test.old_cancelled") ) self.assertIsNotNone( session.get(PrivacyRequest, "privacy_request.test.pending") ) self.assertIsNotNone( session.get(AuditEvent, "audit.test.privacy_retention") ) self.assertIsNotNone( session.scalar( select(AuditEvent).where( AuditEvent.action == "privacy.requests_purged" ) ) ) def test_privacy_administration_requires_active_persisted_admin(self) -> None: requested = self._request_deletion() with Session(self.engine) as session: with self.assertRaises(ServerLicensingError) as denied: self._service(session).preview_deletion( requested["privacyRequestId"], "user.test.owner", "customer_request", "correlation.test.privacy_denied", ) self.assertEqual(denied.exception.code, ServerErrorCode.AUTHORIZATION_DENIED) with Session(self.engine) as session: with self.assertRaises(ValueError): self._service(session).preview_deletion( requested["privacyRequestId"], "user.test.admin", "arbitrary_reason", "correlation.test.privacy_bad_reason", ) with Session(self.engine) as session: with self.assertRaises(ValueError): purge_privacy_requests( session, now=datetime.now(timezone.utc), correlation_id="correlation.test.privacy_bad_retention", retention_days=1, ) if __name__ == "__main__": unittest.main()