"""Qt administration console for product surfaces and guarded serial issuance.""" from __future__ import annotations import argparse import hashlib import os from pathlib import Path from typing import Callable, Mapping, Optional, Sequence from PySide6 import QtCore, QtWidgets 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_server.app.constants import MAX_SERIAL_BATCH_QUANTITY from .admin_cli import ( AdminApiClient, _write_owner_only_json, admin_client_from_environment, ) WINDOW_TITLE = "Licensing Product Administration" FULL_PRODUCT_LABEL = "Full product (no focused profile)" SERIAL_EXPORT_FILTER = "Licensing serial export (*.json);;All files (*)" SUPPORTED_SERIAL_SKU_KINDS = frozenset(("edition", "addon", "trial")) class _OperationSignals(QtCore.QObject): succeeded = QtCore.Signal(object) failed = QtCore.Signal(str) class _AdminOperation(QtCore.QRunnable): def __init__(self, operation: Callable[[], object]) -> None: super().__init__() if not callable(operation): raise TypeError("operation must be callable") self._operation = operation self.signals = _OperationSignals() @QtCore.Slot() def run(self) -> None: try: result = self._operation() except Exception as exc: self.signals.failed.emit(str(exc)) return self.signals.succeeded.emit(result) class SurfaceChecklist(QtWidgets.QWidget): """Searchable check-list that retains checks while rows are filtered.""" def __init__(self, title: str, parent: Optional[QtWidgets.QWidget] = None) -> None: if not isinstance(title, str) or not title.strip(): raise ValueError("title must be non-empty text") super().__init__(parent) self._title = title.strip() layout = QtWidgets.QVBoxLayout(self) heading = QtWidgets.QLabel(self._title) heading.setProperty("role", "heading") layout.addWidget(heading) self.search = QtWidgets.QLineEdit() self.search.setPlaceholderText(f"Filter {self._title.lower()}…") self.search.setClearButtonEnabled(True) layout.addWidget(self.search) self.items = QtWidgets.QListWidget() self.items.setAlternatingRowColors(True) layout.addWidget(self.items, 1) buttons = QtWidgets.QHBoxLayout() self.select_all_button = QtWidgets.QPushButton("Select all") self.select_none_button = QtWidgets.QPushButton("Select none") buttons.addWidget(self.select_all_button) buttons.addWidget(self.select_none_button) buttons.addStretch(1) layout.addLayout(buttons) self.search.textChanged.connect(self._apply_filter) self.select_all_button.clicked.connect(lambda: self._set_all(True)) self.select_none_button.clicked.connect(lambda: self._set_all(False)) def populate(self, surfaces: Sequence[Mapping[str, object]]) -> None: if isinstance(surfaces, (str, bytes)) or not isinstance(surfaces, Sequence): raise TypeError("surfaces must be a sequence of mappings") self.items.clear() for surface in surfaces: if not isinstance(surface, Mapping): raise TypeError("surfaces must contain mappings") surface_id = validate_identifier(surface.get("id"), "surface_id") label = surface.get("label") if not isinstance(label, str) or not label.strip(): raise ValueError("surface label must be non-empty text") item = QtWidgets.QListWidgetItem(f"{label.strip()} · {surface_id}") item.setData(QtCore.Qt.ItemDataRole.UserRole, surface_id) item.setFlags( item.flags() | QtCore.Qt.ItemFlag.ItemIsUserCheckable | QtCore.Qt.ItemFlag.ItemIsEnabled ) item.setCheckState(QtCore.Qt.CheckState.Checked) self.items.addItem(item) self._apply_filter(self.search.text()) def checked_ids(self) -> tuple[str, ...]: values = [] for index in range(self.items.count()): item = self.items.item(index) if item.checkState() == QtCore.Qt.CheckState.Checked: values.append(str(item.data(QtCore.Qt.ItemDataRole.UserRole))) return tuple(sorted(values)) def set_checked_ids(self, surface_ids: Sequence[str]) -> None: if isinstance(surface_ids, (str, bytes)) or not isinstance( surface_ids, Sequence, ): raise TypeError("surface_ids must be a sequence") wanted = {validate_identifier(value, "surface_id") for value in surface_ids} for index in range(self.items.count()): item = self.items.item(index) item.setCheckState( QtCore.Qt.CheckState.Checked if item.data(QtCore.Qt.ItemDataRole.UserRole) in wanted else QtCore.Qt.CheckState.Unchecked ) @QtCore.Slot(str) def _apply_filter(self, text: str) -> None: normalized = text.strip().casefold() for index in range(self.items.count()): item = self.items.item(index) item.setHidden(normalized not in item.text().casefold()) def _set_all(self, checked: bool) -> None: if not isinstance(checked, bool): raise TypeError("checked must be a Boolean") state = ( QtCore.Qt.CheckState.Checked if checked else QtCore.Qt.CheckState.Unchecked ) for index in range(self.items.count()): self.items.item(index).setCheckState(state) class LicensingAdministrationWindow(QtWidgets.QMainWindow): def __init__(self, client: AdminApiClient) -> None: if not isinstance(client, AdminApiClient): raise TypeError("client must be an AdminApiClient") super().__init__() self._client = client self._thread_pool = QtCore.QThreadPool.globalInstance() self._profiles: dict[str, dict[str, object]] = {} self.setWindowTitle(WINDOW_TITLE) self.resize(1180, 820) self._build_ui() QtCore.QTimer.singleShot(0, self.reload) def _build_ui(self) -> None: central = QtWidgets.QWidget() root = QtWidgets.QVBoxLayout(central) explanation = QtWidgets.QLabel( "Create immutable reusable product profiles, bind them to licenses or " "serial batches, and issue time-bounded trial serials. Existing licenses " "continue to use their signed profile even after that profile is archived." ) explanation.setWordWrap(True) root.addWidget(explanation) self.tabs = QtWidgets.QTabWidget() root.addWidget(self.tabs, 1) self._build_profiles_tab() self._build_serials_tab() self.status_label = QtWidgets.QLabel("Connecting to licensing authority…") self.status_label.setWordWrap(True) root.addWidget(self.status_label) self.setCentralWidget(central) def _build_profiles_tab(self) -> None: tab = QtWidgets.QWidget() layout = QtWidgets.QVBoxLayout(tab) form = QtWidgets.QFormLayout() self.profile_name = QtWidgets.QLineEdit() self.profile_description = QtWidgets.QPlainTextEdit() self.profile_description.setMaximumHeight(72) self.profile_reason = QtWidgets.QLineEdit() self.profile_reason.setPlaceholderText("Audited business reason") form.addRow("Profile name", self.profile_name) form.addRow("Description", self.profile_description) form.addRow("Reason", self.profile_reason) layout.addLayout(form) selectors = QtWidgets.QSplitter(QtCore.Qt.Orientation.Horizontal) self.analysis_checklist = SurfaceChecklist("Analyses") self.tab_checklist = SurfaceChecklist("Primary tabs") selectors.addWidget(self.analysis_checklist) selectors.addWidget(self.tab_checklist) selectors.setStretchFactor(0, 3) selectors.setStretchFactor(1, 2) layout.addWidget(selectors, 1) actions = QtWidgets.QHBoxLayout() self.create_profile_button = QtWidgets.QPushButton("Create immutable profile") self.use_template_button = QtWidgets.QPushButton("Use selected as template") self.archive_profile_button = QtWidgets.QPushButton("Archive selected profile") self.reload_button = QtWidgets.QPushButton("Reload") actions.addWidget(self.create_profile_button) actions.addWidget(self.use_template_button) actions.addWidget(self.archive_profile_button) actions.addStretch(1) actions.addWidget(self.reload_button) layout.addLayout(actions) self.profile_table = QtWidgets.QTreeWidget() self.profile_table.setHeaderLabels( ("Name", "Status", "Analyses", "Tabs", "Profile ID") ) self.profile_table.setRootIsDecorated(False) self.profile_table.setAlternatingRowColors(True) layout.addWidget(self.profile_table) assignment = QtWidgets.QGroupBox("Assign profile to an existing license") assignment_form = QtWidgets.QFormLayout(assignment) self.assignment_license_id = QtWidgets.QLineEdit() self.assignment_profile = QtWidgets.QComboBox() self.assignment_reason = QtWidgets.QLineEdit() self.assignment_reason.setPlaceholderText("Audited assignment reason") self.assign_profile_button = QtWidgets.QPushButton("Assign and require refresh") assignment_form.addRow("License ID", self.assignment_license_id) assignment_form.addRow("Profile", self.assignment_profile) assignment_form.addRow("Reason", self.assignment_reason) assignment_form.addRow("", self.assign_profile_button) layout.addWidget(assignment) self.tabs.addTab(tab, "Product profiles") self.create_profile_button.clicked.connect(self._create_profile) self.use_template_button.clicked.connect(self._use_selected_template) self.archive_profile_button.clicked.connect(self._archive_selected_profile) self.reload_button.clicked.connect(self.reload) self.assign_profile_button.clicked.connect(self._assign_profile) def _build_serials_tab(self) -> None: tab = QtWidgets.QWidget() layout = QtWidgets.QVBoxLayout(tab) note = QtWidgets.QLabel( "Edition, add-on, and trial serials are one-time redemption credentials. " "Subscriptions remain account and billing-provider based and are not serials." ) note.setWordWrap(True) layout.addWidget(note) form = QtWidgets.QFormLayout() self.serial_sku = QtWidgets.QComboBox() self.serial_quantity = QtWidgets.QSpinBox() self.serial_quantity.setRange(1, MAX_SERIAL_BATCH_QUANTITY) self.serial_profile = QtWidgets.QComboBox() self.trial_hours = QtWidgets.QSpinBox() self.trial_hours.setRange( TRIAL_MINIMUM_AUTHORITY_DURATION_HOURS, TRIAL_MAXIMUM_TOTAL_DURATION_HOURS, ) self.trial_hours.setValue(24) self.trial_hours.setSuffix(" hours") self.serial_campaign = QtWidgets.QLineEdit() self.serial_reason = QtWidgets.QLineEdit() self.serial_reason.setPlaceholderText("Audited issuance reason") output_row = QtWidgets.QWidget() output_layout = QtWidgets.QHBoxLayout(output_row) output_layout.setContentsMargins(0, 0, 0, 0) self.serial_output = QtWidgets.QLineEdit() self.serial_output.setReadOnly(True) self.choose_output_button = QtWidgets.QPushButton("Choose…") output_layout.addWidget(self.serial_output, 1) output_layout.addWidget(self.choose_output_button) form.addRow("SKU", self.serial_sku) form.addRow("Quantity", self.serial_quantity) form.addRow("Product profile", self.serial_profile) form.addRow("Trial duration", self.trial_hours) form.addRow("Campaign (optional)", self.serial_campaign) form.addRow("Reason", self.serial_reason) form.addRow("Owner-only output", output_row) layout.addLayout(form) self.generate_serials_button = QtWidgets.QPushButton( "Generate and save plaintext serials once" ) layout.addWidget(self.generate_serials_button) layout.addStretch(1) self.tabs.addTab(tab, "Serial issuance") self.serial_sku.currentIndexChanged.connect(self._serial_sku_changed) self.choose_output_button.clicked.connect(self._choose_serial_output) self.generate_serials_button.clicked.connect(self._generate_serials) @QtCore.Slot() def reload(self) -> None: self._set_status("Loading product inventory and active profiles…") self._run_operation( lambda: self._client.request( "GET", "/v1/admin/product-surfaces/inventory", ), self._inventory_loaded, ) self._run_operation( lambda: self._client.request( "GET", "/v1/admin/product-surface-profiles?includeArchived=true", ), self._profiles_loaded, ) self._run_operation( lambda: self._client.request("GET", "/v1/catalog/public"), self._catalog_loaded, ) def _run_operation( self, operation: Callable[[], object], on_success: Callable[[object], None], *, controls: Sequence[QtWidgets.QWidget] = (), ) -> None: if not callable(operation) or not callable(on_success): raise TypeError("operation and on_success must be callable") for control in controls: control.setEnabled(False) worker = _AdminOperation(operation) def succeeded(result: object) -> None: for control in controls: control.setEnabled(True) on_success(result) def failed(message: str) -> None: for control in controls: control.setEnabled(True) self._operation_failed(message) worker.signals.succeeded.connect(succeeded) worker.signals.failed.connect(failed) self._thread_pool.start(worker) def _inventory_loaded(self, result: object) -> None: if not isinstance(result, Mapping) or not isinstance(result.get("surfaces"), list): self._operation_failed("product inventory response is invalid") return analyses = [] tabs = [] for surface in result["surfaces"]: if not isinstance(surface, Mapping): self._operation_failed("product inventory contains an invalid surface") return if surface.get("kind") == "analysis": analyses.append(surface) elif surface.get("kind") == "main_tab": tabs.append(surface) self.analysis_checklist.populate(analyses) self.tab_checklist.populate(tabs) self._set_status( f"Loaded {len(analyses)} analyses and {len(tabs)} primary tabs." ) def _profiles_loaded(self, result: object) -> None: if not isinstance(result, Mapping) or not isinstance(result.get("profiles"), list): self._operation_failed("surface profile response is invalid") return self._profiles.clear() self.profile_table.clear() for value in result["profiles"]: if not isinstance(value, Mapping): self._operation_failed("surface profile response contains an invalid row") return profile = dict(value) profile_id = validate_identifier(profile.get("profileId"), "profile_id") policy = profile.get("policy") if not isinstance(policy, Mapping): self._operation_failed("surface profile has no policy") return self._profiles[profile_id] = profile item = QtWidgets.QTreeWidgetItem( ( str(profile.get("name") or ""), str(profile.get("status") or ""), str(len(policy.get("analysisIds") or [])), str(len(policy.get("mainTabIds") or [])), profile_id, ) ) item.setData(0, QtCore.Qt.ItemDataRole.UserRole, profile_id) self.profile_table.addTopLevelItem(item) self.profile_table.resizeColumnToContents(0) self._populate_profile_combos() def _catalog_loaded(self, result: object) -> None: if not isinstance(result, Mapping) or not isinstance(result.get("skus"), Mapping): self._operation_failed("public catalog response is invalid") return selected_sku = self.serial_sku.currentData(QtCore.Qt.ItemDataRole.UserRole) selected_sku_id = ( selected_sku.get("skuId") if isinstance(selected_sku, Mapping) else None ) self.serial_sku.clear() for sku_id, value in sorted(result["skus"].items()): if not isinstance(value, Mapping): continue kind = value.get("kind") if value.get("active") is not True or kind not in SUPPORTED_SERIAL_SKU_KINDS: continue label = str(value.get("label") or sku_id) self.serial_sku.addItem( f"{label} · {sku_id}", {"skuId": sku_id, "kind": kind}, ) if selected_sku_id is not None: for index in range(self.serial_sku.count()): data = self.serial_sku.itemData(index) if isinstance(data, Mapping) and data.get("skuId") == selected_sku_id: self.serial_sku.setCurrentIndex(index) break self._serial_sku_changed() def _populate_profile_combos(self) -> None: for combo in (self.assignment_profile, self.serial_profile): previous = combo.currentData(QtCore.Qt.ItemDataRole.UserRole) combo.clear() combo.addItem(FULL_PRODUCT_LABEL, None) for profile_id, profile in sorted( self._profiles.items(), key=lambda item: (str(item[1].get("name") or ""), item[0]), ): if profile.get("status") != "active": continue combo.addItem(str(profile.get("name") or profile_id), profile_id) index = combo.findData(previous) if index >= 0: combo.setCurrentIndex(index) @QtCore.Slot() def _create_profile(self) -> None: name = self.profile_name.text().strip() description = self.profile_description.toPlainText().strip() reason = self.profile_reason.text().strip() main_tab_ids = self.tab_checklist.checked_ids() if not name or not description or not reason: self._operation_failed("Profile name, description, and reason are required.") return if not main_tab_ids: self._operation_failed("At least one primary tab must remain included.") return analysis_ids = self.analysis_checklist.checked_ids() body = { "name": name, "description": description, "analysisIds": list(analysis_ids), "mainTabIds": list(main_tab_ids), "reason": reason, "correlationId": "correlation.surface_profile_create." + hashlib.sha256( canonicalize_json( { "name": name, "analysisIds": list(analysis_ids), "mainTabIds": list(main_tab_ids), } ) ).hexdigest()[:32], } self._set_status("Creating immutable product profile…") self._run_operation( lambda: self._client.request( "POST", "/v1/admin/product-surface-profiles", body, ), self._profile_created, controls=(self.create_profile_button,), ) def _profile_created(self, result: object) -> None: if not isinstance(result, Mapping): self._operation_failed("profile creation response is invalid") return self.profile_name.clear() self.profile_description.clear() self.profile_reason.clear() self._set_status(f"Created profile {result.get('profileId')}.") self.reload() @QtCore.Slot() def _use_selected_template(self) -> None: profile = self._selected_profile() if profile is None: self._operation_failed("Select a profile to use as a template.") return policy = profile.get("policy") if not isinstance(policy, Mapping): self._operation_failed("Selected profile has no valid policy.") return self.profile_name.setText(f"{profile.get('name', '')} copy") self.profile_description.setPlainText(str(profile.get("description") or "")) self.analysis_checklist.set_checked_ids(policy.get("analysisIds") or []) self.tab_checklist.set_checked_ids(policy.get("mainTabIds") or []) @QtCore.Slot() def _archive_selected_profile(self) -> None: profile = self._selected_profile() if profile is None: self._operation_failed("Select an active profile to archive.") return if profile.get("status") != "active": self._operation_failed("The selected profile is already archived.") return reason = self.profile_reason.text().strip() if not reason: self._operation_failed("Enter an audited reason before archiving.") return profile_id = str(profile["profileId"]) body = { "reason": reason, "correlationId": "correlation.surface_profile_archive." + hashlib.sha256( canonicalize_json( {"profileId": profile_id, "reason": reason} ) ).hexdigest()[:32], } self._run_operation( lambda: self._client.request( "POST", f"/v1/admin/product-surface-profiles/{profile_id}/archive", body, ), lambda _result: self.reload(), controls=(self.archive_profile_button,), ) def _selected_profile(self) -> Optional[dict[str, object]]: selected = self.profile_table.selectedItems() if len(selected) != 1: return None profile_id = selected[0].data(0, QtCore.Qt.ItemDataRole.UserRole) return self._profiles.get(str(profile_id)) @QtCore.Slot() def _assign_profile(self) -> None: try: license_id = validate_identifier( self.assignment_license_id.text(), "license_id", ) except (TypeError, ValueError) as exc: self._operation_failed(str(exc)) return reason = self.assignment_reason.text().strip() if not reason: self._operation_failed("An audited assignment reason is required.") return profile_id = self.assignment_profile.currentData( QtCore.Qt.ItemDataRole.UserRole ) body = { "surfaceProfileId": profile_id, "reason": reason, "correlationId": "correlation.license_surface_assign." + hashlib.sha256( canonicalize_json( { "licenseId": license_id, "surfaceProfileId": profile_id, "reason": reason, } ) ).hexdigest()[:32], } self._run_operation( lambda: self._client.request( "POST", f"/v1/admin/licenses/{license_id}/surface-profile", body, ), self._profile_assigned, controls=(self.assign_profile_button,), ) def _profile_assigned(self, result: object) -> None: if not isinstance(result, Mapping): self._operation_failed("license assignment response is invalid") return self._set_status( "License profile updated. Active clients will receive it on their next " "successful refresh." ) @QtCore.Slot() def _serial_sku_changed(self) -> None: data = self.serial_sku.currentData(QtCore.Qt.ItemDataRole.UserRole) kind = data.get("kind") if isinstance(data, Mapping) else None is_trial = kind == "trial" supports_profile = kind in ("edition", "trial") self.trial_hours.setEnabled(is_trial) self.serial_profile.setEnabled(supports_profile) @QtCore.Slot() def _choose_serial_output(self) -> None: selected, _selected_filter = QtWidgets.QFileDialog.getSaveFileName( self, "Save one-time serial export", self.serial_output.text(), SERIAL_EXPORT_FILTER, ) if selected: self.serial_output.setText(selected) @QtCore.Slot() def _generate_serials(self) -> None: data = self.serial_sku.currentData(QtCore.Qt.ItemDataRole.UserRole) if not isinstance(data, Mapping): self._operation_failed("Select an active serial-eligible SKU.") return output_text = self.serial_output.text().strip() reason = self.serial_reason.text().strip() if not output_text or not reason: self._operation_failed("Choose an output file and enter an audited reason.") return output_path = Path(output_text).expanduser().resolve() if output_path.exists() or output_path.is_symlink(): self._operation_failed("Output already exists; choose a new file.") return profile_id = self.serial_profile.currentData( QtCore.Qt.ItemDataRole.UserRole ) kind = data.get("kind") body: dict[str, object] = { "skuId": validate_identifier(data.get("skuId"), "sku_id"), "quantity": self.serial_quantity.value(), "reason": reason, "correlationId": "correlation.serial_batch_generate." + hashlib.sha256( canonicalize_json( { "skuId": data.get("skuId"), "quantity": self.serial_quantity.value(), "profileId": profile_id, "trialDurationHours": ( self.trial_hours.value() if kind == "trial" else None ), "reason": reason, } ) ).hexdigest()[:32], } campaign = self.serial_campaign.text().strip() if campaign: body["campaign"] = campaign if kind in ("edition", "trial") and profile_id is not None: body["surfaceProfileId"] = profile_id if kind == "trial": body["trialDurationHours"] = self.trial_hours.value() def generate_and_store() -> object: response = self._client.request( "POST", "/v1/admin/serial-batches", body, ) serials = response.get("serials") if not isinstance(serials, list) or len(serials) != body["quantity"]: raise RuntimeError("serial generation response has an invalid cohort") _write_owner_only_json(str(output_path), response) return { "batchId": response.get("batchId"), "quantity": len(serials), "output": str(output_path), } self._set_status("Generating and securely saving the one-time serial cohort…") self._run_operation( generate_and_store, self._serials_generated, controls=(self.generate_serials_button,), ) def _serials_generated(self, result: object) -> None: if not isinstance(result, Mapping): self._operation_failed("serial generation result is invalid") return self._set_status( f"Generated {result.get('quantity')} serial(s) in batch " f"{result.get('batchId')}. Plaintext exists only in {result.get('output')}." ) self.serial_output.clear() @QtCore.Slot(str) def _operation_failed(self, message: str) -> None: normalized = message.strip() if isinstance(message, str) else "Unknown error" self._set_status(f"Operation failed: {normalized}") QtWidgets.QMessageBox.warning(self, WINDOW_TITLE, normalized) def _set_status(self, message: str) -> None: if not isinstance(message, str) or not message.strip(): raise ValueError("status message must be non-empty text") self.status_label.setText(message.strip()) def _parser() -> argparse.ArgumentParser: return argparse.ArgumentParser( prog="python -m licensing_server.admin_surface_gui", description=( "Open the licensing product-profile and serial administration console. " "The HTTPS origin and owner-only token file are read from the same " "environment variables as licensing_server.admin_cli." ), ) def main( argv: Optional[Sequence[str]] = None, environ: Optional[Mapping[str, 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().parse_args(argv) resolved_environment = os.environ if environ is None else environ application = QtWidgets.QApplication.instance() or QtWidgets.QApplication([]) try: client = admin_client_from_environment(resolved_environment) except Exception as exc: QtWidgets.QMessageBox.critical( None, WINDOW_TITLE, f"Licensing administration could not start: {exc}", ) return 1 window = LicensingAdministrationWindow(client) window.show() return int(application.exec()) if __name__ == "__main__": raise SystemExit(main())