buff/backend/app/utils/timestamp_validator.py

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

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

    @staticmethod
    def get_timestamp_tolerance(db: Session) -> int:
        """获取时间戳容忍秒数"""
        try:
            config = db.query(SystemConfig).filter(SystemConfig.is_deleted == 0).first()
            if config and config.timestamp_tolerance_seconds:
                return config.timestamp_tolerance_seconds
            return 300  # 默认5分钟
        except Exception as e:
            logger.exception("获取时间戳容忍配置失败")
            return 300

    @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())