用哈希链审计免费 LLM 模型输出：防止响应篡改

免费模型的响应不等于证据。

一个普通的 JSON 文件存储了客户端收到的内容，但它无法证明这个文件仍然与原始流程一致。哈希链将每次响应链接到前一条记录，这种链接使得静默编辑、删除和重排序变得可见。

MonkeyCode 提供的免费模型访问是本教程的合适目标。免责声明：本文是 MonkeyCode 产品推广的一部分。你也可以将同样的工作流应用于任何 HTTP 模型端点。如果你的端点是远程的，可以在本地或免费服务器上运行。

为什么普通日志会失败

一次典型的测试运行会为每个 prompt 记录一条日志。你可以编辑其中一条记录而保持其余不变，审查者无法判断哪条记录被改过。你也可以删除中间一条记录而不影响第一条和最后一条。简单的从备份恢复同样无法证明顺序。

哈希链以低成本解决了这个问题。每条记录存储前一条记录的哈希值，当前记录则与自身字段一起做哈希运算。如果任何早期记录被更改，下一条的 prev_hash 就不再匹配，导致从该点向后的链条断裂。

你需要构建两个小的 Python 3 文件：一个文件是模拟模型端点，另一个是哈希链客户端。客户端发送 prompt、提取响应并追加一条链接记录。--verify 标志会重新计算每条哈希并检查每条链接。

这仅使用 Python 标准库，无需任何 Web 框架或数据库。

Step 0: 创建项目

在终端中运行以下命令。

mkdir model-audit
cd model-audit
python3 -m venv .venv
source .venv/bin/activate
python --version

预期输出为 Python 3.10 或更高版本。如果版本较旧，先安装当前版本的 Python 3。

Step 1: 运行模拟端点

保存为 mock_model.py。

from http.server import BaseHTTPRequestHandler, HTTPServer
import json
import time

class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        length = int(self.headers.get('Content-Length', 0))
        body = self.rfile.read(length)
        response = {
            'choices': [
                {
                    'message': {
                        'content': f'mock:{time.time_ns()}'
                    }
                }
            ]
        }
        data = json.dumps(response).encode()
        self.send_response(200)
        self.send_header('Content-Type', 'application/json')
        self.send_header('Content-Length', str(len(data)))
        self.end_headers()
        self.wfile.write(data)

if __name__ == '__main__':
    HTTPServer(('127.0.0.1', 8080), Handler).serve_forever()

python mock_model.py &

curl -sS -X POST http://127.0.0.1:8080/v1/chat/completions -H 'Content-Type: application/json' -d '{"messages":[{"role":"user","content":"ping"}]}'

你应该看到一个包含 content 值的 JSON 对象。

Step 2: 添加哈希链客户端

保存为 audit.py。

"""Tamper-evident log for model responses."""
import argparse
import hashlib
import json
import os
import time
import urllib.request

LOG_FILE = os.environ.get('LOG_FILE', 'audit.jsonl')
GENESIS = 'GENESIS'

def call_model(url, prompt, token):
    body = json.dumps({
        'messages': [{'role': 'user', 'content': prompt}]
    }).encode()
    headers = {'Content-Type': 'application/json'}
    if token:
        headers['Authorization'] = f'Bearer {token}'
    request = urllib.request.Request(url, data=body, headers=headers)
    with urllib.request.urlopen(request, timeout=10) as response:
        return response.read().decode()

def extract_text(raw):
    data = json.loads(raw)
    try:
        return data['choices'][0]['message']['content']
    except (KeyError, IndexError, TypeError):
        return raw

def load_prev():
    if not os.path.exists(LOG_FILE):
        return GENESIS
    with open(LOG_FILE) as f:
        last = None
        for line in f:
            last = json.loads(line)
        return last['hash'] if last else GENESIS

def append_entry(prev_hash, request_id, prompt, response):
    payload = {
        'prev_hash': prev_hash,
        'request_id': request_id,
        'prompt': prompt,
        'response': response,
        'timestamp': time.time_ns(),
    }
    payload['hash'] = hashlib.sha256(
        json.dumps(payload, sort_keys=True).encode()
    ).hexdigest()
    with open(LOG_FILE, 'a') as f:
        f.write(json.dumps(payload, sort_keys=True) + chr(10))
    return payload

def verify_log():
    prev = GENESIS
    with open(LOG_FILE) as f:
        for line in f:
            record = json.loads(line)
            if record.get('prev_hash') != prev:
                return False, f"broken link at {record.get('request_id')}"
            stored = record.get('hash')
            copy = {k: v for k, v in record.items() if k != 'hash'}
            recomputed = hashlib.sha256(
                json.dumps(copy, sort_keys=True).encode()
            ).hexdigest()
            if stored != recomputed:
                return False, f"bad hash at {record.get('request_id')}"
            prev = stored
    return True, 'ok'

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--url', default='http://127.0.0.1:8080/v1/chat/completions')
    parser.add_argument('--token', default=os.environ.get('MODEL_TOKEN', ''))
    parser.add_argument('--prompt', default='ping')
    parser.add_argument('--iterations', type=int, default=1)
    parser.add_argument('--verify', action='store_true')
    args = parser.parse_args()

    if args.verify:
        ok, message = verify_log()
        print(json.dumps({'ok': ok, 'detail': message}))
        raise SystemExit(0 if ok else 1)

    for i in range(args.iterations):
        raw = call_model(args.url, args.prompt, args.token)
        text = extract_text(raw)
        prev = load_prev()
        request_id = f'{int(time.time() * 1000)}-{i}'
        entry = append_entry(prev, request_id, args.prompt, text)
        print(json.dumps({'request_id': entry['request_id'], 'hash': entry['hash'][:12]}))

if __name__ == '__main__':
    main()

客户端仅在端点需要时接受 MODEL_TOKEN。

Step 3: 生成链接记录

运行客户端三次。

python audit.py --prompt "hello" --iterations 3

你应该看到三个短哈希值。检查日志。

cat audit.jsonl

每行包含 prev_hash、prompt、response 和 hash。第一行使用 GENESIS 作为其前一条哈希。

Step 4: 验证链条

运行验证命令。

python audit.py --verify

{"ok": true, "detail": "ok"}

验证通过意味着每条链接和每条哈希都是一致的。

Step 5: 故意破坏链条

在不更新其哈希的情况下编辑最后一条响应。

python - <<'PY'
import json
path = 'audit.jsonl'
lines = open(path).read().splitlines()
record = json.loads(lines[-1])
record['response'] = 'changed after the fact'
lines[-1] = json.dumps(record, sort_keys=True)
open(path, 'w').write(chr(10).join(lines) + chr(10))
PY

再次运行验证。

python audit.py --verify

预期输出报告 ok: false 和一条 bad hash 详情。这个失败正是重点所在。链条无需与外部副本比较就能检测到编辑。

哈希链证明了内部一致性。它不能证明模型输出是正确的，不能验证 schema、真实性或安全性，也不能防止能重写每条哈希的攻击者。如需此类保护，请使用签名记录或只追加存储。

时间戳使用系统墙上时钟。如果需要间隔测量，请使用 time.monotonic_ns()。将日志保存在可信目录中。如果日志写入程序可以被替换，链条就可以被重建。

谁不应该使用此方法

如果需要监管证明，请跳过此方法。如果多个进程同时追加，请跳过。如果日志增长速度快于磁盘轮换计划，请跳过。这些情况请使用数据库或签名日志系统。

对于针对免费模型端点的本地测试，这个脚本已经足够。如果你通过 MonkeyCode 获得免费模型访问，在缓存任何内容之前先从这里开始。