#!/usr/bin/env python3 """Conservative local proof: explicitly listed files only; no Markdown parsing. Usage: python manifest_copy.py --source VAULT_A --manifest paths.txt --target NEW_DIR Add --apply to copy. Each UTF-8 manifest line is one exact POSIX-style vault-relative file path. Empty lines, backslashes, dot segments and duplicate entries are rejected. The target's parent must already exist. The entire source tree is conservatively checked for symlinks and NFC+casefold collisions, including unselected entries. This is a Linux-tested local workflow, not a hostile concurrent-filesystem sandbox. No files are deleted, including after a partially completed failure. """ import argparse import hashlib import json import os from pathlib import Path, PurePosixPath, PureWindowsPath import stat import sys import unicodedata class Rejected(Exception): pass def sha256(path): digest = hashlib.sha256() with path.open('rb') as handle: for block in iter(lambda: handle.read(1024 * 1024), b''): digest.update(block) return digest.hexdigest() def no_symlink_components(path): if '..' in Path(path).parts: raise Rejected('parent traversal is not allowed in input paths') absolute = Path(os.path.abspath(path)) current = Path(absolute.anchor) for component in absolute.parts[1:]: current /= component if current.is_symlink(): raise Rejected('symlink component: ' + str(current)) return absolute def key(path): return '/'.join(unicodedata.normalize('NFC', p).casefold() for p in PurePosixPath(path).parts) def remember_path(seen, rel, kind): normalized = key(rel) old = seen.get(normalized) if old is not None and old != (rel, kind): raise Rejected('NFC+casefold path collision: ' + old[0] + ' / ' + rel) seen[normalized] = (rel, kind) def audit_source(source): seen = {} def walk_error(exc): raise Rejected('cannot audit the complete source tree: ' + str(exc)) for root, dirs, files in os.walk(source, followlinks=False, onerror=walk_error): for name in sorted(dirs + files): path = Path(root) / name metadata = path.lstat() if stat.S_ISLNK(metadata.st_mode): raise Rejected('source tree contains symlink: ' + str(path)) if not (stat.S_ISDIR(metadata.st_mode) or stat.S_ISREG(metadata.st_mode)): raise Rejected('source tree contains non-regular node: ' + str(path)) kind = 'directory' if stat.S_ISDIR(metadata.st_mode) else 'file' remember_path(seen, path.relative_to(source).as_posix(), kind) def parse_manifest(manifest): raw = manifest.read_bytes() try: lines = raw.decode('utf-8').splitlines() except UnicodeDecodeError as exc: raise Rejected('manifest must be UTF-8') from exc if not lines: raise Rejected('empty manifest') seen = {} names = [] for number, line in enumerate(lines, 1): windows = PureWindowsPath(line) parts = line.split('/') if (not line or '\x00' in line or '\\' in line or PurePosixPath(line).is_absolute() or windows.drive or windows.root or any(part in ('', '.', '..') for part in parts)): raise Rejected('invalid relative manifest path at line ' + str(number)) normalized = key(line) if normalized in seen: raise Rejected('duplicate/colliding manifest path: ' + line) seen[normalized] = line names.append(line) return names, hashlib.sha256(raw).hexdigest() def prepare(source_arg, manifest_arg, target_arg): source = no_symlink_components(source_arg) manifest = no_symlink_components(manifest_arg) target = no_symlink_components(target_arg) if not source.is_dir(): raise Rejected('source must be an existing directory') if not manifest.is_file(): raise Rejected('manifest must be an existing regular file') if os.path.lexists(target): raise Rejected('target must not exist, including an empty directory') if not target.parent.is_dir(): raise Rejected('target parent must already be an existing directory') source = source.resolve(strict=True) manifest = manifest.resolve(strict=True) target = target.parent.resolve(strict=True) / target.name if target == source or source in target.parents: raise Rejected('target must not be inside the source vault') audit_source(source) names, manifest_hash = parse_manifest(manifest) entries = [] for name in names: selected = no_symlink_components(source / name) if not selected.is_file() or not stat.S_ISREG(selected.stat().st_mode): raise Rejected('selected path is missing or not a regular file: ' + name) if source not in selected.resolve(strict=True).parents: raise Rejected('selected path escapes source: ' + name) before = selected.stat() digest = sha256(selected) after = selected.stat() if (before.st_size, before.st_mtime_ns, before.st_ino) != (after.st_size, after.st_mtime_ns, after.st_ino): raise Rejected('source changed while planning: ' + name) entries.append({'path': name, 'bytes': after.st_size, 'sha256': digest}) return source, target, manifest_hash, entries def execute(args): source, target, manifest_hash, entries = prepare(args.source, args.manifest, args.target) receipt = {'ok': True, 'mode': 'apply' if args.apply else 'plan', 'source': str(source), 'target': str(target), 'manifestSha256': manifest_hash, 'files': entries, 'copiesOnlyListedFiles': True, 'parsesMarkdown': False, 'provesDependencyCompleteness': False, 'sourceTreeSymlinkAndCollisionAudit': 'entire source tree', 'writesPerformed': False, 'limitation': 'Linux Python tested; no Windows/macOS/Obsidian GUI test; assumes no concurrent filesystem modification.'} if not args.apply: return receipt no_symlink_components(target.parent) target.mkdir(exist_ok=False) receipt['writesPerformed'] = True for entry in entries: src = no_symlink_components(source / entry['path']) if not src.is_file() or sha256(src) != entry['sha256']: raise Rejected('source changed before copy: ' + entry['path']) dst = target / entry['path'] dst.parent.mkdir(parents=True, exist_ok=True) no_symlink_components(dst.parent) with src.open('rb') as reader, dst.open('xb') as writer: for block in iter(lambda: reader.read(1024 * 1024), b''): writer.write(block) output_hash = sha256(dst) source_hash = sha256(src) if output_hash != entry['sha256'] or source_hash != entry['sha256']: raise Rejected('post-copy hash mismatch: ' + entry['path']) if dst.stat().st_size != entry['bytes']: raise Rejected('post-copy byte count mismatch: ' + entry['path']) entry['copiedSha256'] = output_hash entry['sourceSha256AfterCopy'] = source_hash entry['verified'] = True for entry in entries: no_symlink_components(source / entry['path']) if sha256(source / entry['path']) != entry['sha256']: raise Rejected('source changed before final verification: ' + entry['path']) receipt['allListedFilesVerified'] = True return receipt def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--source', required=True) parser.add_argument('--manifest', required=True) parser.add_argument('--target', required=True) parser.add_argument('--apply', action='store_true') args = parser.parse_args() try: result = execute(args) except (Rejected, OSError, ValueError) as exc: print(json.dumps({'ok': False, 'error': str(exc), 'sourceDeleted': False, 'partialTargetMayRemain': bool(args.apply), 'instruction': 'Inspect any partial target; choose a new absent target for another run.'}, ensure_ascii=False)) return 2 print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if __name__ == '__main__': sys.exit(main())