"""Typed Ed25519 motion permits with durable one-use replay protection.

This module is deliberately independent from HTTP, UI and xArm SDK code.  It
accepts no controller commands and exposes no permit-issuing capability.  The
runtime holds only public trust material; a protected external authority owns
the signing key.
"""

from __future__ import annotations

import base64
import binascii
import hashlib
import hmac
import json
import math
import os
import re
import sqlite3
import stat
import subprocess
import tempfile
import threading
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Mapping

PERMIT_SCHEMA = "labcobot.motion-permit.v1"
TRUST_SCHEMA = "labcobot.motion-permit-trust.v1"
SIGNATURE_PREFIX = "ed25519:"
MAX_PERMIT_BYTES = 32 * 1024
MAX_TRUST_BYTES = 64 * 1024
MAX_TRUST_KEYS = 32
MAX_DURATION_MS = 300_000
CLOCK_FUTURE_TOLERANCE_SECONDS = 2.0

_HEX_32 = re.compile(r"^[0-9a-f]{32}$")
_HEX_40 = re.compile(r"^[0-9a-f]{40}$")
_HEX_64 = re.compile(r"^[0-9a-f]{64}$")
_KEY_ID = re.compile(r"^[a-z0-9][a-z0-9._-]{0,63}$")
_DEVICE_UID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$")
_ROBOT_SERIAL = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,63}$")
_VERSION = re.compile(r"^[A-Za-z0-9][A-Za-z0-9.!+_-]{0,79}$")
_PUBLIC_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$")
_SCOPE = re.compile(r"^[a-z][a-z0-9_.:-]{0,63}$")
_TIMESTAMP = re.compile(
    r"^[0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}:[0-9]{2}\.[0-9]{6}Z$"
)

_UNSIGNED_FIELDS = frozenset(
    {
        "action",
        "authorization_epoch",
        "calibration_sha256",
        "device_uid",
        "expires_at",
        "issued_at",
        "maximum_duration_ms",
        "motion_primitive_sha256",
        "nonce",
        "operation_scope",
        "recipe_id",
        "release_tree_sha256",
        "release_version",
        "run_id",
        "runtime_instance_id",
        "safety_profile_sha256",
        "schema",
        "signing_key_id",
        "signing_key_version",
        "source_revision",
        "xarm_serial",
    }
)
_PERMIT_FIELDS = _UNSIGNED_FIELDS | {"signature"}


class MotionPermitError(RuntimeError):
    """A motion permit or its durable replay state is invalid."""


@dataclass(frozen=True)
class MotionIntent:
    """The fixed operation the caller is about to execute."""

    operation_scope: str
    recipe_id: str
    action: str
    run_id: str


@dataclass(frozen=True)
class MotionBindings:
    """Immutable device, robot, release and safety identities."""

    device_uid: str
    xarm_serial: str
    release_version: str
    source_revision: str
    release_tree_sha256: str
    safety_profile_sha256: str
    calibration_sha256: str
    motion_primitive_sha256: str


@dataclass(frozen=True)
class MotionPermitGrant:
    """A consumed permit's local monotonic command deadline."""

    permit_sha256: str
    nonce: str
    deadline_monotonic: float
    maximum_duration_ms: int


class ImmutableMotionBindingProvider:
    """Read current immutable release and commissioned safety identities.

    The active-release manifest is root-owned and verified before the service
    starts.  Reading it afresh at arming means a release switch, calibration,
    profile, primitive, device or robot change makes every older permit fail
    its exact binding check.
    """

    _RELEASE_FIELDS = frozenset(
        {
            "app_root",
            "application_tree_sha256",
            "publisher_trust_anchor_paths",
            "python_executable",
            "python_tree_sha256",
            "schema",
            "source_revision",
            "state_root",
            "version",
        }
    )

    def __init__(
        self,
        *,
        active_release_manifest_path: Path,
        expected_release_owner_uid: int,
        device_uid: str,
        xarm_serial_provider: Callable[[], str],
        safety_profile_sha256: str,
        calibration_sha256: str,
        motion_primitive_sha256: str,
    ) -> None:
        self.active_release_manifest_path = Path(active_release_manifest_path)
        self.expected_release_owner_uid = int(expected_release_owner_uid)
        self.device_uid = device_uid
        self.xarm_serial_provider = xarm_serial_provider
        self.safety_profile_sha256 = safety_profile_sha256
        self.calibration_sha256 = calibration_sha256
        self.motion_primitive_sha256 = motion_primitive_sha256

    def __call__(self) -> MotionBindings:
        payload = _secure_regular_file(
            self.active_release_manifest_path,
            expected_owner_uid=self.expected_release_owner_uid,
            minimum_size=1,
            maximum_size=16 * 1024,
            label="active release manifest",
        )
        try:
            release = json.loads(
                payload.decode("utf-8"),
                object_pairs_hook=_object_pairs,
                parse_constant=lambda _value: (_ for _ in ()).throw(
                    MotionPermitError("active release contains a non-finite value")
                ),
            )
        except (UnicodeError, json.JSONDecodeError) as exc:
            raise MotionPermitError("active release manifest is invalid") from exc
        if (
            not isinstance(release, Mapping)
            or set(release) != self._RELEASE_FIELDS
            or release["schema"] != "labcobot.immutable-release.v1"
        ):
            raise MotionPermitError("active release manifest fields are invalid")
        serial = self.xarm_serial_provider()
        bindings = MotionBindings(
            device_uid=self.device_uid,
            xarm_serial=serial,
            release_version=release["version"],
            source_revision=release["source_revision"],
            release_tree_sha256=release["application_tree_sha256"],
            safety_profile_sha256=self.safety_profile_sha256,
            calibration_sha256=self.calibration_sha256,
            motion_primitive_sha256=self.motion_primitive_sha256,
        )
        MotionPermitAuthority._validate_bindings(bindings)
        return bindings


def canonical_json_bytes(value: Mapping[str, Any]) -> bytes:
    try:
        return json.dumps(
            dict(value),
            allow_nan=False,
            ensure_ascii=False,
            separators=(",", ":"),
            sort_keys=True,
        ).encode("utf-8")
    except (TypeError, UnicodeEncodeError, ValueError) as exc:
        raise MotionPermitError("motion permit is not canonical JSON") from exc


def permit_signing_payload(permit: Mapping[str, Any]) -> bytes:
    """Return the exact canonical bytes covered by the signature."""

    if not isinstance(permit, Mapping) or set(permit) != _PERMIT_FIELDS:
        raise MotionPermitError("motion permit fields are invalid")
    return canonical_json_bytes({name: permit[name] for name in _UNSIGNED_FIELDS})


def canonical_timestamp(value: datetime) -> str:
    if value.tzinfo is None:
        raise MotionPermitError("motion permit timestamp lacks a timezone")
    return (
        value.astimezone(timezone.utc)
        .isoformat(timespec="microseconds")
        .replace("+00:00", "Z")
    )


def _parse_timestamp(value: Any) -> datetime:
    if not isinstance(value, str) or _TIMESTAMP.fullmatch(value) is None:
        raise MotionPermitError("motion permit timestamp is invalid")
    try:
        parsed = datetime.strptime(value, "%Y-%m-%dT%H:%M:%S.%fZ")
    except ValueError as exc:
        raise MotionPermitError("motion permit timestamp is invalid") from exc
    return parsed.replace(tzinfo=timezone.utc)


def _object_pairs(pairs):
    value = {}
    for key, item in pairs:
        if key in value:
            raise MotionPermitError("motion permit contains a duplicate field")
        value[key] = item
    return value


def parse_motion_permit(value: Any) -> Mapping[str, Any]:
    """Parse only a bounded exact mapping or strict UTF-8 JSON object."""

    if isinstance(value, Mapping):
        parsed = dict(value)
    elif isinstance(value, (bytes, bytearray, str)):
        raw = value.encode("utf-8") if isinstance(value, str) else bytes(value)
        if not 1 <= len(raw) <= MAX_PERMIT_BYTES:
            raise MotionPermitError("motion permit size is invalid")
        try:
            parsed = json.loads(
                raw.decode("utf-8"),
                object_pairs_hook=_object_pairs,
                parse_constant=lambda _value: (_ for _ in ()).throw(
                    MotionPermitError("motion permit contains a non-finite value")
                ),
            )
        except (UnicodeError, json.JSONDecodeError) as exc:
            raise MotionPermitError("motion permit JSON is invalid") from exc
    else:
        raise MotionPermitError("motion permit type is invalid")
    if not isinstance(parsed, Mapping) or set(parsed) != _PERMIT_FIELDS:
        raise MotionPermitError("motion permit fields are invalid")
    return dict(parsed)


def _secure_regular_file(
    path: Path,
    *,
    expected_owner_uid: int,
    minimum_size: int,
    maximum_size: int,
    label: str,
) -> bytes:
    try:
        metadata = path.lstat()
        descriptor = os.open(
            str(path),
            os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0),
        )
    except OSError as exc:
        raise MotionPermitError("{} is unavailable".format(label)) from exc
    try:
        opened = os.fstat(descriptor)
        if (
            stat.S_ISLNK(metadata.st_mode)
            or not stat.S_ISREG(metadata.st_mode)
            or not stat.S_ISREG(opened.st_mode)
            or metadata.st_dev != opened.st_dev
            or metadata.st_ino != opened.st_ino
            or opened.st_uid != expected_owner_uid
            or opened.st_mode & 0o022
            or not minimum_size <= opened.st_size <= maximum_size
        ):
            raise MotionPermitError("{} is unsafe".format(label))
        chunks = []
        remaining = maximum_size + 1
        while remaining:
            chunk = os.read(descriptor, min(64 * 1024, remaining))
            if not chunk:
                break
            chunks.append(chunk)
            remaining -= len(chunk)
        payload = b"".join(chunks)
        if not minimum_size <= len(payload) <= maximum_size:
            raise MotionPermitError("{} size is invalid".format(label))
        return payload
    finally:
        os.close(descriptor)


def _trusted_public_key(
    trust_store_path: Path,
    *,
    key_id: str,
    key_version: int,
    expected_owner_uid: int,
) -> Path:
    payload = _secure_regular_file(
        trust_store_path,
        expected_owner_uid=expected_owner_uid,
        minimum_size=1,
        maximum_size=MAX_TRUST_BYTES,
        label="motion permit trust store",
    )
    try:
        trust = json.loads(
            payload.decode("utf-8"),
            object_pairs_hook=_object_pairs,
            parse_constant=lambda _value: (_ for _ in ()).throw(
                MotionPermitError("motion permit trust contains a non-finite value")
            ),
        )
    except (UnicodeError, json.JSONDecodeError) as exc:
        raise MotionPermitError("motion permit trust store is invalid") from exc
    if not isinstance(trust, Mapping) or set(trust) != {
        "keys",
        "minimum_key_version",
        "schema",
    }:
        raise MotionPermitError("motion permit trust fields are invalid")
    minimum_version = trust["minimum_key_version"]
    keys = trust["keys"]
    if (
        trust["schema"] != TRUST_SCHEMA
        or isinstance(minimum_version, bool)
        or not isinstance(minimum_version, int)
        or minimum_version < 1
        or not isinstance(keys, list)
        or not 1 <= len(keys) <= MAX_TRUST_KEYS
    ):
        raise MotionPermitError("motion permit trust policy is invalid")
    identities = set()
    selected = None
    for entry in keys:
        if not isinstance(entry, Mapping) or set(entry) != {
            "algorithm",
            "key_id",
            "key_version",
            "public_key_path",
            "public_key_sha256",
            "status",
        }:
            raise MotionPermitError("motion permit trust key fields are invalid")
        identity = (entry["key_id"], entry["key_version"])
        if identity in identities:
            raise MotionPermitError("motion permit trust key is duplicated")
        identities.add(identity)
        if (
            not isinstance(entry["key_id"], str)
            or _KEY_ID.fullmatch(entry["key_id"]) is None
            or isinstance(entry["key_version"], bool)
            or not isinstance(entry["key_version"], int)
            or not 1 <= entry["key_version"] <= 2_147_483_647
            or entry["algorithm"] != "ed25519"
            or entry["status"] not in {"active", "revoked"}
            or not isinstance(entry["public_key_path"], str)
            or not isinstance(entry["public_key_sha256"], str)
            or _HEX_64.fullmatch(entry["public_key_sha256"]) is None
        ):
            raise MotionPermitError("motion permit trust key policy is invalid")
        if identity == (key_id, key_version):
            selected = entry
    if selected is None:
        raise MotionPermitError("motion permit signing key is unknown")
    if key_version < minimum_version or selected["status"] != "active":
        raise MotionPermitError("motion permit signing key is revoked")
    public_key = Path(selected["public_key_path"])
    if not public_key.is_absolute() or ".." in public_key.parts:
        raise MotionPermitError("motion permit public key path is unsafe")
    key_payload = _secure_regular_file(
        public_key,
        expected_owner_uid=expected_owner_uid,
        minimum_size=64,
        maximum_size=16 * 1024,
        label="motion permit public key",
    )
    if not hmac.compare_digest(
        hashlib.sha256(key_payload).hexdigest(),
        selected["public_key_sha256"],
    ):
        raise MotionPermitError("motion permit public key digest mismatch")
    return public_key


def _decode_signature(value: Any) -> bytes:
    if not isinstance(value, str) or not value.startswith(SIGNATURE_PREFIX):
        raise MotionPermitError("motion permit signature algorithm is invalid")
    encoded = value[len(SIGNATURE_PREFIX) :]
    try:
        raw = base64.b64decode(encoded, validate=True)
    except (binascii.Error, ValueError) as exc:
        raise MotionPermitError("motion permit signature encoding is invalid") from exc
    if len(raw) != 64 or base64.b64encode(raw).decode("ascii") != encoded:
        raise MotionPermitError("motion permit signature encoding is invalid")
    return raw


def _verify_ed25519(
    payload: bytes,
    signature: Any,
    *,
    public_key_path: Path,
    openssl_binary: Path,
    expected_openssl_owner_uid: int,
) -> None:
    executable = _secure_regular_file(
        openssl_binary,
        expected_owner_uid=expected_openssl_owner_uid,
        minimum_size=1,
        maximum_size=16 * 1024 * 1024,
        label="OpenSSL binary",
    )
    del executable
    if not os.access(openssl_binary, os.X_OK):
        raise MotionPermitError("OpenSSL binary is not executable")
    raw_signature = _decode_signature(signature)
    payload_fd, payload_name = tempfile.mkstemp(
        prefix=".motion-permit-", suffix=".json"
    )
    signature_fd, signature_name = tempfile.mkstemp(
        prefix=".motion-permit-", suffix=".sig"
    )
    try:
        with os.fdopen(payload_fd, "wb") as handle:
            handle.write(payload)
            handle.flush()
            os.fsync(handle.fileno())
        with os.fdopen(signature_fd, "wb") as handle:
            handle.write(raw_signature)
            handle.flush()
            os.fsync(handle.fileno())
        os.chmod(payload_name, 0o600)
        os.chmod(signature_name, 0o600)
        try:
            result = subprocess.run(
                [
                    str(openssl_binary),
                    "pkeyutl",
                    "-verify",
                    "-pubin",
                    "-inkey",
                    str(public_key_path),
                    "-rawin",
                    "-in",
                    payload_name,
                    "-sigfile",
                    signature_name,
                ],
                check=False,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                timeout=5,
            )
        except (OSError, subprocess.SubprocessError) as exc:
            raise MotionPermitError("motion permit signature check failed") from exc
        if result.returncode != 0:
            raise MotionPermitError("motion permit signature is invalid")
    finally:
        Path(payload_name).unlink(missing_ok=True)
        Path(signature_name).unlink(missing_ok=True)


class MotionPermitLedger:
    """SQLite-backed boot/epoch/nonce state committed before motor enable."""

    def __init__(
        self,
        path: Path,
        *,
        expected_owner_uid: int | None = None,
        runtime_instance_id: str | None = None,
    ) -> None:
        self.path = Path(path)
        self.expected_owner_uid = (
            os.geteuid() if expected_owner_uid is None else int(expected_owner_uid)
        )
        self._lock = threading.Lock()
        instance = runtime_instance_id or uuid.uuid4().hex
        if _HEX_32.fullmatch(instance) is None:
            raise MotionPermitError("runtime instance identifier is invalid")
        self._prepare_parent()
        self._initialize(instance)

    def _prepare_parent(self) -> None:
        try:
            self.path.parent.mkdir(parents=True, mode=0o700, exist_ok=True)
            metadata = self.path.parent.lstat()
        except OSError as exc:
            raise MotionPermitError(
                "motion permit state directory is unavailable"
            ) from exc
        if (
            stat.S_ISLNK(metadata.st_mode)
            or not stat.S_ISDIR(metadata.st_mode)
            or metadata.st_uid != self.expected_owner_uid
            or metadata.st_mode & 0o077
        ):
            raise MotionPermitError("motion permit state directory is unsafe")
        if self.path.exists() or self.path.is_symlink():
            metadata = self.path.lstat()
            if (
                stat.S_ISLNK(metadata.st_mode)
                or not stat.S_ISREG(metadata.st_mode)
                or metadata.st_uid != self.expected_owner_uid
                or metadata.st_mode & 0o077
            ):
                raise MotionPermitError("motion permit ledger is unsafe")

    def _connect(self, *, allow_create: bool = False) -> sqlite3.Connection:
        if self.path.exists() or self.path.is_symlink():
            metadata = self.path.lstat()
            if (
                stat.S_ISLNK(metadata.st_mode)
                or not stat.S_ISREG(metadata.st_mode)
                or metadata.st_uid != self.expected_owner_uid
                or metadata.st_mode & 0o077
            ):
                raise MotionPermitError("motion permit ledger is unsafe")
        elif not allow_create:
            raise MotionPermitError("motion permit ledger is unavailable")
        connection = sqlite3.connect(str(self.path), timeout=5.0)
        connection.execute("PRAGMA foreign_keys=ON")
        connection.execute("PRAGMA synchronous=FULL")
        connection.execute("PRAGMA journal_mode=DELETE")
        return connection

    def _initialize(self, instance: str) -> None:
        with self._lock:
            connection = self._connect(allow_create=True)
            try:
                connection.execute("BEGIN IMMEDIATE")
                connection.execute(
                    "CREATE TABLE IF NOT EXISTS runtime_state ("
                    "singleton INTEGER PRIMARY KEY CHECK(singleton=1), "
                    "runtime_instance_id TEXT NOT NULL, "
                    "authorization_epoch INTEGER NOT NULL CHECK(authorization_epoch>=0))"
                )
                connection.execute(
                    "CREATE TABLE IF NOT EXISTS consumed_permits ("
                    "nonce TEXT PRIMARY KEY, permit_sha256 TEXT NOT NULL UNIQUE, "
                    "runtime_instance_id TEXT NOT NULL, authorization_epoch INTEGER NOT NULL, "
                    "consumed_at TEXT NOT NULL)"
                )
                connection.execute(
                    "CREATE TABLE IF NOT EXISTS invalidations ("
                    "sequence INTEGER PRIMARY KEY AUTOINCREMENT, "
                    "runtime_instance_id TEXT NOT NULL, authorization_epoch INTEGER NOT NULL, "
                    "reason TEXT NOT NULL, invalidated_at TEXT NOT NULL)"
                )
                connection.execute(
                    "INSERT INTO runtime_state(singleton,runtime_instance_id,authorization_epoch) "
                    "VALUES(1,?,0) ON CONFLICT(singleton) DO UPDATE SET "
                    "runtime_instance_id=excluded.runtime_instance_id, authorization_epoch=0",
                    (instance,),
                )
                connection.commit()
            except Exception as exc:
                connection.rollback()
                raise MotionPermitError(
                    "motion permit ledger initialization failed"
                ) from exc
            finally:
                connection.close()
        os.chmod(self.path, 0o600, follow_symlinks=False)
        directory_descriptor = os.open(
            str(self.path.parent),
            os.O_RDONLY | getattr(os, "O_CLOEXEC", 0),
        )
        try:
            os.fsync(directory_descriptor)
        finally:
            os.close(directory_descriptor)

    def context(self) -> tuple[str, int]:
        with self._lock:
            connection = self._connect()
            try:
                row = connection.execute(
                    "SELECT runtime_instance_id,authorization_epoch FROM runtime_state "
                    "WHERE singleton=1"
                ).fetchone()
            finally:
                connection.close()
        if (
            not isinstance(row, tuple)
            or len(row) != 2
            or _HEX_32.fullmatch(str(row[0])) is None
            or isinstance(row[1], bool)
            or not isinstance(row[1], int)
            or row[1] < 0
        ):
            raise MotionPermitError("motion permit ledger state is invalid")
        return str(row[0]), int(row[1])

    def invalidate(self, reason: str, *, now: datetime) -> tuple[str, int]:
        if not isinstance(reason, str) or _SCOPE.fullmatch(reason) is None:
            raise MotionPermitError("motion permit invalidation reason is invalid")
        with self._lock:
            connection = self._connect()
            try:
                connection.execute("BEGIN IMMEDIATE")
                row = connection.execute(
                    "SELECT runtime_instance_id,authorization_epoch FROM runtime_state "
                    "WHERE singleton=1"
                ).fetchone()
                if row is None:
                    raise MotionPermitError("motion permit ledger state is missing")
                epoch = int(row[1]) + 1
                connection.execute(
                    "UPDATE runtime_state SET authorization_epoch=? WHERE singleton=1",
                    (epoch,),
                )
                connection.execute(
                    "INSERT INTO invalidations(runtime_instance_id,authorization_epoch,reason,invalidated_at) "
                    "VALUES(?,?,?,?)",
                    (row[0], epoch, reason, canonical_timestamp(now)),
                )
                connection.commit()
                return str(row[0]), epoch
            except MotionPermitError:
                connection.rollback()
                raise
            except Exception as exc:
                connection.rollback()
                raise MotionPermitError("motion permit invalidation failed") from exc
            finally:
                connection.close()

    def consume(
        self,
        *,
        nonce: str,
        permit_sha256: str,
        runtime_instance_id: str,
        authorization_epoch: int,
        now: datetime,
    ) -> None:
        if (
            _HEX_32.fullmatch(nonce) is None
            or _HEX_64.fullmatch(permit_sha256) is None
            or _HEX_32.fullmatch(runtime_instance_id) is None
            or isinstance(authorization_epoch, bool)
            or not isinstance(authorization_epoch, int)
            or authorization_epoch < 0
        ):
            raise MotionPermitError("motion permit replay identity is invalid")
        with self._lock:
            connection = self._connect()
            try:
                connection.execute("BEGIN IMMEDIATE")
                row = connection.execute(
                    "SELECT runtime_instance_id,authorization_epoch FROM runtime_state "
                    "WHERE singleton=1"
                ).fetchone()
                if row != (runtime_instance_id, authorization_epoch):
                    raise MotionPermitError("motion permit runtime epoch is stale")
                connection.execute(
                    "INSERT INTO consumed_permits(nonce,permit_sha256,runtime_instance_id,"
                    "authorization_epoch,consumed_at) VALUES(?,?,?,?,?)",
                    (
                        nonce,
                        permit_sha256,
                        runtime_instance_id,
                        authorization_epoch,
                        canonical_timestamp(now),
                    ),
                )
                connection.commit()
            except sqlite3.IntegrityError as exc:
                connection.rollback()
                raise MotionPermitError("motion permit was already consumed") from exc
            except MotionPermitError:
                connection.rollback()
                raise
            except Exception as exc:
                connection.rollback()
                raise MotionPermitError("motion permit consumption failed") from exc
            finally:
                connection.close()


class MotionPermitAuthority:
    """Verify and durably consume signed permits; never issue them."""

    def __init__(
        self,
        *,
        trust_store_path: Path,
        ledger: MotionPermitLedger,
        openssl_binary: Path = Path("/usr/bin/openssl"),
        expected_trust_owner_uid: int = 0,
        expected_openssl_owner_uid: int = 0,
        wall_clock: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
        monotonic: Callable[[], float] = __import__("time").monotonic,
    ) -> None:
        self.trust_store_path = Path(trust_store_path)
        self.ledger = ledger
        self.openssl_binary = Path(openssl_binary)
        self.expected_trust_owner_uid = int(expected_trust_owner_uid)
        self.expected_openssl_owner_uid = int(expected_openssl_owner_uid)
        self.wall_clock = wall_clock
        self.monotonic = monotonic

    def runtime_context(self) -> tuple[str, int]:
        return self.ledger.context()

    def invalidate(self, reason: str) -> tuple[str, int]:
        return self.ledger.invalidate(reason, now=self._now())

    def _now(self) -> datetime:
        value = self.wall_clock()
        if not isinstance(value, datetime) or value.tzinfo is None:
            raise MotionPermitError("motion permit wall clock is invalid")
        return value.astimezone(timezone.utc)

    def verify_and_consume(
        self,
        permit: Any,
        *,
        intent: MotionIntent,
        bindings: MotionBindings,
    ) -> MotionPermitGrant:
        value = parse_motion_permit(permit)
        if value["schema"] != PERMIT_SCHEMA:
            raise MotionPermitError("motion permit schema is invalid")
        self._validate_intent(intent)
        self._validate_bindings(bindings)
        runtime_instance_id, authorization_epoch = self.ledger.context()
        expected = {
            "action": intent.action,
            "authorization_epoch": authorization_epoch,
            "calibration_sha256": bindings.calibration_sha256,
            "device_uid": bindings.device_uid,
            "motion_primitive_sha256": bindings.motion_primitive_sha256,
            "operation_scope": intent.operation_scope,
            "recipe_id": intent.recipe_id,
            "release_tree_sha256": bindings.release_tree_sha256,
            "release_version": bindings.release_version,
            "run_id": intent.run_id,
            "runtime_instance_id": runtime_instance_id,
            "safety_profile_sha256": bindings.safety_profile_sha256,
            "source_revision": bindings.source_revision,
            "xarm_serial": bindings.xarm_serial,
        }
        for name, expected_value in expected.items():
            supplied = value[name]
            if type(supplied) is not type(expected_value) or supplied != expected_value:
                raise MotionPermitError(
                    "motion permit {} binding is invalid".format(name)
                )
        nonce = value["nonce"]
        key_id = value["signing_key_id"]
        key_version = value["signing_key_version"]
        duration = value["maximum_duration_ms"]
        if (
            not isinstance(nonce, str)
            or _HEX_32.fullmatch(nonce) is None
            or not isinstance(key_id, str)
            or _KEY_ID.fullmatch(key_id) is None
            or isinstance(key_version, bool)
            or not isinstance(key_version, int)
            or not 1 <= key_version <= 2_147_483_647
            or isinstance(duration, bool)
            or not isinstance(duration, int)
            or not 1 <= duration <= MAX_DURATION_MS
        ):
            raise MotionPermitError("motion permit identity or duration is invalid")
        issued_at = _parse_timestamp(value["issued_at"])
        expires_at = _parse_timestamp(value["expires_at"])
        interval_ms = int((expires_at - issued_at).total_seconds() * 1000)
        now = self._now()
        if (
            interval_ms != duration
            or issued_at.timestamp() > now.timestamp() + CLOCK_FUTURE_TOLERANCE_SECONDS
            or expires_at <= now
        ):
            raise MotionPermitError("motion permit is expired or has invalid timing")
        payload = permit_signing_payload(value)
        permit_sha256 = hashlib.sha256(canonical_json_bytes(value)).hexdigest()
        public_key = _trusted_public_key(
            self.trust_store_path,
            key_id=key_id,
            key_version=key_version,
            expected_owner_uid=self.expected_trust_owner_uid,
        )
        _verify_ed25519(
            payload,
            value["signature"],
            public_key_path=public_key,
            openssl_binary=self.openssl_binary,
            expected_openssl_owner_uid=self.expected_openssl_owner_uid,
        )
        # Signature verification can be slow on a loaded edge device. Recheck
        # time immediately before committing the durable one-use record.
        consumed_at = self._now()
        remaining = (expires_at - consumed_at).total_seconds()
        if remaining <= 0.0 or not math.isfinite(remaining):
            raise MotionPermitError("motion permit expired during verification")
        self.ledger.consume(
            nonce=nonce,
            permit_sha256=permit_sha256,
            runtime_instance_id=runtime_instance_id,
            authorization_epoch=authorization_epoch,
            now=consumed_at,
        )
        return MotionPermitGrant(
            permit_sha256=permit_sha256,
            nonce=nonce,
            deadline_monotonic=float(self.monotonic()) + remaining,
            maximum_duration_ms=duration,
        )

    @staticmethod
    def _validate_intent(intent: MotionIntent) -> None:
        if not isinstance(intent, MotionIntent):
            raise MotionPermitError("motion intent type is invalid")
        for name in ("recipe_id", "action", "run_id"):
            value = getattr(intent, name)
            if not isinstance(value, str) or _PUBLIC_ID.fullmatch(value) is None:
                raise MotionPermitError("motion intent {} is invalid".format(name))
        if (
            not isinstance(intent.operation_scope, str)
            or _SCOPE.fullmatch(intent.operation_scope) is None
        ):
            raise MotionPermitError("motion operation scope is invalid")

    @staticmethod
    def _validate_bindings(bindings: MotionBindings) -> None:
        if not isinstance(bindings, MotionBindings):
            raise MotionPermitError("motion binding type is invalid")
        values = (
            bindings.device_uid,
            bindings.xarm_serial,
            bindings.release_version,
            bindings.source_revision,
            bindings.release_tree_sha256,
            bindings.safety_profile_sha256,
            bindings.calibration_sha256,
            bindings.motion_primitive_sha256,
        )
        if (
            any(not isinstance(value, str) for value in values)
            or _DEVICE_UID.fullmatch(bindings.device_uid) is None
            or _ROBOT_SERIAL.fullmatch(bindings.xarm_serial) is None
            or _VERSION.fullmatch(bindings.release_version) is None
            or _HEX_40.fullmatch(bindings.source_revision) is None
            or any(
                _HEX_64.fullmatch(value) is None
                for value in (
                    bindings.release_tree_sha256,
                    bindings.safety_profile_sha256,
                    bindings.calibration_sha256,
                    bindings.motion_primitive_sha256,
                )
            )
        ):
            raise MotionPermitError("motion permit immutable bindings are invalid")


__all__ = [
    "ImmutableMotionBindingProvider",
    "MAX_DURATION_MS",
    "MotionBindings",
    "MotionIntent",
    "MotionPermitAuthority",
    "MotionPermitError",
    "MotionPermitGrant",
    "MotionPermitLedger",
    "PERMIT_SCHEMA",
    "TRUST_SCHEMA",
    "canonical_json_bytes",
    "canonical_timestamp",
    "parse_motion_permit",
    "permit_signing_payload",
]
