captcha_service.py 3.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100
  1. """登录图片验证码:内存签发/校验 + PyMuPDF 渲染 PNG(安全需求4)。
  2. - 单实例部署(launcher 单进程)下内存存储足够;
  3. - 4 位字符(剔除易混淆 0O1I),120 秒过期;
  4. - 校验一次性消费(无论成败),防止同一验证码反复试密码;
  5. - 校验忽略大小写与首尾空白。
  6. """
  7. from __future__ import annotations
  8. import base64
  9. import secrets
  10. import threading
  11. import time
  12. import uuid
  13. import random
  14. from dms.common.errors import InvalidCaptchaError
  15. _CAPTCHA_CHARS = "23456789ABCDEFGHJKLMNPQRSTUVWXYZ"
  16. _CODE_LENGTH = 4
  17. _EXPIRES_SECONDS = 120
  18. _MAX_ENTRIES = 1000
  19. _lock = threading.Lock()
  20. # captcha_id -> (code 大写归一, 过期时刻 monotonic)
  21. _entries: dict[str, tuple[str, float]] = {}
  22. def issue() -> dict[str, str]:
  23. """签发新验证码,返回 ``{"captchaId": ..., "image": "data:image/png;base64,..."}``。"""
  24. code = "".join(secrets.choice(_CAPTCHA_CHARS) for _ in range(_CODE_LENGTH))
  25. captcha_id = uuid.uuid4().hex
  26. now = time.monotonic()
  27. with _lock:
  28. _purge_expired(now)
  29. if len(_entries) >= _MAX_ENTRIES:
  30. # 上限保护:极端刷量时整体清空(存量本就即将过期,不影响正常用户)
  31. _entries.clear()
  32. _entries[captcha_id] = (code.upper(), now + _EXPIRES_SECONDS)
  33. return {
  34. "captchaId": captcha_id,
  35. "image": "data:image/png;base64,"
  36. + base64.b64encode(_render_png(code)).decode("ascii"),
  37. }
  38. def verify(captcha_id: object, code: object) -> None:
  39. """校验并消费验证码;缺失/错误/过期均抛 :class:`InvalidCaptchaError`。"""
  40. if not isinstance(captcha_id, str) or not captcha_id:
  41. raise InvalidCaptchaError()
  42. if not isinstance(code, str) or not code.strip():
  43. raise InvalidCaptchaError()
  44. with _lock:
  45. entry = _entries.pop(captcha_id, None)
  46. if entry is None or entry[1] <= time.monotonic():
  47. raise InvalidCaptchaError()
  48. if entry[0] != code.strip().upper():
  49. raise InvalidCaptchaError()
  50. def _purge_expired(now: float) -> None:
  51. for key in [k for k, (_, expires) in _entries.items() if expires <= now]:
  52. _entries.pop(key, None)
  53. def _render_png(code: str) -> bytes:
  54. """用 PyMuPDF(项目既有依赖)绘制验证码位图:随机颜色/基线/字号 + 干扰线。"""
  55. import fitz
  56. width, height = 132, 44
  57. rng = random.Random()
  58. doc = fitz.open()
  59. try:
  60. page = doc.new_page(width=width, height=height)
  61. page.draw_rect(page.rect, color=None, fill=(0.96, 0.98, 1.0))
  62. for _ in range(4):
  63. page.draw_line(
  64. fitz.Point(rng.uniform(0, width), rng.uniform(0, height)),
  65. fitz.Point(rng.uniform(0, width), rng.uniform(0, height)),
  66. color=(rng.random() * 0.6, rng.random() * 0.6, rng.random() * 0.6),
  67. width=1,
  68. )
  69. x = 14.0
  70. for ch in code:
  71. page.insert_text(
  72. fitz.Point(x, rng.uniform(28, 38)),
  73. ch,
  74. fontname="hebo",
  75. fontsize=rng.uniform(22, 26),
  76. color=(
  77. rng.uniform(0, 0.45),
  78. rng.uniform(0, 0.45),
  79. rng.uniform(0, 0.45),
  80. ),
  81. )
  82. x += 27
  83. return page.get_pixmap().tobytes("png")
  84. finally:
  85. doc.close()