Merge pull request #387 from ruby-dlee/codex/nm-scoped-live-proof

fix(azure): harden no-mistakes worker execution

Author: ruby-dlee

.github/workflows/ci.yml

@@ -213,6 +213,8 @@ jobs:
 
   agent-fleet:
     name: Agent Fleet package
+    needs: behavior-test-plan
+    if: needs.behavior-test-plan.outputs.mode == 'full'
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v6

bin/fm-azure-worker-provider.py

@@ -2121,6 +2121,12 @@ def bootstrap_script(action):
     bindings = action["bindings"]
     script = """set -eu
 umask 077
+if ! /usr/bin/getent passwd fmworker >/dev/null; then
+  /usr/sbin/useradd --system --user-group --create-home --home-dir /var/lib/firstmate-worker-user --shell /usr/sbin/nologin fmworker
+fi
+[ "$(/usr/bin/id -u fmworker)" -ne 0 ]
+[ "$(/usr/bin/getent passwd fmworker | /usr/bin/cut -d: -f6)" = /var/lib/firstmate-worker-user ]
+[ "$(/usr/bin/getent passwd fmworker | /usr/bin/cut -d: -f7)" = /usr/sbin/nologin ]
 install -d -m 0755 /usr/local/libexec
 python3 - <<'PY'
 from pathlib import Path

bin/fm-worker-supervisor.py

@@ -13,6 +13,7 @@
 import json
 import os
 from pathlib import Path
+import pwd
 import re
 import shutil
 import subprocess
@@ -312,28 +313,16 @@ def stage_payload(request, worktree, account_home):
         repo = worktree / "repo"
         if repo.is_symlink() or not repo.is_dir():
             raise SupervisorError("existing task-disk repository is unavailable or redirected")
-        top = subprocess.run(
-            ["git", "-C", str(repo), "rev-parse", "--show-toplevel"],
-            stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
-            timeout=GIT_HEAD_TIMEOUT, check=False,
-        )
+        top = git_in(repo, "rev-parse", "--show-toplevel", timeout=GIT_HEAD_TIMEOUT)
         if top.returncode != 0 or Path(top.stdout.decode().strip()).resolve() != repo.resolve():
             raise SupervisorError("existing task-disk repository is not the exact repository root")
-        lineage = subprocess.run(
-            [
-                "git", "-C", str(repo), "merge-base", "--is-ancestor",
-                request["repository_generation"], "HEAD",
-            ],
-            stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
-            timeout=GIT_HEAD_TIMEOUT, check=False,
+        lineage = git_in(
+            repo, "merge-base", "--is-ancestor",
+            request["repository_generation"], "HEAD", timeout=GIT_HEAD_TIMEOUT,
         )
         if lineage.returncode != 0:
             raise SupervisorError("existing task-disk repository lost its dispatched lineage")
-        readable = subprocess.run(
-            ["git", "-C", str(repo), "status", "--porcelain"],
-            stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
-            timeout=GIT_STATUS_TIMEOUT, check=False,
-        )
+        readable = git_in(repo, "status", "--porcelain", timeout=GIT_STATUS_TIMEOUT)
         if readable.returncode != 0:
             raise SupervisorError("existing task-disk working tree is unreadable")
         return repo
@@ -367,11 +356,7 @@ def stage_payload(request, worktree, account_home):
                 clone.stderr.decode("utf-8", errors="replace")[-500:]
             )
         )
-    head = subprocess.run(
-        ["git", "-C", str(repo), "rev-parse", "HEAD"],
-        stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
-        timeout=GIT_HEAD_TIMEOUT, check=False,
-    )
+    head = git_in(repo, "rev-parse", "HEAD", timeout=GIT_HEAD_TIMEOUT)
     if head.returncode != 0 or head.stdout.decode().strip() != request["repository_generation"]:
         raise SupervisorError("staged repository head differs from the bound repository generation")
     if request.get("worker_role") == "no-mistakes":
@@ -387,7 +372,8 @@ def stage_no_mistakes_runtime(source, target, enforce_linux=True):
         if target.is_symlink() or not target.is_dir():
             raise SupervisorError("no-mistakes runtime target is unsafe")
         shutil.rmtree(target)
-    target.mkdir(mode=0o700)
+    target.mkdir(mode=0o755)
+    target.chmod(0o755)
     extracted = {}
     total = 0
     try:
@@ -411,7 +397,7 @@ def stage_no_mistakes_runtime(source, target, enforce_linux=True):
                 if total > 2 * 1024 * 1024 * 1024:
                     raise SupervisorError("no-mistakes runtime expands beyond its bound")
                 destination = target.joinpath(*parts)
-                destination.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+                destination.parent.mkdir(parents=True, exist_ok=True, mode=0o755)
                 destination.write_bytes(body)
                 destination.chmod(member.mode)
                 extracted[member.name] = body
@@ -482,13 +468,69 @@ def stage_no_mistakes_runtime(source, target, enforce_linux=True):
             ):
                 raise SupervisorError(
                     "no-mistakes runtime {} is not Linux amd64".format(path))
+    for directory, directories, _ in os.walk(target):
+        Path(directory).chmod(0o755)
+        for name in directories:
+            child = Path(directory) / name
+            if child.is_symlink():
+                raise SupervisorError("no-mistakes runtime contains a redirected directory")
+            child.chmod(0o755)
+
+
+def no_mistakes_execution_identity():
+    try:
+        identity = pwd.getpwnam("fmworker") if os.geteuid() == 0 else pwd.getpwuid(os.geteuid())
+    except KeyError:
+        raise SupervisorError("no-mistakes service user is unavailable") from None
+    if identity.pw_uid == 0 or identity.pw_gid == 0:
+        raise SupervisorError("no-mistakes service user is privileged")
+    return identity
 
 
-def git_in(repo, *arguments, timeout=BUNDLE_CREATE_TIMEOUT):
+def chown_tree(root, uid, gid):
+    root = Path(root)
+    if root.is_symlink() or not root.is_dir():
+        raise SupervisorError("no-mistakes writable root is unavailable or redirected")
+    os.chown(root, uid, gid, follow_symlinks=False)
+    for directory, directories, files in os.walk(root, followlinks=False):
+        base = Path(directory)
+        os.chown(base, uid, gid, follow_symlinks=False)
+        for name in directories + files:
+            os.chown(base / name, uid, gid, follow_symlinks=False)
+
+
+def prepare_no_mistakes_execution(worktree, worktree_root, account_home, brief):
+    identity = no_mistakes_execution_identity()
+    if os.geteuid() == 0:
+        chown_tree(worktree, identity.pw_uid, identity.pw_gid)
+        chown_tree(account_home, identity.pw_uid, identity.pw_gid)
+    account_home.chmod(0o700)
+    runtime = worktree_root / ".fm-runtime"
+    if runtime.is_symlink() or not runtime.is_dir():
+        raise SupervisorError("no-mistakes runtime root is unavailable or redirected")
+    for directory, directories, _ in os.walk(runtime, followlinks=False):
+        Path(directory).chmod(0o755)
+        for name in directories:
+            child = Path(directory) / name
+            if child.is_symlink():
+                raise SupervisorError("no-mistakes runtime contains a redirected directory")
+            child.chmod(0o755)
+    worktree_root.chmod(0o711)
+    brief.parent.chmod(0o711)
+    if os.geteuid() == 0:
+        os.chown(brief, identity.pw_uid, identity.pw_gid, follow_symlinks=False)
+    brief.chmod(0o400)
+    if os.geteuid() == 0:
+        return {"user": identity.pw_uid, "group": identity.pw_gid, "extra_groups": []}
+    return {}
+
+
+def git_in(repo, *arguments, timeout=BUNDLE_CREATE_TIMEOUT, input_bytes=None, env=None):
     return subprocess.run(
-        ["git", "-C", str(repo), *arguments],
-        stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
-        timeout=timeout, check=False,
+        ["git", "-c", "safe.directory={}".format(repo), "-C", str(repo), *arguments],
+        input=input_bytes, stdin=subprocess.DEVNULL if input_bytes is None else None,
+        stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout, check=False,
+        env=env,
     )
 
 
@@ -589,9 +631,9 @@ def _scratch_artifacts(repo):
 
 
 def _hash_blob(repo, body):
-    result = subprocess.run(
-        ["git", "-C", str(repo), "hash-object", "-w", "--stdin"], input=body,
-        stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=GIT_HEAD_TIMEOUT, check=False,
+    result = git_in(
+        repo, "hash-object", "-w", "--stdin", input_bytes=body,
+        timeout=GIT_HEAD_TIMEOUT,
     )
     if result.returncode != 0:
         raise SupervisorError("returned artifact could not be stored in the repository")
@@ -602,9 +644,8 @@ def _return_commit(repo, base, artifacts, request):
     entries = []
     for name, body in sorted(artifacts.items()):
         entries.append("100644 blob {}\t{}\n".format(_hash_blob(repo, body), name))
-    tree = subprocess.run(
-        ["git", "-C", str(repo), "mktree"], input="".join(entries).encode(),
-        stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=GIT_HEAD_TIMEOUT, check=False,
+    tree = git_in(
+        repo, "mktree", input_bytes="".join(entries).encode(), timeout=GIT_HEAD_TIMEOUT,
     )
     if tree.returncode != 0:
         raise SupervisorError("returned artifact tree could not be created")
@@ -617,11 +658,10 @@ def _return_commit(repo, base, artifacts, request):
         "GIT_AUTHOR_DATE": "@0 +0000",
         "GIT_COMMITTER_DATE": "@0 +0000",
     })
-    committed = subprocess.run(
-        ["git", "-C", str(repo), "commit-tree", tree.stdout.decode().strip(), "-p", base],
-        input=("Firstmate worker return {}\n".format(request["request_digest"])).encode(),
-        stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=GIT_HEAD_TIMEOUT,
-        check=False, env=environment,
+    committed = git_in(
+        repo, "commit-tree", tree.stdout.decode().strip(), "-p", base,
+        input_bytes=("Firstmate worker return {}\n".format(request["request_digest"])).encode(),
+        timeout=GIT_HEAD_TIMEOUT, env=environment,
     )
     if committed.returncode != 0:
         raise SupervisorError("returned artifact commit could not be created")
@@ -945,6 +985,7 @@ def execute(request, worktree, worktree_root):
         "GIT_ASKPASS": "/bin/false",
     }
     argv = request["argv"]
+    execution_identity = {}
     if request.get("worker_role") == "no-mistakes":
         # The no-mistakes guest is already the isolated Azure test boundary.
         # Its project command uses this fixed marker to avoid recursively
@@ -959,11 +1000,13 @@ def execute(request, worktree, worktree_root):
             raise SupervisorError("no-mistakes staged brief is unavailable or redirected")
         argv = list(argv)
         argv[6] = str(brief.resolve())
+        execution_identity = prepare_no_mistakes_execution(
+            worktree, worktree_root, account_home, brief)
     try:
         completed = subprocess.run(
             argv, cwd=str(worktree), env=safe_env,
             stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
-            timeout=request["wall_seconds"], check=False,
+            timeout=request["wall_seconds"], check=False, **execution_identity,
         )
         timed_out = False
         exit_code = completed.returncode

docs/azure-workers.md

@@ -406,8 +406,10 @@ Repair results return one digest-bound single-ref bundle whose head must descend
 The wrapper records a retryable local candidate before cleanup, releases through `service-complete` only after the lifecycle owns the exact execution result, and replays the candidate after a lost response instead of executing the step again.
 Admission, execute recovery, and cleanup use `service-reconcile`, which advances only the caller's exact task generation or replays that task's own pending slot claim rather than converging unrelated fleet work.
 The guest supervisor marks a no-mistakes Azure execution as the already-isolated test boundary, so the repository test command runs the focused service suite directly instead of recursively provisioning the general validation fleet or a Herdr lab.
+The root-owned supervisor stages the job, then runs the no-mistakes process as the dedicated non-root `fmworker` user with no supplementary groups; the sealed runtime remains root-owned and read-only while the exact repository and projected account are writable only by that service identity.
 `bin/fm-azure-service-test-scope.py` owns that focused inventory and the narrow source set eligible for focused pull-request CI; an empty, mixed, or unknown diff and every push to `main` retain the complete behavior suite.
 The wrapper preserves first-seen admission, execute, and cleanup start/completion timestamps in the task's `phase-evidence.json`, so retries keep one stable latency record.
+That evidence is stored at `$FM_HOME/state/no-mistakes-workers/<task>/<generation>/phase-evidence.json` as epoch milliseconds. Subtract each phase's `*_started` value from its `*_completed` value to measure admission, guest execution, and cleanup independently; a missing completion timestamp means that phase has not durably finished and must not be reported as complete.
 No caller chooses an Azure account, sees a credential, invokes `fm-azure-runner.sh`, or bypasses lifecycle cleanup.
 
 The failed retained proof `azr-763d70ab8206` used the generic Azure runner, reached runtime dependency installation, then exited 125 at `guest bootstrap: isolated executor failed` without any structured result.

tests/fm-azure-service-test-scope.test.sh

@@ -90,6 +90,14 @@ workflow_contract() {
   # shellcheck disable=SC2016 # The assertion requires the literal workflow expansion.
   assert_grep 'bin/fm-lint.sh "${files[@]}"' "$CI" \
     "focused CI does not delegate its changed shell files to the canonical lint owner"
+  python3 - "$CI" <<'PY' || fail "focused CI still runs the unrelated Agent Fleet package job"
+from pathlib import Path
+import sys
+body = Path(sys.argv[1]).read_text()
+block = body.split("\n  agent-fleet:\n", 1)[1].split("\n  invariants:\n", 1)[0]
+assert "needs: behavior-test-plan" in block
+assert "if: needs.behavior-test-plan.outputs.mode == 'full'" in block
+PY
   pass "CI runs focused test and lint paths behind stable required checks and preserves full fallback"
 }
 

tests/fm-no-mistakes-worker.test.sh

@@ -320,6 +320,8 @@ target = temporary / "verified-runtime"
 module.stage_no_mistakes_runtime(archive_path, target, enforce_linux=False)
 assert (target / "bin/no-mistakes").read_bytes() == binary
 assert (target / "bin/no-mistakes").stat().st_mode & 0o111
+assert target.stat().st_mode & 0o777 == 0o755
+assert all(path.stat().st_mode & 0o777 == 0o755 for path in target.rglob("*") if path.is_dir())
 PY
 pass "guest supervisor re-verifies the sealed runtime inventory and executable"
 

tests/fm-worker-supervisor.test.sh

@@ -725,6 +725,21 @@ run_supervisor_existing_task_disk_recovery() {
   fm_git_init_commit "$repo"
   base=$(git -C "$repo" rev-parse HEAD)
   printf 'uncommitted scout evidence\n' > "$repo/scratch.txt"
+  python3 - "$repo" <<'PY' \
+    || fail "could not prepare the cross-owner retained repository fixture"
+import os
+from pathlib import Path
+import pwd
+import sys
+
+root = Path(sys.argv[1])
+if os.geteuid() == 0:
+    identity = pwd.getpwnam("nobody")
+    for directory, directories, files in os.walk(root):
+        os.chown(directory, identity.pw_uid, identity.pw_gid)
+        for name in directories + files:
+            os.chown(Path(directory) / name, identity.pw_uid, identity.pw_gid)
+PY
   cat > "$work/.fm-return/data/recover-existing/report.md" <<'REPORT'
 ## Summary
 
@@ -841,7 +856,74 @@ PY
   expect_code 2 "$status" "existing task-disk recovery with foreign lineage should refuse: $out"
   assert_contains "$out" "lost its dispatched lineage" "existing task-disk lineage refusal was not explicit"
   assert_present "$repo/scratch.txt" "a refused existing task-disk recovery removed scout work"
-  pass "existing task-disk recovery returns reports and scratch without restaging or lineage drift"
+  pass "existing task-disk recovery and return Git work across ownership without lineage drift"
+}
+
+run_no_mistakes_privilege_contract() {
+  local tmp
+  fm_test_tmproot_into tmp fm-worker-supervisor-no-mistakes-user
+  python3 - "$SUPERVISOR" "$ROOT/bin/fm-azure-worker-provider.py" "$tmp" <<'PY' \
+    || fail "no-mistakes guest did not execute through its unprivileged service identity"
+import importlib.util
+import os
+from pathlib import Path
+from types import SimpleNamespace
+import sys
+
+supervisor_path, provider_path, temporary = map(Path, sys.argv[1:])
+spec = importlib.util.spec_from_file_location("fm_worker_supervisor", supervisor_path)
+module = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(module)
+
+root = temporary / "work"
+repo = root / "repo"
+account = temporary / "account"
+brief = root / ".fm-task" / "brief.md"
+runtime_dir = root / ".fm-runtime" / "lib" / "pi"
+for directory in (repo, account / "pi-agent", brief.parent, runtime_dir):
+    directory.mkdir(parents=True, exist_ok=True, mode=0o700)
+(repo / "tracked.txt").write_text("tracked\n")
+(account / "pi-agent" / "auth.json").write_text("{}\n")
+brief.write_text("{}\n")
+(runtime_dir / "cli.js").write_text("export {};\n")
+
+module.no_mistakes_execution_identity = lambda: SimpleNamespace(pw_uid=os.getuid(), pw_gid=os.getgid())
+module.os.geteuid = lambda: 0
+options = module.prepare_no_mistakes_execution(repo, root, account, brief)
+assert options == {"user": os.getuid(), "group": os.getgid(), "extra_groups": []}, options
+assert brief.stat().st_mode & 0o777 == 0o400
+assert root.stat().st_mode & 0o111
+assert all(path.stat().st_mode & 0o111 for path in (root / ".fm-runtime").rglob("*") if path.is_dir())
+
+calls = []
+def fake_run(argv, **kwargs):
+    calls.append((argv, kwargs))
+    return SimpleNamespace(returncode=0, stdout=b"", stderr=b"")
+module.subprocess.run = fake_run
+module.os.environ["FM_WORKER_ACCOUNT_HOME"] = str(account)
+request = {
+    "worker_role": "no-mistakes",
+    "argv": ["no-mistakes", "worker", "run", "--role", "test", "--brief", "brief.md", "--result", "outcome.json"],
+    "wall_seconds": 60,
+    "assignment_generation": "asg-1",
+    "request_digest": "a" * 64,
+    "task": "task-1",
+    "task_generation": "gen-1",
+    "cloud_instance_id": "vm-1",
+    "repository_binding": "b" * 64,
+    "repository_generation": "c" * 40,
+}
+module.execute(request, repo, root)
+task_call = next(kwargs for argv, kwargs in calls if argv and argv[0] == "no-mistakes")
+assert task_call["user"] == os.getuid()
+assert task_call["group"] == os.getgid()
+assert task_call["extra_groups"] == []
+
+provider = provider_path.read_text()
+assert "useradd --system --user-group" in provider
+assert "fmworker" in provider
+PY
+  pass "no-mistakes guest runs as a dedicated non-root service user"
 }
 
 run_supervisor_controls
@@ -850,4 +932,5 @@ run_supervisor_steer_controls
 run_supervisor_payload_staging
 run_supervisor_outcome_collection
 run_supervisor_existing_task_disk_recovery
+run_no_mistakes_privilege_contract
 echo "# fm-worker-supervisor.test.sh: all assertions passed"