Bound sealed execution result payloads

docs/sealed-public-execution.md

@@ -16,19 +16,21 @@ Trusted/private side responsibilities:
 2. determine whether the work may execute on a public runner;
 3. derive the least-sufficient public-safe capsule;
 4. generate a per-assignment age X25519 keypair and retain the private identity outside GitHub;
-5. dispatch the public workflow with only the opaque assignment ID, capsule, capsule digest, public recipient, and bounded timeout;
+5. dispatch the public workflow with only the opaque assignment ID, capsule, capsule digest, exact age X25519 public recipient, and bounded timeout;
 6. retrieve the encrypted Actions artifact, verify the receipt/ciphertext digest, decrypt privately, and reconcile useful evidence back to the originating authority;
 7. delete the public artifact/run when it no longer has diagnostic value.
 
 Public runner responsibilities:
 
 1. verify the capsule digest and strict archive bounds;
-2. execute only the capsule's top-level `run.sh` with no private credentials;
-3. capture task stdout/stderr into the private result bundle rather than Actions logs;
-4. collect files written beneath `SEALED_RESULT_DIR`;
-5. package and encrypt the result to the supplied age recipient;
-6. upload only `result.age` plus a minimal `receipt.json` as a seven-day Actions artifact;
-7. preserve task success/failure in the workflow verdict.
+2. validate the exact age X25519 public-recipient shape;
+3. execute only the capsule's top-level `run.sh` with no private credentials;
+4. capture task stdout/stderr into the private result bundle rather than Actions logs;
+5. collect files written beneath `SEALED_RESULT_DIR`;
+6. bound result material before packaging so one task cannot silently consume unbounded artifact storage;
+7. package and encrypt the result to the supplied age recipient;
+8. upload only `result.age` plus a minimal `receipt.json` as a seven-day Actions artifact;
+9. preserve task success/failure in the workflow verdict.
 
 ## Capsule contract
 
@@ -39,7 +41,8 @@ The workflow accepts a gzip-compressed tar archive encoded as base64. Current ha
 - unpacked content: at most 16 MiB;
 - no symlinks, hardlinks, devices, or path traversal;
 - a regular top-level `run.sh` is required;
-- task timeout is at most 7,200 seconds.
+- task timeout is at most 7,200 seconds;
+- recipient must be the exact Bech32 shape used by an age X25519 `age1...` recipient.
 
 At runtime the worker sets:
 
@@ -55,6 +58,18 @@ The public artifact contains only:
 - `result.age` — age-encrypted gzip tar containing execution metadata, captured stdout/stderr, and result files;
 - `receipt.json` — opaque assignment ID, completed/failed status, ciphertext SHA-256/size, public worker revision, and Actions run correlation.
 
+The plaintext result is bounded before sealing:
+
+- captured stdout: at most 1 MiB after deterministic head/tail truncation;
+- captured stderr: at most 1 MiB after deterministic head/tail truncation;
+- substantive result files: at most 252 regular files;
+- one substantive result file: at most 8 MiB;
+- substantive result files plus stored stdout/stderr: at most 16 MiB total.
+
+If substantive result files exceed a hard bound, the worker does **not** upload an arbitrary partial result. It removes those files, writes a small `result-budget-exceeded.json` diagnostic inside the sealed bundle, records the original task exit code separately, and marks the worker execution failed with the bounded result-budget exit status. Stream truncation by itself is diagnostic and does not change an otherwise-successful task verdict.
+
+These limits apply to result material only; they intentionally do not cap task scratch/work files because legitimate public workloads may temporarily acquire large models or datasets during execution. Runtime disk exhaustion remains a separate host/provider boundary and may justify a later control if real reps demonstrate the need.
+
 Retention is set to seven days. This is a pickup/recovery window, not a durable evidence store. Trusted-side reconciliation should normally retrieve and delete the artifact sooner after successful pickup/decryption.
 
 The artifact is transport, not durable project authority. Durable conclusions, accepted evidence, negative results, or follow-on decisions belong back in the originating private/project-native authority.

scripts/sealed_public_execution.py

@@ -26,11 +26,16 @@
 from pathlib import Path
 
 ASSIGNMENT_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{2,95}$")
-AGE_RECIPIENT_RE = re.compile(r"^age1[0-9a-z]+$")
+AGE_RECIPIENT_RE = re.compile(r"^age1[023456789acdefghjklmnpqrstuvwxyz]{58}$")
 MAX_CAPSULE_B64 = 60_000
 MAX_MEMBER_COUNT = 256
 MAX_UNPACKED_BYTES = 16 * 1024 * 1024
 MAX_TIMEOUT_SECONDS = 7_200
+MAX_CAPTURED_STREAM_BYTES = 1 * 1024 * 1024
+MAX_RESULT_FILE_BYTES = 8 * 1024 * 1024
+MAX_RESULT_FILES = 252
+MAX_RESULT_TOTAL_BYTES = 16 * 1024 * 1024
+RESULT_BUDGET_EXIT_CODE = 125
 
 
 class WorkerError(ValueError):
@@ -105,6 +110,97 @@ def safe_extract(capsule: Path, destination: Path) -> None:
         raise WorkerError("capsule must contain a regular top-level run.sh")
 
 
+def _truncate_stream(path: Path, max_bytes: int = MAX_CAPTURED_STREAM_BYTES) -> dict[str, object]:
+    """Bound captured stdout/stderr while retaining both ends for diagnostics."""
+    original_bytes = path.stat().st_size
+    if original_bytes <= max_bytes:
+        return {"original_bytes": original_bytes, "stored_bytes": original_bytes, "truncated": False}
+
+    marker = b"\n...[sealed-public-execution output truncated]...\n"
+    payload_budget = max(0, max_bytes - len(marker))
+    head_bytes = payload_budget // 2
+    tail_bytes = payload_budget - head_bytes
+    with path.open("rb") as fh:
+        head = fh.read(head_bytes)
+        if tail_bytes:
+            fh.seek(-tail_bytes, os.SEEK_END)
+            tail = fh.read(tail_bytes)
+        else:
+            tail = b""
+    path.write_bytes(head + marker + tail)
+    return {
+        "original_bytes": original_bytes,
+        "stored_bytes": path.stat().st_size,
+        "truncated": True,
+    }
+
+
+def enforce_result_budget(result: Path) -> dict[str, object]:
+    """Bound material that will be sealed/uploaded without constraining task scratch data.
+
+    Streams are truncated to a diagnostic head+tail. If substantive result files
+    exceed any hard bound, they are replaced by a small diagnostic rather than
+    uploading a partial result that could be mistaken for complete evidence.
+    """
+    stdout_info = _truncate_stream(result / "stdout.txt")
+    stderr_info = _truncate_stream(result / "stderr.txt")
+    files_root = result / "files"
+
+    violations: list[str] = []
+    file_count = 0
+    file_bytes = 0
+    largest_file_bytes = 0
+
+    for item in sorted(files_root.rglob("*")):
+        if item.is_symlink():
+            violations.append("result files contain a symbolic link")
+            continue
+        if item.is_dir():
+            continue
+        if not item.is_file():
+            violations.append("result files contain a non-regular filesystem entry")
+            continue
+        size = item.stat().st_size
+        file_count += 1
+        file_bytes += size
+        largest_file_bytes = max(largest_file_bytes, size)
+        if size > MAX_RESULT_FILE_BYTES:
+            violations.append("a result file exceeds the per-file byte limit")
+
+    if file_count > MAX_RESULT_FILES:
+        violations.append("result file count exceeds the allowed bound")
+
+    captured_bytes = int(stdout_info["stored_bytes"]) + int(stderr_info["stored_bytes"])
+    if file_bytes + captured_bytes > MAX_RESULT_TOTAL_BYTES:
+        violations.append("result payload exceeds the total byte limit")
+
+    budget = {
+        "limits": {
+            "captured_stream_bytes_each": MAX_CAPTURED_STREAM_BYTES,
+            "result_file_bytes_each": MAX_RESULT_FILE_BYTES,
+            "result_files": MAX_RESULT_FILES,
+            "result_total_bytes": MAX_RESULT_TOTAL_BYTES,
+        },
+        "observed": {
+            "result_files": file_count,
+            "result_file_bytes": file_bytes,
+            "largest_result_file_bytes": largest_file_bytes,
+            "stdout": stdout_info,
+            "stderr": stderr_info,
+        },
+        "exceeded": bool(violations),
+        "violations": sorted(set(violations)),
+    }
+
+    if violations:
+        shutil.rmtree(files_root, ignore_errors=True)
+        files_root.mkdir()
+        (files_root / "result-budget-exceeded.json").write_text(
+            json.dumps(budget, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+        )
+    return budget
+
+
 def package_result(source: Path, destination: Path) -> None:
     with tarfile.open(destination, mode="w:gz") as tf:
         for item in sorted(source.rglob("*")):
@@ -165,13 +261,20 @@ def run_assignment(
                 exit_code = 124
                 timed_out = True
 
+        task_exit_code = exit_code
+        result_budget = enforce_result_budget(result)
+        if result_budget["exceeded"]:
+            exit_code = RESULT_BUDGET_EXIT_CODE
+
         metadata = {
             "schema_version": 1,
             "assignment_id": assignment_id,
             "started_at": started_at,
             "ended_at": _utc_now(),
             "exit_code": exit_code,
+            "task_exit_code": task_exit_code,
             "timed_out": timed_out,
+            "result_budget": result_budget,
             "capsule_sha256": capsule_sha256,
             "worker_repository": os.getenv("GITHUB_REPOSITORY"),
             "worker_revision": os.getenv("GITHUB_SHA"),

tests/test_sealed_public_execution.py

@@ -1,6 +1,7 @@
 import base64
 import hashlib
 import io
+import json
 import tarfile
 import tempfile
 import unittest
@@ -12,6 +13,9 @@
 import sealed_public_execution as worker
 
 
+VALID_AGE_RECIPIENT = "age1" + "q" * 58
+
+
 def make_capsule(entries):
     buffer = io.BytesIO()
     with tarfile.open(fileobj=buffer, mode="w:gz") as tf:
@@ -36,9 +40,10 @@ def test_assignment_ids_are_bounded(self):
             worker._validate_assignment_id("../escape")
 
     def test_age_recipient_shape_is_required(self):
-        self.assertEqual(worker._validate_recipient("age1qqqqqqqq"), "age1qqqqqqqq")
-        with self.assertRaises(worker.WorkerError):
-            worker._validate_recipient("not-a-recipient")
+        self.assertEqual(worker._validate_recipient(VALID_AGE_RECIPIENT), VALID_AGE_RECIPIENT)
+        for invalid in ("not-a-recipient", "age1qqqqqqqq", "AGE1" + "q" * 58):
+            with self.assertRaises(worker.WorkerError):
+                worker._validate_recipient(invalid)
 
     def test_decode_requires_matching_digest(self):
         raw = make_capsule([("run.sh", "echo ok\n", "file")])
@@ -85,6 +90,66 @@ def test_safe_extract_rejects_links(self):
             with self.assertRaises(worker.WorkerError):
                 worker.safe_extract(capsule, root / "work")
 
+    def test_result_budget_truncates_streams_without_failing_small_results(self):
+        with tempfile.TemporaryDirectory() as td:
+            result = Path(td) / "result"
+            files = result / "files"
+            files.mkdir(parents=True)
+            (files / "out.json").write_text("{}\n", encoding="utf-8")
+            large = b"A" * (worker.MAX_CAPTURED_STREAM_BYTES + 4096)
+            (result / "stdout.txt").write_bytes(large)
+            (result / "stderr.txt").write_bytes(b"small\n")
+
+            budget = worker.enforce_result_budget(result)
+
+            self.assertFalse(budget["exceeded"])
+            self.assertTrue(budget["observed"]["stdout"]["truncated"])
+            self.assertLessEqual(
+                (result / "stdout.txt").stat().st_size,
+                worker.MAX_CAPTURED_STREAM_BYTES,
+            )
+            self.assertTrue((files / "out.json").is_file())
+
+    def test_result_budget_replaces_oversized_result_with_diagnostic(self):
+        with tempfile.TemporaryDirectory() as td:
+            result = Path(td) / "result"
+            files = result / "files"
+            files.mkdir(parents=True)
+            (result / "stdout.txt").write_bytes(b"ok\n")
+            (result / "stderr.txt").write_bytes(b"")
+            (files / "too-large.bin").write_bytes(
+                b"X" * (worker.MAX_RESULT_FILE_BYTES + 1)
+            )
+
+            budget = worker.enforce_result_budget(result)
+
+            self.assertTrue(budget["exceeded"])
+            self.assertFalse((files / "too-large.bin").exists())
+            diagnostic = files / "result-budget-exceeded.json"
+            self.assertTrue(diagnostic.is_file())
+            stored = json.loads(diagnostic.read_text(encoding="utf-8"))
+            self.assertTrue(stored["exceeded"])
+            self.assertIn(
+                "a result file exceeds the per-file byte limit",
+                stored["violations"],
+            )
+
+    def test_result_budget_rejects_excessive_file_count(self):
+        with tempfile.TemporaryDirectory() as td:
+            result = Path(td) / "result"
+            files = result / "files"
+            files.mkdir(parents=True)
+            (result / "stdout.txt").write_bytes(b"")
+            (result / "stderr.txt").write_bytes(b"")
+            for index in range(worker.MAX_RESULT_FILES + 1):
+                (files / f"{index:04d}.txt").write_text("x", encoding="utf-8")
+
+            budget = worker.enforce_result_budget(result)
+
+            self.assertTrue(budget["exceeded"])
+            self.assertEqual(len(list(files.iterdir())), 1)
+            self.assertTrue((files / "result-budget-exceeded.json").is_file())
+
 
 if __name__ == "__main__":
     unittest.main()