OrgTJWater/pipeline-lifetime
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: refine email verification login flow

Browse Source
This commit is contained in:
jiang
2026-08-04 18:24:05 +08:00
parent 732c1df031
commit 87c792db44
7 changed files with 323 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/routes.py
+92 -33
View File
@@ -103,8 +103,18 @@ def consume_fresh_authorization(purpose: str) -> None:
session.pop(f"fresh_auth_{purpose}", None)
def render_auth_template(mode: str, status_code: int = 200, captcha: str = ""):
return render_template("login.html", mode=mode, captcha=captcha), status_code
def render_auth_template(
mode: str,
status_code: int = 200,
captcha: str = "",
auth_error_field: str | None = None,
):
return render_template(
"login.html",
mode=mode,
captcha=captcha,
auth_error_field=auth_error_field,
), status_code
def refresh_captcha() -> str:
@@ -112,9 +122,14 @@ def refresh_captcha() -> str:
return session["captcha"]
def render_auth_error(mode: str, message: str, status_code: int = 400):
def render_auth_error(
mode: str,
message: str,
status_code: int = 400,
auth_error_field: str | None = None,
):
flash(message, "error")
return render_auth_template(mode, status_code, refresh_captcha())
return render_auth_template(mode, status_code, refresh_captcha(), auth_error_field)
def captcha_is_valid() -> bool:
@@ -192,12 +207,17 @@ def code_hash(email: str, purpose: str, code: str) -> str:
return hashlib.sha256(f"{email}:{purpose}:{code}".encode()).hexdigest()
def issue_email_code(email: str, purpose: str) -> bool:
def issue_email_code(email: str, purpose: str) -> str | None:
"""Issue an email code, returning a failure reason when it cannot be sent."""
now = utc_now()
resend_at = now - timedelta(seconds=current_app.config["EMAIL_CODE_RESEND_SECONDS"])
recent = (
EmailVerificationCode.query.filter_by(email=email, purpose=purpose)
.filter(EmailVerificationCode.created_at >= resend_at)
.filter(
EmailVerificationCode.created_at >= resend_at,
EmailVerificationCode.expires_at > now,
EmailVerificationCode.used_at.is_(None),
)
.first()
)
ip = request.remote_addr or ""
@@ -205,8 +225,10 @@ def issue_email_code(email: str, purpose: str) -> bool:
EmailVerificationCode.created_at >= now - timedelta(hours=1),
EmailVerificationCode.requested_ip == ip,
).count()
if recent or hourly >= 20:
return False
if recent:
return "resend_wait"
if hourly >= 20:
return "rate_limited"
EmailVerificationCode.query.filter_by(email=email, purpose=purpose, used_at=None).update({"used_at": now})
code = f"{secrets.randbelow(1_000_000):06d}"
record = EmailVerificationCode(
@@ -221,7 +243,7 @@ def issue_email_code(email: str, purpose: str) -> bool:
try:
send_transactional_email(
to=email,
subject=f"供水管道健康评估系统：{EMAIL_CODE_LABELS.get(purpose, '身份验证')}验证码",
subject=f"【管道健康】{EMAIL_CODE_LABELS.get(purpose, '身份验证')}验证码",
html=verification_code_email(
code=code,
minutes=current_app.config["EMAIL_CODE_MINUTES"],
@@ -231,8 +253,16 @@ def issue_email_code(email: str, purpose: str) -> bool:
except (EmailConfigurationError, EmailDeliveryError):
db.session.delete(record)
db.session.commit()
return False
return True
return "delivery_failed"
return None
def email_code_issue_message(reason: str) -> str:
if reason == "resend_wait":
return f"验证码已发送，请 {current_app.config['EMAIL_CODE_RESEND_SECONDS']} 秒后再试。"
if reason == "rate_limited":
return "请求过于频繁，请稍后再试。"
return "验证码发送失败，请稍后重试。"
def consume_email_code(email: str, purpose: str, code: str) -> bool:
@@ -318,17 +348,26 @@ def login():
return redirect(url_for("main.home"))
if request.method == "GET":
return render_auth_template("login", captcha=refresh_captcha())
email, password = normal_email(request.form.get("email", "")), request.form.get("password", "")
if not captcha_is_valid(): return render_auth_error("login", "验证码错误")
user = User.query.filter_by(email=email).first()
identifier, password = request.form.get("identifier", "").strip(), request.form.get("password", "")
if not captcha_is_valid():
return render_auth_error("login", "图形验证码错误", auth_error_field="captcha")
user = User.query.filter(
or_(User.username == identifier, User.email == normal_email(identifier))
).first()
if not user or not user.is_active_account or not user.check_password(password):
return render_auth_error("login", "邮箱或密码错误")
return render_auth_error("login", "用户名、邮箱或密码错误")
remember = bool(request.form.get("remember"))
if trusted_device_for(user):
return login_response(user, remember)
session.update(pending_email=email, pending_user_id=user.id, pending_remember=remember, pending_purpose="login")
if not issue_email_code(email, "login"):
return render_auth_error("login", "验证码发送失败，请稍后重试", 503)
session.update(pending_email=user.email, pending_user_id=user.id, pending_remember=remember, pending_purpose="login")
if reason := issue_email_code(user.email, "login"):
if reason == "resend_wait":
flash("邮箱验证码已发送，请输入验证码完成二次认证。", "info")
return redirect(url_for("main.verify_email", purpose="login"))
return render_auth_error(
"login", email_code_issue_message(reason),
503 if reason == "delivery_failed" else 429,
)
return redirect(url_for("main.verify_email", purpose="login"))
@@ -336,16 +375,24 @@ def login():
def register():
if request.method == "GET": return render_auth_template("register", captcha=refresh_captcha())
username, email, password = request.form.get("username", "").strip(), normal_email(request.form.get("email", "")), request.form.get("password", "")
if not captcha_is_valid(): return render_auth_error("register", "验证码错误")
if not captcha_is_valid():
return render_auth_error("register", "图形验证码错误", auth_error_field="captcha")
if not registration_allowed(): return render_auth_error("register", "当前未开放自助注册，请联系管理员。", 403)
if not username or len(username) > 100: return render_auth_error("register", "显示名不能为空且不能超过100个字符")
if not valid_email(email): return render_auth_error("register", "请输入有效的邮箱地址")
if not valid_password(password): return render_auth_error("register", PASSWORD_RULE_MESSAGE)
if not username or len(username) > 100:
return render_auth_error("register", "显示名不能为空且不能超过100个字符", auth_error_field="username")
if not valid_email(email):
return render_auth_error("register", "请输入有效的邮箱地址", auth_error_field="email")
if not valid_password(password):
return render_auth_error("register", PASSWORD_RULE_MESSAGE, auth_error_field="password")
if User.query.filter((User.username == username) | (User.email == email)).first(): return render_auth_error("register", "显示名或邮箱已被使用")
user = User(username=username, email=email, is_admin=False, is_active_account=False)
user.set_password(password); db.session.add(user); db.session.commit()
session.update(pending_email=email, pending_user_id=user.id, pending_purpose="register")
if not issue_email_code(email, "register"): return render_auth_error("register", "验证码发送失败，请稍后重试", 503)
if reason := issue_email_code(email, "register"):
return render_auth_error(
"register", email_code_issue_message(reason),
503 if reason == "delivery_failed" else 429,
)
return redirect(url_for("main.verify_email", purpose="register"))
@@ -353,24 +400,33 @@ def register():
def verify_email(purpose: str):
if purpose not in EMAIL_CODE_PURPOSES or session.get("pending_purpose") != purpose:
abort(400)
if request.method == "GET": return verification_page("邮箱验证", purpose)
verification_title = "邮箱二次认证" if purpose == "login" else "邮箱验证"
if request.method == "GET": return verification_page(verification_title, purpose)
email = session.get("pending_email", "")
if not consume_email_code(email, purpose, request.form.get("code", "")):
flash("验证码无效、过期或尝试次数已用尽。", "error"); return verification_page("邮箱验证", purpose), 400
flash("邮箱验证码无效、过期或尝试次数已用尽。", "error"); return verification_page(verification_title, purpose), 400
user = db.session.get(User, session.get("pending_user_id"))
if not user or (purpose != "change_email_new" and user.email != email): abort(400)
if purpose == "register":
user.email_verified_at = utc_now(); user.is_active_account = True; db.session.commit(); session.clear(); flash("邮箱验证成功，请登录。", "info"); return redirect(url_for("main.login"))
user.email_verified_at = utc_now()
user.is_active_account = True
db.session.commit()
session.clear()
flash("邮箱验证成功，欢迎使用系统。", "info")
return login_response(user, remember=False)
if purpose == "login":
remember = bool(session.get("pending_remember")); session.clear(); return login_response(user, remember, trust_device=True)
remember = bool(session.get("pending_remember"))
trust_device = bool(request.form.get("trust_device"))
session.clear()
return login_response(user, remember, trust_device=trust_device)
if purpose == "reset":
grant_fresh_authorization(user, "password_reset")
return redirect(url_for("main.set_password"))
if purpose == "change_email_old":
new_email = session.get("new_email", "")
session.update(pending_email=new_email, pending_purpose="change_email_new")
if not issue_email_code(new_email, "change_email_new"):
flash("新邮箱验证码发送失败，请稍后重试。", "error"); return redirect(url_for("main.account_security"))
if reason := issue_email_code(new_email, "change_email_new"):
flash(email_code_issue_message(reason), "error"); return redirect(url_for("main.account_security"))
return redirect(url_for("main.verify_email", purpose="change_email_new"))
if purpose == "change_email_new":
if User.query.filter(User.email == email, User.id != user.id).first():
@@ -384,7 +440,8 @@ def verify_email(purpose: str):
def resend_code(purpose: str):
if purpose not in EMAIL_CODE_PURPOSES or session.get("pending_purpose") != purpose:
abort(400)
if not issue_email_code(session.get("pending_email", ""), purpose): flash("发送过于频繁或服务暂不可用，请稍后再试。", "error")
if reason := issue_email_code(session.get("pending_email", ""), purpose):
flash(email_code_issue_message(reason), "error")
else: flash("验证码已发送，请查收邮箱。", "info")
return redirect(url_for("main.verify_email", purpose=purpose))
@@ -437,10 +494,12 @@ def account_security():
if not valid_email(new_email) or User.query.filter_by(email=new_email).first():
flash("请输入未被使用的有效邮箱地址。", "error"); return redirect(url_for("main.account_security"))
session.update(pending_email=current_user.email, pending_user_id=current_user.id, pending_purpose="change_email_old", new_email=new_email)
if not issue_email_code(current_user.email, "change_email_old"): flash("验证码发送失败，请稍后重试。", "error"); return redirect(url_for("main.account_security"))
if reason := issue_email_code(current_user.email, "change_email_old"):
flash(email_code_issue_message(reason), "error"); return redirect(url_for("main.account_security"))
return redirect(url_for("main.verify_email", purpose="change_email_old"))
session.update(pending_email=current_user.email, pending_user_id=current_user.id, pending_purpose="change_password")
if not issue_email_code(current_user.email, "change_password"): flash("验证码发送失败，请稍后重试。", "error"); return redirect(url_for("main.account_security"))
if reason := issue_email_code(current_user.email, "change_password"):
flash(email_code_issue_message(reason), "error"); return redirect(url_for("main.account_security"))
return redirect(url_for("main.verify_email", purpose="change_password"))
@@ -514,7 +573,7 @@ def admin_password_reset(user_id: int):
try:
send_transactional_email(
to=user.email,
subject="供水管道健康评估系统：请重置密码",
subject="【管道健康】密码重置通知",
html=password_reset_notice_email(
username=user.username,
reset_url=url_for("main.forgot_password", _external=True),