chunk-invariance

本地服务

github.com安全133 天前更新

代码仓库访问官网

关于 chunk-invariance

chunk-invariance 用于检查 Python 流式过滤器在不同分块下的输出是否与整段过滤一致，适合验证脱敏器、护栏和改写器。可选 MCP 服务支持助手执行检查并生成 pytest 测试。

主要功能

01检查不同分块下的输出
02报告首个失败的分块
03支持同步与异步过滤器
04生成 pytest 回归测试
05提供可选 Hypothesis 策略

适用场景

01测试脱敏是否漏掉跨块密钥
02比较流式与整段过滤结果
03添加分块边界回归测试

MCP 相关

可选 MCP 服务提供检查流式函数、生成 pytest 回归测试和解释逐块泄漏的工具。它会在本机独立进程中导入并运行函数。

配置

本地服务的配置在各客户端中通用，复制后粘贴到客户端的 MCP 配置文件即可。

mcp.json · 通用配置

{
  "mcpServers": {
    "ninadphalak-chunk-invariance": {
      "command": "uvx",
      "args": [
        "chunk-invariance"
      ]
    }
  }
}

启动命令uvx chunk-invariance

本地服务需要在你的电脑上安装运行环境（如 Node.js 18+）。各客户端的配置文件位置不同，请参考客户端文档。

安装包

pypichunk-invariance0.2.0stdio

工具

收录的工具列表 · 更新于 1 周前

参数	类型	必填	说明
shape	enum	否	generator：target(chunks) 接收文本块可迭代对象，并生成或返回输出块。per_chunk：target(chunk, state) 返回单个块的输出；state 是在整个流中保留的 dict；通过 flush_target 指定流结束步骤。async_generator：target 接收块的异步可迭代对象，并生成输出。generator \| per_chunk \| async_generator
python	any	否	用于导入用户代码的解释器。默认：如果 project_dir/.venv 或 project_dir/venv 存在，则使用它；否则使用服务器自带的 Python。
splits	any	否	'all-two-part'（默认）：整段文本加上每个单切分点。一个最大为 3 的整数：拆分为最多这么多段的所有方式（较慢）。
target	string	是	流式过滤器，格式为 package.module:function，可从 project_dir 导入，方式与用户测试中的导入方式相同。
samples	any	否	包含过滤器应脱敏或改写的值的输入文本。默认：合成样本（电子邮件、SSN、卡号、API 密钥格式；均为虚构）：['Contact [email protected] about the invoice.', 'SSN on file: 987-65-4321 (fake).', 'Card 4111 1111 1111 1111 exp 12/30 (test card).', 'Use key sk-FAKEab12cd34ef56gh78 for the demo.', 'aws_access_key_id = AKIAIOSFODNN7EXAMPLE', 'token ghp_EXAMPLENOTAREALGITHUBTOKEN0000000000 ok']。请传入符合此过滤器匹配格式的虚构值（例如，按提供商格式输入的密钥），否则检查毫无意义。绝不要使用真实机密或个人数据：样本会在结果和生成的测试文件中回显。
project_dir	string	是	用户项目根目录的绝对路径。target 会从此处（以及其 src/ 目录）导入。
flush_target	any	否	仅适用于 per_chunk：flush(state) 的 package.module:function，在最后一个块之后调用一次。
whole_target	any	否	非流式函数，格式为 package.module:function，接收整段文本并返回过滤后的文本。默认：将流式过滤器作为单个块运行于整段文本；这无法捕获缺少流结束刷新步骤的情况（单个块会丢弃相同的尾部内容），因此如果存在此函数，请提供。
timeout_seconds	number	否	检查进程停止前等待的秒数。

详细介绍

chunk-invariance

流式过滤器（脱敏器、护栏或改写器）只有在以下条件成立时才算正确：无论以何种方式将输入拆分为多个块，拼接后的流式输出都必须等于一次性过滤完整输入的输出。这个包将该规则转化为一条断言。

from chunk_invariance import assert_chunk_invariant

def redact(text):                 # your filter on the whole input
    return text.replace("sk-AbCd1234", "[KEY]")

def redact_stream(chunks):        # your filter on a stream: takes chunks, yields chunks
    for chunk in chunks:
        yield redact(chunk)       # redacts each chunk alone

def test_streaming_redaction():
    assert_chunk_invariant(redact_stream, redact, "use sk-AbCd1234 now")

该测试会失败，并指出导致问题的最小拆分方式：

chunk_invariance.core.ChunkInvarianceError: stream filter is not chunk-invariant
  input:    'use sk-AbCd1234 now'
  split:    ['use s', 'k-AbCd1234 now'] (cut at 5)
  whole:    'use [KEY] now'
  streamed: 'use sk-AbCd1234 now'
  split #6 in the order tried; generated splits run fewest parts first

pip install chunk-invariance
pip install "chunk-invariance[hypothesis]"   # optional Hypothesis strategies
pip install "chunk-invariance[mcp]"          # optional MCP server for AI assistants

它能检测什么

只单独检查每个块的过滤器，因此跨两个块拆分的值两部分都无法匹配。
在每个块前添加上一个块固定长度内容作为缓冲的过滤器。匹配虽然触发了，但值的开头已经随上一个块输出。
缓存长度短于匹配模式、缺少流结束时的刷新，以及输出遭到破坏而非泄露（任何与完整输入输出不一致的情况都会导致测试失败）。

chunk_invariance.examples 提供了同一匹配模式下的一个正确过滤器（hold_back）和两种错误设计（design_a、design_b），作为对照示例。

它不能检测什么

它测试的是可在进程内调用的过滤器，不会测试运行中的网关。网关中的块由上游模型、网络和 SSE 解析器决定。对此，请使用 pii-leak-benchmark，它会将分段响应发送到网关，并检查客户端收到的内容。

它只检查你指定的拆分方式。默认会检查完整输入以及每个单字符位置的拆分，这可以检测被拆成两部分的值。splits="all" 会尝试较短输入的所有拆分方式（最多 16 个字符）；splits=3 会尝试拆分为至多三个部分的所有方式；显式指定的块列表则会逐一按指定方式运行。它无法证明过滤器对于你未提供的输入也正确。

通过 AI 助手使用

你在使用 Claude Code、Cursor 或 Claude Desktop 吗？这个包提供了可选的 MCP 服务。借助它，你可以让助手对自己的流式函数运行逐一拆分检查，然后将测试添加到你的代码库中。

pip install "chunk-invariance[mcp]"      # Python 3.10 or later
claude mcp add chunk-invariance -- chunk-invariance-mcp

对于 Claude Desktop，请将以下内容添加到 claude_desktop_config.json（如果 Claude Desktop 无法在 PATH 中找到命令，请填写命令的完整路径）：

{
  "mcpServers": {
    "chunk-invariance": { "command": "chunk-invariance-mcp" }
  }
}

然后可以这样提问：“检查 myapp/filters.py 中的 redact_stream 是否会泄露跨块拆分的值，并为它添加回归测试。”

该服务提供三个工具。check_stream_filter 会导入你的函数（myapp.filters:redact_stream），将每个样本按所有两段拆分方式流式传入，并告诉你哪种拆分导致问题。例如，输入被拆成 'use s' 和 'k-AbCd1234 now' 时，'sk-AbCd1234' 未经脱敏便输出。write_regression_test 会返回一个基于 assert_chunk_invariant 编写的 pytest 文件，确保持续集成（CI）中也会运行检查。explain 会用一段话解释为什么逐块过滤会导致泄露。

默认样本（电子邮件地址、SSN、卡号和几种 API 密钥格式）都是虚构的。请传入你自己编造的样本，格式应与你的过滤器所脱敏的内容一致，例如按你的服务提供商格式自行编写的密钥。若过滤器没有修改某个样本，则该样本不会测试任何内容，工具会指出这一点。

切勿将真实机密或真实个人数据作为样本传入。样本以及过滤器泄露的任何值都会出现在工具结果中，你的助手会像发送对话的其他内容一样，将结果发送给其模型提供商。你的代码引发异常时的错误消息也会被发送，但你的代码打印的内容不会。相同的样本也会出现在生成的测试文件中。

检查函数时会运行你的代码。服务会在你自己的计算机上导入模块，并在独立进程中运行；超时后（默认 60 秒）会停止该进程。如果项目中有 .venv，服务会使用其中的 Python。服务本身不会发起网络请求、发送遥测数据或写入文件。保存测试文件需要你和你的助手完成。

API

名称	作用
`assert_chunk_invariant(stream_filter, whole_filter, text, *, splits="all-two-part")`	在首次失败的切分处（先检查部分数最少的切分）引发 `ChunkInvarianceError`（一种 `AssertionError`）。返回已检查的切分数量。适用于 `str` 和 `bytes`。
`assert_chunk_invariant_async(...)`	用法相同，适用于异步生成器过滤器，可在异步测试中调用。
`two_part_splits(text)`	将文本切分为两个非空部分的所有方式。
`all_splits(text, max_parts=None)`	将文本切分为最多 `max_parts` 个非空部分的所有方式，整段输入排在最前。
`per_chunk(process, *, state=dict, flush=None)`	将 `process(chunk, state) -> output` 和 `flush(state) -> output` 适配为每条流使用全新状态。
`from_async(async_stream_filter)`	将异步生成器过滤器适配为可用于同步断言的过滤器。
`strategies.splits_of(text, max_parts=None)`	Hypothesis 策略：随机切分 `text`，并逐步缩减为更少的切分点。
`strategies.text_and_splits(texts, max_parts=None)`	Hypothesis 策略：生成 `(text, chunks)` 对。`texts` 本身必须是一个策略，例如 `st.sampled_from([...])`，而不是列表。