1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

server {
    listen 8965;

    # 鉴权接口，仅供 Nginx 内部 auth_request 使用
    location = /auth {
        internal;  # 该接口只能被 Nginx 内部请求，防止外部访问
        # 转发到实际的认证服务
        proxy_pass http://localhost:5001/api/auth/verify;
        # 不转发请求体，提升效率
        proxy_pass_request_body off;
        # 防止后端因 Content-Length 不确定而报错
        proxy_set_header Content-Length "";
        # 将客户端传来的 Authorization 头（JWT Token）传给认证服务
        proxy_set_header Authorization $http_authorization;
        # 传入原始请求路径，供认证服务判断路径是否需要鉴权
        proxy_set_header X-Original-URI $request_uri;
    }

    # 所有 / 路径下的请求都进行认证
    location / {
        # 认证请求会先调用上面的 /auth 接口
        auth_request /auth;
        # 如果认证失败（如返回 401），跳转到自定义处理逻辑
        error_page 401 = @unauthorized;
        # 从 /auth 的响应头中提取用户信息
        auth_request_set $user_id $upstream_http_x_user_id;
        # 把用户信息注入到请求头中，转发给后端业务服务
        proxy_set_header X-User-ID $user_id;
        # 转发到后端服务
        proxy_pass http://localhost:5001;
    }

    # 自定义未授权响应（认证失败时返回）
    location @unauthorized {
        # 返回 401 状态码 + 文本内容
        return 401 "Unauthorized";
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

import jwt
from flask import Flask, request, make_response, jsonify
from jwt import InvalidTokenError

app = Flask(__name__)

SECRET_KEY = "a-string-secret-at-least-256-bits-long"


PUBLIC_PATHS = [
    "/api/public/hello",
    "/api/login",
    "/api/register"
]


@app.route("/api/auth/verify", methods=["GET"])
def verify_token():
    original_uri = request.headers.get("X-Original-URI", "")
    
    if original_uri in PUBLIC_PATHS:
        return "", 200

    
    auth_header = request.headers.get("Authorization")
    if not auth_header or not auth_header.startswith("Bearer "):
        return "Missing or invalid Authorization header", 401

    
    token = auth_header.split(" ", 1)[1]
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        response = make_response("", 200)
        response.headers["X-User-ID"] = str(payload.get("user_id", ""))
        return response
    except InvalidTokenError:
        return "Invalid token", 401


@app.route("/api/hello")
def hello():
    return jsonify({"message": "hello from auth", "user_id": request.headers.get("X-User-ID")})


@app.route("/api/public/hello")
def ping():
    return {"msg": "hello without auth"}


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5001)