专业的编程技术博客社区

网站首页 > 博客文章 正文

FastAPI 中的身份验证和授权:OAuth2 集成

baijin 2024-10-14 08:08:18 博客文章 8 ℃ 0 评论

在现代 Web 应用程序中,保护端点和管理用户身份验证和授权至关重要。FastAPI 为 OAuth2 提供强大的支持,使开发人员能够轻松实现安全的身份验证机制。本博客将指导您在 FastAPI 应用程序中集成 OAuth2,并包含实际演示以巩固您的理解。

1. OAuth2 简介

OAuth2 是一种访问委派的开放标准,通常用于授予网站或应用程序有限的用户信息访问权限,而无需暴露密码。它通过向第三方客户端发出访问令牌来工作,使他们能够代表用户访问受保护的资源。

2. 使用 OAuth2 设置 FastAPI

首先,让我们设置一个基本的 FastAPI 应用程序并安装必要的依赖项。

安装 FastAPI 和 uvicorn:

pip install fastapi uvicorn python-oauthlib

创建 FastAPI 应用程序:

from fastapi import FastAPI


app = FastAPI()


@app.get("/")
def read_root():
    return {"message": "Welcome to the FastAPI OAuth2 example"}

3. 创建 OAuth2 客户端

要实现 OAuth2,您需要一个 OAuth2 提供程序。在本示例中,我们将使用 GitHub 作为 OAuth2 提供程序。您需要在 GitHub 上注册您的应用程序并获取 client_id 和 client_secret。

创建配置文件 (config.py):

import os


class Settings:
    GITHUB_CLIENT_ID: str = os.getenv("GITHUB_CLIENT_ID")
    GITHUB_CLIENT_SECRET: str = os.getenv("GITHUB_CLIENT_SECRET")
    GITHUB_AUTHORIZE_URL: str = "https://github.com/login/oauth/authorize"
    GITHUB_TOKEN_URL: str = "https://github.com/login/oauth/access_token"
    GITHUB_API_URL: str = "https://api.github.com/user"


settings = Settings()

4. 使用 OAuth2 保护端点

在 FastAPI 中集成 OAuth2 (main.py):

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2AuthorizationCodeBearer
from pydantic import BaseModel
import requests


from config import settings


app = FastAPI()


oauth2_scheme = OAuth2AuthorizationCodeBearer(
    authorizationUrl=settings.GITHUB_AUTHORIZE_URL,
    tokenUrl=settings.GITHUB_TOKEN_URL,
    clientId=settings.GITHUB_CLIENT_ID,
    clientSecret=settings.GITHUB_CLIENT_SECRET,
)


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


@app.post("/token", response_model=Token)
def login(code: str):
    response = requests.post(
        settings.GITHUB_TOKEN_URL,
        data={
            'client_id': settings.GITHUB_CLIENT_ID,
            'client_secret': settings.GITHUB_CLIENT_SECRET,
            'code': code,
        },
        headers={'Accept': 'application/json'},
    )
    token = response.json()
    if 'error' in token:
        raise HTTPException(status_code=400, detail=token['error'])
    return Token(access_token=token['access_token'], token_type="bearer")


@app.get("/users/me")
def read_users_me(token: str = Depends(oauth2_scheme)):
    response = requests.get(
        settings.GITHUB_API_URL,
        headers={'Authorization': f'Bearer {token}'},
    )
    user = response.json()
    if 'error' in user:
        raise HTTPException(status_code=400, detail=user['error'])
    return user

5. 测试 OAuth2 集成

要测试 OAuth2 集成,请按照以下步骤操作:

导航到http://localhost:8000/docs 访问 FastAPI 交互式文档。

单击 /token 端点并提供从 GitHub 获取的授权代码

单击 /users/me 端点并使用获取的访问令牌对其进行授权,以从 GitHub 获取用户信息。

演示

获取 OAuth2 授权码:

在浏览器中导航到以下 URL(将 client_id 和 redirect_uri 替换为您的值):

https://github.com/login/oauth/authorize?client_id=<client_id>&redirect_uri=<redirect_uri>&scope=read:user

GitHub 将提示您授权应用程序并使用 code 查询参数将您重定向到 redirect_uri。

将授权码交换为访问令牌:

使用 FastAPI 文档中的 /token 端点将授权码交换为访问令牌。

访问受保护的资源:

使用 FastAPI 文档中的 /users/me 端点,使用获得的访问令牌对其进行授权,并从 GitHub 获取用户信息。

通过实施 OAuth2,您可以确保 FastAPI 应用程序的身份验证安全且可扩展,从而为用户提供无缝且安全的体验。

FastAPI 中 OAuth2 集成的其他演示

除了前面描述的基本实现之外,这里还有一些高级演示,可加深您对 FastAPI 中 OAuth2 集成的理解:

演示 1:刷新令牌

OAuth2 还支持刷新令牌,允许客户端无需重新验证用户身份即可获取新的访问令牌。以下是您可以在 FastAPI 应用程序中实现刷新令牌的方法。

在 Token 模型中添加必要的字段,并更新 /token 端点以处理刷新令牌:

from pydantic import BaseModel


class Token(BaseModel):
    access_token: str
    token_type: str
    refresh_token: str  # Add refresh token field


@app.post("/token", response_model=Token)
def login(code: str):
    response = requests.post(
        settings.GITHUB_TOKEN_URL,
        data={
            'client_id': settings.GITHUB_CLIENT_ID,
            'client_secret': settings.GITHUB_CLIENT_SECRET,
            'code': code,
        },
        headers={'Accept': 'application/json'},
    )
    token = response.json()
    if 'error' in token:
        raise HTTPException(status_code=400, detail=token['error'])
    return Token(
        access_token=token['access_token'],
        token_type="bearer",
        refresh_token=token.get('refresh_token', '')  # Handle refresh token if provided
    )

添加端点以刷新令牌:

@app.post("/refresh-token", response_model=Token)
def refresh_token(refresh_token: str):
    response = requests.post(
        settings.GITHUB_TOKEN_URL,
        data={
            'client_id': settings.GITHUB_CLIENT_ID,
            'client_secret': settings.GITHUB_CLIENT_SECRET,
            'grant_type': 'refresh_token',
            'refresh_token': refresh_token,
        },
        headers={'Accept': 'application/json'},
    )
    token = response.json()
    if 'error' in token:
        raise HTTPException(status_code=400, detail=token['error'])
    return Token(
        access_token=token['access_token'],
        token_type="bearer",
        refresh_token=token.get('refresh_token', '')
    )

演示 2:OAuth2 Scopes

Scopes允许您定义和限制应用程序对用户数据的访问权限。以下是在 FastAPI 中添加和检查范围的方法。

定义 OAuth2 范围并更新 OAuth2AuthorizationCodeBearer 实例:

from fastapi.security import OAuth2AuthorizationCodeBearer


oauth2_scheme = OAuth2AuthorizationCodeBearer(
    authorizationUrl=settings.GITHUB_AUTHORIZE_URL,
    tokenUrl=settings.GITHUB_TOKEN_URL,
    clientId=settings.GITHUB_CLIENT_ID,
    clientSecret=settings.GITHUB_CLIENT_SECRET,
    scopes={"read:user": "Read user information"}
)

添加依赖项以检查所需范围:

from fastapi import Security


def get_current_user(token: str = Security(oauth2_scheme, scopes=["read:user"])):
    response = requests.get(
        settings.GITHUB_API_URL,
        headers={'Authorization': f'Bearer {token}'},
    )
    user = response.json()
    if 'error' in user:
        raise HTTPException(status_code=400, detail=user['error'])
    return user


@app.get("/users/me")
def read_users_me(current_user: dict = Depends(get_current_user)):
    return current_user

演示 3:自定义 OAuth2 中间件

您可以创建自定义中间件,以更灵活地处理 OAuth2 身份验证。此中间件将拦截请求并验证 OAuth2 令牌。

创建自定义 OAuth2 中间件 (middleware.py):

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from fastapi import HTTPException


class OAuth2Middleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        token = request.headers.get("Authorization")
        if token:
            token = token.split(" ")[1]
            response = requests.get(
                settings.GITHUB_API_URL,
                headers={'Authorization': f'Bearer {token}'},
            )
            user = response.json()
            if 'error' in user:
                raise HTTPException(status_code=401, detail="Invalid token")
            request.state.user = user
        else:
            raise HTTPException(status_code=401, detail="Authorization header missing")
        response = await call_next(request)
        return response

将中间件添加到 FastAPI 应用程序 (main.py):

from fastapi.middleware import Middleware
from middleware import OAuth2Middleware


middleware = [
    Middleware(OAuth2Middleware)
]


app = FastAPI(middleware=middleware)


@app.get("/users/me")
def read_users_me(request: Request):
    return request.state.user

演示 4:使用第三方库 (Authlib)

Authlib 是一个功能强大的 Python 处理 OAuth2 库。以下是如何将 Authlib 与 FastAPI 结合使用。

安装 Authlib:

pip install authlib

将 Authlib 与 FastAPI 集成 (main.py):

from fastapi import FastAPI, Depends
from authlib.integrations.starlette_client import OAuth


app = FastAPI()


oauth = OAuth()
oauth.register(
    name='github',
    client_id=settings.GITHUB_CLIENT_ID,
    client_secret=settings.GITHUB_CLIENT_SECRET,
    authorize_url=settings.GITHUB_AUTHORIZE_URL,
    authorize_params=None,
    access_token_url=settings.GITHUB_TOKEN_URL,
    access_token_params=None,
    refresh_token_url=None,
    redirect_uri='http://localhost:8000/auth',
    client_kwargs={'scope': 'read:user'}
)


@app.route('/login')
async def login(request: Request):
    redirect_uri = url_for('auth', _external=True)
    return await oauth.github.authorize_redirect(request, redirect_uri)


@app.route('/auth')
async def auth(request: Request):
    token = await oauth.github.authorize_access_token(request)
    user = await oauth.github.parse_id_token(request, token)
    return {"user": user}


@app.get("/users/me")
def read_users_me(token: str = Depends(oauth2_scheme)):
    return {"message": "User information would be fetched from token"}

这些高级演示展示了 FastAPI 中 OAuth2 集成的不同方面和功能,让您全面了解如何在应用程序中实现安全的身份验证和授权机制。

借助 FastAPI 内置的 OAuth2 支持,在 FastAPI 应用程序中集成 OAuth2 非常简单。通过遵循本博客中概述的步骤,您可以安全地管理应用程序中的用户身份验证和授权。此集成提供了一种安全的方式来访问受保护的资源,而不会危及用户凭据。

写在最后:

  1. 最近的关于安全的主题,暂时告一段落。但是,其实业内人士都知道安全并不是一朝一夕的事情。通常,大多数公司都会有专门的安全团队,做专门的安全审计,包含:代码走查,代码静态扫描,白帽子和安全流程的业务培训,小编所介绍的只是这其中的一小部分而已,除了配合专业人士的工作,作为一个全栈工程师,我们自身也要有安全素养,了解基本的安全常识,掌握安全工具的使用等等
  2. 剧透下:后面还有更多精彩内容,特别是异步处理,监控和测试等,敬请期待

Tags:

本文暂时没有评论,来添加一个吧(●'◡'●)

欢迎 发表评论:

最近发表
标签列表