如何清理支持用户自定义条件的Snowflake查询以降低风险

针对你开发内部工具的场景，除了正则匹配分号和注释，还有以下更可靠的风险控制手段：

1. 从权限层面锁死只读访问 #

给工具使用的Snowflake账号分配只读专属角色（比如基于SNOWFLAKE.READER自定义角色），仅授予目标数据表的SELECT权限，完全禁止INSERT/UPDATE/DELETE/DDL等操作权限。不管用户输入什么恶意语句，没有权限就无法执行，这是最根源的防护手段。

2. 强化资源限制防止DDoS #

除了硬编码返回记录数，还可以通过Snowflake的仓库配置进一步限制资源消耗：

• 给工具专用的Warehouse设置最大查询运行时长：ALTER WAREHOUSE <tool_warehouse> SET MAX_QUERY_RUNTIME = 300 SECONDS;（比如限制为5分钟），避免慢查询长时间占用资源。
• 限制Warehouse的最大规模：比如设置为X-SMALL或SMALL，防止大查询过度消耗计算资源。
• 开启查询结果缓存：Snowflake会自动缓存重复查询的结果，减少数据库重复计算，降低负载。

3. 严格的查询模板拼接与校验 #

不要让用户自由拼接完整查询，而是固定查询的核心结构，仅开放指定片段的输入：

• 固定FROM子句的目标表，禁止用户修改，防止访问敏感数据表。
• 校验SELECT子句：
  • 提前通过DESCRIBE TABLE <target_table>获取目标表的合法列名，检查用户输入的列名均在合法列表内。
  • 禁止SELECT子句包含UNION/JOIN/SUBQUERY（如果业务不需要）、CALL/EXECUTE等关键字，避免用户构造复杂查询或调用存储过程。
• 校验WHERE子句：
  • 同样检查其中的列属于目标表的合法列。
  • 用SQL解析库（比如sqlparse）解析子句，确保仅为合法的条件表达式，不存在多语句注入或恶意语法。

示例代码（SQL解析校验）：

import sqlparse

def validate_where_clause(where_clause):
    parsed = sqlparse.parse(f"WHERE {where_clause}")[0]
    # 检查是否存在DDL/DML关键字
    forbidden_keywords = {'INSERT', 'UPDATE', 'DELETE', 'DROP', 'ALTER', 'EXECUTE', 'CALL'}
    for token in parsed.tokens:
        if token.ttype == sqlparse.tokens.Keyword and token.value.upper() in forbidden_keywords:
            return False
    # 检查是否存在多语句分号
    if ';' in where_clause:
        return False
    return True

4. 给查询打标签并监控 #

给工具发起的所有查询设置Query Tag，比如在连接Snowflake时执行：

conn.cursor().execute("ALTER SESSION SET QUERY_TAG = 'internal_tool_query'")

通过Snowflake的Query History功能，监控带有该标签的所有查询，一旦发现异常语句（比如包含恶意关键字、运行时间过长），可以及时告警或终止。

5. 预处理时过滤危险语法 #

除了正则，还可以直接替换或拒绝包含危险语法的输入：

• 过滤块注释/* */和行注释--，避免用户注释掉后面的LIMIT或其他限制。
• 拒绝包含;的输入，防止多语句注入（比如WHERE id=1; DROP TABLE users;）。

内容的提问来源于stack exchange，提问作者AvitanD