CocoIndex 目标连接器输入安全实战：标识符校验、参数化查询与值转义

CocoIndex 目标连接器输入安全实战：标识符校验、参数化查询与值转义

【免费下载链接】cocoindexIncremental engine for long horizon agents 🌟 Star if you like it!项目地址: https://gitcode.com/GitHub_Trending/co/cocoindex

导读

本文是 CocoIndex 仓库中 target-connector 技能 配套的输入安全指南，聚焦目标连接器（Target Connector）在把用户提供的表名、列名、索引名等标识符与记录 ID、键值等数据值拼进外部系统查询语句时，如何防止注入攻击并保证语义正确。读完本文，你将掌握三类可复用的安全手段——API 入口处的标识符白名单校验、数据值的参数化绑定、无法参数化场景下的类型保留转义，并能在 CocoIndex 各内置连接器（SurrealDB、PostgreSQL、Neo4j、SQLite 等）的源码中找到对应的权威实现与测试证据。

1. 为什么目标连接器需要输入安全

目标连接器的本质工作，是把 CocoIndex 声明式目标状态（TargetState）同步到外部系统：对比期望状态与历史追踪记录后，把差异转成对数据库、文件系统或云存储的增删改操作（参考 SKILL.md 中的 TargetHandler / TargetActionSink 说明）。同步过程免不了把用户提供的名字（表名、列名、索引名）和数据值（记录 ID、主键、内容）写入查询语句，于是出现两类风险：

• 注入风险：名字或值中包含引号、分号、注释符等特殊字符时，直接拼接会导致查询被改写，甚至执行攻击者构造的语句。
• 语义正确性风险：即使没有安全威胁，类型差异（如整数123与字符串"123"）被错误处理，也会导致数据错位。

CocoIndex 的解决思路高度统一：名字走白名单校验，值走参数化绑定，参数化不了的内联场景走保留类型的转义。下文按这三个层次逐一展开，并在每一节给出仓库内的实现与测试佐证。

2. 标识符校验：在 API 入口用白名单正则拒绝一切非法名字

2.1 核心原则：入口处校验，而不是查询构造时转义

用户提供的名字（表、列、索引）会被插值进查询中充当标识符。标识符在大多数数据库里既无法使用绑定参数（如 PostgreSQL 的$1、Cypher 的$param只能绑定值，不能绑定标识符），单纯靠加引号转义又难以堵死所有边界（例如名字本身包含双引号字符）。因此正确做法是在 API 入口尽早校验——只要不匹配"简单标识符"的白名单就立即抛错，让非法名字根本走不到查询构造这一步。

input_safety.md 给出了标准实现：

import re _IDENTIFIER_RE = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]*$") def _validate_identifier(name: str, kind: str) -> None: """Raise ValueError if name is not a safe identifier.""" if not _IDENTIFIER_RE.match(name): raise ValueError( f"Invalid {kind}: {name!r}. " f"Must match [a-zA-Z_][a-zA-Z0-9_]*." )

规则^[a-zA-Z_][a-zA-Z0-9_]*$的含义非常明确：

• 首字符：只能是字母（a-zA-Z）或下划线（_），排除以数字开头的名字；
• 后续字符：只能包含字母、数字、下划线；
• 空字符串、包含空格、连字符、点号、反引号、分号等任何其他字符都会被拒绝。

校验必须放在每一个接受名字的公开方法里，覆盖表名、列名、索引名、主键、schema 名等所有入口：

def table_target(self, table_name: str, ...) -> ...: _validate_identifier(table_name, "table name") ... class TableSchema: def __init__(self, columns: dict[str, ColumnDef], ...) -> None: for col_name in columns: _validate_identifier(col_name, "column name") ...

2.2 仓库实现证据：SurrealDB 连接器

本文档指定的权威实现位于 python/cocoindex/connectors/surrealdb/_target.py。它一字不差地落地了上述模式：

_IDENTIFIER_RE = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]*$") def _validate_identifier(name: str, kind: str) -> None: """Validate that *name* is a safe SurrealQL identifier. Raises :class:`ValueError` if the name contains characters that are not alphanumeric or underscore, or starts with a digit. """ if not _IDENTIFIER_RE.match(name): raise ValueError( f"Invalid SurrealDB {kind}: {name!r}. Must match [a-zA-Z_][a-zA-Z0-9_]*." )

随后在每一处公开 API 入口调用它（同一文件内）：

• TableSchema.__init__对每个列名校验（L356-L359）；
• table_target()校验表名（L1291）；
• relation_target()校验关系表名以及 FROM/TO 两端的表名（L1341-L1347）；
• declare_vector_index()校验向量索引字段名与索引名（L1130-L1133）。

注意"入口处校验"的含金量：即使_create_table/_apply_column_actions在构造DEFINE TABLE、DEFINE FIELD等 SurrealQL 时直接 f-string 拼接这些名字（L959、L1003、L1015、L1038、L1051），由于所有名字都已在入口通过白名单，拼接是安全的——这正是"提前校验代替事后转义"的架构价值。

2.3 全仓库的一致性实践

这一模式并非 SurrealDB 独有，而是整个python/cocoindex/connectors/目录的共同约定，仅正则与错误信息略有差异：

可见"CocoIndex 所有目标连接器共享同一输入安全约定"是有源码依据的结论。

3. 数据值：一律使用参数化查询（绑定变量）

标识符只能靠白名单，但数据值（插入的内容、记录 ID 对应的值、过滤条件中的键值）完全不同——它们应当永远通过参数化查询（绑定变量）传递，绝不直接插值进查询字符串。

3.1 三种主流写法对照

input_safety.md 给出了三种不同数据库的参数化写法：

# Good — parameterized await conn.execute("INSERT INTO t (name) VALUES ($1)", value) # PostgreSQL conn.execute("INSERT INTO t (name) VALUES (?)", (value,)) # SQLite await conn.query("UPSERT t:id CONTENT $content", {"content": val}) # SurrealDB # Bad — string interpolation await conn.execute(f"INSERT INTO t (name) VALUES ('{value}')")

参数化的收益：驱动会在协议层把值与语句分离，值中的引号、分号、反斜杠等字符只被当作数据，不可能逃逸成 SQL/SurrealQL/Cypher 语法，从根上消除注入面。

3.2 仓库中的参数化实践

• SQLite 连接器：_apply_actions中通过?占位符绑定值，例如"INSERT INTO t (name) VALUES (?)"与(value,)的成对使用（参见 python/cocoindex/connectors/sqlite/_target.py）。
• PostgreSQL 连接器：大量使用 asyncpg 的$1绑定。PostgreSQL 还额外处理了一类非注入但致命的边界——text/jsonb列不允许包含 NUL（U+0000）字符，连接器通过_strip_nul/_sanitize_nul在绑定前递归清除字符串、dict 键与嵌套容器中的 NUL（python/cocoindex/connectors/postgres/_target.py）。对应测试 python/tests/connectors/test_postgres_target.py 专门验证text[]数组元素里的 NUL 也会被清除，否则 asyncpg 会抛出ValueError: string cannot contain NUL (0x00) characters。
• Neo4j 连接器：Cypher 生成模块 python/cocoindex/connectors/neo4j/_cypher.py 的模块级约定是"所有值一律通过$-参数绑定"，例如build_node_upsert生成MERGE (n:Label{pk: $key_0, ...}) SET n += $props（L90-L105），键值与属性值全部走$key_N/$props参数，绝无值内联。

4. 值转义：参数化行不通时的最后手段

4.1 何时必须内联值

有些查询语言的语法要求值出现在无法参数化的位置。文档以 SurrealDB 的table:id记录 ID 语法为例：UPSERT person:alice CONTENT {...}中alice是记录 ID 的一部分，不能绑定为$1，只能内联。此时需要自己写转义函数。

4.2 两条硬性要求

• 保留类型区分：整数123与字符串"123"在语义上可能完全不同——SurrealDB 中person:123（数字 ID）与person:123``（字符串 ID）指向不同的记录。转义逻辑必须根据 Python 类型分别输出，不能一律套引号。
• 转义引号字符：目标使用反引号引用字符串时，值内部的反斜杠与反引号都必须转义，防止提前闭合引用。

文档给出的标准实现：

def _format_record_id(value: Any) -> str: """Format a record ID for inline use, preserving type.""" if isinstance(value, (int, float)): return str(value) # bare numeric: 123, 3.14 s = str(value) s = s.replace("\\", "\\\\").replace("`", "\\`") return f"`{s}`" # quoted string: `alice`

4.3 仓库实现与单元测试

这一函数在 python/cocoindex/connectors/surrealdb/_target.py 中逐字存在，并被_SharedRecordApplier._apply_actions用于构造批量 UPSERT / DELETE / RELATE 语句（L493-L531），记录 ID 一律经由_format_record_id内联，内容字段则用json.dumps(content, default=str)序列化后内联进CONTENT子句。

对应的无数据库单元测试位于 python/tests/connectors/test_surrealdb_target.py，逐条验证类型保留与转义正确性：

顺带一提，转义是"按目标语法定制"的：LanceDB 的 DataFusion SQL 删除过滤器需要把单引号翻倍（it's→it''s），而反斜杠原样透传——见 python/tests/connectors/test_lancedb_target.py 的_escape_sql_string测试。转义规则永远取决于目标方言，写连接器时必须按目标语法单独实现与测试。

5. 测试策略：单元测试管转义，集成测试管往返

输入安全逻辑必须可验证。input_safety.md 给出了两层测试要求：

5.1 不需要数据库的单元测试

对安全辅助函数（校验、转义）与 API 入口行为直接做断言：

• 合法标识符通过，非法标识符抛出ValueError；
• 转义对特殊字符、空字符串、数值类型产生正确输出；
• API 入口拒绝坏名字（如TableSchema(columns={"bad-name": ...})）。

这些测试在仓库中有两个典型体现：

SurrealDB 侧（python/tests/connectors/test_surrealdb_target.py）——TestValidateIdentifier用参数化测试验证users、_private、T1、a_b_c通过，而my-table、123abc、空串、has space、back、semi;colon、a.b全部抛ValueError（match="Invalid SurrealDB"）；TestValidateIdentifierAtApiEntryPoints则直接验证TableSchema、table_target、relation_target` 这三个公开入口会拒绝坏名字。

Neo4j 侧（python/tests/connectors/test_neo4j_target.py）——同样的白名单参数化测试，非法集合额外包含X-Y，并验证MERGE (n:Document{filename: $key_0})这类生成结果里标识符以反引号包裹、值以参数绑定（L115-L129）。

5.2 需要真实数据库的集成测试

当目标数据库可用时，应补充特殊字符值走完整 upsert/select 往返的集成测试：写入包含引号、反斜杠、Unicode 的记录 ID 与内容，再读回比对，确认转义/参数化在实际查询链路中没有破坏数据。这类测试的框架可参考 python/tests/connectors/test_surrealdb_target.py 中@requires_surrealdb守卫的集成用例，以及 SKILL.md 的测试章节 给出的test_insert_and_update标准模式（依赖缺失时用pytest.mark.skipif跳过，保证纯环境也能跑完所有安全单元测试）。

6. 编写新目标连接器时的检查清单

结合 input_safety.md 与仓库中十几个连接器的既有实践，开发新目标连接器时可逐项自检：

1. 入口全覆盖：表名、列名、索引名、主键、schema/dataset 名等所有接受用户名字的公开方法，是否都调用了_validate_identifier（参考 surrealDB 的调用点分布）？
2. 值一律参数化：数据值是否全部走绑定变量（$1/?/$name/$props），还是存在直接 f-string 内联值的地方？
3. 内联必转义且保留类型：无法参数化的内联位置（如 SurrealDB 记录 ID、LanceDB 删除过滤器），是否按目标方言转义引号字符，并区分数值与字符串类型？
4. 先校验后拼接：拼接 DDL/DML 时使用的名字是否都已在更早的入口通过白名单，而不是在拼接处做转义补救？
5. 测试双轨并行：无需数据库的单元测试是否覆盖校验/转义的全部边界（特殊字符、空串、数值类型、入口拒收）；有数据库时是否补上特殊字符值的往返集成测试？

遵循这套约定后，你的连接器将与 PostgreSQL、SQLite、SurrealDB、Neo4j 等内置连接器处于同一安全水位：标识符注入在入口被白名单拦死，值注入被参数化消解，方言强制的内联位置有类型保留的精确转义兜底。

7. 延伸阅读

• 本文规范文档：dev/agent-skills/target-connector/input_safety.md
• 连接器实现总纲（TargetHandler / TargetActionSink / 子目标失效策略 / 幂等动作）：dev/agent-skills/target-connector/SKILL.md
• 权威实现一：SurrealDB 连接器 python/cocoindex/connectors/surrealdb/_target.py，含标识符校验、记录 ID 类型保留转义、批量 UPSERT/RELATE 语句构造
• 权威实现二：Neo4j 纯 Cypher 生成模块 python/cocoindex/connectors/neo4j/_cypher.py，展示"标识符入口校验 + 值参数绑定"的模块级约定
• 测试证据：python/tests/connectors/test_surrealdb_target.py、python/tests/connectors/test_neo4j_target.py、python/tests/connectors/test_postgres_target.py

【免费下载链接】cocoindexIncremental engine for long horizon agents 🌟 Star if you like it!项目地址: https://gitcode.com/GitHub_Trending/co/cocoindex