89 lines
4.4 KiB
Python
89 lines
4.4 KiB
Python
"""Verify the optical packet without extracting archives or rewriting evidence."""
|
|
from pathlib import Path, PurePosixPath
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import subprocess
|
|
import zipfile
|
|
|
|
|
|
def require(condition, message):
|
|
if not condition:
|
|
raise AssertionError(message)
|
|
|
|
|
|
def digest(data):
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def safe_name(name):
|
|
path = PurePosixPath(name)
|
|
require(not path.is_absolute() and '..' not in path.parts and '\\' not in name,
|
|
f'Unsafe packet path: {name}')
|
|
return path
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--originals', action='store_true', help='Also compare retained original host files')
|
|
parser.add_argument('--git-index', action='store_true', help='Also compare staged Git bytes')
|
|
args = parser.parse_args()
|
|
packet = Path(__file__).resolve().parent / 'evidence'
|
|
index_bytes = (packet / 'index-final.json').read_bytes()
|
|
index = json.loads(index_bytes)
|
|
require(index['source'] == '38789512626cb24f78f274998977fc8c6cf960d1', 'Unexpected tested source')
|
|
rows = index['files']
|
|
names = [row['path'] for row in rows]
|
|
require(len(names) == len(set(names)) == index['standalone_file_count'], 'Duplicate or missing file inventory')
|
|
actual = {path.relative_to(packet).as_posix() for path in packet.rglob('*') if path.is_file()}
|
|
require(actual == set(names) | {'index-final.json'}, 'Uninventoried or missing packet files')
|
|
total = 0
|
|
for row in rows:
|
|
safe_name(row['path'])
|
|
data = (packet / row['path']).read_bytes()
|
|
require(len(data) == row['bytes'] and digest(data) == row['sha256'], f'Changed packet file: {row["path"]}')
|
|
total += len(data)
|
|
require(total == index['standalone_bytes'], 'Wrong standalone byte total')
|
|
archive_count = archive_bytes = originals = 0
|
|
for archive in index['archives']:
|
|
inventory = json.loads((packet / archive['member_inventory']['path']).read_bytes())
|
|
members = inventory['members']
|
|
expected = {row['path']: row for row in members}
|
|
require(len(expected) == len(members) == archive['member_count'] == inventory['member_count'], 'Duplicate member inventory')
|
|
size = 0
|
|
with zipfile.ZipFile(packet / archive['archive']['path']) as zipped:
|
|
listed = zipped.namelist()
|
|
require(len(listed) == len(set(listed)) and set(listed) == set(expected), 'Wrong or duplicated ZIP members')
|
|
for name in listed:
|
|
safe_name(name)
|
|
row = expected[name]
|
|
data = zipped.read(name)
|
|
require(len(data) == row['bytes'] and digest(data) == row['sha256'], f'Changed archived member: {name}')
|
|
size += len(data)
|
|
if args.originals:
|
|
require(Path(row['original_source_path']).read_bytes() == data, f'Archived bytes differ from original: {name}')
|
|
originals += 1
|
|
require(size == archive['uncompressed_bytes'] == inventory['uncompressed_bytes'], 'Wrong archive byte total')
|
|
archive_count += len(members)
|
|
archive_bytes += size
|
|
require(archive_count == index['archived_member_count'] and archive_bytes == index['archived_uncompressed_bytes'], 'Wrong archive total')
|
|
if args.originals:
|
|
for row in index['copied_sources']:
|
|
require((packet / row['path']).read_bytes() == Path(row['original_source_path']).read_bytes(), f'Copied original differs: {row["path"]}')
|
|
originals += 1
|
|
if args.git_index:
|
|
root = Path(subprocess.check_output(['git', 'rev-parse', '--show-toplevel'], cwd=packet, text=True).strip())
|
|
for path in sorted(actual):
|
|
relative = (packet / path).relative_to(root).as_posix()
|
|
staged = subprocess.check_output(['git', 'show', ':' + relative], cwd=root)
|
|
require(staged == (packet / path).read_bytes(), f'Staged bytes differ: {relative}')
|
|
print(json.dumps({'status': 'PASS', 'tested_source': index['source'],
|
|
'index_sha256': digest(index_bytes), 'packet_files': len(actual),
|
|
'packet_bytes': total + len(index_bytes), 'archive_members': archive_count,
|
|
'uncompressed_archive_bytes': archive_bytes, 'original_files_compared': originals,
|
|
'staged_bytes_checked': args.git_index}, indent=2))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|