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