如何在FastAPI中实现用户身份验证和授权

如何在FastAPI中实现用户身份验证和授权

FastAPI 是一个基于Python的高性能Web框架，它提供了许多强大的功能，如异步支持、自动文档生成和类型提示。在现代Web应用中，用户身份验证和授权是一个非常重要的功能，它们能够保护应用的安全性。在本文中，我们将探讨如何在FastAPI中实现用户身份验证和授权。

1. 安装所需的库

在开始之前，我们首先要安装所需的库。在FastAPI中，通常使用PyJWT库来处理JSON Web Tokens，使用Passlib库来进行密码哈希和验证。我们可以使用以下命令来安装这些库：

pip install fastapi pyjwt passlib

2. 创建用户模型

在我们开始实现身份验证和授权之前，我们需要定义一个用户模型。用户模型通常包含用户名、密码等字段。以下是一个示例用户模型的定义：

from pydantic import BaseModel

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

3. 实现用户注册和登录接口

接下来，我们需要实现用户注册和登录接口。在注册接口中，我们将获取用户名和密码，并将密码进行哈希处理后保存到数据库中。在登录接口中，我们将验证用户提供的用户名和密码是否与数据库中的匹配。以下是一个示例的实现：

from fastapi import FastAPI
from passlib.hash import bcrypt

app = FastAPI()

DATABASE = []

@app.post("/register")
def register_user(user: User):
    # Hash password
    hashed_password = bcrypt.hash(user.password)
    
    # Save user to database
    DATABASE.append({"username": user.username, "password": hashed_password})
    
    return {"message": "User registered successfully"}

@app.post("/login")
def login_user(user: User):
    # Find user in database
    for data in DATABASE:
        if data["username"] == user.username:
            # Check password
            if bcrypt.verify(user.password, data["password"]):
                return {"message": "User logged in successfully"}
    
    return {"message": "Invalid username or password"}

4. 实现身份验证和授权中间件

现在我们已经实现了用户注册和登录接口，接下来我们需要实现身份验证和授权的中间件。这将确保只有在提供有效的令牌的情况下，用户才能访问受保护的路由。

以下是一个示例的身份验证和授权中间件的实现：

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from passlib.hash import bcrypt
from jose import jwt, JWTError

app = FastAPI()

SECRET_KEY = "your-secret-key"

security = HTTPBearer()

@app.post("/register")
def register_user(user: User):
    # ...

@app.post("/login")
def login_user(user: User):
    # ...

def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    try:
        token = credentials.credentials
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        user = payload.get("username")
        return user
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token",
            headers={"WWW-Authenticate": "Bearer"},
        )

@app.get("/protected")
def protected_route(current_user: str = Depends(get_current_user)):
    return {"message": f"Hello, {current_user}"}

5. 生成和验证令牌

最后，我们需要实现一个方法来生成令牌。令牌是一种用于身份验证和授权的安全凭证。在用户成功登录后，我们可以使用该方法生成一个令牌，并将其返回给客户端。

以下是一个示例方法来生成和验证令牌的实现：

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from passlib.hash import bcrypt
from jose import jwt, JWTError, ExpiredSignatureError
from datetime import datetime, timedelta

app = FastAPI()

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

security = HTTPBearer()

@app.post("/register")
def register_user(user: User):
    # ...

@app.post("/login")
def login_user(user: User):
    # ...

def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    try:
        token = credentials.credentials
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user = payload.get("username")
        return user
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token",
            headers={"WWW-Authenticate": "Bearer"},
        )

def create_access_token(username: str):
    expires = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    payload = {"username": username, "exp": expires}
    token = jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
    return token

@app.get("/protected")
def protected_route(current_user: str = Depends(get_current_user)):
    return {"message": f"Hello, {current_user}"}

@app.post("/token")
def get_access_token(user: User):
    # Check username and password
    for data in DATABASE:
        if data["username"] == user.username:
            if bcrypt.verify(user.password, data["password"]):
                # Generate access token
                access_token = create_access_token(user.username)
                return {"access_token": access_token}
    
    raise HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Invalid username or password",
        headers={"WWW-Authenticate": "Bearer"},
    )

综上所述，我们已经了解了如何在FastAPI中实现用户身份验证和授权。通过使用PyJWT库和Passlib库，我们能够安全地处理用户凭证并保护应用程序的安全性。这些示例代码可作为起点，您可以根据自己的需求进行进一步的定制和扩展。希望这篇文章对您有所帮助！

【文章原创作者：美国站群多ip服务器 http://www.558idc.com/mgzq.html欢迎留下您的宝贵建议】