diff --git a/CAPABILITIES.md b/CAPABILITIES.md index 6d2f5b3079..6638049748 100644 --- a/CAPABILITIES.md +++ b/CAPABILITIES.md @@ -316,7 +316,7 @@ improvising or retrying when something is off. | Each package installs on its own within a size budget, checked in CI (privacy without desktop, capture without plotting) | On main ([#1270](https://github.com/OpenAdaptAI/OpenAdapt/pull/1270), [#1271](https://github.com/OpenAdaptAI/OpenAdapt/pull/1271), [#1279](https://github.com/OpenAdaptAI/OpenAdapt/pull/1279)) | CI | `.github/workflows/packages.yml` | | 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 and route read them in memory through one artifact reader and refuse without the key | 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) | +| 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) | | 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) | diff --git a/llms-full.txt b/llms-full.txt index b67e506089..4b1c6986b2 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -719,7 +719,7 @@ improvising or retrying when something is off. | Each package installs on its own within a size budget, checked in CI (privacy without desktop, capture without plotting) | On main ([#1270](https://github.com/OpenAdaptAI/OpenAdapt/pull/1270), [#1271](https://github.com/OpenAdaptAI/OpenAdapt/pull/1271), [#1279](https://github.com/OpenAdaptAI/OpenAdapt/pull/1279)) | CI | `.github/workflows/packages.yml` | | 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 and route read them in memory through one artifact reader and refuse without the key | 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) | +| 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) | | 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) | diff --git a/packages/flow/docs/phi_at_rest.md b/packages/flow/docs/phi_at_rest.md index 575bb703f8..5b4e9dbfe0 100644 --- a/packages/flow/docs/phi_at_rest.md +++ b/packages/flow/docs/phi_at_rest.md @@ -26,7 +26,9 @@ with your other secrets. | 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 | -| Desktop recordings, approval requests, `--record-video`, and the output of `visualize`, `explain`, `teach`, `emit-skill`, `push`, `compose`, `process`, `variants`, `route`, `for-each`, `shadow`, `compile-agent`, `guard`, `bench`, `sanitize`, `redact-report`, `jit`, `--save-healed-to`, `--workers` and `induce --report` | Logged | Held back before the command runs | +| The bundle, skill library and compiled fix of `teach` | Encrypted from encrypted input; otherwise logged | Encrypted | +| The sanitized review copy and approved archive of `push` | Encrypted from encrypted input; otherwise logged | Encrypted | +| Desktop recordings, approval requests, `--record-video`, and the output of `visualize`, `explain`, `emit-skill`, `compose`, `process`, `variants`, `route`, `for-each`, `shadow`, `compile-agent`, `guard`, `bench`, `sanitize`, `redact-report`, `jit`, `--save-healed-to`, `--workers` and `induce --report` | Logged | Held back before the command runs | **Logged** means stderr names the path and one line goes to `~/.openadapt/plaintext-at-rest.jsonl`; when that line can't be written, the @@ -136,8 +138,9 @@ prior certification; certify the sealed destination. ## Inventory -`openadapt_flow.sealed_set.BUNDLE_FILES` and `RECORDING_FILES` are the source -of truth; a test keeps this table in step with them. +`openadapt_flow.sealed_set.BUNDLE_FILES`, `RECORDING_FILES`, `LIBRARY_FILES` +and `SANITIZED_FILES` are the source of truth; a test keeps these tables in +step with them. ### Bundle @@ -169,6 +172,24 @@ of truth; a test keeps this table in step with them. | `distill.json`, `.openadapt-approval.json`, `.openadapt-sanitization.json` | Derived reports and approvals | Encrypted | | Anything else | Unknown | Refused | +### Skill library + +| File | Holds | At rest | +| --- | --- | --- | +| `skills.json` | Every learned program version (target labels, typed literals, screen text, crop paths) and the execution traces it was learned from | Encrypted | +| Anything else | Unknown | Refused | + +### Sanitized review copy + +| File | Holds | At rest | +| --- | --- | --- | +| `.openadapt-sanitization.json` | The redaction manifest: file names, hashes and redaction records | Encrypted | +| `.openadapt-approval.json` | The reviewer's approval | Encrypted | +| `*` | The scrubbed copy of each recording or bundle file, before review | Encrypted | + +The approved archive beside it (`.approved.zip`) is one AES-256-GCM +container under the same key. + ### Run directory of a sealed bundle Every file is encrypted except these structural ones: `.durable_run.claim` @@ -190,9 +211,8 @@ The web recorder seals every frame, event line, the metadata and the network capture in memory and writes only ciphertext, from the first frame. The event log is resealed whole on each append, so the file on disk is always one complete ciphertext. The recording is finalized (indexed) when it finishes. -`record --verify` keeps its system-of-record reads inside it, encrypted; -`record --with-check` says it cannot mark a check in an encrypted recording -yet and marks nothing. +`record --verify` keeps its system-of-record reads inside it, encrypted, and +`record --with-check` marks its check inside it, encrypted. Web recordings are encrypted under a configured key, and under the local key in strict mode. `--no-encrypt` records one in plaintext, with a warning and a @@ -225,9 +245,10 @@ takes tens to hundreds of milliseconds, so encryption adds about 1% or less. `scaffold-verifier`, `seal`, `certify`, `qualify`, `admit`, `run`, `replay`, `approve`, `resume`, `undo`, `visualize` (including a process contract), `explain`, `emit-skill`, `for-each`, `sections`, `check mark`, `compose`, -`process`, `variants` and `route` read sealed bundles, run directories and -encrypted recordings in memory, given the key, and keep what they write inside -a sealed set sealed. Each reads artifact files through one reader, +`process`, `variants`, `route`, `teach`, `push`, `sanitize`, +`review-sanitized` and `approve-sanitized` read sealed bundles, run +directories and encrypted recordings in memory, given the key, and keep what +they write inside a sealed set sealed. Each reads artifact files through one reader, `openadapt_flow.artifacts`: - `compile` of an encrypted recording writes a sealed bundle, under @@ -261,6 +282,33 @@ a sealed set sealed. Each reads artifact files through one reader, (`composition.json`, `process-contract.json`, `variant-set.json`) holds names, digests and pointers; by default it is named on stderr and in the ledger, and strict mode refuses the command before it runs. +- `teach` of a sealed run, fix recording or bundle, and every `teach` in + strict mode, writes the taught bundle sealed (crops, program and repair + candidate), under the same key rules as `compile`. Its skill library + (`.skills`, or `--library`) is a sealed set of its own, and the fix + recording compiles into `_teach_compiled`, a sealed bundle inside the + recording that the recording's index skips. A library or `--out` directory + that already holds plaintext is refused, so nothing learned from encrypted + input lands beside plaintext. An `--out` or library that is, contains, or + is inside any input (by real path or by inode, so a symlink alias counts) + is refused before anything is written. +- `push` and `sanitize` of an encrypted recording or bundle, and every + `push` in strict mode, write the sanitized review copy as a sealed set of + kind `sanitized`, staged in a private (0700) directory beside it and removed + if anything fails. Scrubbing, the second-pass check and the approval rescan + run on the decrypted bytes in memory and are exactly as strict as for + plaintext. `review-sanitized` shows the original and the copy from memory + and writes a reviewer's redactions back encrypted; `approve-sanitized` + encrypts the approved archive; the upload decrypts that archive in memory and + sends the exact approved bytes. + +Refusals never quote what they read. A validation error's quoted input value +is dropped from every refusal the CLI prints. With sealed input or in strict +mode, a `teach` or `push` refusal names only the error's type, unless its text +was built only from the command line (the paths you passed, such as an `--out` +that overlaps an input or a run that came from another bundle). An unexpected +error prints its type instead of a traceback, and a success line names a skill +or workflow only by a short hash. Without the key, each of these refuses before it reads anything and names the variable to set. Any other command given a sealed path refuses with a message @@ -291,8 +339,8 @@ marks as logged. Also: and no `seal`: only `workflow.json` and `templates/` are encrypted there. - The offline Seal of a run (`local-seal`) is not written for an encrypted run directory yet; the run says so on stderr. -- Commands outside the list above, such as `visualize`, `explain`, `teach`, - `emit-skill` and `push`, refuse sealed input rather than read it. +- Commands outside the list above refuse sealed input rather than decrypt it + to disk. ## The identity template is not encryption diff --git a/packages/flow/openadapt_flow/__main__.py b/packages/flow/openadapt_flow/__main__.py index 7004d2e55f..8f1aae14be 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -7867,7 +7867,7 @@ def _cmd_push(args: argparse.Namespace) -> int: if args.json: _print_push_json(_push_json_failure()) return 1 - print(f"push failed: {e}") + print(f"push failed: {_refusal_text(args, e)}") return 1 if args.json: try: @@ -7896,9 +7896,12 @@ def _cmd_push(args: argparse.Namespace) -> int: if result.get("workflow_id"): version = result.get("version") version_number = version.get("version") if isinstance(version, dict) else None - detail = ( - f"name={result.get('workflow_name')!r}, kind={kind}, status={server_status}" + from openadapt_flow import safe_errors + + name = safe_errors.label( + result.get("workflow_name"), private=_private_input(args) ) + detail = f"name={name}, kind={kind}, status={server_status}" if version_number is not None: detail += f", version={version_number}" print(f"Pushed. workflow_id={result['workflow_id']} ({detail}).") @@ -8333,7 +8336,7 @@ def _cmd_teach(args: argparse.Namespace) -> int: library_dir=Path(args.library) if args.library else None, ) except TeachError as e: - print(f"teach cannot run: {e}") + print(f"teach cannot run: {_refusal_text(args, e)}") return 2 print(result.summary()) @@ -8368,7 +8371,7 @@ def _cmd_teach_branch(args: argparse.Namespace) -> int: library_dir=Path(args.library) if args.library else None, ) except TeachError as e: - print(f"teach-branch cannot run: {e}") + print(f"teach-branch cannot run: {_refusal_text(args, e)}") return 2 print(result.summary()) return 0 if result.promoted else 1 @@ -13137,6 +13140,11 @@ def _qualify_default_form(argv: list[str]) -> list[str]: "openadapt_flow.__main__:_cmd_process", "openadapt_flow.__main__:_cmd_variants", "openadapt_flow.__main__:_cmd_route", + "openadapt_flow.__main__:_cmd_teach", + "openadapt_flow.__main__:_cmd_push", + "openadapt_flow.__main__:_cmd_sanitize", + "openadapt_flow.__main__:_cmd_review_sanitized", + "openadapt_flow.__main__:_cmd_approve_sanitized", "openadapt_flow.cli_sections:cmd_sections", "openadapt_flow.cli_check:cmd_check_mark", "openadapt_flow.cli_admit:cmd_admit_run", @@ -13208,7 +13216,8 @@ def _qualify_default_form(argv: list[str]) -> list[str]: #: Sealed-ready commands that seal plaintext input themselves under strict -#: mode (a recording, a compiled, induced or admitted bundle, a run directory). +#: mode (a recording, a compiled, induced or admitted bundle, a run directory, +#: the taught bundle and skill library, the sanitized review copy of a push). _STRICT_SELF_SEALING: frozenset[str] = frozenset( { "openadapt_flow.__main__:_cmd_record", @@ -13218,6 +13227,8 @@ def _qualify_default_form(argv: list[str]) -> list[str]: "openadapt_flow.__main__:_cmd_run", "openadapt_flow.__main__:_cmd_replay", "openadapt_flow.__main__:_cmd_resume", + "openadapt_flow.__main__:_cmd_teach", + "openadapt_flow.__main__:_cmd_push", } ) @@ -13252,9 +13263,17 @@ def _plaintext_directories_in(args: argparse.Namespace) -> list[Path]: return found -#: Commands that write a new bundle sealed when their input bundle is sealed. -_SEALED_OUTPUT_FROM_SEALED_INPUT: frozenset[str] = frozenset( - {"openadapt_flow.__main__:_cmd_for_each"} +#: Commands that write their output sealed when an input is sealed, by the +#: arguments that name those inputs. +_SEALED_OUTPUT_FROM_SEALED_INPUT: dict[str, tuple[str, ...]] = { + "openadapt_flow.__main__:_cmd_for_each": ("bundle",), + "openadapt_flow.__main__:_cmd_teach": ("run_dir", "fix", "bundle"), + "openadapt_flow.__main__:_cmd_push": ("path",), +} + +#: Commands that write their output sealed in strict mode. +_SEALED_OUTPUT_IN_STRICT: frozenset[str] = frozenset( + {"openadapt_flow.__main__:_cmd_teach", "openadapt_flow.__main__:_cmd_push"} ) @@ -13308,12 +13327,15 @@ def _unsealed_output_check(args: argparse.Namespace) -> None: artifact = _UNSEALED_OUTPUT_COMMANDS.get(command) if artifact is None: return + if command in _SEALED_OUTPUT_IN_STRICT and at_rest.strict(): + return # strict mode makes a sealed output if command in _SEALED_OUTPUT_FROM_SEALED_INPUT: from openadapt_flow import sealed_set - source = getattr(args, "bundle", None) - if source and sealed_set.is_sealed_path(source): - return # a sealed input makes a sealed output + for name in _SEALED_OUTPUT_FROM_SEALED_INPUT[command]: + source = getattr(args, name, None) + if source and sealed_set.is_sealed_path(source): + return # a sealed input makes a sealed output target = next( ( getattr(args, name) @@ -13325,37 +13347,148 @@ def _unsealed_output_check(args: argparse.Namespace) -> None: at_rest.unsealed_write(artifact, target) +class _UnclassifiedArgument(ValueError): + """An argument value that is neither a path, a scalar nor a container.""" + + def _sealed_paths_in(args: argparse.Namespace) -> list[Path]: - from openadapt_flow import sealed_set + """The sealed roots named anywhere in the command's arguments. + Values are walked at any depth (lists, tuples, sets, mappings, nested + pairs such as teach-branch's ``--example RUN FIX``). A value that is not + a path, a known scalar or a container raises ``_UnclassifiedArgument``; + ``_private_input`` treats that as private. + """ found: list[Path] = [] + unclassified: list[str] = [] for name, value in vars(args).items(): if name == "func": continue - values = value if isinstance(value, (list, tuple)) else [value] - for item in values: - if not isinstance(item, (str, Path)) or not str(item): - continue - text = str(item) - if "://" in text or "\n" in text or len(text) > 4096: - continue - # NAME=PATH arguments (--child, --admission, --variant) name a - # path after the first "="; check both readings. - candidates = [text] - if isinstance(item, str) and "=" in text: - candidates.append(text.split("=", 1)[1]) - for candidate in candidates: - if not candidate: - continue - path = Path(candidate) - root = ( - path if sealed_set.is_sealed(path) else sealed_set.sealed_root(path) - ) - if root is not None and root not in found: - found.append(root) + # Attributes named "_..." are set by the code (handlers, run state), + # not typed by the user: their paths are still read, but an object + # among them does not make the run private. + internal = name.startswith("_") + _collect_sealed(value, name, found, [] if internal else unclassified) + if unclassified: + raise _UnclassifiedArgument( + f"cannot tell whether {unclassified[0]} names sealed input" + ) return found +def _collect_sealed( + value: object, name: str, found: list[Path], unclassified: list[str] +) -> None: + import enum + + from openadapt_flow import sealed_set + + if value is None or isinstance(value, (bool, int, float, enum.Enum)): + return + if isinstance(value, Mapping): + for key, item in value.items(): + _collect_sealed(key, name, found, unclassified) + _collect_sealed(item, name, found, unclassified) + return + if isinstance(value, (list, tuple, set, frozenset)): + for item in value: + _collect_sealed(item, name, found, unclassified) + return + if not isinstance(value, (str, Path)): + unclassified.append(f"--{name.replace('_', '-')}") + return + text = str(value) + if not text: + return + # Every string is checked against the filesystem, whatever it looks + # like: a newline, "://" or any length can spell a local path. NAME=PATH + # arguments (--child, --admission, --variant) name a path after the + # first "="; check both readings. + candidates = [text] + if isinstance(value, str) and "=" in text: + candidates.append(text.split("=", 1)[1]) + for candidate in candidates: + if not candidate: + continue + try: + root = _sealed_root_named(candidate) + except _UnclassifiedArgument: + unclassified.append(f"--{name.replace('_', '-')}") + continue + if root is None: + root = sealed_set.sealed_root(candidate) # an active writer + if root is not None and root not in found: + found.append(root) + + +def _sealed_root_named(text: str) -> Optional[Path]: + """The sealed set that ``text`` names or lies inside, if any. + + Raises ``_UnclassifiedArgument`` when the filesystem cannot tell (a + header that cannot be checked for a reason other than not existing). + """ + import errno + import os + import stat + + from openadapt_flow import sealed_set + + # The path cannot exist, so nothing sealed can be read through it. + no_such_path = (errno.ENOENT, errno.ENOTDIR, errno.ENAMETOOLONG) + # Windows rejects a name it cannot hold, such as "http:" with a colon + # (ERROR_INVALID_NAME, ERROR_BAD_PATHNAME): no such path can exist either. + invalid_name = (123, 161) + try: + full = os.path.abspath(text) + except ValueError: # an embedded NUL: no such path + return None + current = full + while True: + try: + st = os.stat(os.path.join(current, sealed_set.HEADER)) + except ValueError: + return None + except OSError as exc: + invalid = getattr(exc, "winerror", None) in invalid_name + if exc.errno not in no_such_path and not invalid: + raise _UnclassifiedArgument( + "a path whose sealed header cannot be checked" + ) from exc + else: + if stat.S_ISREG(st.st_mode): + return Path(text) if current == full else Path(current) + parent = os.path.dirname(current) + if parent == current: + return None + current = parent + + +def _refusal_text(args: argparse.Namespace, error: BaseException) -> str: + """What a command prints when it refuses ``error``. + + When the command read sealed input or strict mode is on, any part of the + message may quote decrypted content (a skill id, a workflow name, a + path inside the artifact), so only the kind of error is printed, unless + the error is marked ``public`` (built only from the command line). + Otherwise the message prints without quoted input values. + """ + from openadapt_flow import safe_errors + + if _private_input(args) and getattr(error, "public", False) is not True: + return safe_errors.withheld(error) + return safe_errors.message(error) + + +def _private_input(args: argparse.Namespace) -> bool: + """Whether the command read sealed input or runs in strict mode.""" + from openadapt_flow import at_rest, sealed_set + + try: + return at_rest.strict() or bool(_sealed_paths_in(args)) + except (sealed_set.SealedSetError, ValueError, OSError): + return True # cannot tell: treat it as private + + def _sealed_input_refusal(args: argparse.Namespace) -> Optional[str]: """Refuse a sealed path in a command that cannot keep it sealed, and a sealed path whose key is not configured, before the command runs.""" @@ -13375,8 +13508,9 @@ def _sealed_input_refusal(args: argparse.Namespace) -> Optional[str]: "command. Commands that read sealed input: record, compile, " "induce, scaffold-verifier, seal, certify, qualify, admit, run, " "replay, approve, resume, undo, visualize, explain, emit-skill, " - "for-each, sections, check mark, compose, process, variants and " - "route." + "for-each, sections, check mark, compose, process, variants, " + "route, teach, push, sanitize, review-sanitized and " + "approve-sanitized." ) for root in sealed: kind = sealed_set.header_kind(root) @@ -13435,21 +13569,21 @@ def main(argv: Sequence[str] | None = None) -> int: except (OSError, ValueError) as exc: print(f"REFUSED (deployment_config_changed): {exc}", file=sys.stderr) return 2 - from openadapt_flow import sealed_set + from openadapt_flow import safe_errors, sealed_set try: refusal = _sealed_input_refusal(args) - except sealed_set.SealedSetError as exc: + except (sealed_set.SealedSetError, _UnclassifiedArgument) as exc: refusal = str(exc) if refusal is not None: - print(f"REFUSED: {refusal}", file=sys.stderr) + print(f"REFUSED: {safe_errors.message(refusal)}", file=sys.stderr) return 2 from openadapt_flow import at_rest try: _unsealed_output_check(args) except (at_rest.AtRestRefused, ValueError) as exc: - print(f"REFUSED: {exc}", file=sys.stderr) + print(f"REFUSED: {safe_errors.message(exc)}", file=sys.stderr) return 2 try: code = args.func(args) @@ -13467,7 +13601,7 @@ def main(argv: Sequence[str] | None = None) -> int: # A sealed bundle or encrypted recording that is missing its key, # was changed after sealing, or holds an unknown file, or a write # strict mode refuses: fail closed. - print(f"REFUSED: {exc}", file=sys.stderr) + print(f"REFUSED: {safe_errors.message(exc)}", file=sys.stderr) return 2 except BrowserSupportMissing as exc: # Missing optional browser support is a setup decision, not a crash. @@ -13477,6 +13611,11 @@ def main(argv: Sequence[str] | None = None) -> int: from openadapt_flow.runtime.known_ids import KnownIdAdmissionHalt if not isinstance(exc, KnownIdAdmissionHalt): + if _private_input(args): + # A traceback would quote what the command read from sealed + # input: name the kind of error only. + print(f"FAILED: {safe_errors.withheld(exc)}", file=sys.stderr) + return 1 raise # The configured known-ID list may not be used: halt before any step # runs, with the named reason. diff --git a/packages/flow/openadapt_flow/hosted.py b/packages/flow/openadapt_flow/hosted.py index 4a0f111fdb..ee8e840d9c 100644 --- a/packages/flow/openadapt_flow/hosted.py +++ b/packages/flow/openadapt_flow/hosted.py @@ -38,6 +38,7 @@ from __future__ import annotations import hashlib +import io import ipaddress import json import os @@ -899,7 +900,11 @@ def _is_recording_dir(path: Path) -> bool: def _is_bundle_dir(path: Path) -> bool: - return (path / "workflow.json").is_file() or (path / "workflow.json.enc").is_file() + from openadapt_flow import artifacts + + return artifacts.is_file(path / "workflow.json") or artifacts.is_file( + path / "workflow.json.enc" + ) def find_latest_recording(base: Optional[Path] = None) -> Path: @@ -1221,6 +1226,15 @@ def _verified_archive_snapshot(archive_path: Path, *, expected_sha256: str) -> A raise +def _verified_archive_bytes(data: bytes, *, expected_sha256: str) -> Any: + """An in-memory upload body for an encrypted derivative's archive.""" + if hashlib.sha256(data).hexdigest() != expected_sha256: + raise HostedError( + "Approved immutable archive changed before upload; refusing egress" + ) + return io.BytesIO(data) + + def push( path: Optional[Any] = None, *, @@ -1303,6 +1317,7 @@ def push( raise HostedError(f"PATH is not a directory: {src}") _assert_upload_tree_safe(src) + from openadapt_flow import artifacts from openadapt_flow.sanitized_artifact import ( MANIFEST_NAME, SanitizationError, @@ -1311,11 +1326,12 @@ def push( build_ingest_manifest, load_and_verify_derivative, load_valid_approval, + read_approved_archive, sanitize_artifact, source_tree_sha256, ) - is_derivative = (src / MANIFEST_NAME).is_file() + is_derivative = artifacts.is_file(src / MANIFEST_NAME) if not is_derivative: actual_kind = ( "bundle" if kind == "bundle" and _is_bundle_dir(src) else "recording" @@ -1497,10 +1513,19 @@ def push( ), } try: - archive_snapshot = _verified_archive_snapshot( - archive_path, - expected_sha256=approval["approved_derivative_sha256"], - ) + if artifacts.is_sealed(src): + # An encrypted derivative's archive is decrypted in memory only. + archive_snapshot = _verified_archive_bytes( + read_approved_archive(src), + expected_sha256=approval["approved_derivative_sha256"], + ) + else: + archive_snapshot = _verified_archive_snapshot( + archive_path, + expected_sha256=approval["approved_derivative_sha256"], + ) + except SanitizationError as exc: + raise HostedError(f"Sanitized artifact is not uploadable: {exc}") from exc except HostedError: raise except OSError as exc: diff --git a/packages/flow/openadapt_flow/ir.py b/packages/flow/openadapt_flow/ir.py index 5922d72681..494fa4722f 100644 --- a/packages/flow/openadapt_flow/ir.py +++ b/packages/flow/openadapt_flow/ir.py @@ -3689,6 +3689,13 @@ def load( report = _bv.validate_workflow(wf) report.raise_if(categories=("structure",)) + if not bundle_encrypted and wf.encrypted: + # The flag is authoritative only with an encrypted envelope on + # disk: a plaintext copy (for example a decrypted upload) that + # still says encrypted is plaintext, and loads as plaintext. + wf.encrypted = False + if wf.manifest is not None: + wf.manifest = wf.manifest.model_copy(update={"encrypted": False}) return wf def stamp_certification( diff --git a/packages/flow/openadapt_flow/learning/branch_teach.py b/packages/flow/openadapt_flow/learning/branch_teach.py index 811ac469b5..555549e077 100644 --- a/packages/flow/openadapt_flow/learning/branch_teach.py +++ b/packages/flow/openadapt_flow/learning/branch_teach.py @@ -89,6 +89,7 @@ _spec_from_recording, _texts_seen_before_halt, _was_delivered, + check_out_is_new, check_provenance, load_base_program, load_halt_report, @@ -544,6 +545,13 @@ def teach_branch( "teach-branch needs at least two resolved halts. For one, use " "`openadapt-flow teach`" ) + inputs: dict[str, Path | str] = {"base bundle": bundle} + for i, example in enumerate(examples): + inputs[f"halted run {i + 1}"] = example.run_dir + inputs[f"fix {i + 1}"] = example.fix + for i, negative in enumerate(negatives): + inputs[f"normal run {i + 1}"] = negative + check_out_is_new(out, library=library_dir, **inputs) base, program = load_base_program(bundle) loaded = [_load_example(ex, base, program, Path(bundle)) for ex in examples] n_neg = len(negatives) diff --git a/packages/flow/openadapt_flow/learning/library.py b/packages/flow/openadapt_flow/learning/library.py index 84cf40627f..0a8e43869f 100644 --- a/packages/flow/openadapt_flow/learning/library.py +++ b/packages/flow/openadapt_flow/learning/library.py @@ -13,6 +13,11 @@ Persistence is a single JSON file per library root (``skills.json``): the schema is Pydantic, so a version round-trips a real ``ProgramGraph`` verbatim. No model calls, no I/O beyond the JSON file. + +A learned program can carry patient data, so a library can also be a sealed set +of kind ``library`` (:mod:`openadapt_flow.sealed_set`): ``seal=True`` creates +one, and a root that is already sealed stays sealed. ``skills.json`` is then +encrypted on disk and read back in memory. """ from __future__ import annotations @@ -24,6 +29,7 @@ from pydantic import BaseModel, Field +from openadapt_flow import artifacts, sealed_set from openadapt_flow.ir import ProgramGraph from openadapt_flow.learning.trace import ExecutionTrace @@ -101,12 +107,37 @@ class SkillLibrary: the next). """ - def __init__(self, root: Path | str) -> None: + def __init__( + self, root: Path | str, *, seal: bool = False, key: Optional[str] = None + ) -> None: self.root = Path(root) - self.root.mkdir(parents=True, exist_ok=True) self._skills: dict[str, Skill] = {} + self._writer: Optional[sealed_set.SealedSetWriter] = None + existing = sealed_set.is_sealed(self.root) + if seal or existing: + if not existing and self.root.is_dir() and any(self.root.iterdir()): + raise sealed_set.SealedSetError( + f"{self.root} holds an unencrypted skill library; pass a new " + "library directory so learned programs stay encrypted" + ) + passphrase = sealed_set.passphrase_for("library", key) + if not passphrase: + raise sealed_set.SealedSetKeyError( + sealed_set.missing_key_message(self.root, "library") + ) + self._writer = sealed_set.SealedSetWriter( + self.root, "library", passphrase, resume=True + ) + if not existing: + self._writer.finalize() # an empty library is a complete set + else: + self.root.mkdir(parents=True, exist_ok=True) self._load() + @property + def sealed(self) -> bool: + return self._writer is not None + # -- persistence -------------------------------------------------------- @property @@ -114,15 +145,26 @@ def path(self) -> Path: return self.root / _LIBRARY_FILE def _load(self) -> None: - if not self.path.is_file(): - return - raw = json.loads(self.path.read_text()) + if self._writer is not None: + # Read through the writer, which holds this library's own key. + if not self._writer.is_file(_LIBRARY_FILE): + return + raw = json.loads(self._writer.read_bytes(_LIBRARY_FILE).decode("utf-8")) + else: + if not artifacts.is_file(self.path): + return + raw = json.loads(artifacts.read_text(self.path)) for sid, data in raw.get("skills", {}).items(): self._skills[sid] = Skill.model_validate(data) def save(self) -> Path: payload = {"skills": {sid: s.model_dump() for sid, s in self._skills.items()}} - self.path.write_text(json.dumps(payload, indent=2, default=str)) + text = json.dumps(payload, indent=2, default=str) + if self._writer is not None: + self._writer.write(_LIBRARY_FILE, text.encode("utf-8")) + self._writer.finalize() + else: + self.path.write_text(text) return self.path # -- read --------------------------------------------------------------- diff --git a/packages/flow/openadapt_flow/learning/teach.py b/packages/flow/openadapt_flow/learning/teach.py index 7f11d6eecc..230c5e9334 100644 --- a/packages/flow/openadapt_flow/learning/teach.py +++ b/packages/flow/openadapt_flow/learning/teach.py @@ -61,13 +61,14 @@ from __future__ import annotations import difflib +import os import re -import shutil from pathlib import Path from typing import Optional from pydantic import BaseModel, Field +from openadapt_flow import artifacts, sealed_set from openadapt_flow.ir import ( ActionKind, Anchor, @@ -101,7 +102,15 @@ class TeachError(Exception): """A teach cannot even be attempted (bad inputs), as distinct from a governed REFUSAL to promote a correction (which is a normal, reported - outcome -- see :attr:`TeachResult.promoted`).""" + outcome -- see :attr:`TeachResult.promoted`). + + ``public`` marks a message built only from the command line (the paths + it was given). Any other message may quote what teach read, so the CLI + withholds it when the input is sealed or strict mode is on.""" + + def __init__(self, message: str, *, public: bool = False) -> None: + super().__init__(message) + self.public = public # -- fix demonstration: a scripted correction spec --------------------------- @@ -146,12 +155,12 @@ def load_halt_report(run_dir: Path | str) -> RunReport: """ run = Path(run_dir) report_path = run / "report.json" - if not report_path.is_file(): + if not artifacts.is_file(report_path): raise TeachError( f"no run report at {report_path} -- teach needs the directory of a " "run that HALTED (holding report.json)" ) - report = RunReport.model_validate_json(report_path.read_text(encoding="utf-8")) + report = RunReport.model_validate_json(artifacts.read_text(report_path)) if report.halt is None: raise TeachError( f"the run at {run} did not halt (report.json has no halt observation) " @@ -168,7 +177,9 @@ def load_base_program(bundle_dir: Path | str) -> tuple[Workflow, ProgramGraph]: same governed learn path applies to both. """ bundle = Path(bundle_dir) - if not (bundle / "workflow.json").is_file(): + if not artifacts.is_file(bundle / "workflow.json") and not artifacts.is_file( + bundle / "workflow.json.enc" + ): raise TeachError( f"no bundle at {bundle} -- teach needs the base bundle that halted " "(the one whose replay produced the run report), via --bundle" @@ -185,22 +196,34 @@ def check_provenance( *, run_dir: Path | str, bundle_dir: Path | str, + private: bool = False, ) -> None: """Require that the halted run came from the bundle passed as ``--bundle``. A mismatched pair can't be taught: the correction would be spliced into a program that never ran, and the learn loop would only report a generic refusal. Raises :class:`TeachError` before anything is written. + + With ``private`` (sealed input or strict mode) the message names only the + paths on the command line, never a state id, workflow name or step intent + read from the encrypted report, so the CLI may print it. """ + halt = report.halt + assert halt is not None # guaranteed by load_halt_report + stale = TeachError( + f"the run at {run_dir} did not stop in a program --bundle {bundle_dir} " + "holds. Pass the bundle whose replay stopped.", + public=True, + ) if report.workflow_name != base_workflow.name: + if private: + raise stale raise TeachError( f"the run at {run_dir} came from the workflow " f"{report.workflow_name!r}, but --bundle {bundle_dir} is the " f"workflow {base_workflow.name!r}. Pass the bundle whose replay " "stopped." ) - halt = report.halt - assert halt is not None # guaranteed by load_halt_report graphs = [program, *base_workflow.subflows.values()] if any(halt.state_id in graph.states for graph in graphs): return @@ -212,6 +235,8 @@ def check_provenance( for state in graph.states.values() ): return + if private: + raise stale raise TeachError( f"the step where the run at {run_dir} stopped ({halt.intent!r}) is not " f"in --bundle {bundle_dir}. Pass the bundle whose replay stopped." @@ -413,7 +438,7 @@ def _texts_seen_before_halt( completed = result.ok and not result.skipped if completed or (delivered and result is halted): frames.append(run_dir / result.before_png) - readable = [frame for frame in frames if frame.is_file()] + readable = [frame for frame in frames if artifacts.is_file(frame)] if not readable: return None try: @@ -422,7 +447,7 @@ def _texts_seen_before_halt( seen: list[str] = [] for frame in readable: - for line in ocr(frame.read_bytes()): + for line in ocr(artifacts.open_read(frame)): text = (scrub_text(line.text) or "").strip() if text: seen.append(text) @@ -485,9 +510,13 @@ def _plan_correction( *, run_dir: Path, recorded: dict[str, Step], + private: bool = False, ) -> _Plan: """Build the ``(correction, baseline)`` traces, or a refusal reason. + With ``private`` a refusal names the halted step by its state id, never + by the intent read from the encrypted report. + Places the fix where the obstruction appeared. ``tail`` is the halted step and the remainder of the program. @@ -511,6 +540,7 @@ def _plan_correction( ) halt = report.halt assert halt is not None + step = f"halt {halt.state_id!r}" if private else repr(halt.intent) halted = _halted_result(report) delivered = _was_delivered(halted) halted_step = _halted_step(program, report) @@ -537,7 +567,7 @@ def _plan_correction( if halted_step is None: return _Plan( refusal=( - f"the run stopped after {halt.intent!r} acted, and teach " + f"the run stopped after {step} acted, and teach " "can't find that step in --bundle, so it can't place the " "fix after it" ) @@ -545,7 +575,7 @@ def _plan_correction( if repeats_halted and _can_write(halted_step): return _Plan( refusal=( - f"the run stopped after {halt.intent!r} was delivered, and " + f"the run stopped after {step} was delivered, and " "that step can write to the record. The fix repeats it, " "and nothing shows that the first delivery wrote nothing, " "so the learned workflow could save twice. Check the " @@ -555,7 +585,7 @@ def _plan_correction( if not repeats_halted and halted_step.effects: return _Plan( refusal=( - f"the run stopped after {halt.intent!r} was delivered, and " + f"the run stopped after {step} was delivered, and " "its write is checked against the system of record only " "when its screen check passes. After the fix nothing would " "check that write, so teach won't learn this fix" @@ -680,12 +710,14 @@ def _spec_from_recording( from openadapt_flow.compiler.compile import compile_recording recording = Path(recording_dir) - if not (recording / "events.jsonl").is_file(): + if not artifacts.is_file(recording / "events.jsonl"): raise TeachError( f"{recording} is not a recording directory (no events.jsonl). Pass a " "recording produced by `openadapt-flow record`, or a .json " "correction spec." ) + # Inside an encrypted recording this is a sealed bundle of its own, which + # the recording's set skips (compile_recording seals it). compiled_dir = recording / "_teach_compiled" compiled = compile_recording( recording, compiled_dir, name=f"{report.workflow_name}-fix" @@ -737,6 +769,10 @@ class TeachResult(BaseModel): promoted: bool outcome: LearnOutcome out_bundle: Optional[Path] = None + #: Sealed input or strict mode: the summary withholds what the learn loop + #: read (the skill id, which defaults to the workflow name, the reason, + #: clusters and gate failures). + private: bool = False @property def refused(self) -> bool: @@ -745,11 +781,17 @@ def refused(self) -> bool: def summary(self) -> str: """A short operator-facing verdict (what it learned / why it refused).""" verdict = "LEARNED" if self.promoted else "REFUSED" - lines = [ - f"{verdict}: {self.outcome.action} (skill {self.skill_id!r})", - f" reason: {self.outcome.reason}", - ] - if self.outcome.clusters: + from openadapt_flow import safe_errors + + skill = safe_errors.label(self.skill_id, private=self.private) + lines = [f"{verdict}: {self.outcome.action} (skill {skill})"] + if self.private: + lines.append( + " reason: withheld because the input is sealed or strict mode is on" + ) + else: + lines.append(f" reason: {self.outcome.reason}") + if self.outcome.clusters and not self.private: lines.append(f" clusters: {self.outcome.clusters}") if self.outcome.action != _NOT_ATTEMPTED: lines.append( @@ -760,8 +802,12 @@ def summary(self) -> str: if self.outcome.gate is not None: gate = "passed" if self.outcome.gate.passed else "FAILED" lines.append(f" regression gate: {gate}") - for failure in self.outcome.gate.failures: - lines.append(f" - {failure}") + if self.private and self.outcome.gate.failures: + count = len(self.outcome.gate.failures) + lines.append(f" {count} failure(s), details withheld") + elif not self.private: + for failure in self.outcome.gate.failures: + lines.append(f" - {failure}") if self.promoted and self.out_bundle is not None: lines.append(f" updated bundle written to {self.out_bundle}") lines.append( @@ -778,9 +824,10 @@ def summary(self) -> str: _NOT_ATTEMPTED = "not_attempted" -def _refusal(skill_id: str, reason: str) -> TeachResult: +def _refusal(skill_id: str, reason: str, *, private: bool = False) -> TeachResult: return TeachResult( skill_id=skill_id, + private=private, promoted=False, outcome=LearnOutcome( skill_id=skill_id, @@ -838,7 +885,7 @@ def canary(ctx: CanaryContext) -> tuple[bool, str]: continue if anchor is not None and ( anchor.template in carried - or (bundle_dir / anchor.template).is_file() + or artifacts.is_file(bundle_dir / anchor.template) ): continue return False, ( @@ -852,24 +899,119 @@ def canary(ctx: CanaryContext) -> tuple[bool, str]: return canary +def _write_crop(destination: Path, data: bytes) -> None: + """Write a crop into the taught bundle: encrypted when it is being sealed.""" + if not sealed_set.is_sealed_path(destination): + destination.parent.mkdir(parents=True, exist_ok=True) + sealed_set.write_bytes(destination, data) + + def _copy_templates(src_bundle: Path, out_bundle: Path) -> None: """Carry the base bundle's template crops (subfolders included, such as identifier crops) into the updated bundle so the promoted program can - resolve its anchors on re-run.""" - src_templates = src_bundle / "templates" - if not src_templates.is_dir(): - return - shutil.copytree(src_templates, out_bundle / "templates", dirs_exist_ok=True) + resolve its anchors on re-run. Sealed crops are decrypted in memory.""" + src_templates = Path(os.path.abspath(src_bundle / "templates")) + for template in artifacts.list_files(src_templates): + _write_crop( + out_bundle / "templates" / template.relative_to(src_templates), + artifacts.open_read(template), + ) def _copy_carried(carried: dict[str, Path], out_bundle: Path) -> None: """Copy the fix recording's crops to the paths the spliced step uses.""" for rel, source in carried.items(): - if not source.is_file(): + if not artifacts.is_file(source): raise TeachError(f"the fix recording's crop {source} is missing") - destination = out_bundle / rel - destination.parent.mkdir(parents=True, exist_ok=True) - shutil.copy2(source, destination) + _write_crop(out_bundle / rel, artifacts.open_read(source)) + + +def _seal_outputs(*inputs: Path | str) -> bool: + """Whether teach writes its library and bundle sealed: any input is sealed, + or strict mode (docs/phi_at_rest.md).""" + from openadapt_flow import at_rest + + return at_rest.strict() or any(artifacts.is_sealed(p) for p in inputs) + + +def _check_sealed_out(out_bundle: Path) -> None: + """A sealed taught bundle goes into a new or sealed directory, never over + a plaintext one.""" + if sealed_set.is_sealed(out_bundle): + return + if out_bundle.is_dir() and any(out_bundle.iterdir()): + raise TeachError( + f"--out {out_bundle} already holds an unencrypted bundle; pass a new " + "directory so the taught bundle stays encrypted" + ) + + +def _overlap(a: Path, b: Path) -> bool: + """Is ``a`` the same path as ``b`` (through symlinks, or the same inode), + or inside it, or around it?""" + ra, rb = Path(os.path.realpath(a)), Path(os.path.realpath(b)) + if ra == rb or ra.is_relative_to(rb) or rb.is_relative_to(ra): + return True + try: + return os.path.samefile(a, b) + except OSError: + return False + + +def check_out_is_new( + out: Path | str, *, library: Optional[Path | str] = None, **inputs: Path | str +) -> None: + """Refuse an ``--out`` (or its skill library) that is, holds or sits + inside an input: the base bundle, the halted run or the fix. Teach writes + the taught bundle there, so the base program would be overwritten. Checked + before anything is read or written; a symlink alias counts as the input.""" + out_path = Path(out) + lib = ( + Path(library) + if library is not None + else out_path.parent / f"{out_path.name}.skills" + ) + for target, flag in ((out_path, "--out"), (lib, "the skill library")): + for name, path in inputs.items(): + if _overlap(target, Path(path)): + raise TeachError( + f"{flag} {target} is or overlaps the {name} input {path}. " + "Pass a new directory: teach never writes over its inputs", + public=True, + ) + + +def _write_taught_bundle( + updated: Workflow, + bundle: Path, + out_bundle: Path, + carried: dict[str, Path], + run_dir: Path | str, +) -> None: + # Crops go in first: saving seals a manifest over the files present. + _copy_templates(bundle, out_bundle) + _copy_carried(carried, out_bundle) + updated.save(out_bundle) + # Governed repair promotion (Section 9): the taught bundle is a lifecycle + # CANDIDATE, never an implicitly active bundle. A detached + # repair/candidate.json is written into it; promotion to active requires + # the explicit lifecycle (campaigns, human approval, staged canary) via + # `openadapt-flow repair`. Registration failure fails CLOSED by absence: + # without the record the taught bundle cannot enter the lifecycle at all. + from openadapt_flow.repair.registration import register_bundle_candidate + + try: + register_bundle_candidate( + bundle, + out_bundle, + source="teach", + evidence_run_dir=Path(run_dir), + ) + except Exception as exc: + print( + "repair candidate registration failed (the taught bundle stays " + f"ungoverned and non-promotable): {exc}" + ) def teach( @@ -908,9 +1050,24 @@ def teach( fix, a run that did not come from ``bundle``, or a library at ``library_dir`` seeded from a different bundle). """ + check_out_is_new( + out, + library=library_dir, + **{"halted run": run_dir, "fix": fix, "base bundle": bundle}, + ) + seal = _seal_outputs(run_dir, fix, bundle) + if seal: + _check_sealed_out(Path(out)) report = load_halt_report(run_dir) base_workflow, program = load_base_program(bundle) - check_provenance(report, base_workflow, program, run_dir=run_dir, bundle_dir=bundle) + check_provenance( + report, + base_workflow, + program, + run_dir=run_dir, + bundle_dir=bundle, + private=seal, + ) sid = skill_id or report.workflow_name spec = _load_fix(fix) @@ -919,11 +1076,11 @@ def teach( if spec is None: spec, recorded, carried = _spec_from_recording(Path(fix), report) plan = _plan_correction( - spec, program, report, run_dir=Path(run_dir), recorded=recorded + spec, program, report, run_dir=Path(run_dir), recorded=recorded, private=seal ) if plan.refusal: # Refused before learning: nothing is written, not even a library. - return _refusal(sid, plan.refusal) + return _refusal(sid, plan.refusal, private=seal) assert plan.correction is not None and plan.baseline is not None lib_root = ( @@ -931,7 +1088,12 @@ def teach( if library_dir is not None else Path(out).parent / f"{Path(out).name}.skills" ) - library = SkillLibrary(lib_root) + try: + library = SkillLibrary(lib_root, seal=seal) + except sealed_set.SealedSetError as exc: + # Opening the library names only its directory (``--library`` or the + # one beside ``--out``) and the key it needs. + raise TeachError(str(exc), public=True) from exc _seed_or_check_library( library, sid, program, dict(base_workflow.subflows), bundle_dir=bundle ) @@ -950,7 +1112,7 @@ def teach( # Governed refusal: leave the bundle untouched -- the workflow stays # halting exactly as before the teach was attempted. return TeachResult( - skill_id=sid, promoted=False, outcome=outcome, out_bundle=None + skill_id=sid, promoted=False, outcome=outcome, out_bundle=None, private=seal ) updated = promoted_workflow(library, sid, name=base_workflow.name) @@ -964,30 +1126,25 @@ def teach( } ) out_bundle = Path(out) - # Crops go in first: saving seals a manifest over the files present. - _copy_templates(Path(bundle), out_bundle) - _copy_carried(carried, out_bundle) - updated.save(out_bundle) - # Governed repair promotion (Section 9): the taught bundle is a lifecycle - # CANDIDATE, never an implicitly active bundle. A detached - # repair/candidate.json is written into it; promotion to active requires - # the explicit lifecycle (campaigns, human approval, staged canary) via - # `openadapt-flow repair`. Registration failure fails CLOSED by absence: - # without the record the taught bundle cannot enter the lifecycle at all. - from openadapt_flow.repair.registration import register_bundle_candidate - - try: - register_bundle_candidate( - Path(bundle), + if seal: + # Every crop, the program and the candidate record are encrypted + # before they reach the disk (the same key rules as compile). + passphrase = sealed_set.output_bundle_key() + if not passphrase: + raise TeachError(sealed_set.missing_key_message(out_bundle, "bundle")) + with sealed_set.writing( out_bundle, - source="teach", - evidence_run_dir=Path(run_dir), - ) - except Exception as exc: - print( - "repair candidate registration failed (the taught bundle stays " - f"ungoverned and non-promotable): {exc}" - ) + "bundle", + passphrase, + resume=sealed_set.is_sealed(out_bundle), + ): + _write_taught_bundle(updated, Path(bundle), out_bundle, carried, run_dir) + else: + _write_taught_bundle(updated, Path(bundle), out_bundle, carried, run_dir) return TeachResult( - skill_id=sid, promoted=True, outcome=outcome, out_bundle=out_bundle + skill_id=sid, + promoted=True, + outcome=outcome, + out_bundle=out_bundle, + private=seal, ) diff --git a/packages/flow/openadapt_flow/repair/registration.py b/packages/flow/openadapt_flow/repair/registration.py index d59839e4a2..87f3c9a021 100644 --- a/packages/flow/openadapt_flow/repair/registration.py +++ b/packages/flow/openadapt_flow/repair/registration.py @@ -19,11 +19,13 @@ from __future__ import annotations import json +import os from pathlib import Path from typing import Optional from pydantic import BaseModel +from openadapt_flow import artifacts, sealed_set from openadapt_flow.ir import Workflow from openadapt_flow.qualification import EvidenceRef from openadapt_flow.repair.candidate import ( @@ -104,10 +106,11 @@ def _evidence_from_run_dir( """Collect heal-patch evidence refs + privacy-safe failure fingerprints.""" evidence: list[EvidenceRef] = [] fingerprints: list[FailureFingerprint] = [] + run_dir = Path(os.path.abspath(run_dir)) # list_files names absolute paths heals_dir = run_dir / "heals" - if heals_dir.is_dir(): - for patch_path in sorted(heals_dir.glob("*/patch.json")): - raw = patch_path.read_bytes() + if artifacts.is_dir(heals_dir): + for patch_path in artifacts.list_files(heals_dir, "*/patch.json"): + raw = artifacts.open_read(patch_path) rel = patch_path.relative_to(run_dir).as_posix() evidence.append( EvidenceRef(kind="other", sha256=sha256_hex(raw), relative_path=rel) @@ -126,11 +129,11 @@ def _evidence_from_run_dir( ) ) report_path = run_dir / "report.json" - if report_path.is_file(): + if artifacts.is_file(report_path): evidence.append( EvidenceRef( kind="run_report", - sha256=sha256_hex(report_path.read_bytes()), + sha256=sha256_hex(artifacts.open_read(report_path)), relative_path="report.json", ) ) @@ -223,17 +226,19 @@ def register_bundle_candidate( evidence_run_dir=evidence_run_dir, ) path = detached_candidate_path(proposed_bundle) - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(candidate.model_dump_json(indent=2)) + if not sealed_set.is_sealed_path(proposed_bundle): + path.parent.mkdir(parents=True, exist_ok=True) + # Encrypted when the proposed bundle is sealed (it names anchor text). + sealed_set.write_text(path, candidate.model_dump_json(indent=2)) return path def load_detached_candidate(proposed_bundle: Path | str) -> RepairCandidate: """Load the detached candidate record from a proposed bundle.""" path = detached_candidate_path(proposed_bundle) - if not path.is_file(): + if not artifacts.is_file(path): raise FileNotFoundError( f"no detached repair candidate at {path}; register the bundle " "pair first (openadapt-flow repair register)" ) - return RepairCandidate.model_validate_json(path.read_text()) + return RepairCandidate.model_validate_json(artifacts.read_text(path)) diff --git a/packages/flow/openadapt_flow/safe_errors.py b/packages/flow/openadapt_flow/safe_errors.py new file mode 100644 index 0000000000..ad6e72cfb4 --- /dev/null +++ b/packages/flow/openadapt_flow/safe_errors.py @@ -0,0 +1,58 @@ +"""Error text that never quotes the input it rejected. + +A Pydantic validation error quotes the offending value (``input_value=...``). +When that value was decrypted from a sealed artifact, printing the error +would write private content to stdout, stderr and any log that captures them +(docs/phi_at_rest.md). Every refusal the CLI prints goes through +:func:`message`, which drops those quoted values, and a failure caused by +sealed or strict-mode input prints :func:`withheld` instead of its text. + +Success output follows the same rule: a name read from sealed or strict-mode +input (a workflow name, a skill id) is shown only as :func:`label` gives it, +a short hash that tells two runs apart without quoting the name. +""" + +from __future__ import annotations + +import hashlib +import re + +# Pydantic renders ``[type=..., input_value=, input_type=...]`` on one +# line; a repr never spans lines. Greedy, so a value that itself contains +# ", input_type=" is dropped whole. +_INPUT_VALUE = re.compile(r"input_value=.*, input_type=") +_ANY_INPUT_VALUE = re.compile(r"input_value=[^\n]*") + + +def message(error: BaseException | str) -> str: + """``error`` as text, with every quoted input value removed.""" + text = _INPUT_VALUE.sub("input_type=", str(error)) + return _ANY_INPUT_VALUE.sub("input_value withheld]", text) + + +def root_cause(error: BaseException) -> BaseException: + """The first error in ``error``'s chain.""" + seen = {id(error)} + while True: + cause = error.__cause__ or error.__context__ + if cause is None or id(cause) in seen: + return error + seen.add(id(cause)) + error = cause + + +def withheld(error: BaseException) -> str: + """A refusal that names only the kind of the error that caused it.""" + return ( + f"the input was refused ({type(root_cause(error)).__name__}). The " + "details are withheld because the input is sealed or strict mode is on." + ) + + +def label(name: object, *, private: bool) -> str: + """``name`` for a summary line: quoted, or for private input only a short + hash of it.""" + if not private: + return repr(name) + digest = hashlib.sha256(str(name).encode("utf-8")).hexdigest()[:12] + return f"withheld, sha256 {digest}" diff --git a/packages/flow/openadapt_flow/sanitized_artifact.py b/packages/flow/openadapt_flow/sanitized_artifact.py index 3a9f998c21..a5114a6a1f 100644 --- a/packages/flow/openadapt_flow/sanitized_artifact.py +++ b/packages/flow/openadapt_flow/sanitized_artifact.py @@ -8,6 +8,12 @@ The built-in local reviewer binds only to loopback and serves no remote assets. It is intentionally small: deployments can replace it with a richer desktop UI while retaining the manifest and approval contracts in this module. + +A sealed source (an encrypted recording or bundle), or any source in strict +mode, gets a sealed derivative: a sealed set of kind ``sanitized`` +(:mod:`openadapt_flow.sealed_set`) whose files are decrypted in memory for the +rescan, the review page and the upload, and an approved archive encrypted +beside it. The scrubbing and approval checks are the same for both forms. """ from __future__ import annotations @@ -31,6 +37,7 @@ from datetime import datetime, timezone from functools import lru_cache from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from io import BytesIO from pathlib import Path, PurePosixPath from typing import Any, Optional from urllib.parse import parse_qs @@ -40,7 +47,7 @@ except ModuleNotFoundError: # pragma: no cover - exercised on 3.10 CI import tomli as tomllib -from openadapt_flow import privacy +from openadapt_flow import artifacts, crypto, privacy, sealed_set MANIFEST_NAME = ".openadapt-sanitization.json" APPROVAL_NAME = ".openadapt-approval.json" @@ -82,7 +89,50 @@ def _sha256_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest() +#: Associated data of a sealed derivative's encrypted approved archive. +_ARCHIVE_AAD = b"openadapt-flow/sanitized-archive" + + +def _read(path: Path) -> bytes: + """An artifact file's bytes, decrypted in memory when it is sealed.""" + return artifacts.open_read(path) + + +def _read_text(path: Path) -> str: + """An artifact file's text, with the newline translation of + :meth:`pathlib.Path.read_text` for sealed files too.""" + if not artifacts.is_sealed(path): + return path.read_text(encoding="utf-8") + text = artifacts.read_text(path) + return text.replace("\r\n", "\n").replace("\r", "\n") + + +def _write(path: Path, data: bytes) -> None: + """Write a derivative file: encrypted inside a sealed derivative.""" + if not artifacts.is_sealed(path): + path.parent.mkdir(parents=True, exist_ok=True) + sealed_set.write_bytes(path, data) + + +def _seals_derivative(source: Path) -> bool: + """A sealed source, or strict mode, makes a sealed derivative.""" + from openadapt_flow import at_rest + + return artifacts.is_sealed(source) or at_rest.strict() + + +def _derivative_key(destination: Path) -> str: + passphrase = sealed_set.passphrase_for("sanitized") + if not passphrase: + raise SanitizationError( + sealed_set.missing_key_message(destination, "sanitized") + ) + return passphrase + + def _sha256_file(path: Path) -> str: + if artifacts.is_sealed(path): + return _sha256_bytes(_read(path)) digest = hashlib.sha256() with path.open("rb") as fh: for chunk in iter(lambda: fh.read(1024 * 1024), b""): @@ -111,6 +161,13 @@ def _scrubber_version() -> str: def _source_files(src: Path) -> list[Path]: if not src.is_dir(): raise SanitizationError(f"Artifact is not a directory: {src}") + if sealed_set.is_sealed(src): + # Logical names from the authenticated index; content decrypts later. + base = Path(os.path.abspath(src)) + sealed_files = [src / p.relative_to(base) for p in artifacts.list_files(base)] + if not sealed_files: + raise SanitizationError(f"Artifact has no files: {src}") + return sealed_files files: list[Path] = [] for path in sorted(src.rglob("*")): if path.is_symlink(): @@ -548,7 +605,7 @@ def _scrub_text_file( known_paths: set[str], ) -> tuple[bytes, list[dict[str, Any]]]: try: - raw = source.read_text(encoding="utf-8") + raw = _read_text(source) except UnicodeDecodeError as exc: raise SanitizationError(f"Text artifact is not valid UTF-8: {rel}") from exc if "\x00" in raw: @@ -617,8 +674,11 @@ def _validate_structured_text(suffix: str, text: str, rel: str) -> None: def _forbidden_binary_type(path: Path) -> Optional[str]: - with path.open("rb") as fh: - head = fh.read(512) + if artifacts.is_sealed(path): + head = _read(path)[:512] + else: + with path.open("rb") as fh: + head = fh.read(512) signatures = ( (b"SQLite format 3\x00", "SQLite database"), (b"PK\x03\x04", "ZIP/archive"), @@ -648,7 +708,7 @@ def _scrub_image_file( *, preserve_stable_bytes: bool = False, ) -> tuple[bytes, list[dict[str, Any]]]: - original = source.read_bytes() + original = _read(source) first = privacy.scrub_image_bytes(original, force=True) first, applied = _redact_image_rectangles(first, redactions, rel) second = privacy.scrub_image_bytes(first, force=True) @@ -706,16 +766,25 @@ def _image_metadata_free(data: bytes) -> bool: def _write_manifest(dest: Path, manifest: dict[str, Any]) -> None: - (dest / MANIFEST_NAME).write_text( - json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8" + _write( + dest / MANIFEST_NAME, + (json.dumps(manifest, indent=2, sort_keys=True) + "\n").encode("utf-8"), ) +def _derivative_files(dest: Path) -> list[Path]: + """Every file of a derivative, sorted; logical names when it is sealed.""" + if artifacts.is_sealed(dest): + base = Path(os.path.abspath(dest)) + return sorted(dest / p.relative_to(base) for p in artifacts.list_files(base)) + return sorted(path for path in dest.rglob("*") if path.is_file()) + + def _tree_rows(dest: Path) -> list[tuple[str, str]]: return [ (str(path.relative_to(dest)), _sha256_file(path)) - for path in sorted(dest.rglob("*")) - if path.is_file() and path.name not in _SYSTEM_FILES + for path in _derivative_files(dest) + if path.name not in _SYSTEM_FILES ] @@ -735,8 +804,8 @@ def _reseal_sanitized_bundle(bundle: Path) -> None: provenance.expires_at = None workflow.save(bundle) workflow_py = bundle / "workflow.py" - if workflow_py.is_file(): - workflow_py.write_text(render_workflow_py(workflow), encoding="utf-8") + if artifacts.is_file(workflow_py): + _write(workflow_py, render_workflow_py(workflow).encode("utf-8")) _RECORDING_FRAME_PATH = re.compile( @@ -781,13 +850,13 @@ def sanitize_artifact( if source == destination or source in destination.parents: raise SanitizationError("Destination must not be the source or inside it") files = _source_files(source) - if kind == "bundle" and (source / "workflow.py").is_file(): + if kind == "bundle" and artifacts.is_file(source / "workflow.py"): try: from openadapt_flow.compiler.codegen import render_workflow_py from openadapt_flow.ir import Workflow expected = render_workflow_py(Workflow.load(source)) - actual = (source / "workflow.py").read_text(encoding="utf-8") + actual = _read_text(source / "workflow.py") except Exception as exc: # noqa: BLE001 - normalize bundle validation raise SanitizationError( f"Cannot validate generated workflow.py: {exc}" @@ -797,7 +866,7 @@ def sanitize_artifact( "workflow.py is not the deterministic rendering of workflow.json" ) canonical_bundle = kind == "bundle" and all( - (source / name).is_file() + artifacts.is_file(source / name) for name in ("workflow.json", "workflow.py", "manifest.json") ) scrubber = privacy.get_scrubber() @@ -811,8 +880,26 @@ def sanitize_artifact( raise SanitizationError(f"Destination already exists: {destination}") shutil.rmtree(destination) redactions = _load_redactions(redactions_file) - staging = Path(tempfile.mkdtemp(prefix="openadapt-sanitize-")) / "artifact" - staging.mkdir(parents=True) + sealed = _seals_derivative(source) + writer: Optional[sealed_set.SealedSetWriter] = None + if sealed: + # A sealed derivative is staged beside its destination as a sealed + # set (0700), so only ciphertext reaches the disk, then renamed. + destination.parent.mkdir(parents=True, exist_ok=True) + staging_root = Path( + tempfile.mkdtemp( + prefix=f".{destination.name}.", dir=str(destination.parent) + ) + ) + staging = staging_root / "artifact" + writer = sealed_set.SealedSetWriter( + staging, "sanitized", _derivative_key(destination) + ) + sealed_set.register_writer(writer) + else: + staging_root = Path(tempfile.mkdtemp(prefix="openadapt-sanitize-")) + staging = staging_root / "artifact" + staging.mkdir(parents=True) entries: list[dict[str, Any]] = [] semantics = "preserved" try: @@ -852,7 +939,7 @@ def sanitize_artifact( # Both files are deterministic derivatives of workflow.json. # Preserve them byte-for-byte when workflow data is unchanged; # a load-bearing transform reseals/regenerates them below. - output, manual = path.read_bytes(), [] + output, manual = _read(path), [] handler = "utf8-text" elif suffix in _TEXT_SUFFIXES: output, manual = _scrub_text_file( @@ -879,9 +966,7 @@ def sanitize_artifact( "archive, encrypted, executable, and unknown files require a " "deployment-specific sanitizer and remain local." ) - target = staging / safe_rel - target.parent.mkdir(parents=True, exist_ok=True) - target.write_bytes(output) + _write(staging / safe_rel, output) source_hash = _sha256_file(path) derivative_hash = _sha256_bytes(output) changed = renamed or source_hash != derivative_hash @@ -946,14 +1031,19 @@ def sanitize_artifact( ], } _write_manifest(staging, manifest) - destination.parent.mkdir(parents=True, exist_ok=True) - shutil.move(str(staging), str(destination)) + if writer is not None: + writer.finalize() + sealed_set.unregister_writer(writer) + writer = None + os.rename(staging, destination) + else: + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.move(str(staging), str(destination)) return manifest - except Exception: - shutil.rmtree(staging.parent, ignore_errors=True) - raise finally: - shutil.rmtree(staging.parent, ignore_errors=True) + if writer is not None: + sealed_set.unregister_writer(writer) + shutil.rmtree(staging_root, ignore_errors=True) def load_and_verify_derivative(destination: Path) -> dict[str, Any]: @@ -965,10 +1055,10 @@ def load_and_verify_derivative(destination: Path) -> dict[str, Any]: f"Derivative contains an unreviewed symlink: {path.relative_to(destination)}" ) manifest_path = destination / MANIFEST_NAME - if not manifest_path.is_file(): + if not artifacts.is_file(manifest_path): raise SanitizationError(f"No {MANIFEST_NAME} in {destination}") try: - manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + manifest = json.loads(_read_text(manifest_path)) except json.JSONDecodeError as exc: raise SanitizationError(f"Invalid sanitization manifest: {exc}") from exc if manifest.get("schema_version") != SCHEMA_VERSION: @@ -1006,7 +1096,7 @@ def _rescan_derivative(destination: Path) -> dict[str, Any]: for entry in manifest["files"]: path = destination / entry["path"] if entry["handler"] == "utf8-text": - current_text = path.read_text(encoding="utf-8") + current_text = _read_text(path) if manifest["kind"] == "bundle" and entry["path"] in { "workflow.py", "manifest.json", @@ -1051,7 +1141,7 @@ def _rescan_derivative(destination: Path) -> dict[str, Any]: f"Approval rescan found a remaining identifier in {entry['path']}" ) elif entry["handler"] == "image-ocr-redaction": - current_bytes = path.read_bytes() + current_bytes = _read(path) rescanned_bytes = privacy.scrub_image_bytes(current_bytes, force=True) if not _image_pixels_equal(rescanned_bytes, current_bytes): raise SanitizationError( @@ -1069,15 +1159,35 @@ def approved_archive_path(destination: Path) -> Path: return destination.parent / f"{destination.name}.approved.zip" -def _write_deterministic_archive(destination: Path) -> Path: - """Freeze the reviewed bytes with stable names, order, metadata, and time.""" +def read_approved_archive(destination: Path) -> bytes: + """The approved archive's bytes, decrypted in memory for a sealed + derivative. These are the bytes the approval hashes and push uploads.""" + destination = Path(destination) + raw = approved_archive_path(destination).read_bytes() + if not sealed_set.is_sealed(destination): + return raw + if not crypto.is_encrypted(raw): + raise SanitizationError( + "The approved archive of an encrypted derivative is not encrypted" + ) + try: + return crypto.decrypt_bytes(raw, _derivative_key(destination), aad=_ARCHIVE_AAD) + except crypto.CryptoError as exc: + raise SanitizationError(f"Approved archive cannot be opened: {exc}") from exc + + +def _write_deterministic_archive(destination: Path) -> bytes: + """Freeze the reviewed bytes with stable names, order, metadata, and time. + + The zip is built in memory. A sealed derivative's archive is encrypted + before it is written; the approval hashes the zip bytes either way.""" archive = approved_archive_path(destination) - tmp = archive.with_suffix(".zip.tmp") + buffer = BytesIO() with zipfile.ZipFile( - tmp, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9 + buffer, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9 ) as zf: - for path in sorted(destination.rglob("*")): - if not path.is_file() or path.name == APPROVAL_NAME: + for path in _derivative_files(destination): + if path.name == APPROVAL_NAME: continue rel = path.relative_to(destination).as_posix() info = zipfile.ZipInfo(rel, date_time=(1980, 1, 1, 0, 0, 0)) @@ -1086,12 +1196,20 @@ def _write_deterministic_archive(destination: Path) -> Path: info.create_system = 3 zf.writestr( info, - path.read_bytes(), + _read(path), compress_type=zipfile.ZIP_DEFLATED, compresslevel=9, ) + data = buffer.getvalue() + stored = data + if sealed_set.is_sealed(destination): + stored = crypto.encrypt_bytes( + data, _derivative_key(destination), aad=_ARCHIVE_AAD + ) + tmp = archive.with_suffix(".zip.tmp") + tmp.write_bytes(stored) os.replace(tmp, archive) - return archive + return data def approve_derivative( @@ -1131,8 +1249,8 @@ def approve_derivative( } manifest["approval_verification"] = existing_verification _write_manifest(destination, manifest) - archive = _write_deterministic_archive(destination) - archive_hash = _sha256_file(archive) + archive_bytes = _write_deterministic_archive(destination) + archive_hash = _sha256_bytes(archive_bytes) approval = { "schema_version": SCHEMA_VERSION, "approved_at": _utc_now(), @@ -1142,7 +1260,7 @@ def approve_derivative( "manifest_sha256": _manifest_hash(manifest), "derivative_tree_sha256": manifest["derivative_tree_sha256"], "approved_derivative_sha256": archive_hash, - "approved_archive_size_bytes": archive.stat().st_size, + "approved_archive_size_bytes": len(archive_bytes), "verification": { "verified_at": existing_verification["verified_at"], "method": "full-stable-rescan", @@ -1151,8 +1269,9 @@ def approve_derivative( "unresolved": 0, }, } - approval_path(destination).write_text( - json.dumps(approval, indent=2, sort_keys=True) + "\n", encoding="utf-8" + _write( + approval_path(destination), + (json.dumps(approval, indent=2, sort_keys=True) + "\n").encode("utf-8"), ) return approval @@ -1161,7 +1280,7 @@ def load_valid_approval(destination: Path) -> dict[str, Any]: destination = Path(destination) manifest = load_and_verify_derivative(destination) path = approval_path(destination) - if not path.is_file(): + if not artifacts.is_file(path): archive = approved_archive_path(destination) if archive.is_file(): # The approval binds the archive's SHA-256, so `approve-sanitized` @@ -1180,7 +1299,7 @@ def load_valid_approval(destination: Path) -> dict[str, Any]: "locally, then approve it." ) try: - approval = json.loads(path.read_text(encoding="utf-8")) + approval = json.loads(_read_text(path)) except json.JSONDecodeError as exc: raise SanitizationError(f"Invalid approval: {exc}") from exc if approval.get("manifest_sha256") != _manifest_hash(manifest): @@ -1190,9 +1309,10 @@ def load_valid_approval(destination: Path) -> dict[str, Any]: archive = approved_archive_path(destination) if not archive.is_file(): raise SanitizationError("Approved immutable archive is missing") - if approval.get("approved_derivative_sha256") != _sha256_file(archive): + archive_bytes = read_approved_archive(destination) + if approval.get("approved_derivative_sha256") != _sha256_bytes(archive_bytes): raise SanitizationError("Approved immutable archive changed") - if approval.get("approved_archive_size_bytes") != archive.stat().st_size: + if approval.get("approved_archive_size_bytes") != len(archive_bytes): raise SanitizationError("Approved immutable archive size changed") return approval @@ -1541,7 +1661,7 @@ def build_ingest_manifest(destination: Path) -> dict[str, Any]: def _update_entry(destination: Path, rel: str, redaction: dict[str, Any]) -> None: manifest_path = destination / MANIFEST_NAME try: - manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + manifest = json.loads(_read_text(manifest_path)) except (OSError, json.JSONDecodeError) as exc: raise SanitizationError(f"Cannot update sanitization manifest: {exc}") from exc entry = next((item for item in manifest["files"] if item["path"] == rel), None) @@ -1561,7 +1681,7 @@ def _update_entry(destination: Path, rel: str, redaction: dict[str, Any]) -> Non manifest["runtime_semantics_validated"] = False manifest["trusted_boundary_required_at_runtime"] = True manifest.pop("approval_verification", None) - approval_path(destination).unlink(missing_ok=True) + sealed_set.unlink(approval_path(destination), missing_ok=True) approved_archive_path(destination).unlink(missing_ok=True) _write_manifest(destination, manifest) @@ -1583,7 +1703,7 @@ def _authorized_review_path( if entry is None: raise SanitizationError(f"File is not an authorized {handler} artifact: {rel}") path = (destination / entry["path"]).resolve() - if destination not in path.parents or not path.is_file(): + if destination not in path.parents or not artifacts.is_file(path): raise SanitizationError(f"Review path escapes the derivative: {rel}") return path, entry @@ -1595,11 +1715,11 @@ def add_manual_text_redaction( raise SanitizationError("Manual text redaction cannot be empty") destination = Path(destination) path, _ = _authorized_review_path(destination, rel, handler="utf8-text") - text = path.read_text(encoding="utf-8") + text = _read_text(path) count = text.count(literal) if count == 0: raise SanitizationError(f"Manual redaction text was not found in {rel}") - path.write_text(text.replace(literal, replacement), encoding="utf-8") + _write(path, text.replace(literal, replacement).encode("utf-8")) _update_entry( destination, rel, @@ -1619,11 +1739,11 @@ def add_manual_image_redaction( destination = Path(destination) path, _ = _authorized_review_path(destination, rel, handler="image-ocr-redaction") output, applied = _redact_image_rectangles( - path.read_bytes(), + _read(path), [{"file": rel, "x": x, "y": y, "width": width, "height": height}], rel, ) - path.write_bytes(output) + _write(path, output) redaction = dict(applied[0]) redaction["reviewer_added"] = True _update_entry(destination, rel, redaction) @@ -1648,8 +1768,8 @@ def render_review_html(source: Path, destination: Path, csrf_token: str) -> str: sanitized = destination / entry["path"] rel = html.escape(entry["path"]) if entry["handler"] == "utf8-text": - before = html.escape(original.read_text(encoding="utf-8")[:200_000]) - after = html.escape(sanitized.read_text(encoding="utf-8")[:200_000]) + before = html.escape(_read_text(original)[:200_000]) + after = html.escape(_read_text(sanitized)[:200_000]) preview = f"
{before}
{after}
" form = f"""
@@ -1661,8 +1781,8 @@ def render_review_html(source: Path, destination: Path, csrf_token: str) -> str:
""" else: mime = mimetypes.guess_type(sanitized.name)[0] or "image/png" - before = base64.b64encode(original.read_bytes()).decode("ascii") - after = base64.b64encode(sanitized.read_bytes()).decode("ascii") + before = base64.b64encode(_read(original)).decode("ascii") + after = base64.b64encode(_read(sanitized)).decode("ascii") preview = ( "
" f"original {rel}" diff --git a/packages/flow/openadapt_flow/sealed_set.py b/packages/flow/openadapt_flow/sealed_set.py index 732a44275b..771c248971 100644 --- a/packages/flow/openadapt_flow/sealed_set.py +++ b/packages/flow/openadapt_flow/sealed_set.py @@ -23,8 +23,9 @@ The ``set_digest`` is the SHA-256 of the canonical index. It names the exact file set, and ``seal`` prints it. -Every file kind is classified in :data:`BUNDLE_FILES` and -:data:`RECORDING_FILES`. Writing into a sealed set refuses a file the +Every file kind is classified in :data:`BUNDLE_FILES`, +:data:`RECORDING_FILES`, :data:`RUN_FILES`, :data:`LIBRARY_FILES` and +:data:`SANITIZED_FILES`. Writing into a sealed set refuses a file the inventory does not know, rather than passing it through. ``structural`` files hold only hashes, counts and format markers and stay readable without the key; every other file is encrypted. @@ -63,7 +64,7 @@ FORMAT_VERSION: Final[int] = 1 SEALED_SUFFIX: Final[str] = ".enc" -Kind = Literal["bundle", "recording", "run"] +Kind = Literal["bundle", "recording", "run", "library", "sanitized"] Codec = Literal["set", "plain", "external"] Classification = Literal["phi", "structural"] @@ -269,12 +270,44 @@ class FileKind: ) +#: A skill library written by ``teach`` (:mod:`openadapt_flow.learning.library`). +LIBRARY_FILES: Final[tuple[FileKind, ...]] = ( + FileKind( + "skills.json", + "phi", + "every learned program version (target labels, typed literals, screen " + "text, crop paths) and the execution traces it was learned from", + ), +) + +#: The sanitized review copy ``push`` makes of a sealed source +#: (:mod:`openadapt_flow.sanitized_artifact`). Scrubbing is best effort until a +#: person approves it, so every file stays encrypted. +SANITIZED_FILES: Final[tuple[FileKind, ...]] = ( + FileKind( + ".openadapt-sanitization.json", + "phi", + "the redaction manifest: file names, hashes and redaction records", + ), + FileKind(".openadapt-approval.json", "phi", "the reviewer's approval"), + FileKind( + "*", + "phi", + "the scrubbed copy of each recording or bundle file, before review", + ), +) + +_INVENTORIES: Final[dict[str, tuple[FileKind, ...]]] = { + "bundle": BUNDLE_FILES, + "recording": RECORDING_FILES, + "run": RUN_FILES, + "library": LIBRARY_FILES, + "sanitized": SANITIZED_FILES, +} + + def inventory(kind: Kind) -> tuple[FileKind, ...]: - if kind == "bundle": - return BUNDLE_FILES - if kind == "recording": - return RECORDING_FILES - return RUN_FILES + return _INVENTORIES[kind] def classify(kind: Kind, rel: str) -> Optional[FileKind]: @@ -420,7 +453,7 @@ def read_header(root: Path | str) -> dict[str, object]: raise SealedSetError( f"sealed-set format version {header.get('version')!r} is not supported" ) - if header.get("kind") not in ("bundle", "recording", "run"): + if header.get("kind") not in _INVENTORIES: raise SealedSetError("sealed-set header has no valid kind") return header @@ -1103,8 +1136,10 @@ def require_readable(path: Path | str, what: str) -> None: __all__ = [ "BUNDLE_FILES", "HEADER", + "LIBRARY_FILES", "RECORDING_ENV_KEY", "RECORDING_FILES", + "SANITIZED_FILES", "SealedSet", "SealedSetError", "SealedSetKeyError", diff --git a/packages/flow/tests/test_artifact_reads.py b/packages/flow/tests/test_artifact_reads.py index f9f4848028..ab6f10facc 100644 --- a/packages/flow/tests/test_artifact_reads.py +++ b/packages/flow/tests/test_artifact_reads.py @@ -21,6 +21,7 @@ import ast import json +import os from pathlib import Path import pytest @@ -79,6 +80,43 @@ "_cmd_route", ) CONVERTED["composition.py"] += ("load",) +# teach: the halted run, the fix recording, the base bundle, the library. +CONVERTED["__main__.py"] += ("_cmd_teach",) +CONVERTED["learning/teach.py"] = ( + "load_halt_report", + "load_base_program", + "_texts_seen_before_halt", + "_spec_from_recording", + "_real_target_canary", + "_copy_templates", + "_copy_carried", + "teach", +) +CONVERTED["learning/library.py"] = ("_load",) +# push, and the review and approval of its sanitized copy. +CONVERTED["hosted.py"] = ("push", "_is_bundle_dir") +CONVERTED["sanitized_artifact.py"] = ( + "_source_files", + "_scrub_text_file", + "_scrub_image_file", + "_tree_rows", + "_reseal_sanitized_bundle", + "sanitize_artifact", + "load_and_verify_derivative", + "_rescan_derivative", + "_write_deterministic_archive", + "approve_derivative", + "load_valid_approval", + "_update_entry", + "_authorized_review_path", + "add_manual_text_redaction", + "add_manual_image_redaction", + "render_review_html", +) +CONVERTED["repair/registration.py"] = ( + "_evidence_from_run_dir", + "load_detached_candidate", +) CONVERTED["admitted_composition.py"] += ( "load", "load_child_envelope", @@ -92,6 +130,16 @@ #: (module, function, source of the call) -> why a direct read is right. ALLOWED: dict[tuple[str, str, str], str] = { + ( + "learning/library.py", + "_load", + "self._writer.is_file(_LIBRARY_FILE)", + ): "the sealed library's own writer, opened and hash-checked with its key", + ( + "learning/library.py", + "_load", + "self._writer.read_bytes(_LIBRARY_FILE)", + ): "the sealed library's own writer, opened and hash-checked with its key", ( "__main__.py", "_cmd_record", @@ -128,6 +176,22 @@ "first_screen_probe", "Path(explicit).read_bytes()", ): "the --first-screen PNG a person passes in", + ("hosted.py", "push", "derivative.exists()"): "the review copy's directory", + ( + "sanitized_artifact.py", + "_source_files", + "path.is_file()", + ): "walks a plain source; a sealed one is listed from its index above", + ( + "sanitized_artifact.py", + "sanitize_artifact", + "destination.exists()", + ): "the new derivative directory, not an artifact", + ( + "sanitized_artifact.py", + "load_valid_approval", + "archive.is_file()", + ): "the archive file beside the derivative, read by read_approved_archive", } @@ -445,3 +509,808 @@ def test_name_path_arguments_go_through_the_missing_key_check( err = capsys.readouterr().err assert "OPENADAPT_BUNDLE_KEY" in err and "Nothing was read" in err assert not (tmp_path / "composed").exists() + + +# -- teach on sealed input ---------------------------------------------------- + + +def _seal_dir(plain: Path, root: Path, kind: str) -> Path: + """Copy a plaintext recording or run directory into a sealed set.""" + with sealed_set.writing(root, kind, KEY): # type: ignore[arg-type] + for path in sorted(plain.rglob("*")): + if path.is_file(): + rel = path.relative_to(plain).as_posix() + sealed_set.write_bytes(root / rel, path.read_bytes()) + return root + + +def _teach_inputs(tmp_path: Path) -> tuple[Path, Path, Path]: + """A halted run, a fix recording and the bundle that halted, all sealed.""" + from tests.test_teach_cli import _base_bundle, _dismiss_recording, _halt_run + + plain = tmp_path / "plain" + plain.mkdir() + base = _base_bundle(plain) + _halt_run(base, plain / "run") + _dismiss_recording(plain / "fix") + bundle = tmp_path / "bundle" + assert main(["seal", str(base), "--out", str(bundle)]) == 0 + run = _seal_dir(plain / "run", tmp_path / "run", "run") + fix = _seal_dir(plain / "fix", tmp_path / "fix", "recording") + return run, fix, bundle + + +def _assert_taught_sealed(out: Path, fix: Path) -> None: + for root in (out, out.parent / f"{out.name}.skills", fix / "_teach_compiled"): + assert sealed_set.is_sealed(root), root + _assert_ciphertext_only(root) + assert sealed_set.header_kind(out.parent / f"{out.name}.skills") == "library" + if sealed_set.is_sealed(fix): + # The fix recording still opens: its compiled copy is a separate set. + sealed_set.SealedSet.open(fix, KEY).verify_all() + assert sealed_set.is_file(out / "repair" / "candidate.json") + learned = Workflow.load(out, key=KEY) + clicks = [ + state.step.anchor.click_point + for state in learned.program.states.values() # type: ignore[union-attr] + if state.step is not None and state.step.anchor is not None + ] + assert (140, 108) in clicks # the recorded dismiss click was spliced in + + +def test_teach_reads_sealed_inputs_and_writes_sealed( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys +) -> None: + from openadapt_flow.learning.library import SkillLibrary + + _ready("openadapt_flow.__main__:_cmd_teach") + monkeypatch.setenv(crypto.ENV_KEY, KEY) + run, fix, bundle = _teach_inputs(tmp_path) + before = {p: _files(p) for p in (run, bundle)} + out = tmp_path / "taught" + args = ["teach", str(run), "--fix", str(fix), "--bundle", str(bundle)] + name = Workflow.load(bundle).name + assert name not in str(tmp_path) + capsys.readouterr() + assert main([*args, "--out", str(out)]) == 0 + printed = capsys.readouterr().out + # The summary names no skill id (the workflow name) read from sealed input. + assert "LEARNED" in printed and name not in printed and "withheld" in printed + _assert_taught_sealed(out, fix) + assert {p: _files(p) for p in (run, bundle)} == before + library = SkillLibrary(tmp_path / "taught.skills") + assert library.sealed and library.skill_ids() + # A second teach reads the sealed library back: the fix is learned already. + assert main([*args, "--out", str(out)]) == 1 + printed = capsys.readouterr().out + assert "no_change" in printed and name not in printed + _assert_taught_sealed(out, fix) + + +def test_teach_in_strict_mode_seals_plain_inputs_outputs( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from tests.test_teach_cli import _base_bundle, _dismiss_recording, _halt_run + + monkeypatch.setenv(crypto.ENV_KEY, KEY) + base = _base_bundle(tmp_path) + _halt_run(base, tmp_path / "run") + fix = _dismiss_recording(tmp_path / "fix") + monkeypatch.setenv("OPENADAPT_FLOW_AT_REST", "strict") + out = tmp_path / "taught" + args = ["teach", str(tmp_path / "run"), "--fix", str(fix)] + assert main([*args, "--bundle", str(base), "--out", str(out)]) == 0 + _assert_taught_sealed(out, fix) + + +def test_teach_refuses_a_plaintext_library_for_sealed_input( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys +) -> None: + monkeypatch.setenv(crypto.ENV_KEY, KEY) + run, fix, bundle = _teach_inputs(tmp_path) + library = tmp_path / "plain-library" + library.mkdir() + (library / "skills.json").write_text('{"skills": {}}') + args = ["teach", str(run), "--fix", str(fix), "--bundle", str(bundle)] + out = tmp_path / "taught" + assert main([*args, "--out", str(out), "--library", str(library)]) == 2 + assert "unencrypted skill library" in capsys.readouterr().out + assert not out.exists() + + +_MARKER = "PRIVATE-MARKER-4f1c" + + +def _marker_absent(root: Path, *printed: str) -> None: + """The marker is in no printed text, log record or file under ``root``.""" + for text in printed: + assert _MARKER not in text + for path in root.rglob("*"): + if path.is_file(): + assert _MARKER.encode() not in path.read_bytes(), path + + +@pytest.mark.parametrize("stale", ["state", "workflow"]) +def test_teach_refusal_on_a_sealed_run_never_prints_its_contents( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capfd, caplog, stale: str +) -> None: + """A provenance refusal names only the paths on the command line, never a + state id, step intent or workflow name read from the encrypted report.""" + from tests.test_teach_cli import _base_bundle, _dismiss_recording, _halt_run + + monkeypatch.setenv(crypto.ENV_KEY, KEY) + plain = tmp_path / "plain" + plain.mkdir() + base = _base_bundle(plain) + _halt_run(base, plain / "run") + fix = _dismiss_recording(plain / "fix") + report = json.loads((plain / "run" / "report.json").read_text()) + if stale == "state": + report["halt"]["state_id"] = "s::gone" + report["halt"]["intent"] = _MARKER + else: + report["workflow_name"] = _MARKER + (plain / "run" / "report.json").write_text(json.dumps(report)) + run = _seal_dir(plain / "run", tmp_path / "run", "run") + for path in (plain / "run").rglob("*"): + if path.is_file(): + path.unlink() + monkeypatch.setenv("OPENADAPT_FLOW_AT_REST", "strict") + out = tmp_path / "taught" + args = ["teach", str(run), "--fix", str(fix), "--bundle", str(base)] + capfd.readouterr() + caplog.set_level("DEBUG") + assert main([*args, "--out", str(out)]) == 2 + printed = capfd.readouterr() + _marker_absent(tmp_path, printed.out, printed.err, caplog.text) + assert "teach cannot run" in printed.out + assert "did not stop in a program --bundle" in printed.out + assert repr(report["halt"]["state_id"]) not in printed.out + assert not out.exists() + + +@pytest.mark.parametrize("alias", ["same path", "symlink", "inside"]) +def test_teach_never_writes_over_the_sealed_base( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys, alias: str +) -> None: + monkeypatch.setenv(crypto.ENV_KEY, KEY) + run, fix, bundle = _teach_inputs(tmp_path) + if alias == "same path": + out = bundle + elif alias == "symlink": + out = tmp_path / "alias" + out.symlink_to(bundle, target_is_directory=True) + else: + out = bundle / "taught" + before = {p: _files(p) for p in (run, fix, bundle)} + args = ["teach", str(run), "--fix", str(fix), "--bundle", str(bundle)] + capsys.readouterr() + assert main([*args, "--out", str(out)]) == 2 + printed = capsys.readouterr().out + assert "teach cannot run" in printed and "overlaps the base bundle" in printed + assert {p: _files(p) for p in (run, fix, bundle)} == before + assert not (out.parent / f"{out.name}.skills").exists() + assert not (fix / "_teach_compiled").exists() + assert Workflow.load(bundle, key=KEY).name + + +@pytest.mark.parametrize("which", ["run", "fix"]) +def test_teach_refuses_an_out_that_is_another_input( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys, which: str +) -> None: + monkeypatch.setenv(crypto.ENV_KEY, KEY) + run, fix, bundle = _teach_inputs(tmp_path) + out = {"run": run, "fix": fix}[which] + before = {p: _files(p) for p in (run, fix, bundle)} + args = ["teach", str(run), "--fix", str(fix), "--bundle", str(bundle)] + assert main([*args, "--out", str(out)]) == 2 + assert "overlaps the" in capsys.readouterr().out + assert {p: _files(p) for p in (run, fix, bundle)} == before + + +# -- push of a sealed source -------------------------------------------------- + + +@pytest.fixture +def push_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + import base64 + + from openadapt_flow import hosted, privacy + from tests.test_hosted import _FakeScrubber + + monkeypatch.setenv(crypto.ENV_KEY, KEY) + monkeypatch.setenv("OPENADAPT_HOME", str(tmp_path / "home")) + monkeypatch.delenv(hosted.TOKEN_ENV, raising=False) + monkeypatch.setenv(hosted.DESTINATION_KIND_ENV, "customer-managed") + monkeypatch.setenv(hosted.TRUSTED_HOSTS_ENV, "https://h.test") + monkeypatch.setenv("OPENADAPT_SANITIZATION_POLICY_KEY_ID", "test-policy-key") + monkeypatch.setenv( + "OPENADAPT_SANITIZATION_POLICY_KEY", + base64.b64encode(b"test-policy-secret-material-32-bytes!!").decode(), + ) + monkeypatch.setattr(hosted, "_keyring_token", lambda host: None) + privacy.set_text_scrubber(_FakeScrubber()) + yield + privacy.reset_scrubbers() + + +def _sealed_phi_recording(root: Path) -> Path: + with sealed_set.writing(root, "recording", KEY): + sealed_set.write_text( + root / "events.jsonl", + json.dumps({"kind": "type", "text": "Jane Doe 12345"}) + "\n", + ) + sealed_set.write_text(root / "meta.json", json.dumps({"task": "Jane Doe"})) + return root + + +def _uploaded(monkeypatch: pytest.MonkeyPatch) -> dict[str, bytes]: + import httpx + + from tests.test_hosted import _accepted_recording_post + + sent: dict[str, bytes] = {} + accept = _accepted_recording_post() + + def post(url, **kw): + sent["body"] = kw["files"]["file"][1].read() + return accept(url, **kw) + + monkeypatch.setattr(httpx, "post", post) + return sent + + +def _no_plaintext_under(root: Path) -> None: + for path in root.rglob("*"): + if path.is_file(): + data = path.read_bytes() + assert b"Jane Doe" not in data and b"" not in data, path + + +def test_push_of_a_sealed_recording_keeps_the_review_copy_sealed( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, push_env, capsys +) -> None: + import io + import zipfile + + from openadapt_flow import hosted + from openadapt_flow.sanitized_artifact import ( + add_manual_text_redaction, + approved_archive_path, + render_review_html, + ) + + for name in ("push", "review_sanitized", "approve_sanitized"): + _ready(f"openadapt_flow.__main__:_cmd_{name}") + monkeypatch.setenv(hosted.AUTO_APPROVE_ENV, "false") + rec = _sealed_phi_recording(tmp_path / "rec") + before = _files(rec) + push = ["push", str(rec), "--host", "https://h.test", "--token", "tok"] + assert main(push) == 0 + out = capsys.readouterr().out + derivative = Path(out.split("Sanitized derivative: ")[1].splitlines()[0]) + assert sealed_set.header_kind(derivative) == "sanitized" + _assert_ciphertext_only(derivative) + # Review decrypts in memory; a reviewer's redaction stays encrypted. + page = render_review_html(rec, derivative, "csrf") + assert "Jane Doe" in page and "<PERSON>" in page + add_manual_text_redaction(derivative, "meta.json", "task", "field") + _assert_ciphertext_only(derivative) + approve = ["approve-sanitized", str(derivative), "--original", str(rec)] + assert main([*approve, "--reviewer", "reviewer@example.test"]) == 0 + assert crypto.is_encrypted(approved_archive_path(derivative).read_bytes()) + sent = _uploaded(monkeypatch) + upload = ["push", str(derivative), "--host", "https://h.test", "--token", "tok"] + assert main(upload) == 0 + with zipfile.ZipFile(io.BytesIO(sent["body"])) as zf: + uploaded = {name: zf.read(name) for name in zf.namelist()} + assert b"" in uploaded["events.jsonl"] + assert all(b"Jane Doe" not in data for data in uploaded.values()) + assert b"field" in uploaded["meta.json"] + assert _files(rec) == before + _no_plaintext_under(tmp_path) + + +def test_push_in_strict_mode_seals_the_review_copy_of_a_plain_recording( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, push_env +) -> None: + from openadapt_flow import hosted + + monkeypatch.setenv(hosted.AUTO_APPROVE_ENV, "true") + monkeypatch.setenv("OPENADAPT_FLOW_AT_REST", "strict") + rec = tmp_path / "rec" + rec.mkdir() + (rec / "events.jsonl").write_text(json.dumps({"text": "Jane Doe"}) + "\n") + (rec / "meta.json").write_text("{}") + sent = _uploaded(monkeypatch) + out = tmp_path / "derivative" + push = ["push", str(rec), "--host", "https://h.test", "--token", "tok"] + assert main([*push, "--sanitized-out", str(out)]) == 0 + assert sealed_set.header_kind(out) == "sanitized" + _assert_ciphertext_only(out) + import io + import zipfile + + with zipfile.ZipFile(io.BytesIO(sent["body"])) as zf: + assert b"" in zf.read("events.jsonl") + (rec / "events.jsonl").unlink() + _no_plaintext_under(tmp_path) + + +def _bundle_with_a_bad_action(source: Path, out: Path, *, key=None) -> Path: + """A copy of ``source`` whose first action is the private marker.""" + opened = sealed_set.open_set(source, KEY) + files = {rel: opened.read_bytes(rel) for rel in opened.files("")} + doc = json.loads(files["workflow.json"]) + doc["steps"][0]["action"] = _MARKER + files["workflow.json"] = json.dumps(doc).encode() + if key is None: + for rel, data in files.items(): + (out / rel).parent.mkdir(parents=True, exist_ok=True) + (out / rel).write_bytes(data) + return out + with sealed_set.writing(out, "bundle", key): + for rel, data in files.items(): + sealed_set.write_bytes(out / rel, data) + return out + + +@pytest.mark.parametrize("strict", [True, False], ids=["strict", "sealed input"]) +def test_push_refusal_on_a_sealed_bundle_never_prints_its_contents( + sealed: Path, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + push_env, + capfd, + caplog, + strict: bool, +) -> None: + from openadapt_flow import hosted + + monkeypatch.setenv(hosted.AUTO_APPROVE_ENV, "false") + if strict: + monkeypatch.setenv("OPENADAPT_FLOW_AT_REST", "strict") + bad = _bundle_with_a_bad_action(sealed, tmp_path / "bad", key=KEY) + assert sealed_set.is_sealed(bad) + push = ["push", str(bad), "--kind", "bundle", "--host", "https://h.test"] + capfd.readouterr() + caplog.set_level("DEBUG") + assert main([*push, "--token", "tok"]) == 1 + captured = capfd.readouterr() + _marker_absent(tmp_path, captured.out, captured.err, caplog.text) + assert "push failed: the input was refused (ValidationError)" in captured.out + assert "withheld" in captured.out + + +def test_push_refusal_on_a_plain_bundle_keeps_its_details( + sealed: Path, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + push_env, + capsys, +) -> None: + """Outside strict mode a plaintext input's refusal says what is wrong.""" + from openadapt_flow import hosted + + monkeypatch.setenv(hosted.AUTO_APPROVE_ENV, "false") + bad = _bundle_with_a_bad_action(sealed, tmp_path / "bad") + assert not sealed_set.is_sealed(bad) + push = ["push", str(bad), "--kind", "bundle", "--host", "https://h.test"] + assert main([*push, "--token", "tok"]) == 1 + out = capsys.readouterr().out + assert "Cannot validate generated workflow.py" in out + assert "type=enum" in out and _MARKER not in out # no quoted input value + + +def test_refusal_text_never_quotes_a_validation_input() -> None: + from pydantic import ValidationError + + from openadapt_flow import safe_errors + + try: + Step.model_validate({"id": "s", "intent": "x", "action": _MARKER}) + except ValidationError as exc: + error = exc + assert _MARKER in str(error) + wrapped = RuntimeError(f"Cannot validate: {error}") + for text in (safe_errors.message(error), safe_errors.message(wrapped)): + assert _MARKER not in text and "input_type=str" in text + assert "input_value withheld" in safe_errors.message(f"input_value={_MARKER}") + + +def test_an_unexpected_failure_on_sealed_input_prints_no_traceback( + sealed: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capfd, caplog +) -> None: + """An error no handler expects would print a traceback quoting what the + command read; with sealed input the CLI names its kind only.""" + import openadapt_flow.__main__ as cli + + def _cmd_visualize(args) -> int: + raise RuntimeError(f"read {_MARKER}") + + _cmd_visualize.__module__ = "openadapt_flow.__main__" + monkeypatch.setattr(cli, "_cmd_visualize", _cmd_visualize) + capfd.readouterr() + caplog.set_level("DEBUG") + out = tmp_path / "graph.json" + assert main(["visualize", str(sealed), "--format", "json", "--out", str(out)]) == 1 + printed = capfd.readouterr() + _marker_absent(tmp_path, printed.out, printed.err, caplog.text) + assert "FAILED: the input was refused (RuntimeError)" in printed.err + # Plain input still raises, so its traceback stays available. + with pytest.raises(RuntimeError, match=_MARKER): + main(["visualize", str(_plain_bundle(tmp_path / "p")), "--out", str(out)]) + + +def test_a_sealed_library_reads_back_with_its_own_key( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from openadapt_flow.ir import lift_to_program + from openadapt_flow.learning.library import SkillLibrary + + monkeypatch.setenv(crypto.ENV_KEY, KEY) + own_key = "library-only-test-passphrase-7d2e" + root = tmp_path / "own.skills" + graph = lift_to_program(Workflow.load(_plain_bundle(tmp_path))) + SkillLibrary(root, seal=True, key=own_key).create_skill("own", graph) + _assert_ciphertext_only(root) + assert SkillLibrary(root, key=own_key).skill_ids() == ["own"] + with pytest.raises((sealed_set.SealedSetError, crypto.CryptoError)): + SkillLibrary(root) # the global key does not open it + + +def test_a_plaintext_copy_of_a_sealed_bundle_never_loads_as_encrypted( + sealed: Path, tmp_path: Path +) -> None: + """A decrypted copy (an upload, an export) keeps encrypted: true in its + JSON, but the flag holds only with an encrypted envelope on disk.""" + opened = sealed_set.open_set(sealed, KEY) + copy = tmp_path / "plain-copy" + for rel in opened.files(""): + target = copy / rel + target.parent.mkdir(parents=True, exist_ok=True) + target.write_bytes(opened.read_bytes(rel)) + assert json.loads((copy / "workflow.json").read_text())["encrypted"] is True + loaded = Workflow.load(copy) + assert loaded.encrypted is False + assert loaded.manifest is not None and loaded.manifest.encrypted is False + + +def _named_teach_inputs( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> tuple[Path, Path, Path]: + """Plain teach inputs whose workflow name (the default skill id) is the + private marker.""" + import tests.test_teach_cli as cli + + original = cli._base_workflow + + def named() -> Workflow: + workflow = original() + workflow.name = _MARKER + return workflow + + monkeypatch.setattr(cli, "_base_workflow", named) + base = cli._base_bundle(tmp_path / "in") + cli._halt_run(base, tmp_path / "in" / "run") + fix = cli._dismiss_recording(tmp_path / "in" / "fix") + assert Workflow.load(base).name == _MARKER + return tmp_path / "in" / "run", fix, base + + +def test_strict_teach_success_never_prints_a_name_from_its_input( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capfd, caplog +) -> None: + monkeypatch.setenv(crypto.ENV_KEY, KEY) + run, fix, base = _named_teach_inputs(tmp_path, monkeypatch) + monkeypatch.setenv("OPENADAPT_FLOW_AT_REST", "strict") + out = tmp_path / "taught" + args = ["teach", str(run), "--fix", str(fix), "--bundle", str(base)] + capfd.readouterr() + caplog.set_level("DEBUG") + assert main([*args, "--out", str(out)]) == 0 + learned = capfd.readouterr() + # A second teach reads the library back and refuses with no_change. + assert main([*args, "--out", str(out)]) == 1 + again = capfd.readouterr() + for text in (learned.out, learned.err, again.out, again.err, caplog.text): + assert _MARKER not in text + assert "LEARNED" in learned.out and "no_change" in again.out + assert "(skill withheld, sha256 " in learned.out + for root in (out, out.parent / f"{out.name}.skills", fix / "_teach_compiled"): + _marker_absent(root) + + +def test_teach_success_on_plain_input_names_its_skill( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys +) -> None: + """The control: outside strict mode a plaintext run's summary names it.""" + run, fix, base = _named_teach_inputs(tmp_path, monkeypatch) + out = tmp_path / "taught" + capsys.readouterr() + args = ["teach", str(run), "--fix", str(fix), "--bundle", str(base)] + assert main([*args, "--out", str(out)]) == 0 + assert f"(skill {_MARKER!r})" in capsys.readouterr().out + + +@pytest.mark.parametrize("strict", [True, False], ids=["strict", "plain"]) +def test_push_success_names_the_workflow_only_for_plain_input( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capfd, caplog, strict: bool +) -> None: + from openadapt_flow import hosted + + if strict: + monkeypatch.setenv("OPENADAPT_FLOW_AT_REST", "strict") + pushed = { + "workflow_id": "wf_1", + "workflow_name": _MARKER, + "kind": "bundle", + "status": "active", + } + monkeypatch.setattr(hosted, "push", lambda *a, **kw: pushed) + bundle = tmp_path / "bundle" + bundle.mkdir() + capfd.readouterr() + caplog.set_level("DEBUG") + push = ["push", str(bundle), "--kind", "bundle", "--host", "https://h.test"] + assert main([*push, "--token", "tok"]) == 0 + captured = capfd.readouterr() + assert "Pushed. workflow_id=wf_1" in captured.out + if strict: + for text in (captured.out, captured.err, caplog.text): + assert _MARKER not in text + assert "name=withheld, sha256 " in captured.out + else: + assert f"name={_MARKER!r}" in captured.out + + +def test_strict_library_mismatch_refusal_never_prints_the_skill_id( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capfd, caplog +) -> None: + """A skill library that already holds another program for the skill is + refused without naming the skill id teach read from the input.""" + monkeypatch.setenv(crypto.ENV_KEY, KEY) + run, fix, base = _named_teach_inputs(tmp_path, monkeypatch) + from openadapt_flow.learning.library import SkillLibrary + + monkeypatch.setenv("OPENADAPT_FLOW_AT_REST", "strict") + args = ["teach", str(run), "--fix", str(fix), "--bundle", str(base)] + first = tmp_path / "first" + assert main([*args, "--out", str(first)]) == 0 + # A sealed library that holds the taught program, not the base one, for + # the same skill id. + library = tmp_path / "library" + taught = Workflow.load(first, key=KEY) + SkillLibrary(library, seal=True).create_skill(_MARKER, taught.program) + capfd.readouterr() + caplog.set_level("DEBUG") + second = tmp_path / "second" + assert main([*args, "--library", str(library), "--out", str(second)]) == 2 + printed = capfd.readouterr() + for text in (printed.out, printed.err, caplog.text): + assert _MARKER not in text + assert "teach cannot run: the input was refused (TeachError)" in printed.out + assert not second.exists() + + +@pytest.mark.parametrize("public", [False, True], ids=["read", "command line"]) +def test_strict_teach_refusal_prints_only_command_line_details( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capfd, caplog, public: bool +) -> None: + """Strict teach prints a refusal's text only when the error says it was + built from the command line. (teach-branch is refused in strict mode and + for sealed input before it runs.)""" + from openadapt_flow.learning import teach + from openadapt_flow.learning.teach import TeachError + + def refuse(*_args, **_kwargs): + raise TeachError(f"refused {_MARKER!r}", public=public) + + monkeypatch.setattr(teach, "teach", refuse) + monkeypatch.setenv("OPENADAPT_FLOW_AT_REST", "strict") + for name in ("run", "fix", "bundle"): + (tmp_path / name).mkdir() + run, fix, bundle = (str(tmp_path / n) for n in ("run", "fix", "bundle")) + capfd.readouterr() + caplog.set_level("DEBUG") + out = str(tmp_path / "out") + args = ["teach", run, "--fix", fix, "--bundle", bundle, "--out", out] + assert main(args) == 2 + printed = capfd.readouterr() + assert "teach cannot run" in printed.out + if public: + assert f"refused {_MARKER!r}" in printed.out + else: + for text in (printed.out, printed.err, caplog.text): + assert _MARKER not in text + assert "the input was refused (TeachError)" in printed.out + + +@pytest.mark.parametrize("strict", [True, False], ids=["strict", "plain"]) +def test_push_failure_of_its_own_is_withheld_only_for_private_input( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capfd, caplog, strict: bool +) -> None: + """A push error with no cause can still name a path inside the artifact, + so strict mode withholds it too.""" + from openadapt_flow import hosted + + if strict: + monkeypatch.setenv("OPENADAPT_FLOW_AT_REST", "strict") + + def fail(*_args, **_kwargs): + raise hosted.HostedError(f"cannot read {_MARKER}/steps.json") + + monkeypatch.setattr(hosted, "push", fail) + bundle = tmp_path / "bundle" + bundle.mkdir() + capfd.readouterr() + caplog.set_level("DEBUG") + push = ["push", str(bundle), "--kind", "bundle", "--host", "https://h.test"] + assert main([*push, "--token", "tok"]) == 1 + captured = capfd.readouterr() + if strict: + for text in (captured.out, captured.err, caplog.text): + assert _MARKER not in text + assert "push failed: the input was refused (HostedError)" in captured.out + else: + assert f"push failed: cannot read {_MARKER}/steps.json" in captured.out + + +def test_teach_branch_with_a_sealed_example_is_refused_before_it_runs( + sealed: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capfd, caplog +) -> None: + """--example RUN FIX arrives as a list of pairs; a sealed RUN inside it + is still found, so teach-branch is refused before it reads anything.""" + from openadapt_flow.learning import branch_teach + + calls: list[object] = [] + + def refuse(*args, **kwargs): + calls.append(args) + raise branch_teach.TeachError(f"workflow {_MARKER!r} did not stop here") + + monkeypatch.setattr(branch_teach, "teach_branch", refuse) + base = _plain_bundle(tmp_path / "base") + fix = tmp_path / "fix" + fix.mkdir() + capfd.readouterr() + caplog.set_level("DEBUG") + examples = ["--example", str(sealed), str(fix)] * 2 + out = str(tmp_path / "out") + args = ["teach-branch", *examples, "--bundle", str(base), "--out", out] + assert main(args) == 2 + printed = capfd.readouterr() + for text in (printed.out, printed.err, caplog.text): + assert _MARKER not in text + assert "REFUSED" in printed.err and "does not read encrypted input" in printed.err + assert not calls + + +def test_sealed_paths_are_found_at_any_depth_and_unknown_values_are_private( + sealed: Path, tmp_path: Path +) -> None: + import argparse + + from openadapt_flow.__main__ import _private_input, _sealed_paths_in + + plain = str(tmp_path / "plain") + nested = argparse.Namespace( + example=[[plain, plain], (plain, {"fix": [str(sealed)]})], out=plain + ) + assert _sealed_paths_in(nested) == [sealed] + assert _private_input(nested) + assert not _private_input(argparse.Namespace(example=[[plain, plain]])) + # A value that is neither a path, a scalar nor a container: fail closed. + assert _private_input(argparse.Namespace(example=[[plain, object()]])) + # Objects the code sets on the namespace do not make the run private. + assert not _private_input(argparse.Namespace(_handler=object(), out=plain)) + + +def _odd_spelling(sealed: Path, tmp_path: Path, spelling: str) -> str: + """Move ``sealed`` under a directory whose path is spelled oddly.""" + import shutil + + if spelling == "newline": + parent = tmp_path / "line\nbreak" + parent.mkdir() + shutil.move(str(sealed), str(parent / "sealed")) + return str(parent / "sealed") + parent = tmp_path / "url:" + parent.mkdir() + shutil.move(str(sealed), str(parent / "sealed")) + # "url://sealed" is the same local directory with a redundant slash. + return f"{tmp_path}/url://sealed" + + +@pytest.mark.parametrize("spelling", ["newline", "url-slash"]) +def test_a_sealed_example_under_any_path_spelling_is_refused( + sealed: Path, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capfd, + caplog, + spelling: str, +) -> None: + """A newline or "://" in a path is still a local path: the sealed example + is found and teach-branch is refused before it reads anything.""" + from openadapt_flow.learning import branch_teach + + calls: list[object] = [] + + def refuse(*args, **kwargs): + calls.append(args) + raise branch_teach.TeachError(f"workflow {_MARKER!r} did not stop here") + + monkeypatch.setattr(branch_teach, "teach_branch", refuse) + example = _odd_spelling(sealed, tmp_path, spelling) + base = _plain_bundle(tmp_path / "base") + fix = tmp_path / "fix" + fix.mkdir() + capfd.readouterr() + caplog.set_level("DEBUG") + out = tmp_path / "out" + args = ["teach-branch", "--example", example, str(fix), "--bundle", str(base)] + assert main([*args, "--out", str(out)]) == 2 + printed = capfd.readouterr() + for text in (printed.out, printed.err, caplog.text): + assert _MARKER not in text + assert "REFUSED" in printed.err and "does not read encrypted input" in printed.err + assert not calls + assert not out.exists() + + +def test_a_path_spelled_with_a_newline_or_url_slash_is_private( + sealed: Path, tmp_path: Path +) -> None: + import argparse + + from openadapt_flow.__main__ import _private_input, _sealed_paths_in + + example = _odd_spelling(sealed, tmp_path, "url-slash") + assert _sealed_paths_in(argparse.Namespace(example=[[example, "x"]])) + # Text that is no local path stays public, whatever it looks like. + public = argparse.Namespace( + url="https://example.invalid/a", note="two\nlines", long="x" * 5000 + ) + assert not _private_input(public) + + +@pytest.mark.skipif( + not hasattr(os, "geteuid") or os.geteuid() == 0, reason="root reads any dir" +) +def test_a_path_whose_header_cannot_be_checked_is_private(tmp_path: Path) -> None: + import argparse + + from openadapt_flow.__main__ import _private_input + + locked = tmp_path / "locked" + (locked / "inner").mkdir(parents=True) + locked.chmod(0) + try: + assert _private_input(argparse.Namespace(example=[str(locked / "inner")])) + finally: + locked.chmod(0o700) + + +@pytest.mark.parametrize( + ("winerror", "private"), [(123, False), (161, False), (None, True), (5, True)] +) +def test_a_name_windows_cannot_hold_is_public( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, winerror, private: bool +) -> None: + """A URL argument on Windows (a colon in a name) cannot name a local + path, so it is public; any other error checking the header is private.""" + import argparse + import errno + + from openadapt_flow.__main__ import _private_input + + def stat(path, *args, **kwargs): + exc = OSError(errno.EINVAL, "The filename syntax is incorrect") + if winerror is not None: + exc.winerror = winerror # type: ignore[attr-defined] + raise exc + + monkeypatch.setattr(os, "stat", stat) + args = argparse.Namespace(effects_base_url="http://127.0.0.1:8080") + assert _private_input(args) is private diff --git a/packages/flow/tests/test_sealed_at_rest.py b/packages/flow/tests/test_sealed_at_rest.py index 138d716a28..9c1364a852 100644 --- a/packages/flow/tests/test_sealed_at_rest.py +++ b/packages/flow/tests/test_sealed_at_rest.py @@ -751,5 +751,10 @@ def test_sealed_run_pauses_and_resumes_without_plaintext( def test_phi_at_rest_doc_lists_every_inventory_entry() -> None: doc = (Path(__file__).parents[1] / "docs" / "phi_at_rest.md").read_text() - for entry in (*sealed_set.BUNDLE_FILES, *sealed_set.RECORDING_FILES): + for entry in ( + *sealed_set.BUNDLE_FILES, + *sealed_set.RECORDING_FILES, + *sealed_set.LIBRARY_FILES, + *sealed_set.SANITIZED_FILES, + ): assert f"`{entry.pattern}`" in doc, entry.pattern