authforge.py

import base64
import hashlib
import json
import os
import platform
import re
import secrets
import subprocess
import socket
import sys
import threading
import time
import urllib.error
import urllib.request
import uuid
import warnings
from datetime import datetime, timezone
from typing import Any, Callable, Dict, Iterable, List, Literal, Optional, Sequence, TypedDict, Union

from typing_extensions import NotRequired
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey


PublicKeyArg = Union[str, Sequence[str]]


DEFAULT_API_BASE_URL = "https://auth.authforge.cc"
RATE_LIMIT_RETRY_DELAYS = (2, 5)
NETWORK_RETRY_DELAY = 2
class ValidateLicenseSuccess(TypedDict):
    valid: Literal[True]
    session_token: str
    expires_in: int
    session_data: Dict[str, Any]
    app_variables: Optional[Dict[str, Any]]
    license_variables: Optional[Dict[str, Any]]
    key_id: Optional[str]
    session_expires_at: NotRequired[str]
    license_expires_at: NotRequired[Optional[str]]
    max_hwid_slots: NotRequired[int]
    hwid_count: NotRequired[int]
    license_label: NotRequired[str]


class ValidateLicenseFailure(TypedDict):
    valid: Literal[False]
    code: str
    error: str


ValidateLicenseResult = Union[ValidateLicenseSuccess, ValidateLicenseFailure]


# ---------------------------------------------------------------------------
# Offline license files (`.authforge`)
#
# A cloud-minted, Ed25519-signed document for machines that never phone home.
# This is a SEPARATE mode from the grace period: the grace period continues a
# signed session after one online activation, while an offline file is
# verified locally with only the app public key and the machine HWID. Nothing
# here performs network I/O or starts online check-ins.
# ---------------------------------------------------------------------------

OFFLINE_LICENSE_FILE_VERSION = 1
# How a client authenticated: server session (login) or local file (login_from_file).
SessionKind = Literal["online", "offline"]
_OFFLINE_BEGIN_LICENSE = "-----BEGIN AUTHFORGE LICENSE-----"
_OFFLINE_END_LICENSE = "-----END AUTHFORGE LICENSE-----"
_OFFLINE_BEGIN_SIGNATURE = "-----BEGIN AUTHFORGE SIGNATURE-----"
_OFFLINE_END_SIGNATURE = "-----END AUTHFORGE SIGNATURE-----"
_OFFLINE_BASE64_RE = re.compile(r"^[A-Za-z0-9+/]+={0,2}$")

OFFLINE_LICENSE_ERRORS = (
    "bad_armor",
    "bad_signature",
    "unsupported_version",
    "malformed_payload",
    "wrong_app",
    "expired",
    "hwid_mismatch",
)

# Activation requests (`.authforge-request`): unsigned transport for a HWID.
# Distinct markers from BEGIN AUTHFORGE LICENSE. Not signed; the Checksum
# header is the only integrity check. Keep _SDK_TAG in sync with pyproject.toml.
_ACTIVATION_REQUEST_VERSION = 1
_ACTIVATION_REQUEST_TYP = "authforge-activation-request"
_BEGIN_ACTIVATION_REQUEST = "-----BEGIN AUTHFORGE ACTIVATION REQUEST-----"
_END_ACTIVATION_REQUEST = "-----END AUTHFORGE ACTIVATION REQUEST-----"
_SDK_TAG = "python/2.0.0"
_ARMOR_LINE_WIDTH = 64
_MAX_REQUEST_HWID = 256
_MAX_REQUEST_MACHINE_NAME = 128
_MAX_REQUEST_OS = 64
_MAX_REQUEST_SDK = 64
_MAX_REQUEST_LICENSE_KEY = 64


class OfflineLicense(TypedDict):
    app_id: str
    license_key: str
    jti: str
    key_id: str
    issued_at: str
    expires_at: Optional[str]
    hwid_policy: Dict[str, Any]
    label: NotRequired[str]
    license_expires_at: NotRequired[Optional[str]]
    license_variables: Optional[Dict[str, Any]]
    app_variables: Optional[Dict[str, Any]]
    payload: Dict[str, Any]


class VerifyLicenseFileSuccess(TypedDict):
    ok: Literal[True]
    license: OfflineLicense
    payload_base64: str
    signature_base64: str


class VerifyLicenseFileFailure(TypedDict):
    ok: Literal[False]
    error: str


VerifyLicenseFileResult = Union[VerifyLicenseFileSuccess, VerifyLicenseFileFailure]


class ParsedLicenseFile(TypedDict):
    headers: Dict[str, str]
    payload_base64: str
    signature_base64: str


def _clip_request_field(value: str, max_len: int) -> str:
    return value if len(value) <= max_len else value[:max_len]


def _json_escape_request(value: str) -> str:
    out: List[str] = []
    for ch in value:
        code = ord(ch)
        if ch == "\\":
            out.append("\\\\")
        elif ch == '"':
            out.append('\\"')
        elif ch == "\b":
            out.append("\\b")
        elif ch == "\f":
            out.append("\\f")
        elif ch == "\n":
            out.append("\\n")
        elif ch == "\r":
            out.append("\\r")
        elif ch == "\t":
            out.append("\\t")
        elif code < 0x20:
            out.append(f"\\u00{code:02x}")
        else:
            out.append(ch)
    return '"' + "".join(out) + '"'


def _wrap_armor_64(value: str) -> str:
    return "\n".join(value[i : i + _ARMOR_LINE_WIDTH] for i in range(0, len(value), _ARMOR_LINE_WIDTH))


def _canonical_activation_request_json(
    *,
    app_id: str,
    hwid: str,
    created_at: str,
    machine_name: Optional[str] = None,
    os_name: Optional[str] = None,
    sdk: Optional[str] = None,
    license_key: Optional[str] = None,
) -> str:
    parts = [
        f'"v":{_ACTIVATION_REQUEST_VERSION}',
        f'"typ":{_json_escape_request(_ACTIVATION_REQUEST_TYP)}',
        f'"appId":{_json_escape_request(app_id)}',
        f'"hwid":{_json_escape_request(_clip_request_field(hwid, _MAX_REQUEST_HWID))}',
        f'"createdAt":{_json_escape_request(created_at)}',
    ]
    if machine_name:
        parts.append(
            f'"machineName":{_json_escape_request(_clip_request_field(machine_name, _MAX_REQUEST_MACHINE_NAME))}'
        )
    if os_name:
        parts.append(f'"os":{_json_escape_request(_clip_request_field(os_name, _MAX_REQUEST_OS))}')
    if sdk:
        parts.append(f'"sdk":{_json_escape_request(_clip_request_field(sdk, _MAX_REQUEST_SDK))}')
    if license_key:
        parts.append(
            f'"licenseKey":{_json_escape_request(_clip_request_field(license_key, _MAX_REQUEST_LICENSE_KEY))}'
        )
    return "{" + ",".join(parts) + "}"


def _utc_iso_ms(value: Optional[datetime] = None) -> str:
    now = value or datetime.now(timezone.utc)
    if now.tzinfo is None:
        now = now.replace(tzinfo=timezone.utc)
    now = now.astimezone(timezone.utc)
    return now.strftime("%Y-%m-%dT%H:%M:%S.") + f"{int(now.microsecond / 1000):03d}Z"


def _detect_os_label() -> str:
    system = platform.system()
    release = platform.release()
    if system == "Darwin":
        mac = platform.mac_ver()[0]
        label = f"macOS {mac or release}"
    elif system == "Windows":
        label = f"Windows {release}"
    elif system == "Linux":
        label = f"Linux {release}"
    else:
        label = f"{system} {release}".strip()
    return _clip_request_field(label, _MAX_REQUEST_OS)


def format_activation_request(
    *,
    app_id: str,
    hwid: str,
    created_at: str,
    machine_name: Optional[str] = None,
    os: Optional[str] = None,
    sdk: Optional[str] = None,
    license_key: Optional[str] = None,
) -> str:
    """Armored ``.authforge-request`` text from explicit fields."""
    json_body = _canonical_activation_request_json(
        app_id=app_id,
        hwid=hwid,
        created_at=created_at,
        machine_name=machine_name,
        os_name=os,
        sdk=sdk,
        license_key=license_key,
    )
    payload_b64 = base64.b64encode(json_body.encode("utf-8")).decode("ascii")
    checksum = hashlib.sha256(payload_b64.encode("utf-8")).hexdigest()[:16]
    clean_app = app_id.replace("\r", " ").replace("\n", " ").strip()
    return "\n".join(
        [
            _BEGIN_ACTIVATION_REQUEST,
            f"Version: {_ACTIVATION_REQUEST_VERSION}",
            f"App-Id: {clean_app}",
            f"Checksum: {checksum}",
            "",
            _wrap_armor_64(payload_b64),
            _END_ACTIVATION_REQUEST,
            "",
        ]
    )


def parse_license_file(text: str) -> Optional[ParsedLicenseFile]:
    """Parse armored ``.authforge`` text.

    Returns ``None`` when the armor is malformed. Tolerates CRLF, a UTF-8 BOM,
    any re-wrapping of the base64 body and text before/after the armor.
    ``payload_base64`` is exactly the string the signature covers.
    """
    if not isinstance(text, str):
        return None
    normalized = text.lstrip("\ufeff").replace("\r\n", "\n").replace("\r", "\n")
    lines = normalized.split("\n")

    def _find(marker: str, start: int) -> int:
        for i in range(start, len(lines)):
            if lines[i].strip() == marker:
                return i
        return -1

    begin_idx = _find(_OFFLINE_BEGIN_LICENSE, 0)
    if begin_idx == -1:
        return None
    end_idx = _find(_OFFLINE_END_LICENSE, begin_idx + 1)
    if end_idx == -1:
        return None
    sig_begin_idx = _find(_OFFLINE_BEGIN_SIGNATURE, end_idx + 1)
    if sig_begin_idx == -1:
        return None
    sig_end_idx = _find(_OFFLINE_END_SIGNATURE, sig_begin_idx + 1)
    if sig_end_idx == -1:
        return None

    block = lines[begin_idx + 1 : end_idx]
    blank_idx = -1
    for i, line in enumerate(block):
        if line.strip() == "":
            blank_idx = i
            break
    if blank_idx == -1:
        return None

    headers: Dict[str, str] = {}
    for raw in block[:blank_idx]:
        line = raw.strip()
        colon = line.find(":")
        if colon <= 0:
            return None
        headers[line[:colon].strip()] = line[colon + 1 :].strip()

    payload_base64 = re.sub(r"\s+", "", "".join(block[blank_idx + 1 :]))
    signature_base64 = re.sub(r"\s+", "", "".join(lines[sig_begin_idx + 1 : sig_end_idx]))
    if not payload_base64 or not _OFFLINE_BASE64_RE.match(payload_base64):
        return None
    if not signature_base64 or not _OFFLINE_BASE64_RE.match(signature_base64):
        return None
    return {
        "headers": headers,
        "payload_base64": payload_base64,
        "signature_base64": signature_base64,
    }


def _parse_iso8601_ms(value: str) -> Optional[float]:
    """ISO-8601 -> epoch milliseconds (UTC). Returns None when unparseable."""
    text = value.strip()
    if text.endswith("Z") or text.endswith("z"):
        text = text[:-1] + "+00:00"
    try:
        parsed = datetime.fromisoformat(text)
    except ValueError:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed.timestamp() * 1000.0


def _offline_payload_shape_ok(payload: Dict[str, Any]) -> bool:
    if payload.get("typ") != "authforge-license":
        return False
    for field in ("appId", "licenseKey", "jti", "kid", "issuedAt"):
        value = payload.get(field)
        if not isinstance(value, str) or not value:
            return False
    # `expiresAt` must be present: an explicit null means lifetime, absence is
    # a malformed document (never emitted by the cloud minter).
    if "expiresAt" not in payload:
        return False
    expires_at = payload["expiresAt"]
    if expires_at is not None and (not isinstance(expires_at, str) or not expires_at):
        return False
    hwid = payload.get("hwid")
    if not isinstance(hwid, dict):
        return False
    mode = hwid.get("mode")
    if mode == "bound":
        hwids = hwid.get("hwids")
        if not isinstance(hwids, list) or not hwids:
            return False
        if not all(isinstance(h, str) and h for h in hwids):
            return False
    elif mode != "any":
        return False
    return True


def verify_license_file(
    file: str,
    app_id: str,
    public_key: PublicKeyArg,
    hwid: Optional[str] = None,
    *,
    now: Optional[Union[datetime, float, int]] = None,
) -> VerifyLicenseFileResult:
    """Verify an offline ``.authforge`` license file with NO network access.

    Check order (fixed across every SDK): ``bad_armor`` -> ``bad_signature`` ->
    ``unsupported_version`` -> ``malformed_payload`` -> ``wrong_app`` ->
    ``expired`` -> ``hwid_mismatch``. The signature is verified before the
    payload JSON is decoded so a forged file never reaches the parser.

    ``file`` is the armored text. ``public_key`` accepts the same forms as the
    client constructor (single key, list, or comma-separated). ``now`` may be a
    ``datetime`` or epoch seconds (tests).
    """
    parsed = parse_license_file(file)
    if parsed is None:
        return {"ok": False, "error": "bad_armor"}

    keys = AuthForgeClient._normalize_public_key_list(public_key)
    if not _verify_offline_signature(parsed["payload_base64"], parsed["signature_base64"], keys):
        return {"ok": False, "error": "bad_signature"}

    try:
        payload = json.loads(base64.b64decode(parsed["payload_base64"]).decode("utf-8"))
    except Exception:
        return {"ok": False, "error": "malformed_payload"}
    if not isinstance(payload, dict):
        return {"ok": False, "error": "malformed_payload"}
    version = payload.get("v")
    # ``True == 1`` in Python, so the bool check is required for parity with the
    # other SDKs, which all demand a JSON number here.
    if isinstance(version, bool) or version != OFFLINE_LICENSE_FILE_VERSION:
        return {"ok": False, "error": "unsupported_version"}
    if not _offline_payload_shape_ok(payload):
        return {"ok": False, "error": "malformed_payload"}

    if payload["appId"] != app_id:
        return {"ok": False, "error": "wrong_app"}

    if isinstance(now, datetime):
        now_ms = now.timestamp() * 1000.0
    elif isinstance(now, (int, float)):
        now_ms = float(now) * 1000.0
    else:
        now_ms = time.time() * 1000.0
    expires_at = payload["expiresAt"]
    if expires_at is not None:
        exp_ms = _parse_iso8601_ms(expires_at)
        if exp_ms is None or exp_ms <= now_ms:
            return {"ok": False, "error": "expired"}

    hwid_policy = payload["hwid"]
    if hwid_policy["mode"] == "bound":
        local = hwid.strip() if isinstance(hwid, str) else ""
        if not local or local not in hwid_policy["hwids"]:
            return {"ok": False, "error": "hwid_mismatch"}

    license_info: OfflineLicense = {
        "app_id": payload["appId"],
        "license_key": payload["licenseKey"],
        "jti": payload["jti"],
        "key_id": payload["kid"],
        "issued_at": payload["issuedAt"],
        "expires_at": expires_at,
        "hwid_policy": (
            {"mode": "bound", "hwids": list(hwid_policy["hwids"])}
            if hwid_policy["mode"] == "bound"
            else {"mode": "any"}
        ),
        "license_variables": dict(payload["licenseVariables"])
        if isinstance(payload.get("licenseVariables"), dict)
        else None,
        "app_variables": dict(payload["appVariables"])
        if isinstance(payload.get("appVariables"), dict)
        else None,
        "payload": dict(payload),
    }
    if isinstance(payload.get("label"), str):
        license_info["label"] = payload["label"]
    if "licenseExpiresAt" in payload:
        le = payload["licenseExpiresAt"]
        license_info["license_expires_at"] = le if isinstance(le, str) else None
    return {
        "ok": True,
        "license": license_info,
        "payload_base64": parsed["payload_base64"],
        "signature_base64": parsed["signature_base64"],
    }


def _verify_offline_signature(payload_base64: str, signature_base64: str, keys: Sequence[str]) -> bool:
    try:
        signature_bytes = base64.b64decode(signature_base64, validate=True)
    except Exception:
        return False
    if len(signature_bytes) != 64:
        return False
    message = payload_base64.encode("utf-8")
    for key_b64 in keys:
        try:
            raw = base64.b64decode(key_b64, validate=True)
            if len(raw) != 32:
                continue
            Ed25519PublicKey.from_public_bytes(raw).verify(signature_bytes, message)
            return True
        except Exception:
            continue
    return False


def _write_file_atomic(path: str, data: bytes) -> None:
    """Write ``data`` byte-for-byte to ``path`` via a temp file in the same
    directory and ``os.replace``, creating parent directories. A failure never
    leaves a partial file at ``path`` or changes an existing one."""
    directory = os.path.dirname(os.path.abspath(path))
    os.makedirs(directory, exist_ok=True)
    tmp_path = os.path.join(
        directory, f".{os.path.basename(path)}.{secrets.token_hex(8)}.tmp"
    )
    # O_BINARY (Windows only) keeps the bytes exact; mode 0o666 lets the umask
    # apply as for a normal open().
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0)
    fd = os.open(tmp_path, flags, 0o666)
    try:
        with os.fdopen(fd, "wb") as handle:
            handle.write(data)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(tmp_path, path)
    except BaseException:
        try:
            os.remove(tmp_path)
        except OSError:
            pass
        raise


KNOWN_SERVER_ERRORS = {
    "invalid_app",
    "invalid_key",
    "expired",
    "revoked",
    "hwid_mismatch",
    "no_credits",
    "app_burn_cap_reached",
    "blocked",
    "rate_limited",
    "replay_detected",
    "app_disabled",
    "session_expired",
    "revoke_requires_session",
    "bad_request",
    "malformed_request",
    "demo_quota_exceeded",
    "system_error",
    # POST /auth/offline/activate (self-serve offline activation).
    "offline_activation_disabled",
    "offline_activation_requires_perpetual",
    "offline_activation_requires_seats",
    "offline_activation_limit_reached",
    # SDK code (never sent by the server): the returned file failed local
    # verification. Listed with the server codes for cross-SDK parity.
    "offline_file_rejected",
}

# Well-formed AuthForge verdicts that end the session. This is the only list
# the classification uses: every other code is transient.
DEFINITIVE_ERROR_CODES = frozenset(
    {
        "revoked",
        "expired",
        "hwid_mismatch",
        "blocked",
        "session_expired",
        "malformed_request",
        "app_disabled",
        "invalid_app",
        "signature_mismatch",
        "offline_activation_disabled",
        "offline_activation_requires_perpetual",
        "offline_activation_requires_seats",
        "offline_activation_limit_reached",
        # SDK code: activate_offline() got a file that failed local verification.
        "offline_file_rejected",
    }
)

# Reference list of the named transient codes. Informational only: any code
# outside DEFINITIVE_ERROR_CODES is transient, including every
# http_error_<status>, unexpected_response and codes the server adds later.
TRANSIENT_ERROR_CODES = frozenset(
    {
        "network_error",
        "timeout",
        "rate_limited",
        "system_error",
        "handler_error",
        "invalid_json_response",
        "response_not_json_object",
        "unexpected_response",
        "no_credits",
        "demo_quota_exceeded",
        "app_burn_cap_reached",
        "bad_request",
        "invalid_key",
        "replay_detected",
        "revoke_requires_session",
        "missing_session_token",
        "nonce_mismatch",
        "unknown_error",
        # activate_offline() could not write the verified file; activating the
        # same machine again is free.
        "file_write_failed",
    }
)
_SERVER_ERROR_CODE_RE = re.compile(r"[a-z][a-z0-9_]{0,63}")


class AuthForgeError(ValueError):
    """A failure with a machine-readable ``code``: the server's error code
    (``revoked``, ``hwid_mismatch``, ...) or an SDK code (``network_error``,
    ``timeout``, ``http_error_502``, ...). ``str(exc)`` is the human message,
    which equals ``code`` for server errors."""

    def __init__(self, code: str, message: Optional[str] = None) -> None:
        super().__init__(message if message is not None else code)
        self.code = code

    @property
    def transient(self) -> bool:
        """True when retrying later can succeed (see :func:`is_transient_error`)."""
        return is_transient_error(self.code)

    @property
    def fatal(self) -> bool:
        """True when AuthForge definitively rejected the session or license."""
        return not self.transient


def is_transient_error(error: Union[str, BaseException, None]) -> bool:
    """Classify a failure code (or an exception from ``on_failure``).

    Definitive (fatal) only for the codes in :data:`DEFINITIVE_ERROR_CODES`.
    Everything else is transient: network failures, ``rate_limited``,
    ``no_credits``, ``http_error_<status>``, ``unexpected_response``, unknown
    codes, ... ``None`` and exceptions other than :class:`AuthForgeError`
    carry no AuthForge verdict and are transient too.
    """
    code = error.code if isinstance(error, AuthForgeError) else error
    if not isinstance(code, str):
        return True
    return code not in DEFINITIVE_ERROR_CODES


# HWID v2: sha256("authforge-hwid-v2|" + app_id + "|" + machine_id), where machine_id is
# the OS machine id (macOS IOPlatformUUID, Linux /etc/machine-id, Windows MachineGuid).
# Every SDK derives the same value; conformance is pinned by hwid_vectors.json.
HWID_V2_PREFIX = "authforge-hwid-v2|"
_MACHINE_ID_COMMAND_TIMEOUT = 2
_LINUX_MACHINE_ID_PATHS = ("/etc/machine-id", "/var/lib/dbus/machine-id")
_MACHINE_ID_RE = re.compile(r"[0-9a-f-]+")
_ALL_ZERO_MACHINE_ID_RE = re.compile(r"[0-]+")
_IOREG_PLATFORM_UUID_RE = re.compile(r'"IOPlatformUUID"\s*=\s*"([^"]+)"')
_ASCII_WS_SPLIT_RE = re.compile(r"[ \t\r\n]+")


def hwid_from_machine_id(app_id: str, machine_id: str) -> str:
    """Derive the HWID from an already-normalized machine id (or a ``legacy:`` fallback id)."""
    material = f"{HWID_V2_PREFIX}{app_id}|{machine_id}"
    return hashlib.sha256(material.encode("utf-8")).hexdigest()


def normalize_machine_id(raw: Optional[str]) -> Optional[str]:
    """Trim and lowercase an OS machine id; ``None`` when it is not a usable id."""
    if not isinstance(raw, str):
        return None
    value = raw.strip(" \t\r\n")
    # ASCII-only lowercasing; the character check below rejects anything non-ASCII.
    value = "".join(chr(ord(c) + 32) if "A" <= c <= "Z" else c for c in value)
    if len(value) < 8 or len(value) > 128:
        return None
    if not _MACHINE_ID_RE.fullmatch(value):
        return None
    if _ALL_ZERO_MACHINE_ID_RE.fullmatch(value):
        return None
    return value


def parse_ioreg_platform_uuid(output: Optional[str]) -> Optional[str]:
    """Extract IOPlatformUUID from ``ioreg -rd1 -c IOPlatformExpertDevice`` output."""
    if not isinstance(output, str):
        return None
    match = _IOREG_PLATFORM_UUID_RE.search(output)
    return match.group(1) if match else None


def parse_reg_query_machine_guid(output: Optional[str]) -> Optional[str]:
    """Extract MachineGuid from ``reg.exe query ... /v MachineGuid`` output."""
    if not isinstance(output, str):
        return None
    for line in re.split(r"\r?\n", output):
        tokens = [t for t in _ASCII_WS_SPLIT_RE.split(line) if t]
        if len(tokens) >= 3 and tokens[0].lower() == "machineguid" and tokens[1] == "REG_SZ":
            return tokens[2]
    return None


def _read_os_machine_id() -> Optional[str]:
    """The normalized OS machine id, or ``None``. Never raises."""
    try:
        if sys.platform == "darwin":
            result = subprocess.run(
                ["/usr/sbin/ioreg", "-rd1", "-c", "IOPlatformExpertDevice"],
                capture_output=True,
                text=True,
                timeout=_MACHINE_ID_COMMAND_TIMEOUT,
            )
            if result.returncode != 0:
                return None
            return normalize_machine_id(parse_ioreg_platform_uuid(result.stdout))
        if sys.platform.startswith("linux"):
            for path in _LINUX_MACHINE_ID_PATHS:
                try:
                    with open(path, "r", encoding="utf-8", errors="replace") as handle:
                        machine_id = normalize_machine_id(handle.read())
                except Exception:
                    continue  # missing or unreadable: try the next source
                if machine_id:
                    return machine_id
            return None
        if sys.platform == "win32":
            import winreg

            with winreg.OpenKey(
                winreg.HKEY_LOCAL_MACHINE,
                r"SOFTWARE\Microsoft\Cryptography",
                0,
                winreg.KEY_READ | winreg.KEY_WOW64_64KEY,
            ) as key:
                value, value_type = winreg.QueryValueEx(key, "MachineGuid")
            if value_type != winreg.REG_SZ or not isinstance(value, str):
                return None
            return normalize_machine_id(value)
    except Exception:
        pass  # fall through to the legacy fingerprint
    return None


def _warn(message: str) -> None:
    # sys.stderr is None under pythonw, and a closed stream must not stop check-ins.
    stream = sys.stderr
    if stream is None:
        return
    try:
        print(message, file=stream, flush=True)
    except (OSError, ValueError):
        pass


class AuthForgeClient:
    def __init__(
        self,
        app_id: str,
        app_secret: Optional[str],
        public_key: PublicKeyArg,
        heartbeat_mode: Optional[str] = None,
        heartbeat_interval: int = 900,
        api_base_url: str = DEFAULT_API_BASE_URL,
        on_failure: Optional[Callable[[str, Optional[Exception]], None]] = None,
        request_timeout: int = 15,
        ttl_seconds: Optional[int] = None,
        hwid_override: Optional[str] = None,
        *,
        online_heartbeat: bool = False,
    ) -> None:
        if not app_id or not isinstance(app_id, str):
            raise ValueError("app_id must be a non-empty string")
        # Empty/None is valid for offline-only clients (login_from_file).
        # Online APIs (login, validate_license, self_ban) still require a secret.
        if app_secret is None:
            app_secret = ""
        elif not isinstance(app_secret, str):
            raise ValueError("app_secret must be a string or None")
        public_key_list = self._normalize_public_key_list(public_key)
        if not public_key_list:
            raise ValueError(
                "public_key must be a non-empty base64 string or list of base64 strings"
            )
        # heartbeat_mode is a deprecated shim. By default the client runs on
        # the grace period: after a successful activate/validate, the signed
        # session keeps the app running without contacting AuthForge until
        # the session TTL expires. Online check-ins (periodic
        # /auth/heartbeat calls) are opt-in via online_heartbeat=True.
        mode: Optional[str] = None
        if heartbeat_mode:
            mode = heartbeat_mode.upper()
            if mode not in {"LOCAL", "SERVER"}:
                raise ValueError("heartbeat_mode must be LOCAL or SERVER")
            warnings.warn(
                "heartbeat_mode is deprecated: use online_heartbeat=True for "
                "online check-ins; the default is the grace period behavior",
                DeprecationWarning,
                stacklevel=2,
            )
        if heartbeat_interval < 10:
            raise ValueError("heartbeat_interval must be >= 10")

        self.app_id = app_id
        self.app_secret = app_secret
        # `public_key` is the historical public attribute. We now hold the
        # full trust list to support key rotation, but expose the first entry
        # as `public_key` for callers that read it directly.
        self.public_keys: List[str] = public_key_list
        self.public_key = public_key_list[0]
        # Effective policy: online check-ins are enabled either explicitly or
        # via the legacy heartbeat_mode="SERVER" shim.
        self.online_heartbeat: bool = bool(online_heartbeat) or mode == "SERVER"
        # Back-compat attribute for callers that still read heartbeat_mode.
        self.heartbeat_mode = "SERVER" if self.online_heartbeat else "LOCAL"
        self.heartbeat_interval = int(heartbeat_interval)
        self.api_base_url = api_base_url.rstrip("/")
        self.on_failure = on_failure
        self.request_timeout = request_timeout
        # ttl_seconds is the grace period duration knob: how long the app
        # keeps running on the signed session without contacting AuthForge.
        # None / 0 / negative means "let the server pick its default (24h)".
        # The server clamps requested values to [3600, 604800] (1h to 7d);
        # we don't duplicate the clamp here.
        self.ttl_seconds: Optional[int] = (
            int(ttl_seconds) if isinstance(ttl_seconds, int) and ttl_seconds > 0 else None
        )

        self._lock = threading.Lock()
        self._heartbeat_thread: Optional[threading.Thread] = None
        self._heartbeat_started = False
        self._heartbeat_stop = threading.Event()
        # Bumped whenever the stored session is replaced or cleared, so a
        # heartbeat that started before logout()/login() can't write back.
        self._session_generation = 0

        self._license_key: Optional[str] = None
        self._session_token: Optional[str] = None
        # "online" after login()/validate, "offline" after login_from_file(),
        # None when logged out. Drives is_authenticated(), self_ban() and the
        # heartbeat guard so the two modes can never be confused.
        self._session_kind: Optional[SessionKind] = None
        self._session_expires_in: Optional[int] = None
        self._last_nonce: Optional[str] = None
        self._raw_payload_b64: Optional[str] = None
        self._signature: Optional[str] = None
        self._key_id: Optional[str] = None
        self._session_data: Optional[Dict[str, Any]] = None
        self._app_variables: Optional[Dict[str, Any]] = None
        self._license_variables: Optional[Dict[str, Any]] = None
        self._authenticated = False
        self._offline_license: Optional[Dict[str, Any]] = None
        self._hwid = self._resolve_hwid(hwid_override)
        self._ed25519_public_keys: List[Ed25519PublicKey] = [
            self._load_public_key(k) for k in public_key_list
        ]

    def get_hwid(self) -> str:
        """The HWID this client sends to AuthForge (or ``hwid_override``).

        Customers on air-gapped machines report this value to the operator so
        an offline ``.authforge`` file can be bound to it.
        """
        return self._hwid

    def create_activation_request(
        self,
        *,
        include_machine_name: bool = False,
        machine_name: Optional[str] = None,
        os: Optional[str] = None,
        omit_os: bool = False,
        sdk: Optional[str] = None,
        omit_sdk: bool = False,
        license_key: Optional[str] = None,
        created_at: Optional[str] = None,
    ) -> str:
        """Build an activation request (``.authforge-request``) for this machine.

        No network, no session, no app secret. The HWID is the same value
        :meth:`login` / :meth:`login_from_file` use. ``machine_name`` is omitted
        unless ``include_machine_name`` is true (hostnames are often a person's
        name).
        """
        created = created_at if created_at else _utc_iso_ms()
        name: Optional[str] = None
        if include_machine_name:
            name = machine_name or platform.node() or socket.gethostname()
        os_name: Optional[str] = None if omit_os else (os if os is not None else _detect_os_label())
        sdk_tag: Optional[str] = None if omit_sdk else (sdk if sdk is not None else _SDK_TAG)
        key = license_key if license_key is not None else self._license_key
        return format_activation_request(
            app_id=self.app_id,
            hwid=self._hwid,
            created_at=created,
            machine_name=name,
            os=os_name,
            sdk=sdk_tag,
            license_key=key,
        )

    def write_activation_request(self, path: str, **kwargs: Any) -> None:
        """Write an activation request to ``path`` (UTF-8). Same kwargs as
        :meth:`create_activation_request`."""
        text = self.create_activation_request(**kwargs)
        with open(path, "w", encoding="utf-8", newline="\n") as handle:
            handle.write(text)

    def login_from_file(self, path_or_text: str) -> bool:
        """Authorize from a cloud-minted offline license file (``.authforge``)
        with NO network access. Accepts a filesystem path or the armored text.

        On success the client is authenticated (:meth:`is_authenticated`,
        :meth:`get_session_data`, :meth:`get_app_variables`,
        :meth:`get_license_variables` work) and :meth:`get_offline_license`
        describes the file. No grace-period thread and no online check-ins are
        started - the file's own ``expiresAt`` is the only clock. Online
        :meth:`login` is untouched.

        Failures are reported through ``on_failure("offline_login_failed", exc)``
        and return ``False``; unlike :meth:`login` this never calls
        ``os._exit`` - an unreadable file should not kill an air-gapped
        process without a chance to show the user why.
        """
        try:
            text = self._read_license_file_input(path_or_text)
        except Exception as exc:
            self._fail_soft("offline_login_failed", exc)
            return False
        result = verify_license_file(text, self.app_id, self.public_keys, self._hwid)
        if not result["ok"]:
            self._fail_soft("offline_login_failed", ValueError(result["error"]))
            return False
        self._apply_offline_license(result)
        return True

    def activate_offline(self, license_key: str, path: Optional[str] = None) -> bool:
        """Self-serve offline activation: turn a license key into a lifetime,
        HWID-bound ``.authforge`` file with one online call
        (``POST /auth/offline/activate``).

        Needs the App Secret (unlike air-gapped builds). The server charges the
        app owner 2 credits for a new machine; the same machine activating
        again gets the identical file back for free. The returned file is
        verified locally (signature, app id, this client's HWID) before
        anything is written. When ``path`` is set the file is written there
        atomically, creating parent directories. On success the client is
        authenticated exactly as after :meth:`login_from_file`, and
        ``get_offline_license()["replay"]`` tells whether the server returned
        a file this machine already had.

        Raises :class:`AuthForgeError` on any refusal: ``exc.code`` is the
        server code, ``offline_file_rejected`` (the message carries the
        verifier's reason), ``file_write_failed`` or a network code. Never
        calls ``on_failure`` or ``os._exit``. On failure nothing is written to
        ``path`` and the client is not authenticated by this call; an existing
        session is kept. On success the session is replaced, as with
        :meth:`login_from_file`.
        """
        if not license_key or not isinstance(license_key, str):
            raise ValueError("license_key must be a non-empty string")
        self._require_app_secret()
        target = os.fspath(path) if path else None
        body: Dict[str, Any] = {
            "appId": self.app_id,
            "appSecret": self.app_secret,
            "licenseKey": license_key,
            "hwid": self._hwid,
            "nonce": self._generate_nonce(),
        }
        try:
            # A network failure must raise here, never reach on_failure / os._exit.
            response_obj = self._post_json("/auth/offline/activate", body)
        except AuthForgeError:
            raise
        except Exception as exc:
            raise AuthForgeError("network_error", f"network_error: {exc}") from exc
        if not self._is_success_status(response_obj.get("status")):
            raise self._heartbeat_verdict(response_obj)
        file_text = response_obj.get("file")
        if not isinstance(file_text, str) or not file_text:
            raise AuthForgeError("unexpected_response", "unexpected_response: missing file")

        # The verifier's reason goes in the message only: the server's
        # hwid_mismatch means "no free seat" and must stay distinguishable.
        result = verify_license_file(file_text, self.app_id, self.public_keys, self._hwid)
        if not result["ok"]:
            raise AuthForgeError(
                "offline_file_rejected", f"offline_file_rejected: {result['error']}"
            )

        if target is not None:
            try:
                _write_file_atomic(target, file_text.encode("utf-8"))
            except OSError as exc:
                raise AuthForgeError("file_write_failed", f"file_write_failed: {exc}") from exc

        self._apply_offline_license(result, replay=response_obj.get("replay") is True)
        return True

    def verify_license_file(
        self,
        path_or_text: str,
        *,
        now: Optional[Union[datetime, float, int]] = None,
    ) -> VerifyLicenseFileResult:
        """Verify a ``.authforge`` file with this client's app id, public
        key(s) and HWID without touching session state. Never raises for bad
        input."""
        try:
            text = self._read_license_file_input(path_or_text)
        except Exception as exc:
            return {"ok": False, "error": f"read_error: {exc}"}
        return verify_license_file(text, self.app_id, self.public_keys, self._hwid, now=now)

    def get_offline_license(self) -> Optional[Dict[str, Any]]:
        """Details of the offline file the client authenticated with, or ``None``."""
        with self._lock:
            return dict(self._offline_license) if self._offline_license is not None else None

    @staticmethod
    def _read_license_file_input(path_or_text: str) -> str:
        if not isinstance(path_or_text, str) or not path_or_text:
            raise ValueError("license file must be a path or the armored text")
        if _OFFLINE_BEGIN_LICENSE in path_or_text:
            return path_or_text
        with open(path_or_text, "r", encoding="utf-8") as handle:
            return handle.read()

    def _apply_offline_license(
        self, result: VerifyLicenseFileSuccess, *, replay: Optional[bool] = None
    ) -> None:
        # Stop any online session first so the two modes never overlap.
        self.logout()
        lic = result["license"]
        expires_at = lic["expires_at"]
        expires_ms = _parse_iso8601_ms(expires_at) if isinstance(expires_at, str) else None
        with self._lock:
            self._license_key = lic["license_key"]
            # Offline files carry no server session token. The explicit session
            # kind (not a token sentinel) is what makes is_authenticated() true
            # and keeps self_ban()/heartbeats from ever contacting the server.
            self._session_token = None
            self._session_kind = "offline"
            self._session_expires_in = int(expires_ms / 1000) if expires_ms is not None else None
            self._raw_payload_b64 = result["payload_base64"]
            self._signature = result["signature_base64"]
            self._key_id = lic["key_id"]
            self._session_data = dict(lic["payload"])
            self._app_variables = lic["app_variables"]
            self._license_variables = lic["license_variables"]
            summary: Dict[str, Any] = {
                "license_key": lic["license_key"],
                "jti": lic["jti"],
                "key_id": lic["key_id"],
                "issued_at": lic["issued_at"],
                "expires_at": expires_at,
                "hwid_policy": lic["hwid_policy"],
            }
            if "label" in lic:
                summary["label"] = lic["label"]
            if "license_expires_at" in lic:
                summary["license_expires_at"] = lic["license_expires_at"]
            if replay is not None:
                # Only set by activate_offline(): True when the server returned
                # the file this machine already had (free).
                summary["replay"] = replay
            self._offline_license = summary
            self._authenticated = True

    def _fail_soft(self, reason: str, exc: Optional[Exception]) -> None:
        if self.on_failure is not None:
            try:
                self.on_failure(reason, exc)
            except Exception:
                pass

    def _require_app_secret(self) -> None:
        if not self.app_secret:
            raise ValueError(
                "app_secret is required for online APIs; omit it only when using login_from_file"
            )

    def login(self, license_key: str) -> bool:
        if not license_key or not isinstance(license_key, str):
            raise ValueError("license_key must be a non-empty string")
        self._require_app_secret()

        try:
            self._validate_and_store(license_key)
            self._start_heartbeat_once()
            return True
        except Exception as exc:
            self._fail("login_failed", exc)
            return False

    def validate_license(self, license_key: str) -> ValidateLicenseResult:
        """Validate like :meth:`login` (same /auth/validate + signatures) without storing
        session state or starting the heartbeat thread."""
        if not license_key or not isinstance(license_key, str):
            raise ValueError("license_key must be a non-empty string")
        self._require_app_secret()
        try:
            body: Dict[str, Any] = {
                "appId": self.app_id,
                "appSecret": self.app_secret,
                "licenseKey": license_key,
                "hwid": self._hwid,
                "nonce": self._generate_nonce(),
            }
            if self.ttl_seconds is not None:
                body["ttlSeconds"] = self.ttl_seconds
            response_obj = self._post_json("/auth/validate", body)
            expected_nonce = str(body.get("nonce", "")).strip()
            parsed = self._parse_validate_success(response_obj, expected_nonce)
            result: ValidateLicenseSuccess = {
                "valid": True,
                "session_token": parsed["session_token"],
                "expires_in": parsed["expires_in"],
                "session_data": parsed["session_data"],
                "app_variables": parsed["app_variables"],
                "license_variables": parsed["license_variables"],
                "key_id": parsed["key_id"],
            }
            if "session_expires_at" in parsed:
                result["session_expires_at"] = parsed["session_expires_at"]
            if "license_expires_at" in parsed:
                result["license_expires_at"] = parsed["license_expires_at"]
            if "max_hwid_slots" in parsed:
                result["max_hwid_slots"] = parsed["max_hwid_slots"]
            if "hwid_count" in parsed:
                result["hwid_count"] = parsed["hwid_count"]
            if "license_label" in parsed:
                result["license_label"] = parsed["license_label"]
            return result
        except Exception as exc:
            return {"valid": False, "code": str(exc), "error": str(exc)}

    def self_ban(
        self,
        *,
        license_key: Optional[str] = None,
        session_token: Optional[str] = None,
        revoke_license: bool = True,
        blacklist_hwid: bool = True,
        blacklist_ip: bool = True,
    ) -> Dict[str, Any]:
        resolved_session = (
            session_token.strip()
            if isinstance(session_token, str) and session_token.strip()
            else None
        )
        explicit_license = (
            license_key.strip()
            if isinstance(license_key, str) and license_key.strip()
            else None
        )
        with self._lock:
            current_session = self._session_token
            current_license = self._license_key
            current_kind = self._session_kind
            hwid = self._hwid

        # An offline session has no server session and must never phone home
        # on its own. Callers who pass an explicit license_key/session_token
        # are asking about a *different* credential and get the normal paths.
        if current_kind == "offline" and resolved_session is None and explicit_license is None:
            raise ValueError("offline_session")

        resolved_session = resolved_session or current_session

        # Every failure raises, a network failure as AuthForgeError
        # network_error / timeout; self_ban never calls on_failure / os._exit.
        if resolved_session:
            body: Dict[str, Any] = {
                "appId": self.app_id,
                "sessionToken": resolved_session,
                "hwid": hwid,
                "revokeLicense": bool(revoke_license),
                "blacklistHwid": bool(blacklist_hwid),
                "blacklistIp": bool(blacklist_ip),
            }
            response_obj = self._post_json("/auth/selfban", body)
            if not self._is_success_status(response_obj.get("status")):
                raise ValueError(self._extract_server_error(response_obj))
            return response_obj

        resolved_license = explicit_license or current_license
        if not resolved_license:
            raise ValueError("missing_license_key")
        self._require_app_secret()

        body = {
            "appId": self.app_id,
            "appSecret": self.app_secret,
            "licenseKey": resolved_license,
            "hwid": hwid,
            "nonce": self._generate_nonce(),
            # Pre-session self-ban cannot revoke licenses.
            "revokeLicense": False,
            "blacklistHwid": bool(blacklist_hwid),
            "blacklistIp": bool(blacklist_ip),
        }
        response_obj = self._post_json("/auth/selfban", body)
        if not self._is_success_status(response_obj.get("status")):
            raise ValueError(self._extract_server_error(response_obj))
        return response_obj

    def _start_heartbeat_once(self) -> None:
        with self._lock:
            # Offline sessions have no grace period and no online check-ins:
            # the file's own expires_at is the only clock. Never start a thread.
            if self._heartbeat_started or self._session_kind == "offline":
                return
            # A fresh event per thread: logout() stops this loop even if a
            # later login() starts another one before this one wakes up.
            stop = threading.Event()
            self._heartbeat_stop = stop
            self._heartbeat_started = True
            self._heartbeat_thread = threading.Thread(
                target=self._heartbeat_loop,
                args=(stop,),
                name="AuthForgeHeartbeat",
                daemon=True,
            )
            self._heartbeat_thread.start()

    def _heartbeat_loop(self, stop: Optional[threading.Event] = None) -> None:
        stop = stop if stop is not None else self._heartbeat_stop
        while not stop.wait(self.heartbeat_interval):
            if not self._heartbeat_tick():
                break

    def _heartbeat_tick(self) -> bool:
        """Run one background check. Returns ``False`` when the loop must stop.

        Transient failures are reported (a stderr warning when there is no
        ``on_failure``) and the loop keeps checking in on the next interval.
        Definitive failures drop the stored session first, so
        neither the grace period nor ``is_authenticated()`` keeps the app
        running on it, and ``on_failure`` may call ``login()`` again.
        ``on_failure`` always runs with no SDK lock held.
        """
        with self._lock:
            offline = self._session_kind == "offline"
            generation = self._session_generation
        if offline:
            return False
        try:
            if self.online_heartbeat:
                self._server_heartbeat(generation)
            else:
                self._grace_period_check()
            return True
        except Exception as exc:
            error = self._heartbeat_error(exc)
        with self._lock:
            # The session this check ran for was logged out or replaced
            # meanwhile: its failure is moot and must not touch the new state.
            if generation != self._session_generation:
                return True
            if error.fatal:
                self._clear_session_locked()
        if error.transient and self.on_failure is None:
            _warn(f"AuthForge: background check failed ({error.code}); retrying next interval")
            return True
        self._fail("heartbeat_failed", error)
        return error.transient

    def _heartbeat_error(self, exc: Exception) -> AuthForgeError:
        if isinstance(exc, AuthForgeError):
            error = exc
        else:
            message = str(exc)
            code = message.split(":", 1)[0].strip().lower()
            if not _SERVER_ERROR_CODE_RE.fullmatch(code):
                code = "unknown_error"
            error = AuthForgeError(code, message)
            error.__cause__ = exc
        # A transient failure can't extend the session past its signed TTL.
        if error.transient and self._local_session_expired():
            expired = AuthForgeError("session_expired")
            expired.__cause__ = error
            return expired
        return error

    def _local_session_expired(self) -> bool:
        with self._lock:
            expires_in = self._session_expires_in
        return expires_in is not None and int(time.time()) >= int(expires_in)

    def _server_heartbeat(self, generation: Optional[int] = None) -> None:
        with self._lock:
            session_token = self._session_token
            hwid = self._hwid
        if not session_token:
            raise RuntimeError("missing_session_token")

        body = {
            "appId": self.app_id,
            "sessionToken": session_token,
            "nonce": self._generate_nonce(),
            "hwid": hwid,
        }
        # Network failures surface once, as heartbeat_failed / network_error.
        response_obj = self._post_json("/auth/heartbeat", body)
        if not self._is_success_status(response_obj.get("status")):
            raise self._heartbeat_verdict(response_obj)
        expected_nonce = str(body.get("nonce", "")).strip()
        self._apply_signed_response(
            response_obj,
            expected_nonce=expected_nonce,
            license_key=None,
            context="heartbeat",
            generation=generation,
        )

    def _heartbeat_verdict(self, response_obj: Dict[str, Any]) -> AuthForgeError:
        # Only {"status": "failed", "error": "<code>"} is an AuthForge verdict.
        # Anything else (a proxy's JSON, a half-written body) must not be able
        # to end the session with a definitive code.
        status = response_obj.get("status")
        error = response_obj.get("error")
        if (
            isinstance(status, str)
            and status.strip().lower() == "failed"
            and isinstance(error, str)
            and error.strip()
        ):
            return AuthForgeError(self._extract_server_error(response_obj))
        return AuthForgeError(
            "unexpected_response",
            f"unexpected_response: status={status!r} error={error!r}",
        )

    def _grace_period_check(self) -> None:
        # Grace period enforcement: no network calls. Re-verify the stored
        # signed session payload and fail once the session TTL has expired.
        with self._lock:
            raw_payload_b64 = self._raw_payload_b64
            signature = self._signature
            expires_in = self._session_expires_in
        if not raw_payload_b64 or not signature:
            raise RuntimeError("missing_local_verification_state")

        self._verify_signature(raw_payload_b64, signature)

        if expires_in is None:
            raise RuntimeError("missing_session_expiry")

        now = int(time.time())
        if now >= int(expires_in):
            raise AuthForgeError("session_expired")

    def _validate_and_store(self, license_key: str) -> None:
        body: Dict[str, Any] = {
            "appId": self.app_id,
            "appSecret": self.app_secret,
            "licenseKey": license_key,
            "hwid": self._hwid,
            "nonce": self._generate_nonce(),
        }
        if self.ttl_seconds is not None:
            body["ttlSeconds"] = self.ttl_seconds
        # Network failures surface once, as login_failed / network_error.
        response_obj = self._post_json("/auth/validate", body)
        expected_nonce = str(body.get("nonce", "")).strip()
        self._apply_signed_response(
            response_obj,
            expected_nonce=expected_nonce,
            license_key=license_key,
            context="validate",
        )

    def _parse_validate_success(
        self, response_obj: Dict[str, Any], expected_nonce: str
    ) -> Dict[str, Any]:
        status = response_obj.get("status")
        if not self._is_success_status(status):
            raise AuthForgeError(self._extract_server_error(response_obj))

        raw_payload_b64 = self._require_str(response_obj, "payload")
        signature = self._require_str(response_obj, "signature")
        payload_json = self._decode_payload_json(raw_payload_b64)

        received_nonce = str(payload_json.get("nonce", "")).strip()
        if received_nonce != expected_nonce:
            raise ValueError("nonce_mismatch")

        self._verify_signature(raw_payload_b64, signature)

        session_token = str(payload_json.get("sessionToken", "")).strip()
        if not session_token:
            raise ValueError("missing_sessionToken")
        key_id = response_obj.get("keyId")
        if key_id is not None and not isinstance(key_id, str):
            raise ValueError("invalid_keyId")

        expires_from_token = self._extract_expires_in_from_session_token(session_token)
        expires_from_payload = payload_json.get("expiresIn")

        expires_in = expires_from_token
        if expires_in is None and expires_from_payload is not None:
            expires_in = int(expires_from_payload)
        if expires_in is None:
            raise ValueError("missing_expiresIn")

        out: Dict[str, Any] = {
            "session_token": session_token,
            "expires_in": int(expires_in),
            "session_data": dict(payload_json),
            "app_variables": self._extract_optional_map(payload_json.get("appVariables")),
            "license_variables": self._extract_optional_map(
                payload_json.get("licenseVariables")
            ),
            "key_id": key_id if isinstance(key_id, str) else None,
            "raw_payload_b64": raw_payload_b64,
            "signature": signature,
        }
        se = payload_json.get("sessionExpiresAt")
        if isinstance(se, str) and se != "":
            out["session_expires_at"] = se
        if "licenseExpiresAt" in payload_json:
            le = payload_json["licenseExpiresAt"]
            out["license_expires_at"] = le if isinstance(le, str) else None
        if "maxHwidSlots" in payload_json:
            try:
                out["max_hwid_slots"] = int(payload_json["maxHwidSlots"])
            except (TypeError, ValueError):
                pass
        if "hwidCount" in payload_json:
            try:
                out["hwid_count"] = int(payload_json["hwidCount"])
            except (TypeError, ValueError):
                pass
        ll = payload_json.get("licenseLabel")
        if isinstance(ll, str) and ll != "":
            out["license_label"] = ll
        return out

    def _apply_signed_response(
        self,
        response_obj: Dict[str, Any],
        expected_nonce: str,
        license_key: Optional[str],
        context: str,
        generation: Optional[int] = None,
    ) -> None:
        parsed = self._parse_validate_success(response_obj, expected_nonce)
        _ = context

        with self._lock:
            if generation is not None and generation != self._session_generation:
                return
            if license_key is not None:
                self._session_generation += 1
                self._license_key = license_key
            self._session_token = parsed["session_token"]
            self._session_kind = "online"
            self._session_expires_in = int(parsed["expires_in"])
            self._last_nonce = expected_nonce
            self._raw_payload_b64 = parsed["raw_payload_b64"]
            self._signature = parsed["signature"]
            self._key_id = parsed["key_id"]
            self._session_data = dict(parsed["session_data"])
            self._app_variables = parsed["app_variables"]
            self._license_variables = parsed["license_variables"]
            self._authenticated = True

    def _post_json(self, path: str, data: Dict[str, Any]) -> Dict[str, Any]:
        url = f"{self.api_base_url}{path}"
        body = dict(data)
        rate_attempt = 0
        while True:
            if rate_attempt > 0 and "nonce" in body:
                body["nonce"] = self._generate_nonce()

            network_attempt = 0
            while True:
                payload_bytes = json.dumps(body, separators=(",", ":")).encode("utf-8")
                request = urllib.request.Request(
                    url=url,
                    data=payload_bytes,
                    headers={"Content-Type": "application/json"},
                    method="POST",
                )
                try:
                    with urllib.request.urlopen(request, timeout=self.request_timeout) as response:
                        raw_response = response.read().decode("utf-8")
                        status_code = int(getattr(response, "status", 200))
                    obj = self._parse_response_object(raw_response)
                    data.clear()
                    data.update(body)
                    break
                except urllib.error.HTTPError as exc:
                    status_code = int(exc.code)
                    try:
                        detail = exc.read().decode("utf-8")
                        obj = self._parse_response_object(detail)
                    except Exception:
                        raise AuthForgeError(f"http_error_{status_code}") from exc
                    data.clear()
                    data.update(body)
                    break
                except (urllib.error.URLError, socket.timeout, TimeoutError) as exc:
                    if network_attempt == 0:
                        network_attempt += 1
                        time.sleep(NETWORK_RETRY_DELAY)
                        continue
                    reason = getattr(exc, "reason", exc)
                    code = "timeout" if isinstance(reason, (socket.timeout, TimeoutError)) else "network_error"
                    raise AuthForgeError(code, f"url_error: {exc}") from exc

            # no_credits / app_burn_cap_reached also use HTTP 429 but are not
            # worth retrying; only retry a genuine rate limit.
            server_error = self._extract_server_error(obj)
            is_rate_limited = server_error == "rate_limited" or (
                status_code == 429 and server_error == "unknown_error"
            )
            if is_rate_limited and rate_attempt < len(RATE_LIMIT_RETRY_DELAYS):
                time.sleep(RATE_LIMIT_RETRY_DELAYS[rate_attempt])
                rate_attempt += 1
                continue
            return obj

    def _parse_response_object(self, raw_response: str) -> Dict[str, Any]:
        try:
            obj = json.loads(raw_response)
        except json.JSONDecodeError as exc:
            raise AuthForgeError("invalid_json_response") from exc
        if not isinstance(obj, dict):
            raise AuthForgeError("response_not_json_object")
        return obj

    def _get_hwid(self) -> str:
        machine_id = _read_os_machine_id() or "legacy:" + self._legacy_fingerprint()
        return hwid_from_machine_id(self.app_id, machine_id)

    # Pre-2.0 fingerprint, kept only as the fallback when no OS machine id is readable.
    def _legacy_fingerprint(self) -> str:
        mac = self._safe_mac_address()
        cpu = self._safe_cpu_info()
        disk = self._safe_disk_serial()
        material = f"mac:{mac}|cpu:{cpu}|disk:{disk}"
        return hashlib.sha256(material.encode("utf-8")).hexdigest()

    def _resolve_hwid(self, hwid_override: Optional[str]) -> str:
        if isinstance(hwid_override, str):
            trimmed = hwid_override.strip()
            if trimmed:
                return trimmed
        return self._get_hwid()

    def _safe_mac_address(self) -> str:
        try:
            return f"{uuid.getnode():012x}"
        except Exception:
            return "mac-unavailable"

    def _safe_cpu_info(self) -> str:
        try:
            value = platform.processor() or platform.machine() or "cpu-unavailable"
            return str(value)
        except Exception:
            return "cpu-unavailable"

    def _safe_disk_serial(self) -> str:
        system = platform.system().lower()
        try:
            if "windows" in system:
                return self._run_command(["wmic", "diskdrive", "get", "serialnumber"])
            if "linux" in system:
                out = self._run_command(["lsblk", "-ndo", "SERIAL"])
                if out and out.strip():
                    return out
                return self._run_command(["udevadm", "info", "--query=property", "--name=sda"])
            if "darwin" in system:
                return self._run_command(["system_profiler", "SPStorageDataType"])
        except Exception:
            pass
        return "disk-unavailable"

    def _run_command(self, command: list[str]) -> str:
        try:
            output = subprocess.check_output(
                command,
                stderr=subprocess.DEVNULL,
                timeout=2,
            )
            cleaned = " ".join(output.decode("utf-8", errors="ignore").split())
            return cleaned[:256] if cleaned else "empty"
        except Exception:
            return "unavailable"

    def _decode_payload_json(self, payload_b64: str) -> Dict[str, Any]:
        payload_bytes = self._decode_base64_any(payload_b64)
        try:
            payload_obj = json.loads(payload_bytes.decode("utf-8"))
        except Exception as exc:
            raise ValueError("invalid_payload_json") from exc
        if not isinstance(payload_obj, dict):
            raise ValueError("payload_not_json_object")
        return payload_obj

    def _decode_base64_any(self, value: str) -> bytes:
        padded = self._add_base64_padding(value)
        try:
            return base64.b64decode(padded, validate=False)
        except Exception:
            return base64.urlsafe_b64decode(padded)

    def _extract_expires_in_from_session_token(self, session_token: str) -> Optional[int]:
        payload = self._decode_session_token_body(session_token)
        if payload is None:
            return None
        value = payload.get("exp")
        if value is None:
            return None
        return int(value)

    def _decode_session_token_body(self, session_token: str) -> Optional[Dict[str, Any]]:
        parts = session_token.split(".")
        if len(parts) < 2:
            return None
        padded = self._add_base64_padding(parts[0])
        try:
            decoded = base64.urlsafe_b64decode(padded)
            payload = json.loads(decoded.decode("utf-8"))
        except Exception:
            return None
        if not isinstance(payload, dict):
            return None
        return payload

    def _add_base64_padding(self, text: str) -> str:
        remainder = len(text) % 4
        if remainder == 0:
            return text
        return text + ("=" * (4 - remainder))

    def _load_public_key(self, public_key_b64: str) -> Ed25519PublicKey:
        try:
            public_key_bytes = base64.b64decode(
                self._add_base64_padding(public_key_b64), validate=True
            )
        except Exception as exc:
            raise ValueError("invalid_public_key") from exc
        if len(public_key_bytes) != 32:
            raise ValueError("invalid_public_key_length")
        return Ed25519PublicKey.from_public_bytes(public_key_bytes)

    def _verify_signature(self, raw_payload_b64: str, signature: str) -> None:
        try:
            signature_bytes = base64.b64decode(
                self._add_base64_padding(signature), validate=True
            )
        except Exception as exc:
            raise ValueError("invalid_signature_encoding") from exc
        # During a key rotation the SDK may be pinned to the previous key
        # while a new server-side key signs responses (or vice-versa). Trust
        # any key in the configured list.
        payload_bytes = raw_payload_b64.encode("utf-8")
        for key in self._ed25519_public_keys:
            try:
                key.verify(signature_bytes, payload_bytes)
                return
            except InvalidSignature:
                continue
        raise ValueError("signature_mismatch")

    @staticmethod
    def _normalize_public_key_list(value: PublicKeyArg) -> List[str]:
        """Coerce the public_key constructor arg to a list of base64 strings.

        Accepts:
          - "abc..."                     single-key historical contract
          - ["abc...", "def..."]         current first, previous after
          - "abc...,def..."              env-var convenience form
        """
        keys: List[str] = []
        candidates: Iterable[Any]
        if isinstance(value, str):
            candidates = value.split(",") if "," in value else [value]
        elif isinstance(value, Sequence):
            candidates = value
        else:
            return []
        for entry in candidates:
            if not isinstance(entry, str):
                continue
            trimmed = entry.strip()
            if trimmed:
                keys.append(trimmed)
        return keys

    def _generate_nonce(self) -> str:
        return secrets.token_hex(16)

    def _is_success_status(self, status: Any) -> bool:
        if isinstance(status, bool):
            return status
        if status is None:
            return False
        value = str(status).strip().lower()
        return value in {"ok", "success", "valid", "true", "1"}

    def _require_str(self, obj: Dict[str, Any], key: str) -> str:
        value = obj.get(key)
        if value is None:
            raise ValueError(f"missing_{key}")
        text = str(value)
        if not text:
            raise ValueError(f"empty_{key}")
        return text

    def _extract_server_error(self, obj: Dict[str, Any]) -> str:
        # Pass through codes this SDK version doesn't know yet instead of
        # collapsing them into unknown_error.
        raw = obj.get("error")
        raw_error = raw.strip().lower() if isinstance(raw, str) else ""
        if raw_error in KNOWN_SERVER_ERRORS or _SERVER_ERROR_CODE_RE.fullmatch(raw_error):
            return raw_error
        status = str(obj.get("status", "")).strip().lower()
        if status in KNOWN_SERVER_ERRORS:
            return status
        return "unknown_error"

    def _extract_optional_map(self, value: Any) -> Optional[Dict[str, Any]]:
        if isinstance(value, dict):
            return dict(value)
        return None

    def _fail(self, reason: str, exc: Optional[Exception] = None) -> None:
        if self.on_failure is not None:
            try:
                self.on_failure(reason, exc)
                return
            except Exception:
                pass
        os._exit(1)

    def logout(self) -> None:
        """Clear the session and stop background checks. Never joins the
        heartbeat thread, so it is safe to call from ``on_failure``."""
        with self._lock:
            self._clear_session_locked()

    def _clear_session_locked(self) -> None:
        self._heartbeat_stop.set()
        self._session_generation += 1
        self._license_key = None
        self._session_token = None
        self._session_kind = None
        self._session_expires_in = None
        self._last_nonce = None
        self._raw_payload_b64 = None
        self._signature = None
        self._key_id = None
        self._session_data = None
        self._app_variables = None
        self._license_variables = None
        self._authenticated = False
        self._offline_license = None
        self._heartbeat_started = False
        self._heartbeat_thread = None

    def is_authenticated(self) -> bool:
        """True for an online session (:meth:`login`) or an offline one
        (:meth:`login_from_file`)."""
        with self._lock:
            if not self._authenticated:
                return False
            if self._session_kind == "online":
                return bool(self._session_token)
            return self._session_kind == "offline"

    def get_session_kind(self) -> Optional[SessionKind]:
        """``"online"`` after :meth:`login`, ``"offline"`` after
        :meth:`login_from_file`, ``None`` when logged out."""
        with self._lock:
            return self._session_kind

    def get_session_data(self) -> Optional[Dict[str, Any]]:
        with self._lock:
            return dict(self._session_data) if self._session_data is not None else None

    def get_app_variables(self) -> Optional[Dict[str, Any]]:
        with self._lock:
            return dict(self._app_variables) if self._app_variables is not None else None

    def get_license_variables(self) -> Optional[Dict[str, Any]]:
        with self._lock:
            return dict(self._license_variables) if self._license_variables is not None else None