如何使用 HTTP Basic Auth 作为单独的 FastAPI 服务？

Question

我想实现什么目标？有一项负责 HTTP 基本身份验证（访问）的服务和两项服务（a、b），其中某些端点受到访问服务的保护。

为什么？在存在更多具有受保护端点的服务的情况下，每个服务中都不会重复授权功能。还要在一处进行修改，以防更改为 OAuth2（也许将来）。

我做了什么？ 我按照官方网站上的指南创建了示例服务，该服务运行得很好。

当我尝试将授权移至单独的服务，然后在具有受保护端点的少数其他服务中使用它时，就会出现问题。我不知道该怎么做。你能帮我一下吗？

我尝试过不同的功能设置。没有任何帮助，到目前为止我的代码如下所示：

访问服务

import os
import secrets

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials

security = HTTPBasic()


def authorize(credentials: HTTPBasicCredentials = Depends(security)):
    is_user_ok = secrets.compare_digest(credentials.username, os.getenv('LOGIN'))
    is_pass_ok = secrets.compare_digest(credentials.password, os.getenv('PASSWORD'))

    if not (is_user_ok and is_pass_ok):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail='Incorrect email or password.',
            headers={'WWW-Authenticate': 'Basic'},
        )


app = FastAPI(openapi_url="/api/access/openapi.json", docs_url="/api/access/docs")


@app.get('/api/access/auth', dependencies=[Depends(authorize)])
def auth():
    return {"Granted": True}

服务

import httpx
import os

from fastapi import Depends, FastAPI, HTTPException, status

ACCESS_SERVICE_URL = os.getenv('ACCESS_SERVICE_URL')

app = FastAPI(openapi_url="/api/a/openapi.json", docs_url="/api/a/docs")


def has_access():
    result = httpx.get(os.getenv('ACCESS_SERVICE_URL'))
    if result.status_code == 401:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail='No access to resource. Login first.',
        )


@app.get('/api/a/unprotected_a')
async def unprotected_a():
    return {"Protected": False}


@app.get('/api/a/protected_a', dependencies=[Depends(has_access)])
async def protected_a():
    return {"Protected": True}


@app.get('/api/a/protected_b', dependencies=[Depends(has_access)])
async def protected_b():
    return {"Protected": True}

Answer 1

这里的问题是，当您使用凭据调用 Service_A 时，它会调用 has_access() 函数中的 Access_Service 。

如果你仔细观察，

result = httpx.get(os.getenv('ACCESS_SERVICE_URL'))

您只需进行 GET 调用，而无需将凭据作为此请求的标头转发到 Access_Service。

将所有服务中的 has_access() 重写为

from typing import Optional
from fastapi import Header 

def has_access(authorization: Optional[str] = Header(None)):
    if not authorization:
    raise HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail='No access to resource. Credentials missing!',
    )
    headers = {'Authorization': authorization}
    result = httpx.get(os.getenv('ACCESS_SERVICE_URL'), headers=headers)
    if result.status_code == 401:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail='No access to resource. Login first.',
        )

在您的访问服务中，您错误地将 True 输入为 true，

@app.get('/api/access/auth', dependencies=[Depends(authorize)])
def auth():
    return {"Granted": True} 

我已经克隆了你的存储库并对其进行了测试，它现在可以工作了。请检查并确认。

[编辑] Swagger 不允许使用授权标头进行基本身份验证（https://github.com/tiangolo/fastapi/issues/612）

解决方法（不推荐）

from fastapi.security import HTTPBasic, HTTPBasicCredentials

security = HTTPBasic()

def has_access(credentials: HTTPBasicCredentials = Depends(security), authorization: Optional[str] = Header(None)):