如何在Python中捕获psql子进程的SQL查询并拦截破坏性命令？

核心思路 #

在把SQL命令传给subprocess.run执行前，先对命令字符串做静态检查，直接拦截包含危险关键字的语句，根本不让它进入子进程执行。

具体实现步骤 #

1. 定义危险命令规则 #

先明确要拦截的关键词和模式，比如：

• 直接匹配DELETE、DROP、TRUNCATE这类破坏性DML/DDL命令
• 拦截SELECT * FROM（如果需要限制全表查询）
• 注意要忽略大小写，因为SQL不区分大小写

2. 编写检查函数 #

写一个简单的函数来检测输入的SQL命令是否包含危险内容：

import re

def is_dangerous_sql(sql_command):
    # 转换为小写统一检查
    lower_sql = sql_command.lower()
    # 定义危险关键词和模式
    dangerous_patterns = [
        r'\bdelete\b',
        r'\bdrop\b',
        r'\btruncate\b',
        r'\bselect\s+\*\s+from\b'  # 匹配SELECT * FROM
    ]
    # 遍历模式检查
    for pattern in dangerous_patterns:
        if re.search(pattern, lower_sql):
            return True
    return False

3. 在执行psql前做检查 #

把原来直接用subprocess.run执行的逻辑，加上前置检查：

import subprocess

def execute_psql_safe(sql_command, conn_args):
    # 先检查SQL是否危险
    if is_dangerous_sql(sql_command):
        print(f"拦截危险SQL命令: {sql_command}")
        return None
    
    # 构造psql命令，建议不要用shell=True（避免注入风险）
    psql_cmd = [
        'psql',
        '-h', conn_args['host'],
        '-p', str(conn_args['port']),
        '-U', conn_args['user'],
        '-d', conn_args['dbname'],
        '-c', sql_command
    ]
    
    # 执行安全的命令
    result = subprocess.run(psql_cmd, capture_output=True, text=True)
    print("执行结果:")
    print(result.stdout)
    if result.stderr:
        print("错误信息:")
        print(result.stderr)
    return result

# 示例调用
conn_info = {
    'host': '容器IP',
    'port': 5432,
    'user': 'postgres',
    'dbname': 'test_db'
}

# 测试危险命令
bad_sql = "DELETE FROM users WHERE id=1;"
execute_psql_safe(bad_sql, conn_info)  # 会被拦截

# 测试安全命令
good_sql = "SELECT id, name FROM users LIMIT 10;"
execute_psql_safe(good_sql, conn_info)  # 正常执行

注意事项 #

• 不要用shell=True：这会带来Shell注入风险，直接用列表形式传命令参数更安全
• 静态检查的局限性：如果用户用注释绕开（比如-- DELETE）或者复杂嵌套语句，静态正则可能漏判，更严谨的方式是用SQL解析库（比如sqlparse）来解析SQL语法，准确识别语句类型：

  import sqlparse
  from sqlparse.tokens import DML, Keyword
  
  def is_dangerous_sql_parsed(sql_command):
    parsed = sqlparse.parse(sql_command)[0]
    # 检查是否是DELETE/DROP等命令
    for token in parsed.tokens:
        if token.ttype is Keyword.DML and token.value.lower() in ['delete', 'truncate']:
            return True
        if token.ttype is Keyword.DDL and token.value.lower() == 'drop':
            return True
        # 检查SELECT是否带*
        if token.ttype is Keyword.DML and token.value.lower() == 'select':
            for sub_token in parsed.tokens:
                if sub_token.value == '*':
                    return True
    return False
  

• 权限控制：最好配合PostgreSQL的用户权限设置，比如给连接的数据库用户只赋予SELECT（限制列）、INSERT等必要权限，即使检查漏了，数据库层面也能拦截

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