{escape(username)}，管理员已要求你重置密码。
-请通过下方按钮进入找回密码流程，系统会向本邮箱发送一次性验证码。
+{escape(username)}，请通过下方按钮设置新密码。
+该链接仅可使用一次，并将在 {minutes} 分钟后失效。
""" - return render_transactional_email(title="请重置密码", content=content) + return render_transactional_email(title="重置账户密码", content=content) + + +def registration_invitation_email(*, invitation_url: str, minutes: int) -> str: + """Render a one-time invitation link email for an administrator-created account.""" + safe_url = escape(invitation_url, quote=True) + content = f""" +管理员邀请你加入供水管道健康评估系统。
+请通过下方按钮设置显示名和密码。该链接仅可使用一次，并将在 {minutes} 分钟后失效。
+ """ + return render_transactional_email(title="管理员邀请你注册", content=content) def send_transactional_email(*, to: str, subject: str, html: str) -> dict[str, Any]: diff --git a/app/migrations.py b/app/migrations.py index 3102fc1..dd51fa4 100644 --- a/app/migrations.py +++ b/app/migrations.py @@ -6,17 +6,22 @@ from .extensions import db def upgrade_schema() -> None: """Create new tables and add authentication columns without losing existing data.""" - db.create_all() - inspector = inspect(db.engine) - columns = {item["name"] for item in inspector.get_columns("users")} - additions = { - "email": "VARCHAR(254)", - "email_verified_at": "DATETIME", - "is_active_account": "BOOLEAN NOT NULL DEFAULT 0", - "auth_version": "INTEGER NOT NULL DEFAULT 1", - } - for name, sql_type in additions.items(): - if name not in columns: - db.session.execute(text(f"ALTER TABLE users ADD COLUMN {name} {sql_type}")) - db.session.execute(text("CREATE UNIQUE INDEX IF NOT EXISTS ix_users_email_unique ON users (email) WHERE email IS NOT NULL")) - db.session.commit() + with db.engine.begin() as connection: + # Gunicorn workers can import the application simultaneously. Acquiring + # SQLite's write lock before inspecting or creating tables serializes the + # first-start migration and prevents duplicate CREATE TABLE attempts. + if db.engine.dialect.name == "sqlite": + connection.exec_driver_sql("BEGIN IMMEDIATE") + db.metadata.create_all(bind=connection) + inspector = inspect(connection) + columns = {item["name"] for item in inspector.get_columns("users")} + additions = { + "email": "VARCHAR(254)", + "email_verified_at": "DATETIME", + "is_active_account": "BOOLEAN NOT NULL DEFAULT 0", + "auth_version": "INTEGER NOT NULL DEFAULT 1", + } + for name, sql_type in additions.items(): + if name not in columns: + connection.execute(text(f"ALTER TABLE users ADD COLUMN {name} {sql_type}")) + connection.execute(text("CREATE UNIQUE INDEX IF NOT EXISTS ix_users_email_unique ON users (email) WHERE email IS NOT NULL")) diff --git a/app/models.py b/app/models.py index 4bf40e5..ac22946 100644 --- a/app/models.py +++ b/app/models.py @@ -53,6 +53,20 @@ class PasswordResetToken(db.Model): created_by = db.relationship("User", foreign_keys=[created_by_id]) +class RegistrationInvitation(db.Model): + __tablename__ = "registration_invitations" + + id = db.Column(db.Integer, primary_key=True) + email = db.Column(db.String(254), nullable=False, index=True) + token_hash = db.Column(db.String(64), unique=True, nullable=False, index=True) + expires_at = db.Column(db.DateTime, nullable=False, index=True) + used_at = db.Column(db.DateTime) + created_by_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False) + created_at = db.Column(db.DateTime, default=utc_now, nullable=False) + + created_by = db.relationship("User", foreign_keys=[created_by_id]) + + class EmailVerificationCode(db.Model): __tablename__ = "email_verification_codes" diff --git a/app/routes.py b/app/routes.py index 10d2fcd..32a5352 100644 --- a/app/routes.py +++ b/app/routes.py @@ -16,11 +16,20 @@ from .email import ( EmailConfigurationError, EmailDeliveryError, password_reset_notice_email, + registration_invitation_email, send_transactional_email, verification_code_email, ) from .extensions import db -from .models import AppSetting, EmailVerificationCode, TrustedDevice, UploadRecord, User +from .models import ( + AppSetting, + EmailVerificationCode, + PasswordResetToken, + RegistrationInvitation, + TrustedDevice, + UploadRecord, + User, +) from .prediction import PredictionError, run_prediction from .security import new_captcha from .time_utils import format_datetime_for_timezone, utc_now @@ -35,7 +44,6 @@ EMAIL_RE = re.compile(r"^[^\s@]+@[^\s@]+\.[^\s@]+$") EMAIL_CODE_PURPOSES = { "register", "login", - "reset", "change_password", "change_email_old", "change_email_new", @@ -43,7 +51,6 @@ EMAIL_CODE_PURPOSES = { EMAIL_CODE_LABELS = { "register": "完成注册", "login": "登录确认", - "reset": "重置密码", "change_password": "修改密码", "change_email_old": "确认原邮箱", "change_email_new": "确认新邮箱", @@ -59,6 +66,98 @@ def valid_email(value: str) -> bool: return len(value) <= 254 and bool(EMAIL_RE.fullmatch(value)) +def masked_email(value: str) -> str: + """Return an email suitable for confirmation screens without exposing it.""" + local_part, separator, domain = normal_email(value).partition("@") + if not separator or not local_part or not domain: + return "已绑定邮箱" + if len(local_part) == 1: + visible_local = "*" + elif len(local_part) == 2: + visible_local = f"{local_part[0]}*" + elif len(local_part) <= 4: + visible_local = f"{local_part[0]}**{local_part[-1]}" + else: + visible_local = f"{local_part[:2]}****{local_part[-2:]}" + return f"{visible_local}@{domain}" + + +def token_hash(token: str) -> str: + return hashlib.sha256(token.encode()).hexdigest() + + +def issue_password_reset_token(user: User, created_by: User) -> str: + """Invalidate outstanding links for a user and create a fresh one-time reset link.""" + now = utc_now() + PasswordResetToken.query.filter_by(user_id=user.id, used_at=None).update( + {"expires_at": now}, synchronize_session=False + ) + token = secrets.token_urlsafe(32) + db.session.add( + PasswordResetToken( + user_id=user.id, + created_by_id=created_by.id, + token_hash=token_hash(token), + expires_at=now + timedelta(minutes=current_app.config["PASSWORD_RESET_TOKEN_MINUTES"]), + ) + ) + db.session.commit() + return token + + +def active_password_reset_token(token: str) -> PasswordResetToken | None: + return PasswordResetToken.query.filter( + PasswordResetToken.token_hash == token_hash(token), + PasswordResetToken.used_at.is_(None), + PasswordResetToken.expires_at > utc_now(), + ).first() + + +def issue_registration_invitation(email: str, created_by: User) -> str: + """Create a fresh invitation and invalidate any previous unused invitation.""" + now = utc_now() + RegistrationInvitation.query.filter_by(email=email, used_at=None).update( + {"expires_at": now}, synchronize_session=False + ) + token = secrets.token_urlsafe(32) + db.session.add( + RegistrationInvitation( + email=email, + created_by_id=created_by.id, + token_hash=token_hash(token), + expires_at=now + timedelta(minutes=current_app.config["PASSWORD_RESET_TOKEN_MINUTES"]), + ) + ) + db.session.commit() + return token + + +def active_registration_invitation(token: str) -> RegistrationInvitation | None: + return RegistrationInvitation.query.filter( + RegistrationInvitation.token_hash == token_hash(token), + RegistrationInvitation.used_at.is_(None), + RegistrationInvitation.expires_at > utc_now(), + ).first() + + +def claim_token(model, token_id: int) -> bool: + """Atomically consume a still-valid, one-time link token.""" + return ( + model.query.filter( + model.id == token_id, + model.used_at.is_(None), + model.expires_at > utc_now(), + ).update({"used_at": utc_now()}, synchronize_session=False) + == 1 + ) + + +def revoke_user_authentication(user: User) -> None: + """Invalidate browser sessions and trusted-device cookies for a user.""" + user.revoke_authentication() + TrustedDevice.query.filter_by(user_id=user.id).delete() + + def valid_password(password: str) -> bool: return ( 12 <= len(password) <= 128 @@ -73,6 +172,14 @@ def valid_password(password: str) -> bool: PASSWORD_RULE_MESSAGE = "密码须为12至128位，包含大写字母、小写字母、数字和特殊字符，且不能含空格。" +def valid_new_password(user: User, password: str, confirmation: str) -> bool: + return ( + valid_password(password) + and password == confirmation + and not user.check_password(password) + ) + + def grant_fresh_authorization(user: User, purpose: str) -> None: """Grant a short, session-bound authorization after successful email MFA.""" session[f"fresh_auth_{purpose}"] = { @@ -332,7 +439,7 @@ def verification_page(title: str, purpose: str): "verification.html", title=title, purpose=purpose, - email=session.get("pending_email", ""), + email=masked_email(session.get("pending_email", "")), resend_seconds=current_app.config["EMAIL_CODE_RESEND_SECONDS"], ) @@ -419,9 +526,6 @@ def verify_email(purpose: str): 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") @@ -431,7 +535,7 @@ def verify_email(purpose: str): if purpose == "change_email_new": if User.query.filter(User.email == email, User.id != user.id).first(): flash("该邮箱已被使用。", "error"); return redirect(url_for("main.account_security")) - user.email = email; user.email_verified_at = utc_now(); user.revoke_authentication(); TrustedDevice.query.filter_by(user_id=user.id).delete(); db.session.commit(); logout_user(); session.clear(); flash("邮箱已更新，请重新登录。", "info"); return redirect(url_for("main.login")) + user.email = email; user.email_verified_at = utc_now(); revoke_user_authentication(user); db.session.commit(); logout_user(); session.clear(); flash("邮箱已更新，请重新登录。", "info"); return redirect(url_for("main.login")) grant_fresh_authorization(user, "password_change") return redirect(url_for("main.account_security")) @@ -451,29 +555,89 @@ def forgot_password(): if request.method == "GET": return render_template("forgot_password.html") email = normal_email(request.form.get("email", "")); user = User.query.filter_by(email=email, is_active_account=True).first() if user: - session.update(pending_email=email, pending_user_id=user.id, pending_purpose="reset") - reason = issue_email_code(email, "reset") - if reason in {None, "resend_wait"}: - if reason == "resend_wait": - flash("邮箱验证码已发送，请直接输入验证码继续。", "info") - return redirect(url_for("main.verify_email", purpose="reset")) - flash(email_code_issue_message(reason), "error") + token = issue_password_reset_token(user, user) + try: + send_transactional_email( + to=user.email, + subject="【管道健康】重置密码", + html=password_reset_notice_email( + username=user.username, + reset_url=url_for("main.reset_password_with_token", token=token, _external=True), + minutes=current_app.config["PASSWORD_RESET_TOKEN_MINUTES"], + ), + ) + except (EmailConfigurationError, EmailDeliveryError): + current_app.logger.exception("密码重置邮件发送失败") + flash("若该邮箱已注册，重置链接将发送至邮箱。", "info") + return redirect(url_for("main.forgot_password")) + + +@bp.route("/reset-password/受邀邮箱：{{ email }}。请设置显示名和登录密码以激活账号。
+ {% with messages = get_flashed_messages(with_categories=true) %} + {% for category, message in messages %} +{{ message }}
+ {% endfor %} + {% endwith %} + +即使关闭自助注册，也可向指定邮箱发送一次性注册链接。
+ +输入邮箱后，如账户存在会收到验证码。
输入邮箱后，如账户存在会收到一次性重置链接。
+ + {% with messages = get_flashed_messages(with_categories=true) %} + {% for category, message in messages %} +{{ message }}
+ {% endfor %} + {% endwith %} + + +请设置新的登录密码。提交后，其他登录状态和受信设备将被撤销。
+ {% with messages = get_flashed_messages(with_categories=true) %} + {% for category, message in messages %} +{{ message }}
+ {% endfor %} + {% endwith %} + +