test(scanner): attest running benchmark process identity (#8876)

* test(scanner): attest running benchmark process identity * fix(scanner): reject aliased process receipt data roots * fix(scanner): attest scheduled scan configuration
2026-10-26 22:07:53 +00:00 · 2026-10-24 18:22:11 +09:00
parent 90c5050a0f
commit c8f14dc337
4 changed files with 385 additions and 0 deletions
@@ -184,6 +184,38 @@ that changed sessions. Resize reports the kernel CPU/memory limits and rejects
 a changed PID or mismatched limits. Applying limits alone does not demonstrate
 RustFS adaptation or pass the resource hot-shrink gate.

+#### Running-process evidence
+
+After readiness and before each measurement, run `scripts/scanner_process_receipt.py`
+as the same benchmark user. Supply the immutable service plan, a JSON object with
+the expected non-secret environment, and one `--data-root` per dedicated local
+data directory. The environment must exactly match the script's supported
+configuration keys present in the process; an unknown `RUSTFS_` key fails closed.
+Credential variables are excluded. Keep both input files private and archive
+only the resulting receipt, never the environment file or raw `/proc` contents.
+
+```bash
+python3 scripts/scanner_process_receipt.py node-plan.json public-environment.json --data-root /data/isolated-benchmark
+```
+
+The receipt binds the boot ID, PID/start ticks, actual `/proc/PID/exe` SHA-256 and
+inode, expected arguments/configuration hashes, service identity, observed kernel
+CPU/memory limits, and directory device/inode/capacity. It rejects changes seen
+between its two process observations or replacement of the executable or data
+directories. Configuration values and raw process arguments are not emitted.
+The observations span a reported interval, not an atomic snapshot; they do not
+protect against a compromised benchmark user or administrator. Resource limits
+are observations, including legitimate hot-shrink changes, and must be checked
+against the workload plan by the adapter.
+
+Environment bytes do not establish which configuration the server accepted.
+Listed directories are caller-selected, not proof of the server's open volumes,
+device isolation or SSD/HDD class. Those claims, the binary's source/build
+provenance and the independent data oracle still need separate evidence. The
+receipt explicitly leaves effective configuration, directory-to-server binding
+and storage acceptance unverified. A failed observation produces no receipt;
+retain the failure and reconcile the service before continuing.
+
 The `measure` response echoes the observed `evidence`, `fixed`, `build`,
 `data_dir`, and `background`, plus `sample_count` (1..3600), `elapsed_seconds`,
 and `metrics`. All metrics must be finite nonnegative numbers: `p99_ms`,
@@ -0,0 +1,143 @@
+#!/usr/bin/env python3
+"""Read Linux process evidence for an owned scanner benchmark service."""
+
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path
+import sys
+import time
+import uuid
+
+from scanner_service import Service, local_path, require
+
+PROC = Path("/proc")
+SAFE_ENVIRONMENT = frozenset({
+    "RUSTFS_ADDRESS", "RUSTFS_CONSOLE_ENABLE", "RUSTFS_ERASURE_SET_DRIVE_COUNT",
+    "RUSTFS_STORAGE_CLASS_STANDARD", "RUSTFS_DURABILITY_MODE", "RUSTFS_NEW_BUCKET_DURABILITY_MODE",
+    "RUSTFS_SCANNER_ENABLED", "RUSTFS_HEAL_ENABLED", "RUSTFS_SCANNER_SPEED", "RUSTFS_SCANNER_IDLE_MODE",
+    "RUSTFS_SCANNER_START_DELAY_SECS", "RUSTFS_SCANNER_CYCLE",
+    "RUSTFS_OBJECT_LOCK_DIAG_ENABLE", "RUSTFS_OBS_LOGS_EXPORT_ENABLED", "RUSTFS_OBS_TRACES_EXPORT_ENABLED",
+    "RUSTFS_OBS_METRICS_EXPORT_ENABLED", "RUST_LOG", "NO_PROXY", "no_proxy",
+})
+PRIVATE_ENVIRONMENT = frozenset({
+    "RUSTFS_ACCESS_KEY", "RUSTFS_SECRET_KEY", "RUSTFS_ROOT_USER", "RUSTFS_ROOT_PASSWORD",
+})
+
+
+def read_bounded(path, limit=65536):
+    with path.open("rb") as stream:
+        data = stream.read(limit + 1)
+    require(len(data) <= limit, "oversized process input")
+    return data
+
+
+def fingerprint(value):
+    return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
+
+
+def process_state(pid):
+    root = PROC / str(pid)
+    require(root.stat().st_uid == os.getuid(), "process owner mismatch")
+    stat = read_bounded(root / "stat").decode()
+    # comm is parenthesized and may contain spaces or parentheses.
+    end = stat.rfind(")")
+    require(end >= 0 and stat.split(" ", 1)[0] == str(pid), "invalid process stat")
+    fields = stat[end + 1:].split()
+    require(len(fields) >= 20 and int(fields[19]) > 0, "invalid process start time")
+    return {"start_ticks": int(fields[19]),
+            "arguments": read_bounded(root / "cmdline"),
+            "environment": read_bounded(root / "environ"),
+            "cgroup": read_bounded(root / "cgroup"),
+            "cwd": os.readlink(root / "cwd"),
+            "executable": os.readlink(root / "exe")}
+
+
+def environment_digest(raw, expected):
+    require(type(expected) is dict and set(expected) <= SAFE_ENVIRONMENT
+            and all(type(value) is str for value in expected.values()), "unsupported expected environment")
+    observed = {}
+    for entry in raw.split(b"\0"):
+        if not entry:
+            continue
+        key, value = entry.split(b"=", 1)
+        key = key.decode()
+        if key in PRIVATE_ENVIRONMENT:
+            continue
+        if key.startswith("RUSTFS_") or key in SAFE_ENVIRONMENT:
+            require(key in SAFE_ENVIRONMENT and key not in observed, "unrecognized or duplicate benchmark environment")
+            observed[key] = value.decode()
+    require(observed == expected, "benchmark environment mismatch")
+    return fingerprint(observed)
+
+
+def attest(service, expected_environment, data_roots):
+    started = time.time_ns()
+    monotonic_started = time.monotonic_ns()
+    boot_id = str(uuid.UUID((PROC / "sys/kernel/random/boot_id").read_text().strip()))
+    before = service.observe()
+    require(before["active"] == "active" and before["pid"] > 0, "service is not active")
+    pid = before["pid"]
+    state = process_state(pid)
+    require(state["cwd"] == str(service.root) and state["executable"] == str(service.binary),
+            "process path mismatch")
+    require(state["cgroup"] == ("0::/system.slice/" + service.unit + "\n").encode(), "process cgroup mismatch")
+    arguments = [str(service.binary), *service.plan["args"]]
+    require(state["arguments"] == b"\0".join(arg.encode() for arg in arguments) + b"\0", "process arguments mismatch")
+    config_hash = environment_digest(state["environment"], expected_environment)
+    with (PROC / str(pid) / "exe").open("rb") as stream:
+        identity = os.fstat(stream.fileno())
+        binary_hash = hashlib.file_digest(stream, "sha256").hexdigest()
+    require(binary_hash == service.plan["binary_sha256"], "running binary digest mismatch")
+    require(type(data_roots) is list and data_roots and len(data_roots) == len(set(data_roots)), "invalid data roots")
+    directories = []
+    for name in data_roots:
+        path = local_path(name)
+        info = path.stat()
+        require(path.is_dir() and info.st_uid == os.getuid(), "data root must be an owned directory")
+        usage = os.statvfs(path)
+        directories.append({"path": str(path), "device": info.st_dev, "inode": info.st_ino,
+                            "total_bytes": usage.f_blocks * usage.f_frsize,
+                            "available_bytes": usage.f_bavail * usage.f_frsize})
+    require(len({(directory["device"], directory["inode"]) for directory in directories}) == len(directories),
+            "duplicate data directory identity")
+    require(process_state(pid) == state and service.observe() == before, "process or resource limits changed during observation")
+    current = (PROC / str(pid) / "exe").stat()
+    require((current.st_dev, current.st_ino) == (identity.st_dev, identity.st_ino), "running executable changed")
+    for directory in directories:
+        current = local_path(directory["path"]).stat()
+        require((current.st_dev, current.st_ino) == (directory["device"], directory["inode"]),
+                "data directory changed during observation")
+    ended = time.time_ns()
+    require(ended >= started, "wall clock moved backwards")
+    return {"schema": 1, "kind": "process_receipt", "boot_id": boot_id,
+            "observed_start_unix_ns": started, "observed_end_unix_ns": ended,
+            "elapsed_monotonic_ns": time.monotonic_ns() - monotonic_started,
+            "pid": pid, "process_start_ticks": state["start_ticks"],
+            "clock_ticks_per_second": os.sysconf("SC_CLK_TCK"),
+            "binary_sha256": binary_hash, "executable_device": identity.st_dev, "executable_inode": identity.st_ino,
+            "arguments_sha256": fingerprint(arguments), "environment_sha256": config_hash,
+            "environment_keys": sorted(expected_environment), "service": before, "data_roots": directories,
+            "effective_server_configuration_verified": False, "data_roots_bound_to_server": False,
+            "storage_acceptance_passed": False}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("plan", type=Path)
+    parser.add_argument("environment", type=Path, help="expected non-secret environment JSON; private values are excluded")
+    parser.add_argument("--data-root", action="append", required=True)
+    args = parser.parse_args()
+    try:
+        service = Service(json.loads(read_bounded(args.plan)))
+        result = attest(service, json.loads(read_bounded(args.environment)), args.data_root)
+    except (ValueError, TypeError, AttributeError, KeyError, OSError, IndexError):
+        print("process evidence failed; preserve inputs and reconcile service state", file=sys.stderr)
+        return 1
+    print(json.dumps(result, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -0,0 +1,209 @@
+#!/usr/bin/env python3
+"""Reject incomplete or changing process evidence without exposing configuration."""
+
+import copy
+import hashlib
+import json
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+from unittest.mock import patch
+import uuid
+
+import scanner_process_receipt as receipt
+from scanner_service import Service
+
+
+class ProcessReceiptTest(unittest.TestCase):
+    def setUp(self):
+        temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(temporary.cleanup)
+        self.root = Path(temporary.name).resolve()
+        self.root.chmod(0o700)
+        self.binary = self.root / "binary"
+        self.binary.write_bytes(b"actual executable fixture")
+        self.data = self.root / "data"
+        self.data.mkdir()
+        self.plan = {"run_id": str(uuid.uuid4()), "root": str(self.root), "binary": str(self.binary),
+                     "binary_sha256": hashlib.sha256(self.binary.read_bytes()).hexdigest(),
+                     "environment_file": str(self.root / "private.env"), "args": [str(self.data)],
+                     "runtime_seconds": 60, "cpu_percent": 200, "memory_bytes": 268435456}
+        self.service = Service(self.plan)
+        self.proc = self.root / "proc"
+        self.pid = self.proc / "123"
+        self.pid.mkdir(parents=True)
+        boot = self.proc / "sys/kernel/random/boot_id"
+        boot.parent.mkdir(parents=True)
+        boot.write_text(str(uuid.uuid4()))
+        self.start_ticks = 892351
+        (self.pid / "stat").write_text("123 (name with ) parentheses) " +
+                                      " ".join(["S"] + ["0"] * 18 + [str(self.start_ticks)] + ["0"] * 10))
+        (self.pid / "exe").symlink_to(self.binary)
+        (self.pid / "cwd").symlink_to(self.root)
+        (self.pid / "cmdline").write_bytes((str(self.binary) + "\0" + str(self.data) + "\0").encode())
+        (self.pid / "cgroup").write_text("0::/system.slice/" + self.service.unit + "\n")
+        self.public = {"RUSTFS_SCANNER_ENABLED": "false", "NO_PROXY": "*"}
+        self.environment = b"RUSTFS_SCANNER_ENABLED=false\0NO_PROXY=*\0RUSTFS_SECRET_KEY=private-fixture\0PATH=/bin\0"
+        (self.pid / "environ").write_bytes(self.environment)
+        self.observation = {"unit": self.service.unit, "active": "active", "pid": 123, "stopped": False,
+                            "cpu_max": "200000 100000", "memory_max": 268435456}
+        self.addCleanup(patch.stopall)
+        patch.object(receipt, "PROC", self.proc).start()
+        self.observe = patch.object(self.service, "observe", side_effect=lambda: copy.deepcopy(self.observation)).start()
+
+    def attest(self):
+        return receipt.attest(self.service, self.public, [str(self.data)])
+
+    def test_real_file_identity_and_public_digest_without_private_values(self):
+        result = self.attest()
+        self.assertEqual(result["binary_sha256"], self.plan["binary_sha256"])
+        self.assertEqual(result["executable_inode"], self.binary.stat().st_ino)
+        self.assertEqual(result["data_roots"][0]["inode"], self.data.stat().st_ino)
+        self.assertEqual(result["process_start_ticks"], self.start_ticks)
+        self.assertEqual(result["environment_sha256"], receipt.fingerprint(self.public))
+        self.assertEqual(result["service"], self.observation)
+        self.assertEqual(result["clock_ticks_per_second"], os.sysconf("SC_CLK_TCK"))
+        self.assertFalse(result["effective_server_configuration_verified"])
+        self.assertFalse(result["data_roots_bound_to_server"])
+        self.assertFalse(result["storage_acceptance_passed"])
+        self.assertLessEqual(result["observed_start_unix_ns"], result["observed_end_unix_ns"])
+        self.assertNotIn("private-fixture", json.dumps(result))
+        self.assertNotIn("SECRET", json.dumps(result))
+
+    def test_selected_environment_must_match_exactly(self):
+        cases = [self.environment.replace(b"false", b"true"),
+                 self.environment.replace(b"NO_PROXY=*\0", b""),
+                 self.environment + b"RUSTFS_UNKNOWN=private-fixture\0",
+                 self.environment + b"NO_PROXY=*\0"]
+        for raw in cases:
+            with self.subTest(raw_hash=hashlib.sha256(raw).hexdigest()), self.assertRaises(ValueError):
+                receipt.environment_digest(raw, self.public)
+        for expected in ({"RUSTFS_SECRET_KEY": "private-fixture"}, {"NO_PROXY": 1}, []):
+            with self.assertRaises(ValueError):
+                receipt.environment_digest(self.environment, expected)
+
+    def test_credentials_do_not_change_public_digest(self):
+        changed = self.environment.replace(b"private-fixture", b"different-private-fixture")
+        self.assertEqual(receipt.environment_digest(changed, self.public),
+                         receipt.environment_digest(self.environment, self.public))
+
+    def test_scanner_schedule_is_bound_to_actual_process_environment(self):
+        schedule = {"RUSTFS_SCANNER_START_DELAY_SECS": "1", "RUSTFS_SCANNER_CYCLE": "5"}
+        self.public.update(schedule)
+        raw = self.environment + b"RUSTFS_SCANNER_START_DELAY_SECS=1\0RUSTFS_SCANNER_CYCLE=5\0"
+        (self.pid / "environ").write_bytes(raw)
+        result = self.attest()
+        self.assertEqual(result["environment_sha256"], receipt.fingerprint(self.public))
+        self.assertTrue(set(schedule) <= set(result["environment_keys"]))
+        for key in schedule:
+            with self.subTest(key=key), self.assertRaisesRegex(ValueError, "mismatch"):
+                receipt.environment_digest(raw, {**self.public, key: "60"})
+        with self.assertRaises(ValueError):
+            receipt.environment_digest(raw + b"RUSTFS_SCANNER_UNKNOWN=1\0", self.public)
+
+    def test_inactive_service_is_rejected_before_process_access(self):
+        self.observation.update(active="inactive", pid=0)
+        with self.assertRaisesRegex(ValueError, "not active"):
+            self.attest()
+
+    def test_wrong_binary_content_is_rejected(self):
+        self.binary.write_bytes(b"replaced executable")
+        with self.assertRaisesRegex(ValueError, "binary digest"):
+            self.attest()
+
+    def test_arguments_cgroup_and_paths_are_checked(self):
+        for name, replacement in (("cmdline", b"another-command\0"), ("cgroup", b"0::/another.service\n")):
+            path = self.pid / name
+            original = path.read_bytes()
+            with self.subTest(name=name), self.assertRaises(ValueError):
+                path.write_bytes(replacement)
+                self.attest()
+            path.write_bytes(original)
+        for name in ("cwd", "exe"):
+            path = self.pid / name
+            original = os.readlink(path)
+            path.unlink()
+            path.symlink_to(self.data)
+            with self.subTest(name=name), self.assertRaisesRegex(ValueError, "path mismatch"):
+                self.attest()
+            path.unlink()
+            path.symlink_to(original)
+
+    def test_pid_reuse_and_process_changes_during_hash_are_rejected(self):
+        original = receipt.process_state
+        for key, value in (("start_ticks", self.start_ticks + 1), ("arguments", b"changed\0"),
+                           ("environment", self.environment + b"PRIVATE=changed\0"),
+                           ("cgroup", b"0::/changed\n")):
+            first = original(123)
+            second = dict(first, **{key: value})
+            with self.subTest(key=key), patch.object(receipt, "process_state", side_effect=[first, second]), \
+                    self.assertRaisesRegex(ValueError, "changed during observation"):
+                self.attest()
+
+    def test_changed_pid_or_kernel_limit_is_rejected(self):
+        for key, value in (("pid", 124), ("cpu_max", "100000 100000"), ("memory_max", 134217728)):
+            self.observe.side_effect = [self.observation, dict(self.observation, **{key: value})]
+            with self.subTest(key=key), self.assertRaisesRegex(ValueError, "changed during observation"):
+                self.attest()
+
+    def test_replaced_executable_is_rejected_even_with_same_bytes(self):
+        digest = receipt.hashlib.file_digest
+
+        def replace_after_hash(stream, algorithm):
+            result = digest(stream, algorithm)
+            old = self.root / "previous-binary"
+            self.binary.rename(old)
+            self.binary.write_bytes(old.read_bytes())
+            return result
+
+        with patch.object(receipt.hashlib, "file_digest", side_effect=replace_after_hash), \
+                self.assertRaisesRegex(ValueError, "executable changed"):
+            self.attest()
+
+    def test_replaced_data_directory_during_observation_is_rejected(self):
+        def observe_after_data():
+            if self.observe.call_count == 2:
+                self.data.rename(self.root / "previous-data")
+                self.data.mkdir()
+            return copy.deepcopy(self.observation)
+
+        self.observe.side_effect = observe_after_data
+        with self.assertRaisesRegex(ValueError, "data directory changed"):
+            self.attest()
+
+    def test_missing_duplicate_or_aliased_data_roots_are_rejected(self):
+        link = self.root / "alias"
+        link.symlink_to(self.data)
+        for paths in ([], [str(self.data)] * 2, [str(self.data), str(self.data) + "/."],
+                      [str(link)], [str(self.binary)], [str(self.root / "missing")]):
+            with self.subTest(paths=paths), self.assertRaises((ValueError, OSError)):
+                receipt.attest(self.service, self.public, paths)
+
+    def test_bad_stat_and_oversized_inputs_are_rejected(self):
+        path = self.pid / "stat"
+        for value in ("123 (truncated) S 0", "999 (wrong pid) S " + "0 " * 30,
+                      "123 (no start) S " + "0 " * 30):
+            path.write_text(value)
+            with self.assertRaises(ValueError):
+                receipt.process_state(123)
+        path.write_bytes(b"x" * 65537)
+        with self.assertRaisesRegex(ValueError, "oversized"):
+            receipt.read_bounded(path)
+
+    def test_cli_failure_does_not_echo_secret_or_produce_receipt(self):
+        plan = self.root / "plan.json"
+        plan.write_text(json.dumps(self.plan))
+        public = self.root / "public.json"
+        public.write_text('{"private-fixture":broken}')
+        result = subprocess.run(["python3", receipt.__file__, str(plan), str(public), "--data-root", str(self.data)],
+                                capture_output=True, text=True)
+        self.assertEqual(result.returncode, 1)
+        self.assertEqual(result.stdout, "")
+        self.assertNotIn("private-fixture", result.stderr)
+        self.assertNotIn("Traceback", result.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -321,5 +321,6 @@ fi
 grep -q -- 'unknown arg: --secret-key' "$secret_arg_log"

 "$ROOT_DIR/scripts/python_bin.sh" "$ROOT_DIR/scripts/test_scanner_service.py"
+"$ROOT_DIR/scripts/python_bin.sh" "$ROOT_DIR/scripts/test_scanner_process_receipt.py"
 "$ROOT_DIR/scripts/python_bin.sh" "$ROOT_DIR/scripts/test_scanner_abba.py"
 "$ROOT_DIR/scripts/python_bin.sh" "$ROOT_DIR/scripts/test_scanner_otlp_window.py"