Experiment formal soak exposure source plan 02 amended - kmosoti/FabricO11y GitHub Wiki
R3 narrow source correction: amended pre-flight 02
Prospective: NOT RUN. Clean head ff3c1828be48a3299b9570dcb86e0cc5d457bc7c follows trust-only 81b3592 and its actual 19-method registration red. Production changes only the six pure observation checks and guarded stopped rename, whole-parent cgroup snapshots and two decoded mount refusal calls. All registered tests and protocol bytes remain unchanged. Production delta +196/-3 (net +193); docs +16/-2.
The first cell selects 18 original controls; it deliberately excludes the still-known-red 5460-second default method because helper authority, native begin and the C-081 170/7210 phase amendment are separate pending work. The old default remains 5460 and no R3 runtime claim follows from these source cells. Mount deletion and kernel files are mocked; rename uses tiny owned ordinary files. No real mount, guest, Rust compile, simulator, clock, browser or qualification workload is launched. The existing ten gate names and independent delivery/query/rate oracle bytes are unchanged.
Root executes five serial default-capped cells, stopping at any unexpected failure. Each uses fresh full DATA namespace census, 1 GiB reservation, admission below 95 GB with reserve, 100 GB ceiling, 16 GiB memory high / 20 GiB max / no swap / 1800 seconds for the entire descendant tree. The driver refuses dirty or moved source and simultaneous project units. Success removes owned launcher scratch; failure evidence and exact receipts are preserved. #59 is read after every cell.
The existing locked Bun1.4.0 dependency tree is reused through 106 individual ignored links; no install, dependency mutation or Cargo cache creation. Expected Python control counts are observations to record, not asserted greens. All affected existing source suites run together; the three documentation commands cover the actual docs delta.
C-082 ACK required three additional indirect cgroup consumers: test_query_page_admission, test_release_stress, and test_release_main. They are included in the exact cell 2 below. enter_group.py and self_observation_smoke.py have no source test suite; their native integration remains unverified. This standalone amended plan preserves the original plan as historical proposal.
Pins
/run/media/kmosoti/data/FabricO11y/results/candidate-build-fresh-8fb6f413179b-01/run-step-soak-exposure-source-02.pySHA-25627aa880bbe34c155b29520c242f3c9eaf233dd882a04cc853151fe4f40f85948./run/media/kmosoti/data/FabricO11y/scratch/soak-exposure/tools/qualification/soak_exposure.pySHA-256a3a5e4d444ae0e57839b74f87eaf74146268295770db341b8ac0fda9126b5e81./run/media/kmosoti/data/FabricO11y/scratch/soak-exposure/tools/qualification/preserve_soak_r2.pySHA-256a0cbbbdb5c233cf59d76732b92eb113a79c8409c85b2df792a268f8dd7e14b31./run/media/kmosoti/data/FabricO11y/scratch/soak-exposure/tools/bench/labs/completion/cgroups.pySHA-2561ce92954007c6589b2ad67378a77acb982c86d8060a38890ffdcca6c3a086d20./run/media/kmosoti/data/FabricO11y/scratch/soak-exposure/tools/qualification/soak_tier.pySHA-256534436eb9e601a38f7049641c82683b66d06151b918d6e7c7f55f2b73f812b06./run/media/kmosoti/data/FabricO11y/scratch/soak-exposure/tools/qualification/test_soak_exposure.pySHA-256a4c6bf864c34003ccb4f6fa8a3babf730c6e4167746ae5765af325e85699757d./run/media/kmosoti/data/FabricO11y/scratch/soak-exposure/docs/experiments/benchmarks/soak-protocol-r3.mdSHA-2560f5ed3cbb14f3f144240150fb745a87c268e220f8589ec09091ec8b8f6887199./run/media/kmosoti/data/FabricO11y/scratch/soak-exposure/docs/CURRENT.mdSHA-2561224ad1d622c313f047388461548bbb3e035cbf04c709f71108234e0f973cb37./run/media/kmosoti/data/FabricO11y/scratch/soak-exposure/docs/architecture/deployment.mdSHA-2565b2636130ff9c2dffc96e1d13e8105d39c1cce901e8256780d52dd956e4299a5.
Exact commands
python3 -B /run/media/kmosoti/data/FabricO11y/results/candidate-build-fresh-8fb6f413179b-01/run-step-soak-exposure-source-02.py soak-exposure-source-controls-02 1073741824 env PYTHONPATH=tools/qualification:tools/ui:tools/bench/labs/completion:tools python3 -B -m unittest -v test_soak_exposure.LiteralExposureControls test_soak_exposure.ExposureMechanismAvailability test_soak_exposure.ExposureMechanismControls test_soak_exposure.ExistingSoakSourceControls.test_original_ten_gate_names_remain_exact test_soak_exposure.ExistingSoakSourceControls.test_kernel_snapshot_includes_actual_parent_and_population test_soak_exposure.ExistingSoakSourceControls.test_same_device_mount_refuses_before_inventory_without_deletion test_soak_exposure.ExistingSoakSourceControls.test_new_same_device_mount_refuses_immediately_before_first_unlink
python3 -B /run/media/kmosoti/data/FabricO11y/results/candidate-build-fresh-8fb6f413179b-01/run-step-soak-exposure-source-02.py soak-exposure-affected-controls-02 1073741824 env PYTHONPATH=tools/qualification:tools/ui:tools/bench/labs/completion:tools python3 -B -m unittest -v test_soak_tier test_soak_companion test_soak_observations test_production_access test_prepare_soak test_preserve_soak_r2 test_release_runtime test_release_fixture test_query_corpus test_cgroups test_query_page_admission test_release_stress test_release_main
python3 -B /run/media/kmosoti/data/FabricO11y/results/candidate-build-fresh-8fb6f413179b-01/run-step-soak-exposure-source-02.py soak-exposure-docs-check-02 1073741824 env PATH=/run/media/kmosoti/data/FabricO11y/tools/bun-1.4.0:/usr/local/bin:/usr/bin:/bin BUN_INSTALL_CACHE_DIR=/run/media/kmosoti/data/FabricO11y/tools/docs-bun-cache /run/media/kmosoti/data/FabricO11y/tools/bun-1.4.0/bun tools/docs/check.mjs
python3 -B /run/media/kmosoti/data/FabricO11y/results/candidate-build-fresh-8fb6f413179b-01/run-step-soak-exposure-source-02.py soak-exposure-docs-tests-02 1073741824 env PATH=/run/media/kmosoti/data/FabricO11y/tools/bun-1.4.0:/usr/local/bin:/usr/bin:/bin BUN_INSTALL_CACHE_DIR=/run/media/kmosoti/data/FabricO11y/tools/docs-bun-cache /run/media/kmosoti/data/FabricO11y/tools/bun-1.4.0/bun tools/docs/check.test.mjs
python3 -B /run/media/kmosoti/data/FabricO11y/results/candidate-build-fresh-8fb6f413179b-01/run-step-soak-exposure-source-02.py soak-exposure-docs-hooks-02 1073741824 env PATH=/run/media/kmosoti/data/FabricO11y/tools/bun-1.4.0:/usr/local/bin:/usr/bin:/bin BUN_INSTALL_CACHE_DIR=/run/media/kmosoti/data/FabricO11y/tools/docs-bun-cache python3 -B tools/docs/test_hooks.py
Remaining integration boundaries
These validators accept caller-supplied records, not authenticated native observations. Native callers must bind real begin/end and stop/group observations, independently read receipt hashes, retain complete immutable raw inventories and measure S1 phase maxima with 25% margin. No smoke/FULL, helper replacement, canonical manifest mutation, schedule change or larger deadline is admitted by this plan. The separate C-081 helper/source/170/7210 amendment must be registered and controlled before runtime integration or any build.
Current production sources
tools/qualification/soak_exposure.py
"""Registered R3 observation checks; callers supply independently admitted inputs.
No clock is read by the six validators. Their success is not a native run,
receipt authentication or permission to delete evidence.
"""
import ctypes
import errno
import os
from pathlib import Path
import re
import stat
NS = 1_000_000_000
EDGES = (60, 853, 1646, 2440, 3233, 4026, 4820, 5613, 6406, 7200)
CEILINGS = {'setup': 180, 'exposure': 7200, 'drain_seal': 60,
'shutdown': 20, 'checkpoint': 20, 'closeout': 20}
PARENT_FIELDS = ('memory.high', 'memory.max', 'memory.swap.max', 'pids.max',
'memory.events', 'memory.peak', 'memory.current', 'memory.swap.current',
'pids.current', 'pids.peak', 'cgroup.events', 'cgroup.procs', 'cpu.max', 'cpu.stat', 'io.stat')
def require(condition, reason):
if not condition:
raise ValueError(reason)
def sha256(value):
return isinstance(value, str) and re.fullmatch('[0-9a-f]{64}', value) is not None
def cohort_ns(timestamp, began):
require(type(timestamp) is int and type(began) is int, 'cohort timestamps')
offset = timestamp - began
return next((index for index, (lo, hi) in enumerate(zip(EDGES, EDGES[1:]))
if lo * NS <= offset < hi * NS), None)
def validate_exposure(sim, interval):
require(isinstance(sim, dict) and isinstance(interval, dict), 'exposure evidence')
require(all(type(sim.get(key)) is int for key in ('seconds', 'identities', 'seed', 'began_unix_ns'))
and all(type(interval.get(key)) is int for key in ('start_ns', 'end_ns', 'sim_began_unix_ns', 'last_offered_second')),
'exposure evidence')
edges = interval.get('cohort_edges_s')
require(sim['seconds'] == 7200 and sim['identities'] == 100 and sim['seed'] == 0xA11FA001
and sim['began_unix_ns'] >= 0 and interval['sim_began_unix_ns'] == sim['began_unix_ns']
and 0 <= interval['start_ns'] <= interval['end_ns'] - 7200 * NS
and isinstance(edges, (list, tuple)) and all(type(value) is int for value in edges) and tuple(edges) == EDGES
and interval['last_offered_second'] == 7199 and interval.get('probes_active') is True,
'exposure evidence')
return True
def remaining_ns(phase, launcher_begin, phase_begin, now):
require(phase in CEILINGS and all(type(value) is int for value in (launcher_begin, phase_begin, now))
and 0 <= launcher_begin <= phase_begin <= now, 'phase timestamps')
remaining = min(CEILINGS[phase] * NS - (now - phase_begin), 7500 * NS - (now - launcher_begin))
require(remaining > 0, 'phase budget')
return remaining
def validate_phase_headroom(measurements):
require(isinstance(measurements, dict) and set(measurements) == set(CEILINGS) - {'exposure'}, 'phase headroom')
for phase, record in measurements.items():
require(isinstance(record, dict), 'phase headroom')
maximum = record.get('maximum_ns')
require(type(maximum) is int and maximum > 0
and (5 * maximum + 3) // 4 <= CEILINGS[phase] * NS
and record.get('kind') in ('native-r2', 'full-rate-smoke')
and record.get('stop_confirmed') is True and sha256(record.get('source_sha256')), 'phase headroom')
return True
def validate_checkpoint(measurement, runtime, outer, expected):
refs = {'measurement': measurement, 'runtime': runtime, 'outer': outer}
require(isinstance(expected, dict) and isinstance(expected.get('receipt_sha256'), dict)
and all(isinstance(ref, dict) for ref in refs.values()), 'checkpoint binding')
require(all(sha256(ref.get('sha256')) and ref['sha256'] == expected['receipt_sha256'].get(name)
and isinstance(ref.get('value'), dict) for name, ref in refs.items()), 'checkpoint binding')
m, r, o = (ref['value'] for ref in (measurement, runtime, outer))
child = o.get('command_receipt', {})
require(isinstance(child, dict) and bool(expected.get('unit')) and bool(expected.get('cgroup')),
'checkpoint binding')
exits = r.get('production_exits')
require(bool(expected.get('binding')) and m.get('state') == 'pending_acceptance' and 'passed' not in m
and r.get('state') == 'closed_pending_acceptance' and 'passed' not in r
and m.get('binding') == r.get('binding') == expected['binding']
and r.get('cleanup_ok') is True and r.get('immutable_inputs') is True
and isinstance(exits, dict) and set(exits) == {'server', 'simulator'}
and all(type(value) is int and value == 0 for value in exits.values())
and r.get('forced_shutdown') is False
and o.get('unit') == expected.get('unit') and o.get('stop_confirmed') is True
and bool(expected.get('command')) and o.get('command') == expected['command']
and type(o.get('runtime_max_seconds')) is int and o['runtime_max_seconds'] == 7500
and all(type(value) is int and value == 0 for value in (o.get('exit'), o.get('wrapper_exit'), child.get('exit')))
and child.get('state') == 'completed' and child.get('cgroup') == expected.get('cgroup'), 'checkpoint binding')
return True
def validate_parent(groups, stopped=False):
require(isinstance(groups, dict), 'whole-parent containment')
root = groups.get('.', {})
require(isinstance(root, dict) and all(name in root and isinstance(root[name], str) for name in PARENT_FIELDS),
'whole-parent containment')
require(all(root[name] == value for name, value in {'memory.high': '5368709120',
'memory.max': '6442450944', 'memory.swap.max': '0', 'pids.max': '1536'}.items()), 'whole-parent containment')
try:
events = dict(line.split() for line in root['memory.events'].splitlines())
population = dict(line.split() for line in root['cgroup.events'].splitlines())
require(events.get('oom') == events.get('oom_kill') == '0'
and root['memory.swap.current'] == '0'
and all(0 <= int(root[name]) <= cap for name, cap in
(('memory.peak', 6442450944), ('memory.current', 6442450944),
('pids.peak', 1536), ('pids.current', 1536)))
and population.get('populated') in ('0', '1')
and (not stopped or (population['populated'] == '0' and not root['cgroup.procs'].strip())),
'whole-parent containment')
except (ValueError, TypeError) as error:
raise ValueError('whole-parent containment') from error
return True
def require_no_mounts(root):
root = Path(root).resolve(strict=True)
for line in Path('/proc/self/mountinfo').read_text(errors='surrogateescape').splitlines():
fields = line.split()
require(len(fields) >= 10 and '-' in fields[6:], 'malformed mountinfo')
encoded = fields[4]
require(re.search(r'\\(?!040|011|012|134)', encoded) is None, 'malformed mountinfo escape')
mount = Path(re.sub(r'\\(040|011|012|134)', lambda match: chr(int(match[1], 8)), encoded))
require(mount.is_absolute() and '..' not in mount.parts, 'malformed mountinfo path')
require(not mount.is_relative_to(root), 'mount at or below owned evidence root')
def canonical_directory(path):
path = Path(path)
require(path.is_absolute() and '..' not in path.parts and path.resolve(strict=True) == path,
'stopped checkpoint needs canonical directories')
cursor = Path('/')
for part in path.parts[1:]:
cursor /= part
require(stat.S_ISDIR(cursor.lstat().st_mode), 'stopped checkpoint has linked or non-directory parent')
require(path.stat().st_uid == os.getuid(), 'stopped checkpoint directory is not owned')
return path
def checkpoint_tree(source, destination, *, processes_stopped=True):
require(processes_stopped is True, 'stopped checkpoint requires confirmed processes')
source, destination = Path(source), Path(destination)
require(destination.is_absolute() and '..' not in destination.parts
and not destination.exists() and not destination.is_symlink(), 'fresh checkpoint destination required')
source = canonical_directory(source)
source_parent, destination_parent = canonical_directory(source.parent), canonical_directory(destination.parent)
require(not destination.is_relative_to(source), 'fresh checkpoint cannot be inside its source')
before = source.stat()
require(before.st_dev == destination_parent.stat().st_dev, 'same-filesystem checkpoint required')
require_no_mounts(source)
require_no_mounts(destination_parent)
descriptors = []
try:
for path in (source_parent, destination_parent, source):
descriptors.append(os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC))
parent_fd, destination_fd, source_fd = descriptors
identity = lambda info: (info.st_dev, info.st_ino, info.st_mode, info.st_uid, info.st_gid)
source_stamp = lambda info: (*identity(info), info.st_nlink, info.st_size, info.st_mtime_ns, info.st_ctime_ns)
require(source_stamp(before) == source_stamp(os.fstat(source_fd)) == source_stamp(source.lstat()),
'stopped checkpoint source changed')
require(all(identity(path.lstat()) == identity(os.fstat(fd)) for path, fd in
((source_parent, parent_fd), (destination_parent, destination_fd))), 'stopped checkpoint parent changed')
require(os.fstat(source_fd).st_dev == os.fstat(destination_fd).st_dev, 'same-filesystem checkpoint required')
libc = ctypes.CDLL(None, use_errno=True)
rename = getattr(libc, 'renameat2', None)
require(rename is not None, 'stopped checkpoint requires no-replace rename')
rename.argtypes = (ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint)
rename.restype = ctypes.c_int
if rename(parent_fd, os.fsencode(source.name), destination_fd, os.fsencode(destination.name), 1) != 0:
code = ctypes.get_errno()
if code == errno.EEXIST:
raise ValueError('fresh checkpoint destination already exists')
if code == errno.EXDEV:
raise ValueError('same-filesystem checkpoint required')
raise OSError(code, os.strerror(code))
require(not source.exists() and identity(destination.lstat()) == identity(os.fstat(source_fd)), 'stopped checkpoint rename identity changed')
os.fsync(parent_fd)
os.fsync(destination_fd)
return {'method': 'rename', 'source': str(source), 'destination': str(destination),
'device': before.st_dev, 'inode': before.st_ino}
finally:
for fd in reversed(descriptors):
os.close(fd)
tools/qualification/preserve_soak_r2.py
#!/usr/bin/env python3
"""Preserve, verify, then optionally remove only the completed R2 soak freeze.
The preserve action creates a private archive and per-member manifest. The
cleanup action requires that archive plus a successful frozen-post receipt.
Both actions are intended to run through tools/resource_group.py.
"""
import argparse
import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import shutil
import stat
import sys
import tarfile
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "tools"))
from resource_group import require_limits, require_storage, STORAGE # noqa: E402
from soak_exposure import require_no_mounts # noqa: E402
FREEZE = STORAGE / "scratch/soak-r2-02"
MARKER = ".fabric-soak-freeze-owned"
MARKER_TEXT = "fabric-soak-freeze-r2-v1\n"
FULL_RUN = "alpha-soak-r2-full"
RUNNER_MARKER = ".fabric-alpha-owned"
RUNNER_MARKER_TEXT = "fabric-alpha-runner-v1\n"
EXPECTED_LAUNCHER_ID = "5f9ee6b610a1418bb7ff96d172385516"
EXPECTED_LAUNCHER_UNIT = "fabric-work-" + EXPECTED_LAUNCHER_ID
RESULTS = STORAGE / "results/readiness-continuation-soak-r2-01"
ARCHIVE = RESULTS / "soak-r2-02-full-freeze.tar.gz"
MEMBERS = RESULTS / "soak-r2-02-full-freeze-members.json"
PRESERVE_RECEIPT = RESULTS / "soak-r2-02-full-freeze-preservation.json"
CLEANUP_RECEIPT = RESULTS / "soak-r2-02-full-freeze-cleanup.json"
def sha256_path(path):
digest = hashlib.sha256()
with path.open("rb") as stream:
for block in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
def private_new(path):
return os.fdopen(os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "wb")
def ensure_private_external(path):
require_storage()
RESULTS.mkdir(parents=True, exist_ok=True)
if (RESULTS.is_symlink() or not RESULTS.is_dir()
or RESULTS.stat().st_uid != os.getuid()):
raise ValueError("external evidence directory is not an owned regular directory")
path.parent.resolve(strict=True).relative_to(STORAGE.resolve(strict=True))
if path.exists() or path.is_symlink():
raise ValueError(f"evidence path already exists: {path}")
def require_archive_space(freeze_bytes):
cap = 100_000_000_000
project_bytes = 0
for directory, dirs, files in os.walk(STORAGE, followlinks=False):
for name in list(dirs):
path = Path(directory) / name
if path.is_symlink():
dirs.remove(name)
project_bytes += path.lstat().st_size
for name in files:
path = Path(directory) / name
project_bytes += path.lstat().st_size
reserve = freeze_bytes + 1024**3
if project_bytes + reserve > cap or shutil.disk_usage(STORAGE).free < reserve:
raise ValueError("archive reserve exceeds FabricO11y data budget or free space")
def ownership_checks(postcheck_path, launcher_receipt_path):
storage = STORAGE.resolve(strict=True)
require_no_symlink_components(FREEZE)
owned = FREEZE.resolve(strict=True)
owned.relative_to((storage / "scratch").resolve(strict=True))
if owned != (storage / "scratch/soak-r2-02").resolve(strict=True):
raise ValueError("unexpected freeze path")
marker = owned / MARKER
if marker.is_symlink() or not marker.is_file() or marker.read_text() != MARKER_TEXT:
raise ValueError("exact freeze ownership marker missing")
run = owned / "frozen/target" / FULL_RUN
if run.is_symlink() or not run.is_dir():
raise ValueError("full R2 runner output missing")
run_marker = run / RUNNER_MARKER
if run_marker.is_symlink() or not run_marker.is_file() or run_marker.read_text() != RUNNER_MARKER_TEXT:
raise ValueError("full R2 runner ownership marker missing")
runner_result = json.loads((run / "result.json").read_text())
token = runner_result.get("invocation_id")
if (not isinstance(token, str) or not token or not runner_result.get("process_group_cleanup_ok")
or runner_result.get("stop_reason") is not None):
raise ValueError("full runner did not record complete child-process cleanup")
if tagged_processes(token):
raise ValueError("full-run tagged processes remain live")
postcheck_path = Path(postcheck_path)
launcher_receipt_path = Path(launcher_receipt_path)
for evidence in (postcheck_path, launcher_receipt_path):
if evidence.is_symlink() or not evidence.is_file() or evidence.stat().st_nlink != 1:
raise ValueError("postcheck and launcher receipt must be regular unlinked files")
evidence.resolve(strict=True).relative_to(storage)
launcher = json.loads(launcher_receipt_path.read_text())
child = launcher.get("command_receipt")
command = launcher.get("command")
validate_launcher_receipt(launcher, child, command)
if launcher.get("exit") != child.get("exit"):
raise ValueError("launcher and command exits disagree")
cgroup = Path(child["cgroup"])
if (not cgroup.is_absolute() or cgroup.parts[:4] != ("/", "sys", "fs", "cgroup")
or cgroup.name != EXPECTED_LAUNCHER_UNIT + ".service"
or ".." in cgroup.parts):
raise ValueError("launcher cgroup path does not identify the exact full-soak unit")
if cgroup.exists():
events_path = cgroup / "cgroup.events"
events = dict(line.split() for line in events_path.read_text().splitlines())
if events.get("populated") != "0":
raise ValueError("full-soak unit cgroup remains populated")
postcheck = json.loads(postcheck_path.read_text())
manifest_path = owned / "provenance/manifest.json"
validate_postcheck(postcheck, manifest_path, sha256_path(manifest_path))
return owned, runner_result, launcher, postcheck
def require_no_symlink_components(path):
absolute = Path(os.path.abspath(path))
cursor = Path(absolute.anchor)
for part in absolute.parts[1:]:
cursor = cursor / part
if cursor.is_symlink():
raise ValueError(f"symlink in pinned data path: {cursor}")
if cursor.exists() and not cursor.is_dir():
raise ValueError(f"non-directory in pinned data path: {cursor}")
def validate_launcher_receipt(launcher, child, command):
if (launcher.get("unit") != EXPECTED_LAUNCHER_UNIT
or launcher.get("stop_confirmed") is not True
or not isinstance(child, dict) or child.get("state") != "completed"
or not isinstance(command, list) or not any(FULL_RUN in str(arg) for arg in command)):
raise ValueError("launcher receipt does not prove the named full soak unit stopped")
return True
def validate_postcheck(postcheck, manifest_path, manifest_sha):
if (postcheck.get("passed") is not True or postcheck.get("label") != "full-post"
or postcheck.get("manifest") != str(manifest_path)
or postcheck.get("manifest_sha256") != manifest_sha):
raise ValueError("successful full-post verification of the exact current manifest is required")
return True
def tagged_processes(token):
needle = f"FABRIC_ALPHA_INVOCATION={token}".encode()
found = []
for entry in os.scandir("/proc"):
if not entry.name.isdigit():
continue
try:
proc = Path("/proc") / entry.name
if proc.stat().st_uid != os.getuid():
continue
values = (proc / "environ").read_bytes().split(b"\0")
if needle in values:
state = (proc / "stat").read_text().rpartition(") ")[2].split()[0]
if state != "Z":
found.append(int(entry.name))
except (FileNotFoundError, ProcessLookupError, PermissionError, IndexError):
continue
return found
def tree_manifest(root):
require_no_mounts(root)
result = [{"path": ".", "type": "directory",
"mode": stat.S_IMODE(root.stat().st_mode), "size": 0, "sha256": None}]
for directory, dirs, files in os.walk(root, topdown=True, followlinks=False):
base = Path(directory)
dirs.sort()
files.sort()
for name in dirs:
path = base / name
if path.is_symlink() or not path.is_dir():
raise ValueError(f"unsupported directory entry: {path}")
result.append({"path": path.relative_to(root).as_posix(), "type": "directory",
"mode": stat.S_IMODE(path.stat().st_mode), "size": 0, "sha256": None})
for name in files:
path = base / name
info = path.lstat()
if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1:
raise ValueError(f"archive accepts only unlinked regular files: {path}")
result.append({"path": path.relative_to(root).as_posix(), "type": "file",
"mode": stat.S_IMODE(info.st_mode), "size": info.st_size,
"sha256": sha256_path(path)})
return sorted(result, key=lambda row: row["path"])
def archive_tree(root, expected):
with private_new(ARCHIVE) as raw:
with tarfile.open(fileobj=raw, mode="w:gz", format=tarfile.PAX_FORMAT) as archive:
for row in expected:
if row["path"] == ".":
archive.add(root, arcname=root.name, recursive=False)
else:
path = root / PurePosixPath(row["path"])
archive.add(path, arcname=(root.name + "/" + row["path"]), recursive=False)
raw.flush()
os.fsync(raw.fileno())
os.chmod(ARCHIVE, 0o600)
verify_archive(ARCHIVE, root.name, expected)
def verify_archive(path, top_name, expected):
seen = []
expected_by_path = {row["path"]: row for row in expected}
if len(expected_by_path) != len(expected):
raise ValueError("duplicate paths in expected archive manifest")
with tarfile.open(path, "r:gz") as archive:
for member in archive:
relative = PurePosixPath(member.name)
if (relative.is_absolute() or ".." in relative.parts or not relative.parts
or relative.parts[0] != top_name):
raise ValueError("unsafe archive member path")
rel = PurePosixPath(*relative.parts[1:]).as_posix() if len(relative.parts) > 1 else "."
info = expected_by_path.get(rel)
if info is None or member.issym() or member.islnk() or member.isdev():
raise ValueError(f"unexpected archive member {member.name}")
expected_type = "directory" if member.isdir() else "file" if member.isfile() else None
if expected_type != info["type"] or stat.S_IMODE(member.mode) != info["mode"]:
raise ValueError(f"archive member metadata mismatch: {rel}")
actual_digest = None
actual_size = 0
if member.isfile():
stream = archive.extractfile(member)
if stream is None:
raise ValueError(f"cannot read archived member {rel}")
digest = hashlib.sha256()
for block in iter(lambda: stream.read(1024 * 1024), b""):
actual_size += len(block)
digest.update(block)
actual_digest = digest.hexdigest()
if actual_size != info["size"] or actual_digest != info["sha256"]:
raise ValueError(f"archive bytes differ: {rel}")
seen.append(rel)
wanted = [row["path"] for row in expected]
if sorted(seen) != wanted or len(seen) != len(set(seen)):
raise ValueError("archive members are missing, duplicated or extra")
return {"member_count": len(seen), "archive_sha256": sha256_path(path),
"members_sha256": hashlib.sha256(json.dumps(expected, sort_keys=True,
separators=(",", ":")).encode()).hexdigest()}
def validate_preservation(receipt, manifest, owned, archive_sha, member_sha):
if (receipt.get("state") != "preserved_and_verified"
or receipt.get("archive_sha256") != archive_sha
or manifest.get("freeze") != owned
or manifest.get("archive_sha256") != receipt.get("archive_sha256")
or manifest.get("members_sha256") != receipt.get("members_sha256")
or member_sha != receipt.get("members_sha256")):
raise ValueError("preservation receipt or archive digest mismatch")
return True
def preserve(postcheck, launcher_receipt):
for path in (ARCHIVE, MEMBERS, PRESERVE_RECEIPT, CLEANUP_RECEIPT,
CLEANUP_RECEIPT.with_name(CLEANUP_RECEIPT.name + ".pending")):
ensure_private_external(path)
owned, runner, launcher, check = ownership_checks(postcheck, launcher_receipt)
expected = tree_manifest(owned)
require_archive_space(sum(row["size"] for row in expected if row["type"] == "file"))
archive_tree(owned, expected)
if tree_manifest(owned) != expected:
raise ValueError("owned tree changed during archival")
verification = verify_archive(ARCHIVE, owned.name, expected)
members_payload = {"freeze": str(owned), "members": expected, **verification}
with private_new(MEMBERS) as stream:
stream.write((json.dumps(members_payload, sort_keys=True, indent=2) + "\n").encode())
stream.flush()
os.fsync(stream.fileno())
receipt = {"state": "preserved_and_verified", "freeze": str(owned),
"launcher_unit": launcher["unit"], "launcher_exit": launcher["exit"],
"runner_exit": runner["exit_code"], "postcheck": str(postcheck),
"archive": str(ARCHIVE), "archive_sha256": verification["archive_sha256"],
"members_manifest": str(MEMBERS), "members_sha256": verification["members_sha256"],
"member_count": verification["member_count"],
"credentials_classification": "synthetic owned soak fixtures; archive and receipts mode 0600",
"ownership_manifest_sha256": check["manifest_sha256"]}
with private_new(PRESERVE_RECEIPT) as stream:
stream.write((json.dumps(receipt, sort_keys=True, indent=2) + "\n").encode())
stream.flush()
os.fsync(stream.fileno())
return receipt
def cleanup(postcheck, launcher_receipt):
owned, runner, launcher, check = ownership_checks(postcheck, launcher_receipt)
if (RESULTS.is_symlink() or not RESULTS.is_dir()
or RESULTS.stat().st_uid != os.getuid()):
raise ValueError("external evidence directory is not an owned regular directory")
if any(path.is_symlink() or not path.is_file() or path.stat().st_nlink != 1
for path in (ARCHIVE, MEMBERS, PRESERVE_RECEIPT)):
raise ValueError("verified archive, member manifest and preservation receipt required")
if any(path.stat().st_mode & 0o077 or path.stat().st_uid != os.getuid()
for path in (ARCHIVE, MEMBERS, PRESERVE_RECEIPT)):
raise ValueError("preservation evidence permissions must remain private")
receipt = json.loads(PRESERVE_RECEIPT.read_text())
manifest = json.loads(MEMBERS.read_text())
expected = manifest["members"]
member_digest = hashlib.sha256(json.dumps(expected, sort_keys=True,
separators=(",", ":")).encode()).hexdigest()
validate_preservation(receipt, manifest, str(owned), sha256_path(ARCHIVE), member_digest)
verify_archive(ARCHIVE, owned.name, expected)
if tree_manifest(owned) != expected:
raise ValueError("live freeze differs from verified archive member manifest")
if CLEANUP_RECEIPT.exists() or CLEANUP_RECEIPT.is_symlink():
raise ValueError("cleanup receipt already exists")
pending = CLEANUP_RECEIPT.with_name(CLEANUP_RECEIPT.name + ".pending")
if pending.exists() or pending.is_symlink():
raise ValueError("cleanup receipt temporary path already exists")
if tagged_processes(runner["invocation_id"]):
raise ValueError("full-run process appeared before cleanup")
started = {"state": "cleanup_started", "freeze": str(owned),
"archive_sha256": receipt["archive_sha256"],
"members_manifest_sha256": sha256_path(MEMBERS),
"launcher_unit": launcher["unit"], "postcheck": str(postcheck)}
with private_new(CLEANUP_RECEIPT) as stream:
stream.write((json.dumps(started, sort_keys=True, indent=2) + "\n").encode())
stream.flush()
os.fsync(stream.fileno())
require_no_mounts(owned)
shutil.rmtree(owned)
if owned.exists():
raise ValueError("owned freeze remains after deletion")
cleanup_receipt = {"state": "owned_freeze_removed", "freeze": str(owned),
"archive_sha256": receipt["archive_sha256"],
"members_manifest_sha256": sha256_path(MEMBERS),
"launcher_unit": launcher["unit"], "postcheck": str(postcheck)}
with private_new(pending) as stream:
stream.write((json.dumps(cleanup_receipt, sort_keys=True, indent=2) + "\n").encode())
stream.flush()
os.fsync(stream.fileno())
os.replace(pending, CLEANUP_RECEIPT)
return cleanup_receipt
def main():
require_limits()
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("action", choices=("preserve", "cleanup"))
parser.add_argument("--postcheck", required=True, type=Path)
parser.add_argument("--launcher-receipt", required=True, type=Path)
args = parser.parse_args()
if args.action == "preserve":
result = preserve(args.postcheck, args.launcher_receipt)
else:
result = cleanup(args.postcheck, args.launcher_receipt)
print(json.dumps(result, sort_keys=True))
if __name__ == "__main__":
try:
raise SystemExit(main())
except (OSError, RuntimeError, ValueError, KeyError, TypeError, tarfile.TarError) as error:
print(f"soak preservation: NOT COMPLETE: {error}", file=sys.stderr)
raise SystemExit(2)
tools/bench/labs/completion/cgroups.py
"""Owned delegated descendants only. Parent aggregate cap remains in force."""
import os
from pathlib import Path
ROOT = Path('/sys/fs/cgroup')
def current():
return ROOT / next(x[3:] for x in Path('/proc/self/cgroup').read_text().splitlines()
if x.startswith('0::')).lstrip('/')
def memory_limit_values(maximum, high, page_bytes):
if page_bytes <= 0 or maximum <= 0 or high <= 0:
raise ValueError('memory limits and page size must be positive')
if high > maximum:
raise ValueError('memory.high cannot exceed memory.max')
aligned_max = maximum // page_bytes * page_bytes
aligned_high = high // page_bytes * page_bytes
if aligned_max == 0 or aligned_high == 0 or aligned_high > aligned_max:
raise ValueError('page-rounded memory limits are invalid')
return {'memory.max': str(aligned_max), 'memory.high': str(aligned_high)}
def set_limits(group, maximum, high, tasks, cores):
values = {**memory_limit_values(maximum, high, os.sysconf('SC_PAGE_SIZE')),
'memory.swap.max': '0', 'pids.max': str(tasks),
'cpu.max': f'{int(cores * 100000)} 100000'}
for name, value in values.items():
(group / name).write_text(value)
actual = {k: (group / k).read_text().strip() for k in values}
if actual != values:
raise RuntimeError(f'cgroup enforcement mismatch: {actual} != {values}')
return actual
def delegate():
parent = current()
if not parent.name.startswith('fabric-work-'):
raise RuntimeError('expected fresh owned launcher service')
supervisor = parent / 'supervisor'
supervisor.mkdir()
for pid in (parent / 'cgroup.procs').read_text().split():
(supervisor / 'cgroup.procs').write_text(pid)
(parent / 'cgroup.subtree_control').write_text('+memory +cpu +pids +io')
set_limits(supervisor, 20*2**30, 16*2**30, 1024, 4)
return parent
def subgroup(parent, name, maximum, high, tasks, cores, children=False):
group = parent / name
group.mkdir()
limits = set_limits(group, maximum, high, tasks, cores)
if children:
(group / 'cgroup.subtree_control').write_text('+memory +cpu +pids +io')
return group, limits
def enter(group):
# Called before exec; all subsequent children inherit this exact group.
(Path(group) / 'cgroup.procs').write_text(str(os.getpid()))
if current() != Path(group):
raise RuntimeError('child did not enter its assigned group')
def snapshot(parent):
return {str(g.relative_to(parent)): {k: (g/k).read_text().strip()
for k in ('memory.max','memory.high','memory.swap.max','memory.peak','memory.current',
'memory.events','memory.swap.current','cpu.max','cpu.stat','io.stat',
'pids.max','pids.current','pids.peak','cgroup.events','cgroup.procs')}
for g in (parent, *(g for g in parent.rglob('*') if g.is_dir()))}