#!/usr/bin/env python3
"""
Donjon Platform - License Verification & Tier Management

PRODUCT-SIDE module: verification only.  No private keys, no license
generation.  Licenses are created by the offline admin tool and validated
here using dual ML-DSA-65 (NIST FIPS 204) + Ed25519 signatures.

SECURITY ARCHITECTURE
---------------------
This module embeds PUBLIC verification keys only.  The corresponding
private keys live exclusively in the offline admin tool
(tools/donjon-license-admin.py) which should never be distributed with
the product.  Compromise of this module does NOT compromise signing
capability.

Tier ladder
-----------
community  - free, no license file needed
pro        - signed license required
enterprise - signed license required
managed    - MSSP/MSP tier, signed license required

License format versions
-----------------------
v1 (legacy)  - HMAC-SHA256 signature (backward compat during transition)
v2 (current) - dual post-quantum + classical signatures

SECURITY WARNING: v1 uses a shared HMAC key embedded in the binary.
This is inherently weak (anyone who decompiles the product can forge
v1 licenses).  v1 support exists ONLY for migration and should be
removed once all deployed licenses have been upgraded to v2.
"""

from __future__ import annotations

import base64
import hashlib
import hmac  # kept for v1 backward compat only -- remove with v1 support
import json
import logging
import os
import platform
import secrets
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional

try:
    from dilithium_py.ml_dsa import ML_DSA_65
except ImportError:
    ML_DSA_65 = None  # graceful degradation -- PQC verification unavailable

try:
    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
except ImportError:
    Ed25519PublicKey = None  # classical verification unavailable

try:
    from .paths import paths
except ImportError:
    from paths import paths

logger = logging.getLogger("donjon.licensing")


# ---------------------------------------------------------------------------
# Public keys for license verification (VERIFICATION ONLY -- no private keys)
#
# These base64-encoded public keys are regenerated by the admin tool's
# ``keygen`` command and pasted here during build / deployment.
#
# SECURITY: If an attacker replaces these values, they can forge licenses.
# Protect the product binary with code signing and integrity checks.
#
# Placeholder values -- replaced during build / deployment with real keys.
# ---------------------------------------------------------------------------
_PUBLIC_KEY_CLASSICAL_B64: str = "A1ISZgiW2Xzgqmb+Ke8HU5ScQzh9VK/U4s4nW+b4Qbg="

_PUBLIC_KEY_PQC_B64: str = (
    "Rq4iSkl/4jvYJmx2Fc1Jrpr7ofQRRDvSm2PqsegesDTu7iy0tMbYKNV0rRKugIvk"
    "XsjmoJn6wzLkTFLjCLWNRaSZnYBwh74vvTPT43IafPciMfz8Gjv8AiVZQBvQss5s"
    "sVJYt4ZFaGBHxS+l8lRbnPrCfAK5JvgngP/DorCtOIqFxtVGWA8blO4qa84yKKjG"
    "MNDmDmJYhNDHzeVQsbCU5dvG53X/Guu9tAbHTCBCeSUfzMaKuT3AIFlH7BOP6XjV"
    "JPbkbAzZNsy3jG6vjLzlGNbR+G+aHC6gdZ32CvcLyB02qnVdsvk87uOttvdqxer"
    "ocemHeNB93DOf/8RwPuWrEmQ+ApkcUHGdD7h7WtGnSMtUez/UIzt6SW2fUREXWzt"
    "ADsZZlw8+aQC05NIyGGuhVPSL0O9m0XFJr8VnvU9tjWSChLJLImOjkF4twXMgSYQ"
    "XIxm8ln+AnDLyCBkTqZ89MAzZtITP8zeec9SsTMRwD24kBkxWLcCTFn0J3qVCv6P"
    "4MgTsAN6oLmIf0Q2u3+SdYHnVX8GC6O2509cISOx3EniNDba6qMYgTwZyxGoV3eT"
    "bkEsiAFld+Irn+GgcHi829zBo3iBoJWt2lW/b9eKwYtFozKX6ZZg83GDnZmSPCwO"
    "TMWsv56GHehakvPvGQoH3C44rSTJYY5rPqP4Ourh/jJeXnj/4v2S/lANvjP+n1NH"
    "bOspApJ3K6/NWFag9xwhdQZx9lstmkYkYdQQWXCBa6ZwWXGpiPs1pHpMQFrjyd8q"
    "WmUUtgiFGat9nxuctsoC3XSOjmPiOtIY2yIjHrhHb+wMr6grX8kpT1TOnzMlcHze"
    "akEyQdbUhXZ5Jda9ZoTSNifhFW58PvpN0XJVmUzy79xCyit2dH8fz0+ujamjxvrL"
    "u7RnyaKllStBdwzr2dZIsJ2PreJkTL3QEVXSjauhPjJuwtpWgKHE9XUKfMxVPSNP"
    "/M6KFAMPVm2J49ALL9+BkFnOsPpVibWhF7M6cg5Zz1a1dadsZgnqtJAg/ZlNvuzn"
    "exvI+B5ykWT2x2080V1gGIpwDt0ZmXqZKY+POLrdRKGzNHPB5Y0CYEkhDsbfEBdY"
    "cxZToDNgC3iuxUTt/vJ7EXyg4G3mP7wlZL7EyvvDlMsPg0+9vloJjk4FDY0i/y/W"
    "7TZePRIDP6ZQ+7SbD9lGP6pjLeX3ndLpiEHqhHYnYJVj/BAhXAoQ0H42iPGbGLbZ"
    "76w0zqODiWxYKeFVjM7d9aTXAh4+UYrSFAe0pDi/XMKUf8vnpS8o0Xb7EZmgXL0D"
    "0Y+C2M+xsMtcRiI1BrfOE6NlzVSMpzDaKcmjS93DemlqiACcjiOSiDNQyzv/1K7P"
    "SWAvK2JEamt/hsmeBzwW3fMUaQpYdqHGukp3pGuBUYPQkCy4K5q/IBkxEOqja4n7"
    "zDSP52nO+x3/VOpXwLKajvCgXzu9rdqlYR30zB0W8VQZpeY7DrowgolizNWLAlE3UH"
    "XACue16+V7iwNCPINhQJd6fyjoXgQ7eFYY9BK/32WAAOLcp8bXX3kb4493kqYdBm"
    "G6h8qsX/IP960PCEBQt2mZ3MXBWtFFky0eeWxf902RWCRKMh9Iw8lO22PTO4aCwB"
    "Qy3jyRSGlxJZ8Rc4YS126NMilfTtbz6ga/Prn/WTP6eK51GFp8kOrllOf1Q9AmiC"
    "rf4LDdmMoVJQiLll10570Og/vpm8TaOLjY/B+rHbA7bPlcUIwD1BB15vbmC4aGoc"
    "GJ+ZH2ccQNEOv7Gwe/USmbgYfUKelPN2OzjJcM+YEu4EzxRF+NrRnbAcn1MCHDlF"
    "6LCWm78SO0z+dIqOESIObUCWQWkX3m5brvcttbiLYZk6CjWQwBl25y3V1QfOh8Ho"
    "vLOoWBNGZnQ32yeUZOSed2Mf/b5IPGtAwI3dsmVpAplFRvfB1jB2Fau+dlSp2g9a"
    "6YFP4p7w67bs2B96DTtUthqtC88HGjwJMfPcjAzq02piWV6NfvuM6+g4KwMDlxgY"
    "gad5FWY0jbEFW/XDXwgvcVWZganX3OfnSYMUh+JnzUS4OyU+9fFVCiLX2pscLksc"
    "857fepcfOhCY4brU2UJpPQTlLe6O8FZlov+sVV74BnGo2NwlRn4pNKyDP8nqNakv"
    "5kMB3tZ9Lls89mAcGQt2rZCH1HQ+RXNTaczW+whTwB5ZILNBdkKy8qEoG6iLGy/l"
    "5OKLAo7zBNvddm2rcXdqLbUUIX2lBk/weAgO2ZDWofsxT5C5siZWhFeuMKw+G6lx"
    "/Tf5AzD6hNejD4dQzCnGYFwgmgUHKVkH993G3F7o1lIn5xwLqCE0UuHTBdOLS9Da"
    "buNyrAMM0TWY5n7lKT6gicmW8/y5o4PIKN+zlqvIC5jUc7+3zjy6QXYdRKHK2UBt"
    "9IQpoaA/ikA4tDaniA0wpnVye4uZkbCfYL/W5vyj3xysyFjMXK0Q0dcC7DezCF8+"
    "pYGv0xel3HGHH+F2TJeOhJKar5IjV9d5IWO5k3SrauYNE+lVx4RNVWiMLGg0Labe"
    "NyCCaQyT0PSQ4ACVaSYyp5eL4pQB+5slLcG7LpBCcs4BEm+lllBlO+Zx/yfC6IEe"
    "brEPPvuB7NfJlu8WbUZGMmXpohfW3kQcu+80paAyxQ=="
)

# Legacy HMAC key for v1 backward compatibility only.
# v1 license support DISABLED.  v1 used a hardcoded HMAC key that any customer
# who reads the source could use to forge arbitrary licenses.  All customers
# must use v2 licenses with dual ML-DSA-65 + Ed25519 signatures.
#
# To temporarily re-enable v1 for a migration window, set the environment
# variable DONJON_ALLOW_V1_LICENSE=1.  This should NEVER be set in production.
_LEGACY_HMAC_KEY: bytes = b"donjon-license-signing-key-v1"
_V1_ENABLED: bool = os.environ.get("DONJON_ALLOW_V1_LICENSE") == "1"


# ---------------------------------------------------------------------------
# MSSP-exclusive feature keys
#
# Centralised list so every tier definition can reference them uniformly.
# This avoids typos when adding/removing MSSP features across tiers.
# ---------------------------------------------------------------------------
_MSSP_FEATURES: List[str] = [
    "client_provisioning",
    "client_dashboards",
    "consolidated_rollup",
    "bulk_scan_orchestration",
    "cross_client_reporting",
    "usage_metering_api",
    "license_sub_allocation",
    "per_client_branding",
    "scan_template_library",
    "max_clients",
]


# ---------------------------------------------------------------------------
# Tier definitions
# ---------------------------------------------------------------------------
# Community is genuinely useful for hands-on manual work:
#   - Run any core scanner, see findings, remediate -- no paywall on workflow
#   - Depth limited to quick/standard; deep scans are Pro+
#   - Scans throttled (artificial delay between checks)
#   - No automation (schedules, alert routing, API bulk ops)
#   - Export limited to CSV/JSON (no HTML reports, PDF, SARIF, etc.)
#   - AI analysis is a Pro feature (Community uses built-in template analysis)
#   - Scan history retained 30 days only
#
# Natural upgrade trigger: "I found issues, now I need a client report /
# scheduled scans / deeper analysis."
# ---------------------------------------------------------------------------

TIERS: Dict[str, Dict[str, Any]] = {
    "community": {
        "label": "Community",
        # -- Scanning --
        "scanners": [
            "network", "vulnerability", "web", "ssl",
            "windows", "linux", "compliance",
        ],
        "scan_depths": ["quick", "standard"],
        "scan_throttle_ms": 500,
        "max_targets_per_scan": 16,
        "scan_history_days": 30,
        # -- Automation --
        "scheduled_scans": False,
        "alert_routing": False,
        # -- Reporting --
        "export_formats": ["csv", "json"],
        "compliance_frameworks": 3,
        # -- Notifications --
        "notifications": ["email", "webhook"],
        "sms_notifications": False,
        # -- AI --
        "ai_queries_per_day": 10,         # DEPRECATED/unused: AI is gated by 'ai_analysis' above (no metering)
        "ai_analysis": False,             # AI analysis is a Pro feature; CE uses built-in template analysis
        "ai_scan_summary": False,
        # -- Platform --
        "max_users": 1,
        "api_rate_limit": 100,
        "save_projects": False,
        # -- Enterprise features (disabled) --
        "sso": False,
        "rbac": False,
        "custom_branding": False,
        "multi_tenant": False,
        "audit_log_export": False,
        "priority_support": False,
        # -- MSSP features (disabled) --
        "client_provisioning": False,
        "client_dashboards": False,
        "consolidated_rollup": False,
        "bulk_scan_orchestration": False,
        "cross_client_reporting": False,
        "usage_metering_api": False,
        "license_sub_allocation": False,
        "per_client_branding": False,
        "scan_template_library": False,
        "max_clients": 0,
    },

    "pro": {
        "label": "Pro",
        # -- Scanning --
        "scanners": "all",
        "scan_depths": "all",
        "scan_throttle_ms": 0,
        "max_targets_per_scan": None,     # None = unlimited
        "scan_history_days": None,        # None = unlimited retention
        # -- Automation --
        "scheduled_scans": True,
        "alert_routing": True,
        # -- Reporting --
        "export_formats": "all",
        "compliance_frameworks": None,    # None = unlimited frameworks
        # -- Notifications --
        "notifications": "all",
        "sms_notifications": True,
        # -- AI --
        "ai_queries_per_day": None,       # None = unlimited
        "ai_analysis": True,              # BYO/local AI unlocked, ungated (runs on customer hardware/key)
        "ai_scan_summary": True,
        # -- Platform --
        "max_users": 25,
        "api_rate_limit": 10_000,
        "save_projects": True,
        # -- Enterprise features (disabled) --
        "sso": False,
        "rbac": False,
        "custom_branding": False,
        "multi_tenant": False,
        "audit_log_export": False,
        "priority_support": False,
        # -- MSSP features (disabled) --
        "client_provisioning": False,
        "client_dashboards": False,
        "consolidated_rollup": False,
        "bulk_scan_orchestration": False,
        "cross_client_reporting": False,
        "usage_metering_api": False,
        "license_sub_allocation": False,
        "per_client_branding": False,
        "scan_template_library": False,
        "max_clients": 0,
    },

    "enterprise": {
        "label": "Enterprise",
        # -- Scanning (same as Pro) --
        "scanners": "all",
        "scan_depths": "all",
        "scan_throttle_ms": 0,
        "max_targets_per_scan": None,
        "scan_history_days": None,
        # -- Automation --
        "scheduled_scans": True,
        "alert_routing": True,
        # -- Reporting --
        "export_formats": "all",
        "compliance_frameworks": None,
        # -- Notifications --
        "notifications": "all",
        "sms_notifications": True,
        # -- AI --
        "ai_queries_per_day": None,
        "ai_analysis": True,              # BYO/local AI unlocked, ungated (runs on customer hardware/key)
        "ai_scan_summary": True,
        # -- Platform --
        "max_users": None,               # None = unlimited seats
        "api_rate_limit": None,           # None = unlimited
        "save_projects": True,
        # -- Enterprise-only --
        "sso": True,
        "rbac": True,
        "custom_branding": True,
        "multi_tenant": True,
        "audit_log_export": True,
        "priority_support": True,
        # -- MSSP features (disabled) --
        "client_provisioning": False,
        "client_dashboards": False,
        "consolidated_rollup": False,
        "bulk_scan_orchestration": False,
        "cross_client_reporting": False,
        "usage_metering_api": False,
        "license_sub_allocation": False,
        "per_client_branding": False,
        "scan_template_library": False,
        "max_clients": 0,
    },

    "managed": {
        "label": "Managed (MSSP/MSP)",
        # -- Scanning (same as Enterprise) --
        "scanners": "all",
        "scan_depths": "all",
        "scan_throttle_ms": 0,
        "max_targets_per_scan": None,
        "scan_history_days": None,
        # -- Automation --
        "scheduled_scans": True,
        "alert_routing": True,
        # -- Reporting --
        "export_formats": "all",
        "compliance_frameworks": None,
        # -- Notifications --
        "notifications": "all",
        "sms_notifications": True,
        # -- AI --
        "ai_queries_per_day": None,
        "ai_analysis": True,              # BYO/local AI unlocked, ungated (runs on customer hardware/key)
        "ai_scan_summary": True,
        # -- Platform --
        "max_users": None,
        "api_rate_limit": None,
        "save_projects": True,
        # -- Enterprise features --
        "sso": True,
        "rbac": True,
        "custom_branding": True,
        "multi_tenant": True,
        "audit_log_export": True,
        "priority_support": True,
        # -- MSSP/MSP exclusive --
        "client_provisioning": True,
        "client_dashboards": True,
        "consolidated_rollup": True,
        "bulk_scan_orchestration": True,
        "cross_client_reporting": True,
        "usage_metering_api": True,
        "license_sub_allocation": True,
        "per_client_branding": True,
        "scan_template_library": True,
        "max_clients": None,             # None = unlimited clients
    },
}

# Valid paid tiers that require a license file.
# Using frozenset for O(1) lookup and immutability.
_PAID_TIERS: frozenset = frozenset({"pro", "enterprise", "managed"})

# Maximum license file size to prevent memory exhaustion from a malicious file.
_MAX_LICENSE_FILE_SIZE: int = 1 * 1024 * 1024  # 1 MiB -- generous for a JSON license


# ---------------------------------------------------------------------------
# Machine fingerprinting
# ---------------------------------------------------------------------------

def generate_machine_fingerprint() -> str:
    """Generate a stable machine fingerprint from hardware identifiers.

    Combines MAC address, hostname, platform string, processor, and
    machine-id (Linux) or MachineGuid (Windows) into a single
    ``sha256:<hex>`` digest.

    SECURITY NOTE: Machine fingerprinting is a commercial deterrent, not
    a cryptographic binding.  A user with root/admin access can spoof
    every component.  The fingerprint prevents casual license sharing
    (e.g. copying license.json to a colleague's laptop) but will not
    stop a determined attacker.

    TODO: Consider adding SMBIOS/BIOS serial and TPM-backed attestation
    for higher-assurance environments.
    """
    parts: List[str] = []

    # MAC address (first available NIC).
    # Note: uuid.getnode() may return a random MAC on some virtualised
    # environments; this is acceptable for our purposes.
    mac = uuid.getnode()
    parts.append(f"mac:{mac:012x}")

    # Hostname -- changes if the user renames the machine.
    parts.append(f"host:{platform.node()}")

    # Platform identifier (OS + architecture).
    parts.append(f"platform:{platform.system()}-{platform.machine()}")

    # Processor string -- may be empty on some platforms.
    parts.append(f"cpu:{platform.processor()}")

    # OS-specific machine ID for extra stability across reboots.
    machine_id = _read_machine_id()
    if machine_id:
        parts.append(f"mid:{machine_id}")

    combined = "|".join(parts).encode("utf-8")
    digest = hashlib.sha256(combined).hexdigest()
    return f"sha256:{digest}"


def _read_machine_id() -> str:
    """Read the OS machine ID if available. Returns empty string on failure.

    Linux: /etc/machine-id or /var/lib/dbus/machine-id (systemd-generated)
    Windows: HKLM\\SOFTWARE\\Microsoft\\Cryptography\\MachineGuid
    """
    # Linux paths -- try the primary location first, then the D-Bus fallback.
    for path in ("/etc/machine-id", "/var/lib/dbus/machine-id"):
        try:
            content = Path(path).read_text(encoding="utf-8").strip()
            if content:
                return content
        except OSError:
            continue

    # Windows: read MachineGuid from the registry.
    if platform.system() == "Windows":
        try:
            import winreg
            key = winreg.OpenKey(
                winreg.HKEY_LOCAL_MACHINE,
                r"SOFTWARE\Microsoft\Cryptography",
            )
            try:
                value, _ = winreg.QueryValueEx(key, "MachineGuid")
                return str(value)
            finally:
                # Always close the registry key to avoid resource leaks.
                winreg.CloseKey(key)
        except OSError:
            pass

    return ""


# ---------------------------------------------------------------------------
# License Manager
# ---------------------------------------------------------------------------

class LicenseManager:
    """Reads the license file, verifies signatures, and enforces tier limits.

    Supports two license format versions:
    - v1 (legacy): HMAC-SHA256 signature -- kept for backward compatibility
    - v2 (current): dual ML-DSA-65 + Ed25519 signatures

    Thread Safety: This class is designed for single-threaded use within a
    single process.  The singleton pattern ensures one instance per process.
    If used in a multi-threaded server, external synchronisation is needed
    around ``reload()`` and the AI quota methods.

    SECURITY INVARIANT: On any validation failure, the tier MUST fall back
    to "community".  Never partially trust a license.
    """

    _instance: Optional[LicenseManager] = None

    def __init__(self) -> None:
        self.license_path: Path = paths.data / "license.json"
        self._license: Optional[Dict[str, Any]] = None
        self._tier: str = "community"
        self._ai_query_count: int = 0
        self._ai_query_date: str = ""

        self._load_ai_quota()
        self._load_license()

    # ------------------------------------------------------------------
    # Singleton accessor
    # ------------------------------------------------------------------

    @classmethod
    def get_instance(cls) -> LicenseManager:
        """Return the singleton LicenseManager, creating it on first call.

        Why singleton: The license state (loaded tier, AI quota counter)
        must be consistent across all call sites within the process.
        """
        if cls._instance is None:
            cls._instance = cls()
        return cls._instance

    # ------------------------------------------------------------------
    # License file loading
    # ------------------------------------------------------------------

    def _load_license(self) -> None:
        """Load and validate the license file if it exists.

        On ANY failure (missing file, parse error, invalid signature,
        expired, wrong machine, revoked), we fall back to community tier.
        This is a deliberate security decision: partial trust is dangerous.
        """
        if not self.license_path.exists():
            # Check for a server-issued, SIGNED trial license as fallback.
            #
            # SECURITY (blocker G7): the trial file is a full v2 signed
            # license issued online by the Worker. It is honored ONLY if it
            # passes the SAME dual-signature + expiry + fingerprint +
            # revocation chain as a paid license. An unsigned, tampered,
            # expired, or wrong-machine trial falls back to community.
            trial_path = self.license_path.parent / "trial-license.json"
            if trial_path.exists():
                if self._load_signed_trial(trial_path):
                    return
            self._tier = "community"
            return

        try:
            # Guard against oversized license files.
            file_size = self.license_path.stat().st_size
            if file_size > _MAX_LICENSE_FILE_SIZE:
                logger.warning(
                    "License file too large (%d bytes, max %d) -- community tier",
                    file_size, _MAX_LICENSE_FILE_SIZE,
                )
                self._tier = "community"
                return

            data = json.loads(self.license_path.read_text("utf-8"))
        except (json.JSONDecodeError, OSError) as exc:
            logger.warning("Failed to read license file: %s", exc)
            self._tier = "community"
            return

        if not isinstance(data, dict):
            logger.warning("License file root is not a JSON object -- community tier")
            self._tier = "community"
            return

        # Determine format version.  The admin tool writes "version";
        # accept "format_version" as a legacy alias.
        # SECURITY: Default to version 2 (not 1) to prevent downgrade attacks.
        # An attacker who omits "version" should NOT fall into the weak v1 path.
        version = data.get("version", data.get("format_version", 2))

        if version >= 2:
            valid = self._validate_license_v2(data)
        elif _V1_ENABLED:
            logger.warning(
                "SECURITY: Validating v1 license (HMAC). "
                "v1 is deprecated and should be migrated to v2."
            )
            valid = self._validate_license_v1(data)
        else:
            logger.warning(
                "v1 license rejected -- v1 support is disabled. "
                "Set DONJON_ALLOW_V1_LICENSE=1 to temporarily re-enable, "
                "or upgrade the license to v2 format."
            )
            valid = False

        if not valid:
            logger.warning("License signature verification failed -- community tier")
            self._tier = "community"
            return

        # Check expiry date.
        if not self._check_expiry(data):
            logger.warning("License expired -- community tier")
            self._tier = "community"
            return

        # Check machine fingerprint binding (if present in the license).
        if not self._check_machine_fingerprint(data):
            logger.warning("License not valid for this machine -- community tier")
            self._tier = "community"
            return

        # Check whether the license has been revoked.
        if self._is_revoked(data):
            logger.warning("License has been revoked -- community tier")
            self._tier = "community"
            return

        # Validate the tier value against the known tier table.
        # Why: An attacker could craft a license with tier="superadmin" to
        # see if any code path treats unknown tiers as privileged.
        tier = data.get("tier", "community")
        if tier not in TIERS:
            logger.warning("Unknown tier '%s' in license -- community tier", tier)
            self._tier = "community"
            return

        self._license = data
        self._tier = tier
        logger.info(
            "License loaded: tier=%s org=%s",
            self._tier,
            data.get("organization", ""),
        )

    def _load_signed_trial(self, trial_path: Path) -> bool:
        """Load and FULLY VALIDATE a server-issued signed trial license.

        Returns True (and sets self._tier / self._license) only when the
        trial passes the complete signed-license verification chain:
        dual signature, expiry, machine fingerprint, revocation, and a
        known tier. Returns False on any failure (caller falls back to
        community).

        SECURITY: This is the ONLY trust path for a trial. There is no
        "trust the JSON contents" shortcut — the old unsigned trial that
        let anyone hand-write tier='pro' is gone.
        """
        try:
            file_size = trial_path.stat().st_size
            if file_size > _MAX_LICENSE_FILE_SIZE:
                logger.warning("Trial license file too large -- ignoring")
                return False
            data = json.loads(trial_path.read_text("utf-8"))
        except (json.JSONDecodeError, OSError) as exc:
            logger.debug("Could not read trial license: %s", exc)
            return False

        if not isinstance(data, dict):
            return False

        # A trial MUST be flagged trial=true. This selects the trial
        # signature policy (Ed25519-only) instead of the paid dual-sig
        # policy -- the Cloudflare Worker cannot produce an ML-DSA signature
        # in the Pyodide runtime, so trials are classical-signed only.
        if data.get("trial") is not True:
            logger.info("trial-license.json is not flagged trial -- ignoring")
            return False

        # Verify the Ed25519 signature over the canonical payload, then run
        # the same expiry / fingerprint / revocation / tier checks as paid
        # licenses. This is the ONLY trust path for a trial.
        if not self._validate_trial_signature(data):
            logger.info("Trial signature verification failed -- community tier")
            return False
        if not self._check_expiry(data):
            logger.info("Trial license expired -- community tier")
            return False
        if not self._check_machine_fingerprint(data):
            logger.info("Trial license not valid for this machine -- community tier")
            return False
        if self._is_revoked(data):
            logger.info("Trial license revoked -- community tier")
            return False

        tier = data.get("tier", "community")
        if tier not in TIERS:
            return False

        self._license = data
        self._tier = tier
        logger.info("Signed trial license active: tier=%s", self._tier)
        return True

    def _validate_trial_signature(self, data: Dict[str, Any]) -> bool:
        """Verify a trial's Ed25519 signature over the canonical payload.

        Trials are issued online by the license Worker and signed with
        Ed25519 only (the Worker's Pyodide runtime cannot produce an
        ML-DSA-65 signature). The signing key is the SAME Ed25519 key whose
        public half is embedded as ``_PUBLIC_KEY_CLASSICAL_B64``, so a forged
        or tampered trial cannot pass.

        Returns True only if the classical signature verifies. If the
        Ed25519 backend or public key is unavailable, fails CLOSED (returns
        False) -- we never honour an unverifiable trial.
        """
        sigs = data.get("signatures")
        if not isinstance(sigs, dict):
            return False
        classical_sig_b64 = sigs.get("classical", "")
        if not classical_sig_b64:
            return False

        # Canonical payload: everything except the signatures block.
        payload_dict = {k: v for k, v in data.items() if k != "signatures"}
        payload = json.dumps(
            payload_dict, sort_keys=True, separators=(",", ":")
        ).encode("utf-8")

        try:
            classical_sig = base64.b64decode(classical_sig_b64, validate=True)
        except Exception:
            logger.debug("Trial classical signature is not valid base64")
            return False

        result = self._verify_ed25519(payload, classical_sig)
        # None = backend/key unavailable -> fail closed for trials.
        return result is True

    def reload(self) -> None:
        """Re-read and re-validate the license file from disk.

        Call this after the user uploads a new license file via the UI
        or API, to pick up changes without restarting the process.
        """
        self._license = None
        self._tier = "community"
        self._load_license()

    # ------------------------------------------------------------------
    # Signature verification -- v1 (legacy HMAC)
    # ------------------------------------------------------------------

    def _validate_license_v1(self, data: Dict[str, Any]) -> bool:
        """Validate a v1 license using HMAC-SHA256.

        SECURITY WARNING: v1 uses a shared secret embedded in the binary.
        Anyone who decompiles the product can forge v1 licenses.  This
        method exists ONLY for backward compatibility during migration
        to the v2 dual-signature scheme.

        TODO: Remove this method once all deployed licenses are v2.
        Log a deprecation warning so operators know to upgrade.
        """
        logger.warning(
            "Validating a v1 (HMAC) license -- this format is deprecated "
            "and will be removed in a future release.  Please upgrade to v2."
        )

        signature = data.get("signature", "")
        tier = data.get("tier", "")
        expires = data.get("expires", "")
        org = data.get("organization", "")

        if tier not in _PAID_TIERS:
            return False

        # Constant-time comparison via hmac.compare_digest to prevent
        # timing side-channels that could allow signature guessing.
        payload = f"{tier}:{org}:{expires}".encode("utf-8")
        expected = hmac.new(_LEGACY_HMAC_KEY, payload, hashlib.sha256).hexdigest()
        return hmac.compare_digest(signature, expected)

    # ------------------------------------------------------------------
    # Signature verification -- v2 (dual PQC + classical)
    # ------------------------------------------------------------------

    def _validate_license_v2(self, data: Dict[str, Any]) -> bool:
        """Validate a v2 license with dual ML-DSA-65 + Ed25519 signatures.

        Both signatures must pass for the license to be accepted.
        If either crypto library is missing, that check returns None
        (graceful degradation), but at least one library MUST be present.

        SECURITY: The "belt and suspenders" approach means:
        - If a quantum computer breaks Ed25519, ML-DSA-65 still holds.
        - If a flaw is found in ML-DSA-65, Ed25519 still holds.
        - An attacker must break BOTH to forge a license.
        """
        sigs = data.get("signatures")
        if not isinstance(sigs, dict):
            logger.error("License v2 missing 'signatures' object")
            return False

        classical_sig_b64 = sigs.get("classical", "")
        pqc_sig_b64 = sigs.get("pqc", "")

        if not classical_sig_b64 or not pqc_sig_b64:
            logger.error("License v2 missing one or both signatures")
            return False

        # Build canonical payload: everything except the signatures block.
        # This MUST match the canonical_payload() function in the admin tool.
        # WARNING: Any divergence between admin tool and product-side
        # canonicalisation will silently invalidate all licenses.
        payload_dict = {k: v for k, v in data.items() if k != "signatures"}
        payload = json.dumps(
            payload_dict, sort_keys=True, separators=(",", ":")
        ).encode("utf-8")

        # Decode base64 signatures with strict validation to reject
        # malformed input (e.g. embedded newlines, wrong padding).
        try:
            classical_sig = base64.b64decode(classical_sig_b64, validate=True)
        except Exception:
            logger.error("Classical signature is not valid base64")
            return False

        try:
            pqc_sig = base64.b64decode(pqc_sig_b64, validate=True)
        except Exception:
            logger.error("PQC signature is not valid base64")
            return False

        classical_valid = self._verify_ed25519(payload, classical_sig)
        pqc_valid = self._verify_ml_dsa(payload, pqc_sig)

        # At least one crypto backend must be available.
        if Ed25519PublicKey is None and ML_DSA_65 is None:
            logger.error(
                "No cryptographic backends available -- "
                "install 'cryptography' and/or 'dilithium-py'"
            )
            return False

        # Both must pass.  A None result means the backend was
        # unavailable; we allow that only if the other passed.
        if classical_valid is None and pqc_valid is None:
            return False
        if classical_valid is False or pqc_valid is False:
            return False

        return True

    def _verify_ed25519(
        self, payload: bytes, signature: bytes
    ) -> Optional[bool]:
        """Verify an Ed25519 signature. Returns None if library unavailable.

        Returns:
            True  -- signature is valid
            False -- signature is invalid (library present, verification failed)
            None  -- library or key not available (graceful degradation)
        """
        if Ed25519PublicKey is None:
            logger.warning("Ed25519 verification skipped -- 'cryptography' not installed")
            return None

        if not _PUBLIC_KEY_CLASSICAL_B64:
            logger.warning("Ed25519 public key not embedded -- skipping")
            return None

        try:
            pub_bytes = base64.b64decode(_PUBLIC_KEY_CLASSICAL_B64, validate=True)
            pub_key = Ed25519PublicKey.from_public_bytes(pub_bytes)
            pub_key.verify(signature, payload)
            return True
        except Exception:
            # SECURITY: Log at debug level only.  Do not include the exception
            # message in user-visible output -- it could leak information about
            # the expected signature format.
            logger.debug("Ed25519 verification failed")
            return False

    def _verify_ml_dsa(
        self, payload: bytes, signature: bytes
    ) -> Optional[bool]:
        """Verify an ML-DSA-65 signature. Returns None if library unavailable.

        Returns:
            True  -- signature is valid
            False -- signature is invalid (library present, verification failed)
            None  -- library or key not available (graceful degradation)
        """
        if ML_DSA_65 is None:
            logger.warning("ML-DSA-65 verification skipped -- 'dilithium-py' not installed")
            return None

        if not _PUBLIC_KEY_PQC_B64:
            logger.warning("ML-DSA-65 public key not embedded -- skipping")
            return None

        try:
            # Strip whitespace/padding artifacts from multi-line string concat
            pub_bytes = base64.b64decode(_PUBLIC_KEY_PQC_B64.strip())
            valid = ML_DSA_65.verify(pub_bytes, payload, signature)
            return bool(valid)
        except Exception as exc:
            logger.debug("ML-DSA-65 verification failed: %s", exc)
            return False

    # ------------------------------------------------------------------
    # Expiry, fingerprint, and revocation checks
    # ------------------------------------------------------------------

    def _check_expiry(self, data: Dict[str, Any]) -> bool:
        """Return True if the license has not expired.

        An empty or missing 'expires' field means the license is perpetual.
        Malformed dates are treated as expired (fail closed).
        """
        expires = data.get("expires")
        if not expires:
            return True  # No expiry date means perpetual license

        try:
            # Handle the "Z" suffix that fromisoformat doesn't accept
            # in Python < 3.11.
            exp_str = expires.replace("Z", "+00:00")
            exp_dt = datetime.fromisoformat(exp_str)

            # Always compare in UTC to avoid timezone confusion.
            if exp_dt.tzinfo is not None:
                now = datetime.now(timezone.utc)
            else:
                # If the date has no timezone, assume UTC.
                now = datetime.now(timezone.utc).replace(tzinfo=None)

            return exp_dt >= now
        except (ValueError, TypeError):
            # Malformed date -- fail closed (treat as expired).
            logger.warning("Malformed expiry date: %s", expires)
            return False

    def _check_machine_fingerprint(self, data: Dict[str, Any]) -> bool:
        """If the license carries a fingerprint, verify it matches this machine.

        An EMPTY/absent fingerprint means the license is an intentional
        "floating" license (e.g. an enterprise site license) that is not
        bound to any specific machine -- accepted.

        POLICY (tightened for G7): if the license DECLARES a fingerprint,
        it MUST match this machine. A mismatch FAILS CLOSED (returns False
        -> community tier). The previous "warn but allow" behaviour turned
        every machine-bound license into a floating one, letting a copied
        license.json (or a copied trial) run anywhere. A determined attacker
        with admin rights can still spoof the fingerprint, but casual
        copy-and-share is now blocked.

        SECURITY: Uses constant-time comparison via hmac.compare_digest
        to prevent timing side-channels.  While fingerprints are not
        secret, timing differences could reveal how many characters
        matched, aiding brute-force fingerprint guessing.
        """
        fp = data.get("machine_fingerprint")
        if not fp:
            # No fingerprint = floating license, not bound to a machine.
            return True

        current_fp = generate_machine_fingerprint()
        # Constant-time comparison to prevent timing side-channels.
        if not hmac.compare_digest(str(fp), current_fp):
            logger.warning(
                "Machine fingerprint mismatch: license is bound to a "
                "different machine -- rejecting (community tier). "
                "Expected=%s..., Got=%s...",
                str(fp)[:16], current_fp[:16],
            )
            return False
        return True

    def _is_revoked(self, data: Dict[str, Any]) -> bool:
        """Check whether this license ID appears in the local revocation list.

        The revocation list is a JSON file maintained by the admin tool
        and optionally synced from the Cloudflare Worker.  It supports
        two formats:
        - Flat string array: ["DON-2026-A1B2C", ...]
        - Object array: [{"license_id": "DON-2026-A1B2C", ...}, ...]

        Returns True if revoked, False otherwise.  On any read/parse
        error, returns True (fail closed -- treat as revoked when the
        revocation list cannot be verified).
        """
        license_id = data.get("license_id", "")
        if not license_id:
            return False

        revoked_path = paths.data / "revoked.json"
        if not revoked_path.exists():
            return False

        try:
            # Guard against oversized revocation files.
            file_size = revoked_path.stat().st_size
            if file_size > _MAX_LICENSE_FILE_SIZE:
                logger.warning("SECURITY: Revocation file too large -- treating as revoked (fail closed)")
                return True

            revoked_list = json.loads(revoked_path.read_text("utf-8"))
            if not isinstance(revoked_list, list):
                logger.warning("SECURITY: revoked.json is not a JSON array -- treating as revoked (fail closed)")
                return True

            # Support both flat string arrays and object arrays.
            for entry in revoked_list:
                if isinstance(entry, str) and entry == license_id:
                    return True
                if isinstance(entry, dict) and entry.get("license_id") == license_id:
                    return True
            return False
        except (json.JSONDecodeError, OSError) as exc:
            logger.warning("SECURITY: Failed to read revocation list: %s -- treating as revoked (fail closed)", exc)
            return True

    # ------------------------------------------------------------------
    # Tier queries
    # ------------------------------------------------------------------

    def get_tier(self) -> str:
        """Return the current tier name (e.g. 'community', 'pro')."""
        return self._tier

    def get_tier_label(self) -> str:
        """Return the human-readable display name for the current tier."""
        return TIERS.get(self._tier, TIERS["community"]).get("label", "Community")

    def get_limits(self) -> Dict[str, Any]:
        """Return a copy of the full limits dict for the current tier.

        Returns a copy (via dict()) so callers cannot mutate the
        canonical tier definitions.
        """
        base_limits = dict(TIERS.get(self._tier, TIERS["community"]))

        # Apply per-license feature overrides (e.g. custom max_users
        # for a managed-tier license).
        # SECURITY: Only numeric keys may be overridden, and only downward
        # (more restrictive).  Boolean feature flags (sso, rbac, etc.)
        # MUST NOT be overridable via license overrides.
        _NUMERIC_OVERRIDABLE_KEYS = {
            "max_users", "max_clients", "max_targets_per_scan",
            "ai_queries_per_day", "compliance_frameworks", "scan_throttle_ms",
            "api_rate_limit", "scan_history_days",
        }
        if self._license:
            overrides = self._license.get("features_override")
            if isinstance(overrides, dict):
                for key, value in overrides.items():
                    if key not in _NUMERIC_OVERRIDABLE_KEYS:
                        logger.warning(
                            "SECURITY: features_override tried to set non-numeric key '%s' -- ignored", key)
                        continue
                    if key not in base_limits:
                        continue
                    if not isinstance(value, (int, float)):
                        logger.warning(
                            "SECURITY: features_override key '%s' has non-numeric value -- ignored", key)
                        continue
                    tier_default = base_limits[key]
                    # None means unlimited -- any numeric override is more
                    # restrictive, so allow it.
                    if tier_default is not None:
                        # scan_throttle_ms: higher = more restrictive (slower)
                        if key == "scan_throttle_ms":
                            if value < tier_default:
                                logger.warning(
                                    "SECURITY: features_override '%s' value %s is less restrictive than tier default %s -- ignored",
                                    key, value, tier_default)
                                continue
                        else:
                            # For all other numeric keys: lower = more restrictive
                            if value > tier_default:
                                logger.warning(
                                    "SECURITY: features_override '%s' value %s exceeds tier default %s -- ignored",
                                    key, value, tier_default)
                                continue
                    base_limits[key] = value

        return base_limits

    def get_license_info(self) -> Dict[str, Any]:
        """Return non-secret license metadata for display in the UI.

        SECURITY: This method intentionally omits signature data and
        any internal fields.  Only fields safe for end-user display
        are included.
        """
        if self._license:
            return {
                "tier": self._tier,
                "organization": self._license.get("organization", ""),
                "expires": self._license.get("expires", ""),
                "license_id": self._license.get("license_id", ""),
                "format_version": self._license.get("version",
                                                     self._license.get("format_version", 1)),
                "valid": True,
            }
        return {
            "tier": "community",
            "organization": "",
            "expires": "",
            "license_id": "",
            "format_version": 0,
            "valid": True,
        }

    def days_until_expiry(self) -> Optional[int]:
        """Return number of days until the license expires, or None if perpetual.

        Returns negative values for already-expired licenses.
        Community tier (no license) returns None (no expiry concept).
        """
        if not self._license:
            return None
        expires = self._license.get("expires")
        if not expires:
            return None  # Perpetual license
        try:
            exp_str = expires.replace("Z", "+00:00")
            exp_dt = datetime.fromisoformat(exp_str)
            if exp_dt.tzinfo is not None:
                now = datetime.now(timezone.utc)
            else:
                now = datetime.now(timezone.utc).replace(tzinfo=None)
            delta = exp_dt - now
            return delta.days
        except (ValueError, TypeError):
            return -1  # Malformed date = treat as expired

    # ------------------------------------------------------------------
    # Limit checking
    # ------------------------------------------------------------------

    def check_limit(self, feature: str, current_count: int = 0) -> bool:
        """Check whether the current tier allows *feature*.

        For boolean features (``scheduled_scans``, ``sso``, ...) returns
        the boolean value directly.

        For numeric limits (``max_targets_per_scan``, ``compliance_frameworks``,
        ...) checks *current_count* against the cap.  ``None`` means unlimited.

        For list or ``"all"`` features, always returns True.  Use
        :meth:`check_feature_item` for item-level checks on those.
        """
        limits = self.get_limits()
        value = limits.get(feature)

        if value is None:
            return True  # None = unlimited
        if isinstance(value, bool):
            return value
        if isinstance(value, int):
            return current_count < value
        if isinstance(value, list) or value == "all":
            return True
        # Unknown value type -- default to allowed (fail open for UX).
        return True

    def check_feature_item(self, feature: str, item: str) -> bool:
        """Check whether a specific item is allowed for a list-type feature.

        Examples::

            check_feature_item("scanners", "cloud")
            check_feature_item("export_formats", "pdf")
            check_feature_item("notifications", "slack")
        """
        limits = self.get_limits()
        value = limits.get(feature)
        if value is None or value == "all":
            return True
        if isinstance(value, list):
            return item in value
        return True

    def check_scan_depth(self, depth: str) -> bool:
        """Check whether a scan depth (quick / standard / deep) is allowed."""
        allowed = self.get_limits().get("scan_depths", ["quick", "standard"])
        if allowed == "all":
            return True
        if isinstance(allowed, list):
            return depth in allowed
        return False

    def get_scan_throttle_ms(self) -> int:
        """Return the inter-check delay in milliseconds (0 = full speed).

        Community tier enforces a 500ms delay between individual scan
        checks to rate-limit free usage without blocking functionality.
        """
        return self.get_limits().get("scan_throttle_ms", 500)

    def check_targets_count(self, count: int) -> bool:
        """Check whether *count* targets are within the per-scan limit."""
        cap = self.get_limits().get("max_targets_per_scan")
        if cap is None:
            return True  # None = unlimited
        return count <= cap

    # ------------------------------------------------------------------
    # AI quota (persisted to disk)
    # ------------------------------------------------------------------

    def check_ai_quota(self) -> bool:
        """DEPRECATED. AI access is now gated by the ``ai_analysis`` boolean
        feature (Pro+), not a per-day query cap. AI runs on the customer's own
        model/key (local or BYO) and never touches DonjonSec infrastructure, so
        usage is not metered. Retained as a no-op shim (always True) so any
        remaining caller does not block. Use ``check_limit('ai_analysis')``.
        """
        return True

    def increment_ai_usage(self) -> None:
        """DEPRECATED no-op. AI usage is no longer metered (see
        :meth:`check_ai_quota`). Kept so existing callers do not break.
        """
        return None

    def _load_ai_quota(self) -> None:
        """Load the persisted AI quota counter from disk."""
        quota_file = paths.data / "ai_quota.json"
        if not quota_file.exists():
            return

        try:
            data = json.loads(quota_file.read_text("utf-8"))
            if not isinstance(data, dict):
                return
            today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
            self._ai_query_count = data.get(today, 0)
            self._ai_query_date = today if today in data else ""
        except (json.JSONDecodeError, OSError) as exc:
            logger.debug("Could not load AI quota file: %s", exc)

    def _save_ai_quota(self) -> None:
        """Persist the current AI quota counter to disk.

        Only stores today's count.  Previous days are discarded to keep
        the file small and avoid leaking usage history.
        """
        quota_file = paths.data / "ai_quota.json"
        try:
            quota_file.parent.mkdir(parents=True, exist_ok=True)
            quota_file.write_text(
                json.dumps({self._ai_query_date: self._ai_query_count}),
                encoding="utf-8",
            )
        except OSError as exc:
            logger.debug("Could not save AI quota file: %s", exc)

    # ------------------------------------------------------------------
    # Public validation method (for API endpoints)
    # ------------------------------------------------------------------

    def validate_license(self, data: Dict[str, Any]) -> bool:
        """Public validation method for API endpoints (v1 and v2).

        Runs the full validation chain: signature, expiry, fingerprint,
        revocation, and tier validity.  Returns True only if ALL checks
        pass.

        This method is intended for the license upload/activation API
        endpoint, NOT for internal tier checking (use get_tier() for that).
        """
        if not isinstance(data, dict):
            return False

        version = data.get("version", data.get("format_version", 1))
        if version >= 2:
            if not self._validate_license_v2(data):
                return False
        else:
            if not self._validate_license_v1(data):
                return False
        if not self._check_expiry(data):
            return False
        if not self._check_machine_fingerprint(data):
            return False
        if self._is_revoked(data):
            return False
        return data.get("tier", "community") in TIERS

    # ------------------------------------------------------------------
    # Upgrade messages
    # ------------------------------------------------------------------

    def get_upgrade_message(self, feature: str) -> str:
        """Return a user-friendly upgrade prompt for a blocked feature.

        These messages are displayed in the UI when a user attempts to
        use a feature that is not available at their current tier.
        """
        messages: Dict[str, str] = {
            # -- Scanning --
            "scan_depths": (
                "Deep scans are a Pro feature. Community includes quick "
                "and standard depth. Upgrade to unlock deep scanning."
            ),
            "scan_throttle_ms": (
                "Community scans run at reduced speed. "
                "Upgrade to Pro for full-speed scanning."
            ),
            "max_targets_per_scan": (
                "Community is limited to 16 targets per scan. "
                "Upgrade to Pro for unlimited targets."
            ),
            "scan_history_days": (
                "Community retains scan history for 30 days. "
                "Upgrade to Pro for unlimited retention."
            ),
            "scanners": (
                "This scanner requires Pro. Community includes: "
                "network, vulnerability, web, SSL, Windows, Linux, "
                "and compliance. Upgrade for credential, ASM, SBOM, "
                "container, AD, cloud, and OpenVAS scanners."
            ),
            # -- Automation --
            "scheduled_scans": (
                "Scheduled scans are a Pro feature. "
                "Upgrade to automate recurring security assessments."
            ),
            "alert_routing": (
                "Alert routing and digest rules require Pro. "
                "Community sends all alerts to all channels."
            ),
            # -- Reporting --
            "export_formats": (
                "This export format requires Pro. Community "
                "includes CSV and JSON. Upgrade for HTML, PDF, "
                "SARIF, XML, and compliance-specific formats."
            ),
            "compliance_frameworks": (
                "Community supports 3 compliance frameworks. "
                "Upgrade to Pro for unlimited frameworks."
            ),
            # -- Notifications --
            "notifications": (
                "Slack, Teams, and SMS notifications require Pro. "
                "Community includes email and webhook."
            ),
            "sms_notifications": (
                "SMS/text notifications require a Pro license."
            ),
            # -- AI --
            "ai_analysis": (
                "AI analysis is a Pro feature. Community Edition uses built-in "
                "template-based analysis (no LLM required). Upgrade to Pro to "
                "connect your own model: any OpenAI-compatible endpoint, a local "
                "model (Ollama / llama.cpp / vLLM / LM Studio), or your approved "
                "AI vendor. Bring your own key -- usage is never metered by DonjonSec."
            ),
            "ai_queries_per_day": (
                "AI analysis is a Pro feature. Upgrade to Pro to connect your own "
                "model (local or bring-your-own-key) with no usage metering."
            ),
            "ai_scan_summary": (
                "AI scan summaries are a Pro feature. "
                "Upgrade to auto-generate executive reports."
            ),
            # -- Platform --
            "save_projects": (
                "Saving and resuming projects requires Pro. "
                "Community sessions are temporary."
            ),
            "max_users": (
                "You've reached the user limit for your tier. "
                "Upgrade for more user seats."
            ),
            "api_rate_limit": (
                "You've hit the API rate limit for your tier. "
                "Upgrade for higher throughput."
            ),
            # -- Enterprise --
            "sso": (
                "SAML/OIDC single sign-on requires an Enterprise license."
            ),
            "rbac": (
                "Role-based access control requires an Enterprise license."
            ),
            "custom_branding": (
                "Custom / white-label branding requires Enterprise."
            ),
            "multi_tenant": (
                "Multi-tenant isolation requires an Enterprise license."
            ),
            "audit_log_export": (
                "Audit log export (SIEM integration) requires Enterprise."
            ),
            "priority_support": (
                "Priority support is included with Enterprise and Managed tiers."
            ),
            # -- MSSP / MSP (Managed tier) --
            "client_provisioning": (
                "Automated client provisioning is a Managed (MSSP/MSP) feature. "
                "Upgrade to onboard and configure client tenants at scale."
            ),
            "client_dashboards": (
                "Per-client dashboards require the Managed tier. "
                "Upgrade to give each client their own security overview."
            ),
            "consolidated_rollup": (
                "Consolidated roll-up reporting across all clients requires "
                "the Managed tier."
            ),
            "bulk_scan_orchestration": (
                "Bulk scan orchestration across multiple clients is a "
                "Managed tier feature."
            ),
            "cross_client_reporting": (
                "Cross-client comparative reporting requires the Managed tier."
            ),
            "usage_metering_api": (
                "The usage metering API for billing integration requires "
                "the Managed tier."
            ),
            "license_sub_allocation": (
                "Sub-allocating license seats to clients requires "
                "the Managed tier."
            ),
            "per_client_branding": (
                "Per-client branding and white-labeling requires "
                "the Managed tier."
            ),
            "scan_template_library": (
                "The shared scan template library is a Managed tier feature. "
                "Upgrade to create and distribute scan templates across clients."
            ),
            "max_clients": (
                "You've reached the client limit for your tier. "
                "Upgrade to the Managed tier for unlimited client tenants."
            ),
        }

        return messages.get(
            feature,
            f"This feature requires an upgrade. "
            f"Current tier: {self.get_tier_label()}",
        )


# ---------------------------------------------------------------------------
# Module-level singleton accessor
# ---------------------------------------------------------------------------

_license_manager: Optional[LicenseManager] = None


def get_license_manager() -> LicenseManager:
    """Return the module-level LicenseManager singleton.

    Prefer this function over ``LicenseManager.get_instance()`` for
    backward compatibility with existing call sites.
    """
    global _license_manager
    if _license_manager is None:
        _license_manager = LicenseManager()
    return _license_manager
