219 lines
8.6 KiB
Python
219 lines
8.6 KiB
Python
"""
|
|
captcha — 基于 pymsg 协议（MsgServer）的图形验证码服务。
|
|
|
|
替代原 http.server HTTP 接口，通过 TCP 长度前缀帧收发 JSON。
|
|
协议：每条消息为完整 JSON，``_route`` 字段对应路由名。
|
|
|
|
请求示例::
|
|
|
|
{"_route": "image"} → 生成验证码图片
|
|
{"_route": "verify", "token": "...", "answer": "AB3XY"} → 校验
|
|
{"_route": "health"} → 健康检查
|
|
|
|
响应格式（与 dbv2 一致）::
|
|
|
|
{"success": true, "data": {...}}
|
|
{"success": false, "error": "..."}
|
|
|
|
说明：PNG 图片为二进制，经 base64 编码后放入 JSON 返回；
|
|
调用方（gateway）解码后以 image/png 代理给浏览器。
|
|
"""
|
|
|
|
import base64
|
|
import io
|
|
import json
|
|
import os
|
|
import random
|
|
import secrets
|
|
import string
|
|
import threading
|
|
import time
|
|
|
|
from PIL import Image, ImageDraw, ImageFont
|
|
|
|
# pymsg 导入
|
|
import sys as _sys
|
|
_sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
|
from pymsg import MsgServer # noqa: E402
|
|
|
|
# ── 配置 ────────────────────────────────────────────────────────────────
|
|
|
|
CAPTCHA_LEN = 4
|
|
CAPTCHA_TTL = 300 # Token 有效期（秒）
|
|
|
|
CHARS = string.ascii_letters + string.digits
|
|
CHARS = CHARS.translate(str.maketrans('', '', 'lI1O0o')) # 剔除易混淆字符
|
|
|
|
captcha_store = {} # token -> {'answer': str, 'created_at': float}
|
|
STORE_LOCK = threading.Lock()
|
|
|
|
|
|
# ── 验证码生成 ──────────────────────────────────────────────────────────
|
|
|
|
def generate_text():
|
|
return ''.join(random.choices(CHARS, k=CAPTCHA_LEN))
|
|
|
|
|
|
def generate_image(text):
|
|
width, height = 160, 50
|
|
bg = (random.randint(0,256), random.randint(0,256), random.randint(0,256))
|
|
img = Image.new('RGB', (width, height), bg)
|
|
draw = ImageDraw.Draw(img)
|
|
|
|
for i, ch in enumerate(text):
|
|
color = (random.randint(0,256), random.randint(0,256), random.randint(0,256))
|
|
y = random.randint(2, 12)
|
|
x = 15 + i * 32 + random.randint(-3, 3) # 4 位字符在 160px 宽度内均匀分布
|
|
try:
|
|
font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf",
|
|
random.choice(range(20,40,2)))
|
|
except (IOError, OSError):
|
|
font = ImageFont.load_default()
|
|
|
|
char = Image.new("RGBA", (64, 64), (0,0,0,0))
|
|
draw_char = ImageDraw.Draw(char)
|
|
draw_char.text((20,20), ch, fill=color, font=font)
|
|
rotated_char = char.rotate(random.randint(-30, 30), expand=True)
|
|
img.paste(rotated_char, (x-32, y-32), mask=rotated_char)
|
|
# draw.text((x, y), ch, fill=color, font=font)
|
|
|
|
for _ in range(4):
|
|
draw.line([(random.randint(0, width), random.randint(0, height)),
|
|
(random.randint(0, width), random.randint(0, height))],
|
|
fill=(random.randint(0,256), random.randint(0,256), random.randint(0,256)))
|
|
for _ in range(50):
|
|
draw.point((random.randint(0, width - 1), random.randint(0, height - 1)),
|
|
fill=(random.randint(0,256), random.randint(0,256), random.randint(0,256)))
|
|
|
|
buf = io.BytesIO()
|
|
img.save(buf, format='PNG')
|
|
return buf.getvalue()
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# CaptchaServer — 主服务类
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
class CaptchaServer:
|
|
"""基于 pymsg 的验证码服务，通过 TCP 长连接收发 JSON。"""
|
|
|
|
def __init__(self, addr="127.0.0.1", port=8009):
|
|
self.addr = addr
|
|
self.port = port
|
|
self._server = MsgServer(addr, port)
|
|
|
|
# 路由表
|
|
self._routes = {
|
|
"image": self._handle_image,
|
|
"verify": self._handle_verify,
|
|
"health": self._handle_health,
|
|
}
|
|
|
|
# ── 启动 / 停止 ──────────────────────────────────────────────────
|
|
def start(self) -> None:
|
|
"""启动服务（阻塞当前线程）。"""
|
|
print(f"[captcha] 监听 tcp://{self.addr}:{self.port}")
|
|
self._server.start(self._dispatch)
|
|
|
|
def stop(self) -> None:
|
|
"""从另一线程停止服务。"""
|
|
self._server.stop()
|
|
|
|
# ── 消息调度 -------------------------------------------------------
|
|
def _dispatch(self, sock, addr, text):
|
|
"""解析 JSON 中的 _route，分发给对应的 handler，并回复结果。"""
|
|
try:
|
|
req = json.loads(text)
|
|
except json.JSONDecodeError as e:
|
|
self._reply(sock, False, error=f"JSON 解析失败: {e}")
|
|
return
|
|
|
|
route = req.pop("_route", "")
|
|
handler = self._routes.get(route)
|
|
if handler is None:
|
|
self._reply(sock, False, error=f"未知路由: {route}")
|
|
return
|
|
|
|
try:
|
|
result = handler(req)
|
|
self._reply(sock, True, data=result)
|
|
except Exception as e:
|
|
self._reply(sock, False, error=str(e))
|
|
|
|
def _reply(self, sock, success, data=None, error=None):
|
|
resp = {"success": success}
|
|
if data is not None:
|
|
resp["data"] = data
|
|
if error is not None:
|
|
resp["error"] = error
|
|
self._server.reply(sock, json.dumps(resp, ensure_ascii=False))
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
# 路由处理
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
def _handle_image(self, _req):
|
|
"""生成验证码图片，返回 token 与 base64 编码的 PNG。"""
|
|
text = generate_text()
|
|
token = secrets.token_hex(16)
|
|
|
|
with STORE_LOCK:
|
|
captcha_store[token] = {'answer': text.lower(), 'created_at': time.time()}
|
|
# 惰性清理过期项
|
|
now = time.time()
|
|
expired = [k for k, v in captcha_store.items() if now - v['created_at'] > CAPTCHA_TTL]
|
|
for k in expired:
|
|
del captcha_store[k]
|
|
|
|
png_bytes = generate_image(text)
|
|
return {
|
|
"token": token,
|
|
"image": base64.b64encode(png_bytes).decode('ascii'),
|
|
}
|
|
|
|
def _handle_verify(self, req):
|
|
"""校验答案，Token 一次性消费；恒定时间比较防计时攻击。"""
|
|
token = str(req.get("token", "")).strip()
|
|
answer = str(req.get("answer", "")).strip().lower()
|
|
if not token or not answer:
|
|
return {"valid": False, "error": "token and answer required"}
|
|
|
|
with STORE_LOCK:
|
|
entry = captcha_store.pop(token, None)
|
|
|
|
if not entry:
|
|
return {"valid": False, "error": "invalid or expired token"}
|
|
|
|
expected = entry['answer']
|
|
if len(answer) != len(expected):
|
|
valid = False
|
|
else:
|
|
diff = 0
|
|
for a, b in zip(answer, expected):
|
|
diff |= ord(a) ^ ord(b)
|
|
valid = diff == 0
|
|
|
|
return {"valid": valid}
|
|
|
|
def _handle_health(self, _req):
|
|
return {"status": "ok", "service": "captcha"}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
# 入口
|
|
# ═══════════════════════════════════════════════════════════════════════
|
|
|
|
if __name__ == '__main__':
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="captcha — pymsg 验证码服务")
|
|
parser.add_argument("--addr", default="127.0.0.1", help="监听地址")
|
|
parser.add_argument("--port", type=int, default=8009, help="监听端口")
|
|
args = parser.parse_args()
|
|
|
|
server = CaptchaServer(args.addr, args.port)
|
|
try:
|
|
server.start()
|
|
except KeyboardInterrupt:
|
|
server.stop()
|