diff --git a/docs/operations/scanner-benchmark-runbook.md b/docs/operations/scanner-benchmark-runbook.md index eccff7cc9..fd2dd55d0 100644 --- a/docs/operations/scanner-benchmark-runbook.md +++ b/docs/operations/scanner-benchmark-runbook.md @@ -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`, diff --git a/scripts/scanner_process_receipt.py b/scripts/scanner_process_receipt.py new file mode 100644 index 000000000..93591cdc1 --- /dev/null +++ b/scripts/scanner_process_receipt.py @@ -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()) diff --git a/scripts/test_scanner_process_receipt.py b/scripts/test_scanner_process_receipt.py new file mode 100644 index 000000000..3af6a9925 --- /dev/null +++ b/scripts/test_scanner_process_receipt.py @@ -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() diff --git a/scripts/test_scanner_validation_harness.sh b/scripts/test_scanner_validation_harness.sh index b647db707..573b18386 100755 --- a/scripts/test_scanner_validation_harness.sh +++ b/scripts/test_scanner_validation_harness.sh @@ -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"