P08-03: 配置の通常操作受入
This commit is contained in:
@@ -144,6 +144,93 @@ def input_map(root):
|
||||
if p.is_file() and '__pycache__' not in p.parts and '.scratch' not in p.parts]
|
||||
return {name: sha(root/name) for name in sorted(names)}
|
||||
|
||||
R09_COMMIT = '31aa7706d28a92e313c432d5a62e35d19503dcc3'
|
||||
R09_DOCS = 'docs/plans/original-completion-2026-10-02/execution/p08-03'
|
||||
|
||||
def jpeg_dimensions(data):
|
||||
"""Parse baseline/progressive frame headers, including multi-scan markers.
|
||||
|
||||
This is a structural/dimension check, not a pixel decoder.
|
||||
"""
|
||||
require(data[:2] == b'\xff\xd8' and data[-2:] == b'\xff\xd9', 'invalid JPEG SOI/EOI')
|
||||
pos, dimensions, scans = 2, None, 0
|
||||
entropy = False
|
||||
while pos < len(data):
|
||||
if entropy:
|
||||
pos = data.find(b'\xff', pos)
|
||||
require(pos >= 0, 'truncated JPEG scan')
|
||||
require(data[pos] == 255, 'invalid JPEG marker')
|
||||
while pos < len(data) and data[pos] == 255:
|
||||
pos += 1
|
||||
require(pos < len(data), 'truncated JPEG marker')
|
||||
marker = data[pos]; pos += 1
|
||||
if entropy and (marker == 0 or 0xd0 <= marker <= 0xd7):
|
||||
continue
|
||||
entropy = False
|
||||
if marker == 0xd9:
|
||||
require(pos == len(data) and dimensions is not None and scans > 0, 'invalid JPEG end')
|
||||
return dimensions
|
||||
require(marker not in (0, 0xd8, 1) and not 0xd0 <= marker <= 0xd7,
|
||||
'invalid JPEG standalone marker')
|
||||
require(pos+2 <= len(data), 'truncated JPEG length')
|
||||
length = int.from_bytes(data[pos:pos+2], 'big')
|
||||
require(length >= 2 and pos+length <= len(data)-2, 'truncated JPEG segment')
|
||||
body = data[pos+2:pos+length]; pos += length
|
||||
if marker in (0xc0, 0xc2):
|
||||
require(dimensions is None and len(body) >= 6 and body[0] == 8
|
||||
and body[5] > 0 and len(body) == 6+3*body[5], 'invalid JPEG SOF')
|
||||
height, width = struct.unpack('>HH', body[1:5])
|
||||
require(width > 0 and height > 0, 'invalid JPEG dimensions')
|
||||
dimensions = (width, height)
|
||||
elif 0xc0 <= marker <= 0xcf and marker not in (0xc4, 0xc8, 0xcc):
|
||||
raise ValueError('unsupported JPEG SOF')
|
||||
elif marker == 0xda:
|
||||
require(dimensions is not None and body and body[0] > 0
|
||||
and len(body) == 4+2*body[0], 'invalid JPEG SOS')
|
||||
scans += 1; entropy = True
|
||||
raise ValueError('missing JPEG EOI')
|
||||
|
||||
def r09_images(evidence, index, measurement):
|
||||
require(measurement['original_index_sha256'] == sha(evidence/'index.json'), 'stale image index pin')
|
||||
expected = {f'os-pair/{role}-request-{i:04d}/screen-{side}.png'
|
||||
for role, count in (('reader', 5), ('writer', 64))
|
||||
for i in range(1, count+1) for side in ('before', 'after')}
|
||||
rows = keyed(measurement['images'], 'file')
|
||||
indexed = {n for n in index['artifacts'] if n.startswith('os-pair/') and n.endswith('.png')
|
||||
and n not in ('os-pair/formal/thumbnail.png', 'os-pair/formal.bak/thumbnail.png')}
|
||||
require(set(rows) == indexed == expected, 'missing/extra image entry')
|
||||
sizes = {}
|
||||
for name in sorted(expected):
|
||||
data = (evidence/name).read_bytes()
|
||||
sizes[name] = jpeg_dimensions(data)
|
||||
row, pin = rows[name], index['artifacts'][name]
|
||||
require(hashlib.sha256(data).hexdigest() == row['source_sha256'] == pin['sha256']
|
||||
and len(data) == pin['bytes'], 'image source_sha256 mismatch')
|
||||
require(row['format_by_magic'] == 'JPEG' and type(row['width']) is int
|
||||
and type(row['height']) is int and sizes[name] == (row['width'], row['height']),
|
||||
'image SOF dimension mismatch')
|
||||
for name in expected:
|
||||
if name.endswith('screen-before.png'):
|
||||
require(sizes[name] == sizes[name.replace('screen-before.png', 'screen-after.png')],
|
||||
'before/after dimension drift')
|
||||
|
||||
def r09_collection(evidence, index, pins):
|
||||
original = Path(pins['original_root'])
|
||||
require(original.is_absolute() and pins['native_candidate'] == index['commit'] == R09_COMMIT,
|
||||
'R09 reuse candidate')
|
||||
pinned = [pins['index'], pins['runner_input'], pins['runner_ended']]
|
||||
pinned += [pins['OS'][role]['receipt'] for role in ('writer', 'reader')]
|
||||
for pin in pinned:
|
||||
path = Path(pin['path'])
|
||||
require(path.is_relative_to(original) and not path.is_symlink(), 'collection pin path')
|
||||
require(path.stat().st_size == pin['bytes'] and sha(path) == pin['sha256'], 'collection pin SHA')
|
||||
for name in ['index.json', *index['artifacts']]:
|
||||
require(not Path(name).is_absolute() and '..' not in Path(name).parts, 'collection unsafe path')
|
||||
a, b = original/name, evidence/name
|
||||
require(not a.is_symlink() and a.resolve().is_relative_to(original.resolve())
|
||||
and a.read_bytes() == b.read_bytes(), 'original/canonical byte divergence '+name)
|
||||
return original
|
||||
|
||||
def keyed(rows, key):
|
||||
result = {r[key]: r for r in rows}
|
||||
require(len(result) == len(rows), 'duplicate '+key)
|
||||
@@ -376,12 +463,30 @@ def check(project_root, evidence_dir):
|
||||
require(index['collection_error'] is None, 'collection failed')
|
||||
require(index['conditions'] == CONDITIONS, 'missing condition')
|
||||
actual = {n: v['sha256'] for n, v in runner.source_refs(root).items()}
|
||||
original = evidence
|
||||
pins_path = root/R09_DOCS/'ROOT_NATIVE_R09_COLLECTION_PINS.json'
|
||||
if pins_path.exists():
|
||||
original = r09_collection(evidence, index, load(pins_path))
|
||||
r09_images(evidence, index, load(root/R09_DOCS/'ROOT_R09_READONLY_IMAGE_DIMENSIONS.json'))
|
||||
# Only these Python validator/test files may differ from same31aa.
|
||||
changed = {'tools/check_p08_normal_placement_evidence.py',
|
||||
'tools/test_p08_normal_placement_evidence.py'}
|
||||
added = {'tests/fixtures/p08-03/repair13_images.py'}
|
||||
for name in changed:
|
||||
baseline = subprocess.check_output(['git', '-C', str(root), 'show', R09_COMMIT+':'+name])
|
||||
require(index['source_map'][name] == hashlib.sha256(baseline).hexdigest(), 'reuse source pin')
|
||||
require({n:v for n,v in actual.items() if n not in changed|added} ==
|
||||
{n:v for n,v in index['source_map'].items() if n not in changed}, 'stale reuse source')
|
||||
require({n:v for n,v in input_map(root).items() if n not in added} == index['inputs'], 'stale reuse input')
|
||||
actual = index['source_map']
|
||||
require(index['source_map'] == actual, 'stale source')
|
||||
require(index['inputs'] == input_map(root), 'stale input')
|
||||
require(index['inputs'] == {n:v for n,v in input_map(root).items()
|
||||
if not pins_path.exists() or n != 'tests/fixtures/p08-03/repair13_images.py'}, 'stale input')
|
||||
require(index['spec_sha256'] == SPEC_SHA == hashlib.sha256((root/SPEC).read_text().strip().encode()).hexdigest(), 'immutable spec')
|
||||
require(index['marker_contract_sha256'] == sha(root/CONTRACT), 'stale marker')
|
||||
require(re.fullmatch('[0-9a-f]{40}', index['commit']) is not None, 'commit')
|
||||
require(index['commit'] == subprocess.check_output(['git','-C',str(root),'rev-parse','HEAD'],text=True).strip(), 'stale commit')
|
||||
if not pins_path.exists():
|
||||
require(index['commit'] == subprocess.check_output(['git','-C',str(root),'rev-parse','HEAD'],text=True).strip(), 'stale commit')
|
||||
def artifact(name):
|
||||
require(isinstance(name, str) and name and not Path(name).is_absolute() and '..' not in Path(name).parts, 'unsafe/missing artifact')
|
||||
p = evidence/name
|
||||
@@ -451,7 +556,7 @@ def check(project_root, evidence_dir):
|
||||
require({n:v['sha256'] for n,v in end['source_after'].items()} == actual, 'OS source changed')
|
||||
require(end['profile_before'] == {'existed':False,'files':{}}, 'OS profile not fresh')
|
||||
owners = module('p08_verify_os_owners', HERE/'regression/ordinary_proof_gate.py')
|
||||
actual_owner_proof = owners.ordinary_gate(evidence/prefix/'teardown', pid, evidence/prefix, ['single'])
|
||||
actual_owner_proof = owners.ordinary_gate(original/prefix/'teardown', pid, original/prefix, ['single'])
|
||||
require(end['ordinary_owner_proof'] == actual_owner_proof, 'OS ordinary owner proof mismatch')
|
||||
profile = document('os-pair/'+end['role']+'-profile.json')
|
||||
require(profile['pid'] == pid and Path(profile['user_data_dir']).resolve() == Path(end['profile']).resolve()
|
||||
@@ -667,7 +772,12 @@ def check(project_root, evidence_dir):
|
||||
require(raw['events'] and observed['events'], 'missing OS events')
|
||||
require(raw['collector_pid'] > 0 and raw['collector_pid'] not in os_pids, 'fake OS collector')
|
||||
for key in ('screen_before', 'screen_after'):
|
||||
png(artifact(ack[key]).read_bytes())
|
||||
data = artifact(ack[key]).read_bytes()
|
||||
if pins_path.exists():
|
||||
require(ack[key] == folder+key.replace('_', '-')+'.png', 'R09 screenshot request binding')
|
||||
jpeg_dimensions(data)
|
||||
else:
|
||||
png(data)
|
||||
if request['action'] == 'cell' and 'position' in row:
|
||||
dispatch_gate(row['position'], observed['events'])
|
||||
if row['id'] == 'pan_zoom':
|
||||
|
||||
@@ -524,6 +524,82 @@ class EvidenceTests(unittest.TestCase):
|
||||
def test_mock_native_timeout_restores_project_and_cleans_owned_child(self):
|
||||
self.collector_mock(fail_wait=True)
|
||||
|
||||
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.out, self.original, self.index, self.measurement, self.pins = self.fixture.make(Path(self.temp.name), C.R09_COMMIT)
|
||||
|
||||
def images(self):
|
||||
C.r09_images(self.out, self.index, self.measurement)
|
||||
|
||||
def change_bytes(self, data):
|
||||
row = self.measurement['images'][0]
|
||||
(self.out/row['file']).write_bytes(data)
|
||||
row['source_sha256'] = hashlib.sha256(data).hexdigest()
|
||||
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)))
|
||||
|
||||
def test_truncated_or_non_jpeg(self):
|
||||
good = self.fixture.jpeg()
|
||||
for data in (good[:-1], b'NOT JPEG', b'\xff\xd8\xff\xe0\xff\xff\xff\xd9',
|
||||
good[:15]+b'\xff\xd9', b'\xff\xd8\xff\xd9'):
|
||||
with self.subTest(data=data):
|
||||
self.change_bytes(data)
|
||||
with self.assertRaisesRegex(ValueError, 'JPEG'):
|
||||
self.images()
|
||||
|
||||
def test_sof_dimension_mismatch(self):
|
||||
self.change_bytes(self.fixture.jpeg(height=720))
|
||||
with self.assertRaisesRegex(ValueError,'SOF dimension mismatch'): self.images()
|
||||
|
||||
def test_before_after_dimension_drift(self):
|
||||
self.change_bytes(self.fixture.jpeg(height=753))
|
||||
self.measurement['images'][0]['height'] = 753
|
||||
with self.assertRaisesRegex(ValueError,'before/after dimension drift'): self.images()
|
||||
|
||||
def test_missing_image_entry(self):
|
||||
self.measurement['images'].pop()
|
||||
with self.assertRaisesRegex(ValueError,'missing/extra image entry'): self.images()
|
||||
|
||||
def test_extra_image_entry(self):
|
||||
row = dict(self.measurement['images'][0],file='os-pair/extra.png')
|
||||
self.measurement['images'].append(row)
|
||||
with self.assertRaisesRegex(ValueError,'missing/extra image entry'): self.images()
|
||||
|
||||
def test_duplicate_image_entry(self):
|
||||
self.measurement['images'].append(self.measurement['images'][0])
|
||||
with self.assertRaisesRegex(ValueError,'duplicate file'): self.images()
|
||||
|
||||
def test_extra_indexed_image(self):
|
||||
(self.out/'os-pair/extra.png').write_bytes(self.fixture.jpeg())
|
||||
self.fixture.refresh(self.out,self.index,self.measurement)
|
||||
with self.assertRaisesRegex(ValueError,'missing/extra image entry'): self.images()
|
||||
|
||||
def test_source_sha256_mismatch(self):
|
||||
self.measurement['images'][0]['source_sha256'] = '0'*64
|
||||
with self.assertRaisesRegex(ValueError,'source_sha256 mismatch'): self.images()
|
||||
|
||||
def test_stale_index_pin(self):
|
||||
self.measurement['original_index_sha256'] = '0'*64
|
||||
with self.assertRaisesRegex(ValueError,'stale image index pin'): self.images()
|
||||
|
||||
def test_original_canonical_divergence(self):
|
||||
(self.out/self.measurement['images'][0]['file']).write_bytes(self.fixture.jpeg(height=720))
|
||||
with self.assertRaisesRegex(ValueError,'original/canonical byte divergence'):
|
||||
C.r09_collection(self.out,self.index,self.pins)
|
||||
|
||||
def test_collection_pin_mismatch(self):
|
||||
self.pins['runner_input']['sha256'] = '0'*64
|
||||
with self.assertRaisesRegex(ValueError,'collection pin SHA'):
|
||||
C.r09_collection(self.out,self.index,self.pins)
|
||||
|
||||
|
||||
class RealR05InfrastructureTests(unittest.TestCase):
|
||||
"""Actual native snapshot, never a fabricated acceptance specimen."""
|
||||
def setUp(self):
|
||||
|
||||
Reference in New Issue
Block a user