#!/usr/bin/env python3 """Copy a frozen local browser cache without flattening internal file aliases.""" import hashlib import json import os from pathlib import Path import shutil import stat import sys class CacheCopyError(Exception): pass def regular_reader(file): descriptor = os.open(file, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) if not stat.S_ISREG(os.fstat(descriptor).st_mode): os.close(descriptor) raise CacheCopyError('special_file') return os.fdopen(descriptor, 'rb') def file_digest(file): digest = hashlib.sha256() with regular_reader(file) as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b''): digest.update(chunk) return digest.hexdigest() def snapshot(source): source = Path(source) if source.is_symlink() or not source.is_dir(): raise CacheCopyError('source_not_physical_directory') root = source.resolve(strict=True) entries = {} def walk_error(_error): raise CacheCopyError('source_unreadable') # Classify and validate every link before opening any linked file content. for parent, directories, files in os.walk(root, followlinks=False, onerror=walk_error): for name in sorted(directories + files): file = Path(parent) / name mode = file.lstat().st_mode relative = file.relative_to(root).as_posix() if stat.S_ISLNK(mode): target = os.readlink(file) if os.path.isabs(target): raise CacheCopyError('absolute_link') try: resolved = file.resolve(strict=True) except (OSError, RuntimeError): raise CacheCopyError('broken_or_cyclic_link') from None if not resolved.is_relative_to(root): raise CacheCopyError('external_link') if not resolved.is_file(): raise CacheCopyError('link_target_not_regular_file') entries[relative] = {'kind': 'link', 'target': target, 'resolved': resolved.relative_to(root).as_posix()} elif stat.S_ISDIR(mode): entries[relative] = {'kind': 'directory', 'mode': stat.S_IMODE(mode)} elif stat.S_ISREG(mode): entries[relative] = {'kind': 'file', 'mode': stat.S_IMODE(mode), 'size': file.stat().st_size} else: raise CacheCopyError('special_file') for name, entry in entries.items(): if entry['kind'] == 'file': entry['sha256'] = file_digest(root / name) for entry in entries.values(): if entry['kind'] == 'link': target = entries.get(entry['resolved']) if not target or target['kind'] != 'file': raise CacheCopyError('link_target_not_in_manifest') return {'rootMode': stat.S_IMODE(root.stat().st_mode), 'entries': dict(sorted(entries.items()))} def snapshot_digest(value): return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(',', ':')).encode()).hexdigest() def check_paths(source, destination): if source.is_symlink() or not source.is_dir(): raise CacheCopyError('source_not_physical_directory') source_root = source.resolve(strict=True) target_root = destination.resolve(strict=False) if target_root.is_relative_to(source_root) or source_root.is_relative_to(target_root): raise CacheCopyError('overlapping_directories') if os.path.lexists(destination): raise CacheCopyError('destination_exists') if not destination.parent.is_dir(): raise CacheCopyError('destination_parent_missing') return source_root, target_root def copy_regular(source, destination): # Exclusive destination and no-follow source: never turn a file alias into # an extra loaded library, or reuse a pre-existing destination inode. with regular_reader(source) as reader, destination.open('xb') as writer: metadata = os.fstat(reader.fileno()) shutil.copyfileobj(reader, writer) os.fchmod(writer.fileno(), stat.S_IMODE(metadata.st_mode)) os.utime(destination, ns=(metadata.st_atime_ns, metadata.st_mtime_ns), follow_symlinks=False) def copy_cache(source, destination): source, destination = check_paths(Path(source), Path(destination)) before = snapshot(source) destination.mkdir(mode=0o700) directories = [name for name, entry in before['entries'].items() if entry['kind'] == 'directory'] for name in sorted(directories, key=lambda value: (value.count('/'), value)): (destination / name).mkdir(mode=0o700) for name, entry in before['entries'].items(): if entry['kind'] == 'file': copy_regular(source / name, destination / name) for name, entry in before['entries'].items(): if entry['kind'] == 'link': (destination / name).symlink_to(entry['target']) for name in reversed(sorted(directories, key=lambda value: (value.count('/'), value))): (destination / name).chmod(before['entries'][name]['mode']) destination.chmod(before['rootMode']) after = snapshot(source) copied = snapshot(destination) if before != after: raise CacheCopyError('source_changed') if copied != before: raise CacheCopyError('copy_identity_mismatch') regular = [name for name, entry in before['entries'].items() if entry['kind'] == 'file'] source_inodes = {(entry.st_dev, entry.st_ino) for entry in ((source / name).stat() for name in regular)} if any((entry.st_dev, entry.st_ino) in source_inodes for entry in ((destination / name).stat() for name in regular)): raise CacheCopyError('source_inode_shared') return { 'schemaVersion': 1, 'status': 'PASS', 'scope': 'LOCAL_BROWSER_CACHE_COPY_ONLY', 'sourceUnchanged': True, 'sourceInodesIndependent': True, 'bytesModesAndLinksMatch': True, 'regularFiles': len(regular), 'internalFileLinks': sum(entry['kind'] == 'link' for entry in before['entries'].values()), 'sourceBeforeSha256': snapshot_digest(before), 'sourceAfterSha256': snapshot_digest(after), 'destinationSha256': snapshot_digest(copied), 'manifest': before, 'deploymentEligible': False, } def main(arguments): if len(arguments) != 3: raise CacheCopyError('usage_source_destination_receipt_required') source, destination, receipt = map(Path, arguments) source_root, target_root = check_paths(source, destination) receipt_root = receipt.resolve(strict=False) if receipt_root.is_relative_to(source_root) or receipt_root.is_relative_to(target_root): raise CacheCopyError('receipt_inside_cache') if os.path.lexists(receipt): raise CacheCopyError('receipt_exists') # Preserve failures as new evidence; never erase a partially copied cache. with receipt.open('x', encoding='utf8') as stream: os.fchmod(stream.fileno(), 0o600) try: result = copy_cache(source, destination) except CacheCopyError as error: result = {'status': 'FAIL', 'errorKind': str(error), 'deploymentEligible': False} except OSError: result = {'status': 'FAIL', 'errorKind': 'filesystem_error', 'deploymentEligible': False} json.dump(result, stream, ensure_ascii=False, indent=2) stream.write('\n') print(json.dumps({key: value for key, value in result.items() if key != 'manifest'})) return 0 if result['status'] == 'PASS' else 1 if __name__ == '__main__': try: raise SystemExit(main(sys.argv[1:])) except CacheCopyError as error: print(json.dumps({'status': 'FAIL', 'errorKind': str(error), 'deploymentEligible': False})) raise SystemExit(1) except OSError: print(json.dumps({'status': 'FAIL', 'errorKind': 'filesystem_error', 'deploymentEligible': False})) raise SystemExit(1)