使用oauth2和jwt在FastApi中进行身份验证

使用OAuth2和JWT在FastAPI中进行身份验证是一种常见的安全机制。下面是对这个问题的完善且全面的答案：

身份验证是在Web应用程序中验证用户身份的过程。OAuth2和JWT是两个在身份验证中常用的技术。

OAuth2是一种授权框架，用于授权第三方应用程序访问用户资源，而无需共享用户的凭据。它通过令牌的方式实现授权，包括访问令牌和刷新令牌。访问令牌用于访问受保护的资源，而刷新令牌用于获取新的访问令牌。OAuth2有四种授权类型：授权码模式、密码模式、客户端模式和隐式模式。不同类型的授权适用于不同的应用场景。

JWT（JSON Web Token）是一种用于安全传输信息的开放标准。它是一种紧凑且自包含的方式，用于在各方之间传输信息。JWT由三部分组成：头部、载荷和签名。头部包含关于令牌类型和加密算法的信息，载荷包含实际传输的数据，签名用于验证令牌的完整性。JWT可以用于身份验证和信息交换，因为它可以被验证和信任。

在FastAPI中使用OAuth2和JWT进行身份验证可以按照以下步骤进行：

导入相关模块和库：

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel
from datetime import datetime, timedelta

定义配置项：

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

创建密码哈希上下文：

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

定义用户模型：

class User(BaseModel):
    username: str
    password: str

模拟用户数据库：

fake_users_db = {
    "johnsnow": {
        "username": "johnsnow",
        "full_name": "John Snow",
        "email": "johnsnow@example.com",
        "hashed_password": "$2b$12$AInI7ve5W4wBpZ9C22PvhOv7QyqzxgXt.0EBlF9b95QKqq3Qw8nuq",
    }
}

创建密码哈希函数：

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

创建身份验证函数：

def authenticate_user(username: str, password: str):
    if username in fake_users_db:
        user = fake_users_db[username]
        if not verify_password(password, user["hashed_password"]):
            return False
        return user

def create_access_token(data: dict, expires_delta: timedelta):
    to_encode = data.copy()
    expire = datetime.utcnow() + expires_delta
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

定义登录路由：

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

@app.post("/login")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )

    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user["username"]}, expires_delta=access_token_expires
    )
    
    return {"access_token": access_token, "token_type": "bearer"}

定义受保护的路由：

async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authentication token")
    except JWTError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authentication token")

    user = fake_users_db.get(username)
    if user is None:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
    
    return user

@app.get("/protected")
async def protected_route(user: User = Depends(get_current_user)):
    return {"message": f"Hello, {user['full_name']}!"}

这是在FastAPI中使用OAuth2和JWT进行身份验证的一个基本示例。根据具体需求，你可以进一步定制和优化身份验证过程，如添加访问控制列表（ACL）等。

此外，腾讯云提供了一系列与身份验证和授权相关的产品和服务，例如云访问管理（CAM）和访问管理系统（Cloud Access Management）。你可以根据具体需求选择相应的腾讯云产品并参考其官方文档了解更多详情。

参考链接：