thoxmesh_node/signing.py

"""Ed25519 request signing for the ``meshstack-device-v2`` edge function.

The canonical string and header names here are not our invention: they mirror
``supabase/functions/meshstack-device-v2/index.ts`` in ``ttracx/sadie-mesh``,
which builds

    POST\\n{action}\\n{timestamp}\\n{nonce}\\n{sha256_hex(raw_body)}

and verifies it against ``meshstack_devices_v2.identity_public_key``. Any drift
between this module and that function is a silent authentication failure, so the
canonical form is asserted by ``tests/test_signing.py`` against a vector taken
from the edge function's own construction.

Two properties matter for correctness:

* The body hash is over the **exact bytes sent on the wire**. We therefore
  serialise once and sign that same buffer, rather than re-serialising, because
  ``json.dumps`` key order or separator changes would invalidate the signature.
* The timestamp must land inside the backend's +/-5 minute window, so the node
  clock is checked at startup rather than failing obscurely on first register.
"""

from __future__ import annotations

import base64
import hashlib
import json
import os
import secrets
import time
from dataclasses import dataclass
from typing import Any, Mapping

from .errors import ConfigError

try:
    from cryptography.hazmat.primitives import serialization
    from cryptography.hazmat.primitives.asymmetric.ed25519 import (
        Ed25519PrivateKey,
        Ed25519PublicKey,
    )
except ImportError as _import_error:  # pragma: no cover - environment dependent
    # Re-raised with instructions rather than letting a bare ModuleNotFoundError
    # surface from deep inside `import thoxmesh_node`, where it reads as a broken
    # package rather than one missing dependency.
    raise ConfigError(
        "the 'cryptography' package is required for mesh device signing. Install with:\n"
        "  pip install cryptography\n"
        "It ships prebuilt wheels for every platform this node targets, so this "
        "should not need a compiler."
    ) from _import_error

#: Header names expected by ``meshstack-device-v2``.
HEADER_DEVICE_ID = "x-thox-device-id"
HEADER_TIMESTAMP = "x-thox-timestamp"
HEADER_NONCE = "x-thox-nonce"
HEADER_SIGNATURE = "x-thox-signature"

#: The backend rejects requests more than 5 minutes from its own clock.
CLOCK_SKEW_LIMIT_SECONDS = 5 * 60


def sha256_hex(payload: bytes) -> str:
    """Hex SHA-256, matching the edge function's ``sha256Hex``."""
    return hashlib.sha256(payload).hexdigest()


def canonical_string(action: str, timestamp_ms: str, nonce: str, body_hash: str) -> str:
    """Build the exact string the backend reconstructs and verifies.

    Kept as a standalone function so the test suite can pin the format without
    needing a private key.
    """
    return f"POST\n{action}\n{timestamp_ms}\n{nonce}\n{body_hash}"


@dataclass(frozen=True)
class SignedRequest:
    """A ready-to-send signed request: the exact body bytes plus headers."""

    body: bytes
    headers: dict[str, str]


class DeviceIdentity:
    """A mesh device identity: a device UUID plus its Ed25519 signing key.

    The private key is a real secret. It is loaded from the environment or a file
    and never logged, never serialised into telemetry, and never committed. In
    Colab it is pasted into a form field; in a Space it is a Space secret.
    """

    def __init__(self, device_id: str, private_key: Ed25519PrivateKey) -> None:
        if not device_id:
            raise ConfigError("device_id must not be empty")
        self._device_id = device_id
        self._private_key = private_key

    # ── Construction ─────────────────────────────────────────────────────

    @classmethod
    def from_base64_seed(cls, device_id: str, seed_b64: str) -> "DeviceIdentity":
        """Load from a base64 32-byte Ed25519 seed (the private key material).

        Accepts standard or URL-safe base64, with or without padding, because
        these strings get copy-pasted through notebooks and Space secret fields
        where padding is routinely mangled.
        """
        raw = _b64decode_forgiving(seed_b64, field="private key seed")
        if len(raw) == 64:
            # Some tools emit the libsodium 64-byte "secret key" (seed || public).
            raw = raw[:32]
        if len(raw) != 32:
            raise ConfigError(
                f"Ed25519 private key seed must decode to 32 bytes, got {len(raw)}"
            )
        return cls(device_id, Ed25519PrivateKey.from_private_bytes(raw))

    @classmethod
    def from_pem_file(cls, device_id: str, path: str) -> "DeviceIdentity":
        """Load from an unencrypted PKCS#8 PEM private key file."""
        try:
            with open(path, "rb") as handle:
                key = serialization.load_pem_private_key(handle.read(), password=None)
        except (OSError, ValueError) as exc:
            raise ConfigError(f"could not read Ed25519 private key from {path}: {exc}") from exc
        if not isinstance(key, Ed25519PrivateKey):
            raise ConfigError(f"{path} is not an Ed25519 private key")
        return cls(device_id, key)

    @classmethod
    def generate(cls, device_id: str = "00000000-0000-0000-0000-000000000000") -> "DeviceIdentity":
        """Generate an ephemeral identity. For tests and key-provisioning only.

        A generated key is useless against the live mesh until its public half is
        registered on ``meshstack_devices_v2`` via a pairing session, which
        requires an authenticated mesh owner.
        """
        return cls(device_id, Ed25519PrivateKey.generate())

    # ── Accessors ────────────────────────────────────────────────────────

    @property
    def device_id(self) -> str:
        return self._device_id

    def public_key_base64(self) -> str:
        """Base64 raw public key, the format stored in ``identity_public_key``."""
        raw = self._private_key.public_key().public_bytes(
            encoding=serialization.Encoding.Raw,
            format=serialization.PublicFormat.Raw,
        )
        return base64.b64encode(raw).decode("ascii")

    def private_seed_base64(self) -> str:
        """Base64 raw private seed. Only used by the key-provisioning command."""
        raw = self._private_key.private_bytes(
            encoding=serialization.Encoding.Raw,
            format=serialization.PrivateFormat.Raw,
            encryption_algorithm=serialization.NoEncryption(),
        )
        return base64.b64encode(raw).decode("ascii")

    # ── Signing ──────────────────────────────────────────────────────────

    def sign_request(self, action: str, payload: Mapping[str, Any]) -> SignedRequest:
        """Serialise, hash, sign and return body + headers for one mesh call.

        ``action`` is signed separately from the body even though the body also
        carries it; that is the backend's construction and we match it exactly.
        """
        if not action:
            raise ConfigError("action must not be empty")
        body = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode("utf-8")
        timestamp_ms = str(int(time.time() * 1000))
        nonce = secrets.token_hex(16)
        message = canonical_string(action, timestamp_ms, nonce, sha256_hex(body))
        signature = self._private_key.sign(message.encode("utf-8"))
        headers = {
            "content-type": "application/json",
            HEADER_DEVICE_ID: self._device_id,
            HEADER_TIMESTAMP: timestamp_ms,
            HEADER_NONCE: nonce,
            HEADER_SIGNATURE: base64.b64encode(signature).decode("ascii"),
        }
        return SignedRequest(body=body, headers=headers)

    def __repr__(self) -> str:  # pragma: no cover - never leak key material
        return f"DeviceIdentity(device_id={self._device_id!r})"


def verify_signature(public_key_b64: str, message: str, signature_b64: str) -> bool:
    """Verify a canonical string against a base64 raw public key.

    Present so the smoke test can prove the signature the agent produces is the
    one the backend would accept, without needing the live mesh.
    """
    try:
        public = Ed25519PublicKey.from_public_bytes(_b64decode_forgiving(public_key_b64, field="public key"))
        public.verify(_b64decode_forgiving(signature_b64, field="signature"), message.encode("utf-8"))
    except Exception:
        return False
    return True


def check_clock_skew(server_time_ms: int, *, now_ms: int | None = None) -> float:
    """Return skew in seconds, raising if it exceeds the backend's window.

    Called after the first successful heartbeat, which echoes ``server_time``.
    Colab VMs and Space containers occasionally drift, and a drifted clock
    presents as ``request_timestamp_out_of_window`` on every subsequent call —
    an error whose text does not obviously mean "your clock is wrong".
    """
    current = now_ms if now_ms is not None else int(time.time() * 1000)
    skew = (current - server_time_ms) / 1000.0
    if abs(skew) > CLOCK_SKEW_LIMIT_SECONDS:
        raise ConfigError(
            f"node clock is {skew:.0f}s from mesh time; the mesh rejects skew over "
            f"{CLOCK_SKEW_LIMIT_SECONDS}s. Fix the node clock before registering."
        )
    return skew


def _b64decode_forgiving(value: str, *, field: str) -> bytes:
    """Decode base64 tolerating URL-safe alphabet and missing padding."""
    if not isinstance(value, str) or not value.strip():
        raise ConfigError(f"{field} must be a non-empty base64 string")
    text = value.strip().replace("-", "+").replace("_", "/")
    text += "=" * (-len(text) % 4)
    try:
        return base64.b64decode(text, validate=False)
    except Exception as exc:
        raise ConfigError(f"{field} is not valid base64: {exc}") from exc


def load_identity_from_env(env: Mapping[str, str] | None = None) -> DeviceIdentity | None:
    """Build a ``DeviceIdentity`` from environment, or ``None`` if unconfigured.

    Returning ``None`` rather than raising lets the agent fall back to controller
    transport, which needs no device credentials and is what makes a same-day
    demo possible before a device has been paired.
    """
    source = env if env is not None else os.environ
    device_id = (source.get("THOX_MESH_DEVICE_ID") or "").strip()
    seed = (source.get("THOX_MESH_DEVICE_KEY") or "").strip()
    key_file = (source.get("THOX_MESH_DEVICE_KEY_FILE") or "").strip()
    if not device_id:
        return None
    if seed:
        return DeviceIdentity.from_base64_seed(device_id, seed)
    if key_file:
        return DeviceIdentity.from_pem_file(device_id, key_file)
    raise ConfigError(
        "THOX_MESH_DEVICE_ID is set but no key: provide THOX_MESH_DEVICE_KEY "
        "(base64 seed) or THOX_MESH_DEVICE_KEY_FILE (PEM path)"
    )