Python分页安全案例如何保障分页查询

Python分页安全案例：保障分页查询的完整方案

基础安全分页实现

from flask import Flask, request, jsonify
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
from typing import Optional, Tuple
import re
import hashlib
import time
app = Flask(__name__)
# 数据库配置
DATABASE_URL = "mysql+pymysql://user:password@localhost/dbname"
engine = create_engine(DATABASE_URL)
Session = sessionmaker(bind=engine)
class SecurePaginator:
    """安全分页器"""
    def __init__(self, session):
        self.session = session
    def validate_page_params(self, page: int, per_page: int) -> Tuple[int, int]:
        """验证分页参数"""
        # 设置安全范围
        MAX_PER_PAGE = 100
        MIN_PER_PAGE = 1
        MAX_PAGE = 10000
        # 验证并限制每页数量
        try:
            per_page = int(per_page)
        except (ValueError, TypeError):
            per_page = 20  # 默认值
        per_page = max(MIN_PER_PAGE, min(per_page, MAX_PER_PAGE))
        # 验证并限制页码
        try:
            page = int(page)
        except (ValueError, TypeError):
            page = 1
        page = max(1, min(page, MAX_PAGE))
        return page, per_page
    def paginate_safe(self, query, page: int, per_page: int, 
                     allowed_sort_fields: list = None, 
                     sort_by: str = "", 
                     sort_order: str = "asc"):
        """
        安全分页查询
        """
        # 1. 验证分页参数
        page, per_page = self.validate_page_params(page, per_page)
        # 2. 计算偏移量
        offset = (page - 1) * per_page
        # 3. 安全处理排序
        if allowed_sort_fields and sort_by:
            # 白名单验证
            if sort_by not in allowed_sort_fields:
                sort_by = allowed_sort_fields[0]  # 默认排序字段
            # 验证排序方向
            if sort_order.lower() not in ['asc', 'desc']:
                sort_order = 'asc'
            # 安全拼接排序
            sort_column = f"`{sort_by}`"  # 使用反引号防止注入
            order_clause = f"{sort_column} {sort_order}"
            query = query.order_by(text(order_clause))
        # 4. 执行分页查询
        total = query.count()
        items = query.limit(per_page).offset(offset).all()
        # 5. 计算总页数
        total_pages = (total + per_page - 1) // per_page
        return {
            'items': items,
            'page': page,
            'per_page': per_page,
            'total': total,
            'total_pages': total_pages,
            'has_next': page < total_pages,
            'has_previous': page > 1
        }
# 路由示例
@app.route('/api/users', methods=['GET'])
def get_users():
    """安全的用户分页查询"""
    session = Session()
    try:
        paginator = SecurePaginator(session)
        # 获取并验证参数
        page = request.args.get('page', 1, type=int)
        per_page = request.args.get('per_page', 20, type=int)
        sort_by = request.args.get('sort_by', 'id')
        sort_order = request.args.get('sort_order', 'asc')
        # 允许排序的字段白名单
        allowed_sort_fields = ['id', 'username', 'email', 'created_at']
        # 构建基础查询
        from models import User
        query = session.query(User)
        # 执行安全分页
        result = paginator.paginate_safe(
            query, 
            page, 
            per_page, 
            allowed_sort_fields, 
            sort_by, 
            sort_order
        )
        # 序列化结果
        serialized_items = []
        for user in result['items']:
            serialized_items.append({
                'id': user.id,
                'username': user.username,
                'email': user.email,
                'created_at': user.created_at.isoformat()
            })
        result['items'] = serialized_items
        return jsonify(result)
    except Exception as e:
        return jsonify({'error': str(e)}), 500
    finally:
        session.close()
# 使用ORM的完整示例（SQLAlchemy）
@app.route('/api/orders', methods=['GET'])
def get_orders():
    """使用ORM的安全分页"""
    session = Session()
    try:
        from models import Order
        # 安全获取参数
        page = max(1, int(request.args.get('page', 1)))
        per_page = min(100, max(1, int(request.args.get('per_page', 20))))
        # 使用filter进行安全过滤
        status = request.args.get('status')
        user_id = request.args.get('user_id')
        query = session.query(Order)
        # 安全添加过滤条件
        if status and status in ['pending', 'completed', 'cancelled']:
            query = query.filter(Order.status == status)
        if user_id and user_id.isdigit():
            query = query.filter(Order.user_id == int(user_id))
        # 使用原生SQLAlchemy分页
        pagination = query.paginate(page=page, per_page=per_page, error_out=False)
        return jsonify({
            'items': [order.to_dict() for order in pagination.items],
            'page': pagination.page,
            'per_page': pagination.per_page,
            'total': pagination.total,
            'pages': pagination.pages,
            'has_prev': pagination.has_prev,
            'has_next': pagination.has_next
        })
    except Exception as e:
        return jsonify({'error': str(e)}), 500
    finally:
        session.close()

防SQL注入的游标分页

class CursorBasedPaginator:
    """基于游标的分页，防止偏移量攻击"""
    def __init__(self, session):
        self.session = session
    def paginate_by_cursor(self, model, cursor: Optional[str] = None, 
                          limit: int = 20, direction: str = 'next'):
        """使用游标进行安全分页"""
        # 验证参数
        limit = min(100, max(1, limit))
        # 构建查询
        query = self.session.query(model)
        if direction == 'next' and cursor:
            # 使用WHERE条件代替OFFSET
            query = query.filter(model.id < int(cursor))
            query = query.order_by(model.id.desc())
        elif direction == 'prev' and cursor:
            query = query.filter(model.id > int(cursor))
            query = query.order_by(model.id.asc())
        else:
            query = query.order_by(model.id.desc())
        items = query.limit(limit).all()
        # 生成新的游标
        next_cursor = str(items[-1].id) if items else None
        prev_cursor = str(items[0].id) if items else None
        return {
            'items': [item.to_dict() for item in items],
            'next_cursor': next_cursor,
            'prev_cursor': prev_cursor,
            'limit': limit,
            'has_next': len(items) == limit
        }

防止参数篡改的签名验证

import hmac
import base64
from urllib.parse import quote, unquote
class SignedPaginator:
    """带签名的分页器，防止参数篡改"""
    def __init__(self, secret_key: str = "your-secret-key"):
        self.secret_key = secret_key.encode()
    def generate_page_signature(self, page: int, per_page: int, 
                               timestamp: int = None) -> str:
        """生成分页参数的签名"""
        if not timestamp:
            timestamp = int(time.time())
        # 创建签名字符串
        message = f"{page}:{per_page}:{timestamp}"
        signature = hmac.new(
            self.secret_key,
            message.encode(),
            hashlib.sha256
        ).digest()
        # Base64编码
        return base64.urlsafe_b64encode(signature).decode()
    def validate_page_request(self, page: int, per_page: int, 
                             signature: str, timestamp: int) -> bool:
        """验证分页请求的签名"""
        # 检查时间戳是否过期（5分钟内有效）
        current_time = int(time.time())
        if abs(current_time - timestamp) > 300:  # 5分钟
            return False
        # 重新计算签名并比较
        expected_signature = self.generate_page_signature(page, per_page, timestamp)
        return hmac.compare_digest(expected_signature, signature)
# 使用示例
@app.route('/api/secure/users', methods=['GET'])
def get_secure_users():
    """使用签名的安全分页API"""
    try:
        # 获取签名参数
        page = request.args.get('page', 1, type=int)
        per_page = request.args.get('per_page', 20, type=int)
        signature = request.args.get('sig', '')
        timestamp = request.args.get('ts', 0, type=int)
        # 验证签名
        signed_paginator = SignedPaginator()
        if not signed_paginator.validate_page_request(page, per_page, signature, timestamp):
            return jsonify({'error': 'Invalid or expired request'}), 403
        # 继续处理请求...
        # 这里添加实际的业务逻辑
        return jsonify({'message': 'Request validated'})
    except Exception as e:
        return jsonify({'error': str(e)}), 400

速率限制和防爬虫

from functools import wraps
from collections import defaultdict
import threading
class RateLimiter:
    """速率限制器"""
    def __init__(self):
        self.locks = defaultdict(threading.Lock)
        self.counts = defaultdict(int)
        self.reset_times = defaultdict(float)
    def is_allowed(self, key: str, max_requests: int = 100, 
                   window_seconds: int = 60) -> bool:
        """检查是否允许请求"""
        current_time = time.time()
        with self.locks[key]:
            # 重置计数器（如果窗口已过）
            if current_time - self.reset_times[key] > window_seconds:
                self.counts[key] = 0
                self.reset_times[key] = current_time
            self.counts[key] += 1
            return self.counts[key] <= max_requests
# 装饰器实现
def rate_limit(max_requests=100, window_seconds=60):
    """速率限制装饰器"""
    limiter = RateLimiter()
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            # 使用IP地址作为key
            key = request.remote_addr
            if not limiter.is_allowed(key, max_requests, window_seconds):
                return jsonify({
                    'error': 'Too many requests',
                    'retry_after': window_seconds
                }), 429
            return f(*args, **kwargs)
        return decorated_function
    return decorator
@app.route('/api/rate-limited/users')
@rate_limit(max_requests=30, window_seconds=60)
def get_rate_limited_users():
    """带速率限制的分页API"""
    # 实现逻辑
    pass

综合安全分页示例

# models.py
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
Base = declarative_base()
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    username = Column(String(50), nullable=False)
    email = Column(String(100), nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    def to_dict(self):
        return {
            'id': self.id,
            'username': self.username,
            'email': self.email,
            'created_at': self.created_at.isoformat()
        }
# 完整的安全分页API
@app.route('/api/safe/users', methods=['GET'])
@rate_limit(max_requests=60, window_seconds=60)
def get_safe_users():
    """完整的安全分页API"""
    session = Session()
    try:
        # 1. 参数验证
        page = request.args.get('page', 1, type=int)
        per_page = request.args.get('per_page', 20, type=int)
        # 2. 参数范围限制
        page = max(1, min(page, 10000))
        per_page = max(1, min(per_page, 100))
        # 3. 构建安全查询
        query = session.query(User)
        # 4. 添加安全过滤条件
        search = request.args.get('search', '').strip()
        if search and len(search) <= 50:
            # 使用参数化查询防止SQL注入
            query = query.filter(User.username.like(f'%{search}%'))
        # 5. 安全排序
        sort_by = request.args.get('sort_by', 'id')
        sort_order = request.args.get('sort_order', 'asc')
        allowed_sort = {
            'id': User.id,
            'username': User.username,
            'created_at': User.created_at
        }
        if sort_by in allowed_sort:
            column = allowed_sort[sort_by]
            query = query.order_by(column.desc() if sort_order == 'desc' else column)
        # 6. 执行分页
        total = query.count()
        offset = (page - 1) * per_page
        users = query.limit(per_page).offset(offset).all()
        # 7. 构建响应
        return jsonify({
            'data': [user.to_dict() for user in users],
            'pagination': {
                'page': page,
                'per_page': per_page,
                'total': total,
                'total_pages': (total + per_page - 1) // per_page,
                'has_next': page < (total + per_page - 1) // per_page,
                'has_prev': page > 1
            },
            'meta': {
                'timestamp': datetime.utcnow().isoformat(),
                'request_id': str(uuid.uuid4())[:8]
            }
        })
    except Exception as e:
        return jsonify({'error': 'Internal server error'}), 500
    finally:
        session.close()

安全最佳实践总结

1. 输入验证：

   

   • 限制分页参数范围（如：per_page ≤ 100）
   • 使用类型转换确保参数合法
   • 对排序字段使用白名单

2. SQL注入防护：

   • 使用参数化查询或ORM
   • 避免字符串拼接SQL
   • 对用户输入进行转义

3. 性能防护：

   • 限制最大偏移量
   • 使用索引优化查询
   • 考虑游标分页替代偏移量

4. 访问控制：

   • 实现速率限制
   • 添加请求签名
   • 验证用户权限

5. 错误处理：

   • 不暴露敏感信息
   • 返回统一的错误格式
   • 记录详细的错误日志

这些措施可以确保你的分页查询既安全又高效。