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())

    x_position = secrets.randbelow(200) + 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)

    return {
        'captcha_key': captcha_key,
        'bg_image': bg_base64,
        'slider_image': slider_base64,
        'x_position': x_position,
        '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']

    if abs(stored_x - x_position) <= TOLERANCE:
        default_cache.delete(cache_key)
        return True

    return False


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']

    if abs(stored_x - x_position) > TOLERANCE:
        return False

    if len(trajectory) < TRAJECTORY_MIN_POINTS:
        return False

    if trajectory[-1]['t'] - trajectory[0]['t'] > TRAJECTORY_MAX_DURATION:
        return False

    if not _analyze_trajectory(trajectory):
        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