wangchuanli
3751dffef9
- 新增备份管理页与 API：在线快照、自动周期备份、按份数清理、下载、一键恢复（恢复前自动兜底） - 新增 /profile/export，普通用户可导出本人全部数据（不含 Cookie 明文） - 修复 X-Forwarded-For 可伪造导致三道 IP 防线失效，统一走 client_ip() 取客户端地址 - 取消 admin123 硬编码默认口令，留空则生成随机初始口令并仅打印一次 - .dockerignore 排除 backups/ 并加构建期断言，防止密钥随镜像分发 - 新增会话版本号，改密/停用/删除及恢复备份后其他会话立即失效 - 新增容器资源上限、采集跨度硬顶 31 天、重操作最小间隔与并发 409 - 新增访问日志、HSTS 条件下发、口令黑名单、验证码抗模板匹配、instance.json 0600 - 版本号升至 1.4.0，同步更新 README、SECURITY、.env.example 与 compose 配置
318 行
14 KiB
Python
318 行
14 KiB
Python
# -*- coding: utf-8 -*-
|
||
# SPDX-License-Identifier: MIT
|
||
# Copyright (c) 2026 Wang Chuanli
|
||
|
||
"""图形验证码（自托管，零第三方依赖）。
|
||
|
||
设计要点
|
||
--------
|
||
1. **答案只存在服务端**。下发到浏览器的是一个随机 `captcha_id`，它本身
|
||
不含任何信息。之所以不把答案放进 Flask session：Flask 的 session 是
|
||
**签名而非加密**的（base64 + HMAC），客户端 base64 解开就能读到明文——
|
||
把答案放进去等于把答案直接送给机器人。
|
||
2. **一次性**。校验时无论成功失败都立刻删除该 `captcha_id`，
|
||
防止「一个码刷一万个用户名」的撞库变体。
|
||
3. **光栅图，不是 SVG**。SVG 是文本，答案会以明文出现在页面源码或 DOM 里，
|
||
必须用位图。这里手写 PNG 编码器（zlib 是标准库），点阵字模自带。
|
||
4. **字符集避开易混字符**（0/O、1/I/L），降低正常用户输错概率。
|
||
|
||
字模
|
||
----
|
||
5×7 点阵，`#` 为前景。渲染时**逐字符**随机放大、旋转、切变、加波浪偏移，
|
||
笔画用粗刷子画（旋转后不会出现断点），再叠背景纹理、噪点与压线干扰。
|
||
|
||
强度说明（为什么要这么多花样）
|
||
------------------------------
|
||
字模是固定的，而且就在这份源码里 —— 也就是说攻击者**知道**每个字符长什么样。
|
||
在这种情况下，提高自动识别成本的唯一手段就是让「同一字符的两次渲染」在像素上
|
||
尽量不同：角度、切变、缩放、波形相位、笔画粗细、颜色、干扰线位置全部随机。
|
||
纯模板匹配在这种变形下会失效，必须上带形变增强的模型，成本高一个数量级。
|
||
反过来说，也**不要**指望它能挡住有充足算力、专门针对本站训练的对手 ——
|
||
验证码是「提高成本」而不是「杜绝」。
|
||
"""
|
||
import hmac
|
||
import math
|
||
import random
|
||
import secrets
|
||
import struct
|
||
import zlib
|
||
from datetime import datetime, timedelta
|
||
|
||
ALPHABET = "23456789ABCDEFGHJKMNPQRSTUVWXYZ" # 去掉 0 O 1 I L
|
||
|
||
_FONT = {
|
||
"2": (".###.", "#...#", "....#", "...#.", "..#..", ".#...", "#####"),
|
||
"3": ("####.", "....#", "....#", ".###.", "....#", "....#", "####."),
|
||
"4": ("...#.", "..##.", ".#.#.", "#..#.", "#####", "...#.", "...#."),
|
||
"5": ("#####", "#....", "####.", "....#", "....#", "#...#", ".###."),
|
||
"6": ("..##.", ".#...", "#....", "####.", "#...#", "#...#", ".###."),
|
||
"7": ("#####", "....#", "...#.", "..#..", ".#...", ".#...", ".#..."),
|
||
"8": (".###.", "#...#", "#...#", ".###.", "#...#", "#...#", ".###."),
|
||
"9": (".###.", "#...#", "#...#", ".####", "....#", "...#.", ".##.."),
|
||
"A": ("..#..", ".#.#.", "#...#", "#...#", "#####", "#...#", "#...#"),
|
||
"B": ("####.", "#...#", "#...#", "####.", "#...#", "#...#", "####."),
|
||
"C": (".###.", "#...#", "#....", "#....", "#....", "#...#", ".###."),
|
||
"D": ("###..", "#..#.", "#...#", "#...#", "#...#", "#..#.", "###.."),
|
||
"E": ("#####", "#....", "#....", "####.", "#....", "#....", "#####"),
|
||
"F": ("#####", "#....", "#....", "####.", "#....", "#....", "#...."),
|
||
"G": (".###.", "#...#", "#....", "#.###", "#...#", "#...#", ".###."),
|
||
"H": ("#...#", "#...#", "#...#", "#####", "#...#", "#...#", "#...#"),
|
||
"J": ("..###", "...#.", "...#.", "...#.", "...#.", "#..#.", ".##.."),
|
||
"K": ("#...#", "#..#.", "#.#..", "##...", "#.#..", "#..#.", "#...#"),
|
||
"M": ("#...#", "##.##", "#.#.#", "#...#", "#...#", "#...#", "#...#"),
|
||
"N": ("#...#", "##..#", "#.#.#", "#..##", "#...#", "#...#", "#...#"),
|
||
"P": ("####.", "#...#", "#...#", "####.", "#....", "#....", "#...."),
|
||
"Q": (".###.", "#...#", "#...#", "#...#", "#.#.#", "#..#.", ".##.#"),
|
||
"R": ("####.", "#...#", "#...#", "####.", "#.#..", "#..#.", "#...#"),
|
||
"S": (".####", "#....", "#....", ".###.", "....#", "....#", "####."),
|
||
"T": ("#####", "..#..", "..#..", "..#..", "..#..", "..#..", "..#.."),
|
||
"U": ("#...#", "#...#", "#...#", "#...#", "#...#", "#...#", ".###."),
|
||
"V": ("#...#", "#...#", "#...#", "#...#", "#...#", ".#.#.", "..#.."),
|
||
"W": ("#...#", "#...#", "#...#", "#...#", "#.#.#", "##.##", "#...#"),
|
||
"X": ("#...#", "#...#", ".#.#.", "..#..", ".#.#.", "#...#", "#...#"),
|
||
"Y": ("#...#", "#...#", ".#.#.", "..#..", "..#..", "..#..", "..#.."),
|
||
"Z": ("#####", "....#", "...#.", "..#..", ".#...", "#....", "#####"),
|
||
}
|
||
|
||
GLYPH_W, GLYPH_H = 5, 7
|
||
|
||
# 一次性验证码有效期（秒）。太短用户来不及看，太长给暴力破解留窗口。
|
||
TTL_SECONDS = 300
|
||
# 保留已过期记录多久后清理（仅用于体积控制，不影响安全性）
|
||
PURGE_AFTER_SECONDS = 3600
|
||
|
||
|
||
def random_code(length=4):
|
||
return "".join(secrets.choice(ALPHABET) for _ in range(length))
|
||
|
||
|
||
# ---------------- PNG 编码（手写，无依赖） ----------------
|
||
def _chunk(tag, data):
|
||
return (struct.pack(">I", len(data)) + tag + data
|
||
+ struct.pack(">I", zlib.crc32(tag + data) & 0xFFFFFFFF))
|
||
|
||
|
||
def encode_png(width, height, rgb):
|
||
"""把 RGB 字节串编码成 PNG（8 位真彩，无 alpha）。
|
||
|
||
rgb 长度必须是 width*height*3。每行前面加一个 filter 字节 0（None），
|
||
这是 PNG 对「一行一张扫描线」的强制要求。
|
||
"""
|
||
stride = width * 3
|
||
raw = bytearray()
|
||
for y in range(height):
|
||
raw.append(0)
|
||
raw += rgb[y * stride:(y + 1) * stride]
|
||
ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0)
|
||
return (b"\x89PNG\r\n\x1a\n"
|
||
+ _chunk(b"IHDR", ihdr)
|
||
+ _chunk(b"IDAT", zlib.compress(bytes(raw), 9))
|
||
+ _chunk(b"IEND", b""))
|
||
|
||
|
||
class _Canvas:
|
||
"""极小的 RGB 画布。坐标越界自动丢弃，省得每处调用都判边界。"""
|
||
|
||
def __init__(self, w, h, bg):
|
||
self.w, self.h = w, h
|
||
self.buf = bytearray(bg * (w * h))
|
||
|
||
def dot(self, x, y, color):
|
||
if 0 <= x < self.w and 0 <= y < self.h:
|
||
i = (y * self.w + x) * 3
|
||
self.buf[i:i + 3] = bytes(color)
|
||
|
||
def rect(self, x, y, w, h, color):
|
||
for dy in range(h):
|
||
for dx in range(w):
|
||
self.dot(x + dx, y + dy, color)
|
||
|
||
def line(self, x0, y0, x1, y1, color):
|
||
"""Bresenham 直线。"""
|
||
dx, dy = abs(x1 - x0), -abs(y1 - y0)
|
||
sx = 1 if x0 < x1 else -1
|
||
sy = 1 if y0 < y1 else -1
|
||
err = dx + dy
|
||
while True:
|
||
self.dot(x0, y0, color)
|
||
if x0 == x1 and y0 == y1:
|
||
return
|
||
e2 = 2 * err
|
||
if e2 >= dy:
|
||
err += dy
|
||
x0 += sx
|
||
if e2 <= dx:
|
||
err += dx
|
||
y0 += sy
|
||
|
||
def bytes(self):
|
||
return bytes(self.buf)
|
||
|
||
|
||
def _brush_line(cv, x0, y0, x1, y1, color, r):
|
||
"""用 r×r 方刷画一条线。
|
||
|
||
旋转后的笔画如果只用点阵格逐个平移，会出现锯齿状断点 —— 一圈一圈的
|
||
缝隙正好给「连通域分析」留了把手。这里改成沿线段走样并盖方刷，
|
||
笔画连续，旋转也不散架。
|
||
"""
|
||
steps = int(max(abs(x1 - x0), abs(y1 - y0))) + 1
|
||
o = r // 2
|
||
for i in range(steps + 1):
|
||
t = i / float(steps)
|
||
x = int(round(x0 + (x1 - x0) * t))
|
||
y = int(round(y0 + (y1 - y0) * t))
|
||
cv.rect(x - o, y - o, r, r, color)
|
||
|
||
|
||
def _draw_char(cv, glyph, cx, cy, scale, color, rng):
|
||
"""在 (cx, cy) 为中心画一个字符：随机旋转 + 切变 + 波浪 + 粗笔画。
|
||
|
||
三段变换按「点阵坐标 -> 缩放居中 -> 切变 -> 旋转 -> 波浪纵向偏移」依次施加。
|
||
顺序不能乱：先切变再旋转，得到的才是「斜着写的手写体」而不是「被斜切的旋转体」。
|
||
"""
|
||
ang = rng.uniform(-0.38, 0.38) # 弧度，约 ±22°
|
||
cos_a, sin_a = math.cos(ang), math.sin(ang)
|
||
shear = rng.uniform(-0.32, 0.32)
|
||
amp = rng.uniform(0.0, 2.6) # 波浪振幅（像素）
|
||
period = rng.uniform(18.0, 42.0)
|
||
phase = rng.uniform(0.0, 6.283)
|
||
half_w = GLYPH_W * scale / 2.0
|
||
half_h = GLYPH_H * scale / 2.0
|
||
r = max(2, scale)
|
||
|
||
def place(col, row):
|
||
px = (col + 0.5) * scale - half_w
|
||
py = (row + 0.5) * scale - half_h
|
||
px += shear * py
|
||
x = cx + px * cos_a - py * sin_a
|
||
y = cy + px * sin_a + py * cos_a
|
||
return x, y + amp * math.sin(x / period + phase)
|
||
|
||
for row, bits in enumerate(glyph):
|
||
col = 0
|
||
while col < len(bits):
|
||
if bits[col] != "#":
|
||
col += 1
|
||
continue
|
||
start = col
|
||
while col + 1 < len(bits) and bits[col + 1] == "#":
|
||
col += 1 # 连续的一段合起来画，笔画才连得上
|
||
x0, y0 = place(start, row)
|
||
x1, y1 = place(col, row)
|
||
_brush_line(cv, x0, y0, x1, y1, color, r)
|
||
col += 1
|
||
|
||
|
||
def render(code, width=150, height=56, scale=5, rng=None):
|
||
"""把验证码渲染成 PNG 字节串。
|
||
|
||
字体大小、角度、切变、波浪、颜色、干扰线全部逐次随机 ——
|
||
目标不是「好看」，而是让同一串字符的两次渲染在像素上尽量不同，
|
||
从而让「预存字模 + 模板匹配」这条最便宜的攻击路线失效。
|
||
"""
|
||
rng = rng or random.SystemRandom()
|
||
n = len(code)
|
||
gap = 9
|
||
# 宽度按最大可能字号算，且左右各留够旋转半径 ——
|
||
# 旋转后的字符会往两侧探出约半个字高，留窄了最外侧那个字会被裁掉一截，
|
||
# 而「被裁掉一角的字符」会直接变成一次没道理的输错（体验问题，不是安全问题）。
|
||
text_w = n * GLYPH_W * (scale + 1) + (n - 1) * gap
|
||
need_w = text_w + int(GLYPH_H * (scale + 1) * 0.9) + 8
|
||
if need_w > width:
|
||
width = need_w
|
||
# 高度同理：旋转后的字符比原始点阵高不少，切了顶就等于少一个笔画特征
|
||
need_h = int(GLYPH_H * (scale + 1) * 1.7) + 8
|
||
if need_h > height:
|
||
height = need_h
|
||
x0 = max(5, (width - text_w) // 2)
|
||
y0 = height // 2
|
||
|
||
# 背景不做纯色：纯色底可以用一个阈值把前景整片切出来。
|
||
# 用「两色之间做斜向渐变」能让全局二值化的效果明显变差。
|
||
c1 = tuple(rng.randint(236, 252) for _ in range(3))
|
||
c2 = tuple(rng.randint(214, 240) for _ in range(3))
|
||
slant = rng.uniform(-1.0, 1.0)
|
||
cv = _Canvas(width, height, c1)
|
||
for y in range(height):
|
||
for x in range(width):
|
||
t = (x / float(width - 1 or 1)) * 0.6 + (y / float(height - 1 or 1)) * 0.4
|
||
t = min(1.0, max(0.0, t + slant * 0.15))
|
||
cv.dot(x, y, tuple(int(c1[i] + (c2[i] - c1[i]) * t) for i in range(3)))
|
||
|
||
# 1) 底层干扰线（先画，压在字下面）
|
||
for _ in range(3):
|
||
cv.line(rng.randint(0, width - 1), rng.randint(0, height - 1),
|
||
rng.randint(0, width - 1), rng.randint(0, height - 1),
|
||
tuple(rng.randint(170, 215) for _ in range(3)))
|
||
|
||
# 2) 字符本体
|
||
step = GLYPH_W * (scale + 1) + gap
|
||
for i, ch in enumerate(code):
|
||
glyph = _FONT.get(ch)
|
||
if glyph is None:
|
||
continue
|
||
# 逐字符字号抖动：字符宽度不再一致，按列投影切分就失效了
|
||
s = max(3, scale + rng.choice((-1, 0, 0, 1)))
|
||
cx = x0 + i * step + GLYPH_W * (scale + 1) / 2.0 + rng.uniform(-3.0, 3.0)
|
||
cy = y0 + rng.uniform(-3.0, 3.0)
|
||
color = tuple(rng.randint(15, 95) for _ in range(3))
|
||
_draw_char(cv, glyph, cx, cy, s, color, rng)
|
||
|
||
# 3) 前景噪点：破坏「按连通域找字符」的假设
|
||
for _ in range(70):
|
||
cv.dot(rng.randint(0, width - 1), rng.randint(0, height - 1),
|
||
tuple(rng.randint(90, 195) for _ in range(3)))
|
||
|
||
# 4) 压在字上的干扰线：最有效的反 OCR 手段，但太密人也认不出，
|
||
# 所以刻意控制成 2~3 条细线。
|
||
for _ in range(rng.randint(2, 3)):
|
||
y = rng.randint(2, height - 3)
|
||
cv.line(0, y, width - 1, y + rng.randint(-11, 11),
|
||
tuple(rng.randint(120, 175) for _ in range(3)))
|
||
|
||
return encode_png(width, height, cv.bytes())
|
||
|
||
|
||
# ---------------- 挑战的存储与校验（SQLite） ----------------
|
||
def _fmt(dt):
|
||
return dt.strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
|
||
def purge(conn, now=None):
|
||
"""清掉过期的挑战。每次新建时顺手调用（有 expires_at 索引，代价很小）。"""
|
||
now = now or datetime.now()
|
||
cut = _fmt(now - timedelta(seconds=PURGE_AFTER_SECONDS))
|
||
conn.execute("DELETE FROM captchas WHERE expires_at < ?", (cut,))
|
||
|
||
|
||
def create(conn, purpose, length=4, ttl=TTL_SECONDS, now=None):
|
||
"""新建一个挑战，返回 (captcha_id, code)。code 只应交给渲染函数，不要下发。"""
|
||
now = now or datetime.now()
|
||
code = random_code(length)
|
||
cid = secrets.token_urlsafe(24)
|
||
purge(conn, now)
|
||
conn.execute(
|
||
"INSERT INTO captchas(id,answer,purpose,created_at,expires_at) VALUES(?,?,?,?,?)",
|
||
(cid, code, purpose, _fmt(now), _fmt(now + timedelta(seconds=ttl))))
|
||
return cid, code
|
||
|
||
|
||
def verify(conn, captcha_id, answer, purpose, now=None):
|
||
"""校验并**立即作废**该挑战。返回 True/False。
|
||
|
||
永远不区分「过期」「不存在」「答案错」——对外只回一句人话，
|
||
避免把「这个 id 存在但答错了」这类信息透露给攻击者。
|
||
"""
|
||
if not captcha_id or answer is None:
|
||
return False
|
||
row = conn.execute("SELECT * FROM captchas WHERE id=?", (captcha_id,)).fetchone()
|
||
# 先删后判：无论结果如何都不允许第二次使用同一个 id
|
||
conn.execute("DELETE FROM captchas WHERE id=?", (captcha_id,))
|
||
if row is None or row["purpose"] != purpose:
|
||
return False
|
||
now = now or datetime.now()
|
||
if row["expires_at"] < _fmt(now):
|
||
return False
|
||
return hmac.compare_digest(str(row["answer"]), str(answer).strip().upper())
|