""" 时间戳验证工具 """ 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())