Django中如何通过校验RawSQL阻止删除、更新等操作？

方案一：自定义安全RawSQL子类 #

直接继承Django原生的RawSQL，在初始化阶段添加SQL校验逻辑，替换原有RawSQL的使用场景即可。

from django.db.models.expressions import RawSQL
import re

class SafeRawSQL(RawSQL):
    def __init__(self, sql, params=None, output_field=None):
        cleaned_sql = self._clean_sql(sql)
        # 定义禁止执行的SQL语句前缀
        forbidden_prefixes = ('DELETE', 'UPDATE', 'INSERT INTO', 'ALTER TABLE', 'DROP TABLE')
        
        if any(cleaned_sql.startswith(prefix) for prefix in forbidden_prefixes):
            raise ValueError("禁止执行数据修改类SQL语句")
        
        super().__init__(sql, params, output_field)
    
    @staticmethod
    def _clean_sql(sql):
        # 移除SQL中的单行/多行注释
        sql = re.sub(r'--.*?$', '', sql, flags=re.MULTILINE)
        sql = re.sub(r'/\*.*?\*/', '', sql, flags=re.DOTALL)
        # 合并多余空格并转为大写，统一校验规则
        return ' '.join(sql.strip().split()).upper()

使用时，把代码中所有RawSQL(...)替换为SafeRawSQL(...)，一旦传入修改类SQL，会直接抛出异常终止执行。

方案二：全局拦截数据库游标执行 #

通过自定义数据库后端包装器，拦截所有SQL执行请求，从底层阻止修改类语句。这种方式能覆盖所有SQL执行场景，包括RawSQL、ORM自动生成的SQL（可按需针对性排除）。

1. 自定义数据库后端 #

以SQLite为例，创建自定义包装器：

from django.db.backends.sqlite3.base import SQLiteDatabaseWrapper
from django.db.utils import OperationalError
import re

class SafeSQLiteWrapper(SQLiteDatabaseWrapper):
    def create_cursor(self, name=None):
        cursor = super().create_cursor(name)
        # 包装原生execute和executemany方法
        original_execute = cursor.execute
        original_executemany = cursor.executemany

        def safe_execute(sql, params=None):
            self._validate_sql(sql)
            return original_execute(sql, params)

        def safe_executemany(sql, params=None):
            self._validate_sql(sql)
            return original_executemany(sql, params)

        cursor.execute = safe_execute
        cursor.executemany = safe_executemany
        return cursor

    def _validate_sql(self, sql):
        cleaned_sql = self._clean_sql(sql)
        forbidden_prefixes = ('DELETE', 'UPDATE', 'INSERT INTO', 'ALTER TABLE', 'DROP TABLE')
        
        if any(cleaned_sql.startswith(prefix) for prefix in forbidden_prefixes):
            raise OperationalError("数据库禁止执行数据修改类SQL语句")
    
    @staticmethod
    def _clean_sql(sql):
        # 移除注释并格式化SQL
        sql = re.sub(r'--.*?$', '', sql, flags=re.MULTILINE)
        sql = re.sub(r'/\*.*?\*/', '', sql, flags=re.DOTALL)
        return ' '.join(sql.strip().split()).upper()

如果使用MySQL或PostgreSQL，只需替换对应的父类（比如django.db.backends.mysql.base.DatabaseWrapper）。

2. 配置使用自定义后端 #

在settings.py的DATABASES中指定自定义引擎：

DATABASES = {
    'default': {
        'ENGINE': 'your_app_name.backends.safe_sqlite',  # 替换为你的自定义包装器实际路径
        'NAME': BASE_DIR / 'db.sqlite3',
        # 其他数据库配置参数...
    }
}

方案三：使用装饰器管控RawSQL调用 #

如果只需要管控特定函数中的RawSQL调用，可以用装饰器包裹RawSQL的调用逻辑：

import re

def block_modify_sql(func):
    def wrapper(*args, **kwargs):
        # 假设第一个参数是SQL语句，可根据实际调用结构调整
        sql = args[0] if args else kwargs.get('sql')
        if sql:
            cleaned_sql = _clean_sql(sql)
            forbidden_prefixes = ('DELETE', 'UPDATE', 'INSERT INTO')
            if any(cleaned_sql.startswith(prefix) for prefix in forbidden_prefixes):
                raise ValueError("禁止执行修改类SQL语句")
        return func(*args, **kwargs)
    
    def _clean_sql(sql):
        sql = re.sub(r'--.*?$', '', sql, flags=re.MULTILINE)
        sql = re.sub(r'/\*.*?\*/', '', sql, flags=re.DOTALL)
        return ' '.join(sql.strip().split()).upper()
    
    return wrapper

使用示例：

from django.db.models.expressions import RawSQL

@block_modify_sql
def execute_raw_sql(sql, params=None):
    return RawSQL(sql, params)

# 调用时会自动校验
# execute_raw_sql("DELETE FROM my_table")  # 会抛出异常

额外建议 #

上述方案是应用层的防护，建议同时配置数据库层面的权限控制：给Django使用的数据库账号仅分配SELECT权限，即使应用层校验被绕过，数据库也会直接拒绝修改操作，形成双重安全保障。

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