diff --git a/docs/adr/ADR-427-enforced-confirmation-training-mode.md b/docs/adr/ADR-427-enforced-confirmation-training-mode.md new file mode 100644 index 000000000..282023f42 --- /dev/null +++ b/docs/adr/ADR-427-enforced-confirmation-training-mode.md @@ -0,0 +1,39 @@ +# ADR 427: Enforced independent-confirmation training mode + +Status: accepted for draft research implementation; no model promotion. + +## Context + +ADR 425 added a strict metadata preflight, but it remained advisory and was not +bound to the dataset manifest consumed by a trainer. A passing metadata document +could therefore be paired with different dataset bytes. The predictive paths +were rejected by ADR 424, while the temporal encoder measures only a proxy +embedding objective; neither is an appropriate confirmation entrypoint. + +## Decision + +Add an explicit, opt-in confirmation mode only to the supervised amplitude +trainer. The mode requires a complete metadata plan, exposure ledger, frozen +protocol SHA-256 and frozen dataset-manifest SHA-256. Before importing the +dataset loader, decoding data, creating output, or training, the gate validates +bounded regular files, leakage declarations, exposure history, coverage, and +both frozen hashes. Invalid input exits 1; valid but blocked evidence exits 2. +The dataset loader's returned manifest digest is checked again before training. +Confirmation mode cannot optimize an existing bundle and every receipt remains +`promotionAuthorized: false`. + +Legacy research operation remains available without implying confirmation. +Current ESP-Fi is in the known-exposure set, so it cannot pass as independent +confirmation. Rollback removes the four gate flags and helper without changing +the legacy training path. + +## Consequences + +This closes accidental gate bypass within the selected mode and binds the +declared review to exact manifest bytes. It does not authenticate captures, +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. A concurrent manifest change +may be decoded before the loader's digest mismatch aborts, although it cannot +reach training. Passing software tests makes no sensing-accuracy claim. + diff --git a/scripts/csi_training/confirmation.py b/scripts/csi_training/confirmation.py index 0b3ffdb1f..2fc75eec9 100644 --- a/scripts/csi_training/confirmation.py +++ b/scripts/csi_training/confirmation.py @@ -7,6 +7,7 @@ import re import stat MAX_BYTES = 2 * 1024 * 1024 +MAX_DATASET_MANIFEST_BYTES = 3_000_000 MAX_RECORDS = 4096 SPLITS = ('train', 'development', 'confirmation') AXES = ('room', 'participant', 'session') @@ -126,6 +127,54 @@ def read_json(path): return json.loads(raw, object_pairs_hook=_pairs, parse_constant=invalid) +def regular_sha256(path, maximum=MAX_DATASET_MANIFEST_BYTES): + """Hash one bounded regular file through the descriptor that was checked.""" + if (type(maximum) is not int or not 1 <= maximum <= 16 * 1024 * 1024 + or not hasattr(os, 'O_NOFOLLOW') or not hasattr(os, 'O_NONBLOCK')): + raise ValueError('Safe file hashing unavailable') + fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) + digest_value = hashlib.sha256() + with os.fdopen(fd, 'rb') as handle: + info = os.fstat(handle.fileno()) + if not stat.S_ISREG(info.st_mode) or info.st_size > maximum: + raise ValueError('Expected bounded regular file') + remaining = info.st_size + while remaining: + block = handle.read(min(1024 * 1024, remaining)) + if not block: + raise ValueError('File changed during hashing') + digest_value.update(block) + remaining -= len(block) + if handle.read(1): + raise ValueError('File changed during hashing') + return digest_value.hexdigest() + + +def training_gate(manifest_path, ledger_path, dataset_manifest_path, + expected_protocol_sha256, expected_dataset_manifest_sha256): + """Evaluate declared metadata and bind it to frozen protocol/dataset bytes. + + This remains a metadata gate. It does not authenticate captures, consent, + licensing, source aliases, or the completeness of the exposure ledger. + """ + digest(expected_protocol_sha256) + digest(expected_dataset_manifest_sha256) + report = evaluate(read_json(manifest_path), read_json(ledger_path)) + reasons = set(report['reasons']) + if report['protocol_sha256'] != expected_protocol_sha256: + reasons.add('protocol_digest_mismatch') + actual = regular_sha256(dataset_manifest_path) + if actual != expected_dataset_manifest_sha256: + reasons.add('dataset_manifest_digest_mismatch') + result = dict(report) + result.update({'preflight_passed': not reasons, + 'reasons': sorted(reasons), + 'dataset_manifest_sha256': actual, + 'expected_dataset_manifest_sha256': expected_dataset_manifest_sha256, + 'scope': report['scope'] + ' Dataset bytes are hash-pinned; provenance claims remain unauthenticated.'}) + return result + + def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--manifest', required=True) diff --git a/scripts/csi_training/espfi_train.py b/scripts/csi_training/espfi_train.py index 476b4b2cf..9f5286d46 100644 --- a/scripts/csi_training/espfi_train.py +++ b/scripts/csi_training/espfi_train.py @@ -363,9 +363,37 @@ def main(argv=None): parser.add_argument('--seed', type=int, default=42) parser.add_argument('--max-seconds', type=int, default=600) parser.add_argument('--optimize-bundle', help='Existing statistics bundle; inference only, no training or test evaluation') + parser.add_argument('--confirmation-manifest', help='Independent-confirmation metadata plan (enables enforced gate)') + parser.add_argument('--exposure-ledger', help='Previously exposed dataset revisions for the confirmation gate') + parser.add_argument('--expected-protocol-sha256', help='Frozen protocol digest for the confirmation gate') + parser.add_argument('--expected-dataset-manifest-sha256', help='Frozen digest of --manifest bytes') args = parser.parse_args(argv) + gate_names = ('confirmation_manifest', 'exposure_ledger', 'expected_protocol_sha256', + 'expected_dataset_manifest_sha256') + gate_values = [getattr(args, name) for name in gate_names] + gate_enabled = all(gate_values) + if (any(gate_values) and not gate_enabled) or (gate_enabled and args.optimize_bundle): + print(json.dumps({'status': 'invalid_input', 'promotionAuthorized': False})) + raise SystemExit(1) + preflight = None + if gate_enabled: + from .confirmation import training_gate + try: + preflight = training_gate(args.confirmation_manifest, args.exposure_ledger, + args.manifest, args.expected_protocol_sha256, + args.expected_dataset_manifest_sha256) + except (ValueError, OSError, TypeError, RecursionError, OverflowError): + print(json.dumps({'status': 'invalid_input', 'promotionAuthorized': False})) + raise SystemExit(1) + if not preflight['preflight_passed']: + print(json.dumps(preflight, sort_keys=True)) + raise SystemExit(2) from .espfi import load_dataset partitions, provenance = load_dataset(args.manifest) + if preflight: + if provenance.get('manifest_sha256') != preflight['dataset_manifest_sha256']: + raise ValueError('Dataset manifest changed after confirmation preflight') + provenance = dict(provenance, confirmation_preflight=preflight) if args.optimize_bundle: parent = json.loads((Path(args.optimize_bundle) / 'metrics.json').read_text()) if provenance['manifest_sha256'] != parent['provenance']['manifest_sha256']: diff --git a/scripts/tests/test_csi_training_confirmation.py b/scripts/tests/test_csi_training_confirmation.py index 7efb0b1d9..66d6e6132 100644 --- a/scripts/tests/test_csi_training_confirmation.py +++ b/scripts/tests/test_csi_training_confirmation.py @@ -7,7 +7,7 @@ import subprocess import sys import tempfile import unittest -from csi_training.confirmation import evaluate, read_json, KNOWN_EXPOSED +from csi_training.confirmation import evaluate, read_json, regular_sha256, training_gate, KNOWN_EXPOSED def fixture(): @@ -92,4 +92,28 @@ class ConfirmationTests(unittest.TestCase): self.assertEqual(r.returncode,expected,r.stderr) self.assertFalse(json.loads(r.stdout)['promotionAuthorized']) + def test_training_gate_binds_frozen_protocol_and_dataset(self): + with tempfile.TemporaryDirectory() as d: + m,l=fixture();root=Path(d);dataset=root/'dataset.json' + p=root/'m.json';q=root/'l.json' + dataset.write_text('{"schema_version":1}') + p.write_text(json.dumps(m));q.write_text(json.dumps(l)) + frozen=regular_sha256(dataset) + report=training_gate(p,q,dataset,'a'*64,frozen) + self.assertTrue(report['preflight_passed']) + self.assertFalse(report['promotionAuthorized']) + self.assertNotIn('records',report) + self.assertIn('protocol_digest_mismatch',training_gate(p,q,dataset,'b'*64,frozen)['reasons']) + self.assertIn('dataset_manifest_digest_mismatch',training_gate(p,q,dataset,'a'*64,'0'*64)['reasons']) + + def test_regular_sha256_rejects_unsafe_files(self): + with tempfile.TemporaryDirectory() as d: + root=Path(d);target=root/'target';target.write_bytes(b'ok') + link=root/'link';link.symlink_to(target) + with self.assertRaises(OSError):regular_sha256(link) + fifo=root/'fifo';os.mkfifo(fifo) + with self.assertRaises(ValueError):regular_sha256(fifo) + oversized=root/'oversized';oversized.write_bytes(b'x'*33) + with self.assertRaises(ValueError):regular_sha256(oversized,32) + if __name__=='__main__':unittest.main() diff --git a/scripts/tests/test_espfi_train.py b/scripts/tests/test_espfi_train.py index 7e0a6e3b2..c4db683ee 100644 --- a/scripts/tests/test_espfi_train.py +++ b/scripts/tests/test_espfi_train.py @@ -1,15 +1,19 @@ """Synthetic integrity tests, not RF sensing accuracy evidence.""" import copy import json +import os from pathlib import Path import sys import tempfile import unittest +from unittest.mock import patch import numpy as np import torch sys.path.insert(0, str(Path(__file__).resolve().parents[1])) -from csi_training.espfi_train import AmplitudeBundle, AmplitudeClassifier, optimize, optimize_bundle, scores +from csi_training.espfi_train import AmplitudeBundle, AmplitudeClassifier, main, optimize, optimize_bundle, scores +from csi_training.confirmation import regular_sha256 +from scripts.tests.test_csi_training_confirmation import fixture as confirmation_fixture def fixture(): @@ -107,5 +111,110 @@ class TrainerTests(unittest.TestCase): AmplitudeBundle(path) +class ConfirmationGateIntegrationTests(unittest.TestCase): + def gate_files(self, root): + plan, ledger = confirmation_fixture() + dataset = root / 'dataset.json' + plan_path, ledger_path = root / 'plan.json', root / 'ledger.json' + dataset.write_text('{"schema_version":1}') + plan_path.write_text(json.dumps(plan)) + ledger_path.write_text(json.dumps(ledger)) + return dataset, plan_path, ledger_path + + def argv(self, root, dataset, plan, ledger): + return ['--manifest', str(dataset), '--output', str(root / 'out'), + '--confirmation-manifest', str(plan), '--exposure-ledger', str(ledger), + '--expected-protocol-sha256', 'a' * 64, + '--expected-dataset-manifest-sha256', regular_sha256(dataset)] + + def test_blocked_gate_never_loads_or_trains(self): + with tempfile.TemporaryDirectory() as d: + root=Path(d);dataset,plan,ledger=self.gate_files(root) + value=json.loads(plan.read_text());value['records'][-1]['room']=value['records'][0]['room'] + plan.write_text(json.dumps(value)) + with patch('csi_training.espfi.load_dataset') as load, patch('csi_training.espfi_train.optimize') as train: + with self.assertRaisesRegex(SystemExit, '2'): + main(self.argv(root,dataset,plan,ledger)) + load.assert_not_called();train.assert_not_called() + self.assertFalse((root/'out').exists()) + + def test_invalid_partial_and_hash_mismatch_never_load(self): + with tempfile.TemporaryDirectory() as d: + root=Path(d);dataset,plan,ledger=self.gate_files(root) + cases=[self.argv(root,dataset,plan,ledger)[:-2], + self.argv(root,dataset,plan,ledger)[:-1]+['0'*64]] + plan.write_text('{bad json') + cases.append(self.argv(root,dataset,plan,ledger)) + for args in cases: + with self.subTest(args=args[-2:]), patch('csi_training.espfi.load_dataset') as load, patch('csi_training.espfi_train.optimize') as train: + with self.assertRaises(SystemExit):main(args) + load.assert_not_called();train.assert_not_called() + self.assertFalse((root/'out').exists()) + + def test_unsafe_gate_files_never_load(self): + with tempfile.TemporaryDirectory() as d: + root=Path(d);dataset,plan,ledger=self.gate_files(root) + valid_plan=plan.read_bytes() + unsafe=[] + plan.unlink();plan.symlink_to(ledger);unsafe.append(plan) + for candidate in unsafe: + with patch('csi_training.espfi.load_dataset') as load: + with self.assertRaisesRegex(SystemExit,'1'): + main(self.argv(root,dataset,candidate,ledger)) + load.assert_not_called() + plan.unlink();os.mkfifo(plan) + with patch('csi_training.espfi.load_dataset') as load: + with self.assertRaisesRegex(SystemExit,'1'): + main(self.argv(root,dataset,plan,ledger)) + load.assert_not_called() + plan.unlink();plan.write_bytes(valid_plan+b' '*(2*1024*1024)) + with patch('csi_training.espfi.load_dataset') as load: + with self.assertRaisesRegex(SystemExit,'1'): + main(self.argv(root,dataset,plan,ledger)) + load.assert_not_called() + + def test_protocol_mismatch_never_loads(self): + with tempfile.TemporaryDirectory() as d: + root=Path(d);dataset,plan,ledger=self.gate_files(root) + args=self.argv(root,dataset,plan,ledger) + args[args.index('--expected-protocol-sha256')+1]='b'*64 + with patch('csi_training.espfi.load_dataset') as load, patch('csi_training.espfi_train.optimize') as train: + with self.assertRaisesRegex(SystemExit,'2'):main(args) + load.assert_not_called();train.assert_not_called() + + def test_manifest_race_stops_before_training(self): + with tempfile.TemporaryDirectory() as d: + root=Path(d);dataset,plan,ledger=self.gate_files(root) + parts,provenance=fixture();provenance['manifest_sha256']='f'*64 + with patch('csi_training.espfi.load_dataset',return_value=(parts,provenance)) as load, \ + patch('csi_training.espfi_train.optimize') as train: + with self.assertRaisesRegex(ValueError,'changed'): + main(self.argv(root,dataset,plan,ledger)) + load.assert_called_once();train.assert_not_called() + self.assertFalse((root/'out').exists()) + + def test_gate_rejects_bundle_optimization_before_loading(self): + with tempfile.TemporaryDirectory() as d: + root=Path(d);dataset,plan,ledger=self.gate_files(root) + args=self.argv(root,dataset,plan,ledger)+['--optimize-bundle',str(root/'bundle')] + with patch('csi_training.espfi.load_dataset') as load, patch('csi_training.espfi_train.optimize_bundle') as optimize_existing: + with self.assertRaisesRegex(SystemExit,'1'):main(args) + load.assert_not_called();optimize_existing.assert_not_called() + + def test_passing_gate_attaches_sanitized_provenance(self): + with tempfile.TemporaryDirectory() as d: + root=Path(d);dataset,plan,ledger=self.gate_files(root) + parts,provenance=fixture();provenance['manifest_sha256']=regular_sha256(dataset) + report={'variant':'statistics','test':{},'latency':{},'promotionAuthorized':False} + with patch('csi_training.espfi.load_dataset',return_value=(parts,provenance)) as load, \ + patch('csi_training.espfi_train.optimize',return_value=report) as train: + main(self.argv(root,dataset,plan,ledger)) + load.assert_called_once();train.assert_called_once() + attached=train.call_args.args[1]['confirmation_preflight'] + self.assertTrue(attached['preflight_passed']) + self.assertFalse(attached['promotionAuthorized']) + self.assertNotIn('records',attached) + + if __name__ == '__main__': unittest.main()