DRF如何实现支持括号优先级、多字段组合的安全动态过滤功能

需求说明 #

你需要实现的动态过滤功能要求如下：

• 支持使用括号定义运算优先级，可任意组合模型的可用字段
• 支持运算：and、or、eq（等于）、ne（不等于）、gt（大于）、lt（小于）
• 示例查询语句："(date eq '2016-05-01') AND ((number_of_calories gt 20) OR (number_of_calories lt 10))"

现有方案的问题 #

你当前的实现存在严重的SQL注入风险，靠正则过滤危险SQL关键字的方式完全不可靠，很容易被绕过，比如构造' OR 1=1 --这类查询就能直接拖库，而且直接拼接原生SQL执行的方式本身就不符合Django ORM的安全规范，后续维护成本极高。

最优实现方案 #

完全基于Django ORM的Q对象构造查询，避免原生SQL拼接，从根源上杜绝SQL注入，同时原生支持括号优先级和任意字段组合。

1. 依赖安装 #

用ply库做查询语法的词法和语法解析，比自行写正则拆分更稳定：

pip install ply

2. 核心解析逻辑实现 #

import ply.lex as lex
import ply.yacc as yacc
from django.db.models import Q
from typing import Optional

# 词法规则定义
tokens = (
    'IDENTIFIER',
    'STRING',
    'NUMBER',
    'EQ', 'NE', 'GT', 'LT', 'GTE', 'LTE',
    'AND', 'OR',
    'LPAREN', 'RPAREN',
)

# 运算符匹配
t_EQ = r'eq'
t_NE = r'ne'
t_GT = r'gt'
t_LT = r'lt'
t_GTE = r'gte'
t_LTE = r'lte'
t_AND = r'AND|and'
t_OR = r'OR|or'
t_LPAREN = r'\('
t_RPAREN = r'\)'
t_STRING = r"'[^']*'"
t_NUMBER = r'\d+'
t_ignore = ' \t'

def t_IDENTIFIER(t):
    r'[a-zA-Z_][a-zA-Z0-9_]*'
    keyword_map = {
        'eq': 'EQ', 'ne': 'NE', 'gt': 'GT', 'lt': 'LT',
        'gte': 'GTE', 'lte': 'LTE', 'and': 'AND', 'or': 'OR'
    }
    t.type = keyword_map.get(t.value.lower(), 'IDENTIFIER')
    return t

def t_error(t):
    raise ValueError(f"非法字符: {t.value[0]}")

lexer = lex.lex()

# 语法规则&优先级配置
precedence = (
    ('left', 'OR'),
    ('left', 'AND'),
    ('left', 'EQ', 'NE', 'GT', 'LT', 'GTE', 'LTE'),
)

# 模型字段白名单，替换为你实际允许查询的字段
ALLOWED_FIELDS = {'date', 'number_of_calories', 'name', 'age'}
# 运算符到ORM查询后缀的映射
OPERATOR_MAP = {
    'eq': '',
    'ne': '~',
    'gt': '__gt',
    'lt': '__lt',
    'gte': '__gte',
    'lte': '__lte',
}

def p_expression_binop(p):
    '''expression : expression AND expression
                  | expression OR expression'''
    if p[2].lower() == 'and':
        p[0] = p[1] & p[3]
    elif p[2].lower() == 'or':
        p[0] = p[1] | p[3]

def p_expression_compare_str(p):
    '''expression : IDENTIFIER EQ STRING
                  | IDENTIFIER NE STRING'''
    field = p[1]
    op = p[2].lower()
    value = p[3].strip("'")
    if field not in ALLOWED_FIELDS:
        raise ValueError(f"不允许查询字段: {field}")
    if op == 'ne':
        p[0] = ~Q(**{f"{field}": value})
    else:
        p[0] = Q(**{f"{field}{OPERATOR_MAP[op]}": value})

def p_expression_compare_num(p):
    '''expression : IDENTIFIER GT NUMBER
                  | IDENTIFIER LT NUMBER
                  | IDENTIFIER GTE NUMBER
                  | IDENTIFIER LTE NUMBER'''
    field = p[1]
    op = p[2].lower()
    value = int(p[3])
    if field not in ALLOWED_FIELDS:
        raise ValueError(f"不允许查询字段: {field}")
    p[0] = Q(**{f"{field}{OPERATOR_MAP[op]}": value})

def p_expression_group(p):
    'expression : LPAREN expression RPAREN'
    p[0] = p[2]

def p_error(p):
    raise ValueError("查询语法错误")

parser = yacc.yacc()

def parse_query(query_str: str) -> Optional[Q]:
    try:
        return parser.parse(query_str)
    except:
        return None

3. 视图层调用 #

def get_queryset(self):
    q_string = self.request.data.get('query', '')
    if not q_string:
        return Users.objects.all()
    q_obj = parse_query(q_string)
    if not q_obj:
        # 语法错误/非法字段时返回空或者抛出参数异常
        return Users.objects.none()
    return Users.objects.filter(q_obj)

方案优势 #

• 所有查询参数都由ORM做参数化处理，完全避免SQL注入风险
• 字段白名单校验，避免敏感字段泄露
• 原生支持括号优先级，满足任意字段组合查询的需求
• 后续扩展运算符、关联表查询都很方便，仅需修改解析规则即可

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