580 lines
19 KiB
Python
580 lines
19 KiB
Python
#!/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
|