Experiment formal custody reader green plan 01 - kmosoti/FabricO11y GitHub Wiki
Bounded custody correction source checks
PROSPECTIVE NOT RUN. Actual353 red retained separately; production e12 correction +126/−14/net112,total409≤420/max+120. Root static review found no actionable blocker against the registered finite scope. Full log/common schema only; metric/trace nested leaves opaque. Query oracle unchanged and original payload/SHA stays exact.
Clean final eaf1b3589627416154bafb2615058945b2b74e0e merges exact reviewed PR66db5 safeguards without rebasing. Only CURRENT conflict resolved by keeping PR66 prospective/wiki-linked checker text plus the custody prospective behavior line; historical result paragraph not reintroduced. Original14literal custody controls byte-identical; broader install subtree includes PR66 checker/mutation provenance and filesystem-only A12/continuation controls. No VM/native/Cargo workload.
Default serial16/20GiB/swap0/1800s/full1GiB and fresh95GB complete namespace census everycell; failure stop/post. Shared locked DATA-only Bun dependencies as prior sourcechecks, no install/shared writes. Driver SHA256 dbe53076b4970c6de48a6771156778274d6ab0a1a989980c986fa5a598bdc8f9.
{
"tools/qualification/install/custody_witness.py": "bc4e58836df4b9cd69f4d64c02651ee51bd4421897a0cff16720460d73b18df3",
"tools/qualification/install/test_custody_witness.py": "27ad5a252bfacd51e1625db73c2bb8773c631dd52041a709d265509b8e696e9d",
"tools/qualification/install/test_run_qemu.py": "35bbbc144065a22bd37390aed7325ded8f56b3bdfbee672944fc19292cfba744",
"tools/qualification/query_oracle.py": "fc696ee1173093c258ff4451c4a796dd8e5832761f9e0ff49f89a922b411d23a",
"docs/experiments/formal/installation-release-protocol.md": "7ab2f8a18d7c69cde01a4bf4bd7d7f6e59622683b33ca1c835fff1bce6193482"
}
python3 -B /run/media/kmosoti/data/FabricO11y/results/candidate-build-fresh-8fb6f413179b-01/run-step-custody-reader-green-01.py installation-custody-final-controls-01 1073741824 python3 -B -m unittest discover -s tools/qualification/install -p test_*.py -v
python3 -B /run/media/kmosoti/data/FabricO11y/results/candidate-build-fresh-8fb6f413179b-01/run-step-custody-reader-green-01.py installation-custody-final-docs-check-01 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-custody-reader-green-01.py installation-custody-final-docs-tests-01 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-custody-reader-green-01.py installation-custody-final-docs-hooks-01 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
Implementation diff:
diff --git a/tools/qualification/install/custody_witness.py b/tools/qualification/install/custody_witness.py
index 01cbcd4..2bef1b6 100644
--- a/tools/qualification/install/custody_witness.py
+++ b/tools/qualification/install/custody_witness.py
@@ -18,6 +18,26 @@ MAX_BATCH = 1048576
BINDING = ('run_id', 'candidate_frozen', 'candidate_package_sha256',
'package_source_commit', 'harness_commit')
UNITS = ('fabrico11y-server.service', 'fabrico11y-node.service')
+SCHEMAS = {
+ 'logs': {1: 'resource_logs'},
+ 'resource_logs': {1: 'resource', 2: 'scope_logs', 3: 'utf8'},
+ 'scope_logs': {1: 'scope', 2: 'record', 3: 'utf8'},
+ 'record': {1: 1, 2: 0, 3: 'utf8', 5: 'any', 6: 'kv', 7: 0, 8: 5, 9: 2, 10: 2, 11: 1, 12: 'utf8'},
+ 'resource': {1: 'kv', 2: 0, 3: 'entity'},
+ 'scope': {1: 'utf8', 2: 'utf8', 3: 'kv', 4: 0},
+ 'entity': {1: 'utf8', 2: 'utf8', 3: 'utf8', 4: 'utf8'},
+ 'kv': {1: 'utf8', 2: 'any', 3: 0},
+ 'any': {1: 'utf8', 2: 0, 3: 0, 4: 1, 5: 'array', 6: 'kvlist', 7: 2, 8: 0},
+ 'array': {1: 'any'}, 'kvlist': {1: 'kv'},
+ # Metric/Span nested leaves stay opaque; product ingest owns their semantics.
+ 'metrics': {1: 'resource_metrics'}, 'traces': {1: 'resource_traces'},
+ 'resource_metrics': {1: 2, 2: 'scope_metrics', 3: 2},
+ 'resource_traces': {1: 2, 2: 'scope_traces', 3: 2},
+ 'scope_metrics': {1: 2, 2: 'metric', 3: 2}, 'scope_traces': {1: 2, 2: 'span', 3: 2},
+ 'metric': {1: 2, 2: 2, 3: 2, 5: 2, 7: 2, 9: 2, 10: 2, 11: 2, 12: 2},
+ 'span': {1: 2, 2: 2, 3: 2, 4: 2, 5: 2, 6: 0, 7: 1, 8: 1, 9: 2, 10: 0,
+ 11: 2, 12: 0, 13: 2, 14: 0, 15: 2, 16: 5},
+}
def require(condition, message):
@@ -25,24 +45,114 @@ def require(condition, message):
raise ValueError(message)
-def fields(raw, types):
- try:
- result = oracle.parse_fields(raw)
- require(all(number > 0 for number in result), 'invalid protobuf field')
- for number, values in result.items():
- for wire, value in values:
- require(number not in types or wire == types[number], 'wrong known-field wire type')
- if wire == 0:
- require(0 <= value <= U64, 'protobuf integer overflow')
- return result
- except oracle.MalformedInput as error:
- raise ValueError(str(error)) from error
+def varint(raw, at):
+ value = 0
+ for index in range(10):
+ require(at < len(raw), 'truncated protobuf varint')
+ byte = raw[at]
+ at += 1
+ require(index != 9 or byte <= 1, 'protobuf varint overflow')
+ value |= (byte & 127) << (7 * index)
+ if byte < 128:
+ return value, at
+ raise ValueError('protobuf varint too long')
+
+
+def read_fields(raw, types, depth=0, at=0, group=None):
+ require(0 <= depth <= 100, 'protobuf recursion depth exhausted')
+ result = {}
+ while at < len(raw):
+ key, at = varint(raw, at)
+ number, wire = key >> 3, key & 7
+ require(0 < number < 2**29 and key <= 2**32 - 1, 'invalid protobuf field/tag')
+ require(wire in (0, 1, 2, 3, 4, 5), 'invalid protobuf wire type')
+ if wire == 4:
+ require(group == number, 'unmatched protobuf end group')
+ return result, at
+ spec = types.get(number)
+ require(spec is not None or depth < 100, 'protobuf recursion depth exhausted')
+ expected = 2 if isinstance(spec, str) else spec
+ packed = isinstance(spec, tuple)
+ require(spec is None or wire == expected or (packed and wire in (2, spec[1])),
+ 'wrong known-field wire type')
+ if wire == 3:
+ _, at = read_fields(raw, {}, depth + 1, at, number)
+ continue
+ if wire == 0:
+ value, at = varint(raw, at)
+ else:
+ size, at = varint(raw, at) if wire == 2 else (8 if wire == 1 else 4, at)
+ require(size <= len(raw) - at, 'truncated protobuf length/fixed field')
+ value, at = raw[at:at + size], at + size
+ if packed and wire == 2:
+ require(spec[0] == 'packed' and spec[1] in (0, 1, 5), 'invalid packed specification')
+ if spec[1] == 0:
+ packed_at = 0
+ while packed_at < len(value):
+ _, packed_at = varint(value, packed_at)
+ else:
+ require(len(value) % (8 if spec[1] == 1 else 4) == 0, 'packed fixed width')
+ if group is None:
+ # Last-occurrence order identifies the active known oneof arm.
+ values = result.pop(number, [])
+ values.append((wire, value))
+ result[number] = values
+ require(group is None, 'unterminated protobuf group')
+ return result, at
+
+
+def fields(raw, types, depth=0):
+ return read_fields(raw, types, depth)[0]
+
+
+def validate(raw, schema, depth=0):
+ types = SCHEMAS[schema]
+ for number, values in fields(raw, types, depth).items():
+ spec = types.get(number)
+ if isinstance(spec, str):
+ for _, value in values:
+ if spec == 'utf8':
+ value.decode('utf-8')
+ else:
+ require(depth < 100, 'protobuf message recursion depth exhausted')
+ validate(value, spec, depth + 1)
def one(values, number, default=0):
return values.get(number, [(None, default)])[-1][1]
+def length_field(number, raw):
+ size, encoded = len(raw), bytearray()
+ while size >= 128:
+ encoded.append((size & 127) | 128)
+ size >>= 7
+ return bytes([number << 3 | 2]) + bytes(encoded) + bytes([size]) + raw
+
+
+def string_projection(values):
+ f = fields(b''.join(raw for _, raw in values), SCHEMAS['any'])
+ active = next((number for number in reversed(f) if number in SCHEMAS['any']), None)
+ return length_field(1, one(f, 1, b'')) if active == 1 else b''
+
+
+def log_rows(raw, batch):
+ index = 0
+ for _, resource in fields(raw, SCHEMAS['logs']).get(1, []):
+ for _, scope in fields(resource, SCHEMAS['resource_logs']).get(2, []):
+ for _, record in fields(scope, SCHEMAS['scope_logs']).get(2, []):
+ f = fields(record, SCHEMAS['record'])
+ projection = [b'\x59' + one(f, 11, b'\0' * 8), length_field(5, string_projection(f.get(5, [])))]
+ for _, attribute in f.get(6, []):
+ kv = fields(attribute, SCHEMAS['kv'])
+ projection.append(length_field(6, length_field(1, one(kv, 1, b''))
+ + length_field(2, string_projection(kv.get(2, [])))))
+ row = oracle.decode_log_record(b''.join(projection))
+ yield dict(node='spindle-1', node_id=batch['node_id'].hex(), sequence=batch['sequence'],
+ index=index, observed_ns=row['observed_time_unix_nano'], body=row['body'], attributes=row['attributes'])
+ index += 1
+
+
def cursor(raw):
f = fields(raw, {1: 2, 2: 0, 3: 0, 4: 0, 5: 0, 6: 0, 7: 0, 8: 2})
path = one(f, 1, b'').decode('utf-8')
@@ -74,8 +184,10 @@ def envelope(raw):
for gap in gaps:
gap.decode('utf-8')
cursors = [cursor(value) for _, value in f.get(7, [])]
- batch = oracle.decode_batch(raw)
- rows = oracle.materialize_record('spindle-1', 0, raw).log_rows
+ batch = dict(version=one(f, 1), node_id=one(f, 2, b''), generation=one(f, 3), sequence=one(f, 4))
+ for signal, schema in zip(signals, ('metrics', 'logs', 'traces')):
+ validate(signal, schema)
+ rows = list(log_rows(signals[1], batch))
return batch, cursors, rows
except (oracle.MalformedInput, UnicodeError, TypeError, struct.error) as error:
raise ValueError(str(error)) from error
Exact driver:
import datetime, hashlib, json, os, re, signal, subprocess, sys, time
from pathlib import Path
ROOT = Path('/run/media/kmosoti/data/FabricO11y')
REPO = ROOT / 'scratch/installation-custody-witness'
OUT = ROOT / 'results/service-opening-candidate'
STORE = ROOT / 'scratch/service-opening-census'
FROZEN = 'eaf1b3589627416154bafb2615058945b2b74e0e'
def require(value, reason):
if not value:
raise ValueError(reason)
label, reserve_text, *command = sys.argv[1:]
reserve = int(reserve_text)
delegated = command[:1] == ['--delegate']
if delegated: command = command[1:]
require(re.fullmatch(r'[a-z0-9-]{1,80}', label) and reserve > 0 and command, 'ordinary workload admission guard')
record_path = OUT / (label + '.json')
require(not record_path.exists(), 'unique run label required')
def census():
# Rootless container files can belong to subordinate UIDs. Inspect all of
# them in the same user namespace; never skip unreadable directories.
require(not STORE.is_symlink(), 'ordinary workload admission guard')
require((STORE / '.owned-census-store').read_text() == 'service-opening-census-v1\n', 'ordinary workload admission guard')
args = ['podman', '--root', str(STORE/'graph'), '--runroot', str(STORE/'run'),
'--tmpdir', str(STORE/'tmp'), '--storage-driver', 'vfs',
'unshare', 'python3', '-B', str(OUT/'census.py')]
env = dict(os.environ, TMPDIR=str(STORE/'tmp'), TMP=str(STORE/'tmp'), TEMP=str(STORE/'tmp'))
result = subprocess.run(args, env=env, text=True, capture_output=True, timeout=25)
if result.returncode:
raise RuntimeError(f'strict namespace census failed ({result.returncode}): {result.stderr}')
observed = json.loads(result.stdout)
require(observed['uid'] == 0 and observed['gid'] == 0, 'ordinary workload admission guard')
require(type(observed['data_bytes']) is int and observed['data_bytes'] >= 0, 'ordinary workload admission guard')
return observed
def git(*args):
return subprocess.check_output(['git', *args], cwd=REPO, text=True).strip()
require(git('rev-parse', 'HEAD') == FROZEN and not git('status', '--porcelain'), 'clean frozen source required')
require(not subprocess.check_output(['systemctl', '--user', 'list-units', 'fabric-work-*', '--state=running', '--plain', '--no-legend'], text=True).strip(), 'another project workload is active')
initial = census()
used = initial['data_bytes']
record = {'label': label, 'command': command, 'delegated': delegated, 'frozen': None, 'scope': 'harness controls only; no candidate acceptance evidence', 'harness_revision': FROZEN, 'source_status_before': git('status', '--porcelain'), 'storage_root': str(ROOT), 'admission_bytes': used, 'admission_observation': initial, 'reserved_bytes': reserve, 'observations': [], 'started_utc': datetime.datetime.now(datetime.timezone.utc).isoformat()}
if used + reserve > 95_000_000_000:
record.update(state='not_run', reason='95 GB storage admission threshold')
record_path.write_text(json.dumps(record, indent=2)+'\n')
raise SystemExit('storage admission denied; project command not launched')
receipts = REPO / 'target/resource-containment/runs'
before = set(receipts.glob('*.json'))
env = dict(os.environ, FABRIC_STORAGE_ROOT=str(ROOT), CARGO_HOME=str(ROOT/'toolchain-cache/cargo'), RUSTUP_HOME=str(ROOT/'toolchain-cache/rustup'), CARGO_BUILD_JOBS='2')
env['PATH'] = str(ROOT/'tools/candidate-dpkg/bin') + ':' + env['PATH']
log_path = OUT / (label + '.log')
record.update(state='running', log=str(log_path))
record_path.write_text(json.dumps(record, indent=2)+'\n')
def find_unit():
with log_path.open() as stream:
prefix = stream.read(65536)
match = re.search(r'Running as unit: (fabric-work-[0-9a-f]+)\.service', prefix)
return match.group(1) if match else None
def stop_owned(process):
unit = find_unit()
if unit:
stopped = subprocess.run(['systemctl', '--user', 'stop', unit],
text=True, capture_output=True, timeout=25)
record['stop'] = {'unit': unit, 'exit': stopped.returncode, 'stderr': stopped.stderr}
if stopped.returncode:
# Interrupt the launcher too; its finally block independently
# confirms that this exact service has no surviving descendants.
process.send_signal(signal.SIGINT)
else:
process.send_signal(signal.SIGINT)
record['stop'] = {'unit': None, 'launcher_interrupted': True}
with log_path.open('wb') as log:
launcher = ['python3', '-B', 'tools/resource_group.py', *(['--delegate'] if delegated else []), '--', *command]
process = subprocess.Popen(launcher, cwd=REPO, env=env, stdout=log, stderr=subprocess.STDOUT)
next_census = time.monotonic() + 30
try:
while process.poll() is None:
time.sleep(1)
if time.monotonic() >= next_census:
observed = census()
record['observations'].append({'unix':time.time(), **observed})
record_path.write_text(json.dumps(record, indent=2)+'\n')
if observed['data_bytes'] >= 95_000_000_000:
record['interruption'] = '95 GB threshold reached'
break
next_census = time.monotonic() + 30
except BaseException as error:
record['interruption'] = f'{type(error).__name__}: {error}'
finally:
if record.get('interruption') and process.poll() is None:
try:
stop_owned(process)
except BaseException as error:
record['stop_error'] = f'{type(error).__name__}: {error}'
process.send_signal(signal.SIGINT)
status = process.wait()
record['launcher_exit'] = status
if record.get('interruption'): status = 2
with log_path.open('rb') as stream:
log_digest = hashlib.file_digest(stream, 'sha256').hexdigest()
record.update(state='interrupted' if record.get('interruption') else 'completed', exit=status, source_status_after=git('status', '--porcelain'), finished_utc=datetime.datetime.now(datetime.timezone.utc).isoformat(), log_sha256=log_digest, resource_receipts=[])
archive = OUT/'resource-containment'; archive.mkdir(exist_ok=True)
for p in sorted(set(receipts.glob('*.json')) - before):
value=json.loads(p.read_text())
if value['command'] != command: continue
q=archive/p.name; q.write_bytes(p.read_bytes())
record['resource_receipts'].append({'path':str(q),'sha256':hashlib.sha256(q.read_bytes()).hexdigest(),'exit':value['exit'],'stop_confirmed':value['stop_confirmed'],'retained_failure_evidence':value['retained_failure_evidence']})
if not record['resource_receipts'] or any(not r['stop_confirmed'] or r['exit'] != status for r in record['resource_receipts']):
record['receipt_error'] = 'missing or inconsistent outer resource receipt'
status = 2
record['exit'] = status
record_path.write_text(json.dumps(record, indent=2)+'\n')
print(json.dumps(record, indent=2))
if status: print(log_path.read_text()[-8000:])
raise SystemExit(status)