Notebook

第10章 学生Notebook：受控学习便签助手

模型没有直接写文件的魔法——工具调用是"提议"，宿主程序才"执行"

本Notebook完成以下任务：

  1. 加载工具 Schema，理解参数约束
  2. 运行正常工具调用流程（创建笔记 → 验证存在）
  3. 对比"模型声称已创建"与"日志证明已执行"
  4. 测试5种攻击场景，观察防御机制
  5. 查看审计日志
  6. B档修改：修改一条 Schema 限制或新增一个低风险工具

准备：确认 data/tool_schemas.json 存在。

1. 加载工具 Schema

读取三个工具的定义，理解每个工具的参数约束和权限。

import json
import re
import os
import tempfile
import shutil
from pathlib import Path
from datetime import datetime

# 加载工具 Schema
schema_path = Path('../data/tool_schemas.json')
with open(schema_path, 'r', encoding='utf-8') as f:
    schema_data = json.load(f)

tools = {t['name']: t for t in schema_data['tools']}
global_rules = schema_data['global_rules']

print(f'工具数量: {len(tools)}')
print(f'全局规则: {json.dumps(global_rules, ensure_ascii=False, indent=2)}')
print()
for name, tool in tools.items():
    print(f'工具: {name}')
    print(f'  说明: {tool["description"]}')
    print(f'  权限: {tool["permissions"]}')
    print(f'  风险: {tool["risk_level"]}')
    params = tool['parameters'].get('properties', {})
    if params:
        for pname, pdef in params.items():
            print(f'  参数 {pname}: {pdef.get("type", "?")}', end='')
            if 'pattern' in pdef:
                print(f', 正则: {pdef["pattern"]}', end='')
            if 'maxLength' in pdef:
                print(f', 最大长度: {pdef["maxLength"]}', end='')
            print()
    print()

2. 搭建沙盒与审计日志

创建临时沙盒目录和审计日志。所有文件操作都限制在这个目录中。

运行：观察沙盒路径。

# 创建沙盒目录（使用临时目录，安全隔离）
SANDBOX_ROOT = Path(tempfile.mkdtemp(prefix='ch10_sandbox_'))
AUDIT_LOG = []  # 内存中的审计日志

print(f'沙盒目录: {SANDBOX_ROOT}')
print(f'沙盒目录存在: {SANDBOX_ROOT.exists()}')
print()
print('安全说明:')
print('  - 所有文件操作限制在沙盒目录内')
print('  - 不访问沙盒外的任何文件')
print('  - 每次调用都记录审计日志')
print('  - 课程结束后可安全删除整个沙盒')

3. 实现校验层

这是本章的核心代码：模型只提议，程序来校验。

校验分为四层：Schema 校验 → 路径安全检查 → 内容安全检查 → 权限检查。

运行：阅读完整代码，理解每一层检查什么。

def validate_schema(tool_name, arguments):
    """第一层：Schema 校验 —— 参数类型、格式、长度是否符合工具定义"""
    if tool_name not in tools:
        return False, f'unknown_tool: {tool_name}'
    
    tool = tools[tool_name]
    schema = tool['parameters']
    
    # 检查必填参数
    for req in schema.get('required', []):
        if req not in arguments:
            return False, f'missing_required: {req}'
    
    # 检查额外属性
    if schema.get('additionalProperties') is False:
        allowed = set(schema.get('properties', {}).keys())
        extra = set(arguments.keys()) - allowed
        if extra:
            return False, f'additional_properties: {extra}'
    
    # 检查每个参数的类型和格式
    properties = schema.get('properties', {})
    for pname, pvalue in arguments.items():
        if pname not in properties:
            continue
        pdef = properties[pname]
        
        # 类型检查
        if pdef.get('type') == 'string' and not isinstance(pvalue, str):
            return False, f'type_error: {pname} should be string'
        
        # 正则检查
        if 'pattern' in pdef and isinstance(pvalue, str):
            if not re.match(pdef['pattern'], pvalue):
                return False, f'pattern_error: {pname} does not match {pdef["pattern"]}'
        
        # 长度检查
        if 'maxLength' in pdef and isinstance(pvalue, str):
            if len(pvalue) > pdef['maxLength']:
                return False, f'maxLength_error: {pname} is {len(pvalue)} chars, max {pdef["maxLength"]}'
    
    return True, 'schema_ok'


def validate_path_safety(arguments):
    """第二层：路径安全检查 —— 防止路径穿越"""
    for key, value in arguments.items():
        if not isinstance(value, str):
            continue
        for blocked in ['..', '/', '\\']:
            if blocked in value:
                return False, f'path_traversal: "{blocked}" found in {key}'
    return True, 'path_ok'


def validate_content_safety(arguments):
    """第三层：内容安全检查 —— 防止文档注入"""
    for key, value in arguments.items():
        if not isinstance(value, str):
            continue
        value_lower = value.lower()
        for blocked in global_rules['blocked_patterns']:
            if blocked.lower() in value_lower:
                return False, f'blocked_pattern: "{blocked}" found in {key}'
    return True, 'content_ok'


def validate_all(tool_name, arguments):
    """依次运行四层校验，任何一层失败则拒绝"""
    checks = [
        ('schema', validate_schema(tool_name, arguments)),
        ('path_safety', validate_path_safety(arguments)),
        ('content_safety', validate_content_safety(arguments)),
    ]
    
    for layer_name, (passed, reason) in checks:
        if not passed:
            return False, layer_name, reason
    
    return True, 'all_passed', 'all_ok'


print('校验层已定义:')
print('  1. validate_schema    — 参数类型、格式、长度、额外属性')
print('  2. validate_path_safety  — 路径穿越（.. / \\）')
print('  3. validate_content_safety — 文档注入（<script> exec() 等）')
print('  4. validate_all       — 依次运行以上三层')

4. 实现工具执行与审计日志

只有校验全部通过，才在沙盒中执行。每次调用（无论成功失败）都写入审计日志。

运行：阅读执行逻辑。

def execute_tool(tool_name, arguments):
    """在沙盒中执行工具调用，返回结果"""
    if tool_name == 'create_note':
        filepath = SANDBOX_ROOT / arguments['filename']
        filepath.write_text(arguments['content'], encoding='utf-8')
        return {'status': 'created', 'path': str(filepath), 'size': len(arguments['content'])}
    
    elif tool_name == 'list_files':
        files = [f.name for f in SANDBOX_ROOT.iterdir() if f.is_file()]
        return {'status': 'listed', 'files': sorted(files), 'count': len(files)}
    
    elif tool_name == 'calc':
        expr = arguments['expression']
        # 只允许安全的数学表达式
        if not re.match(r'^[0-9+\-*/().\s]+$', expr):
            return {'status': 'error', 'message': 'invalid_expression'}
        try:
            result = eval(expr, {'__builtins__': {}}, {})  # 安全求值，无内置函数
            return {'status': 'computed', 'expression': expr, 'result': result}
        except Exception as e:
            return {'status': 'error', 'message': str(e)}
    
    return {'status': 'error', 'message': f'unknown_tool: {tool_name}'}


def process_tool_call(tool_name, arguments):
    """完整的工具调用流程：提议 → 校验 → 执行 → 日志"""
    timestamp = datetime.now().isoformat()
    
    # 记录提议
    log_entry = {
        'timestamp': timestamp,
        'tool': tool_name,
        'arguments': {k: (v[:50] + '...' if isinstance(v, str) and len(v) > 50 else v)
                      for k, v in arguments.items()},
        'stages': {}
    }
    
    # 阶段1：校验
    passed, layer, reason = validate_all(tool_name, arguments)
    log_entry['stages']['validation'] = {
        'passed': passed,
        'layer': layer,
        'reason': reason
    }
    
    if not passed:
        log_entry['stages']['execution'] = {'status': 'REJECTED', 'reason': reason}
        log_entry['final'] = 'REJECTED'
        AUDIT_LOG.append(log_entry)
        return {'result': None, 'log': log_entry}
    
    # 阶段2：执行
    try:
        result = execute_tool(tool_name, arguments)
        log_entry['stages']['execution'] = {'status': 'EXECUTED', 'result_summary': str(result)[:100]}
        log_entry['final'] = 'EXECUTED'
    except Exception as e:
        log_entry['stages']['execution'] = {'status': 'ERROR', 'message': str(e)}
        log_entry['final'] = 'ERROR'
        result = {'status': 'error', 'message': str(e)}
    
    AUDIT_LOG.append(log_entry)
    return {'result': result, 'log': log_entry}


print('工具执行引擎已定义:')
print('  process_tool_call(tool_name, arguments)')
print('  → 校验 → 执行 → 审计日志')
print('  返回: {result, log}')

5. 模拟模型输出

在真实场景中，模型会生成 tool_call JSON。这里我们使用预录的模型输出来模拟。

这些 JSON 就是模型"提议"的内容——注意，它们只是文字，不是真正的操作。

运行：查看预录的模型输出。

# 预录的模型输出（模拟 tool_call JSON）
mock_model_outputs = [
    {
        'description': '正常：创建今日学习清单',
        'tool_call': {
            'tool': 'create_note',
            'arguments': {
                'filename': '今日清单.md',
                'content': '1. 复习工具调用流程\n2. 完成攻击测试\n3. 写实验报告'
            }
        }
    },
    {
        'description': '正常：列出沙盒文件',
        'tool_call': {
            'tool': 'list_files',
            'arguments': {}
        }
    },
    {
        'description': '正常：计算学习时长',
        'tool_call': {
            'tool': 'calc',
            'arguments': {
                'expression': '(45 + 30) * 2 / 60'
            }
        }
    },
    {
        'description': '攻击：路径穿越',
        'tool_call': {
            'tool': 'create_note',
            'arguments': {
                'filename': '../../etc/passwd',
                'content': 'hacked'
            }
        }
    },
    {
        'description': '攻击：超长内容',
        'tool_call': {
            'tool': 'create_note',
            'arguments': {
                'filename': 'long_note.md',
                'content': 'A' * 1000
            }
        }
    },
    {
        'description': '攻击：未声明字段',
        'tool_call': {
            'tool': 'create_note',
            'arguments': {
                'filename': 'test.md',
                'content': 'hello',
                'shell': True
            }
        }
    },
    {
        'description': '攻击：文档注入',
        'tool_call': {
            'tool': 'create_note',
            'arguments': {
                'filename': 'injected.md',
                'content': '<script>alert("xss")</script>'
            }
        }
    },
    {
        'description': '攻击：危险计算',
        'tool_call': {
            'tool': 'calc',
            'arguments': {
                'expression': "__import__('os').system('echo hacked')"
            }
        }
    }
]

print(f'预录模型输出数量: {len(mock_model_outputs)}')
print()
for i, item in enumerate(mock_model_outputs):
    tc = item['tool_call']
    print(f'  [{i+1}] {item["description"]}')
    print(f'      tool={tc["tool"]}', end='')
    if tc['arguments']:
        args_preview = {k: (v[:30]+'...' if isinstance(v,str) and len(v)>30 else v)
                        for k, v in tc['arguments'].items()}
        print(f'  args={args_preview}')
    else:
        print('  args={}')

6. 正常调用流程

运行前三个正常调用，观察完整的"提议→校验→执行→日志"四段证据。

运行：观察每次调用的结果。

print('=' * 60)
print('正常调用流程')
print('=' * 60)

for item in mock_model_outputs[:3]:
    tc = item['tool_call']
    print(f'\n--- {item["description"]} ---')
    print(f'提议: tool={tc["tool"]}, args={tc["arguments"]}')
    
    response = process_tool_call(tc['tool'], tc['arguments'])
    log = response['log']
    
    print(f'校验: passed={log["stages"]["validation"]["passed"]}, '
          f'layer={log["stages"]["validation"]["layer"]}')
    print(f'执行: {log["stages"]["execution"]["status"]}')
    if response['result']:
        print(f'结果: {response["result"]}')
    print(f'最终: {log["final"]}')

验证文件确实存在

模型说"我已经创建了文件"，但我们要用 list_files 工具来实际验证。

运行：对比"模型声称"与"日志证明"。

print('=' * 60)
print('"模型声称" vs "日志证明"')
print('=' * 60)

# 模型声称
print('\n模型声称: "我已经创建了今日清单.md"')
print('  → 这只是一段文字，模型没有文件操作能力')

# 日志证明
print('\n日志证明:')
for entry in AUDIT_LOG:
    if entry['tool'] == 'create_note' and entry['final'] == 'EXECUTED':
        print(f'  [{entry["timestamp"]}] create_note '
              f'filename={entry["arguments"]["filename"]} → EXECUTED')

# 实际验证
print('\n实际验证（调用 list_files）:')
response = process_tool_call('list_files', {})
if response['result']:
    print(f'  沙盒中的文件: {response["result"]["files"]}')
    print(f'  文件数量: {response["result"]["count"]}')

# 直接用文件系统验证
print('\n文件系统直接验证:')
actual_files = [f.name for f in SANDBOX_ROOT.iterdir() if f.is_file()]
print(f'  沙盒目录实际文件: {actual_files}')
print(f'  "今日清单.md" 存在: {("今日清单.md" in actual_files)}')

print('\n结论: 只有审计日志 + 文件系统验证才能证明操作确实执行。')
print('      模型的文字输出不等于执行证据。')

思考：

如果模型说"我已经创建了文件"，但审计日志中没有对应记录，可能是什么原因？

在下方写下你的分析。

# 我的分析：
# 
# 可能原因1：
# 可能原因2：
# 可能原因3：

7. 攻击测试

运行5种攻击场景，观察每层防御如何拦截非法调用。

运行：观察每种攻击被哪一层防御拦截。

print('=' * 60)
print('攻击测试')
print('=' * 60)

attack_results = []

for item in mock_model_outputs[3:]:  # 从第4个开始是攻击
    tc = item['tool_call']
    print(f'\n--- {item["description"]} ---')
    
    # 显示攻击输入（截断长内容）
    args_display = {}
    for k, v in tc['arguments'].items():
        if isinstance(v, str) and len(v) > 60:
            args_display[k] = v[:60] + f'... ({len(v)} chars)'
        else:
            args_display[k] = v
    print(f'攻击输入: tool={tc["tool"]}, args={args_display}')
    
    response = process_tool_call(tc['tool'], tc['arguments'])
    log = response['log']
    
    validation = log['stages']['validation']
    execution = log['stages']['execution']
    
    print(f'校验结果: passed={validation["passed"]}')
    if not validation['passed']:
        print(f'拦截层: {validation["layer"]}')
        print(f'拦截原因: {validation["reason"]}')
    print(f'执行状态: {execution["status"]}')
    print(f'最终: {log["final"]}')
    
    attack_results.append({
        'attack': item['description'],
        'blocked': not validation['passed'],
        'layer': validation.get('layer', 'N/A'),
        'reason': validation.get('reason', 'N/A')
    })

print('\n' + '=' * 60)
print('攻击防御汇总')
print('=' * 60)
print(f'{"攻击类型":<20} {"拦截":<6} {"防御层":<18} {"原因"}')
print('-' * 80)
for r in attack_results:
    status = '✓ 拦截' if r['blocked'] else '✗ 放行'
    print(f'{r["attack"]:<20} {status:<6} {r["layer"]:<18} {r["reason"]}')

思考：

  1. 路径穿越攻击被哪一层拦截？如果没有这一层会怎样？
  2. 文档注入攻击被哪一层拦截？为什么内容安全检查很重要？
  3. 未声明字段攻击被哪一层拦截？additionalProperties: false 的作用是什么？

在下方写下你的分析。

# 我的分析：
# 
# 1. 路径穿越：
# 
# 2. 文档注入：
# 
# 3. 未声明字段：

8. 审计日志查看器

查看完整的审计日志。每条记录包含：时间戳、工具名、参数摘要、校验结果、执行结果。

运行：查看所有调用记录。

print('=' * 60)
print('完整审计日志')
print('=' * 60)

for i, entry in enumerate(AUDIT_LOG):
    print(f'\n--- 记录 #{i+1} ---')
    print(f'时间: {entry["timestamp"]}')
    print(f'工具: {entry["tool"]}')
    print(f'参数: {entry["arguments"]}')
    print(f'校验: passed={entry["stages"]["validation"]["passed"]}, '
          f'layer={entry["stages"]["validation"]["layer"]}')
    print(f'执行: {entry["stages"]["execution"]["status"]}')
    print(f'最终: {entry["final"]}')

# 统计
executed = sum(1 for e in AUDIT_LOG if e['final'] == 'EXECUTED')
rejected = sum(1 for e in AUDIT_LOG if e['final'] == 'REJECTED')
print(f'\n--- 统计 ---')
print(f'总调用次数: {len(AUDIT_LOG)}')
print(f'成功执行: {executed}')
print(f'被拒绝: {rejected}')
print(f'拦截率: {rejected}/{len(AUDIT_LOG)} = {rejected/len(AUDIT_LOG)*100:.1f}%')

导出审计日志

将审计日志保存为文件，作为交付物的一部分。

# 导出审计日志为 JSON
audit_path = SANDBOX_ROOT / 'audit_log.json'
with open(audit_path, 'w', encoding='utf-8') as f:
    json.dump(AUDIT_LOG, f, ensure_ascii=False, indent=2)

# 同时导出为可读的文本格式
audit_txt_path = SANDBOX_ROOT / 'audit.log'
with open(audit_txt_path, 'w', encoding='utf-8') as f:
    for entry in AUDIT_LOG:
        ts = entry['timestamp']
        tool = entry['tool']
        final = entry['final']
        args_str = ', '.join(f'{k}={v}' for k, v in entry['arguments'].items())
        reason = ''
        if not entry['stages']['validation']['passed']:
            reason = f' (reason={entry["stages"]["validation"]["reason"]})'
        f.write(f'[{ts}] PROPOSE tool={tool} {args_str} -> {final}{reason}\n')

print(f'审计日志已导出:')
print(f'  JSON: {audit_path}')
print(f'  文本: {audit_txt_path}')
print()
print('文本格式预览:')
print(audit_txt_path.read_text(encoding='utf-8'))

9. B档修改：修改一个变量

从以下两项中选择一项修改，修改前先写预测。

选项A：修改一条 Schema 限制

  • 例如：把 create_note 的内容长度限制从 500 改为 1000
  • 或者：把文件名规则改为也允许 .txt 后缀

选项B：新增一个低风险工具 append_checklist

  • 功能：向已有笔记末尾追加一行待办事项
  • 要求：定义完整的 JSON Schema、权限和风险等级
  • 测试：正常调用 + 至少2种攻击测试

运行前先写预测。

# 我选择：选项___
# 
# 修改前预测：
# 这个修改会扩大什么能力：
# 这个修改会新增什么风险：
# 需要增加什么测试：
#

选项A：修改 Schema 限制

取消下面的注释并修改。

# === 选项A：修改 Schema 限制 ===

# 示例1：修改内容长度限制
# original_max = tools['create_note']['parameters']['properties']['content']['maxLength']
# tools['create_note']['parameters']['properties']['content']['maxLength'] = 1000  # 修改这个值
# print(f'内容长度限制: {original_max} → 1000')

# 示例2：修改文件名规则，允许 .txt
# original_pattern = tools['create_note']['parameters']['properties']['filename']['pattern']
# tools['create_note']['parameters']['properties']['filename']['pattern'] = r'^[a-zA-Z0-9_\u4e00-\u9fa5\-]+\.(md|txt)$'
# print(f'文件名规则: {original_pattern}')
# print(f'         → {tools["create_note"]["parameters"]["properties"]["filename"]["pattern"]}')

# 修改后，重新测试正常用例和攻击用例
# 观察：哪些之前被拒绝的调用现在可以通过了？
# 新增了什么风险？

选项B：新增低风险工具

取消下面的注释并修改。

# === 选项B：新增 append_checklist 工具 ===

# 定义新工具的 Schema
# new_tool = {
#     'name': 'append_checklist',
#     'description': '向已有笔记末尾追加一行待办事项',
#     'parameters': {
#         'type': 'object',
#         'properties': {
#             'filename': {
#                 'type': 'string',
#                 'description': '已有笔记的文件名',
#                 'pattern': '^[a-zA-Z0-9_\\u4e00-\\u9fa5\\-]+\\.md$'
#             },
#             'item': {
#                 'type': 'string',
#                 'description': '要追加的待办事项',
#                 'maxLength': 100
#             }
#         },
#         'required': ['filename', 'item'],
#         'additionalProperties': False
#     },
#     'permissions': {
#         'filesystem': 'append',
#         'scope': 'sandbox_only',
#         'network': 'none'
#     },
#     'risk_level': 'low'
# }

# 注册新工具
# tools['append_checklist'] = new_tool

# 实现执行逻辑（在 execute_tool 函数中添加）
# elif tool_name == 'append_checklist':
#     filepath = SANDBOX_ROOT / arguments['filename']
#     if not filepath.exists():
#         return {'status': 'error', 'message': 'file_not_found'}
#     with open(filepath, 'a', encoding='utf-8') as f:
#         f.write('\n- ' + arguments['item'])
#     return {'status': 'appended', 'file': arguments['filename'], 'item': arguments['item']}

# 测试新工具
# print('测试 append_checklist:')
# r1 = process_tool_call('append_checklist', {'filename': '今日清单.md', 'item': '4. 复习攻击测试'})
# print(f'  正常调用: {r1["log"]["final"]}')
# r2 = process_tool_call('append_checklist', {'filename': '../../etc/passwd', 'item': 'hacked'})
# print(f'  路径穿越: {r2["log"]["final"]}')
# r3 = process_tool_call('append_checklist', {'filename': 'test.md', 'item': 'x'*200})
# print(f'  超长内容: {r3["log"]["final"]}')

修改后验证

修改后重新运行正常用例和攻击用例，确认修改生效且没有引入新问题。

# 修改后验证
# 取消注释并运行你需要的测试

# 重新运行正常用例
# print('修改后的正常用例测试:')
# r = process_tool_call('create_note', {'filename': 'test_b.md', 'content': 'B档修改测试'})
# print(f'  创建笔记: {r["log"]["final"]}')

# 重新运行攻击用例
# print('修改后的攻击测试:')
# r = process_tool_call('create_note', {'filename': '../../etc/passwd', 'content': 'hacked'})
# print(f'  路径穿越: {r["log"]["final"]} (应仍为 REJECTED)')

# 我的观察：
# 修改前后对比：
# 新增风险分析：

10. 清理与验收

清理沙盒目录，运行验收检查。

# 清理沙盒
shutil.rmtree(SANDBOX_ROOT, ignore_errors=True)
print(f'沙盒已清理: {SANDBOX_ROOT}')
print()

# 验收清单
checks = {
    '加载了工具 Schema': True,
    '运行了正常调用流程': True,
    '对比了"模型声称"与"日志证明"': True,
    '运行了攻击测试': True,
    '查看了审计日志': True,
    '写了攻击分析': False,       # 检查第7节的分析单元格
    '完成了B档修改': False,       # 检查第9节是否有修改
}

print('验收清单:')
for check, status in checks.items():
    print(f'  [{"✓" if status else " "}] {check}')

print()
print('请手动把 False 改为 True，确认你完成了对应项目。')
print()
print('关键概念检查:')
print('  1. 工具调用中，模型的角色是 ___，宿主程序的角色是 ___')
print('  2. JSON Schema 的作用是 ___')
print('  3. 路径穿越攻击的原理是 ___，防御方法是 ___')
print('  4. 审计日志的价值是 ___')
print('  5. "模型声称已创建"和"日志证明已执行"的区别是 ___')