import base64 import random import time from app.services.auth import create_access_token from app.utils.redis import get_redis # In-memory fallback when Redis is not available _captcha_store: dict[str, tuple[int, float]] = {} async def generate_slider_captcha() -> dict: """Generate a slider captcha challenge.""" captcha_id = "".join(random.choices("abcdefghijklmnopqrstuvwxyz0123456789", k=16)) redis = get_redis() if redis: await redis.setex(f"captcha:{captcha_id}", 300, "1") else: _captcha_store[captcha_id] = (True, time.time() + 300) bg_data = _create_placeholder_image(300, 150, "#e2e8f0", "拖动滑块到最右侧完成验证") slider_data = _create_placeholder_image(50, 50, "#6366f1", "") return { "captcha_id": captcha_id, "background_image": bg_data, "slider_image": slider_data, "slider_width": 50, } async def verify_slider_captcha(captcha_id: str, x_offset: int) -> str | None: """Verify slider captcha. Return a captcha_token on success.""" redis = get_redis() valid = False if redis: stored = await redis.get(f"captcha:{captcha_id}") if stored: valid = True await redis.delete(f"captcha:{captcha_id}") else: entry = _captcha_store.pop(captcha_id, None) if entry: _, expires = entry if time.time() < expires: valid = True if not valid: return None # User must drag at least 70% of the way (x_offset >= 182 out of 260) if x_offset >= 180: return create_access_token(f"captcha:{captcha_id}") return None def _create_placeholder_image(width: int, height: int, color: str, text: str) -> str: svg = ( f'' f'' f'{text}' f'' ) return f"data:image/svg+xml;base64,{base64.b64encode(svg.encode()).decode()}"