폴더의 두 시점 목록에서 추가·삭제·변경을 구별하기

"""Read-only folder snapshots and strict comparisons (Python 3.12+)."""
import argparse
import hashlib
import json
import os
from pathlib import Path
import re
import stat
import sys
import unicodedata

MAX_JSON = 16 * 1024 * 1024
EMPTY_HASH = hashlib.sha256(b'').hexdigest()
REASONS = {'file_failed', 'scan_failed', 'limit', 'changed_during_read',
           'unsafe_path', 'link_or_reparse', 'special'}
BUCKETS = ('added', 'deleted', 'changed', 'unchanged', 'metadata_changed', 'unverified')


def require(condition, message):
    if not condition:
        raise ValueError(message)


def fields(value, names):
    require(type(value) is dict and set(value) == set(names.split()), 'invalid fields')


def integer(value, low, high):
    require(type(value) is int and low <= value <= high, 'invalid integer')


def safe_path(value):
    require(type(value) is str and 0 < len(value) <= 1024, 'unsafe path')
    require(unicodedata.normalize('NFC', value) == value, 'unsafe path normalization')
    parts = value.split('/')
    require(len(parts) <= 64, 'unsafe path depth')
    for part in parts:
        require(part not in ('', '.', '..') and not part.endswith((' ', '.')) and
                not re.search(r'[\x00-\x1f\x7f\ud800-\udfff<>:"\\|?*]', part) and
                not re.fullmatch(r'(?i:CON|PRN|AUX|NUL|COM[1-9¹²³]|LPT[1-9¹²³])(?:\..*)?', part),
                'unsafe path')
    return value


def excluded(path, prefixes):
    parts = path.split('/')
    return any('/'.join(parts[:i]) in prefixes for i in range(1, len(parts) + 1))


def validate_config(cfg):
    fields(cfg, 'exclude hash max_entries max_file_bytes max_total_bytes links')
    require(type(cfg['hash']) is bool and cfg['links'] == 'fail', 'invalid config')
    require(type(cfg['exclude']) is list, 'invalid exclude')
    for value in cfg['exclude']:
        safe_path(value)
    require(cfg['exclude'] == sorted(set(cfg['exclude'])), 'invalid exclude ordering')
    integer(cfg['max_entries'], 1, 100000)
    integer(cfg['max_file_bytes'], 1, 1024 ** 3)
    integer(cfg['max_total_bytes'], 1, 8 * 1024 ** 3)


def config(exclude=(), do_hash=True, max_entries=10000,
           max_file_bytes=10485760, max_total_bytes=104857600):
    cfg = dict(exclude=sorted(set(exclude)), hash=do_hash, max_entries=max_entries,
               max_file_bytes=max_file_bytes, max_total_bytes=max_total_bytes, links='fail')
    validate_config(cfg)
    return cfg


def validate(data):
    fields(data, 'schema algorithm config complete errors excluded files')
    require(type(data['schema']) is int and data['schema'] == 1 and
            data['algorithm'] == 'sha256', 'unsupported schema/algorithm')
    validate_config(data['config'])
    cfg = data['config']
    prefixes = set(cfg['exclude'])
    require(type(data['complete']) is bool, 'invalid complete')
    for name in ('files', 'errors', 'excluded'):
        limit = 2 * cfg['max_entries'] + 2 if name == 'errors' else cfg['max_entries']
        require(type(data[name]) is list and len(data[name]) <= limit,
                'invalid list or limit')
    require(data['complete'] == (not data['errors']), 'complete/errors conflict')
    error_keys = set()
    for item in data['errors']:
        fields(item, 'path reason')
        if item['path'] is not None:
            safe_path(item['path'])
        require(type(item['reason']) is str and item['reason'] in REASONS, 'invalid error reason')
        error_keys.add((item['path'], item['reason']))
    excluded_paths = set()
    for item in data['excluded']:
        fields(item, 'path reason')
        path = safe_path(item['path'])
        require(path not in excluded_paths, 'duplicate excluded path')
        excluded_paths.add(path)
        require(type(item['reason']) is str and item['reason'] in
                ('configured', 'link_or_reparse', 'special'), 'invalid exclusion')
        require(excluded(path, prefixes) if item['reason'] == 'configured' else
                (path, item['reason']) in error_keys, 'unaccounted exclusion')
    result, aliases, sizes = {}, set(), {}
    for item in data['files']:
        require(type(item) is dict and set(item) in
                ({'path', 'size', 'mtime_ns'}, {'path', 'size', 'mtime_ns', 'sha256'}), 'invalid file fields')
        path = safe_path(item['path'])
        require(path.casefold() not in aliases, 'duplicate path')
        aliases.add(path.casefold())
        require(not excluded(path, prefixes) and path not in excluded_paths, 'excluded file included')
        integer(item['size'], 0, cfg['max_file_bytes'])
        integer(item['mtime_ns'], -(2 ** 63), 2 ** 63 - 1)
        digest = item.get('sha256')
        require(digest is None or type(digest) is str and re.fullmatch('[0-9a-f]{64}', digest),
                'invalid sha256')
        require(cfg['hash'] or digest is None, 'hash/config conflict')
        if digest is not None:
            require((digest == EMPTY_HASH) == (item['size'] == 0), 'hash_size conflict')
            require(digest not in sizes or sizes[digest] == item['size'], 'hash_size conflict')
            sizes[digest] = item['size']
        result[path] = item
    require(len(result) <= cfg['max_entries'], 'entry limit')
    require(sum(item['size'] for item in result.values()) <= cfg['max_total_bytes'], 'total byte limit')
    for path in result:
        parts = path.split('/')
        require(not any('/'.join(parts[:i]).casefold() in aliases for i in range(1, len(parts))),
                'file ancestry conflict')
        require(not excluded(path, excluded_paths), 'excluded subtree included')
    return result


def unique_object(pairs):
    obj = {}
    for key, value in pairs:
        require(key not in obj, 'duplicate JSON key')
        obj[key] = value
    return obj


def load(path):
    flags = os.O_RDONLY | getattr(os, 'O_NONBLOCK', 0) | getattr(os, 'O_NOFOLLOW', 0) | getattr(os, 'O_BINARY', 0)
    with os.fdopen(os.open(path, flags), 'rb') as stream:
        require(stat.S_ISREG(os.fstat(stream.fileno()).st_mode), 'JSON input must be a regular file')
        raw = stream.read(MAX_JSON + 1)
    require(len(raw) <= MAX_JSON, 'JSON byte limit')
    data = json.loads(raw, object_pairs_hook=unique_object,
                      parse_constant=lambda _: require(False, 'invalid JSON constant'))
    validate(data)
    return data


def compare(before, after):
    left, right = validate(before), validate(after)
    require(before['complete'] and after['complete'], 'incomplete snapshot; comparison refused')
    require(before['config'] == after['config'], 'config mismatch; comparison refused')
    sizes = {}
    for item in [*left.values(), *right.values()]:
        digest = item.get('sha256')
        if digest is not None:
            require(digest not in sizes or sizes[digest] == item['size'], 'hash_size conflict')
            sizes[digest] = item['size']
    result = {name: [] for name in BUCKETS}
    result['added'], result['deleted'] = sorted(right.keys() - left.keys()), sorted(left.keys() - right.keys())
    for path in sorted(left.keys() & right.keys()):
        a, b = left[path], right[path]
        if a.get('sha256') is None or b.get('sha256') is None:
            bucket, reason = 'unverified', 'hash_missing'
        elif a['sha256'] != b['sha256']:
            bucket, reason = 'changed', 'sha256_differs'
        elif a['mtime_ns'] != b['mtime_ns']:
            bucket, reason = 'metadata_changed', 'mtime_ns_differs'
        else:
            result['unchanged'].append(path)
            continue
        result[bucket].append(dict(path=path, reason=reason))
    return dict(schema=1, algorithm='sha256', counts={name: len(result[name]) for name in BUCKETS}, **result)


def is_link(info):
    return stat.S_ISLNK(info.st_mode) or bool(getattr(info, 'st_file_attributes', 0) & 0x400)


def stamp(info):
    return tuple(getattr(info, name) for name in
                 ('st_dev', 'st_ino', 'st_mode', 'st_size', 'st_mtime_ns', 'st_ctime_ns'))


def read_file(path, info, remaining, cfg):
    require(info.st_size <= cfg['max_file_bytes'] and info.st_size <= remaining, 'limit')
    digest = None
    if cfg['hash']:
        flags = os.O_RDONLY | getattr(os, 'O_NOFOLLOW', 0) | getattr(os, 'O_NONBLOCK', 0) | getattr(os, 'O_BINARY', 0)
        with os.fdopen(os.open(path, flags), 'rb') as stream:
            opened = os.fstat(stream.fileno())
            require(stat.S_ISREG(opened.st_mode) and not is_link(opened) and stamp(opened) == stamp(info),
                    'changed_during_read')
            hasher, total = hashlib.sha256(), 0
            while total < info.st_size:
                block = stream.read(min(65536, info.st_size - total))
                if not block:
                    break
                total += len(block)
                hasher.update(block)
            require(total == info.st_size and stamp(os.fstat(stream.fileno())) == stamp(info), 'changed_during_read')
            digest = hasher.hexdigest()
    require(stamp(path.lstat()) == stamp(info), 'changed_during_read')
    return dict(size=info.st_size, mtime_ns=info.st_mtime_ns, sha256=digest)


def snapshot(root, cfg):
    validate_config(cfg)
    require('..' not in Path(root).parts, 'input parent hops are not allowed')
    root = Path(os.path.abspath(root))
    for ancestor in reversed((root, *root.parents)):
        require(not is_link(ancestor.lstat()), 'input root/ancestor is a link or reparse point')
    require(stat.S_ISDIR(root.lstat().st_mode), 'input must be a directory')
    data = dict(schema=1, algorithm='sha256', config=cfg, complete=True, errors=[], excluded=[], files=[])
    prefixes = set(cfg['exclude'])
    pending, directories, seen, aliases, used = [root], [], 0, set(), 0
    def error(path, reason):
        data['errors'].append(dict(path=path, reason=reason))
    while pending:
        directory = pending.pop()
        relative_dir = None if directory == root else directory.relative_to(root).as_posix()
        try:
            info = directory.lstat()
            require(stat.S_ISDIR(info.st_mode) and not is_link(info), 'changed_during_read')
            directories.append((directory, stamp(info)))
            with os.scandir(directory) as entries:
                for entry in entries:
                    seen += 1
                    if seen > cfg['max_entries']:
                        error(None, 'limit')
                        pending.clear()
                        break
                    path = Path(entry.path)
                    relative = path.relative_to(root).as_posix()
                    try:
                        safe_path(relative)
                        require(relative.casefold() not in aliases, 'unsafe_path')
                        aliases.add(relative.casefold())
                    except ValueError:
                        error(None, 'unsafe_path')
                        continue
                    try:
                        if excluded(relative, prefixes):
                            data['excluded'].append(dict(path=relative, reason='configured'))
                            continue
                        info = path.lstat()
                        if is_link(info) or not (stat.S_ISDIR(info.st_mode) or stat.S_ISREG(info.st_mode)):
                            reason = 'link_or_reparse' if is_link(info) else 'special'
                            data['excluded'].append(dict(path=relative, reason=reason))
                            error(relative, reason)
                        elif stat.S_ISDIR(info.st_mode):
                            pending.append(path)
                        else:
                            require(info.st_size <= cfg['max_file_bytes'] and
                                    info.st_size <= cfg['max_total_bytes'] - used, 'limit')
                            used += info.st_size  # Reserve even if reading later fails.
                            item = read_file(path, info, info.st_size, cfg)
                            data['files'].append(dict(path=relative, **item))
                    except (OSError, ValueError) as exc:
                        error(relative, str(exc) if str(exc) in ('limit', 'changed_during_read') else 'file_failed')
        except (OSError, ValueError) as exc:
            error(relative_dir, 'changed_during_read' if str(exc) == 'changed_during_read' else 'scan_failed')
    for directory, original in directories:
        try:
            require(stamp(directory.lstat()) == original, 'changed_during_read')
        except (OSError, ValueError):
            error(None if directory == root else directory.relative_to(root).as_posix(), 'changed_during_read')
    data['complete'] = not data['errors']
    for name in ('files', 'excluded'):
        data[name].sort(key=lambda item: item['path'])
    validate(data)
    return data


def outside(output, root):
    require(not Path(output).resolve().is_relative_to(Path(root).resolve()), 'output must be outside input')


def write_json(output, data, root=None):
    if root is not None:
        outside(output, root)
    encoded = (json.dumps(data, ensure_ascii=False, indent=2) + '\n').encode('utf-8')
    require(len(encoded) <= MAX_JSON, 'JSON byte limit')
    with open(output, 'xb') as stream:
        stream.write(encoded)


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    commands = parser.add_subparsers(dest='command', required=True)
    capture = commands.add_parser('snapshot')
    capture.add_argument('root'); capture.add_argument('output')
    capture.add_argument('--exclude', action='append', default=[])
    capture.add_argument('--no-hash', action='store_true')
    for name, default in (('max-entries', 10000), ('max-file-bytes', 10485760), ('max-total-bytes', 104857600)):
        capture.add_argument('--' + name, type=int, default=default)
    diff = commands.add_parser('compare')
    for name in ('before', 'after', 'output'):
        diff.add_argument(name)
    args = parser.parse_args(argv)
    try:
        if args.command == 'snapshot':
            outside(args.output, args.root)
            cfg = config(args.exclude, not args.no_hash, args.max_entries, args.max_file_bytes, args.max_total_bytes)
            data = snapshot(args.root, cfg)
            write_json(args.output, data, args.root)
            print(f"snapshot complete={data['complete']} files={len(data['files'])} errors={len(data['errors'])}")
            return 0 if data['complete'] else 2
        data = compare(load(args.before), load(args.after))
        write_json(args.output, data)
        print('compare ' + json.dumps(data['counts'], ensure_ascii=False))
        return 0
    except (OSError, ValueError, TypeError, RecursionError) as exc:
        print(f'error: {exc}', file=sys.stderr)
        return 2


if __name__ == '__main__':
    sys.exit(main())