"""Neo Wallet API — signing helpers.

Copy this file into your integration. Python >= 3.9 and the `cryptography` package
(`pip install cryptography`) for Ed25519; everything else is the standard library.

It implements, byte for byte, what an integration has to compute:

- the request signature            — Authentication guide
- the transaction-binding hash     — Customer authentication guide
- the webhook signature check      — Webhooks guide

`python neo_wallet_signing.py` checks this file against test-vectors.json, the same vectors
Neo Wallet tests the TypeScript helper and the API itself with.
"""

from __future__ import annotations

import base64
import hashlib
import hmac
import json
import os
import secrets
import time
from dataclasses import dataclass
from typing import Mapping, Optional, Union

from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

Body = Optional[Union[str, bytes]]


def load_signing_key(pem_or_seed_base64: str) -> Ed25519PrivateKey:
    """Loads an Ed25519 key from a PKCS#8 PEM or a base64 32-byte seed (the demo credentials)."""
    value = pem_or_seed_base64.strip()
    if "BEGIN PRIVATE KEY" in value:
        key = serialization.load_pem_private_key(value.encode(), password=None)
        if not isinstance(key, Ed25519PrivateKey):
            raise ValueError("signing key is not Ed25519")
        return key
    seed = base64.b64decode(value)
    if len(seed) != 32:
        raise ValueError("an Ed25519 seed is 32 bytes (base64)")
    return Ed25519PrivateKey.from_private_bytes(seed)


@dataclass(frozen=True)
class Credentials:
    api_key: str  # X-Api-Key, nwak_…
    key_id: str  # X-Key-Id, the kid registered for your public key
    private_key: Ed25519PrivateKey


def _bytes(body: Body) -> bytes:
    if body is None:
        return b""
    return body.encode("utf-8") if isinstance(body, str) else body


def body_sha256_hex(body: Body = None) -> str:
    """Lower-case hex SHA-256 of the exact body bytes; no body hashes as the empty string."""
    return hashlib.sha256(_bytes(body)).hexdigest()


def canonical_request(method: str, path_with_query: str, timestamp: str, body: Body = None) -> str:
    """METHOD \\n PATH+QUERY \\n X-Timestamp \\n sha256hex(body) — no trailing newline."""
    return "\n".join([method.upper(), path_with_query, timestamp, body_sha256_hex(body)])


def sign_request(
    credentials: Credentials,
    method: str,
    path_with_query: str,
    body: Body = None,
    timestamp: Optional[str] = None,
) -> dict:
    """The four authentication headers. A signature is single-use; sign every attempt afresh."""
    ts = timestamp if timestamp is not None else str(int(time.time()))
    canonical = canonical_request(method, path_with_query, ts, body)
    signature = credentials.private_key.sign(canonical.encode("utf-8"))
    return {
        "X-Api-Key": credentials.api_key,
        "X-Key-Id": credentials.key_id,
        "X-Timestamp": ts,
        "X-Signature": base64.b64encode(signature).decode("ascii"),
    }


_CROCKFORD = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"


def ulid(now_ms: Optional[int] = None) -> str:
    """A ULID — the preferred X-Request-Id, and a good Idempotency-Key."""
    ms = now_ms if now_ms is not None else int(time.time() * 1000)
    head = ""
    for _ in range(10):
        head = _CROCKFORD[ms % 32] + head
        ms //= 32
    return head + "".join(_CROCKFORD[b % 32] for b in secrets.token_bytes(16))


def build_request(
    credentials: Credentials,
    method: str,
    path_with_query: str,
    json_body: object = None,
    idempotency_key: Optional[str] = None,
) -> tuple:
    """(headers, body) for one attempt. Reuse the same idempotency_key when retrying a write."""
    method = method.upper()
    body = None if json_body is None else json.dumps(json_body, separators=(",", ":"))
    is_write = method not in ("GET", "HEAD")
    if is_write and not idempotency_key:
        raise ValueError("writes need an Idempotency-Key")
    headers = sign_request(credentials, method, path_with_query, body)
    headers["X-Request-Id"] = ulid()
    if body is not None:
        headers["Content-Type"] = "application/json"
    if is_write:
        headers["Idempotency-Key"] = idempotency_key
    return headers, body


TRANSACTION_BINDING_CANONICAL = (
    "customerUid|walletUid|amountMinor|currency|destinationIdentifierUid|clientReference"
)


def transaction_binding(
    customer_uid: str,
    wallet_uid: str,
    amount_minor: int,
    currency: str,
    target: Optional[str] = None,
    client_reference: Optional[str] = None,
) -> dict:
    """assertion.transactionBinding. target = destinationIdentifierUid or mandateUid."""
    canonical = "|".join(
        [customer_uid, wallet_uid, str(amount_minor), currency, target or "", client_reference or ""]
    )
    return {
        "hash": hashlib.sha256(canonical.encode("utf-8")).hexdigest(),
        "algorithm": "sha256",
        "canonical": TRANSACTION_BINDING_CANONICAL,
    }


def verify_webhook(
    raw_body: bytes,
    headers: Mapping[str, str],
    secrets_by_kid: Mapping[str, str],
    tolerance_seconds: int = 300,
    now: Optional[float] = None,
) -> bool:
    """Webhook signature check over the RAW body bytes; header names are matched case-insensitively."""
    lowered = {k.lower(): v for k, v in headers.items()}
    secret = secrets_by_kid.get(lowered.get("x-webhook-key-id", ""))
    timestamp = lowered.get("x-webhook-timestamp", "")
    presented = lowered.get("x-webhook-signature", "").strip().lower()
    if not secret or not timestamp.isdigit() or len(presented) != 64:
        return False
    expected = hmac.new(secret.encode("utf-8"), timestamp.encode("ascii") + b"." + raw_body, hashlib.sha256)
    if not hmac.compare_digest(presented, expected.hexdigest()):
        return False
    current = now if now is not None else time.time()
    return abs(int(current) - int(timestamp)) <= tolerance_seconds


def _self_test() -> None:
    path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "test-vectors.json")
    with open(path, encoding="utf-8") as handle:
        vectors = json.load(handle)

    signing = vectors["requestSigning"]
    key = load_signing_key(signing["seedBase64"])
    public = key.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw)
    assert base64.b64encode(public).decode() == signing["publicKeyBase64"], "public key"
    credentials = Credentials(signing["apiKey"], signing["keyId"], key)
    for case in signing["cases"]:
        name = f'{case["method"]} {case["pathWithQuery"]}'
        body = case["body"]
        assert body_sha256_hex(body) == case["bodySha256Hex"], name + ": body hash"
        assert canonical_request(case["method"], case["pathWithQuery"], case["timestamp"], body) == case["canonical"], name + ": canonical"
        headers = sign_request(credentials, case["method"], case["pathWithQuery"], body, case["timestamp"])
        assert headers["X-Signature"] == case["signature"], name + ": signature"

    binding = vectors["transactionBinding"]
    given = binding["input"]
    result = transaction_binding(
        given["customerUid"], given["walletUid"], given["amountMinor"], given["currency"],
        given.get("target"), given.get("clientReference"),
    )
    assert result["hash"] == binding["hash"], "transaction binding"

    hook = vectors["webhook"]
    hook_headers = {
        "X-Webhook-Key-Id": hook["keyId"],
        "X-Webhook-Timestamp": hook["timestamp"],
        "X-Webhook-Signature": hook["signature"],
    }
    secrets_by_kid = {hook["keyId"]: hook["secret"]}
    raw = hook["body"].encode("utf-8")
    assert verify_webhook(raw, hook_headers, secrets_by_kid, now=int(hook["timestamp"])), "webhook"
    assert not verify_webhook(raw + b" ", hook_headers, secrets_by_kid, now=int(hook["timestamp"])), "tampered webhook"

    assert len(ulid()) == 26, "ulid"
    print("neo_wallet_signing: all test vectors pass")


if __name__ == "__main__":
    _self_test()