diff --git a/CAPABILITIES.md b/CAPABILITIES.md index 6638049748..6068a996cf 100644 --- a/CAPABILITIES.md +++ b/CAPABILITIES.md @@ -317,7 +317,7 @@ improvising or retrying when something is off. | See a compiled program as a map (every step, exact edge, loop and stop rule, with each step's checks) and play a recorded run over it; one React package that Desktop, app.openadapt.ai and openadapt.ai share | On main ([packages/program-graph](packages/program-graph/README.md)) | Real compiled OpenEMR referral program and a recorded scoreboard run, both on synthetic data | `@openadapt/program-graph`; `openadapt-flow visualize --format json` | | Named sections for every compiled program ("Who: title, first name, last name", "Contact", "Create New Patient"), found without a model from the save, the recorded form structure, the fields and any narration; a person can rename, merge or split them on the review screen | On main | Unit tests; the certified OpenEMR referral bundle and a real recording of it on a local test copy with made-up patients | `openadapt-flow sections`; [SECTIONS.md](packages/flow/docs/SECTIONS.md) | | Encrypt PHI at rest: every bundle file that can hold patient data under one sealed index (unknown files refused), web recordings from the first frame, and the run directories of sealed bundles; compile, induce, scaffold-verifier, qualify, admit, run, resume, undo, visualize, explain, emit-skill, for-each, sections, check mark, compose, process, variants, route, teach and push read them in memory through one artifact reader and refuse without the key; teach keeps its skill library and taught bundle sealed, and push keeps its sanitized review copy and approved archive encrypted until the upload | On main | Synthetic (no-plaintext audit-hook test over record, compile and a VERIFIED run; per-frame cost measured) | `openadapt-flow seal`, `record`; [phi_at_rest.md](packages/flow/docs/phi_at_rest.md) | -| Encryption at rest under a configured or locally generated key (0600): durable checkpoints by default; with `OPENADAPT_FLOW_AT_REST=strict`, every recording, bundle and run directory too. Each plaintext write is logged on stderr and in a ledger that Desktop shows as a notice | On main | Synthetic (key, opt-out, strict-mode and run-directory tests) | `openadapt_flow.at_rest`; [phi_at_rest.md](packages/flow/docs/phi_at_rest.md#default-opt-out-and-strict-mode) | +| Encryption at rest under a configured or locally generated key (0600): durable checkpoints and attended state by default; with `OPENADAPT_FLOW_AT_REST=strict`, every recording, bundle and run directory too. Each plaintext write is logged on stderr and in a ledger that Desktop shows as a notice | On main | Synthetic (key, opt-out, strict-mode, run-directory and attended-state audit-hook tests) | `openadapt_flow.at_rest`; [phi_at_rest.md](packages/flow/docs/phi_at_rest.md#default-opt-out-and-strict-mode) | | Signed receipts (Seals), countersigned and timestamped, offline by default | Released | Unit tests, real TSA tokens | `openadapt-flow seal`, `countersign`, `local-seal` | | Offline Seals signed with a key the customer holds: a key file, the OS keychain, or a PKCS#11 token or HSM; pinned keys with rotation and revocation; no fallback to a generated key | On main ([#1392](https://github.com/OpenAdaptAI/OpenAdapt/pull/1392)) | Unit tests with synthetic data (keychain and PKCS#11 through stand-ins) | `openadapt-flow local-seal --key`, `local-seal key create`, `pin`, `revoke`; `openadapt seal verify FOLDER` checks every sealed run in a folder; [OFFLINE_SEALS.md](packages/flow/docs/OFFLINE_SEALS.md#customer-held-keys) | | Anchor an offline Seal in the public Sigstore Rekor transparency log (opt-in; sends one digest and the public key), and verify the anchor offline from the stored inclusion proof | On main ([#1394](https://github.com/OpenAdaptAI/OpenAdapt/pull/1394)) | Unit tests against a simulated Rekor log; real public Rekor entries and Sigstore TSA tokens for the landing demo's referral runs, verified offline in tests; Ed25519ph matches the RFC 8032 test vector | `openadapt-flow local-seal anchor`, `local-seal verify --require-anchor`; [OFFLINE_SEALS.md](packages/flow/docs/OFFLINE_SEALS.md#optional-anchor-a-seal-in-a-public-transparency-log) | diff --git a/llms-full.txt b/llms-full.txt index 4b1c6986b2..9c66cf8478 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -720,7 +720,7 @@ improvising or retrying when something is off. | See a compiled program as a map (every step, exact edge, loop and stop rule, with each step's checks) and play a recorded run over it; one React package that Desktop, app.openadapt.ai and openadapt.ai share | On main ([packages/program-graph](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/program-graph/README.md)) | Real compiled OpenEMR referral program and a recorded scoreboard run, both on synthetic data | `@openadapt/program-graph`; `openadapt-flow visualize --format json` | | Named sections for every compiled program ("Who: title, first name, last name", "Contact", "Create New Patient"), found without a model from the save, the recorded form structure, the fields and any narration; a person can rename, merge or split them on the review screen | On main | Unit tests; the certified OpenEMR referral bundle and a real recording of it on a local test copy with made-up patients | `openadapt-flow sections`; [SECTIONS.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/SECTIONS.md) | | Encrypt PHI at rest: every bundle file that can hold patient data under one sealed index (unknown files refused), web recordings from the first frame, and the run directories of sealed bundles; compile, induce, scaffold-verifier, qualify, admit, run, resume, undo, visualize, explain, emit-skill, for-each, sections, check mark, compose, process, variants, route, teach and push read them in memory through one artifact reader and refuse without the key; teach keeps its skill library and taught bundle sealed, and push keeps its sanitized review copy and approved archive encrypted until the upload | On main | Synthetic (no-plaintext audit-hook test over record, compile and a VERIFIED run; per-frame cost measured) | `openadapt-flow seal`, `record`; [phi_at_rest.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/phi_at_rest.md) | -| Encryption at rest under a configured or locally generated key (0600): durable checkpoints by default; with `OPENADAPT_FLOW_AT_REST=strict`, every recording, bundle and run directory too. Each plaintext write is logged on stderr and in a ledger that Desktop shows as a notice | On main | Synthetic (key, opt-out, strict-mode and run-directory tests) | `openadapt_flow.at_rest`; [phi_at_rest.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/phi_at_rest.md#default-opt-out-and-strict-mode) | +| Encryption at rest under a configured or locally generated key (0600): durable checkpoints and attended state by default; with `OPENADAPT_FLOW_AT_REST=strict`, every recording, bundle and run directory too. Each plaintext write is logged on stderr and in a ledger that Desktop shows as a notice | On main | Synthetic (key, opt-out, strict-mode, run-directory and attended-state audit-hook tests) | `openadapt_flow.at_rest`; [phi_at_rest.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/phi_at_rest.md#default-opt-out-and-strict-mode) | | Signed receipts (Seals), countersigned and timestamped, offline by default | Released | Unit tests, real TSA tokens | `openadapt-flow seal`, `countersign`, `local-seal` | | Offline Seals signed with a key the customer holds: a key file, the OS keychain, or a PKCS#11 token or HSM; pinned keys with rotation and revocation; no fallback to a generated key | On main ([#1392](https://github.com/OpenAdaptAI/OpenAdapt/pull/1392)) | Unit tests with synthetic data (keychain and PKCS#11 through stand-ins) | `openadapt-flow local-seal --key`, `local-seal key create`, `pin`, `revoke`; `openadapt seal verify FOLDER` checks every sealed run in a folder; [OFFLINE_SEALS.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/OFFLINE_SEALS.md#customer-held-keys) | | Anchor an offline Seal in the public Sigstore Rekor transparency log (opt-in; sends one digest and the public key), and verify the anchor offline from the stored inclusion proof | On main ([#1394](https://github.com/OpenAdaptAI/OpenAdapt/pull/1394)) | Unit tests against a simulated Rekor log; real public Rekor entries and Sigstore TSA tokens for the landing demo's referral runs, verified offline in tests; Ed25519ph matches the RFC 8032 test vector | `openadapt-flow local-seal anchor`, `local-seal verify --require-anchor`; [OFFLINE_SEALS.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/OFFLINE_SEALS.md#optional-anchor-a-seal-in-a-public-transparency-log) | diff --git a/packages/flow/docs/phi_at_rest.md b/packages/flow/docs/phi_at_rest.md index 5b4e9dbfe0..6d956a23fa 100644 --- a/packages/flow/docs/phi_at_rest.md +++ b/packages/flow/docs/phi_at_rest.md @@ -22,7 +22,7 @@ with your other secrets. | | Default | `OPENADAPT_FLOW_AT_REST=strict` | | --- | --- | --- | -| Durable checkpoints | Encrypted | Encrypted | +| Durable checkpoints and attended state | Encrypted | Encrypted | | Web recordings | Encrypted with a configured key; otherwise logged | Encrypted | | Bundles from `compile`, `induce`, `admit` | Encrypted from encrypted input; otherwise logged | Encrypted | | Run directories | Encrypted for a sealed bundle; otherwise logged | Encrypted | @@ -190,6 +190,36 @@ step with them. The approved archive beside it (`.approved.zip`) is one AES-256-GCM container under the same key. +### Durable checkpoints and attended state + +In the default mode, under the configured or local key, the durable store +writes its checkpoints, pauses and approvals as `.enc` AES-256-GCM +containers. The attended store does the same for the pause capability and its +history, the decision log (operator and message), program and reconciliation +receipts, and program-transition control frames. Each file's associated data +names its run-relative path, so a file copied over another path does not +decrypt, and once a run's attended state is encrypted a plaintext file in its +place is refused. The external attended journal stores each decision-log +snapshot encrypted and bound to its run and sequence; its hash chain and HMAC +cover the plaintext. A run whose attended files are plaintext from an earlier +version still resumes, and its first encrypted write converts them. +The key is stretched with scrypt once per run, under a random salt kept in the +run's `.attended_state.encrypted` marker, and each file and journal row gets +its own nonce under keys derived from it, so the cost doesn't grow with the +number of files or journal rows. Each file names the salt it was sealed +under, and a file whose salt is not the one in the run's marker is refused, so +a file copied in from another run under the same key does not read. If two +processes convert the same run at once, the one that loses the race to write +the marker reseals its files under the winner's salt. The replayer classifies +a run's outcome with its own checkpoint key, so a run whose key was passed in +code rather than through the environment still reads its sealed evidence. +Once a run's attended state is encrypted, +`OPENADAPT_FLOW_PLAINTEXT_AT_REST` does not turn it back to plaintext: a write +with no key is refused. Containers written by the earlier per-file format +still read. +`tests/test_attended_at_rest.py` runs attended decisions under the same disk +audit hook and checks that no attended file or journal row holds plaintext. + ### Run directory of a sealed bundle Every file is encrypted except these structural ones: `.durable_run.claim` diff --git a/packages/flow/openadapt_flow/crypto.py b/packages/flow/openadapt_flow/crypto.py index 5c81dd276a..d2dc2bff79 100644 --- a/packages/flow/openadapt_flow/crypto.py +++ b/packages/flow/openadapt_flow/crypto.py @@ -170,6 +170,8 @@ def is_encrypted(data: bytes) -> bool: """ if data[: len(SET_FILE_MAGIC)] == SET_FILE_MAGIC: return True + if data[: len(SALTED_FILE_MAGIC)] == SALTED_FILE_MAGIC: + return True try: obj = json.loads(data) except (ValueError, TypeError): @@ -305,6 +307,25 @@ def open(self, container: bytes, *, aad: bytes) -> bytes: ) from exc +#: Prefix of a container sealed under a key derived once per run (see +#: :mod:`openadapt_flow.runtime.durable.attended_state`): this prefix, the +#: run's 16-byte salt, a random 96-bit nonce, then the AES-256-GCM ciphertext. +SALTED_FILE_MAGIC: Final[bytes] = b"OAFLOW-SALTED-1\n" +SALT_LEN: Final[int] = _SALT_LEN +NONCE_LEN: Final[int] = _NONCE_LEN + + +def derive_data_key(passphrase: str, salt: bytes) -> bytes: + """The 256-bit scrypt key for ``passphrase`` and ``salt``, at the work + factors new containers record. A caller that derives once and caches the + key must keep it in memory only.""" + if not passphrase: + raise MissingKeyError("no encryption passphrase configured") + if len(salt) != _SALT_LEN: + raise CryptoError("a data-key salt must be 16 bytes") + return _derive(passphrase, salt, n=_SCRYPT_N, r=_SCRYPT_R, p=_SCRYPT_P) + + def kdf_parameters() -> dict[str, int]: """The scrypt work factors new containers record.""" return {"kdf_n": _SCRYPT_N, "kdf_r": _SCRYPT_R, "kdf_p": _SCRYPT_P} diff --git a/packages/flow/openadapt_flow/execution_profiles.py b/packages/flow/openadapt_flow/execution_profiles.py index 0ebfe37c9e..ceeb706779 100644 --- a/packages/flow/openadapt_flow/execution_profiles.py +++ b/packages/flow/openadapt_flow/execution_profiles.py @@ -486,6 +486,7 @@ def _program_action_trace( attended_transition_evidence: list[Any] | None = None, business_decision_evidence: list[Any] | None = None, transition_evidence_root: Path | None = None, + transition_evidence_key: str | None = None, transition_predicate_vision: Any | None = None, governed_runtime_inputs_digest: str | None = None, run_id_sha256: str | None = None, @@ -549,7 +550,9 @@ def _program_action_trace( return None from openadapt_flow.runtime.durable.checkpoint import CheckpointStore - manifest = CheckpointStore(transition_evidence_root).read_manifest() + manifest = CheckpointStore( + transition_evidence_root, key=transition_evidence_key + ).read_manifest() if manifest is None: return None trace_base_params = dict(manifest.params) @@ -652,7 +655,11 @@ def _matching_evidence_group( raise ValueError("transition evidence decision rows disagree") return group - def _verified_inventory_bytes(ref: str, expected_sha256: str) -> bytes: + def _verified_inventory_bytes( + ref: str, expected_sha256: str, *, attended: bool = False + ) -> bytes: + if attended: + return _verified_attended_bytes(ref, expected_sha256) if transition_evidence_root is None: raise ValueError("visual transition evidence has no local evidence root") root = transition_evidence_root.resolve() @@ -683,6 +690,52 @@ def _verified_inventory_bytes(ref: str, expected_sha256: str) -> bytes: raise ValueError("transition evidence digest does not match") return payload + def _verified_attended_bytes(ref: str, expected_sha256: str) -> bytes: + """Plaintext of an attended receipt or control file, which may be + encrypted at rest as ``.enc`` (``attended_state``).""" + from openadapt_flow import crypto as _crypto + from openadapt_flow.runtime.durable import attended_state as _state + + if transition_evidence_root is None: + raise ValueError("visual transition evidence has no local evidence root") + from openadapt_flow import sealed_set as _sealed + + root = transition_evidence_root.resolve() + encrypted = root.joinpath(*PurePosixPath(ref + _state.ENC_SUFFIX).parts) + if _sealed.is_sealed_path(root): + # A sealed run directory's files are the sealed-set writer's. + return _verified_inventory_bytes(ref, expected_sha256) + if not (encrypted.exists() or encrypted.is_symlink()): + if (root / _state.MARKER_FILENAME).exists(): + raise ValueError("attended evidence is plaintext where encrypted") + return _verified_inventory_bytes(ref, expected_sha256) + parts = PurePosixPath(ref + _state.ENC_SUFFIX).parts + cursor_path = root + for part in parts: + cursor_path /= part + if cursor_path.is_symlink(): + raise ValueError("transition frame path contains a symlink") + try: + path = encrypted.resolve(strict=True) + path.relative_to(root) + container = path.read_bytes() + except (OSError, ValueError) as exc: + raise ValueError("attended transition evidence is unreadable") from exc + try: + payload = _state.open_file( + root, + root.joinpath(*PurePosixPath(ref).parts), + container, + _crypto.resolve_key(transition_evidence_key), + ) + except _state.AttendedStateUnreadable as exc: + raise ValueError("attended transition evidence does not decrypt") from exc + import hashlib + + if hashlib.sha256(payload).hexdigest() != expected_sha256: + raise ValueError("transition evidence digest does not match") + return payload + def _validated_evidence_target( *, graph_id: str, @@ -835,6 +888,7 @@ def _validated_attended_target( payload = _verified_inventory_bytes( item.receipt_inventory_ref, item.receipt_sha256, + attended=True, ) from openadapt_flow.runtime.durable.attended import ( AttendedActionRefused, @@ -854,9 +908,11 @@ def _validated_attended_target( try: parsed = ProgramTransitionReceipt.model_validate_json(payload) authenticated = AttendedActionStore( - transition_evidence_root + transition_evidence_root, key=transition_evidence_key ).read_program_receipt(item.receipt_pause_id) - manifest = CheckpointStore(transition_evidence_root).read_manifest() + manifest = CheckpointStore( + transition_evidence_root, key=transition_evidence_key + ).read_manifest() except (OSError, ValueError, AttendedActionRefused) as exc: raise ValueError("attended transition receipt does not verify") from exc if ( @@ -895,6 +951,7 @@ def _validated_attended_target( control_payload = _verified_inventory_bytes( item.control_frames_inventory_ref, item.control_frames_sha256, + attended=True, ) try: control_frames = [ @@ -970,8 +1027,12 @@ def _validated_business_decision( ) try: - store = BusinessDecisionStore(transition_evidence_root) - manifest = CheckpointStore(transition_evidence_root).read_manifest() + store = BusinessDecisionStore( + transition_evidence_root, checkpoint_key=transition_evidence_key + ) + manifest = CheckpointStore( + transition_evidence_root, key=transition_evidence_key + ).read_manifest() if manifest is None: raise ValueError("business decision has no durable manifest") expected_bundle = bundle_version(manifest.bundle_dir) @@ -1449,6 +1510,7 @@ def classify_execution_outcome( transition_evidence_root: Path | None = None, transition_predicate_vision: Any | None = None, _qualification_fault_target_step_id: str | None = None, + transition_evidence_key: str | None = None, ) -> ExecutionOutcome: """Classify a completed report without changing legacy ``success``. @@ -1651,6 +1713,7 @@ def classify_execution_outcome( attended_transition_evidence=(report.attended_program_transition_evidence), business_decision_evidence=report.business_decision_evidence, transition_evidence_root=transition_evidence_root, + transition_evidence_key=transition_evidence_key, transition_predicate_vision=transition_predicate_vision, governed_runtime_inputs_digest=report.governed_runtime_inputs_digest, run_id_sha256=report.run_id_sha256, @@ -2207,8 +2270,12 @@ def stamp_execution_outcome( runtime_worklists: Mapping[str, list[dict[str, str]]] | None = None, transition_evidence_root: Path | None = None, transition_predicate_vision: Any | None = None, + transition_evidence_key: str | None = None, ) -> ExecutionOutcome: - """Write the profile and precise outcome into ``report``.""" + """Write the profile and precise outcome into ``report``. + + ``transition_evidence_key`` is the passphrase the run sealed its attended + state under; when it is ``None`` the usual key resolution applies.""" resolved = resolve_execution_profile(profile) report.external_network_calls = _external_network_call_state(report) @@ -2224,6 +2291,7 @@ def stamp_execution_outcome( runtime_worklists=runtime_worklists, transition_evidence_root=transition_evidence_root, transition_predicate_vision=transition_predicate_vision, + transition_evidence_key=transition_evidence_key, ) report.execution_profile = resolved.value report.execution_outcome = outcome.value diff --git a/packages/flow/openadapt_flow/runtime/durable/attended.py b/packages/flow/openadapt_flow/runtime/durable/attended.py index 7068a3f78e..c79fca5a01 100644 --- a/packages/flow/openadapt_flow/runtime/durable/attended.py +++ b/packages/flow/openadapt_flow/runtime/durable/attended.py @@ -51,6 +51,7 @@ project_step_safety, ) from openadapt_flow.runtime.authorization import RuntimeParamScalar +from openadapt_flow.runtime.durable import attended_state as _state from openadapt_flow.runtime.durable.approval import ( ApprovalRecord, ApprovalRequired, @@ -794,8 +795,14 @@ def _attended_step_safety( class AttendedActionStore: """Capability, single-flight lease, and append-only decision persistence.""" - def __init__(self, run_dir: Path | str) -> None: + def __init__(self, run_dir: Path | str, *, key: Optional[str] = None) -> None: self.run_dir = Path(run_dir) + # The at-rest key for this state: the explicit one, else the configured + # key, else the local key (crypto.resolve_key), resolved on first use. + # None under the recorded plaintext opt-out. + self._explicit_key = key + self._resolved_key: Optional[str] = None + self._key_resolved = False self.capability_path = self.run_dir / CAPABILITY_FILENAME self.capability_history_path = self.run_dir / CAPABILITY_HISTORY_FILENAME self.key_path = self.run_dir / CAPABILITY_KEY_FILENAME @@ -837,6 +844,60 @@ def _atomic_write(path: Path, payload: bytes, *, mode: int = 0o600) -> None: except FileNotFoundError: pass + @property + def key(self) -> Optional[str]: + """The at-rest key the attended state is encrypted under.""" + if not self._key_resolved: + from openadapt_flow import crypto as _crypto + + self._resolved_key = _crypto.resolve_key(self._explicit_key) + self._key_resolved = True + return self._resolved_key + + # -- attended state on disk ---------------------------------------------- + # + # A sealed run directory encrypts these files through its sealed-set + # writer (``_atomic_write``). Elsewhere each file is written as + # ``.enc`` under ``key`` (``attended_state``); older plaintext files + # still read until the run's first encrypted write converts them. + + def _state_is_file(self, path: Path) -> bool: + if _sealed.is_sealed_path(path): + return _sealed.is_file(path) + return _state.exists(path) + + def _state_read_bytes(self, path: Path) -> bytes: + if _sealed.is_sealed_path(path): + return _sealed.read_bytes(path) + # An older plaintext file needs no key; don't resolve one for it. + key = self.key if _state.enc_path(path).is_file() else None + return _state.read_bytes(self.run_dir, path, key) + + def _state_read_text(self, path: Path) -> str: + return self._state_read_bytes(path).decode("utf-8") + + def _state_write(self, path: Path, payload: bytes) -> None: + if _sealed.is_sealed_path(path): + self._atomic_write(path, payload) + return + _state.write_bytes( + self.run_dir, + path, + payload, + self.key, + lambda target, data: self._atomic_write(target, data), + ) + + def _state_unlink(self, path: Path) -> None: + if _sealed.is_sealed_path(path): + _sealed.unlink(path) + return + _state.unlink(path) + + def read_state_bytes(self, rel: str) -> bytes: + """Plaintext of one run-relative attended file, in either form.""" + return self._state_read_bytes(self.run_dir.joinpath(*Path(rel).parts)) + def _key(self, *, create: bool) -> bytes: try: key = self.key_path.read_bytes() @@ -998,21 +1059,21 @@ def _signed_capabilities(self) -> list[AttendedPauseCapability]: "the attended capability path must not be a symlink" ) capabilities: list[AttendedPauseCapability] = [] - if _sealed.is_file(self.capability_path): + if self._state_is_file(self.capability_path): try: capabilities.append( AttendedPauseCapability.model_validate_json( - _sealed.read_text(self.capability_path) + self._state_read_text(self.capability_path) ) ) except ValueError as exc: raise AttendedActionRefused( "the current attended capability is invalid" ) from exc - if _sealed.is_file(self.capability_history_path): + if self._state_is_file(self.capability_history_path): try: raw_history = json.loads( - _sealed.read_text(self.capability_history_path) + self._state_read_text(self.capability_history_path) ) if not isinstance(raw_history, list): raise ValueError("history is not a list") @@ -1214,14 +1275,14 @@ def write_reconciliation_receipt( path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) if os.name != "nt": os.chmod(path.parent, 0o700) - if _sealed.is_file(path): + if self._state_is_file(path): existing = self.read_reconciliation_receipt(sealed.pause_id) if existing != sealed: raise AttendedActionRefused( "a different reconciliation receipt already exists for this pause" ) return existing - self._atomic_write(path, sealed.model_dump_json(indent=2).encode("utf-8")) + self._state_write(path, sealed.model_dump_json(indent=2).encode("utf-8")) return sealed def read_reconciliation_receipt( @@ -1234,7 +1295,7 @@ def read_reconciliation_receipt( ) try: receipt = AttendedReconciliationReceipt.model_validate_json( - _sealed.read_text(path) + self._state_read_text(path) ) except (FileNotFoundError, ValueError) as exc: raise AttendedActionRefused( @@ -1279,7 +1340,7 @@ def write_program_receipt( path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) if os.name != "nt": os.chmod(path.parent, 0o700) - if _sealed.is_file(path): + if self._state_is_file(path): existing = self.read_program_receipt(sealed.pause_id) if existing != sealed: raise AttendedActionRefused( @@ -1287,7 +1348,7 @@ def write_program_receipt( "this pause" ) return existing - self._atomic_write(path, sealed.model_dump_json(indent=2).encode("utf-8")) + self._state_write(path, sealed.model_dump_json(indent=2).encode("utf-8")) return sealed def read_program_receipt(self, pause_id: str) -> ProgramTransitionReceipt: @@ -1299,7 +1360,7 @@ def read_program_receipt(self, pause_id: str) -> ProgramTransitionReceipt: ) try: receipt = ProgramTransitionReceipt.model_validate_json( - _sealed.read_text(path) + self._state_read_text(path) ) except (FileNotFoundError, ValueError) as exc: raise AttendedActionRefused( @@ -1383,7 +1444,7 @@ def issue( delivery_state, ) event_sequence = 1 - if _sealed.is_file(self.capability_path): + if self._state_is_file(self.capability_path): existing = self.read() if ( existing.pause_digest == pause_digest @@ -1407,10 +1468,10 @@ def issue( # an append-only history and let the engine issue the new pause; # browser callers still present the exact current digest. history: list[dict[str, Any]] = [] - if _sealed.is_file(self.capability_history_path): + if self._state_is_file(self.capability_history_path): try: raw_history = json.loads( - _sealed.read_text(self.capability_history_path) + self._state_read_text(self.capability_history_path) ) if isinstance(raw_history, list): history = [ @@ -1421,12 +1482,12 @@ def issue( "the attended capability history is invalid" ) from None history.append(existing.model_dump(mode="json")) - self._atomic_write( + self._state_write( self.capability_history_path, json.dumps(history, indent=2, sort_keys=True).encode("utf-8"), ) event_sequence = existing.event_sequence + 1 - elif _sealed.is_file(self.capability_history_path): + elif self._state_is_file(self.capability_history_path): historical = self._signed_capabilities() event_sequence = max(item.event_sequence for item in historical) + 1 now = _now() @@ -1462,7 +1523,7 @@ def issue( allowed_actions=allowed_actions, ) capability.signature = self._sign(capability, create_key=False) - self._atomic_write( + self._state_write( self.capability_path, capability.model_dump_json(indent=2).encode("utf-8"), ) @@ -1476,14 +1537,14 @@ def retire_current(self) -> None: must not remain visible as the current operator action surface. """ - if not _sealed.is_file(self.capability_path): + if not self._state_is_file(self.capability_path): return existing = self.read() history: list[dict[str, Any]] = [] - if _sealed.is_file(self.capability_history_path): + if self._state_is_file(self.capability_history_path): try: raw_history = json.loads( - _sealed.read_text(self.capability_history_path) + self._state_read_text(self.capability_history_path) ) if not isinstance(raw_history, list) or not all( isinstance(item, dict) for item in raw_history @@ -1497,17 +1558,17 @@ def retire_current(self) -> None: existing_payload = existing.model_dump(mode="json") if existing_payload not in history: history.append(existing_payload) - self._atomic_write( + self._state_write( self.capability_history_path, json.dumps(history, indent=2, sort_keys=True).encode("utf-8"), ) - _sealed.unlink(self.capability_path) + self._state_unlink(self.capability_path) self._fsync_parent(self.capability_path) def read(self) -> AttendedPauseCapability: try: capability = AttendedPauseCapability.model_validate_json( - _sealed.read_text(self.capability_path) + self._state_read_text(self.capability_path) ) except (FileNotFoundError, ValueError) as exc: raise AttendedActionRefused( @@ -1595,17 +1656,17 @@ def _read_log_with_head(self) -> tuple[AttendedDecisionLog, str]: CheckpointStore(self.run_dir), ) snapshot, head = authority.read_attended_snapshot( - expected_run_id=capability.run_id + expected_run_id=capability.run_id, key=self.key ) except DurableAuthorityBusy as exc: raise AttendedActionRefused( "the external attended decision journal is unavailable or invalid" ) from exc if snapshot is None: - if _sealed.is_file(self.decisions_path): + if self._state_is_file(self.decisions_path): try: local = AttendedDecisionLog.model_validate_json( - _sealed.read_text(self.decisions_path) + self._state_read_text(self.decisions_path) ) except (OSError, ValueError) as exc: raise AttendedActionRefused( @@ -1626,16 +1687,16 @@ def _read_log_with_head(self) -> tuple[AttendedDecisionLog, str]: payload = snapshot.encode("utf-8") try: local_matches = ( - _sealed.is_file(self.decisions_path) + self._state_is_file(self.decisions_path) and not self.decisions_path.is_symlink() - and _sealed.read_bytes(self.decisions_path) == payload + and self._state_read_bytes(self.decisions_path) == payload ) except OSError: local_matches = False if not local_matches: # The JSON file is a recoverable local projection. The append-only, # HMAC-chained SQLite journal outside the run directory is authority. - self._atomic_write(self.decisions_path, payload) + self._state_write(self.decisions_path, payload) return log, head def _append_log_snapshot( @@ -1661,12 +1722,13 @@ def _append_log_snapshot( expected_run_id=capability.run_id, expected_head_digest=expected_head, snapshot_json=payload, + key=self.key, ) except DurableAuthorityBusy as exc: raise AttendedActionRefused( "the external attended decision journal changed or is unavailable" ) from exc - self._atomic_write(self.decisions_path, payload.encode("utf-8")) + self._state_write(self.decisions_path, payload.encode("utf-8")) def prior(self, request: AttendedActionRequest) -> Optional[AttendedDecision]: request_digest = _digest(request) @@ -2165,7 +2227,7 @@ def issue_attended_capability( "the durable manifest has no stable run identity; cannot issue " "an attended mutation capability" ) - return AttendedActionStore(run_dir).issue( + return AttendedActionStore(run_dir, key=store.key).issue( manifest=manifest, pending=pending, workflow=workflow, @@ -2386,7 +2448,7 @@ def execute_attended_action( "the durable run identity or canonical path changed" ) from exc - actions = AttendedActionStore(run_dir) + actions = AttendedActionStore(run_dir, key=key) prior = actions.prior(request) if prior is not None: if prior.status in {"delivery_started", "delivery_uncertain"}: @@ -2893,7 +2955,7 @@ def checkpoint_human_completed_step( raise AttendedActionRefused( "the human-completed step's independent effect was not confirmed" ) - authenticated = AttendedActionStore(run_dir).signed_capability_for_digest( + authenticated = AttendedActionStore(run_dir, key=key).signed_capability_for_digest( capability.digest ) if authenticated != capability: @@ -3060,9 +3122,9 @@ def _load( raise AttendedActionRefused( "workflow identity changed after attended capability issuance" ) - authenticated = AttendedActionStore(run_dir).signed_capability_for_digest( - capability.digest - ) + authenticated = AttendedActionStore( + run_dir, key=self.key + ).signed_capability_for_digest(capability.digest) if authenticated != capability: raise AttendedActionRefused( "the attended executor did not receive the signed pause capability" @@ -3152,7 +3214,9 @@ def recover_reconciliation_receipt( "workflow identity changed after attended capability issuance" ) if ( - AttendedActionStore(run_dir).signed_capability_for_digest(capability.digest) + AttendedActionStore(run_dir, key=self.key).signed_capability_for_digest( + capability.digest + ) != capability ): raise AttendedActionRefused( @@ -3196,7 +3260,7 @@ def recover_reconciliation_receipt( raise AttendedActionRefused( "the completed durable transition does not bind this reconciliation" ) - AttendedActionStore(run_dir).write_reconciliation_receipt( + AttendedActionStore(run_dir, key=self.key).write_reconciliation_receipt( AttendedReconciliationReceipt( pause_id=capability.pause_id, capability_digest=capability.digest, @@ -3468,7 +3532,7 @@ def _resume_program( cursor_digest=cursor_digest, created_at=capability.issued_at, ) - action_store = AttendedActionStore(run_dir) + action_store = AttendedActionStore(run_dir, key=self.key) receipt = action_store.seal_program_receipt(receipt) attended_effects = effects_for_actuation(state.step, "gui") retained_result = _validated_attended_result( @@ -3673,7 +3737,9 @@ def _resume_program( + [item.decision_index for item in pending.program_exception_evidence_delta] ) receipt_path = action_store._receipt_path(receipt.pause_id) - receipt_sha256 = hashlib.sha256(_sealed.read_bytes(receipt_path)).hexdigest() + receipt_sha256 = hashlib.sha256( + action_store._state_read_bytes(receipt_path) + ).hexdigest() control_payload = json.dumps( [frame.model_dump(mode="json") for frame in pending.program_frames], sort_keys=True, @@ -3696,13 +3762,13 @@ def _resume_program( raise AttendedActionRefused( "the attended transition control path leaves the run directory" ) - if _sealed.is_file(control_path): - if _sealed.read_bytes(control_path) != control_payload: + if action_store._state_is_file(control_path): + if action_store._state_read_bytes(control_path) != control_payload: raise AttendedActionRefused( "the attended transition control digest has other bytes" ) else: - action_store._atomic_write(control_path, control_payload) + action_store._state_write(control_path, control_payload) scope = [ ProgramExecutionScopeFrame( graph_id=frame.graph_id, @@ -3831,7 +3897,7 @@ def _continue_run_locked( store, manifest, workflow = self._load(run_dir, capability) replayer = self.replayer_factory(manifest) self._bind_authorization(replayer, manifest) - attended_store = AttendedActionStore(run_dir) + attended_store = AttendedActionStore(run_dir, key=self.key) if workflow.program is not None: pending, state, params = self._program_context( store, workflow, capability @@ -4005,7 +4071,7 @@ def _reconcile_run_locked( ) replayer = self.replayer_factory(manifest) self._bind_authorization(replayer, manifest) - attended_store = AttendedActionStore(run_dir) + attended_store = AttendedActionStore(run_dir, key=self.key) if workflow.program is not None: pending, state, params = self._program_context( store, workflow, capability diff --git a/packages/flow/openadapt_flow/runtime/durable/attended_state.py b/packages/flow/openadapt_flow/runtime/durable/attended_state.py new file mode 100644 index 0000000000..515c2ba1da --- /dev/null +++ b/packages/flow/openadapt_flow/runtime/durable/attended_state.py @@ -0,0 +1,580 @@ +"""Encryption at rest for attended state. + +The attended store (:mod:`openadapt_flow.runtime.durable.attended`) keeps the +pause capability, its history, the decision log (operator and free-text +message), program and reconciliation receipts and the program-transition +control frames in the run directory. With a key configured (the default: the +configured key, else this machine's local key), each file is written as +``.enc``: the same AES-256-GCM container durable checkpoints use, with +associated data that names the format version and the file's run-relative +path, so a ciphertext copied over another path does not decrypt. + +scrypt runs once per run, not once per file. The run's marker holds one random +salt; the key scrypt derives from the passphrase and that salt is kept in +memory, and HKDF gives one subkey for files and one for journal rows. Each +container carries the salt it was sealed under, so a reader needs only the +passphrase. Containers written before this (one scrypt salt each) still +decrypt. + +Older runs wrote these files in plaintext. A keyed store still reads a +plaintext file when no ``.enc`` form exists, until the run's first encrypted +write. That write converts every plaintext attended file to its encrypted form +and then leaves a marker; from then on a plaintext file standing in for an +encrypted one is refused. + +The external attended journal (:mod:`openadapt_flow.runtime.durable.authority`) +stores each decision-log snapshot with :func:`seal_journal_snapshot`. A sealed +run directory (a sealed bundle's run, or strict mode) encrypts these files +through its sealed-set writer instead, and never reaches this module's writer. +""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import os +import secrets +import threading +from collections import OrderedDict +from pathlib import Path +from typing import Callable, Optional + +from cryptography.exceptions import InvalidTag +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.ciphers.aead import AESGCM +from cryptography.hazmat.primitives.kdf.hkdf import HKDF + +from openadapt_flow import crypto as _crypto +from openadapt_flow import sealed_set as _sealed + +#: Suffix of an encrypted attended file (the checkpoint store uses the same). +ENC_SUFFIX = ".enc" +#: Associated-data prefix for an attended file: format version, then the +#: run-relative path. +FILE_AAD_PREFIX = b"openadapt-flow/attended-state/v1\0" +#: Associated-data prefix for one external journal snapshot. +JOURNAL_AAD_PREFIX = b"openadapt-flow/attended-journal/v1\0" +#: Prefix of an encrypted ``snapshot_json`` value. A stored value without it +#: is a plaintext snapshot from an older run. +JOURNAL_MARKER = "openadapt-flow/attended-journal/v1:" +#: Present once a run directory's attended state is encrypted. It holds no +#: recorded data. +MARKER_FILENAME = ".attended_state.encrypted" +MARKER_PAYLOAD = b'{"schema":"openadapt.flow.attended-state-encryption/v1"}\n' +#: The marker schema that carries the run salt. +MARKER_SCHEMA_V2 = "openadapt.flow.attended-state-encryption/v2" +#: Created once, without replacement, when an older marker is upgraded. The +#: first writer to create it picks the run salt, and every writer copies it +#: into the marker. It is never removed: a writer that read the older marker +#: before the upgrade must still find the salt that was picked. +MARKER_UPGRADE_FILENAME = ".attended_state.encrypted.v2" +#: Associated-data prefixes and the stored-row prefix for containers sealed +#: under the run key. Older (per-container scrypt) forms keep the v1 values. +FILE_AAD_PREFIX_V2 = b"openadapt-flow/attended-state/v2\0" +JOURNAL_AAD_PREFIX_V2 = b"openadapt-flow/attended-journal/v2\0" +JOURNAL_MARKER_V2 = "openadapt-flow/attended-journal/v2:" +#: HKDF ``info`` for the two subkeys of a run key. +FILE_SUBKEY_INFO = b"attended-state/v1" +JOURNAL_SUBKEY_INFO = b"attended-journal/v1" + +#: Plaintext files converted on a run's first encrypted write. +CONVERTED_FILES = ( + "attended_capability.json", + "attended_capability_history.json", + "attended_decisions.json", +) +CONVERTED_DIRS = ( + ".attended_program_receipts", + ".attended_reconciliation_receipts", + "private/program-transition-controls", +) + +Writer = Callable[[Path, bytes], None] + + +class AttendedStateUnreadable(ValueError): + """An attended file can't be authenticated, decrypted or trusted.""" + + +# Older containers each need their own scrypt run, and the attended store +# reads the same file several times per action. Keep a few of their +# authenticated plaintexts, keyed by the exact container and associated data, +# and reuse one only for the same passphrase. Containers sealed under a run +# key never use this cache. +_CACHE: "OrderedDict[tuple[bytes, bytes], tuple[str, bytes]]" = OrderedDict() +_CACHE_LIMIT = 64 +_CACHE_LOCK = threading.Lock() + + +# One scrypt per (passphrase, run salt) per process. Keyed by the run salt; +# an entry is reused only for the same passphrase (compared in constant time, +# as the container cache above does), so a different passphrase derives again. +_RUN_KEYS: "OrderedDict[bytes, tuple[str, tuple[AESGCM, AESGCM]]]" = OrderedDict() +_RUN_KEY_LIMIT = 32 +_RUN_KEY_LOCK = threading.Lock() +# A journal salt for a run directory with no salted marker (a sealed run +# directory, or a run whose first journal row comes before any file write). +_JOURNAL_SALTS: dict[str, bytes] = {} + + +def _run_keys(passphrase: str, salt: bytes) -> tuple[AESGCM, AESGCM]: + """The (file, journal) AEADs for one run salt, deriving them only once.""" + with _RUN_KEY_LOCK: + cached = _RUN_KEYS.get(salt) + if cached is not None: + _RUN_KEYS.move_to_end(salt) + if cached is not None and hmac.compare_digest( + cached[0].encode("utf-8"), passphrase.encode("utf-8") + ): + return cached[1] + master = _crypto.derive_data_key(passphrase, salt) + pair = tuple( + AESGCM( + HKDF(algorithm=hashes.SHA256(), length=32, salt=None, info=info).derive( + master + ) + ) + for info in (FILE_SUBKEY_INFO, JOURNAL_SUBKEY_INFO) + ) + keys = (pair[0], pair[1]) + with _RUN_KEY_LOCK: + _RUN_KEYS[salt] = (passphrase, keys) + while len(_RUN_KEYS) > _RUN_KEY_LIMIT: + _RUN_KEYS.popitem(last=False) + return keys + + +def _seal_salted(aead: AESGCM, salt: bytes, plaintext: bytes, aad: bytes) -> bytes: + nonce = secrets.token_bytes(_crypto.NONCE_LEN) + return ( + _crypto.SALTED_FILE_MAGIC + salt + nonce + aead.encrypt(nonce, plaintext, aad) + ) + + +def _split_salted(container: bytes) -> tuple[bytes, bytes, bytes]: + body = container[len(_crypto.SALTED_FILE_MAGIC) :] + salt_len, nonce_len = _crypto.SALT_LEN, _crypto.NONCE_LEN + salt, nonce = body[:salt_len], body[salt_len : salt_len + nonce_len] + ciphertext = body[salt_len + nonce_len :] + if len(salt) != salt_len or len(nonce) != nonce_len or not ciphertext: + raise AttendedStateUnreadable("a truncated attended container") + return salt, nonce, ciphertext + + +def _open_salted(container: bytes, key: str, aad: bytes, *, journal: bool) -> bytes: + salt, nonce, ciphertext = _split_salted(container) + aead = _run_keys(key, salt)[1 if journal else 0] + try: + return aead.decrypt(nonce, ciphertext, aad) + except InvalidTag as exc: + raise AttendedStateUnreadable( + "attended state does not decrypt: wrong key, or the file was " + "changed or moved from another path" + ) from exc + + +def _read_marker(root: Path) -> tuple[bool, Optional[bytes]]: + """(marker present, its run salt). An older marker has no salt.""" + marker = Path(root) / MARKER_FILENAME + if marker.is_symlink(): + raise AttendedStateUnreadable("the attended-state marker is a symlink") + if not marker.is_file(): + return False, None + data = marker.read_bytes() + if data == MARKER_PAYLOAD: + return True, None + return True, _marker_salt(data) + + +def _marker_salt(data: bytes) -> bytes: + """The run salt in a salted marker payload, or raise.""" + try: + obj = json.loads(data) + if obj.get("schema") != MARKER_SCHEMA_V2: + raise ValueError("unknown marker schema") + salt = base64.b64decode(obj["salt"], validate=True) + except (ValueError, TypeError, KeyError, AttributeError) as exc: + raise AttendedStateUnreadable("the attended-state marker is invalid") from exc + if len(salt) != _crypto.SALT_LEN: + raise AttendedStateUnreadable("the attended-state marker is invalid") + return salt + + +def _marker_payload(salt: bytes) -> bytes: + body = {"schema": MARKER_SCHEMA_V2, "salt": base64.b64encode(salt).decode()} + return json.dumps(body, sort_keys=True, separators=(",", ":")).encode() + b"\n" + + +def enc_path(path: Path) -> Path: + return path.with_name(path.name + ENC_SUFFIX) + + +def file_aad(root: Path, path: Path, prefix: bytes = FILE_AAD_PREFIX) -> bytes: + """Associated data binding a file to its run-relative path.""" + try: + rel = Path(os.path.abspath(path)).relative_to(os.path.abspath(root)) + except ValueError as exc: + raise AttendedStateUnreadable( + "an attended file lies outside its run directory" + ) from exc + return prefix + rel.as_posix().encode("utf-8") + + +def seal(plaintext: bytes, key: str, aad: bytes) -> bytes: + return _crypto.encrypt_bytes(plaintext, key, aad=aad) + + +def _require_key(key: Optional[str]) -> str: + if not key: + raise AttendedStateUnreadable( + "attended state is encrypted and no key is configured; set " + "OPENADAPT_BUNDLE_KEY or use the key it was written with" + ) + return key + + +def open_file(root: Path, path: Path, container: bytes, key: Optional[str]) -> bytes: + """Plaintext of the encrypted form of ``path``, in either container + format, or raise :class:`AttendedStateUnreadable`.""" + key = _require_key(key) + if container.startswith(_crypto.SALTED_FILE_MAGIC): + _refuse_other_run(root, container) + return _open_salted( + container, key, file_aad(root, path, FILE_AAD_PREFIX_V2), journal=False + ) + return open_sealed(container, key, file_aad(root, path)) + + +def _refuse_other_run(root: Path, container: bytes) -> None: + """A salted file belongs to the run whose marker holds its salt. One + copied in from another run under the same key and relative path would + otherwise decrypt, so it is refused, as is a salted file whose run has no + salted marker.""" + _present, run_salt = _read_marker(root) + if run_salt is None or _split_salted(container)[0] != run_salt: + raise AttendedStateUnreadable( + "attended state was not sealed for this run: its salt does not " + "match the run's marker" + ) + + +def open_sealed(container: bytes, key: Optional[str], aad: bytes) -> bytes: + """Authenticate and decrypt one older (per-container scrypt) container, + or raise :class:`AttendedStateUnreadable`.""" + key = _require_key(key) + cache_key = (hashlib.sha256(container).digest(), aad) + with _CACHE_LOCK: + cached = _CACHE.get(cache_key) + if cached is not None: + _CACHE.move_to_end(cache_key) + if cached is not None and hmac.compare_digest( + cached[0].encode("utf-8"), key.encode("utf-8") + ): + return cached[1] + try: + plaintext = _crypto.decrypt_bytes(container, key, aad=aad) + except _crypto.CryptoError as exc: + raise AttendedStateUnreadable( + "attended state does not decrypt: wrong key, or the file was " + "changed or moved from another path" + ) from exc + with _CACHE_LOCK: + _CACHE[cache_key] = (key, plaintext) + while len(_CACHE) > _CACHE_LIMIT: + _CACHE.popitem(last=False) + return plaintext + + +def _refuse_symlink(path: Path) -> None: + if path.is_symlink() or enc_path(path).is_symlink(): + raise AttendedStateUnreadable("an attended file must not be a symlink") + + +def exists(path: Path) -> bool: + """Whether ``path`` exists in either form.""" + return enc_path(path).is_file() or path.is_file() + + +def read_bytes(root: Path, path: Path, key: Optional[str]) -> bytes: + """Plaintext of ``path``: its ``.enc`` form decrypted, else an older + plaintext file. Raises ``FileNotFoundError`` when neither exists.""" + _refuse_symlink(path) + enc = enc_path(path) + if enc.is_file(): + return open_file(root, path, enc.read_bytes(), key) + if path.is_file(): + _refuse_plaintext_in_encrypted_run(root, path) + return path.read_bytes() + raise FileNotFoundError(str(path)) + + +def _is_attended_file(root: Path, path: Path) -> bool: + """Whether ``path`` is one of the files a run encrypts at rest.""" + try: + rel = Path(path).relative_to(Path(root)) + except ValueError: + return False + parent = rel.parent.as_posix() + if parent == ".": + return rel.name in CONVERTED_FILES + return parent in CONVERTED_DIRS + + +def _refuse_plaintext_in_encrypted_run(root: Path, path: Path) -> None: + if not _is_attended_file(root, path): + return + if (Path(root) / MARKER_FILENAME).exists() or ( + Path(root) / MARKER_FILENAME + ).is_symlink(): + raise AttendedStateUnreadable( + "a plaintext attended file stands in for encrypted state" + ) + + +def digest_bytes(root: Path, path: Path, key: Optional[str]) -> Optional[bytes]: + """Plaintext for a state digest, or None when the file is missing. The + digest is the same before and after a file is encrypted. A sealed run + directory's files are the sealed-set writer's, read as they are.""" + if _sealed.is_sealed_path(path): + return path.read_bytes() if path.is_file() else None + enc = enc_path(path) + if enc.is_file(): + return open_file(root, path, enc.read_bytes(), key) + if path.is_file(): + _refuse_plaintext_in_encrypted_run(root, path) + return path.read_bytes() + return None + + +def logical_names(directory: Path, suffix: str = ".json") -> set[str]: + """Names of the files in ``directory`` ending in ``suffix``, in either + form (``x.json`` and ``x.json.enc`` both give ``x.json``).""" + if not directory.is_dir(): + return set() + names: set[str] = set() + for entry in directory.iterdir(): + name = entry.name + if name.endswith(suffix + ENC_SUFFIX): + names.add(name[: -len(ENC_SUFFIX)]) + elif name.endswith(suffix): + names.add(name) + return names + + +def _link_new(path: Path, data: bytes) -> None: + """Create ``path`` with ``data`` unless it already exists.""" + tmp = path.with_name(f".{path.name}.{secrets.token_hex(8)}.tmp") + fd = os.open(tmp, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + try: + with os.fdopen(fd, "wb") as handle: + handle.write(data) + handle.flush() + os.fsync(handle.fileno()) + try: + os.link(tmp, path) + except FileExistsError: + pass + finally: + tmp.unlink(missing_ok=True) + + +def _ensure_encrypted(root: Path, key: str) -> bytes: + """Make the run's attended state encrypted and return its run salt. + + A run with no marker gets a new salt: every plaintext attended file of an + older run is encrypted under it, then the marker is created holding it. + A file whose ``.enc`` form already exists keeps that form: a concurrent + writer may have just replaced it. An older marker without a salt is + replaced by one with a salt, chosen once through the upgrade file even + when writers race; its files keep their older containers.""" + marker = root / MARKER_FILENAME + present, salt = _read_marker(root) + if present and salt is not None: + return salt + if present: + # Two writers can both read the older marker. Each proposes a salt by + # creating the upgrade file without replacement; only the first + # succeeds, and every writer installs that one salt in the marker. + # A writer that picked its own salt and replaced the marker could + # otherwise strand files another writer had just sealed. + upgrade = root / MARKER_UPGRADE_FILENAME + if upgrade.is_symlink(): + raise AttendedStateUnreadable("the attended-state marker is a symlink") + _link_new(upgrade, _marker_payload(_crypto.SetKey.new_salt())) + if upgrade.is_symlink() or not upgrade.is_file(): + raise AttendedStateUnreadable("the attended-state marker is invalid") + payload = upgrade.read_bytes() + chosen = _marker_salt(payload) + tmp = marker.with_name(f".{marker.name}.{secrets.token_hex(8)}.tmp") + _link_new(tmp, payload) + os.replace(tmp, marker) + present, salt = _read_marker(root) + if salt != chosen: + raise AttendedStateUnreadable( + "the attended-state marker changed while it was upgraded" + ) + return chosen + new_salt = _crypto.SetKey.new_salt() + file_key = _run_keys(key, new_salt)[0] + candidates = [root / name for name in CONVERTED_FILES] + for dirname in CONVERTED_DIRS: + directory = root / dirname + if directory.is_dir() and not directory.is_symlink(): + candidates.extend(directory / name for name in logical_names(directory)) + converted: list[tuple[Path, Path, bytes, bytes]] = [] + for path in candidates: + if path.is_symlink() or not path.is_file(): + continue + enc = enc_path(path) + if not enc.exists(): + plaintext = path.read_bytes() + container = _seal_salted( + file_key, + new_salt, + plaintext, + file_aad(root, path, FILE_AAD_PREFIX_V2), + ) + _link_new(enc, container) + converted.append((path, enc, plaintext, container)) + path.unlink(missing_ok=True) + root.mkdir(parents=True, exist_ok=True) + _link_new(marker, _marker_payload(new_salt)) + # A concurrent first write may have created the marker first; its salt + # is the run's. Files this call sealed under its own salt are sealed + # again under the run's, unless another writer has replaced them since. + present, salt = _read_marker(root) + if salt is None: + raise AttendedStateUnreadable("the attended-state marker is invalid") + if salt != new_salt: + run_key = _run_keys(key, salt)[0] + for path, enc, plaintext, container in converted: + if enc.is_symlink() or not enc.is_file() or enc.read_bytes() != container: + continue + tmp = enc.with_name(f".{enc.name}.{secrets.token_hex(8)}.tmp") + _link_new( + tmp, + _seal_salted( + run_key, salt, plaintext, file_aad(root, path, FILE_AAD_PREFIX_V2) + ), + ) + os.replace(tmp, enc) + return salt + + +def write_bytes( + root: Path, path: Path, data: bytes, key: Optional[str], writer: Writer +) -> None: + """Write ``path`` encrypted under ``key`` (as ``.enc``), or in plaintext + when no key resolves (the recorded opt-out). A run whose attended state is + already encrypted stays encrypted: with no key, the write is refused. + ``writer`` performs the atomic replace.""" + _refuse_symlink(path) + root = Path(root) + if key: + salt = _ensure_encrypted(root, key) + container = _seal_salted( + _run_keys(key, salt)[0], + salt, + data, + file_aad(root, path, FILE_AAD_PREFIX_V2), + ) + writer(enc_path(path), container) + path.unlink(missing_ok=True) + return + if (root / MARKER_FILENAME).exists() or (root / MARKER_FILENAME).is_symlink(): + raise AttendedStateUnreadable( + "this run's attended state is encrypted and no key is configured; " + "refusing to write it in plaintext (OPENADAPT_FLOW_PLAINTEXT_AT_REST " + "does not downgrade an encrypted run)" + ) + if enc_path(path).exists(): + raise AttendedStateUnreadable( + "an encrypted attended file exists and no key is configured; " + "refusing to replace it with plaintext" + ) + writer(path, data) + + +def unlink(path: Path) -> None: + path.unlink(missing_ok=True) + enc_path(path).unlink(missing_ok=True) + + +# -- the external journal ---------------------------------------------------- + + +def _journal_aad( + path_key: str, sequence: int, prefix: bytes = JOURNAL_AAD_PREFIX +) -> bytes: + return prefix + f"{path_key}\0{sequence}".encode("utf-8") + + +def _journal_salt(run_dir: Optional[Path], path_key: str) -> bytes: + """The run salt from the run's marker when it has one, else one salt per + run path for this process. Reading the marker never writes the run.""" + if run_dir is not None: + _present, salt = _read_marker(Path(run_dir)) + if salt is not None: + return salt + with _RUN_KEY_LOCK: + salt = _JOURNAL_SALTS.get(path_key) + if salt is None: + salt = _crypto.SetKey.new_salt() + _JOURNAL_SALTS[path_key] = salt + return salt + + +def seal_journal_snapshot( + snapshot_json: str, + key: Optional[str], + *, + path_key: str, + sequence: int, + run_dir: Optional[Path] = None, +) -> str: + """The stored form of one snapshot: encrypted and marked when a key is + set, else the plaintext (the recorded opt-out).""" + if not key: + return snapshot_json + salt = _journal_salt(run_dir, path_key) + container = _seal_salted( + _run_keys(key, salt)[1], + salt, + snapshot_json.encode("utf-8"), + _journal_aad(path_key, sequence, JOURNAL_AAD_PREFIX_V2), + ) + return JOURNAL_MARKER_V2 + base64.b64encode(container).decode("ascii") + + +def is_sealed_journal_snapshot(stored: str) -> bool: + """Whether a stored snapshot is encrypted (either format).""" + return stored.startswith((JOURNAL_MARKER, JOURNAL_MARKER_V2)) + + +def open_journal_snapshot( + stored: str, key: Optional[str], *, path_key: str, sequence: int +) -> str: + """The plaintext of one stored snapshot (older rows are plaintext).""" + if stored.startswith(JOURNAL_MARKER_V2): + try: + container = base64.b64decode( + stored[len(JOURNAL_MARKER_V2) :], validate=True + ) + except ValueError as exc: + raise AttendedStateUnreadable( + "an attended journal row is malformed" + ) from exc + if not container.startswith(_crypto.SALTED_FILE_MAGIC): + raise AttendedStateUnreadable("an attended journal row is malformed") + return _open_salted( + container, + _require_key(key), + _journal_aad(path_key, sequence, JOURNAL_AAD_PREFIX_V2), + journal=True, + ).decode("utf-8") + if not stored.startswith(JOURNAL_MARKER): + return stored + container = stored[len(JOURNAL_MARKER) :].encode("utf-8") + return open_sealed(container, key, _journal_aad(path_key, sequence)).decode("utf-8") diff --git a/packages/flow/openadapt_flow/runtime/durable/authority.py b/packages/flow/openadapt_flow/runtime/durable/authority.py index c5183fb44f..75f07adeb4 100644 --- a/packages/flow/openadapt_flow/runtime/durable/authority.py +++ b/packages/flow/openadapt_flow/runtime/durable/authority.py @@ -1056,8 +1056,17 @@ def _validated_journal_head( *, expected_run_id: str, authority_record: DurableAuthorityRecord, + key: Optional[str] = None, ) -> tuple[Optional[str], str]: - """Verify the full external hash chain and return its latest snapshot.""" + """Verify the full external hash chain and return its latest snapshot. + + A snapshot stored encrypted (``attended_state.is_sealed_journal_snapshot``) is + decrypted under ``key``; an older plaintext row reads as it is. + """ + + from openadapt_flow.runtime.durable import attended_state as _state + + snapshot_key = key rows = connection.execute( "SELECT * FROM attended_journal WHERE path_key = ? ORDER BY sequence", @@ -1072,7 +1081,7 @@ def _validated_journal_head( "the external attended-decision journal tail was restored" ) return None, JOURNAL_GENESIS_DIGEST - key = self._journal_key(create=False) + mac_key = self._journal_key(create=False) previous = JOURNAL_GENESIS_DIGEST latest_snapshot: Optional[str] = None for expected_sequence, row in enumerate(rows, start=1): @@ -1085,7 +1094,17 @@ def _validated_journal_head( raise DurableAuthorityBusy( "the external attended-decision journal chain is invalid" ) - snapshot_json = str(values["snapshot_json"]) + try: + snapshot_json = _state.open_journal_snapshot( + str(values["snapshot_json"]), + snapshot_key, + path_key=self.path_key, + sequence=expected_sequence, + ) + except _state.AttendedStateUnreadable as exc: + raise DurableAuthorityBusy( + "the external attended-decision snapshot does not decrypt" + ) from exc snapshot_digest = ( "sha256:" + hashlib.sha256(snapshot_json.encode("utf-8")).hexdigest() ) @@ -1108,7 +1127,7 @@ def _validated_journal_head( record_mac = ( "hmac-sha256:" + hmac.new( - key, + mac_key, JOURNAL_MAC_DOMAIN + record_digest.encode("ascii"), hashlib.sha256, ).hexdigest() @@ -1216,9 +1235,12 @@ def _validate_snapshot_extension( ) def read_attended_snapshot( - self, *, expected_run_id: str + self, *, expected_run_id: str, key: Optional[str] = None ) -> tuple[Optional[str], str]: - """Read the authenticated journal snapshot and its monotonic head.""" + """Read the authenticated journal snapshot and its monotonic head. + + ``key`` decrypts snapshots stored encrypted. + """ with self._transaction() as connection: record = self._read(connection) @@ -1230,6 +1252,7 @@ def read_attended_snapshot( connection, expected_run_id=expected_run_id, authority_record=record, + key=key, ) def append_attended_snapshot( @@ -1238,8 +1261,17 @@ def append_attended_snapshot( expected_run_id: str, expected_head_digest: str, snapshot_json: str, + key: Optional[str] = None, ) -> str: - """Append one authenticated journal revision in the external authority.""" + """Append one authenticated journal revision in the external authority. + + With ``key`` the snapshot is stored encrypted, bound to this run path + and sequence. The digest and HMAC cover the plaintext snapshot. + """ + + from openadapt_flow.runtime.durable import attended_state as _state + + snapshot_key = key # Parse before the transaction so invalid JSON never enters authority. self._validate_snapshot_extension(None, snapshot_json) @@ -1253,6 +1285,7 @@ def append_attended_snapshot( connection, expected_run_id=expected_run_id, authority_record=record, + key=snapshot_key, ) if head != expected_head_digest: raise DurableAuthorityBusy( @@ -1276,11 +1309,11 @@ def append_attended_snapshot( record_digest = ( "sha256:" + hashlib.sha256(self._canonical(unsigned)).hexdigest() ) - key = self._journal_key(create=True) + mac_key = self._journal_key(create=True) record_mac = ( "hmac-sha256:" + hmac.new( - key, + mac_key, JOURNAL_MAC_DOMAIN + record_digest.encode("ascii"), hashlib.sha256, ).hexdigest() @@ -1299,7 +1332,13 @@ def append_attended_snapshot( expected_run_id, head, snapshot_digest, - snapshot_json, + _state.seal_journal_snapshot( + snapshot_json, + snapshot_key, + path_key=self.path_key, + sequence=sequence, + run_dir=self.run_dir, + ), record_digest, record_mac, created_at, diff --git a/packages/flow/openadapt_flow/runtime/durable/checkpoint.py b/packages/flow/openadapt_flow/runtime/durable/checkpoint.py index bd78532a52..8495cce365 100644 --- a/packages/flow/openadapt_flow/runtime/durable/checkpoint.py +++ b/packages/flow/openadapt_flow/runtime/durable/checkpoint.py @@ -519,6 +519,15 @@ def _read_json(self, path: Path) -> Optional[dict]: return json.loads(path.read_text()) # type: ignore[no-any-return] return None + def _attended_key(self) -> Optional[str]: + """The key attended state is read with: this store's key, else the + one the attended store resolves by default.""" + if self.key: + return self.key + from openadapt_flow import crypto as _crypto + + return _crypto.resolve_key(None) + @staticmethod def model_digest(model: Optional[BaseModel]) -> str: """Return one stable digest for an exact logical durable artifact.""" @@ -556,18 +565,23 @@ def _continuation_state_digest_unlocked(self) -> str: except (OSError, ValueError): claim = None + from openadapt_flow.runtime.durable import attended_state as _state + def file_digest(path: Path) -> Optional[str]: - if not path.is_file(): + # Attended state may be encrypted (``.enc``); the digest is + # over its plaintext, so it doesn't change when a file is sealed. + sealed = _state.enc_path(path).is_file() + key = self._attended_key() if sealed else None + data = _state.digest_bytes(self.run_dir, path, key) + if data is None: return None - return "sha256:" + hashlib.sha256(path.read_bytes()).hexdigest() + return "sha256:" + hashlib.sha256(data).hexdigest() receipts_dir = self.run_dir / ".attended_program_receipts" - receipts = [] - if receipts_dir.is_dir(): - receipts = [ - {"name": path.name, "sha256": file_digest(path)} - for path in sorted(receipts_dir.glob("*.json")) - ] + receipts = [ + {"name": name, "sha256": file_digest(receipts_dir / name)} + for name in sorted(_state.logical_names(receipts_dir)) + ] payload = { "manifest": ( manifest.model_dump(mode="json") if manifest is not None else None @@ -663,6 +677,7 @@ def has_durable_artifacts(self) -> bool: self.run_dir / ".attended_decisions.lock", self.run_dir / ".attended_action.lease", self.run_dir / ".attended_program_receipts", + self.run_dir / ".attended_state.encrypted", ) for path in candidates: if path.exists() or path.with_name(path.name + ENC_SUFFIX).exists(): diff --git a/packages/flow/openadapt_flow/runtime/durable/resume.py b/packages/flow/openadapt_flow/runtime/durable/resume.py index bea335e783..ef38160bb9 100644 --- a/packages/flow/openadapt_flow/runtime/durable/resume.py +++ b/packages/flow/openadapt_flow/runtime/durable/resume.py @@ -1236,7 +1236,9 @@ def _resume_under_lease( checkpoint.model_copy(deep=True) for checkpoint in linear_checkpoints ], "program_checkpoints": [], - "auxiliary_digest": replayer._durable_auxiliary_digest(run_dir, []), + "auxiliary_digest": replayer._durable_auxiliary_digest( + run_dir, [], key=getattr(replayer, "checkpoint_key", None) + ), }, authorizing_approval=approved.model_copy(deep=True), ) @@ -1383,7 +1385,9 @@ def _resume_program( "linear_checkpoints": [], "program_checkpoints": [item.model_copy(deep=True) for item in checkpoints], "auxiliary_digest": replayer._durable_auxiliary_digest( - store.run_dir, checkpoints + store.run_dir, + checkpoints, + key=getattr(replayer, "checkpoint_key", None), ), }, authorizing_approval=approval.model_copy(deep=True), diff --git a/packages/flow/openadapt_flow/runtime/replayer.py b/packages/flow/openadapt_flow/runtime/replayer.py index f1c3180f4b..ef03ef4ca8 100644 --- a/packages/flow/openadapt_flow/runtime/replayer.py +++ b/packages/flow/openadapt_flow/runtime/replayer.py @@ -1393,7 +1393,7 @@ def _durable_resume_payload( "linear_checkpoints": store.checkpoints(), "program_checkpoints": store.program_checkpoints(), "auxiliary_digest": self._durable_auxiliary_digest( - run_dir, store.program_checkpoints() + run_dir, store.program_checkpoints(), key=self.checkpoint_key ), } return { @@ -1425,8 +1425,13 @@ def _durable_resume_payload( def _durable_auxiliary_digest( run_dir: Path, program_checkpoints: list[ProgramCheckpoint], + key: Optional[str] = None, ) -> str: - """Bind namespace, active capability, and every attended receipt.""" + """Bind namespace, active capability, and every attended receipt. + + ``key`` reads attended state encrypted at rest (default: the key + ``crypto.resolve_key`` gives). + """ receipts_dir = Path(run_dir) / ".attended_program_receipts" expected = { @@ -1434,11 +1439,10 @@ def _durable_auxiliary_digest( for checkpoint in program_checkpoints if checkpoint.attended_transition is not None } - actual = ( - {path.name for path in receipts_dir.glob("*.json")} - if receipts_dir.is_dir() - else set() - ) + from openadapt_flow.runtime.durable import attended_state as _state + + # Either form counts: ``.json`` or its encrypted ``.json.enc``. + actual = _state.logical_names(receipts_dir) if actual != expected: from openadapt_flow.runtime.durable.approval import StateDiverged @@ -1455,10 +1459,14 @@ def _durable_auxiliary_digest( for path in paths: digest.update(path.name.encode("utf-8")) digest.update(b"\0") - if path.is_file(): - digest.update(path.read_bytes()) - else: - digest.update(b"") + if _state.enc_path(path).is_file() and not key: + from openadapt_flow import crypto as _crypto + + key = _crypto.resolve_key(None) + # The plaintext, so the digest doesn't change when a file is + # encrypted at rest. + data = _state.digest_bytes(Path(run_dir), path, key) + digest.update(b"" if data is None else data) digest.update(b"\0") return "sha256:" + digest.hexdigest() @@ -3265,6 +3273,7 @@ def _stamp_execution_outcome( runtime_worklists=self._outcome_worklists, transition_evidence_root=run_dir, transition_predicate_vision=self.vision, + transition_evidence_key=self.checkpoint_key, ) def _begin_control_overlay(self, profile: Optional[str]) -> None: diff --git a/packages/flow/scripts/export_public_demo_evidence.py b/packages/flow/scripts/export_public_demo_evidence.py index 3814f5b26a..8554e5b226 100644 --- a/packages/flow/scripts/export_public_demo_evidence.py +++ b/packages/flow/scripts/export_public_demo_evidence.py @@ -113,8 +113,12 @@ "approval.json", "approval.json.enc", "attended_capability.json", + "attended_capability.json.enc", "attended_capability_history.json", + "attended_capability_history.json.enc", "attended_decisions.json", + "attended_decisions.json.enc", + ".attended_state.encrypted", } ) REPO_ROOT = Path(__file__).resolve().parents[1] diff --git a/packages/flow/tests/test_attended_at_rest.py b/packages/flow/tests/test_attended_at_rest.py new file mode 100644 index 0000000000..458528f753 --- /dev/null +++ b/packages/flow/tests/test_attended_at_rest.py @@ -0,0 +1,666 @@ +"""Attended state is ciphertext at rest in the default mode. + +Contract (docs/phi_at_rest.md, "Durable checkpoints and attended state"): + +- With a key resolving (the configured key, else the local key), the pause + capability, its history, the decision log, program receipts and control + frames are written as ``.enc`` and the external journal stores each + snapshot encrypted. No plaintext copy reaches the disk, temporary files + included. +- A run written before this (plaintext attended files and journal rows) still + resumes; its first encrypted write converts the plaintext files. +- Each file is bound to its run-relative path: a ciphertext copied over + another path, or a plaintext file standing in for an encrypted one, is + refused. + +Everything here is synthetic. +""" + +from __future__ import annotations + +import os +import shutil +import sqlite3 +from pathlib import Path + +import pytest +from cryptography.exceptions import InvalidTag + +from openadapt_flow import at_rest, crypto +from openadapt_flow.runtime.durable import attended_state +from openadapt_flow.runtime.durable.attended import ( + AttendedActionRefused, + AttendedActionRequest, + AttendedActionStore, + BoundAttendedExecutor, + execute_attended_action, +) +from openadapt_flow.runtime.durable.checkpoint import CheckpointStore +from openadapt_flow.runtime.replayer import Replayer +from tests.test_attended_actions import ( + _attended_program, + _paused, + _request, + _run_attended_program_to_pause, +) +from tests.test_replayer import FakeBackend, FakeVision, Match +from tests.test_sealed_at_rest import watching_disk + +OPERATOR = "Synthetic Operator Quillfeather" +PLAINTEXT_NAMES = attended_state.CONVERTED_FILES + + +@pytest.fixture +def default_mode(monkeypatch): + """The default at-rest mode: no opt-out, the local key under tmp.""" + monkeypatch.delenv("OPENADAPT_FLOW_PLAINTEXT_AT_REST", raising=False) + monkeypatch.delenv("OPENADAPT_BUNDLE_KEY", raising=False) + at_rest._reset_for_tests() + yield + at_rest._reset_for_tests() + + +def _continue_executor() -> BoundAttendedExecutor: + def factory(_manifest): + vision = FakeVision() + vision.text_results = { + "DONE": Match(point=(10, 10), region=(0, 0, 20, 20), confidence=1.0) + } + return Replayer(FakeBackend(), vision=vision, poll_interval_s=0.0) + + return BoundAttendedExecutor(factory) + + +def _journal_rows() -> list[str]: + db = os.environ["OPENADAPT_DURABLE_AUTHORITY_DB"] + connection = sqlite3.connect(db) + try: + return [ + str(row[0]) + for row in connection.execute( + "SELECT snapshot_json FROM attended_journal ORDER BY sequence" + ) + ] + finally: + connection.close() + + +def _journal_bytes() -> bytes: + db = Path(os.environ["OPENADAPT_DURABLE_AUTHORITY_DB"]) + return b"".join( + path.read_bytes() + for path in (db, Path(f"{db}-wal"), Path(f"{db}-journal")) + if path.is_file() + ) + + +def _attended_leaks(watch, run: Path, markers: list[bytes]) -> list[str]: + """Leaks in attended-state files. The run report and step evidence are + plaintext in the default mode by design (recorded in the ledger), so this + looks only at the attended store's own files.""" + attended = ( + "attended_", + ".attended_", + "private/program-transition-controls", + ) + found = [] + for leak in watch.leaks(markers): + path = leak.split(":", 1)[0].split(" (transient)", 1)[0] + try: + rel = Path(path).resolve().relative_to(run.resolve()).as_posix() + except ValueError: + continue + name = rel.rsplit("/", 1)[-1] + if rel.startswith(attended) or name.startswith(attended): + found.append(leak) + return found + + +def test_default_mode_attended_run_leaves_no_plaintext_attended_state( + tmp_path, default_mode +): + with watching_disk() as watch: + _workflow, _bundle, run, _store, capability = _paused(tmp_path) + decision = execute_attended_action( + run, + _request(capability), + operator=OPERATOR, + decided_by="human", + executor=_continue_executor(), + ) + assert decision.status == "completed" + markers = [ + OPERATOR.encode(), + capability.pause_id.encode(), + capability.signature.split(":", 1)[1].encode(), + ] + assert _attended_leaks(watch, run, markers) == [] + + for name in PLAINTEXT_NAMES: + assert not (run / name).exists(), name + attended_files = [ + path for path in run.iterdir() if path.name.startswith("attended_") + ] + assert run / "attended_decisions.json.enc" in attended_files + for path in attended_files: + assert path.name.endswith(attended_state.ENC_SUFFIX), path.name + assert crypto.is_encrypted(path.read_bytes()) + assert (run / attended_state.MARKER_FILENAME).is_file() + + rows = _journal_rows() + assert rows and all( + row.startswith(attended_state.JOURNAL_MARKER_V2) for row in rows + ) + journal = _journal_bytes() + assert all(marker not in journal for marker in markers) + + # The store and a fresh reader still see the decision. + log = AttendedActionStore(run)._read_log() + assert {item.operator for item in log.decisions} == {OPERATOR} + + +def test_default_mode_program_continue_encrypts_receipts_and_controls( + tmp_path, default_mode +): + workflow = _attended_program() + _bundle, run, _backend, _store, capability = _run_attended_program_to_pause( + tmp_path, workflow + ) + vision = FakeVision() + vision.text_results["DONE"] = Match( + point=(10, 10), region=(0, 0, 20, 20), confidence=1.0 + ) + with watching_disk() as watch: + decision = execute_attended_action( + run, + _request(capability, key="program-continue-at-rest"), + operator=OPERATOR, + executor=BoundAttendedExecutor( + lambda _manifest: Replayer( + FakeBackend(), vision=vision, poll_interval_s=0.0 + ) + ), + ) + assert decision.status == "completed" + # The pause id names the receipt file (and so its associated data); it + # is a random identifier, not recorded data. + markers = [OPERATOR.encode(), capability.signature.split(":", 1)[1].encode()] + assert _attended_leaks(watch, run, markers) == [] + + for dirname in ( + ".attended_program_receipts", + "private/program-transition-controls", + ): + names = sorted(path.name for path in (run / dirname).iterdir()) + assert names and all( + name.endswith(".json" + attended_state.ENC_SUFFIX) for name in names + ), names + checkpoint = CheckpointStore( + run, key=crypto.resolve_key(None) + ).program_checkpoints()[0] + assert AttendedActionStore(run).read_program_receipt(capability.pause_id) == ( + checkpoint.attended_transition + ) + + +def test_older_plaintext_attended_run_still_resumes(tmp_path, monkeypatch): + # Written under the recorded opt-out: plaintext files and journal rows, + # as every run was before attended state was encrypted. + _workflow, _bundle, run, _store, capability = _paused(tmp_path) + escalated = execute_attended_action( + run, + AttendedActionRequest( + capability_digest=capability.digest, + idempotency_key="older-run-escalate-0001", + action="escalate", + disposition="needs_assistance", + ), + operator="earlier operator", + ) + assert escalated.status == "escalated" + assert (run / "attended_capability.json").is_file() + assert (run / "attended_decisions.json").is_file() + assert not any( + attended_state.is_sealed_journal_snapshot(row) for row in _journal_rows() + ) + + monkeypatch.delenv("OPENADAPT_FLOW_PLAINTEXT_AT_REST") + at_rest._reset_for_tests() + decision = execute_attended_action( + run, + _request(capability, key="older-run-continue-0001"), + operator=OPERATOR, + executor=_continue_executor(), + ) + assert decision.status == "completed" + for name in PLAINTEXT_NAMES: + assert not (run / name).exists(), name + assert (run / "attended_decisions.json.enc").is_file() + rows = _journal_rows() + assert not attended_state.is_sealed_journal_snapshot(rows[0]) + assert rows[-1].startswith(attended_state.JOURNAL_MARKER_V2) + log = AttendedActionStore(run)._read_log() + operators = [item.operator for item in log.decisions] + assert operators[0] == "earlier operator" and operators[-1] == OPERATOR + + +def test_encrypted_file_moved_to_another_path_or_replaced_is_refused( + tmp_path, default_mode +): + _workflow, _bundle, run, _store, capability = _paused(tmp_path) + execute_attended_action( + run, + AttendedActionRequest( + capability_digest=capability.digest, + idempotency_key="swap-test-escalate-0001", + action="escalate", + disposition="needs_assistance", + ), + operator=OPERATOR, + ) + store = AttendedActionStore(run) + assert store.read() == capability + capability_enc = run / "attended_capability.json.enc" + decisions_enc = run / "attended_decisions.json.enc" + original = capability_enc.read_bytes() + + # A valid ciphertext of another attended file does not open here. + shutil.copyfile(decisions_enc, capability_enc) + with pytest.raises(AttendedActionRefused): + AttendedActionStore(run).read() + + # A plaintext file standing in for the encrypted one is refused. + capability_enc.unlink() + (run / "attended_capability.json").write_text(capability.model_dump_json(indent=2)) + with pytest.raises(AttendedActionRefused): + AttendedActionStore(run).read() + + capability_enc.write_bytes(original) + (run / "attended_capability.json").unlink() + assert AttendedActionStore(run).read() == capability + + +# -- one key derivation per run, and no downgrade (real filesystem) ---------- + +SYNTHETIC_KEY = "synthetic-attended-test-passphrase" + + +def _write(path: Path, data: bytes) -> None: + tmp = path.with_name(f".{path.name}.tmp") + tmp.write_bytes(data) + os.replace(tmp, path) + + +@pytest.fixture +def counted_scrypt(monkeypatch): + """Count every scrypt run, with the per-run key cache cleared.""" + attended_state._RUN_KEYS.clear() + attended_state._JOURNAL_SALTS.clear() + attended_state._CACHE.clear() + calls: list[int] = [] + real = crypto._derive + + def counting(*args, **kwargs): + calls.append(1) + return real(*args, **kwargs) + + monkeypatch.setattr(crypto, "_derive", counting) + yield calls + attended_state._RUN_KEYS.clear() + attended_state._JOURNAL_SALTS.clear() + attended_state._CACHE.clear() + + +def test_scrypt_runs_once_per_run_not_per_file_or_row(tmp_path, counted_scrypt): + root = tmp_path / "run" + receipts = root / ".attended_program_receipts" + receipts.mkdir(parents=True) + paths = [receipts / f"pause-{index}.json" for index in range(30)] + for index, path in enumerate(paths): + attended_state.write_bytes( + root, path, f'{{"n":{index}}}'.encode(), SYNTHETIC_KEY, _write + ) + rows = [ + attended_state.seal_journal_snapshot( + f'{{"seq":{seq}}}', + SYNTHETIC_KEY, + path_key="synthetic-run", + sequence=seq, + run_dir=root, + ) + for seq in range(30) + ] + for _ in range(3): + for index, path in enumerate(paths): + assert attended_state.read_bytes(root, path, SYNTHETIC_KEY) == ( + f'{{"n":{index}}}'.encode() + ) + assert attended_state.digest_bytes(root, path, SYNTHETIC_KEY) == ( + f'{{"n":{index}}}'.encode() + ) + for seq, row in enumerate(rows): + assert row.startswith(attended_state.JOURNAL_MARKER_V2) + assert ( + attended_state.open_journal_snapshot( + row, SYNTHETIC_KEY, path_key="synthetic-run", sequence=seq + ) + == f'{{"seq":{seq}}}' + ) + assert len(counted_scrypt) == 1 + for path in paths: + container = attended_state.enc_path(path).read_bytes() + assert container.startswith(crypto.SALTED_FILE_MAGIC) + assert crypto.is_encrypted(container) + assert not path.exists() + + +def test_salted_file_is_bound_to_its_path(tmp_path, counted_scrypt): + root = tmp_path / "run" + receipts = root / ".attended_program_receipts" + receipts.mkdir(parents=True) + first, second = receipts / "a.json", receipts / "b.json" + attended_state.write_bytes(root, first, b'{"a":1}', SYNTHETIC_KEY, _write) + attended_state.write_bytes(root, second, b'{"b":2}', SYNTHETIC_KEY, _write) + shutil.copyfile(attended_state.enc_path(first), attended_state.enc_path(second)) + with pytest.raises(attended_state.AttendedStateUnreadable): + attended_state.read_bytes(root, second, SYNTHETIC_KEY) + with pytest.raises(attended_state.AttendedStateUnreadable): + attended_state.read_bytes(root, first, "a-different-synthetic-key") + + +def test_salted_journal_row_is_bound_to_path_and_sequence(counted_scrypt): + row = attended_state.seal_journal_snapshot( + '{"seq":1}', SYNTHETIC_KEY, path_key="synthetic-run", sequence=1 + ) + with pytest.raises(attended_state.AttendedStateUnreadable): + attended_state.open_journal_snapshot( + row, SYNTHETIC_KEY, path_key="synthetic-run", sequence=2 + ) + with pytest.raises(attended_state.AttendedStateUnreadable): + attended_state.open_journal_snapshot( + row, SYNTHETIC_KEY, path_key="other-run", sequence=1 + ) + with pytest.raises(attended_state.AttendedStateUnreadable): + attended_state.open_journal_snapshot( + row, None, path_key="synthetic-run", sequence=1 + ) + + +def test_opt_out_write_does_not_downgrade_an_encrypted_run(tmp_path): + root = tmp_path / "run" + root.mkdir() + path = root / "attended_decisions.json" + attended_state.write_bytes(root, path, b'{"v":1}', SYNTHETIC_KEY, _write) + marker = root / attended_state.MARKER_FILENAME + assert marker.is_file() + with pytest.raises(attended_state.AttendedStateUnreadable, match="encrypted"): + attended_state.write_bytes(root, path, b'{"v":2}', None, _write) + other = root / "attended_capability.json" + with pytest.raises(attended_state.AttendedStateUnreadable, match="encrypted"): + attended_state.write_bytes(root, other, b'{"v":2}', None, _write) + assert marker.is_file() + assert attended_state.enc_path(path).is_file() + assert not path.exists() and not other.exists() + assert attended_state.read_bytes(root, path, SYNTHETIC_KEY) == b'{"v":1}' + + +def test_plaintext_stand_in_is_refused_by_read_and_digest(tmp_path): + root = tmp_path / "run" + receipts = root / ".attended_program_receipts" + receipts.mkdir(parents=True) + path = receipts / "pause.json" + attended_state.write_bytes(root, path, b'{"v":1}', SYNTHETIC_KEY, _write) + attended_state.enc_path(path).unlink() + path.write_bytes(b'{"v":"forged"}') + with pytest.raises(attended_state.AttendedStateUnreadable): + attended_state.read_bytes(root, path, SYNTHETIC_KEY) + with pytest.raises(attended_state.AttendedStateUnreadable): + attended_state.digest_bytes(root, path, SYNTHETIC_KEY) + # A file the run never encrypts (the run claim) is still read as is. + claim = root / ".durable_run.claim" + claim.write_bytes(b"synthetic-claim") + assert attended_state.digest_bytes(root, claim, SYNTHETIC_KEY) == ( + b"synthetic-claim" + ) + + +def test_older_containers_rows_and_marker_still_read(tmp_path, counted_scrypt): + """A run encrypted by the older per-container format still reads, and its + next write moves the marker to the salted format without rewriting the + older files.""" + root = tmp_path / "run" + receipts = root / ".attended_program_receipts" + receipts.mkdir(parents=True) + old = receipts / "old.json" + attended_state.enc_path(old).write_bytes( + attended_state.seal( + b'{"old":true}', SYNTHETIC_KEY, attended_state.file_aad(root, old) + ) + ) + marker = root / attended_state.MARKER_FILENAME + marker.write_bytes(attended_state.MARKER_PAYLOAD) + old_row = attended_state.JOURNAL_MARKER + attended_state.seal( + b'{"seq":0}', + SYNTHETIC_KEY, + attended_state._journal_aad("synthetic-run", 0), + ).decode("utf-8") + + assert attended_state.read_bytes(root, old, SYNTHETIC_KEY) == b'{"old":true}' + assert attended_state.is_sealed_journal_snapshot(old_row) + assert ( + attended_state.open_journal_snapshot( + old_row, SYNTHETIC_KEY, path_key="synthetic-run", sequence=0 + ) + == '{"seq":0}' + ) + + new = receipts / "new.json" + attended_state.write_bytes(root, new, b'{"new":true}', SYNTHETIC_KEY, _write) + present, salt = attended_state._read_marker(root) + assert present and salt is not None + assert ( + attended_state.enc_path(new).read_bytes().startswith(crypto.SALTED_FILE_MAGIC) + ) + assert ( + not attended_state.enc_path(old) + .read_bytes() + .startswith(crypto.SALTED_FILE_MAGIC) + ) + assert attended_state.read_bytes(root, old, SYNTHETIC_KEY) == b'{"old":true}' + assert attended_state.read_bytes(root, new, SYNTHETIC_KEY) == b'{"new":true}' + + +def test_older_plaintext_run_converts_under_one_salt(tmp_path, counted_scrypt): + root = tmp_path / "run" + receipts = root / ".attended_program_receipts" + receipts.mkdir(parents=True) + (root / "attended_decisions.json").write_bytes(b'{"decisions":[]}') + for index in range(10): + (receipts / f"p{index}.json").write_bytes(f'{{"p":{index}}}'.encode()) + attended_state.write_bytes( + root, root / "attended_capability.json", b"{}", SYNTHETIC_KEY, _write + ) + assert not (root / "attended_decisions.json").exists() + for index in range(10): + path = receipts / f"p{index}.json" + assert not path.exists() + assert attended_state.read_bytes(root, path, SYNTHETIC_KEY) == ( + f'{{"p":{index}}}'.encode() + ) + assert len(counted_scrypt) == 1 + + +def test_salted_file_from_another_run_is_refused(tmp_path, counted_scrypt): + """Same key and relative path, but another run's salt: refused.""" + run_a, run_b = tmp_path / "a", tmp_path / "b" + name = ".attended_program_receipts/r.json" + for root, body in ((run_a, b'{"run":"a"}'), (run_b, b'{"run":"b"}')): + (root / ".attended_program_receipts").mkdir(parents=True) + attended_state.write_bytes(root, root / name, body, SYNTHETIC_KEY, _write) + shutil.copyfile( + attended_state.enc_path(run_a / name), attended_state.enc_path(run_b / name) + ) + with pytest.raises(attended_state.AttendedStateUnreadable, match="this run"): + attended_state.read_bytes(run_b, run_b / name, SYNTHETIC_KEY) + with pytest.raises(attended_state.AttendedStateUnreadable, match="this run"): + attended_state.digest_bytes(run_b, run_b / name, SYNTHETIC_KEY) + assert attended_state.read_bytes(run_a, run_a / name, SYNTHETIC_KEY) == ( + b'{"run":"a"}' + ) + # A salted file whose run lost its marker is refused too. + (run_a / attended_state.MARKER_FILENAME).unlink() + with pytest.raises(attended_state.AttendedStateUnreadable, match="this run"): + attended_state.read_bytes(run_a, run_a / name, SYNTHETIC_KEY) + + +def test_conversion_that_loses_the_marker_race_reseals_under_the_run_salt( + tmp_path, counted_scrypt, monkeypatch +): + root = tmp_path / "run" + receipts = root / ".attended_program_receipts" + receipts.mkdir(parents=True) + old = receipts / "old.json" + old.write_bytes(b'{"older":"plaintext"}') + winner_salt = crypto.SetKey.new_salt() + original = attended_state._link_new + marker = root / attended_state.MARKER_FILENAME + + def link_new(path, data): + # Another writer creates the marker just before this one does. + if path == marker and not marker.exists(): + original(path, attended_state._marker_payload(winner_salt)) + original(path, data) + + monkeypatch.setattr(attended_state, "_link_new", link_new) + attended_state.write_bytes( + root, receipts / "new.json", b'{"new":1}', SYNTHETIC_KEY, _write + ) + assert attended_state._read_marker(root) == (True, winner_salt) + assert attended_state.read_bytes(root, old, SYNTHETIC_KEY) == ( + b'{"older":"plaintext"}' + ) + assert attended_state.read_bytes(root, receipts / "new.json", SYNTHETIC_KEY) == ( + b'{"new":1}' + ) + + +def test_concurrent_upgrade_of_an_older_marker_keeps_one_salt( + tmp_path, counted_scrypt, monkeypatch +): + """Two writers both read the older marker. The slower one finishes its + upgrade after the faster one has already sealed a file; both files must + still read, under the one salt the marker ends with.""" + root = tmp_path / "run" + receipts = root / ".attended_program_receipts" + receipts.mkdir(parents=True) + marker = root / attended_state.MARKER_FILENAME + marker.write_bytes(attended_state.MARKER_PAYLOAD) + first = receipts / "first.json" + second = receipts / "second.json" + + attended_state.write_bytes(root, first, b'{"writer":"a"}', SYNTHETIC_KEY, _write) + present, salt_after_first = attended_state._read_marker(root) + assert present and salt_after_first is not None + + # The second writer read the older marker before the first one upgraded it. + original = attended_state._read_marker + stale = {"pending": True} + + def read_marker(run_root): + if stale.pop("pending", False): + return True, None + return original(run_root) + + monkeypatch.setattr(attended_state, "_read_marker", read_marker) + attended_state.write_bytes(root, second, b'{"writer":"b"}', SYNTHETIC_KEY, _write) + monkeypatch.setattr(attended_state, "_read_marker", original) + + assert not stale + assert attended_state._read_marker(root) == (True, salt_after_first) + assert attended_state.read_bytes(root, first, SYNTHETIC_KEY) == b'{"writer":"a"}' + assert attended_state.read_bytes(root, second, SYNTHETIC_KEY) == (b'{"writer":"b"}') + + +def test_racing_writers_on_an_older_marker_both_stay_readable( + tmp_path, counted_scrypt, monkeypatch +): + """Two threads read the older marker at the same moment, then both upgrade + it and seal a file. Every file reads afterwards.""" + import threading + + root = tmp_path / "run" + receipts = root / ".attended_program_receipts" + receipts.mkdir(parents=True) + (root / attended_state.MARKER_FILENAME).write_bytes(attended_state.MARKER_PAYLOAD) + original = attended_state._read_marker + barrier = threading.Barrier(2, timeout=30) + seen: set[int] = set() + + def read_marker(run_root): + result = original(run_root) + ident = threading.get_ident() + if ident not in seen: + seen.add(ident) + barrier.wait() + return result + + monkeypatch.setattr(attended_state, "_read_marker", read_marker) + errors: list[BaseException] = [] + + def writer(name): + try: + attended_state.write_bytes( + root, receipts / f"{name}.json", name.encode(), SYNTHETIC_KEY, _write + ) + except BaseException as exc: # noqa: BLE001 - surfaced below + errors.append(exc) + + threads = [threading.Thread(target=writer, args=(n,)) for n in ("a", "b")] + for thread in threads: + thread.start() + for thread in threads: + thread.join(60) + monkeypatch.setattr(attended_state, "_read_marker", original) + + assert not errors + present, salt = attended_state._read_marker(root) + assert present and salt is not None + for name in ("a", "b"): + path = receipts / f"{name}.json" + assert attended_state.read_bytes(root, path, SYNTHETIC_KEY) == name.encode() + + +def test_replayer_classifies_with_its_own_checkpoint_key(monkeypatch, tmp_path): + """The outcome reader opens attended evidence with the key the run sealed + it under, not only the environment's key.""" + from types import SimpleNamespace + + from openadapt_flow import execution_profiles + + seen: dict = {} + + def stamp(report, workflow, profile, **kwargs): + seen.update(kwargs) + + monkeypatch.setattr(execution_profiles, "stamp_execution_outcome", stamp) + replayer = Replayer(FakeBackend(), vision=FakeVision(), checkpoint_key="k-run") + replayer._outcome_worklists = None + replayer._stamp_execution_outcome( + SimpleNamespace(execution_profile="standard"), object(), tmp_path + ) + assert seen["transition_evidence_key"] == "k-run" + assert seen["transition_evidence_root"] == tmp_path + + +def test_run_key_cache_never_reuses_a_key_for_another_passphrase(counted_scrypt): + salt = b"\x07" * crypto.SALT_LEN + first = attended_state._run_keys("passphrase-one", salt) + assert attended_state._run_keys("passphrase-one", salt) is first + assert len(counted_scrypt) == 1 + other = attended_state._run_keys("passphrase-two", salt) + assert other is not first + assert len(counted_scrypt) == 2 + nonce = b"\x00" * crypto.NONCE_LEN + sealed = first[0].encrypt(nonce, b"attended", b"aad") + with pytest.raises(InvalidTag): + other[0].decrypt(nonce, sealed, b"aad")