#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
"""Host-only Phase-0.9C feasibility and result-protocol model.

This module performs no filesystem, network, compiler, or target operation.
It is not observer source and must never be part of a PS5 target.
"""

from __future__ import annotations

from dataclasses import dataclass, field
import hashlib
import struct


PROOF_STATES = {
    "PROVEN_SAFE",
    "PROVEN_SIDE_EFFECTING",
    "UNPROVEN",
    "NOT_APPLICABLE",
}

STARTUP_PATHS: dict[str, tuple[tuple[str, str], ...]] = {
    "normal_sdk": (
        ("loader_entry_transfer", "PROVEN_SIDE_EFFECTING"),
        ("bss_clear", "PROVEN_SIDE_EFFECTING"),
        ("syscall_kernel_klog_init", "PROVEN_SIDE_EFFECTING"),
        ("libc_isthreaded_write", "PROVEN_SIDE_EFFECTING"),
        ("__patch_init", "PROVEN_SIDE_EFFECTING"),
        ("rtld_init", "PROVEN_SIDE_EFFECTING"),
        ("constructors", "PROVEN_SIDE_EFFECTING"),
        ("main", "UNPROVEN"),
        ("destructors_and_rtld_close", "UNPROVEN"),
        ("terminate_branch", "UNPROVEN"),
    ),
    "freestanding_theoretical": (
        ("loader_entry_transfer", "PROVEN_SIDE_EFFECTING"),
        ("stack_alignment", "UNPROVEN"),
        ("bss_initialization", "UNPROVEN"),
        ("relative_relocations_only", "PROVEN_SAFE"),
        ("tls", "UNPROVEN"),
        ("read_abi", "UNPROVEN"),
        ("monotonic_time_abi", "UNPROVEN"),
        ("return_continuation", "UNPROVEN"),
        ("process_exit_abi", "UNPROVEN"),
        ("cleanup", "UNPROVEN"),
    ),
    "normal_return": (
        ("ret_to_saved_rip", "PROVEN_SIDE_EFFECTING"),
        ("original_process_continuation", "UNPROVEN"),
        ("process_termination", "UNPROVEN"),
        ("loader_cleanup", "UNPROVEN"),
    ),
    "normal_process_exit": (
        ("exit_import_or_raw_syscall", "UNPROVEN"),
        ("process_teardown", "UNPROVEN"),
        ("waitpid_reap", "PROVEN_SAFE"),
        ("exit_status_delivery", "UNPROVEN"),
        ("result_copyout", "UNPROVEN"),
    ),
    "error_or_crash": (
        ("partial_runtime_unwind", "UNPROVEN"),
        ("waitpid_status_discarded", "PROVEN_SIDE_EFFECTING"),
        ("fixed_ambiguous_response", "PROVEN_SIDE_EFFECTING"),
    ),
    "timeout": (
        ("sigterm", "PROVEN_SIDE_EFFECTING"),
        ("sigkill", "PROVEN_SIDE_EFFECTING"),
        ("reap", "PROVEN_SAFE"),
        ("safe_exit", "UNPROVEN"),
    ),
}

PROHIBITED_STARTUP_EFFECTS = {
    "__patch_init",
    "kernel_copyin",
    "kernel_set_ucred_caps",
    "kernel_set_ucred_attrs",
    "syscall_permission_bound_write",
    "dynamic_module_loading",
}

NORMAL_SDK_REACHABLE_PROHIBITED_EFFECTS = {
    "__patch_init",
    "kernel_copyin",
    "kernel_set_ucred_caps",
    "kernel_set_ucred_attrs",
    "syscall_permission_bound_write",
    "dynamic_module_loading",
}

FREESTANDING_DEPENDENCY_CLOSURE = {
    "entry_address": "PROVEN_SAFE",
    "rdi_argument": "PROVEN_SAFE",
    "stack_alignment": "UNPROVEN",
    "saved_rip_continuation": "UNPROVEN",
    "bss_zero_fill": "UNPROVEN",
    "relative_relocation_subset": "PROVEN_SAFE",
    "complete_relocation_set": "UNPROVEN",
    "tls": "UNPROVEN",
    "constructors": "NOT_APPLICABLE",
    "libc": "NOT_APPLICABLE",
    "heap": "NOT_APPLICABLE",
    "callable_read_abi": "UNPROVEN",
    "callable_monotonic_time_abi": "UNPROVEN",
    "process_exit_abi": "UNPROVEN",
    "return_cleanup": "UNPROVEN",
    "bounded_result_copyout": "UNPROVEN",
}

EXIT_PATHS = {
    "return": (
        "ENTRY",
        "RET_TO_SAVED_RIP",
        "ORIGINAL_PROCESS_CONTINUATION_UNPROVEN",
        "NO_SAFE_TERMINAL",
    ),
    "process_exit": (
        "ENTRY",
        "EXIT_ABI_UNPROVEN",
        "CHILD_REAPED_STATUS_DISCARDED",
        "NO_RESULT_COPYOUT",
    ),
    "crash": (
        "ENTRY",
        "FAULT",
        "CHILD_REAPED_STATUS_DISCARDED",
        "AMBIGUOUS_FIXED_RESPONSE",
    ),
    "timeout": (
        "ENTRY",
        "DEADLINE",
        "SIGTERM",
        "SIGKILL",
        "REAPED",
        "INADMISSIBLE_TERMINATION",
    ),
}

SAFE_EXIT_TERMINALS: frozenset[str] = frozenset()

OUTPUT_ARCHITECTURES = {
    "D1_CALLER_OWNED_BOUNDED_BUFFER": (
        "CONCEPT_FEASIBLE_REQUIRES_LOADER_CHANGE_AND_EXIT_PROOF"
    ),
    "D2_EXISTING_REQUEST_RESPONSE": "REJECTED_SEND_ONLY_NO_RESULT_RECEIVE",
    "D3_LOADER_OWNED_STATUS_RECORD": (
        "UNPROVEN_REQUIRES_LOADER_STATE_AND_PROPAGATION_CHANGE"
    ),
    "D4_PROCESS_EXIT_STATUS": "REJECTED_WAIT_STATUS_DISCARDED_AND_AMBIGUOUS",
}


def validate_startup_model() -> list[str]:
    errors: list[str] = []
    required_paths = {
        "normal_sdk",
        "freestanding_theoretical",
        "normal_return",
        "normal_process_exit",
        "error_or_crash",
        "timeout",
    }
    if set(STARTUP_PATHS) != required_paths:
        errors.append("startup path inventory differs")
    for path_name, steps in STARTUP_PATHS.items():
        if not steps:
            errors.append(f"{path_name} has no steps")
        for step, classification in steps:
            if not step or classification not in PROOF_STATES:
                errors.append(f"{path_name} has invalid step {step}")
    if not (
        PROHIBITED_STARTUP_EFFECTS & NORMAL_SDK_REACHABLE_PROHIBITED_EFFECTS
    ):
        errors.append("normal SDK prohibited effects were hidden")
    if SAFE_EXIT_TERMINALS:
        errors.append("an unproven safe exit terminal was added")
    return errors


def freestanding_blockers() -> tuple[str, ...]:
    return tuple(
        key
        for key, classification in FREESTANDING_DEPENDENCY_CLOSURE.items()
        if classification == "UNPROVEN"
    )


def exit_path_is_safe(path: str) -> bool:
    states = EXIT_PATHS[path]
    return bool(states and states[-1] in SAFE_EXIT_TERMINALS)


MAGIC = b"CHG09C01"
PROTOCOL_VERSION = 1
HEADER_SIZE = 256
MAX_OUTPUT_SIZE = 4096
MAX_BODY_SIZE = MAX_OUTPUT_SIZE - HEADER_SIZE
COMPLETION_MARKER = b"COMPLETE"
FIRMWARE_FIELD_SIZE = 8
NONCE_SIZE = 16
REQUEST_ID_SIZE = 16

STATUS_SUCCESS = 1
STATUS_OBSERVER_ERROR = 2
STATUS_TIMEOUT = 3
STATUS_FIRMWARE_CONFLICT = 4

FLAG_TRUNCATED = 1 << 0

CLEANUP_INCOMPLETE = 0
CLEANUP_CLEAN = 1
CLEANUP_FAILED = 2

OFFSET_MAGIC = 0
OFFSET_VERSION = 8
OFFSET_HEADER_SIZE = 10
OFFSET_MAX_OUTPUT = 12
OFFSET_ACTUAL_OUTPUT = 16
OFFSET_OBSERVER_VERSION = 20
OFFSET_STATUS = 24
OFFSET_FLAGS = 28
OFFSET_CLEANUP = 32
OFFSET_RESERVED_WORD = 36
OFFSET_REQUESTED = 40
OFFSET_OBSERVED = 48
OFFSET_UNSUPPORTED = 56
OFFSET_DEADLINE = 64
OFFSET_NONCE = 72
OFFSET_REQUEST_ID = 88
OFFSET_FIRMWARE_ONE = 104
OFFSET_FIRMWARE_TWO = 112
OFFSET_ARTIFACT_HASH = 120
OFFSET_BODY_CHECKSUM = 152
OFFSET_RESULT_CHECKSUM = 184
OFFSET_COMPLETION = 216
OFFSET_RESERVED = 224


@dataclass(frozen=True)
class ResultRequest:
    execution_nonce: bytes
    request_id: bytes
    firmware_source_one: str | None
    firmware_source_two: str | None
    observer_version: int
    requested_capabilities: int
    artifact_sha256: bytes
    deadline_monotonic_ns: int

    def validate(self) -> None:
        if len(self.execution_nonce) != NONCE_SIZE:
            raise ValueError("execution nonce must be 16 bytes")
        if len(self.request_id) != REQUEST_ID_SIZE:
            raise ValueError("request ID must be 16 bytes")
        if len(self.artifact_sha256) != 32:
            raise ValueError("artifact SHA-256 must be 32 bytes")
        if self.observer_version <= 0:
            raise ValueError("observer version must be positive")
        if not 0 <= self.requested_capabilities <= 0xFFFFFFFFFFFFFFFF:
            raise ValueError("requested capability bitmap is out of range")
        if not 0 < self.deadline_monotonic_ns <= 0xFFFFFFFFFFFFFFFF:
            raise ValueError("deadline is out of range")
        _encode_firmware(self.firmware_source_one)
        _encode_firmware(self.firmware_source_two)


def _encode_firmware(value: str | None) -> bytes:
    if value is None:
        return b"\0" * FIRMWARE_FIELD_SIZE
    encoded = value.encode("ascii")
    if not encoded or b"\0" in encoded or len(encoded) >= FIRMWARE_FIELD_SIZE:
        raise ValueError("firmware ID is not canonical")
    return encoded.ljust(FIRMWARE_FIELD_SIZE, b"\0")


def _decode_firmware(value: bytes) -> str | None:
    if value == b"\0" * FIRMWARE_FIELD_SIZE:
        return None
    first_nul = value.find(b"\0")
    if first_nul < 1 or any(value[first_nul:]):
        raise ValueError("firmware ID is not canonically padded")
    return value[:first_nul].decode("ascii")


def _put_u16(buffer: bytearray, offset: int, value: int) -> None:
    struct.pack_into(">H", buffer, offset, value)


def _put_u32(buffer: bytearray, offset: int, value: int) -> None:
    struct.pack_into(">I", buffer, offset, value)


def _put_u64(buffer: bytearray, offset: int, value: int) -> None:
    struct.pack_into(">Q", buffer, offset, value)


def _get_u16(buffer: bytes, offset: int) -> int:
    return struct.unpack_from(">H", buffer, offset)[0]


def _get_u32(buffer: bytes, offset: int) -> int:
    return struct.unpack_from(">I", buffer, offset)[0]


def _get_u64(buffer: bytes, offset: int) -> int:
    return struct.unpack_from(">Q", buffer, offset)[0]


def _result_checksum(buffer: bytes, actual_output_size: int) -> bytes:
    candidate = bytearray(buffer[:actual_output_size])
    candidate[OFFSET_RESULT_CHECKSUM : OFFSET_RESULT_CHECKSUM + 32] = b"\0" * 32
    candidate[OFFSET_COMPLETION : OFFSET_COMPLETION + 8] = b"\0" * 8
    return hashlib.sha256(candidate).digest()


def build_result(
    request: ResultRequest,
    body: bytes,
    *,
    status: int = STATUS_SUCCESS,
    observed_capabilities: int = 0,
    unsupported_capabilities: int = 0,
    cleanup_status: int = CLEANUP_CLEAN,
    truncate: bool = False,
    complete: bool = True,
    protocol_version: int = PROTOCOL_VERSION,
) -> bytes:
    """Build a deterministic host record as if observer then caller finalized it."""

    request.validate()
    if not isinstance(body, bytes):
        raise TypeError("body must be bytes")
    if truncate and len(body) > MAX_BODY_SIZE:
        body = body[:MAX_BODY_SIZE]
    elif len(body) > MAX_BODY_SIZE:
        raise ValueError("body exceeds fixed result buffer")
    for value in (observed_capabilities, unsupported_capabilities):
        if not 0 <= value <= 0xFFFFFFFFFFFFFFFF:
            raise ValueError("capability bitmap is out of range")

    actual_output_size = HEADER_SIZE + len(body)
    if actual_output_size < HEADER_SIZE or actual_output_size > MAX_OUTPUT_SIZE:
        raise ValueError("checked output-size arithmetic failed")

    flags = FLAG_TRUNCATED if truncate else 0
    buffer = bytearray(MAX_OUTPUT_SIZE)
    buffer[OFFSET_MAGIC : OFFSET_MAGIC + 8] = MAGIC
    _put_u16(buffer, OFFSET_VERSION, protocol_version)
    _put_u16(buffer, OFFSET_HEADER_SIZE, HEADER_SIZE)
    _put_u32(buffer, OFFSET_MAX_OUTPUT, MAX_OUTPUT_SIZE)
    _put_u32(buffer, OFFSET_ACTUAL_OUTPUT, actual_output_size)
    _put_u32(buffer, OFFSET_OBSERVER_VERSION, request.observer_version)
    _put_u32(buffer, OFFSET_STATUS, status)
    _put_u32(buffer, OFFSET_FLAGS, flags)
    _put_u32(buffer, OFFSET_CLEANUP, cleanup_status)
    _put_u64(buffer, OFFSET_REQUESTED, request.requested_capabilities)
    _put_u64(buffer, OFFSET_OBSERVED, observed_capabilities)
    _put_u64(buffer, OFFSET_UNSUPPORTED, unsupported_capabilities)
    _put_u64(buffer, OFFSET_DEADLINE, request.deadline_monotonic_ns)
    buffer[OFFSET_NONCE : OFFSET_NONCE + NONCE_SIZE] = request.execution_nonce
    buffer[OFFSET_REQUEST_ID : OFFSET_REQUEST_ID + REQUEST_ID_SIZE] = (
        request.request_id
    )
    buffer[OFFSET_FIRMWARE_ONE : OFFSET_FIRMWARE_ONE + FIRMWARE_FIELD_SIZE] = (
        _encode_firmware(request.firmware_source_one)
    )
    buffer[OFFSET_FIRMWARE_TWO : OFFSET_FIRMWARE_TWO + FIRMWARE_FIELD_SIZE] = (
        _encode_firmware(request.firmware_source_two)
    )
    buffer[OFFSET_ARTIFACT_HASH : OFFSET_ARTIFACT_HASH + 32] = (
        request.artifact_sha256
    )
    buffer[HEADER_SIZE:actual_output_size] = body
    buffer[OFFSET_BODY_CHECKSUM : OFFSET_BODY_CHECKSUM + 32] = hashlib.sha256(
        body
    ).digest()
    buffer[OFFSET_RESULT_CHECKSUM : OFFSET_RESULT_CHECKSUM + 32] = (
        _result_checksum(buffer, actual_output_size)
    )
    if complete:
        buffer[OFFSET_COMPLETION : OFFSET_COMPLETION + 8] = COMPLETION_MARKER
    return bytes(buffer)


def validate_result(
    record: bytes, request: ResultRequest, *, now_monotonic_ns: int
) -> str:
    """Validate one fixed caller-owned record and return a fail-closed decision."""

    try:
        request.validate()
    except (TypeError, ValueError):
        return "BLOCKED_INVALID_REQUEST"
    if len(record) != MAX_OUTPUT_SIZE:
        return "BLOCKED_WRONG_BUFFER_SIZE"
    if record[OFFSET_MAGIC : OFFSET_MAGIC + 8] != MAGIC:
        return "BLOCKED_BAD_MAGIC"
    if _get_u16(record, OFFSET_VERSION) != PROTOCOL_VERSION:
        return "BLOCKED_UNKNOWN_VERSION"
    if _get_u16(record, OFFSET_HEADER_SIZE) != HEADER_SIZE:
        return "BLOCKED_BAD_HEADER_SIZE"
    if _get_u32(record, OFFSET_MAX_OUTPUT) != MAX_OUTPUT_SIZE:
        return "BLOCKED_MAXIMUM_MISMATCH"

    actual_output_size = _get_u32(record, OFFSET_ACTUAL_OUTPUT)
    if not HEADER_SIZE <= actual_output_size <= MAX_OUTPUT_SIZE:
        return "BLOCKED_ACTUAL_SIZE"
    if any(record[actual_output_size:]):
        return "BLOCKED_NONZERO_UNUSED_BYTES"
    if _get_u32(record, OFFSET_RESERVED_WORD) != 0 or any(
        record[OFFSET_RESERVED:HEADER_SIZE]
    ):
        return "BLOCKED_RESERVED_DATA"
    if record[OFFSET_COMPLETION : OFFSET_COMPLETION + 8] != COMPLETION_MARKER:
        return "BLOCKED_INCOMPLETE"
    if (
        record[OFFSET_NONCE : OFFSET_NONCE + NONCE_SIZE]
        != request.execution_nonce
    ):
        return "BLOCKED_STALE_NONCE"
    if (
        record[OFFSET_REQUEST_ID : OFFSET_REQUEST_ID + REQUEST_ID_SIZE]
        != request.request_id
    ):
        return "BLOCKED_STALE_REQUEST_ID"
    if _get_u32(record, OFFSET_OBSERVER_VERSION) != request.observer_version:
        return "BLOCKED_OBSERVER_VERSION"
    if _get_u64(record, OFFSET_DEADLINE) != request.deadline_monotonic_ns:
        return "BLOCKED_DEADLINE_BINDING"
    if now_monotonic_ns > request.deadline_monotonic_ns:
        return "BLOCKED_TIMEOUT"
    if (
        record[OFFSET_ARTIFACT_HASH : OFFSET_ARTIFACT_HASH + 32]
        != request.artifact_sha256
    ):
        return "BLOCKED_ARTIFACT_HASH"

    try:
        firmware_one = _decode_firmware(
            record[
                OFFSET_FIRMWARE_ONE : OFFSET_FIRMWARE_ONE + FIRMWARE_FIELD_SIZE
            ]
        )
        firmware_two = _decode_firmware(
            record[
                OFFSET_FIRMWARE_TWO : OFFSET_FIRMWARE_TWO + FIRMWARE_FIELD_SIZE
            ]
        )
    except (UnicodeDecodeError, ValueError):
        return "BLOCKED_FIRMWARE_ENCODING"
    if firmware_one is None:
        return "BLOCKED_FIRMWARE_SOURCE_1_ABSENT"
    if firmware_two is None:
        return "BLOCKED_FIRMWARE_SOURCE_2_ABSENT"
    if firmware_one != firmware_two:
        return "BLOCKED_FIRMWARE_CONFLICT"
    if firmware_one != "9.60":
        return "BLOCKED_FIRMWARE_MISMATCH"
    if (
        firmware_one != request.firmware_source_one
        or firmware_two != request.firmware_source_two
    ):
        return "BLOCKED_FIRMWARE_BINDING"

    expected_result_checksum = _result_checksum(record, actual_output_size)
    if (
        record[OFFSET_RESULT_CHECKSUM : OFFSET_RESULT_CHECKSUM + 32]
        != expected_result_checksum
    ):
        return "BLOCKED_RESULT_CHECKSUM"
    body = record[HEADER_SIZE:actual_output_size]
    if (
        record[OFFSET_BODY_CHECKSUM : OFFSET_BODY_CHECKSUM + 32]
        != hashlib.sha256(body).digest()
    ):
        return "BLOCKED_BODY_CHECKSUM"
    if _get_u32(record, OFFSET_FLAGS) & ~FLAG_TRUNCATED:
        return "BLOCKED_UNKNOWN_FLAGS"
    if _get_u32(record, OFFSET_FLAGS) & FLAG_TRUNCATED:
        return "BLOCKED_TRUNCATED"
    if _get_u32(record, OFFSET_CLEANUP) != CLEANUP_CLEAN:
        return "BLOCKED_CLEANUP_NOT_PROVEN"
    if _get_u32(record, OFFSET_STATUS) != STATUS_SUCCESS:
        return "BLOCKED_OBSERVER_FAILURE"

    requested = _get_u64(record, OFFSET_REQUESTED)
    observed = _get_u64(record, OFFSET_OBSERVED)
    unsupported = _get_u64(record, OFFSET_UNSUPPORTED)
    if requested != request.requested_capabilities:
        return "BLOCKED_CAPABILITY_REQUEST_BINDING"
    if observed & unsupported:
        return "BLOCKED_CAPABILITY_BITMAP_CONFLICT"
    if (observed | unsupported) & ~requested:
        return "BLOCKED_UNREQUESTED_CAPABILITY"
    if (observed | unsupported) != requested:
        return "BLOCKED_INCOMPLETE_CAPABILITY_RESULT"
    if unsupported:
        return "VALID_RECORD_WITH_UNSUPPORTED_CAPABILITIES"
    return "VALID_COMPLETE_RESULT"


@dataclass
class ResultConsumer:
    consumed: set[tuple[bytes, bytes]] = field(default_factory=set)

    def consume(
        self, record: bytes, request: ResultRequest, *, now_monotonic_ns: int
    ) -> str:
        key = (request.execution_nonce, request.request_id)
        if key in self.consumed:
            return "BLOCKED_DUPLICATE_RESULT"
        result = validate_result(
            record, request, now_monotonic_ns=now_monotonic_ns
        )
        if result in {
            "VALID_COMPLETE_RESULT",
            "VALID_RECORD_WITH_UNSUPPORTED_CAPABILITIES",
        }:
            self.consumed.add(key)
        return result


SIDE_EFFECT_MODEL = {
    "runtime_self_identity": {
        "semantic_readonly",
        "cache_effect_possible",
        "audit_effect_possible",
    },
    "firmware_query": {
        "semantic_readonly",
        "cache_effect_possible",
        "audit_effect_possible",
        "service_or_security_counter_effect_possible",
    },
    "filesystem_metadata": {
        "semantic_readonly",
        "metadata_effect_possible",
        "cache_effect_possible",
        "audit_effect_possible",
        "open_bookkeeping_effect_possible",
        "object_race_possible",
    },
    "filesystem_content_hash": {
        "semantic_readonly",
        "metadata_effect_possible",
        "atime_effect_possible",
        "cache_effect_possible",
        "audit_effect_possible",
        "open_bookkeeping_effect_possible",
        "object_race_possible",
    },
    "process_service_listener_snapshot": {
        "semantic_readonly",
        "cache_effect_possible",
        "audit_effect_possible",
        "counter_effect_possible",
        "process_accounting_effect_possible",
        "object_race_possible",
    },
    "autoload_read": {
        "semantic_readonly",
        "metadata_effect_possible",
        "atime_effect_possible",
        "cache_effect_possible",
        "audit_effect_possible",
        "service_state_effect_possible",
        "object_race_possible",
    },
}


def classify_side_effects(operation: str) -> frozenset[str]:
    return frozenset(SIDE_EFFECT_MODEL[operation])


def is_proven_side_effect_free(operation: str) -> bool:
    _ = SIDE_EFFECT_MODEL[operation]
    return False