67 lines
4.8 KiB
Python
67 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Pinned spike-ihex adapter with checksummed address provenance and explicit node4 coverage."""
|
|
import hashlib,json,pathlib,subprocess,sys
|
|
|
|
def address_inventory(path):
|
|
base=0;rows=[];eof=False;encrypted=False;entries=[];address_records=[]
|
|
if path.stat().st_size>64*1024*1024:raise ValueError('Intel HEX input exceeds bound')
|
|
for number,line in enumerate(path.read_text('ascii').splitlines(),1):
|
|
if not line.strip():continue
|
|
if eof:raise ValueError('record after EOF')
|
|
if not line.startswith(':'):raise ValueError('invalid Intel HEX record')
|
|
raw=bytes.fromhex(line[1:])
|
|
if len(raw)<5 or raw[0]+5!=len(raw) or sum(raw)&255:raise ValueError('invalid Intel HEX count/checksum')
|
|
count=raw[0];address=int.from_bytes(raw[1:3],'big');kind=raw[3];data=raw[4:-1]
|
|
if kind in (2,4):
|
|
if count!=2 or address:raise ValueError('invalid extended address')
|
|
base=int.from_bytes(data,'big')<<(16 if kind==4 else 4)
|
|
address_records.append({'line':number,'type':kind,'base':base})
|
|
elif kind==0:
|
|
start=base+address
|
|
if start+count>64*1024*1024:raise ValueError('firmware address exceeds bounded upstream buffer')
|
|
if len(rows)>=1000000:raise ValueError('firmware record count exceeds bound')
|
|
rows.append({'line':number,'address':start,'bytes':count})
|
|
elif kind==1:
|
|
if count or address:raise ValueError('invalid EOF')
|
|
eof=True
|
|
elif kind in (6,7):encrypted=True
|
|
elif kind in (3,5):
|
|
if count!=4 or address:raise ValueError('invalid entry point record')
|
|
entry=(int.from_bytes(data[:2],'big')<<4)+int.from_bytes(data[2:],'big') if kind==3 else int.from_bytes(data,'big')
|
|
entries.append({'line':number,'type':kind,'address':entry})
|
|
else:raise ValueError('unsupported address/record type')
|
|
if not eof or not rows:raise ValueError('incomplete Intel HEX')
|
|
ordered=sorted(rows,key=lambda r:r['address'])
|
|
if any(a['address']+a['bytes']>b['address'] for a,b in zip(ordered,ordered[1:])):raise ValueError('overlapping firmware records')
|
|
return {'entry_points':entries,'address_records':address_records,'records':rows,'minimum_address':min(r['address'] for r in rows),'maximum_address':max(r['address']+r['bytes'] for r in rows),'encrypted':encrypted}
|
|
|
|
def run(request):
|
|
if request['protocol']!=1:raise ValueError('unsupported protocol')
|
|
settings=request['settings'];tool=pathlib.Path(settings['tool']);source=pathlib.Path(request['input_dir']);output=pathlib.Path(request['output_dir'])
|
|
with tool.open('rb') as f:digest=hashlib.file_digest(f,'sha256').hexdigest()
|
|
if digest!=settings['expected_sha256']:raise ValueError('node tool differs from pinned executable')
|
|
evidence=[];used=0
|
|
for path in sorted(source.rglob('*.hex')):
|
|
if path.is_symlink() or not path.is_file():continue
|
|
rel=path.relative_to(source).as_posix()
|
|
with path.open('rb') as f:identity=hashlib.file_digest(f,'sha256').hexdigest()
|
|
row={'source':rel,'source_sha256':identity,'processing_revision':2,'tool_sha256':digest}
|
|
if path.name.startswith('node4-'):
|
|
row.update(status='unsupported',reason='Pinned spike-ihex does not correctly decode node4-* firmware');evidence.append(row);continue
|
|
try:
|
|
metadata=address_inventory(path)
|
|
dest=output/'firmware'/f'{rel}.bin';dest.parent.mkdir(parents=True,exist_ok=True)
|
|
subprocess.run([str(tool),'--input',str(path),'--output',str(dest)],check=True,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,timeout=60)
|
|
if not dest.is_file() or not dest.stat().st_size:raise ValueError('decoder did not write firmware')
|
|
used+=dest.stat().st_size
|
|
if used>request['workspace_bytes']//2:raise ValueError('node output exceeds budget')
|
|
row.update(status='decoded',output=dest.relative_to(output).as_posix(),address_metadata=metadata,
|
|
output_origin=metadata['minimum_address'] if metadata['encrypted'] else 0)
|
|
except (ValueError,subprocess.SubprocessError) as error:row.update(status='failed',error=str(error))
|
|
evidence.append(row)
|
|
failed=any(r['status']!='decoded' for r in evidence)
|
|
(output/'node-evidence.json').write_text(json.dumps({'schema':1,'nodes':evidence},indent=2))
|
|
return {'protocol':1,'layer':'derived','coverage':'partial' if failed else 'complete','files':[p.relative_to(output).as_posix() for p in sorted(output.rglob('*')) if p.is_file()],'warnings':['Node4 firmware is unsupported; original HEX files retained.'] if failed else []}
|
|
if __name__=='__main__':
|
|
request=json.loads(pathlib.Path(sys.argv[1]).read_text());pathlib.Path(request['result_file']).write_text(json.dumps(run(request)))
|