""" 设备信任服务 管理用户的受信任设备，减少频繁验证 """ from __future__ import annotations from datetime import timedelta from app.config import settings from app.log import logger from app.service.client_detection_service import ClientInfo from app.utils import utcnow from redis.asyncio import Redis class DeviceTrustService: """设备信任服务""" @staticmethod def _get_device_trust_key(user_id: int, device_fingerprint: str) -> str: """获取设备信任的 Redis 键""" return f"device_trust:{user_id}:{device_fingerprint}" @staticmethod def _get_location_trust_key(user_id: int, country_code: str) -> str: """获取位置信任的 Redis 键""" return f"location_trust:{user_id}:{country_code}" @staticmethod def _get_verification_cooldown_key(user_id: int) -> str: """获取验证冷却的 Redis 键""" return f"verification_cooldown:{user_id}" @staticmethod async def is_device_trusted( redis: Redis, user_id: int, device_fingerprint: str, ) -> bool: """ 检查设备是否受信任 Args: redis: Redis 连接 user_id: 用户 ID device_fingerprint: 设备指纹 Returns: bool: 设备是否受信任 """ if not device_fingerprint: return False trust_key = DeviceTrustService._get_device_trust_key(user_id, device_fingerprint) trust_data = await redis.get(trust_key) return trust_data is not None @staticmethod async def is_location_trusted( redis: Redis, user_id: int, country_code: str | None, ) -> bool: """ 检查位置是否受信任 Args: redis: Redis 连接 user_id: 用户 ID country_code: 国家代码 Returns: bool: 位置是否受信任 """ if not country_code: return False trust_key = DeviceTrustService._get_location_trust_key(user_id, country_code) trust_data = await redis.get(trust_key) return trust_data is not None @staticmethod async def is_in_verification_cooldown( redis: Redis, user_id: int, ) -> bool: """ 检查用户是否在验证冷却期内 Args: redis: Redis 连接 user_id: 用户 ID Returns: bool: 是否在冷却期内 """ cooldown_key = DeviceTrustService._get_verification_cooldown_key(user_id) cooldown_data = await redis.get(cooldown_key) return cooldown_data is not None @staticmethod async def trust_device( redis: Redis, user_id: int, device_fingerprint: str, client_info: ClientInfo, trust_duration_days: int | None = None, ) -> None: """ 信任设备 Args: redis: Redis 连接 user_id: 用户 ID device_fingerprint: 设备指纹 client_info: 客户端信息 trust_duration_days: 信任持续天数 """ if not device_fingerprint: return # 使用配置中的默认值 if trust_duration_days is None: trust_duration_days = settings.device_trust_duration_days trust_key = DeviceTrustService._get_device_trust_key(user_id, device_fingerprint) trust_data = { "client_type": client_info.client_type, "platform": client_info.platform or "unknown", "trusted_at": utcnow().isoformat(), } # 设置信任期限 trust_duration_seconds = trust_duration_days * 24 * 3600 await redis.setex(trust_key, trust_duration_seconds, str(trust_data)) logger.info( f"[Device Trust] Device trusted for user {user_id}: " f"{client_info.client_type} on {client_info.platform} " f"(fingerprint: {device_fingerprint[:8]}...)" ) @staticmethod async def trust_location( redis: Redis, user_id: int, country_code: str, trust_duration_days: int | None = None, ) -> None: """ 信任位置 Args: redis: Redis 连接 user_id: 用户 ID country_code: 国家代码 trust_duration_days: 信任持续天数 """ if not country_code: return # 使用配置中的默认值 if trust_duration_days is None: trust_duration_days = settings.location_trust_duration_days trust_key = DeviceTrustService._get_location_trust_key(user_id, country_code) trust_data = { "country_code": country_code, "trusted_at": utcnow().isoformat(), } # 设置信任期限 trust_duration_seconds = trust_duration_days * 24 * 3600 await redis.setex(trust_key, trust_duration_seconds, str(trust_data)) logger.info(f"[Location Trust] Location trusted for user {user_id}: {country_code}") @staticmethod async def set_verification_cooldown( redis: Redis, user_id: int, cooldown_seconds: int, ) -> None: """ 设置验证冷却期 Args: redis: Redis 连接 user_id: 用户 ID cooldown_seconds: 冷却时间（秒） """ cooldown_key = DeviceTrustService._get_verification_cooldown_key(user_id) cooldown_data = { "set_at": utcnow().isoformat(), "expires_at": (utcnow() + timedelta(seconds=cooldown_seconds)).isoformat(), } await redis.setex(cooldown_key, cooldown_seconds, str(cooldown_data)) logger.info(f"[Verification Cooldown] Set cooldown for user {user_id}: {cooldown_seconds}s") @staticmethod async def should_require_verification( redis: Redis, user_id: int, device_fingerprint: str | None, country_code: str | None, client_info: ClientInfo, is_new_location: bool, ) -> tuple[bool, str]: """ 判断是否需要验证 Args: redis: Redis 连接 user_id: 用户 ID device_fingerprint: 设备指纹 country_code: 国家代码 client_info: 客户端信息 is_new_location: 是否为新位置 Returns: tuple[bool, str]: (是否需要验证, 原因) """ # 检查验证冷却期 if await DeviceTrustService.is_in_verification_cooldown(redis, user_id): return False, "用户在验证冷却期内" # 检查设备信任 if device_fingerprint and await DeviceTrustService.is_device_trusted(redis, user_id, device_fingerprint): return False, "设备已受信任" # 检查位置信任 if country_code and await DeviceTrustService.is_location_trusted(redis, user_id, country_code): return False, "位置已受信任" # 受信任的客户端类型降低验证要求 if client_info.is_trusted_client and not is_new_location: return False, "受信任客户端且非新位置" # 如果是新位置登录，需要验证 if is_new_location: return True, "新位置登录需要验证" # 默认不需要验证 return False, "常规登录无需验证" @staticmethod async def mark_verification_successful( redis: Redis, user_id: int, device_fingerprint: str | None, country_code: str | None, client_info: ClientInfo, ) -> None: """ 标记验证成功，更新信任信息 Args: redis: Redis 连接 user_id: 用户 ID device_fingerprint: 设备指纹 country_code: 国家代码 client_info: 客户端信息 """ # 信任设备 if device_fingerprint: await DeviceTrustService.trust_device(redis, user_id, device_fingerprint, client_info) # 信任位置 if country_code: await DeviceTrustService.trust_location(redis, user_id, country_code) # 设置验证冷却期 cooldown_seconds = (client_info.is_trusted_client and 3600) or 1800 # 受信任客户端1小时，其他30分钟 await DeviceTrustService.set_verification_cooldown(redis, user_id, cooldown_seconds) logger.info(f"[Device Trust] Verification successful for user {user_id}, trust updated")