"""Privacy-bounded account export, reviewed erasure, and retention jobs.""" from __future__ import annotations from datetime import datetime, timedelta, timezone import hashlib import hmac from typing import Mapping, Optional, Sequence from sqlalchemy import func, or_, select from sqlalchemy.orm import Session from licensing_shared.canonical_json import canonicalize_json from licensing_shared.constants import validate_identifier from licensing_shared.models import format_rfc3339 from .constants import ( FINGERPRINT_PEPPER_MINIMUM_BYTES, MAX_IDEMPOTENCY_KEY_CHARACTERS, MAX_REASON_CHARACTERS, PRIVACY_DELETION_REASONS, ) from .errors import ServerErrorCode, ServerLicensingError from .models import ( Activation, AuditEvent, Device, DeviceInstallation, Grant, IdempotencyRecord, License, Membership, NotificationDelivery, NotificationFeedbackEvent, NotificationSuppression, Organization, OrganizationInvitation, OrganizationSeat, PrivacyRequest, SerialRedemption, Subscription, SubscriptionItem, User, ) from .security import new_identifier PRIVACY_EXPORT_SCHEMA = "apolon.licensing.account-privacy-export" PRIVACY_EXPORT_SCHEMA_VERSION = 1 PRIVACY_REQUEST_TYPE_DELETION = "deletion" PRIVACY_REQUEST_STATUSES = frozenset(("pending", "cancelled", "completed")) PRIVACY_TERMINAL_SUBSCRIPTION_STATUSES = frozenset( ("canceled", "incomplete_expired") ) PRIVACY_ERASED_ISSUER = "urn:apolon:privacy-erased" PRIVACY_ERASED_DEVICE_NAME = "Removed account device" PRIVACY_ERASURE_RELEASE_REASON = "account_privacy_erasure" MAXIMUM_SELF_SERVICE_PRIVACY_ROWS = 5000 MAXIMUM_PRIVACY_REQUEST_LIST = 100 DEFAULT_PRIVACY_REQUEST_RETENTION_DAYS = 365 MINIMUM_PRIVACY_REQUEST_RETENTION_DAYS = 30 MAXIMUM_PRIVACY_REQUEST_RETENTION_DAYS = 3650 MAXIMUM_PRIVACY_PURGE_ROWS = 1000 ORGANIZATION_INVITATION_EMAIL_DOMAIN = ( b"APOLON-ORGANIZATION-INVITATION-EMAIL-V1\x00" ) def _utc(value: datetime) -> datetime: if not isinstance(value, datetime): raise TypeError("value must be a datetime") if value.tzinfo is None or value.utcoffset() is None: return value.replace(tzinfo=timezone.utc) return value.astimezone(timezone.utc) def _time(value: datetime) -> str: return format_rfc3339(_utc(value), "timestamp") def _optional_time(value: Optional[datetime]) -> Optional[str]: return None if value is None else _time(value) def _bounded_text(value: str, field_name: str, maximum: int) -> str: if not isinstance(field_name, str) or not field_name.strip(): raise ValueError("field_name must be a non-empty string") if isinstance(maximum, bool) or not isinstance(maximum, int) or maximum < 1: raise ValueError("maximum must be a positive integer") if not isinstance(value, str) or not value.strip(): raise ValueError(f"{field_name.strip()} must be a non-empty string") normalized = value.strip() if len(normalized) > maximum: raise ValueError(f"{field_name.strip()} is too long") return normalized def _optional_note(value: Optional[str]) -> Optional[str]: if value is None: return None return _bounded_text(value, "note", MAX_REASON_CHARACTERS) def _request_digest(document: Mapping[str, object]) -> bytes: if not isinstance(document, Mapping): raise TypeError("document must be a mapping") return hashlib.sha256(canonicalize_json(dict(document))).digest() def _email_digest(fingerprint_pepper: bytes, verified_email: str) -> bytes: if not isinstance(fingerprint_pepper, bytes): raise TypeError("fingerprint_pepper must be bytes") if len(fingerprint_pepper) < FINGERPRINT_PEPPER_MINIMUM_BYTES: raise ValueError("fingerprint_pepper is too short") normalized_email = _bounded_text( verified_email, "verified_email", 256, ).casefold() return hmac.new( fingerprint_pepper, ORGANIZATION_INVITATION_EMAIL_DOMAIN + normalized_email.encode("utf-8"), hashlib.sha256, ).digest() def _redaction_digest() -> bytes: return hashlib.sha256(new_identifier("privacy_redaction").encode("ascii")).digest() def _validate_limit(value: int, maximum: int, field_name: str = "limit") -> int: if isinstance(value, bool) or not isinstance(value, int): raise TypeError(f"{field_name} must be an integer") if value < 1 or value > maximum: raise ValueError(f"{field_name} must be between one and {maximum}") return value class AccountPrivacyService: def __init__(self, session: Session, fingerprint_pepper: bytes) -> None: if not isinstance(session, Session): raise TypeError("session must be a SQLAlchemy Session") if not isinstance(fingerprint_pepper, bytes): raise TypeError("fingerprint_pepper must be bytes") if len(fingerprint_pepper) < FINGERPRINT_PEPPER_MINIMUM_BYTES: raise ValueError("fingerprint_pepper is too short") self._session = session self._fingerprint_pepper = fingerprint_pepper def export_account( self, user_id: str, correlation_id: str, ) -> dict[str, object]: normalized_user = validate_identifier(user_id, "user_id") normalized_correlation = validate_identifier( correlation_id, "correlation_id", ) now = datetime.now(timezone.utc) with self._session.begin(): user = self._require_active_user(normalized_user, lock=True) self._session.add( AuditEvent( id=new_identifier("audit"), actor_type="user", actor_id=user.id, action="account.privacy_exported", target_type="user", target_id=user.id, reason="customer_request", correlation_id=normalized_correlation, source_address_digest=None, metadata_json={ "requestType": "export", }, ) ) self._session.flush() return self._export_mapping(user, now) def request_deletion( self, user_id: str, acknowledge_permanent_account_closure: bool, idempotency_key: str, correlation_id: str, ) -> dict[str, object]: normalized_user = validate_identifier(user_id, "user_id") if not isinstance(acknowledge_permanent_account_closure, bool): raise TypeError("acknowledge_permanent_account_closure must be a Boolean") if not acknowledge_permanent_account_closure: raise ValueError("permanent account closure must be acknowledged") normalized_key = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_correlation = validate_identifier( correlation_id, "correlation_id", ) request_digest = _request_digest( { "acknowledgePermanentAccountClosure": True, "requestType": PRIVACY_REQUEST_TYPE_DELETION, } ) now = datetime.now(timezone.utc) with self._session.begin(): user = self._require_active_user(normalized_user, lock=True) replay = self._session.scalar( select(PrivacyRequest) .where( PrivacyRequest.user_id == user.id, PrivacyRequest.idempotency_key == normalized_key, ) .with_for_update() ) if replay is not None: if not hmac.compare_digest(replay.request_digest, request_digest): raise ServerLicensingError( ServerErrorCode.IDEMPOTENCY_CONFLICT, "idempotency key was already used for a different privacy request", status_code=409, ) return self._request_mapping( replay, user, now, idempotent_replay=True, existing_request=True, ) pending = self._session.scalar( select(PrivacyRequest) .where( PrivacyRequest.user_id == user.id, PrivacyRequest.status == "pending", ) .with_for_update() ) if pending is not None: return self._request_mapping( pending, user, now, idempotent_replay=False, existing_request=True, ) privacy_request = PrivacyRequest( id=new_identifier("privacy_request"), user_id=user.id, request_type=PRIVACY_REQUEST_TYPE_DELETION, status="pending", requested_at=now, request_digest=request_digest, idempotency_key=normalized_key, correlation_id=normalized_correlation, 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={}, ) self._session.add(privacy_request) self._session.add( AuditEvent( id=new_identifier("audit"), actor_type="user", actor_id=user.id, action="account.deletion_requested", target_type="privacy_request", target_id=privacy_request.id, reason="customer_request", correlation_id=normalized_correlation, source_address_digest=None, metadata_json={ "privacyRequestId": privacy_request.id, "privacyRequestStatus": privacy_request.status, "requestType": privacy_request.request_type, }, ) ) self._session.flush() return self._request_mapping( privacy_request, user, now, idempotent_replay=False, existing_request=False, ) def list_deletion_requests(self, user_id: str) -> dict[str, object]: normalized_user = validate_identifier(user_id, "user_id") now = datetime.now(timezone.utc) with self._session.begin(): user = self._require_active_user(normalized_user, lock=False) rows = self._bounded_scalars( select(PrivacyRequest) .where(PrivacyRequest.user_id == user.id) .order_by( PrivacyRequest.requested_at.desc(), PrivacyRequest.id.desc(), ), "privacy requests", maximum=MAXIMUM_PRIVACY_REQUEST_LIST, ) return { "requests": [ self._request_mapping( row, user, now, idempotent_replay=False, existing_request=True, ) for row in rows ] } def cancel_deletion_request( self, request_id: str, user_id: str, correlation_id: str, ) -> dict[str, object]: normalized_request = validate_identifier(request_id, "request_id") normalized_user = validate_identifier(user_id, "user_id") normalized_correlation = validate_identifier( correlation_id, "correlation_id", ) now = datetime.now(timezone.utc) with self._session.begin(): user = self._require_active_user(normalized_user, lock=True) privacy_request = self._session.scalar( select(PrivacyRequest) .where(PrivacyRequest.id == normalized_request) .with_for_update() ) if privacy_request is None or privacy_request.user_id != user.id: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "privacy request was not found for the authenticated account", status_code=403, ) if privacy_request.status == "completed": raise ServerLicensingError( ServerErrorCode.CONFLICT, "a completed account deletion cannot be cancelled", status_code=409, ) if privacy_request.status == "cancelled": return self._request_mapping( privacy_request, user, now, idempotent_replay=True, existing_request=True, ) privacy_request.status = "cancelled" privacy_request.cancelled_at = now privacy_request.state_generation += 1 self._session.add( AuditEvent( id=new_identifier("audit"), actor_type="user", actor_id=user.id, action="account.deletion_cancelled", target_type="privacy_request", target_id=privacy_request.id, reason="customer_request", correlation_id=normalized_correlation, source_address_digest=None, metadata_json={ "privacyRequestId": privacy_request.id, "privacyRequestStatus": privacy_request.status, "requestType": privacy_request.request_type, }, ) ) self._session.flush() return self._request_mapping( privacy_request, user, now, idempotent_replay=False, existing_request=True, ) def list_pending_for_administration( self, actor_user_id: str, *, status: Optional[str] = "pending", limit: int = 50, ) -> dict[str, object]: normalized_actor = validate_identifier(actor_user_id, "actor_user_id") if status is not None: if not isinstance(status, str): raise TypeError("status must be a string or None") normalized_status = status.strip().lower() if normalized_status not in PRIVACY_REQUEST_STATUSES: raise ValueError("status is invalid") else: normalized_status = None bounded_limit = _validate_limit( limit, MAXIMUM_PRIVACY_REQUEST_LIST, ) now = datetime.now(timezone.utc) with self._session.begin(): self._require_server_admin(normalized_actor, lock=False) query = select(PrivacyRequest) if normalized_status is not None: query = query.where(PrivacyRequest.status == normalized_status) rows = tuple( self._session.scalars( query.order_by( PrivacyRequest.requested_at, PrivacyRequest.id, ).limit(bounded_limit + 1) ).all() ) returned = rows[:bounded_limit] return { "requests": [ self._administration_mapping(row, now) for row in returned ], "hasMore": len(rows) > bounded_limit, } def preview_deletion( self, request_id: str, actor_user_id: str, reason_code: str, correlation_id: str, *, note: Optional[str] = None, ) -> dict[str, object]: normalized_request = validate_identifier(request_id, "request_id") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_reason = self._deletion_reason(reason_code) validate_identifier(correlation_id, "correlation_id") _optional_note(note) now = datetime.now(timezone.utc) with self._session.begin(): self._require_server_admin(normalized_actor, lock=False) privacy_request = self._require_privacy_request( normalized_request, lock=False, ) if privacy_request.status != "pending": raise ServerLicensingError( ServerErrorCode.CONFLICT, "only a pending account deletion can be previewed", status_code=409, ) user = self._require_user(privacy_request.user_id, lock=False) result = self._deletion_state(privacy_request, user, now) result.update( { "reasonCode": normalized_reason, "executed": False, "idempotentReplay": False, } ) return result def execute_deletion( self, request_id: str, actor_user_id: str, reason_code: str, expected_state_digest: str, idempotency_key: str, correlation_id: str, *, note: Optional[str] = None, ) -> dict[str, object]: normalized_request = validate_identifier(request_id, "request_id") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_reason = self._deletion_reason(reason_code) normalized_state_digest = _bounded_text( expected_state_digest, "expected_state_digest", 64, ) if len(normalized_state_digest) != 64 or any( character not in "0123456789abcdef" for character in normalized_state_digest ): raise ValueError("expected_state_digest must be a lowercase SHA-256 digest") normalized_key = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_correlation = validate_identifier( correlation_id, "correlation_id", ) normalized_note = _optional_note(note) completion_digest = _request_digest( { "privacyRequestId": normalized_request, "reasonCode": normalized_reason, "note": normalized_note, "expectedStateDigest": normalized_state_digest, } ) now = datetime.now(timezone.utc) with self._session.begin(): actor = self._require_server_admin(normalized_actor, lock=True) privacy_request = self._require_privacy_request( normalized_request, lock=True, ) if privacy_request.status == "completed": if ( privacy_request.completion_idempotency_key != normalized_key or privacy_request.completion_request_digest is None or not hmac.compare_digest( privacy_request.completion_request_digest, completion_digest, ) ): raise ServerLicensingError( ServerErrorCode.IDEMPOTENCY_CONFLICT, "account deletion was already completed by another request", status_code=409, ) replay = dict(privacy_request.completion_summary_json) replay["idempotentReplay"] = True return replay if privacy_request.status != "pending": raise ServerLicensingError( ServerErrorCode.CONFLICT, "only a pending account deletion can be completed", status_code=409, ) user = self._require_user(privacy_request.user_id, lock=True) state = self._deletion_state(privacy_request, user, now, lock=True) if state["stateDigest"] != normalized_state_digest: raise ServerLicensingError( ServerErrorCode.CONFLICT, "account deletion state changed after preview; review it again", status_code=409, ) if state["canComplete"] is not True: raise ServerLicensingError( ServerErrorCode.CONFLICT, "account deletion has unresolved billing, ownership, or administrator blockers", status_code=409, ) changes = self._apply_erasure(user, now) privacy_request.status = "completed" privacy_request.state_generation += 1 privacy_request.completed_at = now privacy_request.completed_by_user_id = actor.id privacy_request.completion_reason = normalized_reason privacy_request.completion_idempotency_key = normalized_key privacy_request.completion_request_digest = completion_digest result = { "privacyRequestId": privacy_request.id, "status": privacy_request.status, "completedAt": _time(now), "reasonCode": normalized_reason, "reviewedStateDigest": normalized_state_digest, "preservedOwnedLicenseCount": state["preservedOwnedLicenseCount"], "preservedGrantCount": state["preservedGrantCount"], "preservedSubscriptionRecordCount": state[ "preservedSubscriptionRecordCount" ], "retainedAuditEventCount": state["retainedAuditEventCount"], "erasureMode": "irreversible_pseudonymization", "customerRightsDeleted": False, "existingOfflineDocumentsRemainTimeBound": True, "executed": True, "idempotentReplay": False, **changes, } privacy_request.completion_summary_json = dict(result) self._session.add( AuditEvent( id=new_identifier("audit"), actor_type="user", actor_id=actor.id, action="account.privacy_erased", target_type="privacy_request", target_id=privacy_request.id, reason=normalized_note or normalized_reason, correlation_id=normalized_correlation, source_address_digest=None, metadata_json={ "privacyRequestId": privacy_request.id, "privacyRequestStatus": privacy_request.status, "requestType": privacy_request.request_type, "reasonCode": normalized_reason, "ownedLicenseCount": state[ "preservedOwnedLicenseCount" ], "releasedActivationCount": changes[ "releasedActivationCount" ], "releasedSeatCount": changes["releasedSeatCount"], "removedMembershipCount": changes[ "removedMembershipCount" ], "redactedDeviceCount": changes[ "redactedDeviceCount" ], }, ) ) self._session.flush() return result def _export_mapping(self, user: User, now: datetime) -> dict[str, object]: licenses = self._bounded_scalars( select(License) .where(License.owner_user_id == user.id) .order_by(License.id), "owned licenses", ) license_ids = tuple(row.id for row in licenses) grants = self._rows_for_ids( Grant, Grant.license_id, license_ids, "grants", ) subscriptions = self._rows_for_ids( Subscription, Subscription.license_id, license_ids, "subscriptions", ) subscription_ids = tuple(row.id for row in subscriptions) subscription_items = self._rows_for_ids( SubscriptionItem, SubscriptionItem.subscription_id, subscription_ids, "subscription items", ) devices = self._bounded_scalars( select(Device).where(Device.owner_user_id == user.id).order_by(Device.id), "devices", ) device_ids = tuple(row.id for row in devices) installations = self._rows_for_ids( DeviceInstallation, DeviceInstallation.device_id, device_ids, "device installations", ) activations = self._rows_for_ids( Activation, Activation.device_id, device_ids, "activations", ) memberships = self._bounded_scalars( select(Membership) .where(Membership.user_id == user.id) .order_by(Membership.organization_id, Membership.id), "memberships", ) organization_ids = tuple( sorted({row.organization_id for row in memberships}) ) organizations = self._rows_for_ids( Organization, Organization.id, organization_ids, "organizations", ) seats = self._bounded_scalars( select(OrganizationSeat) .where(OrganizationSeat.assigned_user_id == user.id) .order_by(OrganizationSeat.organization_id, OrganizationSeat.id), "organization seats", ) invitations = self._bounded_scalars( select(OrganizationInvitation) .where( or_( OrganizationInvitation.created_by_user_id == user.id, OrganizationInvitation.accepted_by_user_id == user.id, ) ) .order_by(OrganizationInvitation.id), "organization invitations", ) redemptions = self._bounded_scalars( select(SerialRedemption) .where(SerialRedemption.actor_user_id == user.id) .order_by(SerialRedemption.created_at, SerialRedemption.id), "serial redemptions", ) deliveries = self._bounded_scalars( select(NotificationDelivery) .where(NotificationDelivery.user_id == user.id) .order_by(NotificationDelivery.created_at, NotificationDelivery.id), "notification deliveries", ) suppressions = self._bounded_scalars( select(NotificationSuppression) .where(NotificationSuppression.user_id == user.id) .order_by(NotificationSuppression.created_at, NotificationSuppression.id), "notification suppressions", ) privacy_requests = self._bounded_scalars( select(PrivacyRequest) .where(PrivacyRequest.user_id == user.id) .order_by(PrivacyRequest.requested_at, PrivacyRequest.id), "privacy requests", ) audit_events = self._bounded_scalars( select(AuditEvent) .where( or_( AuditEvent.actor_id == user.id, ( (AuditEvent.target_type == "user") & (AuditEvent.target_id == user.id) ), ) ) .order_by(AuditEvent.occurred_at, AuditEvent.id), "audit events", ) grants_by_license = self._group_rows(grants, "license_id") subscriptions_by_license = self._group_rows(subscriptions, "license_id") items_by_subscription = self._group_rows( subscription_items, "subscription_id", ) installations_by_device = self._group_rows(installations, "device_id") activations_by_device = self._group_rows(activations, "device_id") organization_by_id = {row.id: row for row in organizations} memberships_by_organization = self._group_rows( memberships, "organization_id", ) seats_by_organization = self._group_rows(seats, "organization_id") return { "schema": PRIVACY_EXPORT_SCHEMA, "schemaVersion": PRIVACY_EXPORT_SCHEMA_VERSION, "generatedAt": _time(now), "account": { "userId": user.id, "status": user.status, "verifiedEmail": user.verified_email, "externalIdentity": { "issuer": user.external_issuer, "subject": user.external_subject, }, "createdAt": _time(user.created_at), "updatedAt": _time(user.updated_at), }, "licenses": [ { "licenseId": license_row.id, "productId": license_row.product_id, "subjectType": license_row.subject_type, "status": license_row.status, "devicePolicyId": license_row.device_policy_id, "createdAt": _time(license_row.created_at), "updatedAt": _time(license_row.updated_at), "grants": [ { "grantId": grant.id, "skuId": grant.sku_id, "sourceType": grant.source_type, "sourceReference": grant.source_reference, "status": grant.status, "startsAt": _time(grant.starts_at), "endsAt": _optional_time(grant.ends_at), } for grant in grants_by_license.get(license_row.id, ()) ], "subscriptions": [ { "subscriptionId": subscription.id, "provider": subscription.provider, "providerSubscriptionId": ( subscription.provider_subscription_id ), "providerCustomerId": subscription.provider_customer_id, "status": subscription.status, "currentPeriodStart": _time( subscription.current_period_start ), "currentPeriodEnd": _time( subscription.current_period_end ), "cancelAtPeriodEnd": ( subscription.cancel_at_period_end ), "billingHoldStatus": ( subscription.billing_hold_status ), "items": [ { "skuId": item.sku_id, "quantity": item.quantity, } for item in items_by_subscription.get( subscription.id, (), ) ], } for subscription in subscriptions_by_license.get( license_row.id, (), ) ], } for license_row in licenses ], "devices": [ { "deviceId": device.id, "friendlyName": device.friendly_name, "deviceType": device.device_type, "status": device.status, "evidenceSchemaVersion": device.evidence_schema_version, "evidenceDigests": dict(device.evidence_digests), "lastSeenAt": _time(device.last_seen_at), "installations": [ { "installationId": installation.installation_id, "keyThumbprint": installation.key_thumbprint, "keyProvider": installation.key_provider, "status": installation.status, "lastSeenAt": _time(installation.last_seen_at), } for installation in installations_by_device.get( device.id, (), ) ], "activations": [ { "activationId": activation.id, "licenseId": activation.license_id, "seatId": activation.seat_id, "status": activation.status, "activatedAt": _time(activation.activated_at), "deactivatedAt": _optional_time( activation.deactivated_at ), "releaseReason": activation.release_reason, } for activation in activations_by_device.get(device.id, ()) ], } for device in devices ], "organizations": [ { "organizationId": organization_id, "displayName": organization_by_id[organization_id].display_name, "status": organization_by_id[organization_id].status, "memberships": [ { "membershipId": membership.id, "role": membership.role, "status": membership.status, "validUntil": _optional_time(membership.valid_until), } for membership in memberships_by_organization.get( organization_id, (), ) ], "assignedSeats": [ { "seatId": seat.id, "licenseId": seat.license_id, "status": seat.status, "assignedAt": _optional_time(seat.assigned_at), } for seat in seats_by_organization.get( organization_id, (), ) ], } for organization_id in organization_ids ], "organizationInvitations": [ { "invitationId": invitation.id, "organizationId": invitation.organization_id, "role": invitation.role, "status": invitation.status, "expiresAt": _time(invitation.expires_at), "createdByAccount": invitation.created_by_user_id == user.id, "acceptedByAccount": invitation.accepted_by_user_id == user.id, "acceptedAt": _optional_time(invitation.accepted_at), } for invitation in invitations ], "serialRedemptions": [ { "redemptionId": redemption.id, "serialRecordId": redemption.serial_id, "licenseId": redemption.license_id, "grantId": redemption.grant_id, "activationId": redemption.activation_id, "snapshotId": redemption.snapshot_id, "createdAt": _time(redemption.created_at), } for redemption in redemptions ], "notifications": { "deliveries": [ { "deliveryId": delivery.id, "status": delivery.status, "attempts": delivery.attempts, "providerMessageId": delivery.provider_message_id, "lastErrorCode": delivery.last_error_code, "deliveredAt": _optional_time(delivery.delivered_at), "feedbackStatus": delivery.feedback_status, "feedbackAt": _optional_time(delivery.feedback_at), } for delivery in deliveries ], "suppressions": [ { "suppressionId": suppression.id, "reason": suppression.reason, "active": suppression.active, "suppressedAt": _time(suppression.suppressed_at), "clearedAt": _optional_time(suppression.cleared_at), } for suppression in suppressions ], }, "privacyRequests": [ { "privacyRequestId": privacy_request.id, "requestType": privacy_request.request_type, "status": privacy_request.status, "requestedAt": _time(privacy_request.requested_at), "cancelledAt": _optional_time(privacy_request.cancelled_at), "completedAt": _optional_time(privacy_request.completed_at), "completionReason": privacy_request.completion_reason, } for privacy_request in privacy_requests ], "auditTimeline": [ { "eventId": event.id, "occurredAt": _time(event.occurred_at), "action": event.action, "targetType": event.target_type, "targetId": event.target_id, } for event in audit_events ], "retentionNotice": { "preservedAfterAccountErasure": [ "licenses_and_entitlements", "billing_and_transaction_projection", "security_and_support_audit", ], "neverIncluded": [ "serial_plaintext", "signing_material", "payment_credentials", "raw_provider_payloads", "device_private_keys", "customer_projects_or_measurements", ], }, } def _request_mapping( self, privacy_request: PrivacyRequest, user: User, now: datetime, *, idempotent_replay: bool, existing_request: bool, ) -> dict[str, object]: if not isinstance(privacy_request, PrivacyRequest): raise TypeError("privacy_request must be a PrivacyRequest") if not isinstance(user, User): raise TypeError("user must be a User") if not isinstance(idempotent_replay, bool): raise TypeError("idempotent_replay must be a Boolean") if not isinstance(existing_request, bool): raise TypeError("existing_request must be a Boolean") result = { "privacyRequestId": privacy_request.id, "requestType": privacy_request.request_type, "status": privacy_request.status, "requestedAt": _time(privacy_request.requested_at), "cancelledAt": _optional_time(privacy_request.cancelled_at), "completedAt": _optional_time(privacy_request.completed_at), "stateGeneration": privacy_request.state_generation, "existingRequest": existing_request, "idempotentReplay": idempotent_replay, } if privacy_request.status == "pending": state = self._deletion_state(privacy_request, user, now) result.update( { "canComplete": state["canComplete"], "blockingReasons": state["blockingReasons"], "consequences": state["consequences"], "customerWarnings": state["customerWarnings"], } ) elif privacy_request.status == "completed": result["completionSummary"] = dict( privacy_request.completion_summary_json ) return result def _administration_mapping( self, privacy_request: PrivacyRequest, now: datetime, ) -> dict[str, object]: user = self._require_user(privacy_request.user_id, lock=False) result = { "privacyRequestId": privacy_request.id, "userId": privacy_request.user_id, "requestType": privacy_request.request_type, "status": privacy_request.status, "requestedAt": _time(privacy_request.requested_at), "cancelledAt": _optional_time(privacy_request.cancelled_at), "completedAt": _optional_time(privacy_request.completed_at), "stateGeneration": privacy_request.state_generation, } if privacy_request.status == "pending": state = self._deletion_state(privacy_request, user, now) result.update( { "canComplete": state["canComplete"], "blockingReasons": state["blockingReasons"], "stateDigest": state["stateDigest"], } ) return result def _deletion_state( self, privacy_request: PrivacyRequest, user: User, now: datetime, *, lock: bool = False, ) -> dict[str, object]: if not isinstance(privacy_request, PrivacyRequest): raise TypeError("privacy_request must be a PrivacyRequest") if not isinstance(user, User): raise TypeError("user must be a User") if not isinstance(lock, bool): raise TypeError("lock must be a Boolean") owned_licenses = self._query_rows( select(License) .where(License.owner_user_id == user.id) .order_by(License.id), lock=lock, label="owned licenses", ) license_ids = tuple(row.id for row in owned_licenses) grants = self._rows_for_ids( Grant, Grant.license_id, license_ids, "grants", lock=lock, ) subscriptions = self._rows_for_ids( Subscription, Subscription.license_id, license_ids, "subscriptions", lock=lock, ) blocking_subscriptions = tuple( row for row in subscriptions if row.status not in PRIVACY_TERMINAL_SUBSCRIPTION_STATUSES ) memberships = self._query_rows( select(Membership) .where( Membership.user_id == user.id, Membership.status == "active", ) .order_by(Membership.organization_id, Membership.id), lock=lock, label="memberships", ) last_owner_organizations = [] owner_peer_state = [] for membership in memberships: if ( membership.role != "owner" or ( membership.valid_until is not None and _utc(membership.valid_until) <= now ) ): continue peer_query = ( select(Membership) .where( Membership.organization_id == membership.organization_id, Membership.user_id != user.id, Membership.role == "owner", Membership.status == "active", or_( Membership.valid_until.is_(None), Membership.valid_until > now, ), ) .order_by(Membership.id) ) if lock: peer_query = peer_query.with_for_update() other_owners = tuple(self._session.scalars(peer_query).all()) owner_peer_state.extend( { "id": row.id, "organizationId": row.organization_id, "status": row.status, "validUntil": _optional_time(row.valid_until), } for row in other_owners ) if not other_owners: last_owner_organizations.append(membership.organization_id) seats = self._query_rows( select(OrganizationSeat) .where( OrganizationSeat.assigned_user_id == user.id, OrganizationSeat.status == "assigned", ) .order_by(OrganizationSeat.id), lock=lock, label="organization seats", ) devices = self._query_rows( select(Device) .where(Device.owner_user_id == user.id) .order_by(Device.id), lock=lock, label="devices", ) device_ids = tuple(row.id for row in devices) seat_ids = tuple(row.id for row in seats) activation_query = select(Activation).where(Activation.status == "active") activation_predicates = [] if device_ids: activation_predicates.append(Activation.device_id.in_(device_ids)) if seat_ids: activation_predicates.append(Activation.seat_id.in_(seat_ids)) activations = ( () if not activation_predicates else self._query_rows( activation_query.where(or_(*activation_predicates)).order_by( Activation.id ), lock=lock, label="activations", ) ) notification_delivery_count = self._count( NotificationDelivery, NotificationDelivery.user_id == user.id, ) notification_suppression_count = self._count( NotificationSuppression, NotificationSuppression.user_id == user.id, ) idempotency_count = self._count( IdempotencyRecord, IdempotencyRecord.subject_key == user.id, ) retained_audit_count = self._count( AuditEvent, or_( AuditEvent.actor_id == user.id, ( (AuditEvent.target_type == "user") & (AuditEvent.target_id == user.id) ), ), ) blockers = [] if user.is_server_admin: blockers.append( { "code": "server_administrator_must_be_transferred", "resourceIds": [user.id], } ) if blocking_subscriptions: blockers.append( { "code": "subscription_must_be_cancelled_or_resolved", "resourceIds": sorted(row.id for row in blocking_subscriptions), } ) if last_owner_organizations: blockers.append( { "code": "organization_ownership_must_be_transferred", "resourceIds": sorted(set(last_owner_organizations)), } ) state_document = { "privacyRequestId": privacy_request.id, "requestStatus": privacy_request.status, "stateGeneration": privacy_request.state_generation, "userId": user.id, "userStatus": user.status, "isServerAdmin": user.is_server_admin, "licenses": [ {"id": row.id, "status": row.status} for row in owned_licenses ], "grants": [ {"id": row.id, "status": row.status} for row in grants ], "subscriptions": [ {"id": row.id, "status": row.status} for row in subscriptions ], "memberships": [ { "id": row.id, "organizationId": row.organization_id, "role": row.role, "status": row.status, "validUntil": _optional_time(row.valid_until), } for row in memberships ], "otherActiveOwners": sorted( owner_peer_state, key=lambda value: value["id"], ), "seats": [ { "id": row.id, "status": row.status, "assignmentGeneration": row.assignment_generation, } for row in seats ], "devices": [ {"id": row.id, "status": row.status} for row in devices ], "activations": [ {"id": row.id, "status": row.status} for row in activations ], "notificationDeliveryCount": notification_delivery_count, "notificationSuppressionCount": notification_suppression_count, "idempotencyCount": idempotency_count, "blockers": blockers, } consequences = { "preservedOwnedLicenseCount": len(owned_licenses), "preservedGrantCount": len(grants), "preservedSubscriptionRecordCount": len(subscriptions), "willRemoveMembershipCount": len(memberships), "willReleaseSeatCount": len(seats), "willReleaseActivationCount": len(activations), "willRedactDeviceCount": len(devices), "willDeleteNotificationDeliveryCount": notification_delivery_count, "willDeleteNotificationSuppressionCount": notification_suppression_count, "willDeleteIdempotencyRecordCount": idempotency_count, "retainedAuditEventCount": retained_audit_count, } return { "privacyRequestId": privacy_request.id, "userId": user.id, "status": privacy_request.status, "canComplete": not blockers, "blockingReasons": blockers, "stateDigest": hashlib.sha256( canonicalize_json(state_document) ).hexdigest(), "preservedOwnedLicenseCount": len(owned_licenses), "preservedGrantCount": len(grants), "preservedSubscriptionRecordCount": len(subscriptions), "retainedAuditEventCount": retained_audit_count, "consequences": consequences, "customerWarnings": [ "account_sign_in_and_license_recovery_will_end", "owned_licenses_grants_and_transaction_records_are_preserved", "existing_offline_documents_may_remain_valid_until_expiry", "a_later_sign_in_creates_a_new_unlinked_account", ], } def _apply_erasure(self, user: User, now: datetime) -> dict[str, int]: devices = self._query_rows( select(Device) .where(Device.owner_user_id == user.id) .order_by(Device.id), lock=True, label="devices", ) device_ids = tuple(row.id for row in devices) seats = self._query_rows( select(OrganizationSeat) .where( OrganizationSeat.assigned_user_id == user.id, OrganizationSeat.status == "assigned", ) .order_by(OrganizationSeat.id), lock=True, label="organization seats", ) seat_ids = tuple(row.id for row in seats) activation_predicates = [] if device_ids: activation_predicates.append(Activation.device_id.in_(device_ids)) if seat_ids: activation_predicates.append(Activation.seat_id.in_(seat_ids)) activations = ( () if not activation_predicates else self._query_rows( select(Activation) .where( Activation.status == "active", or_(*activation_predicates), ) .order_by(Activation.id), lock=True, label="activations", ) ) for activation in activations: activation.status = "deactivated" activation.deactivated_at = now activation.release_reason = PRIVACY_ERASURE_RELEASE_REASON installations = self._rows_for_ids( DeviceInstallation, DeviceInstallation.device_id, device_ids, "device installations", lock=True, ) for installation in installations: installation.installation_id = new_identifier("erased_installation") installation.public_key_der = b"\x00" installation.key_thumbprint = new_identifier("erased_key") installation.key_provider = "privacy_erased" installation.status = "released" installation.last_seen_at = now for device in devices: device.friendly_name = PRIVACY_ERASED_DEVICE_NAME device.status = "released" device.evidence_digests = {} device.last_seen_at = now for seat in seats: seat.assigned_user_id = None seat.status = "available" seat.assigned_at = None seat.released_at = now seat.last_reassigned_at = now seat.assignment_generation += 1 memberships = self._query_rows( select(Membership) .where( Membership.user_id == user.id, Membership.status == "active", ) .order_by(Membership.id), lock=True, label="memberships", ) for membership in memberships: membership.status = "removed" membership.valid_until = now invitation_count = 0 if user.verified_email is not None: invited_email_digest = _email_digest( self._fingerprint_pepper, user.verified_email, ) invitations = self._query_rows( select(OrganizationInvitation) .where( OrganizationInvitation.invited_email_digest == invited_email_digest ) .order_by(OrganizationInvitation.id), lock=True, label="organization invitations", ) for invitation in invitations: invitation.invited_email_digest = _redaction_digest() if invitation.status == "pending": invitation.status = "cancelled" invitation.token_digest = _redaction_digest() invitation_count += 1 deliveries = self._query_rows( select(NotificationDelivery) .where(NotificationDelivery.user_id == user.id) .order_by(NotificationDelivery.id), lock=True, label="notification deliveries", ) delivery_ids = tuple(row.id for row in deliveries) feedback_rows = self._rows_for_ids( NotificationFeedbackEvent, NotificationFeedbackEvent.delivery_id, delivery_ids, "notification feedback", lock=True, ) for feedback in feedback_rows: feedback.delivery_id = None feedback.provider_message_id = None for delivery in deliveries: self._session.delete(delivery) suppressions = self._query_rows( select(NotificationSuppression) .where(NotificationSuppression.user_id == user.id) .order_by(NotificationSuppression.id), lock=True, label="notification suppressions", ) for suppression in suppressions: self._session.delete(suppression) idempotency_records = self._query_rows( select(IdempotencyRecord) .where(IdempotencyRecord.subject_key == user.id) .order_by(IdempotencyRecord.id), lock=True, label="idempotency records", ) for record in idempotency_records: self._session.delete(record) user.external_issuer = PRIVACY_ERASED_ISSUER user.external_subject = new_identifier("erased_subject") user.verified_email = None user.status = "privacy_erased" user.is_server_admin = False return { "releasedActivationCount": len(activations), "redactedDeviceCount": len(devices), "redactedInstallationCount": len(installations), "releasedSeatCount": len(seats), "removedMembershipCount": len(memberships), "redactedInvitationCount": invitation_count, "deletedNotificationDeliveryCount": len(deliveries), "deletedNotificationSuppressionCount": len(suppressions), "deletedIdempotencyRecordCount": len(idempotency_records), } def _require_active_user(self, user_id: str, *, lock: bool) -> User: user = self._require_user(user_id, lock=lock) if user.status != "active": raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "account is not active", status_code=403, ) return user def _require_user(self, user_id: str, *, lock: bool) -> User: query = select(User).where(User.id == user_id) if lock: query = query.with_for_update() user = self._session.scalar(query) if user is None: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "account was not found", status_code=403, ) return user def _require_server_admin(self, user_id: str, *, lock: bool) -> User: user = self._require_active_user(user_id, lock=lock) if not user.is_server_admin: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "privacy administration requires an active administrator", status_code=403, ) return user def _require_privacy_request( self, request_id: str, *, lock: bool, ) -> PrivacyRequest: query = select(PrivacyRequest).where(PrivacyRequest.id == request_id) if lock: query = query.with_for_update() privacy_request = self._session.scalar(query) if privacy_request is None: raise ServerLicensingError( ServerErrorCode.CONFLICT, "privacy request was not found", status_code=404, ) return privacy_request @staticmethod def _deletion_reason(value: str) -> str: normalized = _bounded_text(value, "reason_code", 64) if normalized not in PRIVACY_DELETION_REASONS: raise ValueError("reason_code is invalid for account deletion") return normalized def _bounded_scalars( self, query, label: str, *, maximum: int = MAXIMUM_SELF_SERVICE_PRIVACY_ROWS, ) -> tuple[object, ...]: _validate_limit(maximum, MAXIMUM_SELF_SERVICE_PRIVACY_ROWS, "maximum") rows = tuple(self._session.scalars(query.limit(maximum + 1)).all()) if len(rows) > maximum: raise ServerLicensingError( ServerErrorCode.CONFLICT, f"{label} exceed the self-service privacy bound; use support-assisted export", status_code=409, ) return rows def _query_rows( self, query, *, lock: bool, label: str, ) -> tuple[object, ...]: if not isinstance(lock, bool): raise TypeError("lock must be a Boolean") if lock: query = query.with_for_update() return self._bounded_scalars(query, label) def _rows_for_ids( self, model, field, identifiers: Sequence[str], label: str, *, lock: bool = False, ) -> tuple[object, ...]: if isinstance(identifiers, (str, bytes)) or not isinstance( identifiers, Sequence, ): raise TypeError("identifiers must be a sequence") if not identifiers: return () query = select(model).where(field.in_(tuple(identifiers))).order_by(model.id) return self._query_rows(query, lock=lock, label=label) @staticmethod def _group_rows( rows: Sequence[object], attribute_name: str, ) -> dict[str, tuple[object, ...]]: if isinstance(rows, (str, bytes)) or not isinstance(rows, Sequence): raise TypeError("rows must be a sequence") if not isinstance(attribute_name, str) or not attribute_name.strip(): raise ValueError("attribute_name must be a non-empty string") grouped: dict[str, list[object]] = {} for row in rows: key = getattr(row, attribute_name) if not isinstance(key, str): raise TypeError("grouping attribute must be a string") grouped.setdefault(key, []).append(row) return {key: tuple(values) for key, values in grouped.items()} def _count(self, model, predicate) -> int: value = self._session.scalar( select(func.count()).select_from(model).where(predicate) ) if isinstance(value, bool) or not isinstance(value, int) or value < 0: raise RuntimeError("privacy row count is invalid") return value def purge_privacy_requests( session: Session, *, now: datetime, correlation_id: str, retention_days: int = DEFAULT_PRIVACY_REQUEST_RETENTION_DAYS, maximum_requests: int = MAXIMUM_PRIVACY_PURGE_ROWS, ) -> dict[str, object]: if not isinstance(session, Session): raise TypeError("session must be a SQLAlchemy Session") normalized_now = _utc(now) normalized_correlation = validate_identifier( correlation_id, "correlation_id", ) if isinstance(retention_days, bool) or not isinstance(retention_days, int): raise TypeError("retention_days must be an integer") if ( retention_days < MINIMUM_PRIVACY_REQUEST_RETENTION_DAYS or retention_days > MAXIMUM_PRIVACY_REQUEST_RETENTION_DAYS ): raise ValueError( "retention_days must be between " f"{MINIMUM_PRIVACY_REQUEST_RETENTION_DAYS} and " f"{MAXIMUM_PRIVACY_REQUEST_RETENTION_DAYS}" ) bounded_maximum = _validate_limit( maximum_requests, MAXIMUM_PRIVACY_PURGE_ROWS, "maximum_requests", ) cutoff = normalized_now - timedelta(days=retention_days) with session.begin(): rows = tuple( session.scalars( select(PrivacyRequest) .where( or_( ( (PrivacyRequest.status == "completed") & (PrivacyRequest.completed_at < cutoff) ), ( (PrivacyRequest.status == "cancelled") & (PrivacyRequest.cancelled_at < cutoff) ), ) ) .order_by(PrivacyRequest.requested_at, PrivacyRequest.id) .limit(bounded_maximum) .with_for_update() ).all() ) for row in rows: session.delete(row) retained_audit_count = session.scalar( select(func.count()).select_from(AuditEvent) ) session.add( AuditEvent( id=new_identifier("audit"), actor_type="system", actor_id=None, action="privacy.requests_purged", target_type="privacy_retention", target_id=None, reason="retention_policy", correlation_id=normalized_correlation, source_address_digest=None, metadata_json={ "purgedPrivacyRequestCount": len(rows), "retainedAuditEventCount": retained_audit_count, "retentionDays": retention_days, }, ) ) return { "purgedPrivacyRequestCount": len(rows), "retainedAuditEventCount": retained_audit_count, "retentionDays": retention_days, "cutoff": _time(cutoff), } __all__ = [ "AccountPrivacyService", "DEFAULT_PRIVACY_REQUEST_RETENTION_DAYS", "MAXIMUM_PRIVACY_PURGE_ROWS", "PRIVACY_EXPORT_SCHEMA", "PRIVACY_EXPORT_SCHEMA_VERSION", "purge_privacy_requests", ]