如何在Databricks SQL Connector中安全实现动态添加列以解决SQL注入漏洞

你的问题核心在于动态列定义属于SQL语法的一部分，无法通过参数绑定直接安全传递——这也是为什么你之前用IDENTIFIER()处理表名可行，但列名部分没法复用同样逻辑的原因。不过我们可以通过强化输入验证+最小权限原则来彻底解决SQL注入风险，同时保留动态添加列的功能。以下是具体方案：

一、完善输入验证：从基础检查到全维度安全校验 #

你现有的is_safe_identifier是很好的起点，但我们需要扩展它，覆盖列名、数据类型的全场景验证，确保只有合法的输入才能进入SQL拼接环节：

1. 强化标识符安全检查 #

首先，我们可以保留基础的标识符规则，同时处理可能的特殊情况（比如需要用反引号包裹的非标准列名，前提是列名不含反引号避免注入）：

def is_safe_identifier(self, identifier):
    # 验证标准标识符（字母/数字/下划线），或不含反引号的非标准标识符
    standard_pattern = r'^[a-zA-Z0-9_]+$'
    quoted_safe_pattern = r'^[^`]+$'
    return re.match(standard_pattern, identifier) is not None or re.match(quoted_safe_pattern, identifier) is not None

2. 验证数据类型合法性 #

Databricks SQL的列类型是固定的集合，我们可以维护一个允许的类型列表，拒绝任何不在列表中的输入：

def is_valid_datatype(self, dtype):
    allowed_types = {
        'STRING', 'INT', 'INTEGER', 'BIGINT', 'DATE', 
        'TIMESTAMP', 'BOOLEAN', 'DOUBLE', 'FLOAT', 'DECIMAL'
    }
    return dtype.upper() in allowed_types

二、修改核心业务代码：安全拼接动态列 #

在add_config函数中，我们先对所有输入（库名、表名、列名、数据类型）做严格校验，只有全部通过后才拼接SQL语句：

def add_config(self, user_name, action, schema_name, table_name, field_names, field_datatypes):
    # 1. 验证库名和表名的安全性
    if not all([self.is_safe_identifier(schema_name), self.is_safe_identifier(table_name)]):
        raise ValueError("Invalid schema or table name - only alphanumeric, underscore, or non-backtick characters allowed")
    
    # 2. 验证列名和数据类型的合法性，同时生成安全的列定义字符串
    columns_parts = []
    if len(field_names) != len(field_datatypes):
        raise ValueError("Field names and data types count mismatch")
    
    for name, dtype in zip(field_names, field_datatypes):
        if not self.is_safe_identifier(name):
            raise ValueError(f"Invalid column name: {name} - only alphanumeric, underscore, or non-backtick characters allowed")
        if not self.is_valid_datatype(dtype):
            raise ValueError(f"Invalid data type: {dtype} - must be one of {', '.join(self.allowed_types)}")
        
        # 用反引号包裹列名，兼容非标准标识符（如果需要）
        columns_parts.append(f"`{name}` {dtype.upper()}")
    
    columns_str = ", ".join(columns_parts)
    
    # 3. 执行ALTER语句（表名仍用IDENTIFIER参数化）
    alter_query = "ALTER TABLE IDENTIFIER(:sch || '.' || :tab) ADD COLUMNS ({})".format(columns_str)
    cursor.execute(alter_query, {'sch': schema_name,'tab': table_name})
    self.conn.commit()

三、额外安全加固：最小权限与审计 #

除了代码层面的验证，我们还可以通过环境配置进一步降低风险：

• 最小权限原则：确保连接Databricks的SQL账号仅拥有ALTER TABLE的必要权限，没有DROP TABLE、CREATE DATABASE等高风险权限，即使发生意外注入，危害也会被限制。
• 操作审计：记录所有ALTER操作的详细信息（操作人、时间、库表名、添加的列），便于事后追踪和排查。
• 拒绝保留字：可以额外添加一个检查，确保列名不是Databricks SQL的保留字（比如SELECT、FROM、WHERE等），避免语法错误或潜在风险。

为什么这个方案能避免SQL注入？ #

因为所有进入SQL拼接环节的输入都经过了严格的校验：

• 列名只能是字母、数字、下划线，或者不含反引号的字符串，无法包含SQL注入需要的特殊字符（比如;、--、DROP等）。
• 数据类型被限制在合法的Databricks类型列表中，无法注入恶意SQL片段。
• 库表名依然用IDENTIFIER()参数化，避免这部分的注入风险。

这种方案既保留了你需要的动态添加列的功能，又彻底解决了SQL注入的问题——这也是处理动态SQL语法元素（如表名、列名）的行业通用安全方案。

备注：内容来源于stack exchange，提问作者Vinay Yogeesh