Files
archy/scripts/cloud-test-artifact.py

70 lines
2.8 KiB
Python

#!/usr/bin/env python3
"""Source/compiler/runtime-bound local test artifact; not independent approval."""
import hashlib
import json
from pathlib import Path
import shutil
import stat
import sys
def source_hash(root):
digest = hashlib.sha256()
for path in sorted(Path(root).rglob('*')):
if not path.is_file() and not path.is_symlink():
continue
digest.update(str(path.relative_to(root)).encode() + b'\0')
digest.update(str(path.lstat().st_mode).encode() + b'\0')
digest.update(str(path.readlink()).encode() if path.is_symlink() else path.read_bytes())
digest.update(b'\0')
return digest.hexdigest()
def regular(path):
if not stat.S_ISREG(path.lstat().st_mode):
raise ValueError('Artifact must be a regular file, not a symlink')
return path
def transfer(mode, source, compiler, runtime, binary, directory):
root = Path(directory)
receipt_file = root / 'artifact.json'
artifact = root / 'test-binary'
expected = dict(source_sha256=source_hash(Path(source)), compiler_image=compiler, runtime_image=runtime)
if mode == 'export':
binary = regular(Path(binary))
with binary.open('rb') as src, artifact.open('xb') as dest:
shutil.copyfileobj(src, dest)
artifact.chmod(0o500)
expected.update(binary_sha256=hashlib.sha256(artifact.read_bytes()).hexdigest(),
status='compile-only; execution qualification recorded separately')
with receipt_file.open('x') as out:
json.dump(expected, out, indent=2)
out.write('\n')
elif mode == 'reuse':
receipt = json.loads(regular(receipt_file).read_text())
if not all(receipt.get(key) == value for key, value in expected.items()):
raise ValueError('Artifact source/compiler/runtime binding mismatch')
data = regular(artifact).read_bytes()
if not data or hashlib.sha256(data).hexdigest() != receipt.get('binary_sha256'):
raise ValueError('Artifact binary hash mismatch')
# Destination is the runner-created empty private scratch file.
destination = regular(Path(binary))
if destination.stat().st_size:
raise ValueError('Artifact destination must be empty')
destination.write_bytes(data)
else:
raise ValueError('Unknown artifact action')
if __name__ == '__main__':
try:
if len(sys.argv) != 8:
raise ValueError('Expected action/source/compiler/runtime/binary/directory plus binding version')
if sys.argv[7] != 'v1':
raise ValueError('Unknown binding version')
transfer(*sys.argv[1:7])
except (ValueError, OSError, TypeError) as error:
print(f'Test artifact refused: {error}', file=sys.stderr)
sys.exit(2)