不要用危险的字符拼接sql查询

解决 SQL 注入问题，在于不去拼接字符串构造 SQL，而是使用“预编译 + 参数绑定”的方式（即参数化查询）来处理 SQL 请求。抑或是使用ORM。

什么是 SQL 注入？

SQL 注入（SQL Injection）是指攻击者通过构造恶意输入，插入到 SQL 查询语句中，从而操控数据库执行非预期的操作，如绕过登录、删除数据等。
例如：

username = input("请输入用户名：")
sql = "SELECT * FROM users WHERE username = '" + username + "';"

当用户输入 admin' OR '1'='1 时，SQL 变成：

SELECT * FROM users WHERE username = 'admin' OR '1'='1';

这会导致永远为真，绕过认证。

解决方案1：预编译+参数绑定（Prepared Statements）

将 SQL 语句结构提前固定好，只在后续绑定参数，而不是拼接字符串，数据库系统会提前“准备”好 SQL 的执行计划。

import psycopg2

conn = psycopg2.connect(database="testdb", user="postgres", password="yourpassword")
cursor = conn.cursor()

# 用户输入
username = input("请输入用户名：")

# ✅ 安全的方式：参数化查询，防止SQL注入
sql = "SELECT * FROM users WHERE username = %s;"
cursor.execute(sql, (username,))  # 第二个参数是 tuple，即使用户输入恶意内容也不会被执行为SQL代码

rows = cursor.fetchall()
print(rows)

这里的 %s 是占位符，(username,) 是绑定参数。PostgreSQL 会自动对输入内容进行转义和类型处理，用户输入的内容只会当作“数据”，不会被解释为 SQL 代码。

解决方案2: ORM（Object-Relational Mapping）

ORM 是一种将数据库表映射为程序中的对象 的工具，它允许你用“面向对象的方式”来操作数据库，而不是手写 SQL。

优点包括：

• 代码更简洁、抽象
• 更容易维护和迁移数据库
• 内置 参数化查询机制，天然避免 SQL 注入

一个python+postgresql的简单的例子：

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base

# 创建数据库连接
engine = create_engine("postgresql://postgres:yourpassword@localhost/testdb")
Session = sessionmaker(bind=engine)
session = Session()

Base = declarative_base()

# 定义一个 User 模型
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String)

# 查询一个用户（安全）
username_input = input("请输入用户名：")
user = session.query(User).filter(User.username == username_input).first()
print(user)

当然，orm实际上更有益于我们进行数据库控制，即增删改查等操作：

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base

# 数据库连接设置（请修改为你的实际用户名、密码和数据库名）
engine = create_engine("postgresql://postgres:yourpassword@localhost/testdb")
Session = sessionmaker(bind=engine)
session = Session()

# 定义 ORM 基类
Base = declarative_base()

# 定义 User 模型（对应 users 表）
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String)
    email = Column(String)

# 创建表（如果尚不存在）
Base.metadata.create_all(engine)

# ---------- 增（Insert） ----------
new_user = User(username="alice", email="alice@example.com")
session.add(new_user)
session.commit()
print(f"插入用户 ID：{new_user.id}")

# ---------- 查（Query） ----------
# 查询所有用户
print("所有用户：")
users = session.query(User).all()
for user in users:
    print(user.id, user.username, user.email)

# 查询特定用户
user = session.query(User).filter(User.username == "alice").first()
if user:
    print("找到用户：", user.username, user.email)

# ---------- 改（Update） ----------
if user:
    user.email = "alice_updated@example.com"
    session.commit()
    print("已更新用户邮箱：", user.email)

# ---------- 删（Delete） ----------
if user:
    session.delete(user)
    session.commit()
    print("已删除用户：", user.username)

# ---------- 关闭会话 ----------
session.close()