"""R626:生成无凭证合成复现包,在独立目录/进程实际回放并核验。""" from hashlib import sha256 import json from pathlib import Path import subprocess import sys from tempfile import TemporaryDirectory from zipfile import ZipFile STRATEGY = '''def run(prices, config): cash, shares = config['cash'], 0 fills = [] for t in range(1, len(prices)-1): target = int(prices[t] > prices[t-1]) qty = target-shares if qty: fills.append([t+1, qty, prices[t+1]]) cash -= qty*prices[t+1] shares = target return {'ending': cash+shares*prices[-1], 'fills': fills} ''' RUNNER = '''from pathlib import Path from hashlib import sha256 import json, runpy root=Path(__file__).resolve().parent manifest=json.loads((root/'manifest.json').read_text()) for name,expected in manifest['sha256'].items(): path=root/name if not path.exists(): raise RuntimeError('missing required file: '+name) if sha256(path.read_bytes()).hexdigest()!=expected: raise RuntimeError('digest mismatch: '+name) run=runpy.run_path(str(root/'strategy.py'))['run'] result=run(json.loads((root/'data.json').read_text()),json.loads((root/'config.json').read_text())) assert result==json.loads((root/'expected.json').read_text()) print(json.dumps(result,sort_keys=True)) ''' def main(): expected={'ending':1003,'fills':[[2,1,101],[3,-1,104]]} files={'strategy.py':STRATEGY.encode(),'runner.py':RUNNER.encode(), 'data.json':json.dumps([100,102,101,104]).encode(), 'config.json':json.dumps({'cash':1000}).encode(), 'expected.json':json.dumps(expected).encode(), 'README.txt':b'Python 3.12+ standard library. Synthetic redistributable data. Run python -I -B runner.py. No credentials.'} manifest={'runtime':'Python 3.12+ standard library','data_rights':'synthetic fixture', 'sha256':{name:sha256(blob).hexdigest() for name,blob in files.items()}} with TemporaryDirectory(prefix='r626-package-') as temp: root=Path(temp);destination=root/'independent';destination.mkdir() archive=root/'package.zip' with ZipFile(archive,'w') as out: for name,blob in files.items():out.writestr(name,blob) out.writestr('manifest.json',json.dumps(manifest)) with ZipFile(archive) as inp: for name in inp.namelist(): if Path(name).is_absolute() or '..' in Path(name).parts:raise ValueError('unsafe archive path') (destination/name).write_bytes(inp.read(name)) command=[sys.executable,'-I','-B',str(destination/'runner.py')] good=subprocess.run(command,cwd=root,capture_output=True,text=True,timeout=5) assert good.returncode==0 and json.loads(good.stdout)==expected (destination/'data.json').write_text('[1,2,3]') bad=subprocess.run(command,cwd=root,capture_output=True,text=True,timeout=5) assert bad.returncode!=0 and 'digest mismatch: data.json' in bad.stderr (destination/'data.json').write_bytes(files['data.json']) limited=root/'restricted_data_case';limited.mkdir() for name,blob in files.items(): if name!='data.json':(limited/name).write_bytes(blob) (limited/'manifest.json').write_text(json.dumps(manifest)) missing=subprocess.run([sys.executable,'-I','-B',str(limited/'runner.py')],cwd=root,capture_output=True,text=True,timeout=5) assert missing.returncode!=0 and 'missing required file: data.json' in missing.stderr print('independent_directory=True isolated_child=True ending=1003 fills=[[2, 1, 101], [3, -1, 104]]') print('hashed_files=6 tampered_data=rejected unavailable_data=precondition_blocked credentials=none') if __name__=='__main__':main()