feat(security): isolate bounded work and support required auth actions

tests/bounded_process_fixtures.py

@@ -0,0 +1,69 @@
+"""Synthetic operations, imported only by isolated tests; no application effects."""
+from __future__ import annotations
+
+import json
+import os
+import re
+import subprocess
+import sys
+import time
+
+
+def echo(payload: bytes) -> bytes:
+    return payload
+
+
+def wait(payload: bytes) -> bytes:
+    time.sleep(float(payload))
+    return b"done"
+
+
+def regex_stall(_payload: bytes) -> bytes:
+    re.fullmatch(r"(a+)+$", "a" * 100 + "!")
+    return b"unreachable"
+
+
+def allocate(_payload: bytes) -> bytes:
+    return b"x" * (256 * 1024 * 1024)
+
+
+def too_much_stdout(_payload: bytes) -> bytes:
+    while True:
+        os.write(1, b"x" * 65536)
+
+
+def too_much_stderr(_payload: bytes) -> bytes:
+    while True:
+        os.write(2, b"sensitive synthetic log" * 4096)
+
+
+def fail(_payload: bytes) -> bytes:
+    raise ValueError("private synthetic data must not become an error response")
+
+
+def close_pipes_then_wait(_payload: bytes) -> bytes:
+    os.close(1)
+    os.close(2)
+    time.sleep(30)
+    return b""
+
+
+def observe(_payload: bytes) -> bytes:
+    import resource
+
+    return json.dumps({
+        "pid": os.getpid(), "pgid": os.getpgrp(), "sid": os.getsid(0),
+        "cpu": resource.getrlimit(resource.RLIMIT_CPU),
+        "memory": resource.getrlimit(resource.RLIMIT_AS),
+        "file": resource.getrlimit(resource.RLIMIT_FSIZE),
+        "core": resource.getrlimit(resource.RLIMIT_CORE),
+        "env": sorted(os.environ), "cwd": os.getcwd(),
+    }).encode()
+
+
+def child_with_closed_pipes(_payload: bytes) -> bytes:
+    child = subprocess.Popen(
+        [sys.executable, "-I", "-c", "import time; time.sleep(30)"],
+        stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+    )
+    return str(child.pid).encode()