"""Privileged licensing authority maintenance commands. Serial plaintext is written once to an explicit owner-only file and is never printed to stdout or stored by the database. """ from __future__ import annotations import argparse from datetime import datetime, timezone import hashlib import json import os from pathlib import Path import signal import sys import threading from typing import Mapping, Optional, Sequence from sqlalchemy import func, select from sqlalchemy.orm import Session from licensing_shared.catalog import LicensingCatalog, SkuKind, load_builtin_catalog from licensing_shared.canonical_json import canonicalize_json from licensing_shared.constants import ( TRIAL_MAXIMUM_TOTAL_DURATION_HOURS, TRIAL_MINIMUM_AUTHORITY_DURATION_HOURS, validate_identifier, ) from licensing_shared.models import ( SignedLicenseDocument, format_rfc3339, parse_rfc3339, ) from .app.config import ( DatabaseWorkerSettings, NotificationWorkerSettings, ServerSettings, SubscriptionWorkerSettings, ) from .app.billing import StripeBillingProvider from .app.catalog_administration import CatalogAdministrationService from .app.database import create_database_engine from .app.deployment_readiness import ( DEFAULT_MAXIMUM_BACKLOG_AGE_MINUTES, DEPLOYMENT_COMPONENT_API, DEPLOYMENT_COMPONENTS, DEPLOYMENT_COMPONENT_SUBSCRIPTION_WORKER, inspect_deployment_readiness, ) from .app.models import ( NotificationDelivery, NotificationFeedbackEvent, NotificationSuppression, OutboxEvent, ) from .app.notifications import ( CUSTOMER_NOTIFICATION_OUTBOX_EVENT_TYPE, DEFAULT_NOTIFICATION_CLAIM_TIMEOUT_MINUTES, DEFAULT_NOTIFICATION_FEEDBACK_RETENTION_DAYS, DEFAULT_NOTIFICATION_MAXIMUM_AGE_DAYS, NOTIFICATION_SUPPRESSION_CLEAR_REASONS, NotificationFeedbackType, SmtpCustomerNotificationProvider, clear_notification_suppression, process_customer_notification_outbox, purge_notification_feedback_events, recover_stale_customer_notification_claims, record_notification_feedback, requeue_failed_customer_notifications, ) from .app.constants import ( CATALOG_RELEASE_REASONS, PRIVACY_DELETION_REASONS, SIGNING_KEY_CHANGE_REASONS, SIGNING_KEY_COMPROMISE_REASONS, ) from .app.security import Ed25519SnapshotSigner, new_identifier from .app.privacy import ( AccountPrivacyService, DEFAULT_PRIVACY_REQUEST_RETENTION_DAYS, MAXIMUM_PRIVACY_PURGE_ROWS, purge_privacy_requests, ) from .app.signing_key_administration import SigningKeyAdministrationService from .app.services import LicensingService, SerialGenerationResult from .app.subscriptions import ( process_subscription_outbox, requeue_failed_subscription_outbox, ) from .app.subscription_maintenance import ( DEFAULT_FAILED_PROVIDER_RETENTION_DAYS, DEFAULT_PROCESSED_PROVIDER_RETENTION_DAYS, purge_subscription_provider_records, reconcile_stripe_subscriptions, ) CLI_SUCCESS = 0 CLI_FAILURE = 1 CLI_USAGE_ERROR = 2 SERIAL_OUTPUT_SCHEMA = "apolon.licensing.serial-batch-export" SERIAL_OUTPUT_SCHEMA_VERSION = 1 SERIAL_OUTPUT_LINE_TERMINATOR = b"\r\n" DEFAULT_OUTBOX_BATCH_SIZE = 100 DEFAULT_OUTBOX_POLL_SECONDS = 5.0 MAXIMUM_OUTBOX_POLL_SECONDS = 300.0 class _SerialGenerationOnlySigner: def __init__(self, key_id: str) -> None: self._key_id = validate_identifier(key_id, "key_id") @property def key_id(self) -> str: return self._key_id def public_key_bytes(self) -> bytes: raise RuntimeError("serial issuance does not expose signing key material") def sign_payload(self, _payload: Mapping[str, object]) -> SignedLicenseDocument: raise RuntimeError("serial issuance cannot sign entitlement documents") def _write_owner_only_serial_export(path: Path, document: Mapping[str, object]) -> None: if not isinstance(path, Path): raise TypeError("path must be a Path") if not path.name: raise ValueError("path must identify an output file") if not isinstance(document, Mapping): raise TypeError("document must be a mapping") parent = path.parent if not parent.is_dir() or parent.is_symlink(): raise ValueError("serial output parent must be an existing non-symlink directory") body = json.dumps( dict(document), ensure_ascii=True, indent=2, sort_keys=True, ).encode("utf-8") + SERIAL_OUTPUT_LINE_TERMINATOR flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL if hasattr(os, "O_NOFOLLOW"): flags |= os.O_NOFOLLOW descriptor = os.open(path, flags, 0o600) try: with os.fdopen(descriptor, "wb", closefd=True) as output: output.write(body) output.flush() os.fsync(output.fileno()) except Exception: try: path.unlink(missing_ok=True) except OSError: pass raise if os.name != "nt" and path.stat().st_mode & 0o077: path.unlink(missing_ok=True) raise PermissionError("serial export permissions are not owner-only") def _settings() -> ServerSettings: return ServerSettings.from_environment() def _database_worker_settings() -> DatabaseWorkerSettings: return DatabaseWorkerSettings.from_environment() def _subscription_worker_settings() -> SubscriptionWorkerSettings: return SubscriptionWorkerSettings.from_environment() def _notification_worker_settings() -> NotificationWorkerSettings: return NotificationWorkerSettings.from_environment() def _configured_signer(settings: ServerSettings) -> Ed25519SnapshotSigner: if not isinstance(settings, ServerSettings): raise TypeError("settings must be ServerSettings") return Ed25519SnapshotSigner.from_private_key_file( settings.signing_key_id, settings.signing_private_key_path, require_owner_only=settings.allow_production_file_signer, ) def _list_serial_skus(_arguments: argparse.Namespace) -> int: catalog = load_builtin_catalog() rows = [] for sku in catalog.skus.values(): if sku.active and sku.kind in ( SkuKind.EDITION, SkuKind.ADDON, SkuKind.TRIAL, ): rows.append( { "skuId": sku.sku_id, "label": sku.label, "kind": sku.kind.value, "accountRequired": sku.account_required, } ) print(json.dumps(rows, ensure_ascii=True, indent=2, sort_keys=True)) return CLI_SUCCESS def _require_published_runtime_catalog( session: Session, catalog: LicensingCatalog, ) -> None: if not isinstance(session, Session): raise TypeError("session must be a SQLAlchemy Session") if not isinstance(catalog, LicensingCatalog): raise TypeError("catalog must be LicensingCatalog") status = CatalogAdministrationService(session, catalog).status() if status["catalogReady"] is not True: raise RuntimeError( "runtime catalog is not the exact active reviewed publication" ) def _generate_serials(arguments: argparse.Namespace) -> int: if not isinstance(arguments, argparse.Namespace): raise TypeError("arguments must be an argparse Namespace") settings = _settings() output_path = Path(arguments.output).expanduser().resolve() if output_path.exists() or output_path.is_symlink(): raise FileExistsError("serial output already exists; choose a new file") deadline = ( None if arguments.redemption_deadline is None else parse_rfc3339(arguments.redemption_deadline, "redemption_deadline") ) engine = create_database_engine(settings) export_written = False def export_plaintext(result: SerialGenerationResult) -> None: nonlocal export_written _write_owner_only_serial_export( output_path, { "schema": SERIAL_OUTPUT_SCHEMA, "schemaVersion": SERIAL_OUTPUT_SCHEMA_VERSION, "batchId": result.batch_id, "skuId": result.sku_id, "quantity": len(result.serials), "surfaceProfileId": result.surface_profile_id, "trialDurationHours": result.trial_duration_hours, "createdAt": datetime.now(timezone.utc) .isoformat() .replace("+00:00", "Z"), "serials": list(result.serials), "warning": ( "Shown once. Deliver securely, then remove this file according to policy." ), }, ) export_written = True try: try: with Session(engine) as session: catalog = load_builtin_catalog() _require_published_runtime_catalog(session, catalog) result = LicensingService( session, catalog, _SerialGenerationOnlySigner(settings.signing_key_id), settings.serial_pepper, settings.fingerprint_pepper, ).generate_serial_batch( arguments.sku, arguments.quantity, arguments.admin_user_id, arguments.reason, new_identifier("correlation"), campaign=arguments.campaign, redemption_deadline=deadline, plaintext_exporter=export_plaintext, surface_profile_id=arguments.surface_profile_id, trial_duration_hours=arguments.trial_duration_hours, ) except Exception: if export_written: output_path.unlink(missing_ok=True) raise finally: engine.dispose() print( f"Generated batch {result.batch_id} with {len(result.serials)} serial(s). " f"Plaintext was written only to {output_path}." ) return CLI_SUCCESS def _export_signing_public_key(arguments: argparse.Namespace) -> int: if not isinstance(arguments, argparse.Namespace): raise TypeError("arguments must be an argparse Namespace") settings = _settings() signer = _configured_signer(settings) engine = create_database_engine(settings) try: with Session(engine) as session: status = SigningKeyAdministrationService(session, signer).status() finally: engine.dispose() matching_keys = [ value for value in status["keys"] if isinstance(value, Mapping) and value.get("keyId") == signer.key_id ] if len(matching_keys) != 1 or matching_keys[0].get("status") not in ( "staged", "active", ) or matching_keys[0].get("publicKeySha256") != status.get( "configuredPublicKeySha256" ): raise RuntimeError( "configured signer must have staged or active public metadata before export" ) registered_key = matching_keys[0] output_path = Path(arguments.output).expanduser().resolve() _write_owner_only_serial_export( output_path, { "schema": "apolon.licensing.public-key-export", "schemaVersion": 1, "keyId": signer.key_id, "publicKey": registered_key["publicKey"], "notBefore": registered_key["notBefore"], "expiresAt": registered_key["expiresAt"], }, ) print(f"Public verification key {signer.key_id} was written to {output_path}.") return CLI_SUCCESS def _signing_command_identifiers( action: str, payload: Mapping[str, object], ) -> tuple[str, str]: normalized_action = validate_identifier(action, "action") if not isinstance(payload, Mapping): raise TypeError("payload must be a mapping") digest = hashlib.sha256(canonicalize_json(dict(payload))).hexdigest()[:32] return ( f"signing-key-{normalized_action}-{digest}", f"correlation.signing_key_{normalized_action}.{digest}", ) def _catalog_command_identifiers( action: str, payload: Mapping[str, object], ) -> tuple[str, str]: normalized_action = validate_identifier(action, "action") if not isinstance(payload, Mapping): raise TypeError("payload must be a mapping") digest = hashlib.sha256(canonicalize_json(dict(payload))).hexdigest()[:32] return ( f"catalog-{normalized_action}-{digest}", f"correlation.catalog_{normalized_action}.{digest}", ) def _privacy_command_identifiers( action: str, payload: Mapping[str, object], ) -> tuple[str, str]: normalized_action = validate_identifier(action, "action") if not isinstance(payload, Mapping): raise TypeError("payload must be a mapping") digest = hashlib.sha256(canonicalize_json(dict(payload))).hexdigest()[:32] return ( f"privacy-{normalized_action}-{digest}", f"correlation.privacy_{normalized_action}.{digest}", ) def _catalog_status(_arguments: argparse.Namespace) -> int: settings = _settings() catalog = load_builtin_catalog() engine = create_database_engine(settings) try: with Session(engine) as session: status = CatalogAdministrationService(session, catalog).status() finally: engine.dispose() print(json.dumps(status, ensure_ascii=True, indent=2, sort_keys=True)) return CLI_SUCCESS if status["catalogReady"] is True else CLI_FAILURE def _stage_current_catalog(arguments: argparse.Namespace) -> int: if not isinstance(arguments, argparse.Namespace): raise TypeError("arguments must be an argparse Namespace") settings = _settings() catalog = load_builtin_catalog() request = { "actorUserId": arguments.admin_user_id, "catalogSha256": catalog.sha256(), "note": arguments.note, "reasonCode": arguments.reason_code, "revision": catalog.revision, } idempotency_key, correlation_id = _catalog_command_identifiers( "stage", request, ) engine = create_database_engine(settings) try: with Session(engine) as session: result = CatalogAdministrationService( session, catalog, ).stage_candidate( catalog.to_mapping(), arguments.admin_user_id, arguments.reason_code, idempotency_key, correlation_id, note=arguments.note, ) finally: engine.dispose() print(json.dumps(result, ensure_ascii=True, indent=2, sort_keys=True)) return CLI_SUCCESS def _publish_current_catalog(arguments: argparse.Namespace) -> int: if not isinstance(arguments, argparse.Namespace): raise TypeError("arguments must be an argparse Namespace") settings = _settings() catalog = load_builtin_catalog() correlation_payload = { "actorUserId": arguments.admin_user_id, "catalogSha256": catalog.sha256(), "note": arguments.note, "reasonCode": arguments.reason_code, "revision": catalog.revision, } _unused_idempotency, correlation_id = _catalog_command_identifiers( "publish", correlation_payload, ) engine = create_database_engine(settings) try: with Session(engine) as session: preview = CatalogAdministrationService( session, catalog, ).preview_publication( catalog.revision, arguments.admin_user_id, arguments.reason_code, correlation_id, note=arguments.note, ) state_digest = preview.get("stateDigest") if not isinstance(state_digest, str): raise RuntimeError("catalog publication preview has no state digest") if preview.get("canPublish") is not True: raise RuntimeError("current catalog is not a publishable staged candidate") if not arguments.confirm: print( json.dumps( { "confirmationRequired": True, "requiredStateDigest": state_digest, "publication": preview, }, ensure_ascii=True, indent=2, sort_keys=True, ) ) return CLI_SUCCESS if arguments.expected_state_digest is None: raise ValueError("--expected-state-digest is required with --confirm") if arguments.expected_state_digest != state_digest: raise RuntimeError( "catalog registry changed after review; preview and confirm again" ) idempotency_key, _unused_correlation = _catalog_command_identifiers( "publish", { **correlation_payload, "expectedStateDigest": state_digest, }, ) with Session(engine) as session: result = CatalogAdministrationService( session, catalog, ).publish_candidate( catalog.revision, arguments.admin_user_id, arguments.reason_code, state_digest, idempotency_key, correlation_id, note=arguments.note, ) finally: engine.dispose() print(json.dumps(result, ensure_ascii=True, indent=2, sort_keys=True)) return CLI_SUCCESS def _signing_status(_arguments: argparse.Namespace) -> int: settings = _settings() signer = _configured_signer(settings) engine = create_database_engine(settings) try: with Session(engine) as session: status = SigningKeyAdministrationService(session, signer).status() finally: engine.dispose() print(json.dumps(status, ensure_ascii=True, indent=2, sort_keys=True)) return CLI_SUCCESS if status["issuanceReady"] is True else CLI_FAILURE def _register_signing_key(arguments: argparse.Namespace) -> int: if not isinstance(arguments, argparse.Namespace): raise TypeError("arguments must be an argparse Namespace") settings = _settings() signer = _configured_signer(settings) not_before = parse_rfc3339(arguments.not_before, "not_before") expires_at = ( None if arguments.expires_at is None else parse_rfc3339(arguments.expires_at, "expires_at") ) request = { "keyId": signer.key_id, "notBefore": format_rfc3339(not_before, "not_before"), "expiresAt": ( None if expires_at is None else format_rfc3339(expires_at, "expires_at") ), "actorUserId": arguments.admin_user_id, "reasonCode": arguments.reason_code, "note": arguments.note, } idempotency_key, correlation_id = _signing_command_identifiers( "register", request, ) engine = create_database_engine(settings) try: with Session(engine) as session: result = SigningKeyAdministrationService( session, signer, ).register_current_key( not_before, arguments.admin_user_id, arguments.reason_code, idempotency_key, correlation_id, expires_at=expires_at, note=arguments.note, ) finally: engine.dispose() print(json.dumps(result, ensure_ascii=True, indent=2, sort_keys=True)) return CLI_SUCCESS def _activate_signing_key(arguments: argparse.Namespace) -> int: if not isinstance(arguments, argparse.Namespace): raise TypeError("arguments must be an argparse Namespace") settings = _settings() signer = _configured_signer(settings) correlation_payload = { "keyId": signer.key_id, "actorUserId": arguments.admin_user_id, "reasonCode": arguments.reason_code, "note": arguments.note, } _unused_idempotency, correlation_id = _signing_command_identifiers( "activate", correlation_payload, ) engine = create_database_engine(settings) try: with Session(engine) as session: preview = SigningKeyAdministrationService( session, signer, ).preview_current_key_activation( arguments.admin_user_id, arguments.reason_code, correlation_id, note=arguments.note, ) state_digest = preview.get("stateDigest") if not isinstance(state_digest, str): raise RuntimeError("signing-key activation preview has no state digest") if preview.get("canActivateConfiguredKey") is not True: raise RuntimeError( "configured signer is not a valid staged activation candidate" ) if not arguments.confirm: print( json.dumps( { "confirmationRequired": True, "requiredStateDigest": state_digest, "rotation": preview, }, ensure_ascii=True, indent=2, sort_keys=True, ) ) return CLI_SUCCESS if arguments.expected_state_digest is None: raise ValueError("--expected-state-digest is required with --confirm") if arguments.expected_state_digest != state_digest: raise RuntimeError( "signing-key registry changed after review; preview and confirm again" ) idempotency_key, _unused_correlation = _signing_command_identifiers( "activate", { **correlation_payload, "expectedStateDigest": state_digest, }, ) with Session(engine) as session: result = SigningKeyAdministrationService( session, signer, ).activate_current_key( arguments.admin_user_id, arguments.reason_code, state_digest, idempotency_key, correlation_id, note=arguments.note, ) finally: engine.dispose() print(json.dumps(result, ensure_ascii=True, indent=2, sort_keys=True)) return CLI_SUCCESS def _compromise_signing_key(arguments: argparse.Namespace) -> int: if not isinstance(arguments, argparse.Namespace): raise TypeError("arguments must be an argparse Namespace") settings = _settings() signer = _configured_signer(settings) correlation_payload = { "keyId": arguments.key_id, "actorUserId": arguments.admin_user_id, "reasonCode": arguments.reason_code, "note": arguments.note, } _unused_idempotency, correlation_id = _signing_command_identifiers( "compromise", correlation_payload, ) engine = create_database_engine(settings) try: with Session(engine) as session: preview = SigningKeyAdministrationService( session, signer, ).preview_key_compromise( arguments.key_id, arguments.admin_user_id, arguments.reason_code, correlation_id, note=arguments.note, ) state_digest = preview.get("stateDigest") if not isinstance(state_digest, str): raise RuntimeError("signing-key compromise preview has no state digest") if not arguments.confirm: if preview.get("canCompromise") is not True: raise RuntimeError("signing key is already marked compromised") print( json.dumps( { "confirmationRequired": True, "requiredStateDigest": state_digest, "compromise": preview, }, ensure_ascii=True, indent=2, sort_keys=True, ) ) return CLI_SUCCESS if arguments.expected_state_digest is None: raise ValueError("--expected-state-digest is required with --confirm") idempotency_key, _unused_correlation = _signing_command_identifiers( "compromise", { **correlation_payload, "expectedStateDigest": arguments.expected_state_digest, }, ) with Session(engine) as session: result = SigningKeyAdministrationService( session, signer, ).compromise_key( arguments.key_id, arguments.admin_user_id, arguments.reason_code, arguments.expected_state_digest, idempotency_key, correlation_id, note=arguments.note, ) finally: engine.dispose() print(json.dumps(result, ensure_ascii=True, indent=2, sort_keys=True)) return CLI_SUCCESS def _run_outbox_once(arguments: argparse.Namespace) -> int: settings = _subscription_worker_settings() billing_incident_provider = ( None if settings.stripe_secret_key is None else StripeBillingProvider(settings.stripe_secret_key) ) engine = create_database_engine(settings) try: with Session(engine) as session: catalog = load_builtin_catalog() _require_published_runtime_catalog(session, catalog) processed = process_subscription_outbox( session, catalog, now=datetime.now(timezone.utc), maximum_events=arguments.maximum_events, billing_incident_provider=billing_incident_provider, ) finally: engine.dispose() print(f"Processed {processed} licensing provider outbox event(s).") return CLI_SUCCESS def _run_outbox_worker(arguments: argparse.Namespace) -> int: if isinstance(arguments.poll_seconds, bool): raise TypeError("poll_seconds must be a number") poll_seconds = float(arguments.poll_seconds) if poll_seconds <= 0.0 or poll_seconds > MAXIMUM_OUTBOX_POLL_SECONDS: raise ValueError( f"poll_seconds must be between zero and {MAXIMUM_OUTBOX_POLL_SECONDS:g}" ) settings = _subscription_worker_settings() billing_incident_provider = ( None if settings.stripe_secret_key is None else StripeBillingProvider(settings.stripe_secret_key) ) engine = create_database_engine(settings) stop_event = threading.Event() def request_stop(_signal_number: int, _frame: object) -> None: stop_event.set() previous_sigterm = signal.signal(signal.SIGTERM, request_stop) previous_sigint = signal.signal(signal.SIGINT, request_stop) try: catalog = load_builtin_catalog() while not stop_event.is_set(): with Session(engine) as session: _require_published_runtime_catalog(session, catalog) processed = process_subscription_outbox( session, catalog, now=datetime.now(timezone.utc), maximum_events=arguments.maximum_events, billing_incident_provider=billing_incident_provider, ) if processed: print( f"Processed {processed} licensing provider outbox event(s).", flush=True, ) stop_event.wait(poll_seconds) finally: signal.signal(signal.SIGTERM, previous_sigterm) signal.signal(signal.SIGINT, previous_sigint) engine.dispose() return CLI_SUCCESS def _retry_failed_outbox(arguments: argparse.Namespace) -> int: settings = _database_worker_settings() engine = create_database_engine(settings) try: with Session(engine) as session: requeued = requeue_failed_subscription_outbox( session, now=datetime.now(timezone.utc), maximum_events=arguments.maximum_events, maximum_attempts=arguments.maximum_attempts, ) finally: engine.dispose() print(f"Requeued {requeued} failed subscription outbox event(s).") return CLI_SUCCESS def _outbox_status(_arguments: argparse.Namespace) -> int: settings = _database_worker_settings() engine = create_database_engine(settings) try: with Session(engine) as session: rows = session.execute( select(OutboxEvent.status, func.count()).group_by(OutboxEvent.status) ).all() finally: engine.dispose() print(json.dumps(dict(rows), ensure_ascii=True, indent=2, sort_keys=True)) return CLI_SUCCESS def _notification_provider( settings: NotificationWorkerSettings, ) -> SmtpCustomerNotificationProvider: if not isinstance(settings, NotificationWorkerSettings): raise TypeError("settings must be NotificationWorkerSettings") if not settings.email_notifications_enabled: raise ValueError("LICENSING_EMAIL_NOTIFICATIONS_ENABLED=true is required") if any( value is None for value in ( settings.smtp_host, settings.smtp_username, settings.smtp_password, settings.email_from_address, settings.account_portal_url, ) ): raise ValueError("email notification settings are incomplete") assert settings.smtp_host is not None assert settings.smtp_username is not None assert settings.smtp_password is not None assert settings.email_from_address is not None return SmtpCustomerNotificationProvider( settings.smtp_host, settings.smtp_port, settings.smtp_tls_mode, settings.smtp_username, settings.smtp_password, settings.email_from_address, ) def _run_notifications_once(arguments: argparse.Namespace) -> int: settings = _notification_worker_settings() provider = _notification_provider(settings) assert settings.account_portal_url is not None engine = create_database_engine(settings) try: with Session(engine) as session: recovered = recover_stale_customer_notification_claims( session, now=datetime.now(timezone.utc), claim_timeout_minutes=arguments.claim_timeout_minutes, maximum_events=arguments.maximum_events, ) processed = process_customer_notification_outbox( session, load_builtin_catalog(), provider, settings.account_portal_url, now=datetime.now(timezone.utc), maximum_events=arguments.maximum_events, maximum_age_days=arguments.maximum_age_days, notification_pepper=settings.notification_pepper, ) finally: engine.dispose() print( f"Recovered {recovered} stale notification claim(s); " f"processed {processed} notification event(s)." ) return CLI_SUCCESS def _run_notifications_worker(arguments: argparse.Namespace) -> int: if isinstance(arguments.poll_seconds, bool): raise TypeError("poll_seconds must be a number") poll_seconds = float(arguments.poll_seconds) if poll_seconds <= 0.0 or poll_seconds > MAXIMUM_OUTBOX_POLL_SECONDS: raise ValueError( f"poll_seconds must be between zero and {MAXIMUM_OUTBOX_POLL_SECONDS:g}" ) settings = _notification_worker_settings() provider = _notification_provider(settings) assert settings.account_portal_url is not None engine = create_database_engine(settings) stop_event = threading.Event() def request_stop(_signal_number: int, _frame: object) -> None: stop_event.set() previous_sigterm = signal.signal(signal.SIGTERM, request_stop) previous_sigint = signal.signal(signal.SIGINT, request_stop) try: catalog = load_builtin_catalog() while not stop_event.is_set(): with Session(engine) as session: recovered = recover_stale_customer_notification_claims( session, now=datetime.now(timezone.utc), claim_timeout_minutes=arguments.claim_timeout_minutes, maximum_events=arguments.maximum_events, ) processed = process_customer_notification_outbox( session, catalog, provider, settings.account_portal_url, now=datetime.now(timezone.utc), maximum_events=arguments.maximum_events, maximum_age_days=arguments.maximum_age_days, notification_pepper=settings.notification_pepper, ) if recovered or processed: print( f"Recovered {recovered} stale notification claim(s); " f"processed {processed} notification event(s).", flush=True, ) stop_event.wait(poll_seconds) finally: signal.signal(signal.SIGTERM, previous_sigterm) signal.signal(signal.SIGINT, previous_sigint) engine.dispose() return CLI_SUCCESS def _retry_failed_notifications(arguments: argparse.Namespace) -> int: settings = _database_worker_settings() engine = create_database_engine(settings) try: with Session(engine) as session: requeued = requeue_failed_customer_notifications( session, now=datetime.now(timezone.utc), maximum_events=arguments.maximum_events, maximum_attempts=arguments.maximum_attempts, ) finally: engine.dispose() print(f"Requeued {requeued} failed customer notification(s).") return CLI_SUCCESS def _recover_stale_notifications(arguments: argparse.Namespace) -> int: settings = _database_worker_settings() engine = create_database_engine(settings) try: with Session(engine) as session: recovered = recover_stale_customer_notification_claims( session, now=datetime.now(timezone.utc), claim_timeout_minutes=arguments.claim_timeout_minutes, maximum_events=arguments.maximum_events, ) finally: engine.dispose() print(f"Recovered {recovered} stale customer notification claim(s).") return CLI_SUCCESS def _notification_status(_arguments: argparse.Namespace) -> int: settings = _database_worker_settings() engine = create_database_engine(settings) try: with Session(engine) as session: outbox_rows = session.execute( select(OutboxEvent.status, func.count()) .where( OutboxEvent.event_type == CUSTOMER_NOTIFICATION_OUTBOX_EVENT_TYPE ) .group_by(OutboxEvent.status) ).all() delivery_rows = session.execute( select(NotificationDelivery.status, func.count()).group_by( NotificationDelivery.status ) ).all() feedback_rows = session.execute( select(NotificationFeedbackEvent.feedback_type, func.count()).group_by( NotificationFeedbackEvent.feedback_type ) ).all() suppression_rows = session.execute( select(NotificationSuppression.active, func.count()).group_by( NotificationSuppression.active ) ).all() finally: engine.dispose() print( json.dumps( { "deliveries": dict(delivery_rows), "feedback": dict(feedback_rows), "outbox": dict(outbox_rows), "suppressions": { "active": next( ( count for active, count in suppression_rows if active ), 0, ), "cleared": next( ( count for active, count in suppression_rows if not active ), 0, ), }, }, ensure_ascii=True, indent=2, sort_keys=True, ) ) return CLI_SUCCESS def _record_notification_feedback(arguments: argparse.Namespace) -> int: if not isinstance(arguments, argparse.Namespace): raise TypeError("arguments must be an argparse Namespace") settings = _database_worker_settings() engine = create_database_engine(settings) try: with Session(engine) as session: recorded = record_notification_feedback( session, arguments.provider, arguments.provider_event_id, NotificationFeedbackType(arguments.feedback_type), occurred_at=parse_rfc3339(arguments.occurred_at, "occurred_at"), now=datetime.now(timezone.utc), delivery_id=arguments.delivery_id, provider_message_id=arguments.provider_message_id, ) finally: engine.dispose() print( "Recorded notification feedback." if recorded else "Notification feedback was already recorded." ) return CLI_SUCCESS def _clear_notification_suppression(arguments: argparse.Namespace) -> int: if not isinstance(arguments, argparse.Namespace): raise TypeError("arguments must be an argparse Namespace") settings = _notification_worker_settings() assert settings.notification_pepper is not None engine = create_database_engine(settings) try: with Session(engine) as session: cleared = clear_notification_suppression( session, arguments.user_id, arguments.admin_user_id, arguments.reason, settings.notification_pepper, now=datetime.now(timezone.utc), ) finally: engine.dispose() print( "Cleared the current verified-address suppression." if cleared else "No active suppression exists for the current verified address." ) return CLI_SUCCESS def _purge_notification_feedback(arguments: argparse.Namespace) -> int: if not isinstance(arguments, argparse.Namespace): raise TypeError("arguments must be an argparse Namespace") settings = _database_worker_settings() engine = create_database_engine(settings) try: with Session(engine) as session: deleted = purge_notification_feedback_events( session, now=datetime.now(timezone.utc), retention_days=arguments.retention_days, maximum_events=arguments.maximum_events, ) finally: engine.dispose() print(f"Purged {deleted} expired notification feedback event(s).") return CLI_SUCCESS def _deployment_doctor(arguments: argparse.Namespace) -> int: if not isinstance(arguments, argparse.Namespace): raise TypeError("arguments must be an argparse Namespace") if arguments.component == DEPLOYMENT_COMPONENT_API: settings = _settings() signer = Ed25519SnapshotSigner.from_private_key_file( settings.signing_key_id, settings.signing_private_key_path, require_owner_only=settings.allow_production_file_signer, ) elif arguments.component == DEPLOYMENT_COMPONENT_SUBSCRIPTION_WORKER: settings = _subscription_worker_settings() signer = None else: settings = _notification_worker_settings() signer = None engine = create_database_engine(settings) try: report = inspect_deployment_readiness( engine, settings, load_builtin_catalog(), signer, now=datetime.now(timezone.utc), production_required=not arguments.nonproduction_diagnostic, require_commerce=arguments.require_commerce, maximum_backlog_age_minutes=arguments.maximum_backlog_age_minutes, component=arguments.component, ) finally: engine.dispose() document = report.to_mapping() if arguments.output is None: print(json.dumps(document, ensure_ascii=True, indent=2, sort_keys=True)) else: output_path = Path(arguments.output).expanduser().resolve() _write_owner_only_serial_export(output_path, document) print(f"Deployment readiness evidence was written to {output_path}.") return CLI_SUCCESS if report.ready else CLI_FAILURE def _reconcile_subscriptions(arguments: argparse.Namespace) -> int: settings = _subscription_worker_settings() if settings.stripe_secret_key is None: raise ValueError("LICENSING_STRIPE_SECRET_KEY[_FILE] is required for reconciliation") engine = create_database_engine(settings) try: with Session(engine) as session: catalog = load_builtin_catalog() _require_published_runtime_catalog(session, catalog) result = reconcile_stripe_subscriptions( session, catalog, StripeBillingProvider(settings.stripe_secret_key), settings.stripe_price_sku_map, now=datetime.now(timezone.utc), correlation_id=new_identifier("correlation"), maximum_subscriptions=arguments.maximum_subscriptions, ) finally: engine.dispose() print(json.dumps(result.to_mapping(), ensure_ascii=True, indent=2, sort_keys=True)) return CLI_FAILURE if result.failed else CLI_SUCCESS def _purge_subscription_records(arguments: argparse.Namespace) -> int: settings = _database_worker_settings() engine = create_database_engine(settings) try: with Session(engine) as session: result = purge_subscription_provider_records( session, now=datetime.now(timezone.utc), correlation_id=new_identifier("correlation"), processed_retention_days=arguments.processed_retention_days, failed_retention_days=arguments.failed_retention_days, ) finally: engine.dispose() print(json.dumps(result.to_mapping(), ensure_ascii=True, indent=2, sort_keys=True)) return CLI_SUCCESS def _list_privacy_requests(arguments: argparse.Namespace) -> int: if not isinstance(arguments, argparse.Namespace): raise TypeError("arguments must be an argparse Namespace") settings = _settings() engine = create_database_engine(settings) try: with Session(engine) as session: result = AccountPrivacyService( session, settings.fingerprint_pepper, ).list_pending_for_administration( arguments.admin_user_id, status=arguments.status, limit=arguments.limit, ) finally: engine.dispose() print(json.dumps(result, ensure_ascii=True, indent=2, sort_keys=True)) return CLI_SUCCESS def _erase_privacy_account(arguments: argparse.Namespace) -> int: if not isinstance(arguments, argparse.Namespace): raise TypeError("arguments must be an argparse Namespace") settings = _settings() engine = create_database_engine(settings) try: with Session(engine) as session: service = AccountPrivacyService( session, settings.fingerprint_pepper, ) if not arguments.confirm: result = service.preview_deletion( arguments.request_id, arguments.admin_user_id, arguments.reason_code, new_identifier("correlation"), note=arguments.note, ) else: if arguments.expected_state_digest is None: raise ValueError( "--expected-state-digest is required with --confirm" ) identifiers = _privacy_command_identifiers( "erase", { "privacyRequestId": arguments.request_id, "reasonCode": arguments.reason_code, "note": arguments.note, "expectedStateDigest": arguments.expected_state_digest, }, ) result = service.execute_deletion( arguments.request_id, arguments.admin_user_id, arguments.reason_code, arguments.expected_state_digest, identifiers[0], identifiers[1], note=arguments.note, ) finally: engine.dispose() print(json.dumps(result, ensure_ascii=True, indent=2, sort_keys=True)) return CLI_SUCCESS if result.get("canComplete", True) else CLI_FAILURE def _purge_privacy_request_records(arguments: argparse.Namespace) -> int: if not isinstance(arguments, argparse.Namespace): raise TypeError("arguments must be an argparse Namespace") settings = _database_worker_settings() engine = create_database_engine(settings) try: with Session(engine) as session: result = purge_privacy_requests( session, now=datetime.now(timezone.utc), correlation_id=new_identifier("correlation"), retention_days=arguments.retention_days, maximum_requests=arguments.maximum_requests, ) finally: engine.dispose() print(json.dumps(result, ensure_ascii=True, indent=2, sort_keys=True)) return CLI_SUCCESS def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="python -m licensing_server.cli") commands = parser.add_subparsers(dest="command", required=True) serials = commands.add_parser("serials", help="Manage one-time serial issuance") serial_commands = serials.add_subparsers(dest="serial_command", required=True) list_skus = serial_commands.add_parser("list-skus", help="List serial-eligible SKUs") list_skus.set_defaults(handler=_list_serial_skus) generate = serial_commands.add_parser("generate", help="Generate a guarded serial batch") generate.add_argument("--sku", required=True) generate.add_argument("--quantity", required=True, type=int) generate.add_argument("--admin-user-id", required=True) generate.add_argument("--reason", required=True) generate.add_argument("--campaign") generate.add_argument("--redemption-deadline") generate.add_argument("--surface-profile-id") generate.add_argument( "--trial-duration-hours", type=int, choices=range( TRIAL_MINIMUM_AUTHORITY_DURATION_HOURS, TRIAL_MAXIMUM_TOTAL_DURATION_HOURS + 1, ), metavar=( f"{TRIAL_MINIMUM_AUTHORITY_DURATION_HOURS}.." f"{TRIAL_MAXIMUM_TOTAL_DURATION_HOURS}" ), help="Exact trial duration in hours; required only for a trial SKU.", ) generate.add_argument("--output", required=True) generate.set_defaults(handler=_generate_serials) catalog = commands.add_parser( "catalog", help="Operate the reviewed runtime catalog release registry", ) catalog_commands = catalog.add_subparsers(dest="catalog_command", required=True) catalog_status = catalog_commands.add_parser("status") catalog_status.set_defaults(handler=_catalog_status) catalog_stage = catalog_commands.add_parser( "stage-current", help="Register the exact built-in catalog as a staged release", ) catalog_stage.add_argument("--admin-user-id", required=True) catalog_stage.add_argument( "--reason-code", choices=tuple(sorted(CATALOG_RELEASE_REASONS)), required=True, ) catalog_stage.add_argument("--note") catalog_stage.set_defaults(handler=_stage_current_catalog) catalog_publish = catalog_commands.add_parser( "publish-current", help="Preview or publish the exact built-in staged catalog", ) catalog_publish.add_argument("--admin-user-id", required=True) catalog_publish.add_argument( "--reason-code", choices=tuple(sorted(CATALOG_RELEASE_REASONS)), required=True, ) catalog_publish.add_argument("--note") catalog_publish.add_argument("--expected-state-digest") catalog_publish.add_argument( "--confirm", action="store_true", help="Publish the reviewed catalog; without this flag the command previews only", ) catalog_publish.set_defaults(handler=_publish_current_catalog) signing = commands.add_parser( "signing", help="Operate isolated public signer metadata and rotation state", ) signing_commands = signing.add_subparsers(dest="signing_command", required=True) signing_status = signing_commands.add_parser("status") signing_status.set_defaults(handler=_signing_status) signing_register = signing_commands.add_parser( "register-current", help="Register the configured signer's public metadata as staged", ) signing_register.add_argument("--admin-user-id", required=True) signing_register.add_argument( "--reason-code", choices=tuple(sorted(SIGNING_KEY_CHANGE_REASONS)), required=True, ) signing_register.add_argument("--note") signing_register.add_argument("--not-before", required=True) signing_register.add_argument("--expires-at") signing_register.set_defaults(handler=_register_signing_key) signing_activate = signing_commands.add_parser( "activate-current", help="Preview or activate the exact configured staged signer", ) signing_activate.add_argument("--admin-user-id", required=True) signing_activate.add_argument( "--reason-code", choices=tuple(sorted(SIGNING_KEY_CHANGE_REASONS)), required=True, ) signing_activate.add_argument("--note") signing_activate.add_argument("--expected-state-digest") signing_activate.add_argument( "--confirm", action="store_true", help="Execute the reviewed rotation; without this flag the command previews only", ) signing_activate.set_defaults(handler=_activate_signing_key) signing_compromise = signing_commands.add_parser( "mark-compromised", help="Preview or mark an exact public signing-key record compromised", ) signing_compromise.add_argument("--key-id", required=True) signing_compromise.add_argument("--admin-user-id", required=True) signing_compromise.add_argument( "--reason-code", choices=tuple(sorted(SIGNING_KEY_COMPROMISE_REASONS)), required=True, ) signing_compromise.add_argument("--note") signing_compromise.add_argument("--expected-state-digest") signing_compromise.add_argument( "--confirm", action="store_true", help="Mark the reviewed key compromised; without this flag the command previews only", ) signing_compromise.set_defaults(handler=_compromise_signing_key) export_public = signing_commands.add_parser("export-public-key") export_public.add_argument("--output", required=True) export_public.set_defaults(handler=_export_signing_public_key) outbox = commands.add_parser("outbox", help="Operate provider billing projections") outbox_commands = outbox.add_subparsers(dest="outbox_command", required=True) run_once = outbox_commands.add_parser("run-once") run_once.add_argument( "--maximum-events", type=int, default=DEFAULT_OUTBOX_BATCH_SIZE, ) run_once.set_defaults(handler=_run_outbox_once) worker = outbox_commands.add_parser("worker") worker.add_argument( "--maximum-events", type=int, default=DEFAULT_OUTBOX_BATCH_SIZE, ) worker.add_argument( "--poll-seconds", type=float, default=DEFAULT_OUTBOX_POLL_SECONDS, ) worker.set_defaults(handler=_run_outbox_worker) retry = outbox_commands.add_parser("retry-failed") retry.add_argument( "--maximum-events", type=int, default=DEFAULT_OUTBOX_BATCH_SIZE, ) retry.add_argument("--maximum-attempts", type=int, default=10) retry.set_defaults(handler=_retry_failed_outbox) status = outbox_commands.add_parser("status") status.set_defaults(handler=_outbox_status) notifications = commands.add_parser( "notifications", help="Deliver privacy-bounded transactional licensing email", ) notification_commands = notifications.add_subparsers( dest="notification_command", required=True, ) notification_once = notification_commands.add_parser("run-once") notification_once.add_argument( "--maximum-events", type=int, default=DEFAULT_OUTBOX_BATCH_SIZE, ) notification_once.add_argument( "--maximum-age-days", type=int, default=DEFAULT_NOTIFICATION_MAXIMUM_AGE_DAYS, ) notification_once.add_argument( "--claim-timeout-minutes", type=int, default=DEFAULT_NOTIFICATION_CLAIM_TIMEOUT_MINUTES, ) notification_once.set_defaults(handler=_run_notifications_once) notification_worker = notification_commands.add_parser("worker") notification_worker.add_argument( "--maximum-events", type=int, default=DEFAULT_OUTBOX_BATCH_SIZE, ) notification_worker.add_argument( "--maximum-age-days", type=int, default=DEFAULT_NOTIFICATION_MAXIMUM_AGE_DAYS, ) notification_worker.add_argument( "--claim-timeout-minutes", type=int, default=DEFAULT_NOTIFICATION_CLAIM_TIMEOUT_MINUTES, ) notification_worker.add_argument( "--poll-seconds", type=float, default=DEFAULT_OUTBOX_POLL_SECONDS, ) notification_worker.set_defaults(handler=_run_notifications_worker) notification_retry = notification_commands.add_parser("retry-failed") notification_retry.add_argument( "--maximum-events", type=int, default=DEFAULT_OUTBOX_BATCH_SIZE, ) notification_retry.add_argument("--maximum-attempts", type=int, default=10) notification_retry.set_defaults(handler=_retry_failed_notifications) notification_recover = notification_commands.add_parser("recover-stale") notification_recover.add_argument( "--maximum-events", type=int, default=DEFAULT_OUTBOX_BATCH_SIZE, ) notification_recover.add_argument( "--claim-timeout-minutes", type=int, default=DEFAULT_NOTIFICATION_CLAIM_TIMEOUT_MINUTES, ) notification_recover.set_defaults(handler=_recover_stale_notifications) notification_status = notification_commands.add_parser("status") notification_status.set_defaults(handler=_notification_status) notification_feedback = notification_commands.add_parser( "record-feedback", help="Record already-authenticated delayed provider feedback", ) notification_feedback.add_argument("--provider", required=True) notification_feedback.add_argument("--provider-event-id", required=True) notification_feedback.add_argument( "--feedback-type", choices=tuple(value.value for value in NotificationFeedbackType), required=True, ) notification_feedback.add_argument("--occurred-at", required=True) notification_feedback.add_argument("--delivery-id") notification_feedback.add_argument("--provider-message-id") notification_feedback.set_defaults(handler=_record_notification_feedback) notification_clear = notification_commands.add_parser( "clear-suppression", help="Clear suppression for a user's current verified address", ) notification_clear.add_argument("--user-id", required=True) notification_clear.add_argument("--admin-user-id", required=True) notification_clear.add_argument( "--reason", choices=tuple(sorted(NOTIFICATION_SUPPRESSION_CLEAR_REASONS)), required=True, ) notification_clear.set_defaults(handler=_clear_notification_suppression) notification_purge = notification_commands.add_parser( "purge-feedback", help="Purge expired normalized feedback while retaining suppressions", ) notification_purge.add_argument( "--retention-days", type=int, default=DEFAULT_NOTIFICATION_FEEDBACK_RETENTION_DAYS, ) notification_purge.add_argument( "--maximum-events", type=int, default=DEFAULT_OUTBOX_BATCH_SIZE, ) notification_purge.set_defaults(handler=_purge_notification_feedback) deployment = commands.add_parser( "deployment", help="Run read-only production deployment qualification", ) deployment_commands = deployment.add_subparsers( dest="deployment_command", required=True, ) deployment_doctor = deployment_commands.add_parser("doctor") deployment_doctor.add_argument( "--maximum-backlog-age-minutes", type=int, default=DEFAULT_MAXIMUM_BACKLOG_AGE_MINUTES, ) deployment_doctor.add_argument( "--component", choices=tuple(sorted(DEPLOYMENT_COMPONENTS)), default=DEPLOYMENT_COMPONENT_API, ) deployment_doctor.add_argument("--require-commerce", action="store_true") deployment_doctor.add_argument( "--nonproduction-diagnostic", action="store_true", help="Permit a non-production environment; output is not release evidence.", ) deployment_doctor.add_argument("--output") deployment_doctor.set_defaults(handler=_deployment_doctor) subscriptions = commands.add_parser( "subscriptions", help="Reconcile provider state and apply privacy retention", ) subscription_commands = subscriptions.add_subparsers( dest="subscription_command", required=True, ) reconcile = subscription_commands.add_parser("reconcile") reconcile.add_argument("--maximum-subscriptions", type=int, default=100) reconcile.set_defaults(handler=_reconcile_subscriptions) purge = subscription_commands.add_parser("purge-provider-records") purge.add_argument( "--processed-retention-days", type=int, default=DEFAULT_PROCESSED_PROVIDER_RETENTION_DAYS, ) purge.add_argument( "--failed-retention-days", type=int, default=DEFAULT_FAILED_PROVIDER_RETENTION_DAYS, ) purge.set_defaults(handler=_purge_subscription_records) privacy = commands.add_parser( "privacy", help="Review account erasure and apply privacy retention", ) privacy_commands = privacy.add_subparsers( dest="privacy_command", required=True, ) privacy_list = privacy_commands.add_parser( "list", help="List privacy requests without exposing customer identity claims", ) privacy_list.add_argument("--admin-user-id", required=True) privacy_list.add_argument( "--status", choices=("pending", "cancelled", "completed"), default="pending", ) privacy_list.add_argument("--limit", type=int, default=50) privacy_list.set_defaults(handler=_list_privacy_requests) privacy_erase = privacy_commands.add_parser( "erase-account", help="Preview or execute one exact customer-requested account erasure", ) privacy_erase.add_argument("--request-id", required=True) privacy_erase.add_argument("--admin-user-id", required=True) privacy_erase.add_argument( "--reason-code", choices=tuple(sorted(PRIVACY_DELETION_REASONS)), required=True, ) privacy_erase.add_argument("--note") privacy_erase.add_argument("--expected-state-digest") privacy_erase.add_argument( "--confirm", action="store_true", help="Execute the reviewed erasure; without this flag the command previews only", ) privacy_erase.set_defaults(handler=_erase_privacy_account) privacy_purge = privacy_commands.add_parser( "purge-request-records", help="Purge resolved privacy requests while retaining security audit", ) privacy_purge.add_argument( "--retention-days", type=int, default=DEFAULT_PRIVACY_REQUEST_RETENTION_DAYS, ) privacy_purge.add_argument( "--maximum-requests", type=int, default=MAXIMUM_PRIVACY_PURGE_ROWS, ) privacy_purge.set_defaults(handler=_purge_privacy_request_records) return parser def main(argv: Optional[Sequence[str]] = None) -> int: if argv is not None and ( isinstance(argv, (str, bytes)) or not isinstance(argv, Sequence) ): raise TypeError("argv must be a sequence of strings or None") parser = _parser() arguments = parser.parse_args(argv) handler = getattr(arguments, "handler", None) if not callable(handler): parser.error("a command is required") return CLI_USAGE_ERROR try: return int(handler(arguments)) except KeyboardInterrupt: return CLI_FAILURE except Exception as exc: print(f"Licensing command failed: {exc}", file=sys.stderr) return CLI_FAILURE if __name__ == "__main__": raise SystemExit(main())