"""Production operator console for the guarded licensing administration API.""" from __future__ import annotations import argparse import hashlib import json import os from pathlib import Path import stat import sys from typing import Callable, Mapping, MutableMapping, Optional, Sequence from urllib.error import HTTPError, URLError from urllib.parse import urlencode, urlsplit, urlunsplit from urllib.request import Request, urlopen as standard_urlopen from licensing_shared.canonical_json import canonicalize_json, parse_bounded_json from licensing_shared.catalog import load_catalog_document from licensing_shared.constants import ( MAX_CATALOG_DOCUMENT_BYTES, TRIAL_MAXIMUM_TOTAL_DURATION_HOURS, TRIAL_MINIMUM_AUTHORITY_DURATION_HOURS, validate_identifier, ) from licensing_shared.product_surfaces import ( PRODUCT_SURFACE_KIND_ANALYSIS, PRODUCT_SURFACE_KIND_MAIN_TAB, validate_product_surface_ids, ) from licensing_server.app.constants import ( CATALOG_RELEASE_REASONS, PRIVACY_DELETION_REASONS, SERIAL_BATCH_REVOCATION_REASONS, ) CLI_SUCCESS = 0 CLI_FAILURE = 1 CLI_USAGE_ERROR = 2 ADMIN_API_URL_ENVIRONMENT = "LICENSING_ADMIN_API_BASE_URL" ADMIN_TOKEN_FILE_ENVIRONMENT = "LICENSING_ADMIN_ACCESS_TOKEN_FILE" MAXIMUM_ADMIN_TOKEN_BYTES = 16 * 1024 MAXIMUM_ADMIN_RESPONSE_BYTES = 16 * 1024 * 1024 ADMIN_OUTPUT_LINE_TERMINATOR = b"\r\n" def _admin_base_url(value: str) -> str: if not isinstance(value, str): raise TypeError("admin API base URL must be a string") parsed = urlsplit(value.strip()) if ( parsed.scheme != "https" or not parsed.netloc or parsed.username is not None or parsed.password is not None or parsed.path not in ("", "/") or parsed.query or parsed.fragment ): raise ValueError("admin API base URL must be an HTTPS origin without credentials") return urlunsplit((parsed.scheme, parsed.netloc, "", "", "")) def _access_token(path_value: str) -> str: if not isinstance(path_value, str) or not path_value.strip(): raise ValueError("admin access-token file path is required") source_path = Path(path_value.strip()).expanduser() if source_path.is_symlink(): raise ValueError("admin access-token path must be a regular non-symlink file") path = source_path.resolve() if not path.is_file(): raise ValueError("admin access-token path must be a regular non-symlink file") metadata = path.stat() if metadata.st_size < 1 or metadata.st_size > MAXIMUM_ADMIN_TOKEN_BYTES: raise ValueError("admin access-token file is empty or too large") if os.name != "nt" and stat.S_IMODE(metadata.st_mode) & 0o077: raise PermissionError("admin access-token file must be owner-only") token = path.read_text(encoding="utf-8").strip() if not token or any(character.isspace() for character in token): raise ValueError("admin access token must be a non-empty single token") return token class AdminApiClient: def __init__( self, base_url: str, access_token: str, *, urlopen: Callable[..., object] = standard_urlopen, ) -> None: self._base_url = _admin_base_url(base_url) if not isinstance(access_token, str) or not access_token.strip(): raise ValueError("access_token must be a non-empty string") if any(character.isspace() for character in access_token.strip()): raise ValueError("access_token must be a single token") if not callable(urlopen): raise TypeError("urlopen must be callable") self._access_token = access_token.strip() self._urlopen = urlopen def request( self, method: str, path: str, body: Optional[Mapping[str, object]] = None, ) -> dict[str, object]: if not isinstance(method, str) or method not in ("GET", "POST"): raise ValueError("method must be GET or POST") if not isinstance(path, str) or not path.startswith("/v1/"): raise ValueError("path must be a versioned licensing API path") if body is not None and not isinstance(body, Mapping): raise TypeError("body must be a mapping or None") payload = None if body is None else canonicalize_json(body) request = Request( self._base_url + path, data=payload, method=method, headers={ "Accept": "application/json", "Authorization": f"Bearer {self._access_token}", "Content-Type": "application/json", "Cache-Control": "no-store", }, ) try: with self._urlopen(request, timeout=30.0) as response: raw = response.read(MAXIMUM_ADMIN_RESPONSE_BYTES + 1) status_code = int(getattr(response, "status", 200)) except HTTPError as exc: raw = exc.read(MAXIMUM_ADMIN_RESPONSE_BYTES + 1) status_code = int(exc.code) except (TimeoutError, URLError) as exc: raise RuntimeError(f"licensing administration API is unavailable: {exc}") from exc if len(raw) > MAXIMUM_ADMIN_RESPONSE_BYTES: raise RuntimeError("licensing administration response is too large") parsed = parse_bounded_json(raw, maximum_bytes=MAXIMUM_ADMIN_RESPONSE_BYTES) if not isinstance(parsed, Mapping): raise RuntimeError("licensing administration response is not an object") result = dict(parsed) if status_code < 200 or status_code >= 300: code = str(result.get("code") or "request_failed") message = str(result.get("message") or "administration request failed") raise RuntimeError(f"{code}: {message}") return result def _client( environ: Mapping[str, str], urlopen: Callable[..., object], ) -> AdminApiClient: if not isinstance(environ, Mapping): raise TypeError("environ must be a mapping") base_url = environ.get(ADMIN_API_URL_ENVIRONMENT) token_file = environ.get(ADMIN_TOKEN_FILE_ENVIRONMENT) if base_url is None: raise ValueError(f"{ADMIN_API_URL_ENVIRONMENT} is required") if token_file is None: raise ValueError(f"{ADMIN_TOKEN_FILE_ENVIRONMENT} is required") return AdminApiClient( base_url, _access_token(token_file), urlopen=urlopen, ) def admin_client_from_environment( environ: Mapping[str, str], urlopen: Callable[..., object] = standard_urlopen, ) -> AdminApiClient: """Create the guarded admin client without exposing a token on the command line.""" return _client(environ, urlopen) def _decision_defaults( decision: str, request_id: str, generation: int, reason_code: str, note: Optional[str], ) -> tuple[str, str]: digest = hashlib.sha256( canonicalize_json( { "decision": validate_identifier(decision, "decision"), "requestId": validate_identifier(request_id, "request_id"), "generation": generation, "reasonCode": validate_identifier(reason_code, "reason_code"), "note": note, } ) ).hexdigest()[:32] return f"admin-offline-{decision}-{digest}", f"correlation.offline_{decision}.{digest}" def _serial_batch_revocation_defaults( batch_id: str, state_digest: str, reason_code: str, note: Optional[str], ) -> tuple[str, str]: correlation_id = _serial_batch_revocation_correlation( batch_id, reason_code, note, ) normalized_batch = validate_identifier(batch_id, "batch_id") normalized_reason = validate_identifier(reason_code, "reason_code") if ( not isinstance(state_digest, str) or len(state_digest) != 64 or state_digest != state_digest.lower() or any(character not in "0123456789abcdef" for character in state_digest) ): raise ValueError("state digest must be 64 lowercase hexadecimal characters") idempotency_digest = hashlib.sha256( canonicalize_json( { "batchId": normalized_batch, "stateDigest": state_digest, "reasonCode": normalized_reason, "note": note, } ) ).hexdigest()[:32] return f"admin-serial-batch-revoke-{idempotency_digest}", correlation_id def _serial_batch_revocation_correlation( batch_id: str, reason_code: str, note: Optional[str], ) -> str: normalized_batch = validate_identifier(batch_id, "batch_id") normalized_reason = validate_identifier(reason_code, "reason_code") if normalized_reason not in SERIAL_BATCH_REVOCATION_REASONS: raise ValueError("reason_code is not valid for serial batch revocation") if note is not None and (not isinstance(note, str) or not note.strip()): raise ValueError("note must be a non-empty string or None") correlation_digest = hashlib.sha256( canonicalize_json( { "batchId": normalized_batch, "reasonCode": normalized_reason, "note": note, } ) ).hexdigest()[:32] return f"correlation.serial_batch_revoke.{correlation_digest}" def _privacy_deletion_defaults( request_id: str, state_digest: str, reason_code: str, note: Optional[str], ) -> tuple[str, str]: normalized_request = validate_identifier(request_id, "request_id") normalized_reason = validate_identifier(reason_code, "reason_code") if normalized_reason not in PRIVACY_DELETION_REASONS: raise ValueError("reason_code is not valid for account deletion") if note is not None and (not isinstance(note, str) or not note.strip()): raise ValueError("note must be a non-empty string or None") if ( not isinstance(state_digest, str) or len(state_digest) != 64 or state_digest != state_digest.lower() or any(character not in "0123456789abcdef" for character in state_digest) ): raise ValueError("state digest must be 64 lowercase hexadecimal characters") digest = hashlib.sha256( canonicalize_json( { "privacyRequestId": normalized_request, "stateDigest": state_digest, "reasonCode": normalized_reason, "note": note, } ) ).hexdigest()[:32] return ( f"admin-privacy-erase-{digest}", f"correlation.privacy_erase.{digest}", ) def _catalog_operation_defaults( action: str, revision: int, catalog_sha256: str, reason_code: str, note: Optional[str], *, state_digest: Optional[str] = None, ) -> tuple[str, str]: normalized_action = validate_identifier(action, "action") if isinstance(revision, bool) or not isinstance(revision, int) or revision < 1: raise ValueError("revision must be a positive integer") for value, field_name in ( (catalog_sha256, "catalog_sha256"), (state_digest, "state_digest"), ): if value is None and field_name == "state_digest": continue if ( not isinstance(value, str) or len(value) != 64 or value != value.lower() or any(character not in "0123456789abcdef" for character in value) ): raise ValueError( f"{field_name} must be 64 lowercase hexadecimal characters" ) normalized_reason = validate_identifier(reason_code, "reason_code") if normalized_reason not in CATALOG_RELEASE_REASONS: raise ValueError("reason_code is not valid for catalog administration") if note is not None and (not isinstance(note, str) or not note.strip()): raise ValueError("note must be a non-empty string or None") payload = { "action": normalized_action, "catalogSha256": catalog_sha256, "note": note, "reasonCode": normalized_reason, "revision": revision, "stateDigest": state_digest, } digest = hashlib.sha256(canonicalize_json(payload)).hexdigest()[:32] return ( f"admin-catalog-{normalized_action}-{digest}", f"correlation.catalog_{normalized_action}.{digest}", ) def _read_catalog_candidate(path_value: str) -> dict[str, object]: if not isinstance(path_value, str) or not path_value.strip(): raise ValueError("catalog candidate path is required") source_path = Path(path_value).expanduser() if source_path.is_symlink(): raise ValueError("catalog candidate must be a regular non-symlink file") path = source_path.resolve() if not path.is_file(): raise ValueError("catalog candidate must be a regular non-symlink file") metadata = path.stat() if metadata.st_size < 1 or metadata.st_size > MAX_CATALOG_DOCUMENT_BYTES: raise ValueError("catalog candidate is empty or too large") catalog = load_catalog_document(path.read_bytes()) return catalog.to_mapping() def _write_owner_only_json(path_value: str, document: Mapping[str, object]) -> None: if not isinstance(path_value, str) or not path_value.strip(): raise ValueError("output path is required") if not isinstance(document, Mapping): raise TypeError("document must be a mapping") source_path = Path(path_value).expanduser() if source_path.is_symlink(): raise FileExistsError("output already exists; choose a new file") path = source_path.resolve() if path.exists(): raise FileExistsError("output already exists; choose a new file") if not path.parent.is_dir() or path.parent.is_symlink(): raise ValueError("output parent must be an existing non-symlink directory") body = json.dumps(dict(document), indent=2, sort_keys=True).encode("utf-8") 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 + ADMIN_OUTPUT_LINE_TERMINATOR) output.flush() os.fsync(output.fileno()) except Exception: path.unlink(missing_ok=True) raise if os.name != "nt" and path.stat().st_mode & 0o077: path.unlink(missing_ok=True) raise PermissionError("admin export permissions are not owner-only") def _offline_list(arguments: argparse.Namespace, client: AdminApiClient) -> dict[str, object]: query: MutableMapping[str, object] = {"limit": arguments.limit} if arguments.status is not None: query["status"] = arguments.status return client.request("GET", "/v1/admin/offline-requests?" + urlencode(query)) def _offline_inspect( arguments: argparse.Namespace, client: AdminApiClient, ) -> dict[str, object]: request_id = validate_identifier(arguments.request_id, "request_id") return client.request("GET", f"/v1/admin/offline-requests/{request_id}") def _offline_decide( arguments: argparse.Namespace, client: AdminApiClient, ) -> dict[str, object]: preview = _offline_inspect(arguments, client) if preview.get("status") != "pending": raise RuntimeError("only a pending, unexpired offline request can be decided") generation = preview.get("decisionGeneration") if isinstance(generation, bool) or not isinstance(generation, int) or generation < 0: raise RuntimeError("offline request preview has an invalid decision generation") if not arguments.confirm: return { "confirmationRequired": True, "plannedDecision": arguments.offline_command, "reasonCode": arguments.reason_code, "request": preview, } normalized_request = validate_identifier(arguments.request_id, "request_id") idempotency_key, correlation_id = _decision_defaults( arguments.offline_command, normalized_request, generation, arguments.reason_code, arguments.note, ) return client.request( "POST", f"/v1/admin/offline-requests/{normalized_request}/{arguments.offline_command}", { "reasonCode": arguments.reason_code, "note": arguments.note, "expectedDecisionGeneration": generation, "idempotencyKey": idempotency_key, "correlationId": correlation_id, }, ) def _license_inspect( arguments: argparse.Namespace, client: AdminApiClient, ) -> dict[str, object]: license_id = validate_identifier(arguments.license_id, "license_id") return client.request("GET", f"/v1/admin/licenses/{license_id}") def _surface_inventory( _arguments: argparse.Namespace, client: AdminApiClient, ) -> dict[str, object]: return client.request("GET", "/v1/admin/product-surfaces/inventory") def _surface_profiles( arguments: argparse.Namespace, client: AdminApiClient, ) -> dict[str, object]: query = urlencode({"includeArchived": bool(arguments.include_archived)}) return client.request( "GET", "/v1/admin/product-surface-profiles?" + query, ) def _surface_profile_create( arguments: argparse.Namespace, client: AdminApiClient, ) -> dict[str, object]: analysis_ids = validate_product_surface_ids( PRODUCT_SURFACE_KIND_ANALYSIS, tuple(sorted(set(arguments.analysis_id))), ) main_tab_ids = validate_product_surface_ids( PRODUCT_SURFACE_KIND_MAIN_TAB, tuple(sorted(set(arguments.main_tab_id))), require_nonempty=True, ) correlation_id = "correlation.surface_profile_create." + hashlib.sha256( canonicalize_json( { "name": arguments.name, "description": arguments.description, "analysisIds": list(analysis_ids), "mainTabIds": list(main_tab_ids), "reason": arguments.reason, } ) ).hexdigest()[:32] return client.request( "POST", "/v1/admin/product-surface-profiles", { "name": arguments.name, "description": arguments.description, "analysisIds": list(analysis_ids), "mainTabIds": list(main_tab_ids), "reason": arguments.reason, "correlationId": correlation_id, }, ) def _surface_profile_archive( arguments: argparse.Namespace, client: AdminApiClient, ) -> dict[str, object]: profile_id = validate_identifier(arguments.profile_id, "profile_id") correlation_id = "correlation.surface_profile_archive." + hashlib.sha256( canonicalize_json( { "profileId": profile_id, "reason": arguments.reason, } ) ).hexdigest()[:32] return client.request( "POST", f"/v1/admin/product-surface-profiles/{profile_id}/archive", { "reason": arguments.reason, "correlationId": correlation_id, }, ) def _license_surface_profile_assign( arguments: argparse.Namespace, client: AdminApiClient, ) -> dict[str, object]: license_id = validate_identifier(arguments.license_id, "license_id") profile_id = ( None if arguments.profile_id is None else validate_identifier(arguments.profile_id, "profile_id") ) correlation_id = "correlation.license_surface_assign." + hashlib.sha256( canonicalize_json( { "licenseId": license_id, "surfaceProfileId": profile_id, "reason": arguments.reason, } ) ).hexdigest()[:32] return client.request( "POST", f"/v1/admin/licenses/{license_id}/surface-profile", { "surfaceProfileId": profile_id, "reason": arguments.reason, "correlationId": correlation_id, }, ) def _serial_batch_generate( arguments: argparse.Namespace, client: AdminApiClient, ) -> dict[str, object]: request_body: dict[str, object] = { "skuId": validate_identifier(arguments.sku_id, "sku_id"), "quantity": arguments.quantity, "reason": arguments.reason, "correlationId": "correlation.serial_batch_generate." + hashlib.sha256( canonicalize_json( { "skuId": arguments.sku_id, "quantity": arguments.quantity, "reason": arguments.reason, "campaign": arguments.campaign, "redemptionDeadline": arguments.redemption_deadline, "surfaceProfileId": arguments.surface_profile_id, "trialDurationHours": arguments.trial_duration_hours, } ) ).hexdigest()[:32], } for argument_name, field_name in ( ("campaign", "campaign"), ("redemption_deadline", "redemptionDeadline"), ("surface_profile_id", "surfaceProfileId"), ("trial_duration_hours", "trialDurationHours"), ): value = getattr(arguments, argument_name) if value is not None: request_body[field_name] = value response = client.request("POST", "/v1/admin/serial-batches", request_body) serials = response.get("serials") if not isinstance(serials, list) or len(serials) != arguments.quantity: raise RuntimeError("serial generation response has an invalid serial cohort") _write_owner_only_json(arguments.output, response) return { "batchId": response.get("batchId"), "skuId": response.get("skuId"), "quantity": len(serials), "surfaceProfileId": response.get("surfaceProfileId"), "trialDurationHours": response.get("trialDurationHours"), "output": str(Path(arguments.output).expanduser().resolve()), "warning": "Plaintext serials were written only to the owner-only output file.", } def _catalog_status( _arguments: argparse.Namespace, client: AdminApiClient, ) -> dict[str, object]: return client.request("GET", "/v1/admin/catalog/releases") def _catalog_stage( arguments: argparse.Namespace, client: AdminApiClient, ) -> dict[str, object]: document = _read_catalog_candidate(arguments.input) revision = document.get("revision") if isinstance(revision, bool) or not isinstance(revision, int): raise RuntimeError("validated catalog candidate has no revision") catalog_sha256 = hashlib.sha256(canonicalize_json(document)).hexdigest() idempotency_key, correlation_id = _catalog_operation_defaults( "stage", revision, catalog_sha256, arguments.reason_code, arguments.note, ) return client.request( "POST", "/v1/admin/catalog/releases/stage", { "document": document, "reasonCode": arguments.reason_code, "note": arguments.note, "idempotencyKey": idempotency_key, "correlationId": correlation_id, }, ) def _catalog_publish( arguments: argparse.Namespace, client: AdminApiClient, ) -> dict[str, object]: if isinstance(arguments.revision, bool) or arguments.revision < 1: raise ValueError("revision must be a positive integer") status = client.request("GET", "/v1/admin/catalog/releases") releases = status.get("releases") if not isinstance(releases, list): raise RuntimeError("catalog registry status has no release list") candidates = [ value for value in releases if isinstance(value, Mapping) and value.get("revision") == arguments.revision ] if len(candidates) != 1 or not isinstance( candidates[0].get("catalogSha256"), str, ): raise RuntimeError("catalog candidate revision was not found") catalog_sha256 = str(candidates[0]["catalogSha256"]) _unused_idempotency, correlation_id = _catalog_operation_defaults( "publish", arguments.revision, catalog_sha256, arguments.reason_code, arguments.note, ) preview = client.request( "POST", f"/v1/admin/catalog/releases/{arguments.revision}/publication-preview", { "reasonCode": arguments.reason_code, "note": arguments.note, "correlationId": correlation_id, }, ) 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("catalog candidate is not publishable") if not arguments.confirm: return { "confirmationRequired": True, "requiredStateDigest": state_digest, "publication": preview, } 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, correlation_id = _catalog_operation_defaults( "publish", arguments.revision, catalog_sha256, arguments.reason_code, arguments.note, state_digest=state_digest, ) return client.request( "POST", f"/v1/admin/catalog/releases/{arguments.revision}/publish", { "reasonCode": arguments.reason_code, "note": arguments.note, "expectedStateDigest": state_digest, "idempotencyKey": idempotency_key, "correlationId": correlation_id, }, ) def _serial_batch_revoke( arguments: argparse.Namespace, client: AdminApiClient, ) -> dict[str, object]: batch_id = validate_identifier(arguments.batch_id, "batch_id") correlation_id = _serial_batch_revocation_correlation( batch_id, arguments.reason_code, arguments.note, ) preview = client.request( "POST", f"/v1/admin/serial-batches/{batch_id}/revocation-preview", { "reasonCode": arguments.reason_code, "note": arguments.note, "correlationId": correlation_id, }, ) state_digest = preview.get("stateDigest") if not isinstance(state_digest, str): raise RuntimeError("serial batch preview has no state digest") _serial_batch_revocation_defaults( batch_id, state_digest, arguments.reason_code, arguments.note, ) if preview.get("canExecute") is not True: raise RuntimeError("serial batch preview has no active serials to revoke") if not arguments.confirm: return { "confirmationRequired": True, "requiredStateDigest": state_digest, "batch": preview, } 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("serial batch changed after review; preview and confirm again") idempotency_key, correlation_id = _serial_batch_revocation_defaults( batch_id, state_digest, arguments.reason_code, arguments.note, ) return client.request( "POST", f"/v1/admin/serial-batches/{batch_id}/revoke", { "reasonCode": arguments.reason_code, "note": arguments.note, "expectedStateDigest": state_digest, "idempotencyKey": idempotency_key, "correlationId": correlation_id, }, ) def _privacy_list( arguments: argparse.Namespace, client: AdminApiClient, ) -> dict[str, object]: query = urlencode( { "status": arguments.status, "limit": arguments.limit, } ) return client.request( "GET", f"/v1/admin/privacy/deletion-requests?{query}", ) def _privacy_erase( arguments: argparse.Namespace, client: AdminApiClient, ) -> dict[str, object]: request_id = validate_identifier(arguments.request_id, "request_id") preview_correlation = "correlation.privacy_preview." + hashlib.sha256( canonicalize_json( { "privacyRequestId": request_id, "reasonCode": arguments.reason_code, "note": arguments.note, } ) ).hexdigest()[:32] preview = client.request( "POST", f"/v1/admin/privacy/deletion-requests/{request_id}/preview", { "reasonCode": arguments.reason_code, "note": arguments.note, "correlationId": preview_correlation, }, ) state_digest = preview.get("stateDigest") if not isinstance(state_digest, str): raise RuntimeError("account deletion preview has no state digest") if not arguments.confirm: return { "confirmationRequired": preview.get("canComplete") is True, "requiredStateDigest": state_digest, "privacyRequest": preview, } if preview.get("canComplete") is not True: raise RuntimeError("account deletion still has unresolved blockers") 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("account deletion changed after review; preview and confirm again") idempotency_key, correlation_id = _privacy_deletion_defaults( request_id, state_digest, arguments.reason_code, arguments.note, ) return client.request( "POST", f"/v1/admin/privacy/deletion-requests/{request_id}/execute", { "reasonCode": arguments.reason_code, "note": arguments.note, "expectedStateDigest": state_digest, "idempotencyKey": idempotency_key, "correlationId": correlation_id, }, ) def _audit_body(arguments: argparse.Namespace) -> dict[str, object]: result: dict[str, object] = {"limit": arguments.limit} for argument_name, field_name in ( ("action", "action"), ("target_type", "targetType"), ("target_id", "targetId"), ("actor_id", "actorId"), ("occurred_from", "occurredFrom"), ("occurred_to", "occurredTo"), ): value = getattr(arguments, argument_name) if value is not None: result[field_name] = value return result def _audit_search( arguments: argparse.Namespace, client: AdminApiClient, ) -> dict[str, object]: body = _audit_body(arguments) if arguments.cursor is not None: body["cursor"] = arguments.cursor return client.request("POST", "/v1/admin/audit-events/search", body) def _audit_export( arguments: argparse.Namespace, client: AdminApiClient, ) -> dict[str, object]: body = _audit_body(arguments) body["correlationId"] = "correlation.audit_export." + hashlib.sha256( canonicalize_json(body) ).hexdigest()[:32] document = client.request("POST", "/v1/admin/audit-events/export", body) _write_owner_only_json(arguments.output, document) return { "status": "exported", "output": str(Path(arguments.output).expanduser().resolve()), "quantity": len(document.get("events", [])), "sha256": document.get("sha256"), } def _add_audit_filters(parser: argparse.ArgumentParser, *, export: bool) -> None: parser.add_argument("--action") parser.add_argument("--target-type") parser.add_argument("--target-id") parser.add_argument("--actor-id") parser.add_argument("--occurred-from") parser.add_argument("--occurred-to") parser.add_argument("--limit", type=int, default=5000 if export else 100) def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="python -m licensing_server.admin_cli") commands = parser.add_subparsers(dest="command", required=True) offline = commands.add_parser("offline", help="Inspect and decide offline trials") offline_commands = offline.add_subparsers(dest="offline_command", required=True) offline_list = offline_commands.add_parser("list") offline_list.add_argument( "--status", choices=("pending", "approved", "rejected", "expired"), ) offline_list.add_argument("--limit", type=int, default=50) offline_list.set_defaults(handler=_offline_list) offline_inspect = offline_commands.add_parser("inspect") offline_inspect.add_argument("request_id") offline_inspect.set_defaults(handler=_offline_inspect) for decision, choices in ( ("approve", ("evaluation_request", "accessibility_accommodation", "support_recovery")), ("reject", ("trial_policy_ineligible", "duplicate_request", "unsupported_environment")), ): decision_parser = offline_commands.add_parser(decision) decision_parser.add_argument("request_id") decision_parser.add_argument("--reason-code", required=True, choices=choices) decision_parser.add_argument("--note") decision_parser.add_argument( "--confirm", action="store_true", help="Execute the displayed decision; without this flag the command previews only", ) decision_parser.set_defaults(handler=_offline_decide) license_command = commands.add_parser("license", help="Inspect a license") license_commands = license_command.add_subparsers(dest="license_command", required=True) license_inspect = license_commands.add_parser("inspect") license_inspect.add_argument("license_id") license_inspect.set_defaults(handler=_license_inspect) license_surface = license_commands.add_parser( "assign-surface-profile", help="Assign an immutable surface profile, or omit it to restore the full surface", ) license_surface.add_argument("license_id") license_surface.add_argument("--profile-id") license_surface.add_argument("--reason", required=True) license_surface.set_defaults(handler=_license_surface_profile_assign) surfaces = commands.add_parser( "surfaces", help="Inspect the shipped surface inventory and manage reusable profiles", ) surface_commands = surfaces.add_subparsers( dest="surface_command", required=True, ) surface_inventory = surface_commands.add_parser("inventory") surface_inventory.set_defaults(handler=_surface_inventory) surface_list = surface_commands.add_parser("list-profiles") surface_list.add_argument("--include-archived", action="store_true") surface_list.set_defaults(handler=_surface_profiles) surface_create = surface_commands.add_parser("create-profile") surface_create.add_argument("--name", required=True) surface_create.add_argument("--description", required=True) surface_create.add_argument("--analysis-id", action="append", default=[]) surface_create.add_argument("--main-tab-id", action="append", default=[]) surface_create.add_argument("--reason", required=True) surface_create.set_defaults(handler=_surface_profile_create) surface_archive = surface_commands.add_parser("archive-profile") surface_archive.add_argument("profile_id") surface_archive.add_argument("--reason", required=True) surface_archive.set_defaults(handler=_surface_profile_archive) catalog = commands.add_parser( "catalog", help="Validate, stage and publish catalog releases", ) 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") catalog_stage.add_argument("--input", 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=_catalog_stage) catalog_publish = catalog_commands.add_parser("publish") catalog_publish.add_argument("revision", type=int) 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 candidate; without this flag the command previews only", ) catalog_publish.set_defaults(handler=_catalog_publish) serial_batch = commands.add_parser( "serial-batch", help="Preview and revoke serial issuance batches", ) serial_batch_commands = serial_batch.add_subparsers( dest="serial_batch_command", required=True, ) serial_batch_generate = serial_batch_commands.add_parser( "generate", help="Generate one guarded edition, add-on, or bounded trial serial batch", ) serial_batch_generate.add_argument("--sku-id", required=True) serial_batch_generate.add_argument("--quantity", type=int, required=True) serial_batch_generate.add_argument("--reason", required=True) serial_batch_generate.add_argument("--campaign") serial_batch_generate.add_argument("--redemption-deadline") serial_batch_generate.add_argument("--surface-profile-id") serial_batch_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.", ) serial_batch_generate.add_argument("--output", required=True) serial_batch_generate.set_defaults(handler=_serial_batch_generate) serial_batch_revoke = serial_batch_commands.add_parser("revoke") serial_batch_revoke.add_argument("batch_id") serial_batch_revoke.add_argument( "--reason-code", required=True, choices=tuple(sorted(SERIAL_BATCH_REVOCATION_REASONS)), ) serial_batch_revoke.add_argument("--note") serial_batch_revoke.add_argument("--expected-state-digest") serial_batch_revoke.add_argument( "--confirm", action="store_true", help="Execute the reviewed revocation; without this flag the command previews only", ) serial_batch_revoke.set_defaults(handler=_serial_batch_revoke) privacy = commands.add_parser( "privacy", help="Inspect and complete customer-requested account erasure", ) privacy_commands = privacy.add_subparsers( dest="privacy_command", required=True, ) privacy_list = privacy_commands.add_parser("list") 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=_privacy_list) privacy_erase = privacy_commands.add_parser("erase-account") privacy_erase.add_argument("request_id") 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=_privacy_erase) audit = commands.add_parser("audit", help="Search or export privacy-bounded audit data") audit_commands = audit.add_subparsers(dest="audit_command", required=True) audit_search = audit_commands.add_parser("search") _add_audit_filters(audit_search, export=False) audit_search.add_argument("--cursor") audit_search.set_defaults(handler=_audit_search) audit_export = audit_commands.add_parser("export") _add_audit_filters(audit_export, export=True) audit_export.add_argument("--output", required=True) audit_export.set_defaults(handler=_audit_export) return parser def main( argv: Optional[Sequence[str]] = None, environ: Optional[Mapping[str, str]] = None, urlopen: Optional[Callable[..., object]] = 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") resolved_environment = os.environ if environ is None else environ resolved_urlopen = standard_urlopen if urlopen is None else urlopen 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: client = _client(resolved_environment, resolved_urlopen) result = handler(arguments, client) print(json.dumps(result, ensure_ascii=True, indent=2, sort_keys=True)) return CLI_SUCCESS except KeyboardInterrupt: return CLI_FAILURE except Exception as exc: print(f"Licensing administration failed: {exc}", file=sys.stderr) return CLI_FAILURE if __name__ == "__main__": raise SystemExit(main())