qing/micro_servers
1
3 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: captcha 改造为 pymsg/TCP 服务，gateway 新增图形码代理路由

Browse Source 
captcha 参照 dbv2 统一为 pymsg 协议（image/verify/health 路由），与主干 RPC 栈对齐；gateway 新增 GET /captcha/image 做 HTTP→pymsg 转换，解码 base64 PNG 并以原契约（image/png + X-Captcha-Token）返回浏览器。
This commit is contained in:
qing committed 2026-07-28 09:54:08 +08:00
1 parent d06546f6bf
commit 44c1e5ef72
4 files changed
+307 -224
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

captcha/README.md
+75 -123
View File
@@ -1,24 +1,24 @@
# CAPTCHA 验证码服务
一个轻量级的图片验证码服务，仅依赖 Python 标准库实现 HTTP（无需 Flask），配合 Pillow 生成验证码图片。提供图片生成、Token 校验、健康检查三个接口，可作为后端服务的验证码组件独立部署。
一个轻量级的图片验证码服务，基于自研 [pymsg](../pymsg/) 协议（TCP 长度前缀帧 + JSON），配合 Pillow 生成验证码图片。提供图片生成、Token 校验、健康检查三个路由，由 `gateway` 经 pymsg 调用并代理给浏览器。
服务由 `core`（PMC）的 `init.txt` 拉起，监听 `127.0.0.1:8009`，仅允许本机/内网调用；`gateway` 在登录与注册中可选地启用 CAPTCHA 校验（见 `gateway/README.md`）。
服务由 `core`（PMC）的 `init.txt` 拉起，监听 `127.0.0.1:8009`（TCP），仅允许本机/内网调用。
## 功能特性
- **图片验证码生成**：5 位字符（字母 + 数字），自动排除易混淆字符（`l I 1 O 0 o`）
- **Token 机制**：每次生成图片返回唯一 Token（32 位十六进制），校验时**一次性消费**，防止重放
- **过期自动清理**：Token 默认 300 秒（5 分钟）失效；`/captcha/image` 请求时顺带清理过期项，`/captcha/verify` 时若 Token 已不在存储即判为失效
- **过期自动清理**：Token 默认 300 秒（5 分钟）失效；`image` 路由请求时顺带清理过期项
- **线程安全**：使用锁（`STORE_LOCK`）保护内存存储，支持并发请求
- **恒定时间比较**：答案校验使用逐字符异或（`diff |= ord(a) ^ ord(b)`），避免计时攻击
- **干扰元素**：随机字符颜色、垂直位置偏移、干扰线与噪点，提升机器识别难度
- **零框架依赖**：HTTP 层基于标准库 `http.server`，除 Pillow 外无第三方依赖
- **零框架依赖**：基于 pymsg（项目内置库），除 Pillow 外无第三方依赖
## 技术栈
- Python 3
- [Pillow](https://python-pillow.org/)（PIL）—— 图片绘制
- Python 标准库：`http.server`、`threading`、`secrets`、`json`、`urllib`
- [pymsg](../pymsg/) —— TCP 长度前缀帧协议（`[4字节大端长度][UTF-8 JSON]`）
## 安装
@@ -35,158 +35,109 @@ pip install Pillow
cd captcha && ./_run.sh
# 方式二：直接运行
python3 captcha_service.py
python3 captcha_service.py # 默认 127.0.0.1:8009
python3 captcha_service.py --port 8009 --addr 127.0.0.1
```
服务启动后监听 `http://127.0.0.1:8009`。
## 协议与路由
## 配置项
所有可调参数均为 `captcha_service.py` 顶部的模块级常量：
| 常量 | 默认值 | 说明 |
| ------------- | ------------ | --------------------------------- |
| `CAPTCHA_LEN` | `5` | 验证码字符长度 |
| `CAPTCHA_TTL` | `300` | Token 有效期（秒） |
| `ADDR` | `127.0.0.1` | 监听地址 |
| `PORT` | `8009` | 监听端口 |
## API 接口
### 接口总览
| 方法 | 路径 | 说明 |
| ---- | ----------------- | --------------------------------------------- |
| GET | `/captcha/image` | 生成验证码图片，响应头下发 `X-Captcha-Token` |
| POST | `/captcha/verify` | 校验用户输入，一次性消费 Token |
| GET | `/health` | 健康检查 |
---
### 1. 获取验证码图片
```
GET /captcha/image
```
生成一组新验证码（5 位字符 + 对应 Token），返回 PNG 图片二进制流，并在响应头 `X-Captcha-Token` 中携带本次校验所需的 Token。
**请求**：无参数、无请求体。
**响应头**：
| 头字段 | 取值 | 说明 |
| ----------------- | --------------- | --------------------------------- |
| `Content-Type` | `image/png` | 图片格式 |
| `Content-Length` | 数值 | 图片字节数 |
| `X-Captcha-Token` | 32 位十六进制串 | 本次验证码的 Token，校验时必须回传 |
**响应体**：二进制 PNG 图片，尺寸 160 × 50。
**示例**：
```bash
curl -D - http://127.0.0.1:8009/captcha/image -o captcha.png
```
响应头示例：
```
Content-Type: image/png
Content-Length: 3210
X-Captcha-Token: 3f9a1c2b4d5e6f7a8b9c0d1e2f3a4b5c
```
> Token 有效期 300 秒，且**一次性使用**——无论校验成功与否，第一次 `/captcha/verify` 调用后即被消费。
---
### 2. 校验验证码
```
POST /captcha/verify
Content-Type: application/json
```
提交 `/captcha/image` 下发的 Token 与用户输入，比对答案（大小写不敏感）。**Token 在调用后被立即移除（一次性）。**
**请求体字段**：
| 字段 | 类型 | 必填 | 说明 |
| ------- | ------ | ---- | ------------------------------------------ |
| `token` | string | 是 | `GET /captcha/image` 响应头中的 `X-Captcha-Token` |
| `answer` | string | 是 | 用户输入的验证码文本，校验前统一转小写，大小写不敏感 |
**请求体示例**：
服务基于 pymsg 协议：每条消息为完整 JSON，`_route` 字段指定路由。响应格式与 db 服务一致：
```json
{
"token": "3f9a1c2b4d5e6f7a8b9c0d1e2f3a4b5c",
"answer": "AB3XY"
}
{"success": true, "data": {...}}
{"success": false, "error": "..."}
```
| _route | 说明 |
| ------- | ------------------------------------------ |
| `image` | 生成验证码图片（base64 PNG + Token） |
| `verify` | 校验用户输入，一次性消费 Token |
| `health` | 健康检查 |
### 1. image — 生成验证码图片
**请求**：
```json
{"_route": "image"}
```
**响应**：
| HTTP 状态码 | 响应体 | 触发条件 |
| ----------- | ------------------------------------------ | ---------------------------------------------- |
| `200` | `{ "valid": true }` | Token 有效且答案正确 |
| `200` | `{ "valid": false }` | Token 有效但**答案错误**（Token 已消费） |
| `400` | `{ "valid": false, "error": "invalid json" }` | 请求体非合法 JSON |
| `400` | `{ "valid": false, "error": "token and answer required" }` | 缺少 `token` 或 `answer` 字段 |
| `400` | `{ "valid": false, "error": "invalid or expired token" }` | Token 无效、已消费或已过期 |
> ⚠️ **注意区分**：答案错误与 Token 错误是两种不同的结果。答案错误（Token 仍有效）返回 `200 {valid:false}`；而 Token 缺失/失效/过期会直接返回 `400`。消费方（如 gateway）应据此分别处理：前者提示"验证码错误"，后者提示"请重新获取验证码"。
**示例**：
```bash
curl -X POST http://127.0.0.1:8009/captcha/verify \
-H 'Content-Type: application/json' \
-d '{"token":"3f9a1c2b4d5e6f7a8b9c0d1e2f3a4b5c","answer":"AB3XY"}'
```json
{
"success": true,
"data": {
"token": "3f9a1c2b4d5e6f7a8b9c0d1e2f3a4b5c",
"image": "<base64 编码的 PNG，160×50>"
}
}
```
---
> Token 有效期 300 秒，且**一次性使用**——无论校验成功与否，第一次 `verify` 调用后即被消费。
### 3. 健康检查
### 2. verify — 校验验证码
```
GET /health
```
**响应**（HTTP 200）：
**请求**：
```json
{ "status": "ok", "service": "captcha" }
{"_route": "verify", "token": "3f9a1c2b4d5e6f7a8b9c0d1e2f3a4b5c", "answer": "AB3XY"}
```
---
**响应**（均为 `success: true`，业务结果在 `data` 中区分）：
| 响应 data | 触发条件 |
| ------------------------------------------------------------- | ------------------------------ |
| `{ "valid": true }` | Token 有效且答案正确 |
| `{ "valid": false }` | Token 有效但**答案错误**（Token 已消费） |
| `{ "valid": false, "error": "token and answer required" }` | 缺少 `token` 或 `answer` 字段 |
| `{ "valid": false, "error": "invalid or expired token" }` | Token 无效、已消费或已过期 |
> ⚠️ **注意区分**：答案错误与 Token 错误是两种不同的结果。消费方（如 gateway）应分别处理：前者提示"验证码错误"，后者提示"请重新获取验证码"。
### 3. health — 健康检查
**请求**：`{"_route": "health"}`
**响应**：`{"success": true, "data": {"status": "ok", "service": "captcha"}}`
### 调用示例（Python）
```python
import json, sys
sys.path.insert(0, "../pymsg")
from pymsg import MsgClient
c = MsgClient("127.0.0.1", 8009, "my-service")
resp = json.loads(c.call(json.dumps({"_route": "image"}), timeout=5))
token = resp["data"]["token"]
png_b64 = resp["data"]["image"]
```
## 数据模型与 Token 生命周期
- **存储**：进程内存字典 `captcha_store`，键为 Token，值为 `{ answer: <小写答案>, created_at: <Unix 时间戳> }`。
- **生成**：`GET /captcha/image` 时写入一条记录，并返回对应 Token。
- **消费**：`POST /captcha/verify` 时 `pop` 移除该 Token，无论答案对错，均不可再次使用。
- **过期**：超过 `CAPTCHA_TTL`（300 秒）即视为失效。`/captcha/image` 会在每次生成时顺带清理过期项；`/captcha/verify` 时若 Token 不在存储中（过期、未生成或已消费），统一返回 `400 invalid or expired token`。
- **生成**：`image` 路由时写入一条记录，并返回对应 Token 与 base64 PNG。
- **消费**：`verify` 路由时 `pop` 移除该 Token，无论答案对错，均不可再次使用。
- **过期**：超过 `CAPTCHA_TTL`（300 秒）即视为失效。`image` 路由每次生成时顺带清理过期项；`verify` 时若 Token 不在存储中（过期、未生成或已消费），统一返回 `invalid or expired token`。
## 集成关系
- 由 `core`（`PMC`）通过 `init.txt` 中的 `bash /opt/micro_servers/captcha/_run.sh` 拉起并守护。
- `gateway` 提供 **`GET /captcha/image`** 代理路由：经 pymsg 调本服务的 `image` 路由，将 base64 PNG 解码后以 `image/png` 响应体 + `X-Captcha-Token` 响应头返回给浏览器（`Cache-Control: no-store`）。
- `gateway` 在 `/login` 与 `/register` 中可选地启用 CAPTCHA 校验（以 `captcha_token` + `captcha_answer` 字段传入，见 `gateway/README.md`）。
- 在集中式日志链路中，`captcha` 作为 `service` 名之一，可经 `log_client` SDK 推送日志至 `db` 的 `/log/ingest`。
## 典型使用流程
1. 客户端调用 `GET /captcha/image`，保存返回的图片与响应头中的 `X-Captcha-Token`。
1. 前端请求 `GET https://<host>/captcha/image`（经 waf → gateway → captcha），展示图片并保存 `X-Captcha-Token`。
2. 用户根据图片内容输入验证码文本。
3. 客户端调用 `POST /captcha/verify`，提交 `token` 与用户输入的 `answer`。
4. 服务端比对答案（小写化、恒定时间比较），返回 `valid` 结果；Token 随即作废（无论对错）。
3. 前端在登录/注册请求中携带 `captcha_token` + `captcha_answer`。
4. 服务端比对答案（小写化、恒定时间比较），Token 随即作废（无论对错）。
## 项目结构
```
captcha/
├── captcha_service.py # 服务主程序（HTTP 服务 + 验证码生成与校验）
├── captcha_service.py # 服务主程序（pymsg 服务 + 验证码生成与校验）
├── _run.sh # 启动脚本（cd 后 exec python3 captcha_service.py）
└── README.md
```
@@ -194,6 +145,7 @@ captcha/
## 注意事项
- 验证码存储在进程内存中，服务重启后所有 Token 失效；**不支持多实例水平扩展**（如需分布式部署，应将 `captcha_store` 替换为 Redis 等外部存储，并相应调整过期清理逻辑）。
- 默认仅监听 `127.0.0.1`，仅供本机或内网调用；如需对外暴露，请配合反向代理（如 `gateway`/`waf`）并做好限流。
- 默认仅监听 `127.0.0.1`，仅供本机或内网调用；浏览器侧图片获取统一走 `gateway` 的 `GET /captcha/image` 代理。
- 答案校验不区分大小写；Token 一经校验即从内存中移除（一次性使用）。
- 字体缺失时自动回退到 PIL 默认字体，可能影响验证码观感，部署前建议确认 DejaVu 字体已安装。
- 旧版 HTTP 接口（http.server）已下线，协议与 db 服务统一为 pymsg/TCP。
captcha/captcha_service.py
+160 -92
View File
@@ -1,144 +1,212 @@
"""
CAPTCHA Image Service (stdlib only, no Flask dependency)
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 os
import base64
import io
import json
import time
import os
import random
import string
import secrets
import string
import threading
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
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 = 5
CAPTCHA_TTL = 300
ADDR = '127.0.0.1'
PORT = 8009
CAPTCHA_TTL = 300 # Token 有效期（秒）
CHARS = string.ascii_letters + string.digits
CHARS = CHARS.translate(str.maketrans('', '', 'lI1O0o'))
CHARS = CHARS.translate(str.maketrans('', '', 'lI1O0o')) # 剔除易混淆字符
captcha_store = {}
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.choice([(255,255,255), (240,240,240), (230,240,250)])
bg = random.choice([(255, 255, 255), (240, 240, 240), (230, 240, 250)])
img = Image.new('RGB', (width, height), bg)
draw = ImageDraw.Draw(img)
for i, ch in enumerate(text):
color = random.choice([(30,30,30),(50,80,150),(150,50,50),(50,130,50)])
color = random.choice([(30, 30, 30), (50, 80, 150), (150, 50, 50), (50, 130, 50)])
y = random.randint(2, 12)
x = 10 + i * 28 + random.randint(-3, 3)
try:
font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf",
random.choice([28,30,32,34]))
random.choice([28, 30, 32, 34]))
except (IOError, OSError):
font = ImageFont.load_default()
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.choice([(30,30,30),(50,80,150)]), width=1)
draw.line([(random.randint(0, width), random.randint(0, height)),
(random.randint(0, width), random.randint(0, height))],
fill=random.choice([(30, 30, 30), (50, 80, 150)]), width=1)
for _ in range(50):
draw.point((random.randint(0,width-1), random.randint(0,height-1)),
fill=random.choice([(30,30,30),(50,80,150)]))
draw.point((random.randint(0, width - 1), random.randint(0, height - 1)),
fill=random.choice([(30, 30, 30), (50, 80, 150)]))
buf = io.BytesIO()
img.save(buf, format='PNG')
buf.seek(0)
return buf
return buf.getvalue()
def json_response(handler, data, status=200):
body = json.dumps(data).encode('utf-8')
handler.send_response(status)
handler.send_header('Content-Type', 'application/json')
handler.send_header('Content-Length', str(len(body)))
handler.end_headers()
handler.wfile.write(body)
# ═══════════════════════════════════════════════════════════════════════
# CaptchaServer — 主服务类
# ═══════════════════════════════════════════════════════════════════════
class CaptchaServer:
"""基于 pymsg 的验证码服务，通过 TCP 长连接收发 JSON。"""
class CaptchaHandler(BaseHTTPRequestHandler):
def log_message(self, format, *args):
pass
def __init__(self, addr="127.0.0.1", port=8009):
self.addr = addr
self.port = port
self._server = MsgServer(addr, port)
def do_GET(self):
parsed = urlparse(self.path)
if parsed.path == '/captcha/image':
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]
# 路由表
self._routes = {
"image": self._handle_image,
"verify": self._handle_verify,
"health": self._handle_health,
}
img_buf = generate_image(text)
data = img_buf.read()
self.send_response(200)
self.send_header('Content-Type', 'image/png')
self.send_header('Content-Length', str(len(data)))
self.send_header('X-Captcha-Token', token)
self.end_headers()
self.wfile.write(data)
elif parsed.path == '/health':
json_response(self, {'status': 'ok', 'service': 'captcha'})
# ── 启动 / 停止 ──────────────────────────────────────────────────
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:
self.send_response(404)
self.end_headers()
diff = 0
for a, b in zip(answer, expected):
diff |= ord(a) ^ ord(b)
valid = diff == 0
def do_POST(self):
parsed = urlparse(self.path)
if parsed.path == '/captcha/verify':
length = int(self.headers.get('Content-Length', 0))
body = self.rfile.read(length)
try:
data = json.loads(body)
except json.JSONDecodeError:
json_response(self, {'valid': False, 'error': 'invalid json'}, 400)
return
return {"valid": valid}
token = data.get('token', '').strip()
answer = data.get('answer', '').strip().lower()
if not token or not answer:
json_response(self, {'valid': False, 'error': 'token and answer required'}, 400)
return
def _handle_health(self, _req):
return {"status": "ok", "service": "captcha"}
with STORE_LOCK:
entry = captcha_store.pop(token, None)
if not entry:
json_response(self, {'valid': False, 'error': 'invalid or expired token'}, 400)
return
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
json_response(self, {'valid': valid})
else:
self.send_response(404)
self.end_headers()
# ═══════════════════════════════════════════════════════════════════════
# 入口
# ═══════════════════════════════════════════════════════════════════════
if __name__ == '__main__':
server = HTTPServer((ADDR, PORT), CaptchaHandler)
print(f'CAPTCHA service listening at http://{ADDR}:{PORT}')
server.serve_forever()
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()