原创

Python 写一个逐文件 SHA-256 清单:从生成到校验完整交付目录

把目录发给另一台机器,文件名和数量都对,却仍不能证明内容完整。单独给压缩包算一个摘要可以验证那一份压缩文件;需要定位到具体哪个文件变化,则可以制作逐文件清单。

这次写一个小而完整的命令行工具:为静态平面目录记录每个文件的字节数与SHA-256,校验时同时检查缺失、多出、大小和摘要变化。它不递归,目录成员中的符号链接和子目录直接拒绝;这份范围是为了让操作对象明确,而不是宣称一百行代码能替代备份软件。

使用Python 3.12,只依赖标准库。目录必须在生成与校验期间保持稳定,不应由不可信进程同时修改。摘要清单保存到目录外,并通过另一个可信渠道保管;攻击者若能同时改文件和清单,普通哈希无法证明来源。

先写出清单里究竟保存什么

每项只有name、size和sha256,名字限定为单层文件名。清单还带格式版本与算法名称,避免未来改了编码或算法,却让旧校验器继续按原规则猜。内容以原始字节读取,不改变换行、编码或文件名大小写。

保存为file_manifest.py:

import argparse
import hashlib
import json
import os
import re
import stat
import sys
from pathlib import Path

MAX_FILES = 1000
MAX_MANIFEST = 1024 * 1024
HEX256 = re.compile(r'[0-9a-f]{64}\Z')


def flat_name(name):
    return (isinstance(name, str) and name not in ('', '.', '..')
            and '/' not in name and chr(92) not in name and chr(0) not in name)


def members(root):
    result = {}
    for path in root.iterdir():
        if not flat_name(path.name):
            raise ValueError('invalid flat file name')
        if path.is_symlink() or not path.is_file():
            raise ValueError(f'only regular flat files allowed: {path.name}')
        result[path.name] = path
        if len(result) > MAX_FILES:
            raise ValueError('too many files')
    return result


def fingerprint(path):
    digest = hashlib.sha256()
    with path.open('rb') as stream:
        start = os.fstat(stream.fileno())
        if not stat.S_ISREG(start.st_mode):
            raise ValueError(f'not a regular file: {path.name}')
        total = 0
        while block := stream.read(1024 * 1024):
            total += len(block)
            digest.update(block)
        end = os.fstat(stream.fileno())
    if (start.st_size, start.st_mtime_ns) != (end.st_size, end.st_mtime_ns):
        raise ValueError(f'file changed while reading: {path.name}')
    if total != end.st_size:
        raise ValueError(f'byte count changed: {path.name}')
    return {'name': path.name, 'size': total, 'sha256': digest.hexdigest()}


def load_manifest(path):
    with path.open('rb') as stream:
        raw = stream.read(MAX_MANIFEST + 1)
    if len(raw) > MAX_MANIFEST:
        raise ValueError('manifest too large')
    value = json.loads(raw)
    if (not isinstance(value, dict) or type(value.get('version')) is not int
            or value['version'] != 1 or value.get('algorithm') != 'sha256'):
        raise ValueError('unsupported manifest format')
    records = value.get('files')
    if not isinstance(records, list) or len(records) > MAX_FILES:
        raise ValueError('invalid record list')
    seen = set()
    for record in records:
        if not isinstance(record, dict) or set(record) != {'name', 'size', 'sha256'}:
            raise ValueError('invalid record fields')
        name, size, digest = record['name'], record['size'], record['sha256']
        if not flat_name(name):
            raise ValueError('invalid flat file name')
        if name in seen or type(size) is not int or size < 0:
            raise ValueError('duplicate name or invalid size')
        if not isinstance(digest, str) or not HEX256.fullmatch(digest):
            raise ValueError('invalid sha256 text')
        seen.add(name)
    return records


def main():
    parser = argparse.ArgumentParser(description='Static flat-directory integrity manifest')
    parser.add_argument('command', choices=['create', 'check'])
    parser.add_argument('directory', type=Path)
    parser.add_argument('manifest', type=Path)
    args = parser.parse_args()
    root = args.directory.resolve(strict=True)
    if not root.is_dir():
        raise ValueError('directory required')
    manifest = args.manifest.parent.resolve(strict=True) / args.manifest.name
    if manifest.is_symlink():
        raise ValueError('manifest must not be a symbolic link')
    if manifest == root or root in manifest.parents:
        raise ValueError('manifest must be outside the directory')
    actual = members(root)
    if args.command == 'create':
        records = [fingerprint(actual[name]) for name in sorted(actual)]
        document = {'version': 1, 'algorithm': 'sha256', 'files': records}
        text = json.dumps(document, ensure_ascii=False, indent=2) + '\n'
        if len(text.encode('utf-8')) > MAX_MANIFEST:
            raise ValueError('generated manifest exceeds limit')
        with manifest.open('x', encoding='utf-8') as stream:
            stream.write(text)
        print(f'created: {len(records)} files')
        return
    records = load_manifest(manifest)
    expected = {record['name']: record for record in records}
    missing = sorted(expected.keys() - actual.keys())
    extra = sorted(actual.keys() - expected.keys())
    if missing or extra:
        raise ValueError(f'membership mismatch; missing={missing}; extra={extra}')
    for name in sorted(expected):
        if fingerprint(actual[name]) != expected[name]:
            raise ValueError(f'content mismatch: {name}')
    print(f'verified: {len(records)} files')


if __name__ == '__main__':
    try:
        main()
    except (OSError, ValueError) as error:
        print(f'error: {error}', file=sys.stderr)
        raise SystemExit(1)

代码按固定大小块更新哈希,不需要把每个文件整个读到内存。块大小影响读取方式,不改变同一字节序列的SHA-256结果;不要把一次read返回多少字节误当作摘要分段协议。

记录同时保存大小,是为了错误定位与输入核对,不是说大小相同就可以省略摘要。两个不同文件完全可能字节数相同,而同一个内容的改名又会改变清单成员关系。

创建两份内容,再验证正常路径

在一个新的实验目录执行:

mkdir payload
printf 'alpha\n' > payload/a.txt
printf 'beta\n' > payload/b.txt
python3 file_manifest.py create payload manifest.json
python3 file_manifest.py check payload manifest.json

应依次显示created:2files和verified:2files。打开manifest.json,可以看见每个文件的原始字节数、十六进制摘要,以及明确的version与algorithm。

清单位于payload外,所以不会出现“算完清单,再把清单自己加入清单”的递归问题。程序也主动拒绝把输出放到输入目录内,而不是悄悄忽略某个固定文件名;忽略规则一旦变化,很容易漏掉本应核对的业务文件。

create使用独占创建,已有manifest.json会失败。重新生成应选择新清单路径或明确处理旧文件,避免一次误操作覆盖先前可信基线。这个写入没有宣称原子发布:磁盘写入失败仍可能留下不完整清单,需要检查并移除失败产物,再重新生成。

同长度修改,也必须被发现

将beta改成BETA,字节数不变:

printf 'BETA\n' > payload/b.txt
python3 file_manifest.py check payload manifest.json

命令应以非零退出,提示b.txt内容不符。此时不要立即重建清单让检查“恢复绿色”。先判断改变是否经过授权,保留旧清单,决定这是预期新版本还是传输损坏。清单校验只能报告差异,不能替发布流程批准变化。

恢复内容,再增加一个没有登记的文件:

printf 'beta\n' > payload/b.txt
printf 'new\n' > payload/extra.txt
python3 file_manifest.py check payload manifest.json

这次应报告extra。只遍历清单里原有条目会漏掉新增文件,因此程序把两边名字集合也比较。删除原文件则进入missing路径,不能因为剩余文件摘要都正确就宣布整个目录一致。

生成与校验必须使用同一份名字规则。这里抽成flat_name,既检查目录成员,也检查清单记录;为避免跨平台分隔符歧义,Unix上本可作为普通名称的反斜线也明确拒绝。否则create可能产出一份check随后不接受的清单。输入根路径会先规范化,成员类型拒绝与根路径解析是两项独立规则。

资源范围与并发边界写在明处

程序最多处理一千个平面文件,并限制清单读取字节,避免一份元数据文件无限增长。它没有限制每个内容文件大小,读取是流式的,超大文件仍会花费相应I/O时间。需要操作时间预算时,应由运行环境与任务管理提供,不能把小内存读取说成低成本读取。

符号链接与子目录都会拒绝,不会默默跳过。真实交付若需要递归,应设计相对路径规范、链接策略、重复路径、大小写差异与跨平台名称规则,再升级格式版本。仅把iterdir换成rglob,不会自动补全这些语义。

fingerprint在同一打开文件描述上比较读取前后的大小与修改时间,能发现一部分读取期间变化,却不是文件系统快照。并发修改方若改写后恢复相同元数据,或者在路径检查与打开之间替换对象,这份代码没有提供敌对环境下的完整防护。因此实验要求输入目录静态、受控;需要强一致备份时,应使用可信快照或停写边界。

对超大目录,多次成员读取之间也可能新增或删掉文件。逐文件摘要正确不能证明“同一瞬间的整个目录”成立。应让清单生成依附一个稳定发布目录,生成完成后再整体交付,不要直接拿线上持续写入目录做确定版本基线。

完整性与真实性别混成一项证明

SHA-256回答给定字节的指纹。若文件与清单都从同一个不可信位置下载,攻击者可以同时改两者;校验通过只说明它们彼此一致。可信发布可以通过签名或经过认证的渠道交付清单,并在接收端核对来源。

清单也不是备份,里面没有保存文件内容,发现损坏以后无法从摘要恢复原文。实际交付应保留可重新取得的源文件,并把版本、清单和内容对应起来。

最后删除extra.txt,再运行check,应恢复成功。再试把目录中的一项换成子目录,确认工具拒绝这份范围变化。正常、同长度改动、成员变化、对象类型变化各走一条清楚路径,才说明这份小工具检查的是已声明的交付集合。

正文到此结束
评论插件初始化中...
Loading...