chunyu_project/utils/slider_captcha.py

import io
import math
import secrets
import time
import uuid
import base64
from dataclasses import dataclass
from typing import List, Dict, Optional

from PIL import Image, ImageDraw
from django.core.cache import caches

default_cache = caches['default']

SLIDER_WIDTH = 60
SLIDER_HEIGHT = 60
TOLERANCE = 5
CAPTCHA_TIMEOUT = 300
TRAJECTORY_MIN_POINTS = 5
TRAJECTORY_MAX_DURATION = 10000


class SliderCaptchaError(Exception):
    pass


@dataclass
class SliderPosition:
    x: int
    y: int


def _generate_background_image(width: int = 300, height: int = 150) -> Image.Image:
    image = Image.new('RGB', (width, height), (245, 245, 245))
    draw = ImageDraw.Draw(image)

    for _ in range(5):
        x1 = secrets.randbelow(width)
        y1 = secrets.randbelow(height)
        x2 = secrets.randbelow(width)
        y2 = secrets.randbelow(height)
        color = tuple(secrets.randbelow(200) for _ in range(3))
        draw.line([(x1, y1), (x2, y2)], fill=color, width=1)

    for _ in range(50):
        x = secrets.randbelow(width)
        y = secrets.randbelow(height)
        color = tuple(secrets.randbelow(200) for _ in range(3))
        draw.point((x, y), fill=color)

    return image


def _generate_slider_image(width: int = SLIDER_WIDTH, height: int = SLIDER_HEIGHT) -> Image.Image:
    image = Image.new('RGBA', (width, height), (0, 0, 0, 0))
    draw = ImageDraw.Draw(image)

    draw.rectangle([(0, 0), (width-1, height-1)], fill=(0, 100, 255, 200))
    draw.rectangle([(5, 5), (width-6, height-6)], fill=(0, 120, 255, 180))

    notch_x = width // 2 - 5
    notch_y = height // 2 - 5
    draw.arc([(notch_x, notch_y), (notch_x+10, notch_y+10)], 0, 360, fill=(255, 255, 255), width=2)

    return image


def _cut_slider_from_background(bg_image: Image.Image, position: SliderPosition) -> Image.Image:
    slider_img = bg_image.crop((
        position.x,
        position.y,
        position.x + SLIDER_WIDTH,
        position.y + SLIDER_HEIGHT
    ))
    return slider_img


def _image_to_base64(image: Image.Image) -> str:
    buffer = io.BytesIO()
    image.save(buffer, format='PNG')
    image_base64 = base64.b64encode(buffer.getvalue()).decode('utf-8')
    return f'data:image/png;base64,{image_base64}'


def generate_slider_captcha() -> Dict:
    captcha_key = str(uuid.uuid4())

    # 上限 239：保证缺口在客户端可拖动范围内可达（滑块行程 = 图宽300 - 块宽60 = 240，含 ±5 容差）
    x_position = secrets.randbelow(190) + 50
    y_position = secrets.randbelow(50) + 30

    bg_image = _generate_background_image()

    slider_img = _cut_slider_from_background(bg_image, SliderPosition(x_position, y_position))

    draw = ImageDraw.Draw(bg_image)
    draw.rectangle([
        (x_position, y_position),
        (x_position + SLIDER_WIDTH, y_position + SLIDER_HEIGHT)
    ], fill=(200, 200, 200))

    bg_base64 = _image_to_base64(bg_image)
    slider_base64 = _image_to_base64(slider_img)

    cache_data = {
        'x_position': x_position,
        'y_position': y_position,
        'timestamp': time.time()
    }
    default_cache.set(f'slider_captcha_{captcha_key}', cache_data, timeout=CAPTCHA_TIMEOUT)

    # 安全修复：坐标答案（x）只存服务端缓存，绝不下发客户端（否则脚本可直接读答案绕过）；
    # y_position 非答案，仅用于客户端垂直摆放拼图块
    return {
        'captcha_key': captcha_key,
        'bg_image': bg_base64,
        'slider_image': slider_base64,
        'y_position': y_position,
    }


def verify_slider_captcha(captcha_key: str, x_position: int) -> bool:
    cache_key = f'slider_captcha_{captcha_key}'
    cached_data = default_cache.get(cache_key)

    if cached_data is None:
        raise SliderCaptchaError('验证码已过期')

    stored_x = cached_data['x_position']

    # 安全修复：无论成败都销毁 key，杜绝同一 key 的穷举重试
    default_cache.delete(cache_key)
    return abs(stored_x - x_position) <= TOLERANCE


def verify_slider_trajectory(
    captcha_key: str,
    x_position: int,
    trajectory: List[Dict[str, int]]
) -> bool:
    cache_key = f'slider_captcha_{captcha_key}'
    cached_data = default_cache.get(cache_key)

    if cached_data is None:
        raise SliderCaptchaError('验证码已过期')

    stored_x = cached_data['x_position']

    # 安全修复：x 错误或轨迹校验失败均立即销毁 key，客户端需重新获取验证码
    if abs(stored_x - x_position) > TOLERANCE:
        default_cache.delete(cache_key)
        return False

    if len(trajectory) < TRAJECTORY_MIN_POINTS:
        default_cache.delete(cache_key)
        return False

    if trajectory[-1]['t'] - trajectory[0]['t'] > TRAJECTORY_MAX_DURATION:
        default_cache.delete(cache_key)
        return False

    if not _analyze_trajectory(trajectory):
        default_cache.delete(cache_key)
        return False

    default_cache.delete(cache_key)
    return True


def _analyze_trajectory(trajectory: List[Dict[str, int]]) -> bool:
    if len(trajectory) < 3:
        return False

    velocities = []
    for i in range(1, len(trajectory)):
        dx = trajectory[i]['x'] - trajectory[i-1]['x']
        dt = trajectory[i]['t'] - trajectory[i-1]['t']
        if dt == 0:
            continue
        velocities.append(dx / dt)

    if len(velocities) < 2:
        return False

    velocity_changes = []
    for i in range(1, len(velocities)):
        velocity_changes.append(abs(velocities[i] - velocities[i-1]))

    avg_change = sum(velocity_changes) / len(velocity_changes)
    if avg_change < 0.001:
        return False

    has_pause = False
    for i in range(1, len(trajectory)):
        dt = trajectory[i]['t'] - trajectory[i-1]['t']
        if dt > 50:
            has_pause = True
            break

    return has_pause