safety: implement Phase 0 (F1-F5) and Phase 1 (safety layer + Stream 1)

src/cipherrescue/orchestration/__init__.py

@@ -5,21 +5,22 @@
 
     INIT → ENUMERATE → DETECT → DIAGNOSE → AUTH → SELECT → CONFIRM → EXECUTE → REPORT
 
-State invariants:
-    - No write can occur before CONFIRM state (WriteBlocker enforced).
-    - Every transition is logged to AuditLog.
-    - AUTH failure returns to DETECT, not ENUMERATE (device persists).
-    - CONFIRM requires explicit operator acknowledgement in the TUI.
+State invariants enforced by transition():
+    - Adjacency: only permitted edges in VALID_TRANSITIONS may be traversed.
+    - AUTH failure returns to DETECT (not ENUMERATE) — device context persists.
+    - EXECUTE requires backup_token is not None (Theorem 6.1).
+    - Any state may transition to ABORTED.
+    - ABORTED is a terminal state — no transitions out.
+    - Every permitted and rejected transition is logged to AuditLog.
 
 Reference: spec §6 (Layer 6 — Orchestration Engine), Theorem 6.1
            (Orchestration Safety Property).
-
-Status: STUB — state machine implementation pending.
 """
 
 from __future__ import annotations
 
 import logging
+import os
 import uuid
 from enum import Enum, auto
 
@@ -41,32 +42,91 @@ class SessionState(Enum):
     ABORTED = auto()
 
 
+# Permitted transition graph per spec §6.
+# AUTH → DETECT is the AUTH-failure path (device persists, retry from detection).
+VALID_TRANSITIONS: dict[SessionState, set[SessionState]] = {
+    SessionState.INIT:      {SessionState.ENUMERATE, SessionState.ABORTED},
+    SessionState.ENUMERATE: {SessionState.DETECT, SessionState.ABORTED},
+    SessionState.DETECT:    {SessionState.DIAGNOSE, SessionState.ABORTED},
+    SessionState.DIAGNOSE:  {SessionState.AUTH, SessionState.ABORTED},
+    SessionState.AUTH:      {SessionState.SELECT, SessionState.DETECT, SessionState.ABORTED},
+    SessionState.SELECT:    {SessionState.CONFIRM, SessionState.ABORTED},
+    SessionState.CONFIRM:   {SessionState.EXECUTE, SessionState.ABORTED},
+    SessionState.EXECUTE:   {SessionState.REPORT, SessionState.ABORTED},
+    SessionState.REPORT:    {SessionState.ABORTED},
+    SessionState.ABORTED:   set(),
+}
+
+
+class CipherRescueError(Exception):
+    """Base exception for all CipherRescue errors."""
+
+
+class InvalidTransitionError(CipherRescueError):
+    """Raised when a transition violates the session state machine graph."""
+
+
+class MissingBackupTokenError(InvalidTransitionError):
+    """Raised when EXECUTE is attempted without a registered backup_token."""
+
+
 class SessionContext:
     """
     Holds all mutable state for a single recovery session.
 
     Attributes:
-        session_id:     Unique session identifier (UUID4).
-        state:          Current state machine state.
-        device_path:    Target block device path.
-        audit_log:      Append-only tamper-evident log for this session.
-        diagnosis:      SCPRSolution from Layer 5 (set after DIAGNOSE).
-        auth_token:     AuthToken from the plugin (set after AUTH).
-        backup_token:   BackupToken from BackupManager (set after CONFIRM).
+        session_id:      Unique session identifier (UUID4).
+        session_key:     32-byte random key for HMAC signing (audit log + backup tokens).
+        state:           Current state machine state.
+        device_path:     Target block device path.
+        audit_log:       Append-only tamper-evident log for this session.
+        diagnosis:       SCPRSolution from Layer 5 (set after DIAGNOSE).
+        auth_token:      AuthToken from the plugin (set after AUTH).
+        backup_token:    BackupToken from BackupManager (set after CONFIRM).
         selected_action: Action chosen by operator (set in SELECT).
     """
 
     def __init__(self, authority: Authority) -> None:
         self.session_id = str(uuid.uuid4())
+        self.session_key: bytes = os.urandom(32)
         self.state = SessionState.INIT
         self.device_path: str = ""
-        self.audit_log = AuditLog(self.session_id, authority)
+        self.audit_log = AuditLog(
+            self.session_id, authority, session_key=self.session_key
+        )
         self.diagnosis = None
         self.auth_token = None
         self.backup_token = None
         self.selected_action = None
 
     def transition(self, target: SessionState) -> None:
+        """
+        Attempt to move the session to target state.
+
+        Raises:
+            InvalidTransitionError:   If target is not in the permitted
+                                      successor set for the current state.
+            MissingBackupTokenError:  If transitioning to EXECUTE without
+                                      a registered backup_token (Theorem 6.1).
+
+        Rejected transitions are logged to the audit log before raising.
+        """
+        permitted = VALID_TRANSITIONS.get(self.state, set())
+
+        if target not in permitted:
+            self.audit_log.log_rejected_transition(self.state.name, target.name)
+            raise InvalidTransitionError(
+                f"Invalid transition {self.state.name} → {target.name}. "
+                f"Permitted: {', '.join(s.name for s in permitted) or 'none (terminal state)'}."
+            )
+
+        if target == SessionState.EXECUTE and self.backup_token is None:
+            self.audit_log.log_rejected_transition(self.state.name, target.name)
+            raise MissingBackupTokenError(
+                "Cannot transition to EXECUTE: backup_token is None. "
+                "BackupManager.create_backup() must complete before execution."
+            )
+
         logger.info(
             "Session %s: %s → %s", self.session_id, self.state.name, target.name
         )
@@ -92,5 +152,7 @@ def start_session(self, authority: Authority) -> SessionContext:
 
     def abort(self, reason: str = "") -> None:
         if self.context:
-            logger.warning("Session %s aborted: %s", self.context.session_id, reason)
+            logger.warning(
+                "Session %s aborted: %s", self.context.session_id, reason
+            )
             self.context.transition(SessionState.ABORTED)

src/cipherrescue/plugins/__init__.py

@@ -38,8 +38,11 @@ class AuthToken:
     Proof that the operator has successfully authenticated to the scheme.
 
     Issued by SchemePlugin.authenticate().  Required for any action that
-    reads key material or modifies the device.  Zeroed immediately after
-    use via ctypes.memset (see spec §7.2).
+    reads key material or modifies the device.
+
+    Status: credential zeroing (mlock + ctypes.memset per spec §7.2)
+    is not yet implemented.  Scheduled for Phase 1 implementation.
+    Tracked in GitHub issue: CipherRescue Phase 1 — credential zeroing.
     """
 
     scheme: str

src/cipherrescue/plugins/luks2_plugin.py

@@ -19,8 +19,8 @@
 import logging
 from typing import Any
 
-from ...safety.write_blocker import BackupToken
-from .._base import Action, AuthToken, PluginError, SchemePlugin
+from ..safety.write_blocker import BackupToken
+from . import Action, AuthToken, PluginError, SchemePlugin
 
 logger = logging.getLogger(__name__)
 

src/cipherrescue/safety/__init__.py

@@ -1,5 +1,6 @@
 """Layer 3 — Safety & Audit Layer."""
 
+from .backup_manager import BackupError, BackupManager
 from .write_blocker import BackupToken, WriteBlocker
 
-__all__ = ["WriteBlocker", "BackupToken"]
+__all__ = ["WriteBlocker", "BackupToken", "BackupManager", "BackupError"]

src/cipherrescue/safety/audit_log.py

@@ -3,23 +3,26 @@
 
 Implements Theorem 3.3 (Log Tamper Evidence):
 
-    The audit log is hash-chained: each entry eₙ includes
-    H(eₙ₋₁ ‖ payload), making retrospective alteration detectable.
+    The audit log is HMAC-keyed and hash-chained: each entry eₙ includes
+    H(eₙ₋₁ ‖ payload) and mac_n = HMAC_K(entry_hash_n), making
+    retrospective alteration detectable even by an attacker who knows
+    the chain structure but not the session key K.
 
 Each session produces a self-contained, append-only log that records:
     - Authority declaration (device owner / authorised representative /
       law enforcement / corporate IT)
-    - Every state machine transition
+    - Every state machine transition (including rejected transitions)
     - All write operations and their BackupTokens
     - The full SCPRSolution (optimal reasons + dual weights)
     - Any uncovered signals flagged for operator review
 
-Status: STUB — hash chaining and DFXML export pending implementation.
+Status: hash chaining and HMAC keying implemented. DFXML export pending.
 """
 
 from __future__ import annotations
 
 import hashlib
+import hmac as _hmac
 import json
 import time
 from dataclasses import dataclass, field
@@ -42,6 +45,7 @@ class LogEntry:
     payload: dict[str, Any]
     prev_hash: str
     entry_hash: str = field(init=False)
+    mac: str = field(init=False, default="")
 
     def __post_init__(self) -> None:
         raw = json.dumps(
@@ -56,20 +60,33 @@ def __post_init__(self) -> None:
         )
         self.entry_hash = hashlib.sha256(raw.encode()).hexdigest()
 
+    def set_mac(self, session_key: bytes) -> None:
+        """Compute and set HMAC_K(entry_hash). Called by AuditLog._append()."""
+        self.mac = _hmac.new(
+            session_key, self.entry_hash.encode(), hashlib.sha256
+        ).hexdigest()
+
 
 class AuditLog:
     """
     Tamper-evident, append-only audit log for a recovery session.
 
-    The chain can be verified at any time with verify_chain().
-    Any modification to a historical entry breaks all subsequent hashes.
+    Each entry is SHA-256 hash-chained and HMAC'd under the session key K
+    (Definition 3.1). verify_chain() checks both the hash linkage and the
+    per-entry MAC, so the chain is not regeneratable without K.
     """
 
     GENESIS_HASH = "0" * 64
 
-    def __init__(self, session_id: str, authority: Authority) -> None:
+    def __init__(
+        self,
+        session_id: str,
+        authority: Authority,
+        session_key: bytes = b"",
+    ) -> None:
         self.session_id = session_id
         self.authority = authority
+        self._session_key = session_key
         self._entries: list[LogEntry] = []
         self._append(
             "SESSION_OPEN",
@@ -88,12 +105,19 @@ def _append(self, event_type: str, payload: dict[str, Any]) -> LogEntry:
             payload=payload,
             prev_hash=prev,
         )
+        entry.set_mac(self._session_key)
         self._entries.append(entry)
         return entry
 
     def log_state_transition(self, from_state: str, to_state: str) -> None:
         self._append("STATE_TRANSITION", {"from": from_state, "to": to_state})
 
+    def log_rejected_transition(self, from_state: str, to_state: str) -> None:
+        """Log an attempted transition that was rejected by the state machine."""
+        self._append(
+            "REJECTED_TRANSITION", {"from": from_state, "to": to_state}
+        )
+
     def log_diagnosis(self, solution_repr: str, uncovered_count: int) -> None:
         self._append(
             "SCPR_DIAGNOSIS",
@@ -114,12 +138,19 @@ def log_write(self, device: str, backup_sha256: str, action: str) -> None:
         )
 
     def verify_chain(self) -> bool:
-        """Return True iff the entire chain is unmodified."""
+        """
+        Return True iff the entire chain is unmodified.
+
+        Checks per entry (in order):
+          1. Hash-chain linkage: entry.prev_hash == previous entry's entry_hash.
+          2. Payload integrity: recomputed entry_hash matches stored entry_hash.
+          3. MAC integrity: HMAC_K(entry_hash) matches stored entry.mac.
+        """
         for i, entry in enumerate(self._entries):
             prev = self._entries[i - 1].entry_hash if i > 0 else self.GENESIS_HASH
             if entry.prev_hash != prev:
                 return False
-            # Recompute hash and compare
+
             raw = json.dumps(
                 {
                     "seq": entry.sequence,
@@ -132,6 +163,13 @@ def verify_chain(self) -> bool:
             )
             if hashlib.sha256(raw.encode()).hexdigest() != entry.entry_hash:
                 return False
+
+            expected_mac = _hmac.new(
+                self._session_key, entry.entry_hash.encode(), hashlib.sha256
+            ).hexdigest()
+            if not _hmac.compare_digest(expected_mac, entry.mac):
+                return False
+
         return True
 
     def export_json(self) -> str: