""" 时间戳验证工具 """ 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.config_key == 'timestamp_tolerance_seconds', SystemConfig.is_deleted == 0 ).first() if config: # 根据配置类型转换值 if config.config_type == 'int': return int(config.config_value) else: return int(config.config_value) 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())