用户认证

Flask-Login 提供了用户会话管理功能，简化了认证流程。

安装

pip install flask-login flask-bcrypt

基本配置

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager
from flask_bcrypt import Bcrypt

app = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key'

db = SQLAlchemy(app)
bcrypt = Bcrypt(app)

# 配置 LoginManager
login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'auth.login'  # 登录页面路由
login_manager.login_message = '请先登录'

用户模型

from flask_login import UserMixin

class User(UserMixin, db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    password_hash = db.Column(db.String(128))
    
    def set_password(self, password):
        self.password_hash = bcrypt.generate_password_hash(password).decode('utf-8')
    
    def check_password(self, password):
        return bcrypt.check_password_hash(self.password_hash, password)

# 用户加载回调
@login_manager.user_loader
def load_user(user_id):
    return User.query.get(int(user_id))

登录和登出

from flask import Blueprint, render_template, redirect, url_for, flash, request
from flask_login import login_user, logout_user, login_required, current_user
from urllib.parse import urlparse

auth_bp = Blueprint('auth', __name__)

@auth_bp.route('/login', methods=['GET', 'POST'])
def login():
    if current_user.is_authenticated:
        return redirect(url_for('main.index'))
    
    form = LoginForm()
    if form.validate_on_submit():
        user = User.query.filter_by(username=form.username.data).first()
        
        if user and user.check_password(form.password.data):
            login_user(user, remember=form.remember_me.data)
            next_page = request.args.get('next')
            
            # 安全检查
            if not next_page or urlparse(next_page).netloc != '':
                next_page = url_for('main.index')
            
            flash('登录成功！', 'success')
            return redirect(next_page)
        
        flash('用户名或密码错误', 'danger')
    
    return render_template('auth/login.html', form=form)

@auth_bp.route('/logout')
def logout():
    logout_user()
    flash('已退出登录', 'info')
    return redirect(url_for('main.index'))

注册

@auth_bp.route('/register', methods=['GET', 'POST'])
def register():
    if current_user.is_authenticated:
        return redirect(url_for('main.index'))
    
    form = RegistrationForm()
    if form.validate_on_submit():
        user = User(
            username=form.username.data,
            email=form.email.data
        )
        user.set_password(form.password.data)
        
        db.session.add(user)
        db.session.commit()
        
        flash('注册成功！请登录', 'success')
        return redirect(url_for('auth.login'))
    
    return render_template('auth/register.html', form=form)

保护路由

from flask_login import login_required, current_user

@app.route('/profile')
@login_required
def profile():
    return render_template('profile.html', user=current_user)

@app.route('/admin')
@login_required
def admin():
    if not current_user.is_admin:
        abort(403)
    return render_template('admin.html')

自定义装饰器

from functools import wraps
from flask import abort
from flask_login import current_user

def admin_required(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if not current_user.is_authenticated or not current_user.is_admin:
            abort(403)
        return f(*args, **kwargs)
    return decorated_function

@app.route('/admin/dashboard')
@login_required
@admin_required
def admin_dashboard():
    return render_template('admin/dashboard.html')

记住我功能

# 登录时启用记住我
login_user(user, remember=True)

# 配置记住我 cookie 持续时间
from datetime import timedelta
app.config['REMEMBER_COOKIE_DURATION'] = timedelta(days=30)

密码重置

from itsdangerous import URLSafeTimedSerializer

# 生成重置令牌
def generate_reset_token(user):
    serializer = URLSafeTimedSerializer(app.config['SECRET_KEY'])
    return serializer.dumps(user.email, salt='password-reset-salt')

# 验证令牌
def verify_reset_token(token, expiration=3600):
    serializer = URLSafeTimedSerializer(app.config['SECRET_KEY'])
    try:
        email = serializer.loads(token, salt='password-reset-salt', max_age=expiration)
    except:
        return None
    return User.query.filter_by(email=email).first()

# 请求重置
@auth_bp.route('/reset-password', methods=['GET', 'POST'])
def reset_request():
    form = RequestResetForm()
    if form.validate_on_submit():
        user = User.query.filter_by(email=form.email.data).first()
        if user:
            token = generate_reset_token(user)
            # 发送邮件...
            flash('重置链接已发送到您的邮箱', 'info')
        return redirect(url_for('auth.login'))
    return render_template('auth/reset_request.html', form=form)

# 重置密码
@auth_bp.route('/reset-password/<token>', methods=['GET', 'POST'])
def reset_token(token):
    user = verify_reset_token(token)
    if user is None:
        flash('链接无效或已过期', 'warning')
        return redirect(url_for('auth.reset_request'))
    
    form = ResetPasswordForm()
    if form.validate_on_submit():
        user.set_password(form.password.data)
        db.session.commit()
        flash('密码已重置，请登录', 'success')
        return redirect(url_for('auth.login'))
    
    return render_template('auth/reset_token.html', form=form)

模板中使用

<!-- 检查登录状态 -->
{% if current_user.is_authenticated %}
    <p>欢迎, {{ current_user.username }}!</p>
    <a href="{{ url_for('auth.logout') }}">退出</a>
{% else %}
    <a href="{{ url_for('auth.login') }}">登录</a>
    <a href="{{ url_for('auth.register') }}">注册</a>
{% endif %}

<!-- 检查权限 -->
{% if current_user.is_admin %}
    <a href="{{ url_for('admin.dashboard') }}">管理后台</a>
{% endif %}

下一步

学习使用 Blueprints 组织大型应用。