"""
Foliox License - Python client, v2.

Checks a license key against your license server. Every answer is verified with the server's Ed25519 public key
before it is believed. Requires: ``pip install cryptography`` (Python 3.9+).

    from license_client import LicenseClient

    client = LicenseClient(
        base_url="https://YOUR-SERVER",
        product="YOUR_PRODUCT_ID",                 # shown on the product's page in the dashboard
        public_key_b64="PASTE_THE_PUBLIC_KEY",     # from the docs page; it is public, not a secret
    )
    result = client.check(user_entered_key)        # activates on first run, verifies afterwards
    if not result.valid:
        raise SystemExit(f"Not licensed: {result.reason}")

What this client does for you (and why):

* HTTPS only. A plain ``http://`` address is refused (loopback addresses excepted, for local development).
  Certificates are always verified, TLS 1.2 or newer, and redirects are never followed - a redirect could send
  the key somewhere else.
* Nothing is trusted before its signature is checked: Ed25519 over the exact payload, then the product it names,
  then its age (``max_response_age``) so an old captured "valid" answer can't be replayed.
* Transient trouble (network errors, timeouts, HTTP 429/5xx) is retried with exponential backoff and jitter inside
  a hard overall deadline. A definite answer from the server - revoked, expired, wrong device - is never retried
  and never softened by the cache.
* If the server can't be reached, a locally stored signed answer keeps the app running for the grace period the
  server granted (72 hours), and never past the key's own expiry. The stored answer is bound to this device, this
  product and this key, the raw key is never written to disk, and a system clock set back is detected.
* ``check()`` never raises for runtime conditions: it always returns a ``LicenseResult``. Bad configuration
  raises ``LicenseConfigError`` immediately, when the client is constructed.

Offline grace and the cache are resilience features, not a security boundary: whoever controls the machine
controls its clock and its files. See the security model in the docs for what a license check can and can't do.
"""

from __future__ import annotations

import base64
import datetime as _dt
import hashlib
import http.client
import json
import os
import platform
import random
import re
import socket
import ssl
import subprocess
import tempfile
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Optional

from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from cryptography.hazmat.primitives.serialization import load_der_public_key

__version__ = "2.0.0"
__all__ = [
    "LicenseClient",
    "LicenseResult",
    "LicenseError",
    "LicenseConfigError",
    "LicenseInvalid",
    "compute_hwid",
    "is_valid_key_format",
    "normalize_key",
]

_USER_AGENT = f"foliox-license-python/{__version__}"
_MAX_BODY = 64 * 1024  # a signed answer is ~500 bytes; anything bigger is not our server
_LOOPBACK_HOSTS = {"localhost", "127.0.0.1", "::1"}
_PRODUCT_RE = re.compile(r"^[a-z0-9-]{1,64}$")
_HWID_RE = re.compile(r"^[a-f0-9]{64}$")
_CLOCK_TOLERANCE = 300.0  # seconds a clock may run behind its own last-seen time before we call it rolled back
_MAX_RETRY_AFTER = 5.0


# ---------------------------------------------------------------------------
# Errors and results
# ---------------------------------------------------------------------------


class LicenseError(Exception):
    """Base class for everything this module raises."""


class LicenseConfigError(LicenseError, ValueError):
    """The client was configured with something unsafe or malformed (raised at construction)."""


class LicenseInvalid(LicenseError):
    """Raised by ``LicenseClient.require()`` when the license is not valid. ``.result`` has the details."""

    def __init__(self, result: "LicenseResult"):
        super().__init__(f"License is not valid: {result.reason}")
        self.result = result


@dataclass
class LicenseResult:
    """The outcome of a check.

    ``reason`` is only meaningful when ``valid`` is False. Server reasons: ``not_found``, ``revoked``,
    ``expired``, ``hwid_mismatch``, ``not_activated``, ``malformed_key``, ``blocked``, ``suspended``.
    Client-side reasons: ``network_error``, ``rate_limited``, ``signature_invalid``, ``invalid_response``,
    ``invalid_request``, ``product_mismatch``, ``clock_skew``, ``clock_rollback``.
    """

    valid: bool
    reason: Optional[str]
    status: Optional[str]
    expires_at: Optional[str]
    from_offline_cache: bool = False
    key_id: Optional[int] = None
    checked_at: Optional[str] = None  # the server's own timestamp for the answer (signed)
    message: Optional[str] = None  # human-readable detail for logs; never contains the key

    def __bool__(self) -> bool:  # `if client.check(key):` reads naturally
        return self.valid


# ---------------------------------------------------------------------------
# Key format (identical to the server's, so typos are caught without a round trip)
# ---------------------------------------------------------------------------

_ALPHABET = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
_KEY_RE = re.compile(r"^[%s]{5}(-[%s]{5}){3}-[%s]$" % (_ALPHABET, _ALPHABET, _ALPHABET))


def normalize_key(key: str) -> str:
    """Trims, upper-cases and drops anything that can't be part of a key (spaces, quotes, zero-width characters)."""
    return re.sub(r"[^0-9A-Z-]", "", str(key or "").strip().upper())


def is_valid_key_format(key: str) -> bool:
    """True when ``key`` has the right shape and its check character is correct."""
    clean = normalize_key(key)
    if not _KEY_RE.match(clean):
        return False
    flat = clean.replace("-", "")
    body, check = flat[:-1], flat[-1]
    total = sum(_ALPHABET.index(ch) * (i + 1) for i, ch in enumerate(body))
    return _ALPHABET[total % len(_ALPHABET)] == check


# ---------------------------------------------------------------------------
# HWID: a SHA-256 hash of a platform-stable machine identifier. Only the hash
# ever leaves the device - raw hardware identifiers are never transmitted.
# ---------------------------------------------------------------------------


def compute_hwid() -> str:
    raw = _raw_machine_id()
    return hashlib.sha256(raw.strip().encode("utf-8")).hexdigest()


def _raw_machine_id() -> str:
    system = platform.system()
    if system == "Windows":
        try:
            import winreg

            key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\Microsoft\Cryptography")
            value, _ = winreg.QueryValueEx(key, "MachineGuid")
            return value
        except OSError:
            pass
    elif system == "Linux":
        for path in ("/etc/machine-id", "/var/lib/dbus/machine-id"):
            if os.path.exists(path):
                return Path(path).read_text().strip()
    elif system == "Darwin":
        try:
            out = subprocess.check_output(["ioreg", "-rd1", "-c", "IOPlatformExpertDevice"], text=True)
            for line in out.splitlines():
                if "IOPlatformUUID" in line:
                    return line.split('"')[-2]
        except (OSError, subprocess.CalledProcessError):
            pass

    # Last-resort fallback: known to be less stable (changes if the network adapter changes) but
    # guarantees compute_hwid() never raises.
    import uuid

    return f"fallback-{uuid.getnode()}"


# ---------------------------------------------------------------------------
# Transport
# ---------------------------------------------------------------------------


class _NoRedirect(urllib.request.HTTPRedirectHandler):
    """A license server never redirects its API. Following one could hand the key to another host."""

    def redirect_request(self, req, fp, code, msg, headers, newurl):  # noqa: D401
        return None


class _Transient(Exception):
    def __init__(self, message: str, kind: str = "network_error", retry: bool = True, retry_after: Optional[float] = None):
        super().__init__(message)
        self.kind = kind
        self.retry = retry
        self.retry_after = retry_after


class _Unreachable(Exception):
    """Could not get an answer (after retrying): fall back to the offline grace period."""

    def __init__(self, message: str, kind: str = "network_error"):
        super().__init__(message)
        self.kind = kind


class _Rejected(Exception):
    """The server's answer can't be trusted or used: a definite failure, never softened by the cache."""

    def __init__(self, reason: str, message: str):
        super().__init__(message)
        self.reason = reason


def _retry_after_seconds(headers) -> Optional[float]:
    raw = headers.get("Retry-After") if headers else None
    try:
        return max(0.0, min(float(raw), _MAX_RETRY_AFTER)) if raw is not None else None
    except (TypeError, ValueError):
        return None


def _canonical(value) -> bytes:
    """Must match the server byte for byte: keys sorted at every level, no whitespace."""

    def order(v):
        if isinstance(v, dict):
            return {k: order(v[k]) for k in sorted(v)}
        if isinstance(v, list):
            return [order(x) for x in v]
        return v

    return json.dumps(order(value), separators=(",", ":"), ensure_ascii=False, allow_nan=False).encode("utf-8")


def _parse_time(text) -> Optional[float]:
    if not isinstance(text, str):
        return None
    try:
        return _dt.datetime.fromisoformat(text.replace("Z", "+00:00")).timestamp()
    except ValueError:
        return None


def _iso(epoch: float) -> str:
    return _dt.datetime.fromtimestamp(epoch, _dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.") + f"{int(epoch * 1000) % 1000:03d}Z"


# ---------------------------------------------------------------------------
# The client
# ---------------------------------------------------------------------------


class LicenseClient:
    def __init__(
        self,
        base_url: str,
        product: str,
        public_key_b64: str,
        cache_path: Optional[str] = None,
        timeout: float = 6.0,
        *,
        max_retries: int = 2,
        backoff_base: float = 0.4,
        backoff_cap: float = 4.0,
        total_deadline: float = 20.0,
        max_response_age: Optional[float] = 900.0,
        hwid: Optional[str] = None,
        allow_insecure_loopback: bool = True,
        user_agent: Optional[str] = None,
        _now: Callable[[], float] = time.time,
        _sleep: Callable[[float], None] = time.sleep,
    ):
        self._base = self._check_base_url(base_url, allow_insecure_loopback)
        if not isinstance(product, str) or not _PRODUCT_RE.match(product):
            raise LicenseConfigError("product must be the product ID from your dashboard (lowercase letters, digits and hyphens)")
        self.product = product
        self._public_key = self._load_public_key(public_key_b64)
        if hwid is not None and not _HWID_RE.match(str(hwid)):
            raise LicenseConfigError("hwid must be a lowercase SHA-256 hex string (64 characters)")
        self._hwid_override = hwid
        if not (0 < timeout <= 120):
            raise LicenseConfigError("timeout must be between 0 and 120 seconds")
        self._timeout = float(timeout)
        self._max_retries = max(0, min(int(max_retries), 5))
        self._backoff_base = max(0.0, float(backoff_base))
        self._backoff_cap = max(self._backoff_base, float(backoff_cap))
        self._total_deadline = max(1.0, float(total_deadline))
        self._max_age = float(max_response_age) if max_response_age else None
        self.cache_path = Path(cache_path) if cache_path else Path.home() / f".{product}" / "license_cache.json"
        self._now = _now
        self._sleep = _sleep
        self._lock = threading.Lock()

        context = ssl.create_default_context()
        context.minimum_version = ssl.TLSVersion.TLSv1_2
        self._opener = urllib.request.build_opener(urllib.request.HTTPSHandler(context=context), _NoRedirect())
        self._user_agent = user_agent or _USER_AGENT

    def __repr__(self) -> str:  # holds no secrets, but keep logs tidy anyway
        return f"LicenseClient(base_url={self._base!r}, product={self.product!r})"

    # -- configuration ------------------------------------------------------------------------------------------

    @staticmethod
    def _check_base_url(base_url: str, allow_insecure_loopback: bool) -> str:
        try:
            parts = urllib.parse.urlsplit(str(base_url or "").strip())
            host = (parts.hostname or "").lower()
        except ValueError as exc:
            raise LicenseConfigError("base_url must be a full address such as https://license.example.com") from exc
        if parts.scheme not in ("https", "http") or not host:
            raise LicenseConfigError("base_url must be a full address such as https://license.example.com")
        if parts.username or parts.password or parts.query or parts.fragment:
            raise LicenseConfigError("base_url must not contain credentials, a query string or a fragment")
        if parts.scheme == "http" and not (allow_insecure_loopback and host in _LOOPBACK_HOSTS):
            raise LicenseConfigError("base_url must use https:// - a license key must never cross the network in the clear")
        return f"{parts.scheme}://{parts.netloc}{parts.path.rstrip('/')}"

    @staticmethod
    def _load_public_key(public_key_b64: str) -> Ed25519PublicKey:
        try:
            key = load_der_public_key(base64.b64decode(str(public_key_b64), validate=True))
        except Exception as exc:  # noqa: BLE001 - any parse failure is a config error
            raise LicenseConfigError("public_key_b64 is not a valid base64 public key") from exc
        if not isinstance(key, Ed25519PublicKey):
            raise LicenseConfigError("public_key_b64 is not an Ed25519 key")
        return key

    # -- public API ---------------------------------------------------------------------------------------------

    def check(self, key: str, *, activate: bool = True) -> LicenseResult:
        """Verifies ``key`` on this device - binding it on first use when ``activate`` is true - and returns the outcome."""
        normalized = normalize_key(key)
        if not is_valid_key_format(normalized):
            return LicenseResult(False, "malformed_key", None, None, message="The key is not in the expected format")
        hwid = self._hwid_override or compute_hwid()
        deadline = time.monotonic() + self._total_deadline
        try:
            payload, envelope = self._exchange("/api/v1/verify", normalized, hwid, deadline)
            if activate and payload.get("reason") == "not_activated":
                payload, envelope = self._exchange("/api/v1/activate", normalized, hwid, deadline)
        except _Unreachable as exc:
            return self._offline(normalized, hwid, exc)
        except _Rejected as exc:
            return LicenseResult(False, exc.reason, None, None, message=str(exc))

        result = self._result(payload)
        if result.valid:
            self._cache_store(normalized, hwid, envelope)
        else:
            self._cache_clear()  # a definite "no" must not be resurrected by the offline grace period
        return result

    def require(self, key: str) -> LicenseResult:
        """Like ``check`` but raises ``LicenseInvalid`` unless the license is valid."""
        result = self.check(key)
        if not result.valid:
            raise LicenseInvalid(result)
        return result

    def clear_cache(self) -> None:
        """Forget the stored offline answer (for example when the user signs out or transfers the license)."""
        self._cache_clear()

    # -- one verified exchange ----------------------------------------------------------------------------------

    def _exchange(self, path: str, key: str, hwid: str, deadline: float):
        status, headers, raw = self._post(path, {"product": self.product, "key": key, "hwid": hwid}, deadline)
        try:
            envelope = json.loads(raw.decode("utf-8"))
        except (UnicodeDecodeError, ValueError) as exc:
            raise _Rejected("invalid_response", "The server's answer was not JSON - is base_url your license server?") from exc
        payload = self._verified_payload(envelope)
        return payload, envelope

    def _verified_payload(self, envelope) -> dict:
        if not isinstance(envelope, dict) or not isinstance(envelope.get("payload"), dict) or not isinstance(envelope.get("signature"), str):
            raise _Rejected("invalid_response", "The server's answer had an unexpected shape")
        payload = envelope["payload"]
        try:
            signature = base64.b64decode(envelope["signature"], validate=True)
            self._public_key.verify(signature, _canonical(payload))
        except (InvalidSignature, ValueError, TypeError) as exc:
            raise _Rejected("signature_invalid", "The answer's signature does not match the public key") from exc
        if payload.get("product_slug") != self.product:
            raise _Rejected("product_mismatch", "The answer is for a different product")
        if not isinstance(payload.get("valid"), bool):
            raise _Rejected("invalid_response", "The signed answer has no valid/invalid verdict")
        issued = _parse_time(payload.get("issued_at"))
        if issued is None:
            raise _Rejected("invalid_response", "The signed answer has no timestamp")
        if self._max_age is not None and abs(self._now() - issued) > self._max_age:
            raise _Rejected("clock_skew", "The answer's timestamp is too far from this computer's clock - fix the system time (or a stale answer was replayed)")
        return payload

    def _result(self, payload: dict, *, from_cache: bool = False) -> LicenseResult:
        valid = payload["valid"]
        return LicenseResult(
            valid=valid,
            reason=None if valid else (payload.get("reason") or "invalid"),
            status=payload.get("status"),
            expires_at=payload.get("expires_at"),
            from_offline_cache=from_cache,
            key_id=payload.get("key_id") if isinstance(payload.get("key_id"), int) else None,
            checked_at=payload.get("issued_at"),
        )

    # -- HTTP with retries --------------------------------------------------------------------------------------

    def _post(self, path: str, body: dict, deadline: float):
        last: Optional[_Transient] = None
        for attempt in range(self._max_retries + 1):
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                break
            try:
                return self._post_once(path, body, min(self._timeout, remaining))
            except _Transient as exc:
                last = exc
                if not exc.retry or attempt == self._max_retries:
                    break
                if exc.retry_after is not None:
                    wait = exc.retry_after
                else:  # exponential backoff with full jitter
                    wait = random.uniform(0.0, min(self._backoff_cap, self._backoff_base * (2**attempt)))
                if time.monotonic() + wait >= deadline:
                    break
                self._sleep(wait)
        raise _Unreachable(str(last) if last else "no time left to reach the license server", last.kind if last else "network_error")

    def _post_once(self, path: str, body: dict, timeout: float):
        request = urllib.request.Request(
            self._base + path,
            data=json.dumps(body, separators=(",", ":")).encode("utf-8"),
            method="POST",
            headers={"Content-Type": "application/json", "Accept": "application/json", "User-Agent": self._user_agent},
        )
        try:
            with self._opener.open(request, timeout=timeout) as response:
                status, headers, raw = response.status, response.headers, response.read(_MAX_BODY + 1)
        except urllib.error.HTTPError as err:
            try:
                status, headers, raw = err.code, err.headers, err.read(_MAX_BODY + 1)
            except (OSError, http.client.HTTPException):
                status, headers, raw = err.code, err.headers, b""
            finally:
                err.close()
        except ssl.SSLCertVerificationError as exc:
            raise _Transient("TLS certificate verification failed", retry=False) from exc
        except urllib.error.URLError as exc:
            if isinstance(exc.reason, ssl.SSLCertVerificationError):
                raise _Transient("TLS certificate verification failed", retry=False) from exc
            raise _Transient(f"could not reach the license server ({exc.reason})") from exc
        except (socket.timeout, TimeoutError, ConnectionError, ssl.SSLError, OSError, http.client.HTTPException) as exc:
            raise _Transient(f"could not reach the license server ({exc.__class__.__name__})") from exc

        if len(raw) > _MAX_BODY:
            raise _Rejected("invalid_response", "The server's answer was unexpectedly large")
        if status == 200:
            return status, headers, raw
        if status == 429:
            raise _Transient("rate limited by the license server", kind="rate_limited", retry_after=_retry_after_seconds(headers))
        if status in (500, 502, 503, 504):
            raise _Transient(f"license server error (HTTP {status})")
        if status == 400:
            raise _Rejected("invalid_request", "The license server rejected the request as malformed")
        if 300 <= status < 400:
            raise _Rejected("invalid_response", "The server redirected the request - check that base_url is exactly your license server's address")
        raise _Rejected("invalid_response", f"Unexpected HTTP {status} from the license server")

    # -- offline grace ------------------------------------------------------------------------------------------

    def _fingerprint(self, key: str) -> str:
        return hashlib.sha256(f"foliox-license-cache:{self.product}:{key}".encode("utf-8")).hexdigest()[:32]

    def _offline(self, key: str, hwid: str, cause: _Unreachable) -> LicenseResult:
        failure = LicenseResult(False, cause.kind, None, None, message=str(cause))
        with self._lock:
            entry = self._cache_read()
            if not entry or entry.get("hwid") != hwid or entry.get("key_fp") != self._fingerprint(key):
                return failure
            try:
                envelope = entry["envelope"]
                payload = self._trusted_cached_payload(envelope)
            except _Rejected:
                return failure
            now = self._now()
            seen = _parse_time(entry.get("last_seen"))
            if seen is not None and now < seen - _CLOCK_TOLERANCE:
                return LicenseResult(False, "clock_rollback", None, None, message="The system clock is earlier than the last time this license was seen")
            if not payload["valid"]:
                return failure
            grace = _parse_time(payload.get("offline_grace_until"))
            expires = _parse_time(payload.get("expires_at")) if payload.get("expires_at") else None
            limit = min(t for t in (grace, expires) if t is not None) if grace is not None else None
            if limit is None or now >= limit:
                return failure
            entry["last_seen"] = _iso(max(now, seen or 0))
            self._cache_write(entry)
        result = self._result(payload, from_cache=True)
        result.message = f"Working offline from a signed answer ({cause})"
        return result

    def _trusted_cached_payload(self, envelope) -> dict:
        """Re-verifies a stored answer. It may be old, so the age limit does not apply - the grace period does."""
        if not isinstance(envelope, dict) or not isinstance(envelope.get("payload"), dict) or not isinstance(envelope.get("signature"), str):
            raise _Rejected("invalid_response", "cache entry malformed")
        payload = envelope["payload"]
        try:
            self._public_key.verify(base64.b64decode(envelope["signature"], validate=True), _canonical(payload))
        except (InvalidSignature, ValueError, TypeError) as exc:
            raise _Rejected("signature_invalid", "cache entry signature invalid") from exc
        if payload.get("product_slug") != self.product or not isinstance(payload.get("valid"), bool):
            raise _Rejected("product_mismatch", "cache entry is for something else")
        return payload

    def _cache_read(self) -> Optional[dict]:
        try:
            if self.cache_path.stat().st_size > _MAX_BODY:
                return None
            entry = json.loads(self.cache_path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            return None
        if not isinstance(entry, dict) or entry.get("v") != 2 or entry.get("product") != self.product:
            return None
        return entry

    def _cache_store(self, key: str, hwid: str, envelope: dict) -> None:
        with self._lock:
            self._cache_write({"v": 2, "product": self.product, "hwid": hwid, "key_fp": self._fingerprint(key), "envelope": envelope, "last_seen": _iso(self._now())})

    def _cache_write(self, entry: dict) -> None:
        """Atomic, owner-only. Best-effort: a read-only disk must never make a license check fail."""
        try:
            directory = self.cache_path.parent
            directory.mkdir(parents=True, exist_ok=True, mode=0o700)
            fd, tmp = tempfile.mkstemp(prefix=".license-", suffix=".tmp", dir=str(directory))
            try:
                with os.fdopen(fd, "w", encoding="utf-8") as handle:
                    json.dump(entry, handle, separators=(",", ":"))
                os.replace(tmp, self.cache_path)
            except BaseException:
                try:
                    os.unlink(tmp)
                except OSError:
                    pass
                raise
        except OSError:
            pass

    def _cache_clear(self) -> None:
        with self._lock:
            try:
                self.cache_path.unlink()
            except OSError:
                pass