python_filecheck.py (10611B)
1 #!/usr/bin/env python3 2 """Installed Python consumer: no checkout/compiler/cache at runtime.""" 3 import argparse 4 from dataclasses import asdict 5 import hashlib 6 import json 7 import os 8 from pathlib import Path 9 import resource 10 import shutil 11 import subprocess 12 import sys 13 import tempfile 14 import time 15 from unittest.mock import patch 16 17 from process_rss import run_guarded 18 19 20 def reuse(directory): 21 from tricu_filecheck import FileChecker, FileProblem 22 directory = Path(directory) 23 source = directory/'reused.bin' 24 body = bytes(range(256))*4 25 source.write_bytes(body) 26 started = time.monotonic() 27 with FileChecker() as checker: 28 worker = checker.worker 29 context = checker.context.identity 30 for _ in range(8): 31 checker.hash_file(source) 32 fd_count = len(list(Path('/proc/self/fd').iterdir())) 33 for _ in range(128): 34 # Native IO, not a Python hashing/read fallback. Bundle loading has 35 # already completed; Python file opens during hashing are prohibited. 36 with patch('builtins.open', side_effect=AssertionError('Python opened worker input')), patch.object(Path, 'open', side_effect=AssertionError('Python opened worker input')): 37 actual = checker.hash_file(source) 38 assert asdict(actual) == dict(bytes=len(body), sha256=hashlib.sha256(body).hexdigest()) 39 try: 40 checker.hash_file(directory/'missing') 41 except FileProblem as error: 42 assert error.problem.category == 'not-found' 43 assert error.problem.operation == 'path/stat' 44 else: 45 raise AssertionError('missing file accepted') 46 assert checker.worker == worker and checker.context.identity == context 47 assert len(list(Path('/proc/self/fd').iterdir())) == fd_count 48 large = directory/'large.bin' 49 payload = bytes(range(256))*4096 50 large.write_bytes(payload) 51 assert checker.hash_file(large).sha256 == hashlib.sha256(payload).hexdigest() 52 for invalid in (-1, 2**63, 1.5, True): 53 try: 54 checker.hash_file(source, invalid) 55 except ValueError: 56 pass 57 else: 58 raise AssertionError('invalid byte limit accepted') 59 checker.close() 60 return dict(warmup_calls=8, successful_repeated_calls=128, expected_error_calls=128, 61 large_file_bytes=len(payload), seconds=time.monotonic()-started, 62 peak_rss_kib=resource.getrusage(resource.RUSAGE_SELF).ru_maxrss, 63 descriptor_count=fd_count) 64 65 66 def acceptance(args): 67 report = dict(checks=[]) 68 with tempfile.TemporaryDirectory(prefix='filecheck-consumer-') as work: 69 work = Path(work) 70 data, caller, home = work/'data', work/'unrelated-cwd', work/'empty-home' 71 for directory in (data, caller, home): 72 directory.mkdir() 73 env = dict(os.environ, HOME=str(home), XDG_CACHE_HOME=str(home/'cache'), 74 PATH='/no-runtime-tools', PYTHONDONTWRITEBYTECODE='1') 75 for variable in ('TRICU_LIB', 'ARBORICX_LIB', 'PYTHONPATH', 'tricu_datadir'): 76 env.pop(variable, None) 77 # CLI runs have no source tree, tool lookup, compiler or writable cache. 78 caller.chmod(0o555) 79 home.chmod(0o555) 80 def run(*arguments, code=0, environment=None): 81 result = subprocess.run([str(args.app), *map(str, arguments)], cwd=caller, 82 env=environment or env, text=True, capture_output=True, timeout=30) 83 assert result.returncode == code, (arguments, result.returncode, result.stdout, result.stderr) 84 assert not result.stderr, result.stderr 85 return json.loads(result.stdout) 86 contents = {'empty': b'', 'boundary.bin': bytes(range(256))*16+b'\0', 87 'Unicode 🌳\nname.bin': b'\xff\x00\x80binary\r\n'} 88 for name, body in contents.items(): 89 (data/name).write_bytes(body) 90 (data/'link').symlink_to('boundary.bin') 91 contents['link'] = contents['boundary.bin'] 92 manifest = data/'checks.json' 93 actual = run('record', manifest, *contents) 94 assert actual['ok'] is True 95 expected = dict(schema='tricu.filecheck.v1', algorithm='sha256', files=[ 96 dict(path=name, bytes=len(body), sha256=hashlib.sha256(body).hexdigest()) for name, body in contents.items()]) 97 assert json.loads(manifest.read_text()) == expected 98 assert run('verify', manifest)['ok'] is True 99 report['checks'].append('installed source-free record/verify; exact hashes and byte counts') 100 import tricu_filecheck 101 installed = Path(tricu_filecheck.__file__).parent 102 assert (installed/'worker.arboricx').is_file() 103 assert not list(installed.rglob('*.tri')) 104 original = manifest.read_bytes() 105 assert run('record', manifest, *contents, code=2)['ok'] is False 106 assert manifest.read_bytes() == original 107 assert not list(data.glob('.filecheck-*')) 108 report['checks'].append('existing manifest preserved; staging cleaned') 109 # Same-length mutations ensure verification actually compares hashes. 110 (data/'boundary.bin').write_bytes(b'x'*len(contents['boundary.bin'])) 111 changed = run('verify', manifest, code=1) 112 assert [r['path'] for r in changed['files'] if r['status']=='changed'] == ['boundary.bin','link'] 113 (data/'empty').unlink() 114 failed = run('verify', manifest, code=2) 115 assert failed['files'][0]['error']['category'] == 'not-found' 116 assert failed['files'][0]['error']['operation'] == 'path/stat' 117 assert failed['files'][1]['status'] == 'changed' 118 report['checks'].append('changed and missing files distinguished; remaining entries checked') 119 (data/'empty').write_bytes(b'') 120 exact = run('record', data/'exact.json', 'empty', '--max-bytes', 0) 121 assert exact['files'][0]['bytes'] == 0 122 exact_boundary = run('record', data/'exact-boundary.json', 'boundary.bin', '--max-bytes', 4097) 123 assert exact_boundary['files'][0]['bytes'] == 4097 124 limit = run('record', data/'over-limit.json', 'boundary.bin', '--max-bytes', 4096, code=2) 125 assert limit['files'][0]['error']['category'] == 'invalid-bound' 126 assert not (data/'over-limit.json').exists() 127 (data/'directory').mkdir() 128 os.mkfifo(data/'fifo') 129 for name in ('directory', 'fifo'): 130 failure = run('record', data/f'{name}.json', name, code=2) 131 assert failure['files'][0]['error']['category'] == 'invalid-argument' 132 assert os.geteuid() != 0, 'permission control requires an unprivileged test process' 133 denied = data/'denied' 134 denied.write_bytes(b'secret') 135 denied.chmod(0) 136 try: 137 failure = run('record', data/'denied.json', 'denied', code=2) 138 assert failure['files'][0]['error']['category'] == 'os-permission-denied' 139 finally: 140 denied.chmod(0o600) 141 mixed = run('record', data/'partial.json', 'empty', 'missing', code=2) 142 assert [row['status'] for row in mixed['files']] == ['ok','error'] 143 assert not (data/'partial.json').exists() 144 report['checks'].append('zero/excess budget, non-files, permission failure and no partial manifest') 145 # Validate all manifest syntax/fields before executing the worker. 146 bad = data/'bad.json' 147 cases = ['{', '{"schema":1,"schema":2}', json.dumps(dict(expected, algorithm='md5')), 148 json.dumps(dict(expected, files=[expected['files'][0]]*2)), 149 json.dumps(dict(expected, files=[dict(expected['files'][0], path='/absolute')])), 150 json.dumps(dict(expected, files=[dict(expected['files'][0], path='bad\0path')])), 151 json.dumps(dict(expected, files=[dict(expected['files'][0], bytes=True)])), 152 json.dumps(dict(expected, files=[dict(expected['files'][0], sha256='no')])), 153 json.dumps(dict(expected, files=[])), ' '* (1024*1024+1)] 154 for text in cases: 155 bad.write_text(text) 156 assert run('verify', bad, code=2)['ok'] is False 157 assert run('record', data/'duplicate.json', 'empty', 'empty', code=2)['ok'] is False 158 report['checks'].append('strict manifest validation; duplicate paths/keys and size bound') 159 # Corrupt/missing packaged workers fail closed, with no source/compiler 160 # lookup fallback. The copied installed module keeps its original pin. 161 copied = work/'copied-package' 162 package = copied/'tricu_filecheck' 163 shutil.copytree(Path(tricu_filecheck.__file__).parent, package) 164 for item in [package, *package.rglob('*')]: 165 item.chmod(0o755 if item.is_dir() else 0o644) 166 bundle = package/'worker.arboricx' 167 bundle.write_bytes(bundle.read_bytes()[:32]) 168 command = [sys.executable, '-c', 169 'import sys; sys.path.insert(0,sys.argv.pop(1)); from tricu_filecheck.cli import main; sys.exit(main())', 170 str(copied), 'record', str(data/'corrupt.json'), 'empty'] 171 result = subprocess.run(command, cwd=caller, text=True, capture_output=True, timeout=15) 172 assert result.returncode == 2 and 'checksum mismatch' in result.stdout 173 assert not (data/'corrupt.json').exists() 174 bundle.unlink() 175 result = subprocess.run(command, cwd=caller, text=True, capture_output=True, timeout=15) 176 assert result.returncode == 2 and not (data/'corrupt.json').exists() 177 report['checks'].append('corrupt and missing installed bundle rejected without fallback') 178 repeated, peak = run_guarded([sys.executable, str(Path(__file__).resolve()), '--reuse', str(data)], 179 memory_mib=512, seconds=90) 180 report['reuse'] = json.loads(repeated) 181 report['reuse']['sampled_peak_rss_kib'] = peak 182 assert not list(caller.iterdir()) and not list(home.iterdir()) 183 caller.chmod(0o755) 184 home.chmod(0o755) 185 return report 186 187 188 def main(): 189 parser=argparse.ArgumentParser(description=__doc__) 190 parser.add_argument('--app', type=Path) 191 parser.add_argument('--report', type=Path) 192 parser.add_argument('--reuse', type=Path) 193 args=parser.parse_args() 194 if args.reuse: 195 print(json.dumps(reuse(args.reuse))) 196 else: 197 report=acceptance(args) 198 args.report.write_text(json.dumps(report,ensure_ascii=False,indent=2)+'\n') 199 print(json.dumps(report,ensure_ascii=False)) 200 201 202 if __name__=='__main__': 203 main()