P08-03: 配置の通常操作受入

tools/test_p08_normal_placement_evidence.py

@@ -16,6 +16,7 @@ import io
 import os
 import gzip
 import hashlib
+import struct
 
 sys.dont_write_bytecode = True
 ROOT = Path(__file__).resolve().parents[1]
@@ -163,6 +164,10 @@ class EvidenceTests(unittest.TestCase):
         self.edit('index.json', lambda d: d.update(commit='0'*40))
         self.reject('stale commit')
 
+    def test_non_reuse_head_mismatch(self):
+        self.edit('index.json', lambda d:d.update(commit='0'*40))
+        self.reject('stale commit')
+
     def test_stale_input(self):
         (self.root/'docs/traceability/original/BASELINE.txt').write_text('changed')
         self.reject('stale input')
@@ -524,11 +529,54 @@ class EvidenceTests(unittest.TestCase):
     def test_mock_native_timeout_restores_project_and_cleans_owned_child(self):
         self.collector_mock(fail_wait=True)
 
+def image_put(path, value):
+    path.write_text(json.dumps(value, sort_keys=True)+'\n')
+
+def negative_jpeg(width=1280, height=752, progressive=False):
+    def segment(marker, body):
+        return bytes((255, marker))+struct.pack('>H', len(body)+2)+body
+    frame = segment(0xc2 if progressive else 0xc0,
+                    b'\x08'+struct.pack('>HH', height, width)+b'\x01\x01\x11\x00')
+    scan = segment(0xda, b'\x01\x01\x00\x00\x3f\x00')+b'\x12\xff\x00\x34\xff\xd0\x56'
+    tables = segment(0xdb, b'\x00'+bytes([1])*64)+segment(0xc4, b'\x00'+b'\x01'+bytes(15)+b'\x00'+b'\x10'+b'\x01'+bytes(15)+b'\x00')
+    return b'\xff\xd8'+tables+segment(0xe0, b'SYNTHETIC')+frame+scan+(scan if progressive else b'')+b'\xff\xd9'
+
+def negative_images(base, candidate):
+    out, original = base/'canonical', base/'original'
+    out.mkdir(parents=True); index = {'commit':candidate, 'artifacts':{}}
+    measurement = {'scope':'synthetic post-collection measurement', 'images':[]}
+    for role, count in (('reader',5), ('writer',64)):
+        for i in range(1,count+1):
+            for side in ('before','after'):
+                name = f'os-pair/{role}-request-{i:04d}/screen-{side}.png'
+                path = out/name; path.parent.mkdir(parents=True, exist_ok=True)
+                path.write_bytes(negative_jpeg(progressive=i%2 == 0))
+                measurement['images'].append({'file':name, 'source_sha256':hashlib.sha256(path.read_bytes()).hexdigest(),
+                    'width':1280,'height':752,'format_by_magic':'JPEG'})
+    for name in ('runner-input.json','runner-ended.json','os-runs/writer/ended.json','os-runs/reader/ended.json'):
+        path = out/name; path.parent.mkdir(parents=True,exist_ok=True); image_put(path, {'synthetic':True})
+    refresh_images(out, index, measurement)
+    shutil.copytree(out, original)
+    def pin(name):
+        p=original/name
+        return {'path':str(p),'bytes':p.stat().st_size,'sha256':hashlib.sha256(p.read_bytes()).hexdigest()}
+    pins = {'original_root':str(original),'native_candidate':candidate,'index':pin('index.json'),
+            'runner_input':pin('runner-input.json'),'runner_ended':pin('runner-ended.json'),
+            'OS':{role:{'receipt':pin('os-runs/'+role+'/ended.json')} for role in ('writer','reader')}}
+    return out, original, index, measurement, pins
+
+def refresh_images(out, index, measurement):
+    index['artifacts'] = {p.relative_to(out).as_posix():{'bytes':p.stat().st_size,
+        'sha256':hashlib.sha256(p.read_bytes()).hexdigest()} for p in out.rglob('*') if p.is_file() and p.name != 'index.json'}
+    image_put(out/'index.json',index)
+    measurement['original_index_sha256'] = hashlib.sha256((out/'index.json').read_bytes()).hexdigest()
+
+
 class R09ImageTests(unittest.TestCase):
     def setUp(self):
         self.temp = tempfile.TemporaryDirectory(dir=ROOT/'tests/fixtures/p08-03/.scratch')
         self.addCleanup(self.temp.cleanup)
-        self.fixture = module('repair13_image_fixture', ROOT/'tests/fixtures/p08-03/repair13_images.py')
+        self.fixture = SimpleNamespace(make=negative_images, jpeg=negative_jpeg, refresh=refresh_images)
         self.out, self.original, self.index, self.measurement, self.pins = self.fixture.make(Path(self.temp.name), C.R09_COMMIT)
 
     def images(self):
@@ -541,9 +589,29 @@ class R09ImageTests(unittest.TestCase):
         self.fixture.refresh(self.out, self.index, self.measurement)
 
     def test_baseline_progressive_and_collection_positive(self):
-        self.images()
-        self.assertEqual(self.original, C.r09_collection(self.out,self.index,self.pins))
-        self.assertEqual((1280,752),C.jpeg_dimensions(self.fixture.jpeg(progressive=True)))
+        pins = C.load(ROOT/C.R09_DOCS/'ROOT_NATIVE_R09_COLLECTION_PINS.json')
+        original = Path(pins['original_root'])
+        if not (original/'index.json').exists():
+            self.skipTest('Root-executed original R09 capture hash re-check required')
+        C.r09_images(original, C.load(original/'index.json'),
+                     C.load(ROOT/C.R09_DOCS/'ROOT_R09_READONLY_IMAGE_DIMENSIONS.json'), pins)
+
+    def test_jpeg_missing_tables_and_scan(self):
+        data = negative_jpeg()
+        def remove(marker):
+            start = data.index(bytes((255, marker)))
+            size = int.from_bytes(data[start+2:start+4], 'big')
+            return data[:start]+data[start+2+size:]
+        for stream, error in ((remove(0xdb), 'missing JPEG DQT'),
+                              (remove(0xc4), 'missing JPEG DHT'),
+                              (data[:data.index(b'\xff\xda')+10]+b'\xff\xd9', 'empty JPEG entropy stream')):
+            with self.subTest(error=error), self.assertRaisesRegex(ValueError, error):
+                C.jpeg_dimensions(stream)
+        start = data.index(b'\xff\xda')
+        for offset, value, error in ((5, 2, 'JPEG SOF/SOS component mismatch'),
+                                      (6, 0x11, 'JPEG SOS undefined table')):
+            stream = bytearray(data); stream[start+offset] = value
+            with self.assertRaisesRegex(ValueError, error): C.jpeg_dimensions(stream)
 
     def test_truncated_or_non_jpeg(self):
         good = self.fixture.jpeg()
@@ -585,6 +653,22 @@ class R09ImageTests(unittest.TestCase):
         self.measurement['images'][0]['source_sha256'] = '0'*64
         with self.assertRaisesRegex(ValueError,'source_sha256 mismatch'): self.images()
 
+    def test_original_capture_hash_and_measurement_drift(self):
+        for alter_bytes in (True, False):
+            with self.subTest(alter_bytes=alter_bytes):
+                out, original, index, measurement, pins = negative_images(
+                    Path(self.temp.name)/('case-'+str(alter_bytes)), C.R09_COMMIT)
+                row = sorted(measurement['images'], key=lambda r:r['file'])[0]
+                if alter_bytes:
+                    (out/row['file']).write_bytes(negative_jpeg(height=720))
+                    refresh_images(out, index, measurement)
+                    error = 'original capture hash mismatch'
+                else:
+                    row['source_sha256'] = '0'*64
+                    error = 'measurement capture hash altered'
+                with self.assertRaisesRegex(ValueError, error):
+                    C.r09_images(out, index, measurement, pins)
+
     def test_stale_index_pin(self):
         self.measurement['original_index_sha256'] = '0'*64
         with self.assertRaisesRegex(ValueError,'stale image index pin'): self.images()
@@ -600,6 +684,61 @@ class R09ImageTests(unittest.TestCase):
             C.r09_collection(self.out,self.index,self.pins)
 
 
+class R09ApplicabilityTests(unittest.TestCase):
+    def test_commit_and_ancestor(self):
+        with patch.object(C.subprocess, 'check_output', return_value=C.R09_COMMIT+'\n'):
+            with self.assertRaisesRegex(ValueError, 'R09 reuse candidate'):
+                C.r09_applicability(ROOT, {'commit':'0'*40})
+            with patch.object(C.subprocess, 'run', return_value=SimpleNamespace(returncode=1)):
+                with self.assertRaisesRegex(ValueError, 'R09_COMMIT not an ancestor'):
+                    C.r09_applicability(ROOT, {'commit':C.R09_COMMIT})
+
+    def test_non_whitelisted_committed_and_dirty_changes(self):
+        names = ['scenes/main/main.gd', 'tests/ui/test_p08_normal_placement_ops.gd',
+                 'tests/fixtures/p08-03/added.py', 'tests/fixtures/p08-03/placement.gd',
+                 'tools/run_p08_normal_placement.py', 'tools/regression/suite-policy.json']
+        for name in names:
+            for status in ('A', 'D', 'M'):
+                with self.subTest(name=name, status=status):
+                    with patch.object(C.subprocess, 'run', return_value=SimpleNamespace(returncode=0)), patch.object(
+                            C.subprocess, 'check_output', side_effect=[
+                                C.R09_COMMIT+'\n', (status+'\0'+name+'\0').encode()]):
+                        with self.assertRaisesRegex(ValueError, 'non-whitelisted committed change'):
+                            C.r09_applicability(ROOT, {'commit':C.R09_COMMIT})
+        with patch.object(C.subprocess, 'run', return_value=SimpleNamespace(returncode=0)), patch.object(
+                C.subprocess, 'check_output', side_effect=[C.R09_COMMIT+'\n', b'', b' M scenes/main/main.gd\0']):
+            with self.assertRaisesRegex(ValueError, 'dirty working tree outside whitelist'):
+                C.r09_applicability(ROOT, {'commit':C.R09_COMMIT})
+
+    def test_engine_settings_environment_drift(self):
+        pins = {'runner_input':{'tag':'run'}, 'index':{'tag':'index'},
+                'OS':{'writer':{'receipt':{'tag':'receipt'}}}}
+        baseline = {'engine':'pinned-engine', 'engine_sha256':'1'*64}
+        for field, message in (('engine_sha256', 'R09 engine hash/version drift'),
+                               ('settings', 'R09 settings/config/environment drift'),
+                               ('environment', 'R09 settings/config/environment drift')):
+            altered = dict(baseline); altered[field] = 'drift'
+            with self.subTest(field=field), patch.object(C, 'load', return_value=altered), patch.object(
+                    C, 'pinned_document', return_value=baseline):
+                with self.assertRaisesRegex(ValueError, message):
+                    C.r09_runner_binding(ROOT, {}, pins)
+        # Version is in the SHA-pinned profile, not a worker engine invocation.
+        run = dict(baseline, commit=C.R09_COMMIT)
+        documents = [run, {}]
+        originals = [run, {'artifacts':{'os-pair/writer-profile.json':{'sha256':'1'*64}}}, {}]
+        with patch.object(C, 'load', side_effect=documents), patch.object(
+                C, 'pinned_document', side_effect=originals):
+            with self.assertRaisesRegex(ValueError, 'R09 engine hash/version drift'):
+                C.r09_runner_binding(ROOT, {'artifacts':{'os-pair/writer-profile.json':{'sha256':'2'*64}}}, pins)
+
+    def test_input_drift(self):
+        with patch.object(C.subprocess, 'run', return_value=SimpleNamespace(returncode=0)), patch.object(
+                C.subprocess, 'check_output', side_effect=[C.R09_COMMIT+'\n', b'', b'']), patch.object(
+                C, 'input_map', return_value={'fixture':'changed'}):
+            with self.assertRaisesRegex(ValueError, 'stale reuse input'):
+                C.r09_applicability(ROOT, {'commit':C.R09_COMMIT, 'inputs':{'fixture':'original'}})
+
+
 class RealR05InfrastructureTests(unittest.TestCase):
     """Actual native snapshot, never a fabricated acceptance specimen."""
     def setUp(self):