diff --git a/docs/adr/ADR-427-enforced-confirmation-training-mode.md b/docs/adr/ADR-427-enforced-confirmation-training-mode.md index 3887db789..6bc66d8d8 100644 --- a/docs/adr/ADR-427-enforced-confirmation-training-mode.md +++ b/docs/adr/ADR-427-enforced-confirmation-training-mode.md @@ -29,7 +29,9 @@ records the supervised loader will consume without assigning activity semantics. input exits 1; valid but blocked evidence exits 2. The dataset loader parses and hashes the manifest through one checked descriptor, then compares that digest before resolving or decoding the referenced dataset -bundle. Its returned digest is checked again before training. +bundle. In confirmation mode, the NPZ final component is also opened once as a +bounded regular file without following symlinks; the same descriptor is hashed, +inspected and decoded. Its returned digest is checked again before training. Confirmation mode cannot optimize an existing bundle and every receipt remains `promotionAuthorized: false`. @@ -45,9 +47,10 @@ declared review to exact manifest bytes and retained record identities. It does consent, licenses, aliases, stable physical identities, or exposure-ledger completeness. The inputs are unsigned. The manifest hash binds declarations and referenced payload hashes, not provenance truth. Parent directories remain a -trusted deployment boundary, and the separately referenced NPZ still has its -own digest/open race boundary. Passing software tests makes no sensing-accuracy -claim. +trusted deployment boundary. Descriptor reuse prevents final-component pathname +replacement, but does not freeze bytes against a writer that can mutate the same +inode. Such writers are outside this local-file trust boundary. Passing software +tests makes no sensing-accuracy claim. The current ESP-Fi adapter pins the canonical dataset name and source revision, but its upstream records do not establish physical session or device identity. diff --git a/scripts/csi_training/espfi.py b/scripts/csi_training/espfi.py index 9becd61f6..8cf395cee 100644 --- a/scripts/csi_training/espfi.py +++ b/scripts/csi_training/espfi.py @@ -5,11 +5,14 @@ and supplied benchmark class directories disagree about activity semantics. """ from __future__ import annotations import argparse +from contextlib import contextmanager import hashlib import io import json +import os from pathlib import Path, PurePosixPath import re +import stat import zipfile import numpy as np @@ -29,6 +32,45 @@ def digest(path): return hashlib.file_digest(f, 'sha256').hexdigest() +def _stream_digest(stream): + stream.seek(0) + value = hashlib.file_digest(stream, 'sha256').hexdigest() + stream.seek(0) + return value + + +@contextmanager +def _open_regular_bundle(path): + """Open one bounded final-component descriptor for hash and decode. + + Parent directories remain a trusted deployment boundary. O_NONBLOCK keeps a + substituted FIFO from blocking before fstat; O_NOFOLLOW rejects a final + symlink. A descriptor prevents pathname replacement between digest and + decode, but does not prevent a writer from modifying the same inode. + """ + if not hasattr(os, 'O_NOFOLLOW') or not hasattr(os, 'O_NONBLOCK'): + raise ValueError('secure dataset bundle intake unavailable') + flags = os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK + if hasattr(os, 'O_CLOEXEC'): + flags |= os.O_CLOEXEC + if hasattr(os, 'O_BINARY'): + flags |= os.O_BINARY + try: + fd = os.open(path, flags) + except OSError as exc: + raise ValueError('dataset bundle unavailable or symlinked') from exc + try: + info = os.fstat(fd) + if not stat.S_ISREG(info.st_mode) or not 0 < info.st_size <= MAX_NPZ: + raise ValueError('dataset bundle must be a bounded regular file') + with os.fdopen(fd, 'rb', closefd=False) as stream: + yield stream + except OSError as exc: + raise ValueError('dataset bundle read failed') from exc + finally: + os.close(fd) + + def parse_identity(name): p = PurePosixPath(name) if p.is_absolute() or '..' in p.parts or '\\' in name: @@ -202,10 +244,50 @@ def _load_partitions(manifest, splits, expected_manifest_sha256=None): duplicate = occurrences[record['signal_sha256']] > 1 if record['split'] != (None if duplicate else split_identity(ident)): raise ValueError('incorrect source split assignment') - npz = (path.parent / meta['npz']).resolve() - if npz.parent != path.parent or npz.stat().st_size > MAX_NPZ or digest(npz) != meta['npz_sha256']: - raise ValueError('unsafe or modified dataset bundle') - with zipfile.ZipFile(npz) as z: + npz_name, npz_sha256 = meta.get('npz'), meta.get('npz_sha256') + if (not isinstance(npz_name, str) or not npz_name or len(npz_name) > 255 + or Path(npz_name).name != npz_name or '\\' in npz_name + or npz_name in ('.', '..') or not isinstance(npz_sha256, str) + or not re.fullmatch('[0-9a-f]{64}', npz_sha256)): + raise ValueError('unsafe dataset bundle declaration') + npz = path.parent / npz_name + if expected_manifest_sha256 is not None: + with _open_regular_bundle(npz) as bundle: + if _stream_digest(bundle) != npz_sha256: + raise ValueError('unsafe or modified dataset bundle') + _inspect_bundle(bundle, meta, records) + parts, groups, signals, identities_seen = _decode_bundle( + bundle, meta, lookup, splits) + else: + resolved_npz = npz.resolve() + if (resolved_npz.parent != path.parent or resolved_npz.stat().st_size > MAX_NPZ + or digest(resolved_npz) != npz_sha256): + raise ValueError('unsafe or modified dataset bundle') + _inspect_bundle(resolved_npz, meta, records) + parts, groups, signals, identities_seen = _decode_bundle( + resolved_npz, meta, lookup, splits) + for a, b in (('train', 'validation'), ('train', 'test'), ('validation', 'test')): + if a in groups and b in groups and any(groups[a][i] & groups[b][i] for i in (0, 1)): + raise ValueError('room or participant leakage') + if identities_seen != {i for i, r in lookup.items() if r['split'] in splits}: + raise ValueError('incomplete retained population') + expected_counts = {s: sum(r['split'] == s for r in records) + for s in ('train', 'validation', 'test')} + if meta.get('counts') != expected_counts: + raise ValueError('split counts mismatch') + meta['manifest_sha256'] = manifest_sha256 + meta['loaded_partitions'] = list(splits) + return parts, meta + + +def _rewind(bundle): + if hasattr(bundle, 'seek'): + bundle.seek(0) + + +def _inspect_bundle(bundle, meta, records): + _rewind(bundle) + with zipfile.ZipFile(bundle) as z: infos = z.infolist() expected_names = {s + suffix + '.npy' for s in ('train', 'validation', 'test') for suffix in ('_x', '_y', '_identity')} @@ -229,8 +311,13 @@ def _load_partitions(manifest, splits, expected_manifest_sha256=None): if (shape != expected_shape or dtype != expected_dtype or fortran or f.tell() + int(np.prod(shape)) * dtype.itemsize != info.file_size): raise ValueError('invalid numpy header or payload size') + _rewind(bundle) + + +def _decode_bundle(bundle, meta, lookup, splits): parts, groups, signals, identities_seen = {}, {}, set(), set() - with np.load(npz, allow_pickle=False) as data: + _rewind(bundle) + with np.load(bundle, allow_pickle=False) as data: for split in splits: x, y, identities = (data[split + suffix] for suffix in ('_x', '_y', '_identity')) if (x.dtype != np.float32 or x.ndim != 3 or x.shape[1:] != (52, meta['window_samples']) @@ -254,18 +341,7 @@ def _load_partitions(manifest, splits, expected_manifest_sha256=None): signals.add(h) groups[split] = (set(identities[:, 0]), set(identities[:, 1])) parts[split] = {'x': x, 'y': y, 'identity': identities} - for a, b in (('train', 'validation'), ('train', 'test'), ('validation', 'test')): - if a in groups and b in groups and any(groups[a][i] & groups[b][i] for i in (0, 1)): - raise ValueError('room or participant leakage') - if identities_seen != {i for i, r in lookup.items() if r['split'] in splits}: - raise ValueError('incomplete retained population') - expected_counts = {s: sum(r['split'] == s for r in records) - for s in ('train', 'validation', 'test')} - if meta.get('counts') != expected_counts: - raise ValueError('split counts mismatch') - meta['manifest_sha256'] = manifest_sha256 - meta['loaded_partitions'] = list(splits) - return parts, meta + return parts, groups, signals, identities_seen def main(): diff --git a/scripts/tests/test_espfi.py b/scripts/tests/test_espfi.py index e40efabae..3e497fefc 100644 --- a/scripts/tests/test_espfi.py +++ b/scripts/tests/test_espfi.py @@ -1,6 +1,7 @@ """Synthetic intake boundary tests. No captured CSI enters the repository.""" import io import hashlib +import os import zipfile import json import tempfile @@ -130,6 +131,82 @@ class EspfiTests(unittest.TestCase): espfi.load_dataset(link, expected_manifest_sha256=expected) load.assert_not_called() + def test_confirmed_bundle_digest_mismatch_stops_before_decode(self): + with tempfile.TemporaryDirectory() as tmp: + manifest, _, _, meta = self.fixture(tmp) + meta['npz_sha256'] = '0' * 64 + manifest.write_text(json.dumps(meta)) + expected = espfi.digest(manifest) + with patch('csi_training.espfi.zipfile.ZipFile') as archive, \ + patch('csi_training.espfi.np.load') as load: + with self.assertRaisesRegex(ValueError, 'modified'): + espfi.load_dataset(manifest, expected_manifest_sha256=expected) + archive.assert_not_called();load.assert_not_called() + + def test_confirmed_bundle_reuses_descriptor_after_path_swap(self): + if os.name != 'posix': + self.skipTest('POSIX rename semantics required') + with tempfile.TemporaryDirectory() as tmp: + manifest, npz, _, meta = self.fixture(tmp) + expected = espfi.digest(manifest) + original_npz_sha256 = meta['npz_sha256'] + replacement = Path(tmp) / 'replacement.npz' + replacement.write_bytes(b'unverified replacement') + original_digest = espfi._stream_digest + def digest_then_swap(stream): + value = original_digest(stream) + replacement.replace(npz) + return value + with patch('csi_training.espfi._stream_digest', side_effect=digest_then_swap): + parts, provenance = espfi.load_dataset( + manifest, expected_manifest_sha256=expected) + self.assertEqual(len(parts['train']['y']), 840) + self.assertEqual(provenance['npz_sha256'], original_npz_sha256) + self.assertEqual(npz.read_bytes(), b'unverified replacement') + + def test_confirmed_bundle_uses_same_stream_for_zip_and_numpy(self): + with tempfile.TemporaryDirectory() as tmp: + manifest, _, _, _ = self.fixture(tmp) + expected = espfi.digest(manifest) + original_zip, original_load = zipfile.ZipFile, np.load + consumers = [] + def inspect(source, *args, **kwargs): + consumers.append(('zip', source)) + self.assertFalse(source.closed) + return original_zip(source, *args, **kwargs) + def decode(source, *args, **kwargs): + consumers.append(('numpy', source)) + self.assertFalse(source.closed) + return original_load(source, *args, **kwargs) + with patch('csi_training.espfi.zipfile.ZipFile', side_effect=inspect), \ + patch('csi_training.espfi.np.load', side_effect=decode): + espfi.load_dataset(manifest, expected_manifest_sha256=expected) + self.assertEqual({kind for kind, _ in consumers}, {'zip', 'numpy'}) + self.assertTrue(all(source is consumers[0][1] + for _, source in consumers)) + + def test_confirmed_bundle_rejects_nonregular_before_decode(self): + if not hasattr(os, 'mkfifo'): + self.skipTest('FIFO creation unavailable') + for kind in ('symlink', 'fifo', 'directory'): + with self.subTest(kind=kind), tempfile.TemporaryDirectory() as tmp: + manifest, npz, _, _ = self.fixture(tmp) + expected = espfi.digest(manifest) + target = Path(tmp) / 'original.npz' + npz.replace(target) + if kind == 'symlink': + npz.symlink_to(target) + elif kind == 'fifo': + os.mkfifo(npz) + else: + npz.mkdir() + with patch('csi_training.espfi.zipfile.ZipFile') as archive, \ + patch('csi_training.espfi.np.load') as load: + with self.assertRaisesRegex(ValueError, 'bundle'): + espfi.load_dataset( + manifest, expected_manifest_sha256=expected) + archive.assert_not_called();load.assert_not_called() + def test_development_never_deserializes_test_arrays(self): with tempfile.TemporaryDirectory() as tmp: manifest, _, _, _ = self.fixture(tmp)