1009 lines
31 KiB
Python
1009 lines
31 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
"""Host-only virtual simulator for the Phase-0.9A anti-brick design.
|
|
|
|
The model never opens a target artifact, touches a device path, starts a
|
|
compiler, or performs network I/O. All objects and writes are logical records
|
|
held in memory. A successful host-model result is not PS5 hardware evidence.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import copy
|
|
import json
|
|
from dataclasses import asdict, dataclass
|
|
from typing import Any, Iterable
|
|
|
|
|
|
STATES = (
|
|
"OFFLINE_ONLY",
|
|
"OBSERVATION_NOT_AUTHORIZED",
|
|
"OBSERVATION_AUTHORIZED",
|
|
"DEVICE_IDENTITY_OBSERVED",
|
|
"LIVE_OBJECTS_VERIFIED",
|
|
"BACKUP_NOT_PRESENT",
|
|
"BACKUP_CREATION_NOT_AUTHORIZED",
|
|
"BACKUP_CREATION_AUTHORIZED",
|
|
"BACKUP_CREATED",
|
|
"BACKUP_REOPENED_AND_VERIFIED",
|
|
"RECOVERY_PATH_VERIFIED",
|
|
"CANDIDATE_NOT_AUTHORIZED",
|
|
"CANDIDATE_APPROVED_OFFLINE",
|
|
"STAGING_NOT_AUTHORIZED",
|
|
"STAGING_AUTHORIZED",
|
|
"CANDIDATE_STAGED",
|
|
"CANDIDATE_REOPENED_AND_VERIFIED",
|
|
"TARGET_NOT_QUIESCENT",
|
|
"TARGET_QUIESCENT",
|
|
"SWITCH_NOT_AUTHORIZED",
|
|
"SWITCH_AUTHORIZED",
|
|
"SWITCH_IN_PROGRESS",
|
|
"POST_SWITCH_VERIFY",
|
|
"MANUAL_EXECUTION_NOT_AUTHORIZED",
|
|
"MANUAL_EXECUTION_AUTHORIZED",
|
|
"ONE_SHOT_EXECUTION",
|
|
"ACCEPTED",
|
|
"ROLLBACK_REQUIRED",
|
|
"ROLLBACK_AUTHORIZED",
|
|
"ROLLBACK_IN_PROGRESS",
|
|
"ROLLBACK_VERIFIED",
|
|
"BLOCKED",
|
|
)
|
|
|
|
ALLOWED_FORWARD_TRANSITIONS = {
|
|
("OFFLINE_ONLY", "OBSERVATION_NOT_AUTHORIZED"),
|
|
("OBSERVATION_NOT_AUTHORIZED", "OBSERVATION_AUTHORIZED"),
|
|
("OBSERVATION_AUTHORIZED", "DEVICE_IDENTITY_OBSERVED"),
|
|
("DEVICE_IDENTITY_OBSERVED", "LIVE_OBJECTS_VERIFIED"),
|
|
("LIVE_OBJECTS_VERIFIED", "BACKUP_NOT_PRESENT"),
|
|
("BACKUP_NOT_PRESENT", "BACKUP_CREATION_NOT_AUTHORIZED"),
|
|
("BACKUP_CREATION_NOT_AUTHORIZED", "BACKUP_CREATION_AUTHORIZED"),
|
|
("BACKUP_CREATION_AUTHORIZED", "BACKUP_CREATED"),
|
|
("BACKUP_CREATED", "BACKUP_REOPENED_AND_VERIFIED"),
|
|
("BACKUP_REOPENED_AND_VERIFIED", "RECOVERY_PATH_VERIFIED"),
|
|
("RECOVERY_PATH_VERIFIED", "CANDIDATE_NOT_AUTHORIZED"),
|
|
("CANDIDATE_NOT_AUTHORIZED", "CANDIDATE_APPROVED_OFFLINE"),
|
|
("CANDIDATE_APPROVED_OFFLINE", "STAGING_NOT_AUTHORIZED"),
|
|
("STAGING_NOT_AUTHORIZED", "STAGING_AUTHORIZED"),
|
|
("STAGING_AUTHORIZED", "CANDIDATE_STAGED"),
|
|
("CANDIDATE_STAGED", "CANDIDATE_REOPENED_AND_VERIFIED"),
|
|
("CANDIDATE_REOPENED_AND_VERIFIED", "TARGET_NOT_QUIESCENT"),
|
|
("TARGET_NOT_QUIESCENT", "TARGET_QUIESCENT"),
|
|
("TARGET_QUIESCENT", "SWITCH_NOT_AUTHORIZED"),
|
|
("SWITCH_NOT_AUTHORIZED", "SWITCH_AUTHORIZED"),
|
|
("SWITCH_AUTHORIZED", "SWITCH_IN_PROGRESS"),
|
|
("SWITCH_IN_PROGRESS", "POST_SWITCH_VERIFY"),
|
|
("POST_SWITCH_VERIFY", "MANUAL_EXECUTION_NOT_AUTHORIZED"),
|
|
("MANUAL_EXECUTION_NOT_AUTHORIZED", "MANUAL_EXECUTION_AUTHORIZED"),
|
|
("MANUAL_EXECUTION_AUTHORIZED", "ONE_SHOT_EXECUTION"),
|
|
("ONE_SHOT_EXECUTION", "ACCEPTED"),
|
|
("POST_SWITCH_VERIFY", "ROLLBACK_REQUIRED"),
|
|
("ROLLBACK_REQUIRED", "ROLLBACK_AUTHORIZED"),
|
|
("ROLLBACK_AUTHORIZED", "ROLLBACK_IN_PROGRESS"),
|
|
("ROLLBACK_IN_PROGRESS", "ROLLBACK_VERIFIED"),
|
|
}
|
|
|
|
COMPONENTS = {
|
|
"hardened_elfldr": {
|
|
"stock_size": 397000,
|
|
"stock_sha256": (
|
|
"092d16ee0ede0c494947efd38d1a17bbd7cc4b022d3858ea898833c188c703e8"
|
|
),
|
|
"candidate_size": 397000,
|
|
"candidate_sha256": (
|
|
"63e810982471eb40cae3a20aa9df9a0a02892f420e429874fae4e99aa400b561"
|
|
),
|
|
"candidate_source_commit": "197623058f509eddde18868dafcb92fdcac66464",
|
|
},
|
|
"controlled_payload_manager": {
|
|
"stock_size": 2050320,
|
|
"stock_sha256": (
|
|
"518740adbacccb9094fadb07dd424c53ee290f38306449ccc9d6957fdf813c0b"
|
|
),
|
|
"candidate_size": 99560,
|
|
"candidate_sha256": (
|
|
"8fecf8241a46246eddbd21e8bb4d875f5d76f1f4f4c6a11384df1f131aa5e5b1"
|
|
),
|
|
"candidate_source_commit": "e23d94ff91233aa770e2342800c1467875bdef44",
|
|
},
|
|
}
|
|
|
|
FAULTS = (
|
|
"wrong_preimage_hash",
|
|
"wrong_size",
|
|
"missing_live_path",
|
|
"missing_mount_id",
|
|
"missing_object_id",
|
|
"reference_only_preimage",
|
|
"object_swap_after_preflight",
|
|
"symlink_substitution",
|
|
"backup_same_object_as_live",
|
|
"short_backup_write",
|
|
"backup_hash_mismatch",
|
|
"backup_not_reopened",
|
|
"insufficient_space",
|
|
"candidate_hash_mismatch",
|
|
"target_process_active",
|
|
"autoload_active",
|
|
"retry_active",
|
|
"switch_primitive_unknown",
|
|
"atomic_switch_failure",
|
|
"directory_durability_unknown",
|
|
"live_verification_failure",
|
|
"rollback_hash_mismatch",
|
|
"recovery_depends_on_replaced_component",
|
|
"wrong_component_artifact_mapping",
|
|
"lifecycle_probe_candidate",
|
|
"authorization_missing",
|
|
"authorization_wrong_hash",
|
|
"authorization_expired",
|
|
"second_component_before_first_accepted",
|
|
"timeout",
|
|
"unknown_firmware",
|
|
"in_place_overwrite",
|
|
"two_step_rename_gap",
|
|
)
|
|
|
|
POWER_LOSS_BOUNDARIES = (
|
|
"before_backup_write",
|
|
"during_backup_write",
|
|
"after_backup_write_before_flush",
|
|
"after_flush_before_reopen_hash",
|
|
"during_candidate_staging",
|
|
"after_candidate_staging_before_verification",
|
|
"immediately_before_live_switch",
|
|
"during_live_switch",
|
|
"immediately_after_live_switch",
|
|
"after_switch_before_live_hash",
|
|
"after_live_hash_before_execution",
|
|
"during_first_manual_execution",
|
|
"during_rollback",
|
|
"after_rollback_before_recovery_verification",
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class VirtualObject:
|
|
"""Logical file identity; it is never materialized on the host."""
|
|
|
|
path_token: str | None
|
|
mount_id: str | None
|
|
object_id: str | None
|
|
object_type: str
|
|
size: int
|
|
sha256: str
|
|
complete: bool = True
|
|
durable: bool = True
|
|
reopened: bool = False
|
|
|
|
|
|
@dataclass
|
|
class SyntheticAuthorization:
|
|
"""A host-model gate token that explicitly grants no real-world authority."""
|
|
|
|
action: str
|
|
component: str
|
|
artifact_sha256: str
|
|
gate_token_present: bool
|
|
expires_at_tick: int
|
|
synthetic_host_model_only: bool = True
|
|
|
|
def valid(
|
|
self, action: str, component: str, artifact_sha256: str, now_tick: int
|
|
) -> bool:
|
|
return (
|
|
self.synthetic_host_model_only
|
|
and self.gate_token_present
|
|
and self.action == action
|
|
and self.component == component
|
|
and self.artifact_sha256 == artifact_sha256
|
|
and now_tick < self.expires_at_tick
|
|
)
|
|
|
|
|
|
class TransitionError(RuntimeError):
|
|
"""Raised when the state machine would skip a mandatory gate."""
|
|
|
|
|
|
class StateMachine:
|
|
def __init__(self) -> None:
|
|
self.state = "OFFLINE_ONLY"
|
|
self.history = [self.state]
|
|
|
|
def transition(self, destination: str) -> None:
|
|
if destination not in STATES:
|
|
raise TransitionError(f"unknown state: {destination}")
|
|
if destination == "BLOCKED":
|
|
self.state = destination
|
|
self.history.append(destination)
|
|
return
|
|
if (self.state, destination) not in ALLOWED_FORWARD_TRANSITIONS:
|
|
raise TransitionError(
|
|
f"forbidden transition: {self.state} -> {destination}"
|
|
)
|
|
self.state = destination
|
|
self.history.append(destination)
|
|
|
|
|
|
def _authorization(
|
|
action: str,
|
|
component: str,
|
|
digest: str,
|
|
faults: set[str],
|
|
) -> SyntheticAuthorization:
|
|
gate_token_present = "authorization_missing" not in faults
|
|
scoped_digest = "0" * 64 if "authorization_wrong_hash" in faults else digest
|
|
expiry = 0 if "authorization_expired" in faults else 100
|
|
return SyntheticAuthorization(
|
|
action=action,
|
|
component=component,
|
|
artifact_sha256=scoped_digest,
|
|
gate_token_present=gate_token_present,
|
|
expires_at_tick=expiry,
|
|
)
|
|
|
|
|
|
def _base_objects(component: str) -> tuple[VirtualObject, VirtualObject]:
|
|
identity = COMPONENTS[component]
|
|
live = VirtualObject(
|
|
path_token=f"LIVE_SLOT::{component}",
|
|
mount_id=f"MOUNT::{component}",
|
|
object_id=f"OBJECT::stock::{component}",
|
|
object_type="regular",
|
|
size=identity["stock_size"],
|
|
sha256=identity["stock_sha256"],
|
|
)
|
|
candidate = VirtualObject(
|
|
path_token=f"STAGING_SLOT::{component}",
|
|
mount_id=f"MOUNT::{component}",
|
|
object_id=f"OBJECT::candidate::{component}",
|
|
object_type="regular",
|
|
size=identity["candidate_size"],
|
|
sha256=identity["candidate_sha256"],
|
|
)
|
|
return live, candidate
|
|
|
|
|
|
def _report(
|
|
machine: StateMachine,
|
|
component: str,
|
|
outcome: str,
|
|
blockers: list[dict[str, str]],
|
|
live: VirtualObject,
|
|
backup: VirtualObject | None,
|
|
candidate: VirtualObject,
|
|
writes: list[str],
|
|
risk: str | None = None,
|
|
) -> dict[str, Any]:
|
|
identity = COMPONENTS[component]
|
|
old_live_intact = (
|
|
live.complete
|
|
and live.object_type == "regular"
|
|
and live.path_token == f"LIVE_SLOT::{component}"
|
|
and live.mount_id == f"MOUNT::{component}"
|
|
and live.object_id
|
|
in {
|
|
f"OBJECT::stock::{component}",
|
|
f"OBJECT::restored::{component}",
|
|
}
|
|
and live.sha256 == identity["stock_sha256"]
|
|
and live.size == identity["stock_size"]
|
|
)
|
|
verified_new_live = (
|
|
live.complete
|
|
and live.object_type == "regular"
|
|
and live.path_token == f"LIVE_SLOT::{component}"
|
|
and live.mount_id == f"MOUNT::{component}"
|
|
and live.object_id == f"OBJECT::live-candidate::{component}"
|
|
and live.sha256 == identity["candidate_sha256"]
|
|
and live.size == identity["candidate_size"]
|
|
)
|
|
verified_backup_intact = (
|
|
backup is not None
|
|
and backup.complete
|
|
and backup.reopened
|
|
and backup.sha256 == identity["stock_sha256"]
|
|
and backup.size == identity["stock_size"]
|
|
and backup.object_id != live.object_id
|
|
)
|
|
crash_invariant = (
|
|
"A_OLD_LIVE_COMPLETE"
|
|
if old_live_intact
|
|
else (
|
|
"B_NEW_LIVE_COMPLETE_AND_BACKUP_INTACT"
|
|
if verified_new_live and verified_backup_intact
|
|
else "C_REJECTED_UNSAFE_OR_UNPROVEN"
|
|
)
|
|
)
|
|
return {
|
|
"schema_contract": "chimera-gfx-phase-0.9-host-simulator-v1",
|
|
"phase": "PHASE_0_9A_OFFLINE_ANTI_BRICK",
|
|
"status": outcome,
|
|
"component": component,
|
|
"state": machine.state,
|
|
"state_history": machine.history,
|
|
"blockers": blockers,
|
|
"risk": risk,
|
|
"virtual_writes": writes,
|
|
"virtual_live": asdict(live),
|
|
"virtual_backup": asdict(backup) if backup is not None else None,
|
|
"virtual_candidate": asdict(candidate),
|
|
"crash_invariant": crash_invariant,
|
|
"host_simulation_only": True,
|
|
"hardware_evidence": False,
|
|
"ps5_connected": False,
|
|
"device_write_performed": False,
|
|
"files_transferred": False,
|
|
"target_execution_performed": False,
|
|
"target_artifact_created": False,
|
|
"production_installation_code": False,
|
|
"synthetic_host_model_authorization_states": [
|
|
state
|
|
for state in machine.history
|
|
if state
|
|
in {
|
|
"OBSERVATION_AUTHORIZED",
|
|
"BACKUP_CREATION_AUTHORIZED",
|
|
"STAGING_AUTHORIZED",
|
|
"SWITCH_AUTHORIZED",
|
|
"MANUAL_EXECUTION_AUTHORIZED",
|
|
"ROLLBACK_AUTHORIZED",
|
|
}
|
|
],
|
|
"real_world_authority": False,
|
|
"installation_authorized": False,
|
|
"lifecycle_authorized": False,
|
|
"execution_authorized": False,
|
|
"transfer_authorized": False,
|
|
"automatic_retry": False,
|
|
"autoload": False,
|
|
}
|
|
|
|
|
|
def _blocked(
|
|
machine: StateMachine,
|
|
component: str,
|
|
code: str,
|
|
detail: str,
|
|
live: VirtualObject,
|
|
backup: VirtualObject | None,
|
|
candidate: VirtualObject,
|
|
writes: list[str],
|
|
risk: str = "HIGH",
|
|
) -> dict[str, Any]:
|
|
machine.transition("BLOCKED")
|
|
return _report(
|
|
machine,
|
|
component,
|
|
"BLOCKED",
|
|
[{"code": code, "detail": detail}],
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
risk,
|
|
)
|
|
|
|
|
|
def simulate_transaction(
|
|
component: str,
|
|
faults: Iterable[str] = (),
|
|
*,
|
|
prove_virtual_switch_model: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Run one component transaction entirely in memory.
|
|
|
|
``prove_virtual_switch_model`` proves only the synthetic model primitive.
|
|
It never promotes a PS5 filesystem property.
|
|
"""
|
|
|
|
if component not in COMPONENTS:
|
|
raise ValueError("component must be hardened_elfldr or controlled_payload_manager")
|
|
fault_set = set(faults)
|
|
unknown = fault_set.difference(FAULTS)
|
|
if unknown:
|
|
raise ValueError(f"unknown fault(s): {', '.join(sorted(unknown))}")
|
|
|
|
identity = COMPONENTS[component]
|
|
live, candidate = _base_objects(component)
|
|
expected = copy.deepcopy(live)
|
|
backup: VirtualObject | None = None
|
|
writes: list[str] = []
|
|
machine = StateMachine()
|
|
machine.transition("OBSERVATION_NOT_AUTHORIZED")
|
|
|
|
observation_auth = _authorization(
|
|
"observation", component, identity["stock_sha256"], fault_set
|
|
)
|
|
if not observation_auth.valid(
|
|
"observation", component, identity["stock_sha256"], now_tick=1
|
|
):
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"OBSERVATION_AUTHORIZATION_INVALID",
|
|
"The synthetic observation input is missing, expired or hash-mismatched.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
)
|
|
machine.transition("OBSERVATION_AUTHORIZED")
|
|
|
|
if "unknown_firmware" in fault_set:
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"UNKNOWN_FIRMWARE",
|
|
"An unknown firmware always stops before mutation.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
)
|
|
machine.transition("DEVICE_IDENTITY_OBSERVED")
|
|
|
|
if "missing_live_path" in fault_set:
|
|
live.path_token = None
|
|
if "missing_mount_id" in fault_set:
|
|
live.mount_id = None
|
|
if "missing_object_id" in fault_set:
|
|
live.object_id = None
|
|
if "wrong_preimage_hash" in fault_set:
|
|
live.sha256 = "1" * 64
|
|
if "wrong_size" in fault_set:
|
|
live.size += 1
|
|
if "symlink_substitution" in fault_set:
|
|
live.object_type = "symlink"
|
|
if "reference_only_preimage" in fault_set:
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"REFERENCE_ONLY_PREIMAGE",
|
|
"A stock reference was not promoted by an exact stable observation.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
)
|
|
|
|
exact_identity = all(
|
|
(
|
|
live.path_token == expected.path_token,
|
|
live.mount_id == expected.mount_id,
|
|
live.object_id == expected.object_id,
|
|
live.object_type == "regular",
|
|
live.size == expected.size,
|
|
live.sha256 == expected.sha256,
|
|
live.complete,
|
|
)
|
|
)
|
|
if not exact_identity:
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"LIVE_IDENTITY_MISMATCH",
|
|
"Path, mount, object, type, size and hash must all match.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
)
|
|
machine.transition("LIVE_OBJECTS_VERIFIED")
|
|
machine.transition("BACKUP_NOT_PRESENT")
|
|
machine.transition("BACKUP_CREATION_NOT_AUTHORIZED")
|
|
|
|
backup_auth = _authorization(
|
|
"backup_creation", component, identity["stock_sha256"], fault_set
|
|
)
|
|
if not backup_auth.valid(
|
|
"backup_creation", component, identity["stock_sha256"], now_tick=1
|
|
):
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"BACKUP_AUTHORIZATION_INVALID",
|
|
"Missing, expired or differently hash-bound backup authorization.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
)
|
|
machine.transition("BACKUP_CREATION_AUTHORIZED")
|
|
|
|
if "object_swap_after_preflight" in fault_set:
|
|
live.object_id = "OBJECT::substituted"
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"OBJECT_CHANGED_AFTER_PREFLIGHT",
|
|
"The stable live object identity changed before the virtual write.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
)
|
|
if "insufficient_space" in fault_set:
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"INSUFFICIENT_SPACE",
|
|
"Capacity and metadata reserve are insufficient.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
)
|
|
if "backup_same_object_as_live" in fault_set:
|
|
backup = copy.deepcopy(live)
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"BACKUP_NOT_SEPARATE",
|
|
"The backup resolves to the live object.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
"CATASTROPHIC",
|
|
)
|
|
|
|
backup = VirtualObject(
|
|
path_token=f"BACKUP_SLOT::{component}",
|
|
mount_id=f"BACKUP_MOUNT::{component}",
|
|
object_id=f"OBJECT::backup::{component}",
|
|
object_type="regular",
|
|
size=live.size,
|
|
sha256=live.sha256,
|
|
durable=True,
|
|
)
|
|
writes.append("virtual_backup_create")
|
|
machine.transition("BACKUP_CREATED")
|
|
if "short_backup_write" in fault_set:
|
|
backup.size -= 1
|
|
backup.complete = False
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"SHORT_BACKUP_WRITE",
|
|
"A partial backup is never promotable.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
"CATASTROPHIC",
|
|
)
|
|
if "backup_hash_mismatch" in fault_set:
|
|
backup.sha256 = "2" * 64
|
|
if "backup_not_reopened" not in fault_set:
|
|
backup.reopened = True
|
|
if (
|
|
backup.sha256 != expected.sha256
|
|
or backup.size != expected.size
|
|
or not backup.reopened
|
|
):
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"BACKUP_REOPEN_VERIFY_FAILED",
|
|
"The separate backup must survive close, reopen, size and hash checks.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
"CATASTROPHIC",
|
|
)
|
|
machine.transition("BACKUP_REOPENED_AND_VERIFIED")
|
|
|
|
if "recovery_depends_on_replaced_component" in fault_set:
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"RECOVERY_NOT_INDEPENDENT",
|
|
"Recovery depends on the component being replaced.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
"CATASTROPHIC",
|
|
)
|
|
machine.transition("RECOVERY_PATH_VERIFIED")
|
|
machine.transition("CANDIDATE_NOT_AUTHORIZED")
|
|
|
|
if "lifecycle_probe_candidate" in fault_set:
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"LIFECYCLE_PROBE_NOT_INSTALLABLE",
|
|
"The lifecycle probe is excluded from both component transactions.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
)
|
|
if "wrong_component_artifact_mapping" in fault_set:
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"COMPONENT_ARTIFACT_MAPPING_MISMATCH",
|
|
"The candidate is bound to a different component.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
)
|
|
if "second_component_before_first_accepted" in fault_set:
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"COMPONENT_SEQUENCE_VIOLATION",
|
|
"A second component cannot start before separate acceptance or rollback.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
)
|
|
machine.transition("CANDIDATE_APPROVED_OFFLINE")
|
|
machine.transition("STAGING_NOT_AUTHORIZED")
|
|
stage_auth = _authorization(
|
|
"staging", component, identity["candidate_sha256"], fault_set
|
|
)
|
|
if not stage_auth.valid(
|
|
"staging", component, identity["candidate_sha256"], now_tick=1
|
|
):
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"STAGING_AUTHORIZATION_INVALID",
|
|
"Staging approval is missing, expired or bound to different bytes.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
)
|
|
machine.transition("STAGING_AUTHORIZED")
|
|
|
|
if "candidate_hash_mismatch" in fault_set:
|
|
candidate.sha256 = "3" * 64
|
|
writes.append("virtual_candidate_stage")
|
|
machine.transition("CANDIDATE_STAGED")
|
|
candidate.reopened = True
|
|
if (
|
|
candidate.sha256 != identity["candidate_sha256"]
|
|
or candidate.size != identity["candidate_size"]
|
|
or not candidate.complete
|
|
or not candidate.reopened
|
|
):
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"CANDIDATE_REOPEN_VERIFY_FAILED",
|
|
"Staged candidate bytes do not match the offline approved artifact.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
)
|
|
machine.transition("CANDIDATE_REOPENED_AND_VERIFIED")
|
|
machine.transition("TARGET_NOT_QUIESCENT")
|
|
|
|
if "target_process_active" in fault_set:
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"TARGET_NOT_QUIESCENT",
|
|
"The target process or service is still active.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
)
|
|
if "autoload_active" in fault_set:
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"AUTOLOAD_ACTIVE",
|
|
"Autoload is forbidden.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
)
|
|
if "retry_active" in fault_set:
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"AUTOMATIC_RETRY_ACTIVE",
|
|
"Automatic retry is forbidden.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
)
|
|
if "timeout" in fault_set:
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"TIMEOUT_STOP",
|
|
"A timeout stops and never retries.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
)
|
|
machine.transition("TARGET_QUIESCENT")
|
|
machine.transition("SWITCH_NOT_AUTHORIZED")
|
|
|
|
switch_auth = _authorization(
|
|
"switch", component, identity["candidate_sha256"], fault_set
|
|
)
|
|
if not switch_auth.valid(
|
|
"switch", component, identity["candidate_sha256"], now_tick=1
|
|
):
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"SWITCH_AUTHORIZATION_INVALID",
|
|
"Switch approval is separate and hash-bound.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
)
|
|
machine.transition("SWITCH_AUTHORIZED")
|
|
|
|
platform_unproven = (
|
|
not prove_virtual_switch_model
|
|
or "switch_primitive_unknown" in fault_set
|
|
or "directory_durability_unknown" in fault_set
|
|
or "in_place_overwrite" in fault_set
|
|
or "two_step_rename_gap" in fault_set
|
|
)
|
|
if platform_unproven:
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"NO_PROVEN_POWER_LOSS_SAFE_SWITCH",
|
|
"Atomicity and directory durability are not proven for the platform.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
"CATASTROPHIC",
|
|
)
|
|
|
|
machine.transition("SWITCH_IN_PROGRESS")
|
|
if "atomic_switch_failure" in fault_set:
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"ATOMIC_SWITCH_FAILED",
|
|
"The virtual atomic switch failed and left the old live object intact.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
"CATASTROPHIC",
|
|
)
|
|
|
|
live = copy.deepcopy(candidate)
|
|
live.path_token = expected.path_token
|
|
live.object_id = f"OBJECT::live-candidate::{component}"
|
|
writes.append("virtual_atomic_switch")
|
|
machine.transition("POST_SWITCH_VERIFY")
|
|
|
|
if "live_verification_failure" in fault_set:
|
|
machine.transition("ROLLBACK_REQUIRED")
|
|
return _report(
|
|
machine,
|
|
component,
|
|
"ROLLBACK_REQUIRED",
|
|
[
|
|
{
|
|
"code": "POST_SWITCH_VERIFY_FAILED",
|
|
"detail": (
|
|
"A separate rollback authorization is required; "
|
|
"execution remains forbidden."
|
|
),
|
|
}
|
|
],
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
"CATASTROPHIC",
|
|
)
|
|
|
|
if "rollback_hash_mismatch" in fault_set:
|
|
machine.transition("ROLLBACK_REQUIRED")
|
|
rollback_auth = _authorization(
|
|
"rollback", component, identity["stock_sha256"], set()
|
|
)
|
|
if not rollback_auth.valid(
|
|
"rollback", component, identity["stock_sha256"], now_tick=1
|
|
):
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"ROLLBACK_AUTHORIZATION_INVALID",
|
|
"Rollback authorization is absent.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
"CATASTROPHIC",
|
|
)
|
|
machine.transition("ROLLBACK_AUTHORIZED")
|
|
machine.transition("ROLLBACK_IN_PROGRESS")
|
|
live = copy.deepcopy(backup)
|
|
live.path_token = expected.path_token
|
|
live.object_id = f"OBJECT::restored::{component}"
|
|
writes.append("virtual_atomic_rollback")
|
|
if "rollback_hash_mismatch" in fault_set:
|
|
live.sha256 = "4" * 64
|
|
return _blocked(
|
|
machine,
|
|
component,
|
|
"ROLLBACK_VERIFY_FAILED",
|
|
"Rollback hash mismatch is catastrophic and remains blocked.",
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
"CATASTROPHIC",
|
|
)
|
|
machine.transition("ROLLBACK_VERIFIED")
|
|
return _report(
|
|
machine,
|
|
component,
|
|
"ROLLBACK_VERIFIED",
|
|
[],
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
)
|
|
|
|
machine.transition("MANUAL_EXECUTION_NOT_AUTHORIZED")
|
|
return _report(
|
|
machine,
|
|
component,
|
|
"DESIGN_MODEL_STOP_BEFORE_EXECUTION",
|
|
[
|
|
{
|
|
"code": "MANUAL_EXECUTION_NOT_AUTHORIZED",
|
|
"detail": "Execution is a later separate gate and is not simulated here.",
|
|
}
|
|
],
|
|
live,
|
|
backup,
|
|
candidate,
|
|
writes,
|
|
)
|
|
|
|
|
|
def simulate_power_loss_boundary(
|
|
component: str, boundary: str, *, prove_virtual_switch_model: bool = False
|
|
) -> dict[str, Any]:
|
|
"""Classify one abstract interruption without performing any real write."""
|
|
|
|
if component not in COMPONENTS:
|
|
raise ValueError("unknown component")
|
|
if boundary not in POWER_LOSS_BOUNDARIES:
|
|
raise ValueError("unknown power-loss boundary")
|
|
if not prove_virtual_switch_model:
|
|
return {
|
|
"boundary": boundary,
|
|
"result": "UNPROVEN",
|
|
"implementation_blocked": True,
|
|
"reason": "PS5 atomicity and durability are not proven",
|
|
"host_simulation_only": True,
|
|
"hardware_evidence": False,
|
|
"automatic_start": False,
|
|
"automatic_retry": False,
|
|
"crash_invariant": "C_REJECTED_UNSAFE_OR_UNPROVEN",
|
|
}
|
|
after_switch = boundary in {
|
|
"immediately_after_live_switch",
|
|
"after_switch_before_live_hash",
|
|
"after_live_hash_before_execution",
|
|
"during_first_manual_execution",
|
|
"during_rollback",
|
|
}
|
|
return {
|
|
"boundary": boundary,
|
|
"result": "SAFE_IN_VIRTUAL_MODEL_ONLY",
|
|
"implementation_blocked": False,
|
|
"reason": "the synthetic atomic primitive chooses one complete identity",
|
|
"host_simulation_only": True,
|
|
"hardware_evidence": False,
|
|
"automatic_start": False,
|
|
"automatic_retry": False,
|
|
"crash_invariant": (
|
|
"B_NEW_LIVE_COMPLETE_AND_BACKUP_INTACT"
|
|
if after_switch
|
|
else "A_OLD_LIVE_COMPLETE"
|
|
),
|
|
}
|
|
|
|
|
|
def run_fault_suite(component: str) -> dict[str, Any]:
|
|
results: list[dict[str, Any]] = []
|
|
for fault in FAULTS:
|
|
report = simulate_transaction(
|
|
component,
|
|
[fault],
|
|
prove_virtual_switch_model=True,
|
|
)
|
|
results.append(
|
|
{
|
|
"fault": fault,
|
|
"status": report["status"],
|
|
"state": report["state"],
|
|
"risk": report["risk"],
|
|
"crash_invariant": report["crash_invariant"],
|
|
"target_execution_performed": report[
|
|
"target_execution_performed"
|
|
],
|
|
"automatic_retry": report["automatic_retry"],
|
|
}
|
|
)
|
|
power_loss = [
|
|
simulate_power_loss_boundary(
|
|
component, boundary, prove_virtual_switch_model=True
|
|
)
|
|
for boundary in POWER_LOSS_BOUNDARIES
|
|
]
|
|
return {
|
|
"schema_contract": "chimera-gfx-phase-0.9-host-fault-suite-v1",
|
|
"phase": "PHASE_0_9A_OFFLINE_ANTI_BRICK",
|
|
"component": component,
|
|
"fault_results": results,
|
|
"power_loss_results": power_loss,
|
|
"host_simulation_only": True,
|
|
"hardware_evidence": False,
|
|
"ps5_connected": False,
|
|
"device_write_performed": False,
|
|
"files_transferred": False,
|
|
"target_execution_performed": False,
|
|
"target_artifact_created": False,
|
|
"automatic_retry": False,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument(
|
|
"--component",
|
|
choices=tuple(COMPONENTS),
|
|
required=True,
|
|
)
|
|
parser.add_argument("--fault", choices=FAULTS, action="append", default=[])
|
|
parser.add_argument(
|
|
"--power-loss-boundary",
|
|
choices=POWER_LOSS_BOUNDARIES,
|
|
)
|
|
parser.add_argument("--run-fault-suite", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
if args.run_fault_suite and (args.fault or args.power_loss_boundary):
|
|
parser.error("--run-fault-suite cannot be combined with another scenario")
|
|
if args.power_loss_boundary and args.fault:
|
|
parser.error("--power-loss-boundary cannot be combined with --fault")
|
|
|
|
if args.run_fault_suite:
|
|
report = run_fault_suite(args.component)
|
|
elif args.power_loss_boundary:
|
|
report = simulate_power_loss_boundary(
|
|
args.component, args.power_loss_boundary
|
|
)
|
|
else:
|
|
report = simulate_transaction(args.component, args.fault)
|
|
print(json.dumps(report, indent=2, sort_keys=True))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|