from __future__ import annotations from pathlib import Path from typing import Optional from PySide6.QtWidgets import QApplication from SRC.Pages.license_activation_dialog import LicenseActivationDialog from SRC.Services.license_api_client import LicenseApiClient, LicenseApiError from SRC.Services.license_machine import build_machine_fingerprint from SRC.Services.license_models import LicenseEnvelope, LicenseRuntimeState from SRC.Services.license_state import ( build_default_state, is_current_eula_accepted, mark_eula_accepted, parse_utc, update_state_after_validation, utc_now, ) from SRC.Services.license_storage import ( LicenseStorageError, clear_runtime_state, ensure_runtime_state, get_license_file_path, load_license_envelope, load_license_envelope_from_file, save_license_envelope, save_runtime_state, ) from SRC.Services.license_validator import validate_license class LicenseGuardResult: def __init__(self, is_allowed: bool, message: str = '') -> None: self.is_allowed = is_allowed self.message = message class OnlineCheckResult: def __init__( self, *, server_contacted: bool, license_allowed: bool, message: Optional[str] = None, ) -> None: self.server_contacted = server_contacted self.license_allowed = license_allowed self.message = message or '' def ensure_license_or_prompt(app: QApplication) -> LicenseGuardResult: api_client = LicenseApiClient() eula_text = _load_eula_text() try: machine = build_machine_fingerprint() state = ensure_runtime_state(machine.fingerprint) envelope = load_license_envelope() except LicenseStorageError as exc: return LicenseGuardResult(False, f'Le stockage local de licence est invalide : {exc}') online_result = OnlineCheckResult(server_contacted=False, license_allowed=True) if envelope is not None and api_client.is_configured(): envelope, state, online_result = try_online_check(envelope, state, api_client) validation, updated_state = validate_license(envelope, state) if online_result.server_contacted and not online_result.license_allowed: validation = type(validation)( False, 'ONLINE_CHECK_REJECTED', online_result.message or 'La licence a été refusée par le serveur.', envelope=validation.envelope, payload=validation.payload, now_utc=validation.now_utc, days_until_expiry=validation.days_until_expiry, ) updated_state = state if validation.is_valid and state is not None and is_current_eula_accepted(state): if updated_state is not None: save_runtime_state(updated_state) return LicenseGuardResult(True, validation.message) dialog = LicenseActivationDialog( validation_result=validation, existing_license_key=_extract_existing_license_key(envelope), eula_text=eula_text, eula_required_only=bool(validation.is_valid and state is not None and not is_current_eula_accepted(state)), ) while dialog.exec(): action = dialog.requested_action if action == 'retry': try: machine = build_machine_fingerprint() state = ensure_runtime_state(machine.fingerprint) if dialog.eula_accepted: state = mark_eula_accepted(state) save_runtime_state(state) envelope = load_license_envelope() except LicenseStorageError as exc: dialog.set_status_message(f'Stockage local invalide : {exc}', is_error=True) continue online_result = OnlineCheckResult(server_contacted=False, license_allowed=True) if envelope is not None and api_client.is_configured(): envelope, state, online_result = try_online_check(envelope, state, api_client) validation, updated_state = validate_license(envelope, state) if online_result.server_contacted and not online_result.license_allowed: validation = type(validation)( False, 'ONLINE_CHECK_REJECTED', online_result.message or 'La licence a été refusée par le serveur.', envelope=validation.envelope, payload=validation.payload, now_utc=validation.now_utc, days_until_expiry=validation.days_until_expiry, ) updated_state = state if validation.is_valid and state is not None and is_current_eula_accepted(state): if updated_state is not None: save_runtime_state(updated_state) return LicenseGuardResult(True, validation.message) if validation.is_valid and state is not None and dialog.eula_accepted: state = mark_eula_accepted(state) save_runtime_state(state) return LicenseGuardResult(True, validation.message) dialog.set_status_message(validation.message, is_error=True) continue if action == 'import': selected_file = dialog.selected_license_file if not selected_file: dialog.set_status_message("Aucun fichier de licence n'a été sélectionné.", is_error=True) continue try: install_local_license_file(selected_file) machine = build_machine_fingerprint() state = ensure_runtime_state(machine.fingerprint) if dialog.eula_accepted: state = mark_eula_accepted(state) save_runtime_state(state) envelope = load_license_envelope() except LicenseStorageError as exc: dialog.set_status_message(f'Stockage local invalide : {exc}', is_error=True) continue validation, updated_state = validate_license(envelope, state) if validation.is_valid: if updated_state is not None: refreshed = update_state_after_validation( updated_state, now_utc=utc_now(), license_hash=updated_state.last_valid_license_hash or '', mark_successful_check=True, ) if dialog.eula_accepted: refreshed = mark_eula_accepted(refreshed) save_runtime_state(refreshed) return LicenseGuardResult(True, validation.message) dialog.set_status_message(validation.message, is_error=True) continue if action == 'activate': license_key = dialog.license_key if not license_key: dialog.set_status_message("Saisissez une clé de licence pour lancer l'activation en ligne.", is_error=True) continue if not api_client.is_configured(): dialog.set_status_message( "Le service d'activation n'est pas configuré sur ce poste.", is_error=True, ) continue try: machine = build_machine_fingerprint() state = ensure_runtime_state(machine.fingerprint) activation = activate_online(license_key=license_key, state=state, api_client=api_client) except LicenseStorageError as exc: dialog.set_status_message(f'Stockage local invalide : {exc}', is_error=True) continue except LicenseApiError as exc: dialog.set_status_message(str(exc), is_error=True) continue if not activation.success or activation.envelope is None: dialog.set_status_message(activation.message or 'Activation refusée par le serveur.', is_error=True) continue save_license_envelope(activation.envelope) state = state or build_default_state(machine.fingerprint) state = _stamp_successful_check(state, activation.envelope, activation.server_time_utc) if dialog.eula_accepted: state = mark_eula_accepted(state) save_runtime_state(state) validation, updated_state = validate_license(activation.envelope, state) if validation.is_valid: if updated_state is not None: if dialog.eula_accepted: updated_state = mark_eula_accepted(updated_state) save_runtime_state(updated_state) return LicenseGuardResult(True, activation.message or validation.message) dialog.set_status_message(validation.message, is_error=True) continue dialog.set_status_message("Action inconnue sur l'écran d'activation.", is_error=True) return LicenseGuardResult(False, validation.message) def install_local_license_file(source_file: str | Path) -> Path: source_path = Path(source_file) if not source_path.exists(): raise FileNotFoundError(source_file) envelope = load_license_envelope_from_file(source_path) if envelope is None: raise LicenseStorageError('Le fichier de licence importé est vide ou invalide.') save_license_envelope(envelope) clear_runtime_state() return get_license_file_path() def activate_online(*, license_key: str, state: LicenseRuntimeState, api_client: LicenseApiClient) -> object: machine = build_machine_fingerprint() return api_client.activate_license( license_key=license_key, installation_id=state.installation_id or '', machine_fingerprint=machine.fingerprint, fingerprint_data=machine.components, ) def try_online_check( envelope: Optional[LicenseEnvelope], state: Optional[LicenseRuntimeState], api_client: LicenseApiClient, ) -> tuple[Optional[LicenseEnvelope], Optional[LicenseRuntimeState], OnlineCheckResult]: if envelope is None: return envelope, state, OnlineCheckResult(server_contacted=False, license_allowed=True) license_key = envelope.payload.license_key.strip() if not license_key: return envelope, state, OnlineCheckResult(server_contacted=False, license_allowed=True) machine = build_machine_fingerprint() working_state = state or ensure_runtime_state(machine.fingerprint) try: response = api_client.check_license( license_key=license_key, installation_id=working_state.installation_id or '', machine_fingerprint=machine.fingerprint, current_license_id=envelope.payload.license_id, fingerprint_data=machine.components, ) except LicenseApiError: return envelope, working_state, OnlineCheckResult(server_contacted=False, license_allowed=True) if not response.success or response.envelope is None: return envelope, working_state, OnlineCheckResult( server_contacted=True, license_allowed=False, message=response.message or 'La licence a été refusée par le serveur.', ) save_license_envelope(response.envelope) working_state = _stamp_successful_check(working_state, response.envelope, response.server_time_utc) save_runtime_state(working_state) return response.envelope, working_state, OnlineCheckResult( server_contacted=True, license_allowed=True, message=response.message, ) def _stamp_successful_check( state: LicenseRuntimeState, envelope: LicenseEnvelope, server_time_utc: Optional[str], ) -> LicenseRuntimeState: reference_time = parse_utc(server_time_utc) or utc_now() license_hash = envelope.payload.license_id or envelope.payload.license_key return update_state_after_validation( state, now_utc=reference_time, license_hash=license_hash, mark_successful_check=True, ) def _extract_existing_license_key(envelope: Optional[LicenseEnvelope]) -> str: if envelope is None: return '' return envelope.payload.license_key.strip() def _load_eula_text() -> str: path = Path(__file__).resolve().parents[1] / 'Assets' / 'eula_dps_trail.txt' if not path.exists(): return 'Le contrat de licence utilisateur final est indisponible sur ce poste.' return path.read_text(encoding='utf-8').strip()