Check recorded run conditions; limit run output; hash inputs once

docs/archive.md

@@ -71,7 +71,7 @@ Records are serialized as canonical JSON: sorted keys, compact separators, UTF-8
 {"locator":"/corpus/doc.pdf","schema":"stele.source","schema_version":1}
 ```
 
-A Snapshot record contains only fields derived from the content. It is published only after every blob it covers has been re-hashed (the file, or the tree object and each file it lists), so its presence means the Snapshot was complete and intact in the store when it was recorded. A stored blob corrupted to other bytes of the same length is refused, not certified.
+A Snapshot record contains only fields derived from the content. It is published only after every blob it covers has been re-hashed (the file, or the tree object and each file it lists), so its presence means the Snapshot was complete and intact in the store when it was recorded. A stored blob corrupted to other bytes of the same length is refused, not certified. Ingesting a staged input verifies each blob as it is stored (new bytes are hashed while written, and a copy already stored is re-hashed), so `snapshot_staged_input` publishes the record without hashing the input a second time.
 
 A Source's `source_id` is the SHA-256 of its canonical record. Linking a Source to its Snapshots is a per-run fact: each ledger record stores its run's Snapshot digest (`source_hash`) and `source_id` (see [ledger.md](ledger.md)).
 

docs/containment.md

@@ -125,6 +125,14 @@ as the entrypoint of the image, so it must resolve inside the image (e.g.
   If the Stele process is killed, a running Docker container is not stopped
   (Podman gets `--timeout` as a backstop).
 
+**What a run leaves in `/stele/output` is limited too.** When the output is
+collected, before anything is stored in the archive, a run whose regular
+files total more than `SandboxConfig.max_output_bytes` (default 4 GiB) or
+number more than `max_output_files` (default 100,000) fails with
+`FailureReason.OUTPUT_LIMIT`, and its output is removed. While the parser
+runs, its writes are bounded only by the host filesystem that holds
+`artifact_dir`; put that directory on a volume sized for the parsers you run.
+
 **Parser output is captured within a fixed size.** Whatever a parser prints
 is buffered in the Stele process, outside every limit the sandbox puts on the
 parser (a container's memory limit does not cover the engine CLI relaying its

docs/ledger.md

@@ -70,7 +70,7 @@ A second record for the same `run_id` raises `DuplicateRunError`.
 | `parser` | `ParserIdentity`: `name`, `version`, and the executable's digest: `image_digest` (OCI backend) or `module_sha256` (Wasm backends). The digests are measured by the run, not asserted by the caller. |
 | `parser_config` | The configuration passed to the parser, stored as canonical JSON |
 | `backend` | Sandbox backend that ran the parser |
-| `run_conditions` | `RunConditions`: the `device` (`cpu` or `gpu`) and the `memory`, `cpus`, `pids_limit` and `timeout_seconds` the run executed under. Recorded by Stele's runner (`record_parser_run`), so a replay runs the parser the same way (#30). `None` for runs recorded without them and for records made before schema version 5. |
+| `run_conditions` | `RunConditions`: the `device` (`cpu` or `gpu`) and the `memory`, `cpus`, `pids_limit` and `timeout_seconds` the run executed under. Recorded by Stele's runner (`record_parser_run`), so a replay runs the parser the same way (#30). Every stated condition must match the limits the backend reported applying (`SandboxResult.telemetry.limits`); a mismatch, or a condition the backend does not report, raises `ProvenanceError` and nothing is recorded. `None` for runs recorded without them and for records made before schema version 5. |
 | `artifact_dir` | Host path of the run's output directory (a location, not evidence) |
 | `artifact_manifest` | `{relative POSIX path: sha256}` for every artifact |
 | `artifact_hash` | Digest of the bundle's tree object in the archive (`sha256_manifest(manifest)`) |

stele/archive/ingest.py

@@ -48,7 +48,10 @@ def snapshot_staged_input(store: BlobStore, staged: StagedInput) -> Snapshot:
         raise IntegrityError(
             f"snapshot digest {snapshot.digest} differs from staged hash {staged.sha256}"
         )
-    return store.put_snapshot(snapshot)
+    # Every blob above was stored with expected_digest (and the tree object
+    # was just written from its bytes), so the content is already verified:
+    # publish the record without hashing the whole input a second time.
+    return store._record_snapshot(snapshot)
 
 
 def ingest_artifacts(

stele/archive/store.py

@@ -408,6 +408,16 @@ def put_snapshot(self, snapshot: Snapshot) -> Snapshot:
         in the store when it was recorded. Recording it again is a no-op.
         """
         self._check_snapshot_content(snapshot)
+        return self._record_snapshot(snapshot)
+
+    def _record_snapshot(self, snapshot: Snapshot) -> Snapshot:
+        """Publish a Snapshot record whose content the caller has just verified.
+
+        For stele.archive.ingest only: it stores every blob with
+        expected_digest (which hashes new bytes as they are written and
+        re-verifies a copy already stored), so hashing them all again in
+        put_snapshot would only double the cost of ingesting an input.
+        """
         final = self._sharded(self._snapshots / snapshot.kind.value, snapshot.digest)
         self._install_record(final, snapshot.to_canonical())
         return snapshot

stele/containment/artifacts.py

@@ -11,6 +11,10 @@ class UnsafeArtifactError(ValueError):
     """Raised when parser output contains a symlink or non-regular entry."""
 
 
+class OutputLimitError(ValueError):
+    """Raised when parser output exceeds the run's size or file-count limit."""
+
+
 def _lstat(path: Path) -> os.stat_result:
     # A directory that is listable but not searchable (e.g. chmod 0444) lets
     # os.walk see names whose lstat then fails; treat that like unreadable.
@@ -23,12 +27,19 @@ def _lstat(path: Path) -> os.stat_result:
         ) from exc
 
 
-def collect_artifact_paths(artifact_dir: Path) -> list[Path]:
+def collect_artifact_paths(
+    artifact_dir: Path,
+    *,
+    max_bytes: int | None = None,
+    max_files: int | None = None,
+) -> list[Path]:
     """Return regular files under artifact_dir without following symlinks.
 
     The parser controls everything created below artifact_dir, so every entry is
     checked with lstat after the sandbox exits. Symlinks, devices, FIFOs,
-    sockets, and any non-directory/non-regular entry fail closed.
+    sockets, and any non-directory/non-regular entry fail closed. Output over
+    max_bytes in total or max_files files raises OutputLimitError, before
+    any of it is read.
     """
     artifact_dir = Path(artifact_dir)
 
@@ -49,6 +60,7 @@ def _unreadable(exc: OSError) -> None:
         ) from exc
 
     artifacts: list[Path] = []
+    total = 0
     for root, dirnames, filenames in os.walk(
         artifact_dir, onerror=_unreadable, followlinks=False
     ):
@@ -70,6 +82,11 @@ def _unreadable(exc: OSError) -> None:
                     f"unsafe artifact file entry (symlink/non-regular): {path}"
                 )
             artifacts.append(path)
+            total += entry.st_size
+            if max_files is not None and len(artifacts) > max_files:
+                raise OutputLimitError(f"the parser wrote more than {max_files} files")
+            if max_bytes is not None and total > max_bytes:
+                raise OutputLimitError(f"the parser wrote more than {max_bytes} bytes of output")
 
     return sorted(artifacts, key=lambda path: str(path.relative_to(artifact_dir)))
 

stele/containment/runner.py

@@ -26,7 +26,12 @@
 from typing import TYPE_CHECKING
 from uuid import uuid4
 
-from .artifacts import UnsafeArtifactError, collect_artifact_paths, discard_artifact_dir_contents
+from .artifacts import (
+    OutputLimitError,
+    UnsafeArtifactError,
+    collect_artifact_paths,
+    discard_artifact_dir_contents,
+)
 from .backend import (
     BWRAP_MISSING as _BWRAP_MISSING,
     ExecutionOutcome,
@@ -156,9 +161,15 @@ def run_in_sandbox(
     artifact_paths: list[Path] = []
     if failure is None:
         try:
-            artifact_paths = collect_artifact_paths(config.artifact_dir)
+            artifact_paths = collect_artifact_paths(
+                config.artifact_dir,
+                max_bytes=config.max_output_bytes,
+                max_files=config.max_output_files,
+            )
         except UnsafeArtifactError as exc:
             failure = RunFailure(FailureReason.UNSAFE_ARTIFACT, str(exc), exit_code=outcome.exit_code)
+        except OutputLimitError as exc:
+            failure = RunFailure(FailureReason.OUTPUT_LIMIT, str(exc), exit_code=outcome.exit_code)
     if failure is not None:
         artifact_paths = []
         _remove_output(config.artifact_dir, created_artifact_dir)

stele/containment/sandbox.py

@@ -30,6 +30,11 @@
 _EPHEMERAL_TMPFS = ["/home", "/mnt", "/root", "/run", "/var"]
 
 
+# Defaults for SandboxConfig.max_output_bytes / max_output_files.
+DEFAULT_MAX_OUTPUT_BYTES = 4 * 1024 ** 3
+DEFAULT_MAX_OUTPUT_FILES = 100_000
+
+
 @dataclass
 class SandboxConfig:
     """Everything needed to construct and run one sandboxed parser invocation."""
@@ -71,6 +76,14 @@ class SandboxConfig:
     # run. Set False to allow writes only under /stele/output.
     writable_scratch: bool = True
 
+    # Limits on what the run may leave in /stele/output. Checked when the
+    # output is collected, before anything is stored in the archive; a run
+    # over either limit fails with FailureReason.OUTPUT_LIMIT and its output
+    # is removed. (While the parser runs, its writes are bounded only by the
+    # host filesystem holding artifact_dir.)
+    max_output_bytes: int = DEFAULT_MAX_OUTPUT_BYTES
+    max_output_files: int = DEFAULT_MAX_OUTPUT_FILES
+
 
 @dataclass(frozen=True)
 class LandlockLaunch:

stele/containment/telemetry.py

@@ -32,6 +32,7 @@ class FailureReason(str, enum.Enum):
     ENGINE_ERROR = "engine_error"        # the sandbox could not start the parser (container
                                          # engine errors 125-127, Wasm load/link failure)
     UNSAFE_ARTIFACT = "unsafe_artifact"  # output held a symlink, FIFO, device or unreadable dir
+    OUTPUT_LIMIT = "output_limit"        # output exceeded max_output_bytes / max_output_files
 
 
 @dataclass(frozen=True)

stele/ledger/models.py

@@ -70,8 +70,10 @@ class RunConditions:
     """How a run was executed: the device and the limits it ran under.
 
     Recorded by Stele's own runner (stele.parsers.replay.record_parser_run
-    takes it from the ParserRun, which applied it), never asserted by a
-    caller, so a replay can run the parser the same way (roadmap #30).
+    takes it from the ParserRun, which applied it) so a replay can run the
+    parser the same way (roadmap #30). The ledger checks every stated
+    condition against the limits the backend reported applying, so a caller
+    cannot record conditions the run did not execute under.
     """
 
     device: str = "cpu"                 # "cpu" or "gpu": which image variant ran

stele/ledger/transaction.py

@@ -85,6 +85,61 @@ def _check_input(sandbox_result: SandboxResult) -> None:
         )
 
 
+# RunConditions field → the key under which a backend reports the limit it
+# applied (ExecutionOutcome.limits, carried as SandboxResult.telemetry.limits).
+_APPLIED_LIMITS = {
+    "memory": "memory",
+    "cpus": "cpus",
+    "pids_limit": "pids",
+    "timeout_seconds": "timeout_seconds",
+}
+
+
+def _check_run_conditions(
+    sandbox_result: SandboxResult, run_conditions: RunConditions | None
+) -> None:
+    """Refuse run conditions the backend does not confirm it applied.
+
+    A replay runs the parser under the recorded conditions, so they must be
+    what the run executed under, not what a caller says it did. Every stated
+    condition is compared with the limits the backend reported; one it did
+    not report cannot be vouched for and is refused too.
+    """
+    if run_conditions is None:
+        return
+    telemetry = sandbox_result.telemetry
+    applied = dict(telemetry.limits) if telemetry is not None else {}
+    problems: list[str] = []
+
+    if "gpu" in applied:
+        ran_on = "gpu" if applied["gpu"] else "cpu"
+        if ran_on != run_conditions.device:
+            problems.append(f"device {run_conditions.device!r}, but the run used the {ran_on}")
+    elif run_conditions.device == "gpu":
+        problems.append("device 'gpu', but the backend reports no GPU")
+
+    for field_name, key in _APPLIED_LIMITS.items():
+        stated = getattr(run_conditions, field_name)
+        if stated is None:
+            continue
+        if key not in applied:
+            problems.append(f"{field_name}={stated!r}, which the backend does not report applying")
+        elif not _same_limit(stated, applied[key]):
+            problems.append(f"{field_name}={stated!r}, but the backend applied {applied[key]!r}")
+
+    if problems:
+        raise ProvenanceError(
+            f"run_id={sandbox_result.run_id}: the stated run conditions are not the ones "
+            "the run executed under: " + "; ".join(problems)
+        )
+
+
+def _same_limit(stated: Any, applied: Any) -> bool:
+    if isinstance(stated, float) and isinstance(applied, (int, float)) and not isinstance(applied, bool):
+        return float(applied) == stated
+    return type(stated) is type(applied) and stated == applied
+
+
 @contextmanager
 def ledger_transaction(
     store: LedgerStore,
@@ -111,7 +166,10 @@ def ledger_transaction(
 
     run_conditions is the device and limits Stele's runner applied to the
     run (stele.parsers.replay.record_parser_run passes them from the
-    ParserRun); a replay uses them to run the parser the same way.
+    ParserRun); a replay uses them to run the parser the same way. Each one
+    must match the limits the backend reported applying
+    (sandbox_result.telemetry.limits), or ProvenanceError is raised and
+    nothing is recorded.
 
     Yields the ArtifactRecord in PENDING state.
     """
@@ -130,6 +188,7 @@ def ledger_transaction(
         )
 
     _check_input(sandbox_result)
+    _check_run_conditions(sandbox_result, run_conditions)
     identity = measured_parser_identity(parser, sandbox_result)
 
     record = store.create_pending(

tests/test_archive.py

@@ -569,6 +569,33 @@ def test_empty_file_and_empty_directory_do_not_collide(
         assert store.get_snapshot(f.digest, "file").kind is SnapshotKind.FILE
         assert store.get_snapshot(d.digest, "tree").kind is SnapshotKind.TREE
 
+    def test_ingest_hashes_the_input_once(
+        self, store: BlobStore, tmp_path: Path, monkeypatch
+    ) -> None:
+        # Storing each blob with expected_digest already verifies it, so the
+        # snapshot is published without put_snapshot's second full re-hash.
+        def second_pass(snapshot):
+            raise AssertionError("the input was hashed a second time")
+
+        monkeypatch.setattr(store, "_check_snapshot_content", second_pass)
+        src = tmp_path / "corpus"
+        _tree(src)
+        snap = snapshot_staged_input(store, stage_input(src, tmp_path / "stage"))
+        assert store.get_snapshot(snap.digest, SnapshotKind.TREE) == snap
+        with pytest.raises(AssertionError, match="second time"):
+            store.put_snapshot(snap)  # the public entry point still re-hashes
+
+    def test_ingest_still_refuses_a_corrupt_stored_copy(
+        self, store: BlobStore, tmp_path: Path
+    ) -> None:
+        src = tmp_path / "doc.txt"
+        src.write_bytes(b"AAAA")
+        digest = store.put_bytes(b"AAAA")
+        overwrite(blob_file(store, digest), b"BBBB")  # same size, other bytes
+        with pytest.raises(IntegrityError, match="corrupt"):
+            snapshot_staged_input(store, stage_input(src, tmp_path / "stage"))
+        assert not store.has_snapshot(digest, SnapshotKind.FILE)
+
     def test_snapshot_refuses_bytes_that_differ_from_staging(
         self, store: BlobStore, tmp_path: Path
     ) -> None:

tests/test_parsers.py

@@ -37,7 +37,8 @@
 )
 from stele.containment.oci import DEFAULT_OCI_IMAGE, OciBackend
 from stele.ledger.models import ParserIdentity as ParserIdentityRecord
-from stele.ledger.models import RunConditions
+from stele.ledger.models import ArtifactState, RunConditions
+from stele.ledger.store import ProvenanceError
 from stele.ledger.transaction import record_run
 from stele.parsers import (
     CONFIG_ENV,
@@ -118,6 +119,12 @@ def execute(self, config):
         return ExecutionOutcome(
             exit_code=self.exit_code, stdout="", stderr="boom: bad page 3\n",
             wall_time_seconds=0.1, timed_out=self.timed_out, image_digest=DIGEST,
+            # Reported like OciBackend's: the ledger checks recorded run
+            # conditions against them.
+            limits={
+                "timeout_seconds": config.timeout_seconds, "memory": self.memory,
+                "cpus": self.cpus, "pids": self.pids_limit, "gpu": self.gpu,
+            },
         )
 
 
@@ -474,6 +481,53 @@ def test_moved_block_or_changed_text_is_diverged(self, doc: Path, tmp_path: Path
 
     # -- run conditions (roadmap #30) ---------------------------------------
 
+    def _unrecorded_run(self, tmp_path: Path, doc: Path, backend: FakeBackend):
+        ledger = open_ledger(tmp_path / "ledger.db")
+        run = run_parser(self.POLICY_PARSER, doc, tmp_path / "out", store=ledger.archive,
+                         backend=backend)
+        assert run.succeeded, run.failure
+        return ledger, run
+
+    @pytest.mark.parametrize(
+        "stated, complaint",
+        [
+            (RunConditions(device="gpu"), "device 'gpu', but the run used the cpu"),
+            (RunConditions(memory="8g"), "memory='8g', but the backend applied '512m'"),
+            (RunConditions(cpus=16), "cpus=16.0, but the backend applied 2.5"),
+            (RunConditions(pids_limit=4096), "pids_limit=4096"),
+            (RunConditions(timeout_seconds=5), "timeout_seconds=5"),
+        ],
+        ids=["device", "memory", "cpus", "pids", "timeout"],
+    )
+    def test_stated_run_conditions_must_be_the_applied_ones(
+        self, doc: Path, tmp_path: Path, stated: RunConditions, complaint: str
+    ) -> None:
+        ledger, run = self._unrecorded_run(tmp_path, doc, FakeBackend(write=self._files(self.LAYOUT)))
+        with pytest.raises(ProvenanceError, match="not the ones the run executed under") as info:
+            record_run(ledger, run.result, parser=ParserIdentityRecord("fake", "1.0"),
+                       parser_config=dict(run.identity.config), run_conditions=stated)
+        assert complaint in str(info.value)
+        assert ledger.list_by_states(list(ArtifactState)) == []  # nothing recorded
+
+    def test_conditions_a_backend_does_not_report_are_refused(
+        self, doc: Path, tmp_path: Path
+    ) -> None:
+        class Silent(FakeBackend):
+            def execute(self, config):
+                outcome = super().execute(config)
+                return replace(outcome, limits={"timeout_seconds": config.timeout_seconds})
+
+        ledger, run = self._unrecorded_run(tmp_path, doc, Silent(write=self._files(self.LAYOUT)))
+        identity = ParserIdentityRecord("fake", "1.0")
+        with pytest.raises(ProvenanceError, match="does not report applying"):
+            record_run(ledger, run.result, parser=identity, parser_config={},
+                       run_conditions=RunConditions(memory="512m"))
+        # What it did report can be recorded.
+        timeout = run.identity.timeout_seconds
+        record = record_run(ledger, run.result, parser=identity, parser_config={},
+                            run_conditions=RunConditions(timeout_seconds=timeout))
+        assert record.run_conditions == RunConditions(timeout_seconds=timeout)
+
     GPU_PARSER = replace(POLICY_PARSER, gpu="optional", gpu_image="localhost/stele/fake-gpu:1.0")
 
     def _record_with(self, tmp_path: Path, doc: Path, parser: ParserImage, backend: FakeBackend):