buff3.0/backend-admin/app/utils/timestamp_validator.py

"""
时间戳验证工具
"""
import time
from typing import Optional
from sqlalchemy.orm import Session
from app.core.logger import logger

# 时间戳容忍秒数（硬编码）
TIMESTAMP_TOLERANCE_SECONDS = 300  # 默认5分钟

class TimestampValidator:
    """时间戳验证器"""

    @staticmethod
    def get_timestamp_tolerance(db: Session) -> int:
        """获取时间戳容忍秒数"""
        # 硬编码时间戳容忍值
        return TIMESTAMP_TOLERANCE_SECONDS

    @staticmethod
    def validate_timestamp(request_timestamp: int, db: Session) -> tuple[bool, str]:
        """
        验证请求时间戳是否有效

        Args:
            request_timestamp: 请求中的时间戳
            db: 数据库会话

        Returns:
            tuple: (是否有效, 错误信息)
        """
        try:
            # 获取当前时间戳
            current_timestamp = int(time.time())

            # 获取容忍时间范围
            tolerance = TimestampValidator.get_timestamp_tolerance(db)

            # 计算时间差
            time_diff = abs(current_timestamp - request_timestamp)

            # 检查是否在容忍范围内
            if time_diff > tolerance:
                logger.warning(f"时间戳验证失败: 请求时间戳={request_timestamp}, 当前时间戳={current_timestamp}, 差值={time_diff}秒, 容忍范围={tolerance}秒")
                return False, f"时间戳超出容忍范围({tolerance}秒)"

            logger.debug(f"时间戳验证通过: 请求时间戳={request_timestamp}, 当前时间戳={current_timestamp}, 差值={time_diff}秒")
            return True, ""

        except Exception as e:
            logger.exception("时间戳验证异常")
            return False, f"时间戳验证异常: {str(e)}"

    @staticmethod
    def generate_timestamp() -> int:
        """生成当前时间戳"""
        return int(time.time())