Python中实现JWT认证的完整指南(从生成到验证)

目录

• 引言
• 一、JWT 是什么？为什么需要它？
• 传统 session 与 JWT 对比
• 二、JWT 的结构解析
• 三、python 中实现 JWT
• 1. 安装 PyJWT 包
• 2. 生成 JWT
• 3. 验证 JWT
• 4. 错误处理大全
• 四、高级应用场景
• 1. 双令牌系统（Access + Refresh）
• 详细说明表格：
• 异常处理补充表：
• 2. 与 FastAPI/Django 集成
• 五、安全最佳实践
• 六、性能优化技巧
• 算法性能比较表（执行时间，越小越好）
• 详细对比表格：
• 七、完整示例：FastAPI 实现
• 结语

引言

jsON Web Tokens (JWT) 是现代 Web 开发中广泛使用的身份验证机制。本文将用生动的方式带你全面了解 JWT 在 Python 中的实现，包括生成、验证和各种相关方法，通过丰富的比喻、表格和流程图帮助你彻底掌握 JWT。

一、JWT 是什么？为什么需要它？

想象你去游乐园，入园时会得到一个手环。这个手环：

• 包含信息：显示你的门票类型（VIP/普通）
• 防伪设计：有特殊图案难以伪造
• 有效期：只在当天有效

JWT 就是这样的数字手环：

传统 session 与 JWT 对比

二、JWT 的结构解析

一个 JWT 看起来像编程客栈这样：xxxxx.yyyyy.zzzzz

就像三明治分三层：

Header (面包上层)

{
  "alg": "HS256",  // 签名算法(HMAC SHA256)
  "typ": "JWT"     // 类型标识
}

Payload (馅料)

{
  "sub": "user123",  // 主题(用户ID)
  "name": "张三",
  "admin": true,
  "iat": 1516239022  // 签发时间
}

Signature (面包下层+防伪标记)

HMACSHA256(
  base64UrlEncode(header) + "." +
  base64UrlEncode(payloandroidad),
  密钥
)

生成流程：

[Header] → base64编码 → xxxxx
[Payload] → base64编码 → yyyyy
[xxxxx.yyyyy + 密钥] → 签名算法 → zzzzz
最终令牌：xxxxx.yyyyy.zzzzz

三、Python 中实现 JWT

1. 安装 PyJWT 包

pip install pyjwt

2. 生成 JWT

import jwt
import datetime
from datetime import timezone

# 建议从环境变量读取
SECRET_KEY = "your_super_secret_key"

def generate_jwt(user_id: str, username: str, role: str) -> str:
    """生成JWT令牌"""
    payload = {
        "sub": user_id,  # 标准字段：主题
        "name": username,
        "role": role,
        "iat": datetime.datetime.now(tz=timezone.utc),  # 签发时间
        "exp": datetime.datetime.now(tz=timezone.utc) + datetime.timedelta(hours=1)  # 过期时间
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

参数详解表：

标准声明字段(建议)：

3. 验证 JWT

def verify_jwt(token: str) -> dict:
    """验证JWT令牌"""
    try:
        payload = jwt.decode(
            token, 
            SECRET_KEY, 
            algorithms=["HS256"],
            audience="your-app",  # 验证接收方
            issuer="auth-server"  # 验证签发方
        )
        return payload
    except jwt.PyJWTError as e:
        print(f"Token验证失败: {type(e).__name__}: {e}")
        return None

验证流程示意图：

4. 错误处理大全

from fastapi import HTTPException, status

def validate_token(token: str):
    try:
        return jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token已过期",
            headers={"WWW-Authenticate": "Bearer"}
        )
    except jwt.InvalidTokenError as e:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=f"无效Token: {str(e)}",
            headers={"WWW-Authenticate": "Bearer"}
        )

异常类型表：

四、高级应用场景

1. 双令牌系统（Access + Refresh）

以下是基于你的 JWT 认证流程整理的表格表示：

详细说明表格：

异常处理补充表：

实现代码：

def generate_token_pair(user_id: str):
    """生成令牌对"""
    access_payload = {
        "sub": user_id,
        "type": "access",
        "exp": datetime.datetime.now(tz=timezone.utc) + datetime.timedelta(minutes=15)
    }
    refresh_payload = {
        "sub": user_id,
        "type": "refresh",
        "exp": datetime.datetime.now(tz=timezone.utc) + datetime.timedelta(days=7)
    }
    
    access_token = jwt.encode(access_payload, SECRET_KEY, algorithm="HS256")
    refresh_token = jwt.encode(refresh_payload, SECRET_KEY, algorithm="HS256")
    
    return {
        "access_token": access_token,
        "refresh_token": refresh_token
    }

def refresh_access_token(refresh_token: str):
    """使用refresh_token获取新access_token"""
    try:
        payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=["HS256"])
        if payload.get("type") != "refresh":
            raise HTTPException(status_code=400, detail="无效的refresh token类型")
        
        new_payload = {
            "sub": payload["sub"],
            "type": "access",
            "exp": datetime.datetime.now(tz=timezone.utc) + datetime.timedelta(minutes=15)
        }
        return jwt.encode(new_payload, SECRET_KEY, algorithm="HS256")
    
    except jwt.PyJWTError as e:
        raise HTTPException(status_code=401, detail=f"refresh token无效: {str(e)}")

2. 与 FastAPI/Django 集成

FastAPI 示例：

from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

security = HTTPBearer()

app = FastAPI()

async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return payload
    except jwt.PyJWTError:
        raise HTTPException(status_code=401, detail="无效或过期的token")

@app.get("/protected")
async def protected_route(user: dict = Depends(get_current_user)):
    return {"message": f"你好, {user['sub']}!", "user_data": user}

Django 示例：

from django.http import JsonResponse
from functools import wraps

def jwt_required(view_func):
    @wraps(view_func)
    def wrapper(request, *args, **kwargs):
        auth_header = request.headers.get('Authorization')
        if not auth_header or not auth_header.startswith('Bearer '):
            return JsonResponse({"error": "未提供token"}, status=401)
        
        token = auth_header.split(' ')[1]
        try:
            request.user = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
            return view_func(request, *args, **kwargs)
        except jwt.ExpiredSignatureErrphpor:
            return JsonResponse({"error": "token已过期"}, status=401)
        except jwt.InvalidTokenError:
            return JsonResponse({"error": "无效token"}, status=401)
    return wrapper

@jwt_required
def protected_view(request):
    return JsonResponse({"data": "受保护的内容", "user": request.user})

五、安全最佳实践

密钥管理：

• 使用环境变量存储密钥

import os
SECRET_KEY = os.getenv("JWT_SECRET_KEY", "fallback-secret")

• 定期轮换密钥（密钥版本控制）

keys = {
    "2023": "old-secret",
    "2024": "current-secret"
}

增强安全措施：

def generate_secure_token(user, ip):
    """生成带IP绑定的token"""
    payload = {
        "sub": user.id,
        "ip": ip,  # 绑定客户端IP
        "jti": str(uuid.uuid4())  # 唯一标识防止重放
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

def verify_secure_token(token, ip):
    payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
    if payload.get("ip") != ip:
        raise ValueError("IP地址不匹配")
    return payload

黑名单实现：

from Redis import Redis

redis = Redis(host='localhost', port=6379)

def revoke_token(jti: str, expire_in: int):
    """将token加入黑名单"""
    redis.setex(f"blacklist:{jti}", expire_in, "revoked")

def is_revoked(jti: str) -> bool:
    """检查是否在黑名单"""
    return bool(redis.exists(f"blacklist:{jti}"))

六、性能优化技巧

选择更快的算法：

算法性能比较表（执行时间，越小越好）

详细对比表格：

减少payload大小：

# 不推荐 - payload过大
payload = {**user.__dict__, "iat": ..., "exp": ...}

# 推荐 - 只存储必要信息
payload = {
    "sub": user.id,
    "role": user.role,
    "iat": ...,
    "exp": ...
}

异步签名验证：

import asyncio
from jwt import PyJWT

async def async_verify(token: str):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(
        None, 
        lambda: jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
    )

七、完整示例：FastAPI 实现

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
import jwt
import datetime
from datetime import timezone
from typing import Optional

app = FastAPI()
security = HTTPBearer()

# 配置
SECRET_KEY = "your-secret-key-here"  # 生产环境应从环境变量获取
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE = datetime.timedelta(minutes=15)
REFRESH_TOKEN_EXPIRE = datetime.timedelta(days=7)

# 模拟数据库
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "password": "secret123",
        "role": "user"
    },
    "admin": {
        "username": "admin",
        "password": "admin123",
        "role": "admin"
    }
}

class Token(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str

class User(BaseModel):
    username: str
    role: Optional[str] = None

def create_tokens(username: str):
    """创建access和refresh令牌对"""
    access_payload = {
        "sub": username,
        "role": fake_users_db[username]["role"],
        "type": "access",
        "exp": datetime.datetime.now(tz=timezone.utc) + ACCESS_TOKEN_EXPIRE
    }
    refresh_payload = {
        "sub": username,
        "type": "refresh",
        "exp": datetime.datetime.now(tz=timezone.utc) + REFRESH_TOKEN_EXPIRE
    }
    
    access_token = jwt.encode(access_payload, SECRET_KEY, algorithm=ALGORITHM)
    refresh_token = jwt.encode(refresh_payload, SECRET_KEY, algorithm=ALGORITHM)
    
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer"
    }

async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    """依赖项：验证JWT并返回当前用户"""
    token = credentials.credentials
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        if payload.get("type") != "access":
            raise HTTPException(status_code=400, detail="需要access token")
        username = payload.get("sub")
        if username not in fake_users_db:
            raise HTTPException(status_code=404, detail="用户不存在")
        return User(username=username, role=payload.get("role"))
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token已过期",
            headers={"WWW-Authenticate": "Bearer"}
        )
    except jwt.InvalidTokenError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效Token",
            headers={"WWW-Authenticate": "Bearer"}
        )

@app.post("/login")
async def login(username: str, password: str):
    """登录接口"""
    if username not in fake_users_db or fake_users_db[username]["password"] != password:
        raise HTTPException(status_code=400, detail="用户名或密码错误")
    return create_tokens(username)

@app.post("/refresh")
async def refresh_token(refresh_token: str):
    """刷新access token"""
    try:
        payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
        if payload.get("type") != "refresh":
            raise HTTPException(status_code=400, detail="需要refresh token")
        return create_tokens(payload.get("sub"))
    except jwt.PyJWTError as e:
        raise HTTPException(status_code=401, detail=str(e))

@app.get("/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    """获取当前用户信息"""
    return current_user

@app.get("/admin")
async def admin_route(current_user: User = Depends(get_current_user)):
    """管理员专属路由"""
    if current_user.role != "admin":
        raise HTTPException(status_code=403, detail="无权访问")
    return {"message": "欢迎管理员"}

结语

通过本文，你已经掌握了：

• JWT 的核心原理和结构
• Python 中生成和验证 JWT 的完整方法
• 各种高级应用场景和安全实践
• 与流行框架的集成方式

记住这些黄金法则：

• 最小化payload：只存储必要信息
• 保护密钥：像保护银行密码一样保护你的密钥
• 合理设置有效期：平衡安全性和用户体验
• 考虑刷新机制：提升用户体验同时保持安全

现在，你已经准备好为你的 Python 应用实现强大的 JWT 认证系统了！

以上就是Python中JWT认证的完整指南(从生成到验证)的详细内容，更多关于Python JWT认证指南的资料请关注编程客栈(www.devze.com)其它相关文章！