用户认证¶

# app/__init__.py
from flask_login import LoginManager

login_manager = LoginManager()

def create_app():
    app = Flask(__name__)

    # 配置Flask-Login
    login_manager.init_app(app)
    login_manager.login_view = 'auth.login'
    login_manager.login_message = '请先登录'
    login_manager.login_message_category = 'info'

    return app

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

# app/models.py
from flask_login import UserMixin
from werkzeug.security import generate_password_hash, check_password_hash

class User(UserMixin, db.Model):
    """用户模型"""
    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(20), unique=True, nullable=False, index=True)
    email = db.Column(db.String(120), unique=True, nullable=False, index=True)
    password_hash = db.Column(db.String(128))
    created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))  # lambda匿名函数：简洁的单行函数
    is_admin = db.Column(db.Boolean, default=False)
    is_active = db.Column(db.Boolean, default=True)

    def set_password(self, password):
        """设置密码"""
        self.password_hash = generate_password_hash(password)

    def check_password(self, password):
        """验证密码"""
        return check_password_hash(self.password_hash, password)

    def __repr__(self):
        return f'<User {self.username}>'

# app/auth/routes.py
from flask import render_template, redirect, url_for, flash, request
from flask_login import login_user, logout_user, current_user
from app.auth.forms import LoginForm
from app.models import User

@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):
            if not user.is_active:
                flash('账号已被禁用', 'error')
                return render_template('auth/login.html', form=form)

            login_user(user, remember=form.remember_me.data)
            flash('登录成功！', 'success')

            next_page = request.args.get('next')
            if not next_page or not next_page.startswith('/'):
                next_page = url_for('main.index')

            return redirect(next_page)
        else:
            flash('用户名或密码错误', 'error')

    return render_template('auth/login.html', form=form)

@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)

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

# config.py
from datetime import timedelta

class Config:
    # 会话配置
    SESSION_COOKIE_SECURE = True  # 仅HTTPS
    SESSION_COOKIE_HTTPONLY = True  # 禁止JavaScript访问
    SESSION_COOKIE_SAMESITE = 'Lax'  # 防止CSRF
    PERMANENT_SESSION_LIFETIME = timedelta(days=7)  # 会话有效期

# app/auth/forms.py
class LoginForm(FlaskForm):
    username = StringField('用户名',
                       validators=[DataRequired()])
    password = PasswordField('密码',
                          validators=[DataRequired()])
    remember_me = BooleanField('记住我')
    submit = SubmitField('登录')

# 登录时使用remember_me
login_user(user, remember=form.remember_me.data)

# app/decorators.py
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):  # *args接收任意位置参数；**kwargs接收任意关键字参数
        if not current_user.is_authenticated:
            abort(403)
        if not current_user.is_admin:
            abort(403)
        return f(*args, **kwargs)
    return decorated_function

def permission_required(permission):
    """权限检查装饰器"""
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            if not current_user.is_authenticated:
                abort(403)
            if not current_user.has_permission(permission):
                abort(403)
            return f(*args, **kwargs)
        return decorated_function
    return decorator

# app/models.py
class User(UserMixin, db.Model):
    # ... 其他字段 ...

    def has_permission(self, permission):
        """检查用户是否有指定权限"""
        return self.is_admin or permission in self.permissions

    def can_edit_post(self, post):
        """检查是否可以编辑文章"""
        return self.id == post.author_id or self.is_admin

    def can_delete_post(self, post):
        """检查是否可以删除文章"""
        return self.id == post.author_id or self.is_admin

# app/auth/routes.py
from flask_mail import Message
from app import mail

def send_password_reset_email(user):
    """发送密码重置邮件"""
    token = user.get_reset_token()
    msg = Message('重置密码',
                 recipients=[user.email])
    msg.body = f'''重置密码的链接：
{url_for('auth.reset_password', token=token, _external=True)}

如果你没有请求重置密码，请忽略此邮件。
'''
    mail.send(msg)

@auth_bp.route('/reset_password_request', methods=['GET', 'POST'])
def reset_password_request():
    if current_user.is_authenticated:
        return redirect(url_for('main.index'))

    form = ResetPasswordRequestForm()

    if form.validate_on_submit():
        user = User.query.filter_by(email=form.email.data).first()
        if user:
            send_password_reset_email(user)
        flash('如果邮箱存在，重置链接已发送', 'info')
        return redirect(url_for('auth.login'))

    return render_template('auth/reset_password_request.html', form=form)

@auth_bp.route('/reset_password/<token>', methods=['GET', 'POST'])
def reset_password(token):
    if current_user.is_authenticated:
        return redirect(url_for('main.index'))

    user = User.verify_reset_token(token)
    if not user:
        flash('无效或过期的重置链接', 'error')
        return redirect(url_for('auth.reset_password_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_password.html', form=form)

# 认证安全最佳实践
auth_security = {
    '密码哈希': '使用安全的密码哈希算法',
    'HTTPS': '强制使用HTTPS',
    'CSRF保护': '启用CSRF保护',
    '会话安全': '配置安全的会话选项',
    '密码策略': '实施强密码策略'
}

# 认证用户体验
auth_ux = {
    '记住我': '提供记住我功能',
    '密码重置': '实现密码重置功能',
    '邮箱验证': '验证用户邮箱',
    '社交登录': '支持第三方登录',
    '双因素认证': '提供2FA选项'
}