"""Guarded administrator operations for serial issuance batches.""" from __future__ import annotations from datetime import datetime, timedelta, timezone import hashlib import hmac import re from typing import Callable, Mapping, Optional, Sequence from sqlalchemy import select from sqlalchemy.orm import Session from licensing_shared.canonical_json import canonicalize_json from licensing_shared.constants import validate_identifier from .constants import ( DEFAULT_IDEMPOTENCY_DAYS, MAX_IDEMPOTENCY_KEY_CHARACTERS, MAX_REASON_CHARACTERS, SERIAL_BATCH_REVOCATION_REASONS, ) from .errors import ServerErrorCode, ServerLicensingError from .models import AuditEvent, IdempotencyRecord, Serial, SerialBatch, User from .security import new_identifier SERIAL_BATCH_REVOCATION_SCOPE = "admin.serial_batch.revoke" SERIAL_BATCH_STATE_DIGEST_PATTERN = re.compile(r"^[0-9a-f]{64}$") SERIAL_BATCH_STATUSES = frozenset(("active", "revoked")) SERIAL_STATUSES = frozenset(("active", "redeemed", "revoked")) 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 _bounded_text(value: str, field_name: str, maximum: int) -> str: if not isinstance(value, str): raise TypeError(f"{field_name} must be a string") normalized = value.strip() if not normalized or len(normalized) > maximum: raise ValueError(f"{field_name} is empty or 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 _reason_code(value: str) -> str: normalized = validate_identifier(value, "reason_code") if normalized not in SERIAL_BATCH_REVOCATION_REASONS: raise ValueError("reason_code is not allowed for serial batch revocation") return normalized def _state_digest(value: Mapping[str, object]) -> str: if not isinstance(value, Mapping): raise TypeError("value must be a mapping") return hashlib.sha256(canonicalize_json(value)).hexdigest() class SerialAdministrationService: def __init__( self, session: Session, *, now_factory: Callable[[], datetime] = lambda: datetime.now(timezone.utc), ) -> None: if not isinstance(session, Session): raise TypeError("session must be a SQLAlchemy Session") if not callable(now_factory): raise TypeError("now_factory must be callable") self._session = session self._now_factory = now_factory def _now(self) -> datetime: return _utc(self._now_factory()) def preview_batch_revocation( self, batch_id: str, actor_user_id: str, reason_code: str, correlation_id: str, *, note: Optional[str] = None, ) -> dict[str, object]: normalized_batch = validate_identifier(batch_id, "batch_id") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_reason = _reason_code(reason_code) _optional_note(note) normalized_correlation = validate_identifier(correlation_id, "correlation_id") with self._session.begin(): self._require_server_admin(normalized_actor) batch = self._session.get(SerialBatch, normalized_batch) if batch is None: raise ServerLicensingError( ServerErrorCode.SERIAL_BATCH_NOT_FOUND, "serial batch was not found", status_code=404, ) serials = self._serials(batch.id, lock=False) result = self._preview_mapping(batch, serials) result["reasonCode"] = normalized_reason result["correlationId"] = normalized_correlation return result def revoke_batch( self, batch_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_batch = validate_identifier(batch_id, "batch_id") normalized_actor = validate_identifier(actor_user_id, "actor_user_id") normalized_reason = _reason_code(reason_code) normalized_note = _optional_note(note) normalized_state_digest = _bounded_text( expected_state_digest, "expected_state_digest", 64, ) if SERIAL_BATCH_STATE_DIGEST_PATTERN.fullmatch(normalized_state_digest) is None: raise ValueError("expected_state_digest must be 64 lowercase hexadecimal characters") normalized_key = _bounded_text( idempotency_key, "idempotency_key", MAX_IDEMPOTENCY_KEY_CHARACTERS, ) normalized_correlation = validate_identifier(correlation_id, "correlation_id") request_digest = hashlib.sha256( canonicalize_json( { "batchId": normalized_batch, "actorUserId": normalized_actor, "reasonCode": normalized_reason, "note": normalized_note, "expectedStateDigest": normalized_state_digest, "correlationId": normalized_correlation, } ) ).digest() now = self._now() with self._session.begin(): actor = self._require_server_admin(normalized_actor) batch = self._session.scalar( select(SerialBatch) .where(SerialBatch.id == normalized_batch) .with_for_update() ) if batch is None: raise ServerLicensingError( ServerErrorCode.SERIAL_BATCH_NOT_FOUND, "serial batch was not found", status_code=404, ) replay = self._load_idempotent_response( normalized_batch, normalized_key, request_digest, now, ) if replay is not None: result = dict(replay) result["idempotentReplay"] = True return result serials = self._serials(batch.id, lock=True) preview = self._preview_mapping(batch, serials) if not hmac.compare_digest( str(preview["stateDigest"]), normalized_state_digest, ): raise ServerLicensingError( ServerErrorCode.CONFLICT, "serial batch changed after the revocation preview; preview again", status_code=409, ) if batch.status != "active" or not preview["canExecute"]: raise ServerLicensingError( ServerErrorCode.CONFLICT, "serial batch has no active serials to revoke", status_code=409, ) newly_revoked = 0 for serial in serials: if serial.status == "active": serial.status = "revoked" newly_revoked += 1 batch.status = "revoked" batch.revoked_at = now batch.revoked_by_user_id = actor.id batch.revocation_reason = normalized_reason result = self._preview_mapping(batch, serials) result.update( { "reasonCode": normalized_reason, "correlationId": normalized_correlation, "previousStateDigest": normalized_state_digest, "newlyRevokedSerialCount": newly_revoked, "preservedRedeemedLicenseCount": result[ "redeemedLicenseCount" ], "executed": True, "idempotentReplay": False, } ) self._store_idempotent_response( normalized_batch, normalized_key, request_digest, result, now, ) self._session.add( AuditEvent( id=new_identifier("audit"), actor_type="user", actor_id=actor.id, action="serial.batch_revoked", target_type="serial_batch", target_id=batch.id, reason=normalized_note or normalized_reason, correlation_id=normalized_correlation, source_address_digest=None, metadata_json={ "activeSerialCount": result["activeSerialCount"], "redeemedLicenseCount": result["redeemedLicenseCount"], "redeemedSerialCount": result["redeemedSerialCount"], "revokedSerialCount": result["revokedSerialCount"], "wouldRevokeSerialCount": newly_revoked, "reasonCode": normalized_reason, "skuId": batch.sku_id, "stateDigest": result["stateDigest"], }, ) ) return result def _require_server_admin(self, actor_user_id: str) -> User: actor = self._session.get(User, actor_user_id) if actor is None or actor.status != "active" or not actor.is_server_admin: raise ServerLicensingError( ServerErrorCode.AUTHORIZATION_DENIED, "serial batch administration requires an active administrator", status_code=403, ) return actor def _serials(self, batch_id: str, *, lock: bool) -> Sequence[Serial]: query = select(Serial).where(Serial.batch_id == batch_id).order_by(Serial.id) if lock: query = query.with_for_update() return tuple(self._session.scalars(query).all()) def _preview_mapping( self, batch: SerialBatch, serials: Sequence[Serial], ) -> dict[str, object]: if batch.status not in SERIAL_BATCH_STATUSES: self._invalid_state("serial batch has an invalid status") if len(serials) != batch.quantity: self._invalid_state("serial batch quantity does not match its serial rows") if batch.status == "active" and any( value is not None for value in ( batch.revoked_at, batch.revoked_by_user_id, batch.revocation_reason, ) ): self._invalid_state("active serial batch contains revocation metadata") if batch.status == "revoked" and any( value is None for value in ( batch.revoked_at, batch.revoked_by_user_id, batch.revocation_reason, ) ): self._invalid_state("revoked serial batch is missing revocation metadata") counts = {status: 0 for status in SERIAL_STATUSES} redeemed_license_ids = set() serial_state = [] for serial in serials: if serial.status not in SERIAL_STATUSES: self._invalid_state("serial batch contains an invalid serial status") if serial.status == "redeemed": if serial.redeemed_license_id is None: self._invalid_state("redeemed serial is missing its license") redeemed_license_ids.add(serial.redeemed_license_id) elif serial.redeemed_license_id is not None: self._invalid_state("unredeemed serial unexpectedly references a license") counts[serial.status] += 1 serial_state.append( { "serialId": serial.id, "status": serial.status, "redeemedLicenseId": serial.redeemed_license_id, } ) state = { "batchId": batch.id, "batchStatus": batch.status, "revokedAt": ( None if batch.revoked_at is None else _utc(batch.revoked_at).isoformat() ), "revokedByUserId": batch.revoked_by_user_id, "revocationReason": batch.revocation_reason, "serials": serial_state, } active_count = counts["active"] return { "batchId": batch.id, "skuId": batch.sku_id, "batchStatus": batch.status, "quantity": batch.quantity, "activeSerialCount": active_count, "redeemedSerialCount": counts["redeemed"], "revokedSerialCount": counts["revoked"], "redeemedLicenseCount": len(redeemed_license_ids), "wouldRevokeSerialCount": active_count, "stateDigest": _state_digest(state), "canExecute": batch.status == "active" and active_count > 0, } def _load_idempotent_response( self, batch_id: str, idempotency_key: str, request_digest: bytes, now: datetime, ) -> Optional[dict[str, object]]: record = self._session.scalar( select(IdempotencyRecord) .where( IdempotencyRecord.scope == SERIAL_BATCH_REVOCATION_SCOPE, IdempotencyRecord.subject_key == batch_id, IdempotencyRecord.idempotency_key == idempotency_key, ) .with_for_update() ) if record is None: return None if _utc(record.expires_at) <= now: self._session.delete(record) self._session.flush() return None if not hmac.compare_digest(record.request_digest, request_digest): raise ServerLicensingError( ServerErrorCode.IDEMPOTENCY_CONFLICT, "idempotency key was already used for a different request", status_code=409, ) return record.response_json def _store_idempotent_response( self, batch_id: str, idempotency_key: str, request_digest: bytes, response: dict[str, object], now: datetime, ) -> None: self._session.add( IdempotencyRecord( id=new_identifier("idempotency"), scope=SERIAL_BATCH_REVOCATION_SCOPE, subject_key=batch_id, idempotency_key=idempotency_key, request_digest=request_digest, response_status=200, response_json=response, expires_at=now + timedelta(days=DEFAULT_IDEMPOTENCY_DAYS), ) ) @staticmethod def _invalid_state(message: str) -> None: raise ServerLicensingError( ServerErrorCode.INTERNAL_ERROR, message, status_code=500, ) __all__ = [ "SERIAL_BATCH_REVOCATION_SCOPE", "SerialAdministrationService", ]