"""Fail-closed commissioning evidence for unattended OTA installation.

The artifact validated here is deliberately not a robot-motion permit.  It
only lets the existing guarded OTA scheduler attempt its normal admission
checks.  Manual update checks, Universal STOP, and every robot-control gate
remain independent.
"""

from __future__ import annotations

import hashlib
import hmac
import json
import os
import re
import stat
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Mapping, Optional, Tuple

COMMISSIONING_SCHEMA = "labcobot.ota-auto-apply-commissioning.v1"
MAX_ARTIFACT_BYTES = 64 * 1024
ROOT_OWNER_UID = 0
REQUIRED_P0_ISSUES = (298, 299, 300, 301, 302, 303, 308, 311, 312, 315)
_SHA256 = re.compile(r"[0-9a-f]{64}")
_SOURCE_REVISION = re.compile(r"(?:[0-9a-f]{40}|[0-9a-f]{64})")
_VERSION = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,99}")
_UTC_TIMESTAMP = re.compile(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z")

COMMISSIONING_BLOCKERS = (
    "rollback_switch_active",
    "commissioning_path_missing",
    "commissioning_path_unsafe",
    "commissioning_digest_missing",
    "commissioning_digest_mismatch",
    "commissioning_artifact_invalid",
    "commissioning_expired",
    "commissioning_device_mismatch",
    "commissioning_release_mismatch",
    "commissioning_camera_mismatch",
    "commissioning_controller_mismatch",
    "commissioning_boot_supervisor_mismatch",
    "commissioning_filesystem_mismatch",
    "commissioning_lvgl_mismatch",
    "commissioning_evidence_incomplete",
)


class OTACommissioningError(ValueError):
    """The commissioning artifact is unsafe or internally malformed."""


@dataclass(frozen=True)
class OTACommissioningGate:
    """Bounded, secret-free result consumed by settings and diagnostics."""

    requested: bool
    permitted: bool
    blockers: Tuple[str, ...] = ()
    artifact_sha256: str = ""

    def __post_init__(self) -> None:
        if not isinstance(self.requested, bool) or not isinstance(self.permitted, bool):
            raise TypeError("commissioning gate flags must be booleans")
        blockers = tuple(self.blockers)
        if (
            len(blockers) > len(COMMISSIONING_BLOCKERS)
            or len(set(blockers)) != len(blockers)
            or any(value not in COMMISSIONING_BLOCKERS for value in blockers)
        ):
            raise ValueError("commissioning blockers are invalid")
        if blockers != tuple(
            value for value in COMMISSIONING_BLOCKERS if value in blockers
        ):
            raise ValueError("commissioning blockers are not canonical")
        if self.permitted and (blockers or not self.requested):
            raise ValueError("permitted commissioning gate is inconsistent")
        if self.artifact_sha256 and _SHA256.fullmatch(self.artifact_sha256) is None:
            raise ValueError("commissioning artifact digest is invalid")


def _reject_constant(value: str) -> None:
    raise OTACommissioningError("non-finite JSON is forbidden: {}".format(value))


def _object_pairs(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise OTACommissioningError("duplicate JSON field")
        result[key] = value
    return result


def _canonical_bytes(value: Any) -> bytes:
    return (
        json.dumps(
            value,
            ensure_ascii=False,
            allow_nan=False,
            separators=(",", ":"),
            sort_keys=True,
        )
        + "\n"
    ).encode("utf-8")


def _exact_mapping(value: Any, fields: set[str], label: str) -> Mapping[str, Any]:
    if not isinstance(value, Mapping) or set(value) != fields:
        raise OTACommissioningError("{} fields are invalid".format(label))
    return value


def _text(value: Any, label: str, *, maximum: int = 256) -> str:
    if (
        not isinstance(value, str)
        or not value
        or value != value.strip()
        or len(value) > maximum
        or any(ord(character) < 32 or ord(character) == 127 for character in value)
    ):
        raise OTACommissioningError("{} is invalid".format(label))
    return value


def _sha256(value: Any, label: str) -> str:
    if not isinstance(value, str) or _SHA256.fullmatch(value) is None:
        raise OTACommissioningError("{} is invalid".format(label))
    return value


def _timestamp(value: Any, label: str) -> datetime:
    if not isinstance(value, str) or _UTC_TIMESTAMP.fullmatch(value) is None:
        raise OTACommissioningError("{} is invalid".format(label))
    try:
        return datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ").replace(
            tzinfo=timezone.utc
        )
    except ValueError as exc:
        raise OTACommissioningError("{} is invalid".format(label)) from exc


def _read_root_file(
    path: Path,
    *,
    maximum: int,
    expected_owner_uid: int,
) -> bytes:
    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(str(path), flags)
    try:
        metadata = os.fstat(descriptor)
        if (
            not stat.S_ISREG(metadata.st_mode)
            or metadata.st_uid != expected_owner_uid
            or metadata.st_nlink != 1
            or metadata.st_mode & 0o022
            or metadata.st_size <= 0
            or metadata.st_size > maximum
        ):
            raise OTACommissioningError("commissioning file metadata is unsafe")
        chunks = []
        remaining = maximum + 1
        while remaining > 0:
            chunk = os.read(descriptor, min(remaining, 64 * 1024))
            if not chunk:
                break
            chunks.append(chunk)
            remaining -= len(chunk)
        payload = b"".join(chunks)
        if len(payload) != metadata.st_size or not payload or len(payload) > maximum:
            raise OTACommissioningError("commissioning file changed while reading")
        final_metadata = os.fstat(descriptor)
        stable_fields = (
            "st_dev",
            "st_ino",
            "st_mode",
            "st_nlink",
            "st_uid",
            "st_gid",
            "st_size",
            "st_mtime_ns",
        )
        if any(
            getattr(final_metadata, name) != getattr(metadata, name)
            for name in stable_fields
        ):
            raise OTACommissioningError("commissioning file changed while reading")
        return payload
    finally:
        os.close(descriptor)


def _regular_file_sha256(path: Path, *, expected_owner_uid: int) -> str:
    return hashlib.sha256(
        _read_root_file(
            path,
            maximum=4 * 1024 * 1024,
            expected_owner_uid=expected_owner_uid,
        )
    ).hexdigest()


def _load_artifact(
    path: Path,
    pinned_sha256: str,
    *,
    expected_owner_uid: int,
) -> tuple[Mapping[str, Any], str]:
    payload = _read_root_file(
        path,
        maximum=MAX_ARTIFACT_BYTES,
        expected_owner_uid=expected_owner_uid,
    )
    digest = hashlib.sha256(payload).hexdigest()
    if not hmac.compare_digest(digest, pinned_sha256):
        raise OTACommissioningError("commissioning artifact digest mismatch")
    try:
        text = payload.decode("utf-8")
    except UnicodeDecodeError as exc:
        raise OTACommissioningError("commissioning artifact is not UTF-8") from exc
    try:
        artifact = json.loads(
            text,
            object_pairs_hook=_object_pairs,
            parse_constant=_reject_constant,
        )
    except json.JSONDecodeError as exc:
        raise OTACommissioningError("commissioning artifact is malformed") from exc
    if not isinstance(artifact, Mapping) or _canonical_bytes(artifact) != payload:
        raise OTACommissioningError("commissioning artifact is not canonical")
    return artifact, digest


def _installed_source_revision(settings: Any, *, expected_owner_uid: int) -> str:
    payload = _read_root_file(
        Path(settings.install_root) / ".labcobot-source-revision",
        maximum=128,
        expected_owner_uid=expected_owner_uid,
    )
    try:
        value = payload.decode("ascii").strip()
    except UnicodeDecodeError as exc:
        raise OTACommissioningError("installed source revision is invalid") from exc
    if (
        payload != (value + "\n").encode("ascii")
        or _SOURCE_REVISION.fullmatch(value) is None
    ):
        raise OTACommissioningError("installed source revision is invalid")
    return value


def _semantic_blockers(
    artifact: Mapping[str, Any],
    settings: Any,
    *,
    now: datetime,
    expected_owner_uid: int,
) -> Tuple[str, ...]:
    root = _exact_mapping(
        artifact,
        {
            "approval",
            "boot_supervisor",
            "cameras",
            "controller",
            "device",
            "filesystem_mount_profile",
            "motion_authority",
            "ota_scheduler_admission",
            "p0_evidence",
            "release",
            "schema",
            "touchscreen_lvgl",
        },
        "commissioning artifact",
    )
    if (
        root["schema"] != COMMISSIONING_SCHEMA
        or root["motion_authority"] != "none"
        or root["ota_scheduler_admission"] is not True
    ):
        raise OTACommissioningError("commissioning authority is invalid")

    blockers = set()
    device = _exact_mapping(root["device"], {"uid"}, "device")
    if _text(device["uid"], "device uid", maximum=128) != settings.device_uid:
        blockers.add("commissioning_device_mismatch")

    release = _exact_mapping(root["release"], {"source_revision", "version"}, "release")
    source_revision = _text(release["source_revision"], "source revision", maximum=64)
    version = _text(release["version"], "release version", maximum=100)
    if (
        _SOURCE_REVISION.fullmatch(source_revision) is None
        or _VERSION.fullmatch(version) is None
    ):
        raise OTACommissioningError("release binding is invalid")
    try:
        installed_source = _installed_source_revision(
            settings, expected_owner_uid=expected_owner_uid
        )
    except (OSError, OTACommissioningError):
        installed_source = ""
    if source_revision != installed_source or version != settings.current_version:
        blockers.add("commissioning_release_mismatch")

    cameras = _exact_mapping(root["cameras"], {"tool", "workspace"}, "cameras")
    camera_values = (
        (
            "tool",
            settings.tool_camera_device_id,
            settings.tool_camera_source_id,
            settings.eye_in_hand_calibration_sha256,
        ),
        (
            "workspace",
            settings.workspace_camera_device_id,
            settings.workspace_camera_source_id,
            settings.workspace_depth_calibration_sha256,
        ),
    )
    for role, device_id, source_id, calibration_sha256 in camera_values:
        binding = _exact_mapping(
            cameras[role],
            {"calibration_sha256", "device_id", "source_id"},
            "{} camera".format(role),
        )
        if (
            _text(binding["device_id"], "camera device id", maximum=128) != device_id
            or _text(binding["source_id"], "camera source id", maximum=128) != source_id
            or _sha256(binding["calibration_sha256"], "camera calibration")
            != calibration_sha256
        ):
            blockers.add("commissioning_camera_mismatch")

    controller = _exact_mapping(
        root["controller"],
        {
            "axis_count",
            "device_type",
            "emergency_stop_verified",
            "model",
            "stopped_state_procedure_sha256",
            "stopped_state_verified",
        },
        "controller",
    )
    stopped_procedure_sha256 = _sha256(
        controller["stopped_state_procedure_sha256"],
        "controller stopped-state procedure",
    )
    if (
        controller["model"] != "lite6"
        or controller["axis_count"] != 6
        or controller["device_type"] != 9
        or controller["emergency_stop_verified"] is not True
        or controller["stopped_state_verified"] is not True
        or settings.xarm_model != "lite6"
        or not settings.xarm_metrics_enabled
        or stopped_procedure_sha256 != settings.ota_controller_stopped_procedure_sha256
    ):
        blockers.add("commissioning_controller_mismatch")

    boot = _exact_mapping(root["boot_supervisor"], {"sha256"}, "boot supervisor")
    expected_boot_sha = _sha256(boot["sha256"], "boot supervisor digest")
    try:
        actual_boot_sha = _regular_file_sha256(
            Path(settings.ota_boot_supervisor_path),
            expected_owner_uid=expected_owner_uid,
        )
    except (OSError, OTACommissioningError):
        actual_boot_sha = ""
    if expected_boot_sha != actual_boot_sha:
        blockers.add("commissioning_boot_supervisor_mismatch")

    filesystem = _exact_mapping(
        root["filesystem_mount_profile"], {"sha256"}, "filesystem mount profile"
    )
    filesystem_sha256 = _sha256(filesystem["sha256"], "filesystem mount profile digest")
    if filesystem_sha256 != settings.ota_filesystem_mount_profile_sha256:
        blockers.add("commissioning_filesystem_mismatch")

    lvgl = _exact_mapping(
        root["touchscreen_lvgl"],
        {"build_info_sha256", "test_evidence_sha256"},
        "touchscreen LVGL",
    )
    expected_lvgl_sha = _sha256(lvgl["build_info_sha256"], "LVGL build digest")
    _sha256(lvgl["test_evidence_sha256"], "LVGL test evidence digest")
    try:
        actual_lvgl_sha = _regular_file_sha256(
            Path(settings.ota_lvgl_build_info_path),
            expected_owner_uid=expected_owner_uid,
        )
    except (OSError, OTACommissioningError):
        actual_lvgl_sha = ""
    if expected_lvgl_sha != actual_lvgl_sha:
        blockers.add("commissioning_lvgl_mismatch")

    p0_evidence = root["p0_evidence"]
    if not isinstance(p0_evidence, list) or len(p0_evidence) != len(REQUIRED_P0_ISSUES):
        raise OTACommissioningError("P0 evidence is incomplete")
    evidence_issues = []
    for evidence in p0_evidence:
        item = _exact_mapping(
            evidence, {"issue", "test_evidence_sha256"}, "P0 evidence"
        )
        if isinstance(item["issue"], bool) or not isinstance(item["issue"], int):
            raise OTACommissioningError("P0 issue is invalid")
        evidence_issues.append(item["issue"])
        _sha256(item["test_evidence_sha256"], "P0 test evidence")
    if tuple(evidence_issues) != REQUIRED_P0_ISSUES:
        blockers.add("commissioning_evidence_incomplete")

    approval = _exact_mapping(
        root["approval"], {"approved_at", "approver", "expires_at"}, "approval"
    )
    _text(approval["approver"], "approver", maximum=254)
    approved_at = _timestamp(approval["approved_at"], "approval timestamp")
    expires_at = _timestamp(approval["expires_at"], "approval expiry")
    if (
        approved_at > now + timedelta(minutes=5)
        or expires_at <= approved_at
        or expires_at > approved_at + timedelta(days=366)
    ):
        raise OTACommissioningError("approval interval is invalid")
    if expires_at <= now:
        blockers.add("commissioning_expired")

    return tuple(value for value in COMMISSIONING_BLOCKERS if value in blockers)


def evaluate_auto_apply_commissioning(
    settings: Any,
    *,
    now: Optional[datetime] = None,
    expected_owner_uid: Optional[int] = None,
) -> OTACommissioningGate:
    """Evaluate the current artifact without exposing paths or approver data."""

    requested = bool(getattr(settings, "ota_auto_apply", False))
    if bool(getattr(settings, "ota_auto_apply_rollback", False)):
        return OTACommissioningGate(
            requested=requested,
            permitted=False,
            blockers=("rollback_switch_active",),
        )
    if not requested:
        return OTACommissioningGate(requested=False, permitted=False)

    raw_path = str(getattr(settings, "ota_commissioning_artifact_path", "") or "")
    if not raw_path:
        return OTACommissioningGate(
            requested=True,
            permitted=False,
            blockers=("commissioning_path_missing",),
        )
    path = Path(raw_path)
    if not path.is_absolute() or ".." in path.parts:
        return OTACommissioningGate(
            requested=True,
            permitted=False,
            blockers=("commissioning_path_unsafe",),
        )
    pinned = str(getattr(settings, "ota_commissioning_artifact_sha256", "") or "")
    if _SHA256.fullmatch(pinned) is None:
        return OTACommissioningGate(
            requested=True,
            permitted=False,
            blockers=("commissioning_digest_missing",),
        )

    owner_uid = ROOT_OWNER_UID if expected_owner_uid is None else expected_owner_uid
    try:
        artifact, digest = _load_artifact(
            path,
            pinned,
            expected_owner_uid=owner_uid,
        )
    except OSError:
        return OTACommissioningGate(
            requested=True,
            permitted=False,
            blockers=("commissioning_path_unsafe",),
        )
    except OTACommissioningError as exc:
        blocker = (
            "commissioning_digest_mismatch"
            if "digest mismatch" in str(exc)
            else "commissioning_artifact_invalid"
        )
        return OTACommissioningGate(
            requested=True,
            permitted=False,
            blockers=(blocker,),
        )

    current = now or datetime.now(timezone.utc)
    if current.tzinfo is None:
        current = current.replace(tzinfo=timezone.utc)
    else:
        current = current.astimezone(timezone.utc)
    try:
        blockers = _semantic_blockers(
            artifact,
            settings,
            now=current,
            expected_owner_uid=owner_uid,
        )
    except (OSError, OTACommissioningError, TypeError, ValueError):
        blockers = ("commissioning_artifact_invalid",)
    return OTACommissioningGate(
        requested=True,
        permitted=not blockers,
        blockers=blockers,
        artifact_sha256=digest,
    )


__all__ = [
    "COMMISSIONING_BLOCKERS",
    "COMMISSIONING_SCHEMA",
    "OTACommissioningError",
    "OTACommissioningGate",
    "REQUIRED_P0_ISSUES",
    "evaluate_auto_apply_commissioning",
]
