第 8 章 · 认证与鉴权（JWT）

from werkzeug.security import generate_password_hash, check_password_hash

h1 = generate_password_hash("123456")
h2 = generate_password_hash("123456")   # 再算一次，结果和 h1 不同

print(h1)
print("两次相同吗：", h1 == h2)
print("校验正确密码：", check_password_hash(h1, "123456"))
print("校验错误密码：", check_password_hash(h1, "654321"))

import re
from flask import Blueprint, request, jsonify
from werkzeug.security import generate_password_hash
from ..models import db, User

user_bp = Blueprint("user", __name__, url_prefix="/api/user")

# 用户名：3-20 位，字母 / 数字 / 下划线（对齐接口文档 2.1）
USERNAME_RE = re.compile(r"^\w{3,20}$")


@user_bp.post("/register")
def register():
    data = request.get_json(silent=True) or {}
    username = (data.get("username") or "").strip()
    password = data.get("password") or ""
    nickname = (data.get("nickname") or "").strip() or username
    email = (data.get("email") or "").strip() or None

    # ① 参数校验：不通过就返回 400 + 可读提示
    if not USERNAME_RE.match(username):
        return jsonify(code=400, msg="用户名需 3-20 位字母、数字或下划线", data=None)
    if not (6 <= len(password) <= 20):
        return jsonify(code=400, msg="密码长度需 6-20 位", data=None)
    if email and ("@" not in email or "." not in email.split("@")[-1]):
        return jsonify(code=400, msg="邮箱格式不正确", data=None)

    # ② 唯一性校验：用户名、邮箱都不能重复
    if User.query.filter_by(username=username).first():
        return jsonify(code=400, msg="用户名已存在", data=None)
    if email and User.query.filter_by(email=email).first():
        return jsonify(code=400, msg="邮箱已被注册", data=None)

    # ③ 密码哈希后写库——这一行是整章的底线，绝不能存明文
    user = User(
        username=username,
        password=generate_password_hash(password),
        nickname=nickname,
        email=email,
    )
    db.session.add(user)
    db.session.commit()

    return jsonify(code=200, msg="注册成功", data={"user_id": user.id})

{
  "code": 400,
  "msg": "用户名已存在",
  "data": null
}

from datetime import timedelta
from flask import Flask, jsonify
from flask_jwt_extended import JWTManager

jwt = JWTManager()


def create_app():
    app = Flask(__name__)
    app.config["SECRET_KEY"] = "dev-secret-change-me"
    # JWT 的签名密钥：生产环境务必从环境变量读，不要写死在代码里
    app.config["JWT_SECRET_KEY"] = "jwt-secret-change-me"
    app.config["JWT_ACCESS_TOKEN_EXPIRES"] = timedelta(hours=2)  # 默认 15 分钟，这里放宽

    jwt.init_app(app)

    # 统一 401 响应：注意这里不传 status code，HTTP 状态码仍是 200（对齐接口文档）
    @jwt.unauthorized_loader
    def no_token(reason):
        return jsonify(code=401, msg="请先登录", data=None)

    @jwt.expired_token_loader
    def token_expired(header, payload):
        return jsonify(code=401, msg="登录已过期，请重新登录", data=None)

    from .user.views import user_bp
    app.register_blueprint(user_bp)
    return app

from werkzeug.security import check_password_hash
from flask_jwt_extended import create_access_token


@user_bp.post("/login")
def login():
    data = request.get_json(silent=True) or {}
    username = (data.get("username") or "").strip()
    password = data.get("password") or ""
    if not username or not password:
        return jsonify(code=400, msg="用户名和密码不能为空", data=None)

    user = User.query.filter_by(username=username).first()

    # 用户不存在 与 密码错误 走同一句提示，避免暴露「这个账号存不存在」
    if user is None or not check_password_hash(user.password, password):
        return jsonify(code=400, msg="用户名或密码错误", data=None)
    if user.status != 1:
        return jsonify(code=403, msg="账号已被禁用", data=None)

    # 身份标识必须是字符串：Flask-JWT-Extended 4.6+ 要求 sub 声明为字符串
    token = create_access_token(identity=str(user.id))

    return jsonify(code=200, msg="登录成功", data={
        "token": token,
        "user": {
            "id": user.id,
            "username": user.username,
            "nickname": user.nickname,
            "avatar": user.avatar,
        },
    })

# 原始 token（示意，实际更长）
eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9 . eyJzdWIiOiIzIiwiZXhwIjoxNzk5MDAwMDAwfQ . 3fQ9x...

# 第 1 段 header（Base64 解码后）
{ "alg": "HS256", "typ": "JWT" }

# 第 2 段 payload（Base64 解码后）—— 明文可读！
{ "sub": "3", "exp": 1799000000 }

from flask_jwt_extended import jwt_required, get_jwt_identity


@user_bp.get("/profile")
@jwt_required()                          # 没带合法 token 会直接在这里被拦下
def profile():
    user_id = int(get_jwt_identity())    # 取回登录时写入的 user_id（字符串 → int）
    user = db.session.get(User, user_id)
    if user is None:
        return jsonify(code=404, msg="用户不存在", data=None)

    return jsonify(code=200, msg="success", data={
        "id": user.id,
        "username": user.username,
        "nickname": user.nickname,
        "email": user.email,
        "avatar": user.avatar,
        "intro": user.intro,
        "gender": user.gender,
        "created_at": user.created_at.strftime("%Y-%m-%d %H:%M:%S"),
    })

from functools import wraps
from flask import g, jsonify
from flask_jwt_extended import verify_jwt_in_request, get_jwt_identity
from ..models import db, User


def login_required(fn):
    """需要登录的接口装饰器：校验 JWT → 查出 User → 挂到 g.user。"""
    @wraps(fn)
    def wrapper(*args, **kwargs):
        # ① 校验请求头里的 token（不带 / 过期 / 伪造都会抛异常）
        try:
            verify_jwt_in_request()
        except Exception:
            return jsonify(code=401, msg="请先登录", data=None)

        # ② 取 user_id 并查库
        user = db.session.get(User, int(get_jwt_identity()))
        if user is None or user.status != 1:
            return jsonify(code=401, msg="账号不存在或已被禁用", data=None)

        # ③ 放进 g，供视图与后续逻辑使用（呼应第 4 章的 g 对象）
        g.user = user
        return fn(*args, **kwargs)

    return wrapper

from flask import Blueprint, request, jsonify, g
from ..utils.auth import login_required
from ..models import db, Comment

comment_bp = Blueprint("comment", __name__, url_prefix="/api/comment")


@comment_bp.post("/add")
@login_required                       # 一行搞定鉴权 + 取当前用户
def add_comment():
    data = request.get_json(silent=True) or {}
    article_id = data.get("article_id")
    content = (data.get("content") or "").strip()
    if not article_id or not content:
        return jsonify(code=400, msg="缺少参数：article_id / content", data=None)

    comment = Comment(
        user_id=g.user.id,            # 当前用户直接来自 g，不用前端传、也传不了
        article_id=article_id,
        parent_id=data.get("parent_id"),
        content=content,
    )
    db.session.add(comment)
    db.session.commit()
    return jsonify(code=200, msg="发表成功", data={"comment_id": comment.id})

@user_bp.post("/logout")
@login_required
def logout():
    # JWT 无状态：真正的「退出」由前端删除 localStorage 里的 token 完成。
    # 后端这一步的作用是记录日志；若要立刻作废 token，可走下面的黑名单方案。
    return jsonify(code=200, msg="已退出", data=None)

from flask_jwt_extended import get_jwt, jwt_required

# 思路：把「要作废的 token 的唯一编号 jti」存进 Redis，校验时先查黑名单
# 1) 签发时不用改，token 自带 jti
# 2) 退出时把 jti 写进黑名单，过期时间设成 token 剩余寿命
# 3) 注册校验回调，命中黑名单就拒绝
blacklist = set()   # 生产环境用 Redis，这里用集合示意

@jwt.token_in_blocklist_loader
def is_revoked(header, payload):
    return payload["jti"] in blacklist

# 退出时：blacklist.add(get_jwt()["jti"])

// 请求：POST {{base_url}}/user/login
// 把响应里的 token 存进当前环境，供其它接口复用

const res = pm.response.json();

pm.test("登录成功", function () {
    pm.expect(res.code).to.eql(200);
});

if (res.code === 200) {
    pm.environment.set("token", res.data.token);
    console.log("token 已保存到环境变量");
}

// 提示：先点「Send」跑一次登录，环境里才会出现 token

Key            Value
Authorization  Bearer {{token}}
Content-Type   application/json

# 说明：
# 1) Bearer 与 {{token}} 之间是「一个空格」，不是冒号
# 2) {{token}} 是 Postman 的环境变量引用语法，发送时会自动替换成真实值
# 3) 若手抖漏了空格，或写成 token: xxx，都会拿到 401 请先登录