如何安全允许用户在FastAPI应用中运行原生SQL查询？附示例

允许用户执行任意原生SQL本身存在极高风险（比如删表、资源耗尽、数据泄露），结合你的学习类应用场景，必须通过权限隔离、资源限制、输入校验三层措施来降低风险，以下是具体方案和代码示例：

核心安全前置配置 #

1. 创建PostgreSQL只读专用用户 #

绝对不能使用拥有写权限的数据库用户连接，必须为训练场景创建只读用户，并限制资源：

-- 创建训练专用只读用户
CREATE USER training_user WITH PASSWORD 'your_secure_password';

-- 仅授予指定训练表的SELECT权限
GRANT SELECT ON public.training_table TO training_user;
-- 若有多个训练表，批量授权
GRANT SELECT ON ALL TABLES IN SCHEMA public TO training_user;

-- 设置资源限制（防止恶意查询耗尽系统资源）
ALTER ROLE training_user SET statement_timeout = '5000ms'; -- 单查询超时5秒
ALTER ROLE training_user SET work_mem = '64MB'; -- 单查询内存上限64MB
ALTER ROLE training_user SET max_parallel_workers_per_gather = 0; -- 禁用并行查询

2. 数据库环境隔离 #

将训练数据放在独立的数据库/schema中，完全与生产环境物理隔离，避免影响核心业务数据。

代码实现示例 #

方案一：同步模式（psycopg3） #

from fastapi import FastAPI, HTTPException
from fastapi.templating import Jinja2Templates
from starlette.requests import Request
import psycopg
from psycopg.rows import dict_row
import logging

app = FastAPI()
templates = Jinja2Templates(directory="templates")

# 配置只读数据库连接
DB_CONFIG = {
    "dbname": "training_db",
    "user": "training_user",
    "password": "your_secure_password",
    "host": "localhost"
}

# 配置日志审计（记录所有用户执行的SQL）
logging.basicConfig(filename='sql_queries.log', level=logging.INFO)
logger = logging.getLogger(__name__)

def validate_sql(sql: str) -> bool:
    """基础SQL校验：仅允许SELECT开头的只读查询"""
    stripped_sql = sql.strip().upper()
    # 限制查询类型
    if not stripped_sql.startswith("SELECT"):
        return False
    # 过滤高危关键字（补充更多场景可自行添加）
    forbidden = ["DROP", "TRUNCATE", "ALTER", "CREATE", "INSERT", "DELETE", "UPDATE", "EXECUTE"]
    return not any(keyword in stripped_sql for keyword in forbidden)

@app.get("/")
async def index(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

@app.post("/run-sql")
def run_sql(sql: str):
    if not validate_sql(sql):
        raise HTTPException(status_code=400, detail="仅允许执行只读SELECT查询，禁止包含危险操作关键字")
    
    # 记录审计日志
    logger.info(f"User executed SQL: {sql}")
    
    try:
        with psycopg.connect(**DB_CONFIG) as conn:
            with conn.cursor(row_factory=dict_row) as cur:
                cur.execute(sql)
                result = cur.fetchall()
                columns = [desc[0] for desc in cur.description] if cur.description else []
                return {"columns": columns, "data": result}
    except psycopg.Error as e:
        raise HTTPException(status_code=400, detail=f"SQL执行错误: {str(e)}")

方案二：异步模式（asyncpg） #

from fastapi import FastAPI, HTTPException
from fastapi.templating import Jinja2Templates
from starlette.requests import Request
import asyncpg
import logging

app = FastAPI()
templates = Jinja2Templates(directory="templates")

DB_CONFIG = {
    "database": "training_db",
    "user": "training_user",
    "password": "your_secure_password",
    "host": "localhost"
}

logger = logging.getLogger(__name__)
logging.basicConfig(filename='sql_queries.log', level=logging.INFO)

async def validate_sql(sql: str) -> bool:
    stripped_sql = sql.strip().upper()
    if not stripped_sql.startswith("SELECT"):
        return False
    forbidden = ["DROP", "TRUNCATE", "ALTER", "CREATE", "INSERT", "DELETE", "UPDATE"]
    return not any(keyword in stripped_sql for keyword in forbidden)

@app.get("/")
async def index(request: Request):
    return templates.TemplateResponse("index.html", {"request": request})

@app.post("/run-sql")
async def run_sql(sql: str):
    if not await validate_sql(sql):
        raise HTTPException(status_code=400, detail="仅允许执行只读SELECT查询，禁止包含危险操作关键字")
    
    logger.info(f"User executed SQL: {sql}")
    
    try:
        conn = await asyncpg.connect(**DB_CONFIG)
        try:
            result = await conn.fetch(sql)
            result_dicts = [dict(row) for row in result]
            columns = result[0].keys() if result else []
            return {"columns": columns, "data": result_dicts}
        finally:
            await conn.close()
    except asyncpg.PostgresError as e:
        raise HTTPException(status_code=400, detail=f"SQL执行错误: {str(e)}")

Jinja2前端模板（集成ACE编辑器） #

创建templates/index.html：

<!DOCTYPE html>
<html>
<head>
    <title>SQL训练平台</title>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/ace/1.32.3/ace.js"></script>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/ace/1.32.3/mode-sql.js"></script>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/ace/1.32.3/theme-monokai.js"></script>
    <style>
        #sql-editor { width: 100%; height: 300px; margin-bottom: 10px; }
        #result-table { border-collapse: collapse; width: 100%; margin-top: 20px; display: none; }
        #result-table th, td { border: 1px solid #ddd; padding: 8px; }
        #result-table th { background-color: #f2f2f2; }
        .error { color: red; margin: 10px 0; }
    </style>
</head>
<body>
    <h1>SQL训练查询工具</h1>
    <div id="sql-editor">SELECT * FROM training_table LIMIT 10;</div>
    <button onclick="runQuery()">执行查询</button>
    <div class="error" id="error-box"></div>
    <table id="result-table">
        <thead id="table-header"></thead>
        <tbody id="table-body"></tbody>
    </table>

    <script>
        const editor = ace.edit("sql-editor");
        editor.setTheme("ace/theme/monokai");
        editor.session.setMode("ace/mode/sql");

        async function runQuery() {
            const sql = editor.getValue();
            const errorBox = document.getElementById("error-box");
            const resultTable = document.getElementById("result-table");
            const header = document.getElementById("table-header");
            const body = document.getElementById("table-body");

            // 清空历史内容
            errorBox.textContent = "";
            resultTable.style.display = "none";
            header.innerHTML = "";
            body.innerHTML = "";

            try {
                const res = await fetch("/run-sql", {
                    method: "POST",
                    headers: { "Content-Type": "application/x-www-form-urlencoded" },
                    body: `sql=${encodeURIComponent(sql)}`
                });

                if (!res.ok) {
                    const errData = await res.json();
                    throw new Error(errData.detail);
                }

                const data = await res.json();
                if (!data.data.length) {
                    errorBox.textContent = "查询无结果";
                    return;
                }

                // 渲染表头
                const headerRow = document.createElement("tr");
                data.columns.forEach(col => {
                    const th = document.createElement("th");
                    th.textContent = col;
                    headerRow.appendChild(th);
                });
                header.appendChild(headerRow);

                // 渲染数据行
                data.data.forEach(row => {
                    const tr = document.createElement("tr");
                    data.columns.forEach(col => {
                        const td = document.createElement("td");
                        td.textContent = row[col];
                        tr.appendChild(td);
                    });
                    body.appendChild(tr);
                });

                resultTable.style.display = "table";
            } catch (err) {
                errorBox.textContent = err.message;
            }
        }
    </script>
</body>
</html>

关键注意事项 #

1. 权限最小化是核心：SQL校验只能作为辅助，真正的安全屏障是数据库用户的只读权限，哪怕用户绕过校验，也无法执行写操作。
2. 资源限制必须设置：防止用户执行SELECT * FROM huge_table这类耗尽内存或CPU的查询。
3. 审计日志不可少：记录所有用户的SQL操作，方便后续排查异常行为。

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