SQL 注入防御策略

1. SQL 注入防御策略 (Defense Strategies)

1.1 核心防御：参数化查询（Prepared Statements）

1.1.1 什么是参数化查询

参数化查询是将 SQL 语句的结构和数据分离的技术。SQL 语句在执行前被预编译，数据作为参数传入，不会被解释为 SQL 代码的一部分。 工作原理：

1. 数据库驱动程序发送 SQL 语句结构到数据库服务器
2. 数据库服务器编译并缓存该语句结构
3. 用户输入作为参数绑定到已编译的语句
4. 数据库知道这些值是数据，不会被解释为 SQL 代码

1.1.2 Python (PyMySQL)

 import pymysql
 def safe_login(username, password):
  connection = pymysql.connect(
  host='localhost',
  user='root',
  password='password',
  database='test'
  )
  try:
  cursor = connection.cursor()
  # 使用参数化查询
  sql = "SELECT * FROM users WHERE username = %s AND password = %s"
  # 注意：%s 是占位符，不是字符串格式化
  cursor.execute(sql, (username, password))
  result = cursor.fetchone()
  return result
  finally:
  connection.close()
 # 高级用法：多次执行同一查询
 def batch_insert(users):
  connection = pymysql.connect(host='localhost', user='root', password='password', database='test')
  try:
  cursor = connection.cursor()
  sql = "INSERT INTO users (username, email) VALUES (%s, %s)"
  # 批量插入
  cursor.executemany(sql, users)
  connection.commit()
  finally:
  connection.close()

1.1.3 Python (SQLAlchemy ORM)

 from sqlalchemy import create_engine, text
 from sqlalchemy.orm import sessionmaker
 engine = create_engine('mysql+pymysql://root:password@localhost/test')
 Session = sessionmaker(bind=engine)
 # 方式一：使用 text() 和参数
 def safe_login_orm(username, password):
  with engine.connect() as conn:
  sql = text("SELECT * FROM users WHERE username = :username AND password = :password")
  result = conn.execute(sql, {"username": username, "password": password})
  return result.fetchone()
 # 方式二：使用 ORM 查询（更安全、更推荐）
 def safe_login_orm2(username, password):
  session = Session()
  try:
  user = session.query(User).filter(
  User.username == username,
  User.password == password
  ).first()
  return user
  finally:
  session.close()
 # 方式三：使用 filter_by
 def get_user_by_id(user_id):
  session = Session()
  try:
  user = session.query(User).filter_by(id=user_id).first()
  return user
  finally:
  session.close()

1.1.4 PHP (PDO)

 <?php
 function safe_login($username, $password) {
  $pdo = new PDO('mysql:host=localhost;dbname=test', 'root', 'password');
  // 使用预处理语句
  $stmt = $pdo->prepare("SELECT * FROM users WHERE username = ? AND password = ?");
  $stmt->execute([$username, $password]);
  return $stmt->fetch();
 }
 // 命名参数方式
 function safe_login2($username, $password) {
  $pdo = new PDO('mysql:host=localhost;dbname=test', 'root', 'password');
  $stmt = $pdo->prepare("SELECT * FROM users WHERE username = :username AND password = :password");
  $stmt->execute([
  ':username' => $username,
  ':password' => $password
  ]);
  return $stmt->fetch();
 }
 // 绑定参数类型
 function safe_insert($username, $email) {
  $pdo = new PDO('mysql:host=localhost;dbname=test', 'root', 'password');
  $stmt = $pdo->prepare("INSERT INTO users (username, email) VALUES (:username, :email)");
  $stmt->bindParam(':username', $username, PDO::PARAM_STR);
  $stmt->bindParam(':email', $email, PDO::PARAM_STR);
  $stmt->execute();
  return $stmt->rowCount();
 }
 ?

1.1.5 Java (JDBC)

 import java.sql.*;
 public class Login {
  public User safeLogin(String username, String password) throws SQLException {
  String url = "jdbc:mysql://localhost:3306/test";
  Connection conn = DriverManager.getConnection(url, "root", "password");
  // 使用 PreparedStatement
  String sql = "SELECT * FROM users WHERE username = ? AND password = ?";
  PreparedStatement pstmt = conn.prepareStatement(sql);
  pstmt.setString(1, username);
  pstmt.setString(2, password);
  ResultSet rs = pstmt.executeQuery();
  User user = null;
  if (rs.next()) {
  user = new User();
  user.setId(rs.getInt("id"));
  user.setUsername(rs.getString("username"));
  user.setPassword(rs.getString("password"));
  }
  rs.close();
  pstmt.close();
  conn.close();
  return user;
  }
  // 批量插入
  public void batchInsert(List<User> users) throws SQLException {
  String url = "jdbc:mysql://localhost:3306/test";
  Connection conn = DriverManager.getConnection(url, "root", "password");
  String sql = "INSERT INTO users (username, email) VALUES (?, ?)";
  PreparedStatement pstmt = conn.prepareStatement(sql);
  conn.setAutoCommit(false);
  for (User user : users) {
  pstmt.setString(1, user.getUsername());
  pstmt.setString(2, user.getEmail());
  pstmt.addBatch();
  }
  pstmt.executeBatch();
  conn.commit();
  pstmt.close();
  conn.close();
  }
 }

1.1.6 Java (MyBatis)

 <!-- 安全的 XML 映射文件 -->
 <select id="getUser" resultType="User">
  SELECT * FROM users
  WHERE username = #{username} AND password = #{password}
 </select>
 <!-- 注意：
  #{param} 使用参数化查询（安全）
  ${param} 直接拼接字符串（危险）
 -
 <!-- 危险示例 -->
 <select id="getUserDangerous" resultType="User">
  SELECT * FROM users
  WHERE username = ${username} -- 危险！
 </select>

 // Java 接口
 public interface UserMapper {
  User getUser(@Param("username") String username, @Param("password") String password);
  List<User> getUsersByIds(@Param("ids") List<Integer> ids);
 }
 // 使用示例
 SqlSession session = sqlSessionFactory.openSession();
 try {
  UserMapper mapper = session.getMapper(UserMapper.class);
  User user = mapper.getUser("admin", "password");
 }
  session.close();
 }

1.2 使用 ORM 框架

1.2.1 Python (Django ORM)

 from django.contrib.auth.models import User
 # Django ORM 自动使用参数化查询
 def login(request):
  username = request.POST.get('username')
  password = request.POST.get('password')
  # 安全的查询方式
  user = User.objects.filter(username=username, password=password).first()
  return user
 # 更安全的做法：使用 authenticate（推荐）
 from django.contrib.auth import authenticate
 def login(request):
  username = request.POST.get('username')
  password = request.POST.get('password')
  # Django 的 authenticate 会自动处理密码哈希
  user = authenticate(username=username, password=password)
  if user is not None:
  login(request, user)
  return
  return False
 # 使用 Q 对象进行复杂查询
 from django.db.models import Q
 def search_users(query):
  users = User.objects.filter(
  Q(username__icontains=query) | Q(email__icontains=query)
  )
  return users

1.2.2 Python (SQLAlchemy)

 from sqlalchemy import Column, Integer, String
 from sqlalchemy.ext.declarative import declarative_base
 Base = declarative_base()
 class User(Base):
  __tablename__ = 'users'
  id = Column(Integer, primary_key=True)
  username = Column(String(50))
  password = Column(String(50))
  email = Column(String(100))
 # 查询
 session = Session()
 user = session.query(User).filter(
  User.username == username,
  User.password == password
 )
 # 使用 filter_by
 user = session.query(User).filter_by(username=username).first()
 # 使用 get
 user = session.query(User).get(user_id)
 # 复杂查询
 from sqlalchemy import and_, or_
 results = session.query(User).filter(
  and_(
  User.username.like('%admin%'),
  or_(User.email.is_(None), User.email != '')
  )
 )

1.2.3 Java (JPA/Hibernate)

 import javax.persistence.EntityManager;
 import javax.persistence.PersistenceContext;
 import javax.persistence.Query;
 import org.springframework.stereotype.Repository;
 @Repository
 public class UserRepository {
  @PersistenceContext
  private EntityManager entityManager;
  // 使用 JPQL 参数化查询
  public User findByUsernameAndPassword(String username, String password) {
  String jpql = "SELECT u FROM User u WHERE u.username = :username AND u.password = :password";
  return entityManager.createQuery(jpql, User.class)
  .setParameter("username", username)
  .setParameter("password", password)
  .getSingleResult();
  }
  // 使用命名参数
  public User findByUsername(String username) {
  String jpql = "SELECT u FROM User u WHERE u.username = :username";
  return entityManager.createQuery(jpql, User.class)
  .setParameter("username", username)
  .getSingleResult();
  }
  // 安全的数据更新
  public void updatePassword(Long userId, String newPassword) {
  String jpql = "UPDATE User u SET u.password = :password WHERE u.id = :id";
  entityManager.createQuery(jpql)
  .setParameter("password", newPassword)
  .setParameter("id", userId)
  .executeUpdate();
  }
 }

1.3 输入验证与过滤

1.3.1 白名单验证

 def safe_get_product(product_id):
  # 白名单验证：只允许数字
  if not product_id.isdigit():
  return None
  # 或者使用正则表达式
  import re
  if not re.match(r'^\d+$', product_id):
  return None
  # 再使用参数化查询
  sql = "SELECT * FROM products WHERE id = %s"
  cursor.execute(sql, (product_id,))
  return cursor.fetchone()
 # 更严格的验证
 def safe_get_user(user_id):
  # 类型检查
  try:
  user_id = int(user_id)
  except (ValueError, TypeError):
  return None
  # 范围检查
  if user_id <= 0 or user_id > 1000000:
  return None
  sql = "SELECT * FROM users WHERE id = %s"
  cursor.execute(sql, (user_id,))
  return cursor.fetchone()

1.3.2 类型转换

 def safe_calculate(a, b, operation):
  try:
  a = float(a)
  b = float(b)
  except (ValueError, TypeError):
  return None
  if operation == 'add':
  return a + b
  elif operation == 'subtract':
  return a - b
  elif operation == 'multiply':
  return a * b
  elif operation == 'divide':
  if b == 0:
  return None
  return a / b
  else:
  return None
 def safe_get_user(user_id):
  try:
  user_id = int(user_id)
  except (ValueError, TypeError):
  return None
  sql = "SELECT * FROM users WHERE id = %s"
  cursor.execute(sql, (user_id,))
  return cursor.fetchone()

1.3.3 输入长度限制

 def safe_login(username, password):
  # 限制输入长度
  if len(username) > 50 or len(password) > 50:
  return None
  # 去除前后空白
  username = username.strip()
  password = password.strip()
  # 检查是否为空
  if not username or not password:
  return None
  # 再使用参数化查询
  sql = "SELECT * FROM users WHERE username = %s AND password = %s"
  cursor.execute(sql, (username, password))
  return cursor.fetchone()

1.4 存储过程（谨慎使用）

1.4.1 安全的存储过程

 DELIMITER //
 CREATE PROCEDURE GetUser(IN p_username VARCHAR(50), IN p_password VARCHAR(50))
 BEGIN
  -- 使用参数，不拼接字符串
  SELECT * FROM users WHERE username = p_username AND password = p_password;
 END //
 DELIMITER ;
 -
 CALL GetUser('admin', '123456');

1.4.2 危险的存储过程

 -
 DELIMITER //
 CREATE PROCEDURE DangerousGetUser(IN p_username VARCHAR(50))
 BEGIN
  SET @sql = CONCAT('SELECT * FROM users WHERE username = ''', p_username, '''');
  PREPARE stmt FROM @sql;
  EXECUTE stmt;
  DEALLOCATE PREPARE stmt;
 END //
 DELIMITER ;
 -

1.5 权限控制

1.5.1 最小权限原则

 -
 CREATE USER 'app_user'@'localhost' IDENTIFIED BY 'strong_password';
 -
 GRANT SELECT, INSERT, UPDATE, DELETE ON test_db.* TO 'app_user'@'localhost';
 -
 REVOKE FILE, SUPER, PROCESS ON *.* FROM 'app_user'@'localhost';
 -
 FLUSH PRIVILEGES;

1.5.2 权限矩阵

1.5.3 权限建议

• 不要使用 root 用户运行应用程序
• 创建专用用户，只授予必要的权限
• 限制用户的访问范围（特定数据库、特定表）
• 禁止 FILE 权限（防止读写文件）
• 禁止 SUPER 权限（防止修改服务器配置）
• 禁止 PROCESS 权限（防止查看其他连接）

1.6 错误信息处理

1.6.1 PHP 错误处理

 <?php
 // 危险：暴露详细错误信息
 mysqli_query($conn, $sql) or die(mysqli_error($conn));
 // 安全：记录错误，返回通用信息
 try {
  mysqli_query($conn, $sql);
 }
  // 记录错误到日志文件
  error_log($e->getMessage());
  // 关闭错误显示
  ini_set('display_errors', 0);
  // 返回通用错误信息
  echo "系统错误，请稍后重试";
 }
 // 生产环境应该这样设置
 ini_set('display_errors', 0);
 error_reporting(E_ALL);
 log_errors = On
 error_log = /var/log/php_errors.log
 ?

1.6.2 Python 错误处理

 import logging
 # 配置日志
 logging.basicConfig(
  filename='app.log',
  level=logging.ERROR,
  format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
 )
 def safe_query(sql, params):
  try:
  cursor.execute(sql, params)
  return cursor.fetchall()
  except Exception as e:
  # 记录详细错误
  logging.error(f"SQL Error: {e}", exc_info=True)
  # 在开发环境可以打印详细错误
  if app.debug:
  print(f"Database Error: {e}")
  # 返回通用错误
  raise Exception("数据库错误，请稍后重试")
 # Flask 示例
 from flask import Flask, jsonify
 import traceback
 app = Flask(__name__)
 @app.errorhandler(500)
 def internal_error(error):
  # 不暴露详细错误信息
  app.logger.error(str(error))
  return jsonify({"error": "Internal server error"}), 500

1.6.3 Java 错误处理

 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 public class UserDAO {
  private static final Logger logger = LoggerFactory.getLogger(UserDAO.class);
  public User findByUsername(String username) {
  String jpql = "SELECT u FROM User u WHERE u.username = :username";
  try {
  return entityManager.createQuery(jpql, User.class)
  .setParameter("username", username)
  .getSingleResult();
  } catch (NoResultException e) {
  return null;
  } catch (Exception e) {
  // 记录详细错误
  logger.error("Error finding user: " + username, e);
  // 抛出通用异常
  throw new DataAccessException("Database error occurred");
  }
  }
 }

1.7 Web 应用防火墙（WAF）

1.7.1 WAF 的作用

• 拦截常见的 SQL 注入 Payload
• 提供额外的安全层
• 即使代码存在漏洞，也能提供保护

1.7.2 常见 WAF 产品

1.7.3 ModSecurity 规则示例

 # 阻止常见 SQL 注入 Payload
 SecRule ARGS "@rx (union.*select|select.*from|insert.*into|update.*set|delete.*from)" \
  "phase:2,deny,status:403,msg:'SQL Injection Attack'"
 # 阻止单引号
 SecRule ARGS "@rx '" \
  "phase:2,deny,status:403,msg:'Single Quote Detected'"
 # 阻止注释符
 SecRule ARGS "@rx (--|#|/\*)" \
  "phase:2,deny,status:403,msg:'SQL Comment Detected'"
 # 阻止关键字组合
 SecRule ARGS "@rx (?i:(and|or).*[\d\s]*[=<>]|having|union.*select)" \
  "phase:2,deny,status:403,msg:'SQL Injection Pattern'"

1.7.4 Cloudflare WAF 规则

 -
 (http.request.uri.path contains "login" and cf.threat_score > 15)
 -
 (cf.threat_score > 50 and not cf.client.bot)

1.8 数据库层面防护

1.8.1 启用 SQL 日志

 -
 SET GLOBAL general_log = 'ON';
 SET GLOBAL general_log_file = '/var/log/mysql/query.log';
 -
 SET GLOBAL log_output = 'TABLE';
 SET GLOBAL general_log = 'ON';
 -
 SET GLOBAL slow_query_log = 'ON';
 SET GLOBAL long_query_time = 1;
 SET GLOBAL slow_query_log_file = '/var/log/mysql/slow.log';
 -
 SELECT * FROM mysql.general_log;
 SELECT * FROM mysql.slow_log;

1.8.2 定期审计

• 定期检查 SQL 日志
• 监控异常查询
• 审计数据库访问
• 监控用户权限变化

1.8.3 数据加密

• 传输加密：使用 SSL/TLS
• 存储加密：敏感数据加密存储
• 密码哈希：使用 bcrypt、Argon2 等强哈希算法

 import bcrypt
 # 密码哈希
 def hash_password(password):
  salt = bcrypt.gensalt()
  hashed = bcrypt.hashpw(password.encode('utf-8'), salt)
  return hashed
 # 密码验证
 def verify_password(password, hashed):
  return bcrypt.checkpw(password.encode('utf-8'), hashed)
 # 使用 argon2
 import argon2
 def hash_password_argon2(password):
  ph = argon2.PasswordHasher()
  return ph.hash(password)
 def verify_password_argon2(password, hashed):
  ph = argon2.PasswordHasher()
  try:
  return ph.verify(hashed, password)
  except argon2.exceptions.VerifyMismatch:
  return False

2. SQL 注入防御最佳实践 (Best Practices)

2.1 开发最佳实践

2.1.1 代码规范

推荐做法：

 [推荐] 使用参数化查询
 [推荐] 使用 ORM 框架
 [推荐] 输入验证（白名单）
 [推荐] 类型转换
 [推荐] 长度限制
 [推荐] 错误处理
 [推荐] 代码审查
 [推荐] 安全测试

禁止做法：

 [禁止] 直接拼接 SQL
 [禁止] 使用字符串格式化
 [禁止] 信任用户输入
 [禁止] 暴露详细错误
 [禁止] 使用 root 用户
 [禁止] 过度授权

2.1.2 安全开发流程

1. 需求阶段：考虑安全需求
2. 设计阶段：安全设计，威胁建模
3. 编码阶段：遵循安全编码规范
4. 测试阶段：安全测试，包括渗透测试
5. 部署阶段：安全配置
6. 运维阶段：安全监控，定期审计

2.2 测试最佳实践

2.2.1 安全测试清单

• 所有输入点测试
• 参数化查询验证
• 输入验证测试
• 权限控制测试
• 错误信息测试
• SQLMap 扫描
• 代码审计
• 渗透测试

2.2.2 自动化测试

 import pytest
 import requests
 class TestSQLInjection:
  @pytest.fixture
  def base_url(self):
  return "http://example.com"
  def test_login_endpoint(self, base_url):
  """测试登录端点的 SQL 注入"""
  payloads = [
  "' OR '1'='1",
  "' --",
  "1' UNION SELECT 1,2,3 --",
  "1' AND SLEEP(5) --",
  ]
  for payload in payloads:
  response = requests.post(
  f"{base_url}/login",
  data={"username": payload, "password": "test"}
  )
  # 检查是否返回了不应该返回的数据
  assert "管理员" not in response.text
  assert "admin" not in response.text.lower() or response.status_code == 401
  def test_search_endpoint(self, base_url):
  """测试搜索端点"""
  response = requests.get(f"{base_url}/search", params={"q": "' OR 1=1 --"})
  # 验证不会返回所有数据
  assert response.status_code == 400 or len(response.json()) == 0

2.3 运维最佳实践

2.3.1 服务器配置

• 使用最新版本的 MySQL
• 定期更新补丁
• 启用防火墙
• 配置安全组
• 禁用不必要的功能
• 关闭远程访问（如果不是必需的）

2.3.2 监控与审计

• 启用日志
• 定期备份
• 实时监控
• 定期审计
• 应急响应

2.3.3 安全配置清单

 # MySQL 安全配置
 # 1. 禁用远程 root 登录
 mysql> DELETE FROM mysql.user WHERE User='root' AND Host NOT IN ('localhost', '127.0.0.1');
 # 2. 删除匿名用户
 mysql> DELETE FROM mysql.user WHERE User='';
 # 3. 设置密码策略
 mysql> SET GLOBAL validate_password_policy=STRONG;
 # 4. 限制用户连接
 mysql> CREATE USER 'app'@'localhost' WITH MAX_CONNECTIONS_PER_HOUR 100;
 # 5. 启用审计日志（企业版）
 mysql> INSTALL PLUGIN audit_log SONAME 'audit_log.so';

3. SQL 注入常见问题 (FAQ)

Q1：参数化查询能完全防止 SQL 注入吗？

A：是的，参数化查询是防止 SQL 注入最有效的方法。但需要注意：

• 不要在参数化查询中拼接表名或列名
• 表名/列名需要使用白名单验证
• 存储过程的参数化也需要注意动态 SQL 的使用

Q2：使用了 ORM 就一定安全吗？

A：不一定。如果 ORM 被错误使用，仍然可能存在 SQL 注入：

 # 危险：使用 raw SQL 拼接
 session.execute(f"SELECT * FROM users WHERE id = {user_id}")
 # 安全：使用 ORM 查询
 session.query(User).filter(User.id == user_id).first()
 # 危险：使用 filter with text
 session.query(User).filter(text(f"id = {user_id}")).first()

Q3：转义单引号能防止 SQL 注入吗？

A：不完全能。存在以下绕过方式：

• 宽字节注入
• 二次注入
• 数字类型注入（不需要单引号）
• 字符集问题

Q4：如何防止表名/列名动态拼接的注入？

A：使用白名单验证：

 def safe_query(table_name, column_name, value):
  # 白名单验证
  allowed_tables = ['users', 'products', 'orders']
  allowed_columns = ['id', 'name', 'price']
  if table_name not in allowed_tables:
  raise ValueError("Invalid table name")
  if column_name not in allowed_columns:
  raise ValueError("Invalid column name")
  # 使用参数化查询
  sql = f"SELECT * FROM {table_name} WHERE {column_name} = %s"
  cursor.execute(sql, (value,))
  return cursor.fetchall()

Q5：SQL 注入只存在于 Web 应用吗？

A：不是。任何使用数据库且用户输入可控的应用都可能存在 SQL 注入：

• 桌面应用
• 移动应用
• API 服务
• 脚本工具
• 命令行工具

Q6：如何检测已有的 SQL 注入漏洞？

A：可以使用以下方法：

1. 代码审计
2. SQLMap 等自动化工具扫描
3. 手动渗透测试
4. Web 应用安全扫描器

Q7：云数据库是否还需要担心 SQL 注入？

A：是的。云数据库同样面临 SQL 注入风险，因为：

• 应用程序代码可能存在注入漏洞
• 云不等于安全
• 需要在应用层面做好防护

4. 总结 (Summary)

4.1 核心要点

4.2 防御 Checklist

• 使用参数化查询
• 使用 ORM 框架
• 输入验证（白名单）
• 类型转换
• 长度限制
• 最小权限原则
• 错误信息处理
• 定期安全测试
• 代码审计
• 监控与审计

4.3 学习建议

1. 理解原理：深入理解 SQL 注入的原理和机制
2. 实践操作：搭建测试环境，进行攻防演练
3. 学习工具：掌握 SQLMap、Burp Suite 等工具
4. 代码审计：学习如何查找和修复漏洞
5. 持续学习：关注新的攻击技术和防御方法

更新日志 (Changelog)

• 2026-05-27: 拆分为独立文件，添加元数据，版本升级至 v1.0.0
• 2026-05-03: 创建 SQL 注入安全防御文档