如何在Python中避免SQL Server Upsert查询的SQL注入？

你的问题核心是避免用f字符串直接拼接SQL，同时不依赖存储过程。要解决这个问题，得拆分两类输入分别处理：表名、列名这类「标识符」只能用白名单验证，而数值、字符串这类「值」可以用参数化查询。下面是具体修改方案：

关键防护逻辑 #

1. 白名单验证标识符：数据库驱动不支持参数化表名、列名，所以必须提前定义允许访问的表和列，拒绝任何不在白名单内的输入，防止攻击者注入恶意语句（比如'; DROP TABLE users--）。
2. 参数化处理值：所有用户可控的数值、字符串都用数据库驱动支持的占位符（pymssql用%s），由驱动自动转义特殊字符，彻底避免注入。

修改后的完整函数 #

def sql_gen(tv, kv, join_kv, col_inst, val_inst, val_upd):
    # 1. 配置业务允许的表和列白名单（必须根据实际情况修改）
    ALLOWED_TABLES = ['your_target_table']  # 替换成你实际的表名
    ALLOWED_COLUMNS = ['id', 'name', 'age', 'email']  # 替换成你实际的列名

    # 验证表名合法性
    if tv not in ALLOWED_TABLES:
        raise ValueError(f"非法表名：{tv}")
    
    # 验证WHERE子句的列名
    if kv not in ALLOWED_COLUMNS:
        raise ValueError(f"非法列名：{kv}")
    
    # 验证插入列的合法性
    insert_cols = [col.strip() for col in col_inst.split(',')]
    for col in insert_cols:
        if col not in ALLOWED_COLUMNS:
            raise ValueError(f"非法插入列：{col}")
    
    # 验证更新列的合法性
    update_cols = [clause.strip().split('=')[0] for clause in val_upd.split(',')]
    for col in update_cols:
        if col not in ALLOWED_COLUMNS:
            raise ValueError(f"非法更新列：{col}")

    # 2. 构造参数化SQL语句
    # 生成INSERT对应的占位符（和插入列数量一致）
    insert_placeholders = ', '.join(['%s'] * len(insert_cols))
    # 生成UPDATE对应的占位符（每个列赋值用 列名=%s）
    update_placeholders = ', '.join([f"{col.strip()}=%s" for col in update_cols])

    sqlstmt = """
    IF NOT EXISTS (
            SELECT *
            FROM {tv}
            WHERE {kv} = %s
            )
        INSERT {tv} (
        {col_inst}
        )
        VALUES (
        {insert_placeholders}
        )
    ELSE
        UPDATE {tv}
        SET {update_placeholders}
        WHERE {kv} = %s;
    """.format(
        tv=tv,
        kv=kv,
        col_inst=col_inst,
        insert_placeholders=insert_placeholders,
        update_placeholders=update_placeholders
    )

    # 3. 整理参数列表（顺序必须和SQL里的占位符一一对应）
    params = [join_kv]  # 第一个占位符：IF NOT EXISTS里的WHERE条件值
    params.extend([val.strip() for val in val_inst.split(',')])  # INSERT的所有值
    params.extend([clause.strip().split('=')[1].strip() for clause in val_upd.split(',')])  # UPDATE的所有值
    params.append(join_kv)  # 最后一个占位符：UPDATE里的WHERE条件值

    # 4. 执行SQL（补充资源释放逻辑）
    engine = create_engine(f"mssql+pymssql://{username}:{password}@{server}/{database}")
    connection = engine.raw_connection()
    try:
        cursor = connection.cursor()
        cursor.execute(sqlstmt, params)
        connection.commit()
    finally:
        cursor.close()
        connection.close()  # 确保连接关闭，避免资源泄漏

重要注意事项 #

• 白名单必须严格配置：ALLOWED_TABLES和ALLOWED_COLUMNS要完全匹配你的业务需求，不能包含无关表/列，也不能留空。
• 不要硬编码凭证：把username、password、server、database这些敏感信息放到环境变量里，比如用os.getenv("DB_USER")获取，避免代码泄露风险。
• 参数顺序要对应：params列表的顺序必须和SQL中%s出现的顺序完全一致，否则会导致数据错误。

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