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

Question

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

为什么？ 在每个服务中会有更多具有受保护端点的服务不重复 authorize function 的情况下。也可以在一个地方进行修改，以防更改为 OAuth2（可能在将来）。

我做了什么？ 我按照 official website and created example service 上的指南进行操作，效果非常好。

问题发生在我尝试移动授权以分离service然后在其他几个具有受保护端点的服务。我不知道该怎么做。你能帮帮我吗？

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

访问服务

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 调用，而没有将此请求的凭据作为 header 转发给 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 不允许授权 header 进行基本身份验证 (https://github.com/tiangolo/fastapi/issues/612)

Work-Around（不推荐）

from fastapi.security import HTTPBasic, HTTPBasicCredentials

security = HTTPBasic()

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

Answer 2

感谢 Soumojit Ghosh 的回答和 FastAPI Issue 1037 我想出了应该如何修改我的代码。 a-service 修改后：

import httpx
import os

from fastapi import Depends, FastAPI, Header, HTTPException, status
from typing import Optional
from fastapi.security import HTTPBasicCredentials, HTTPBearer

security = HTTPBearer()

ACCESS_SERVICE_URL = os.getenv('ACCESS_SERVICE_URL')

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


def has_access(credentials: HTTPBasicCredentials = Depends(security)):
    response = httpx.get(os.getenv('ACCESS_SERVICE_URL'), headers={'Authorization': credentials.credentials})
    if response.status_code == 401:
        raise HTTPException(status_code=401)


@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}

现在header可以通过SwaggerUI发送了。单击授权，然后在值字段中输入它。要从登录名和密码生成 header，您可以使用例如 this tool。它看起来像：Basic YWRtaW46cGFzc3dvcmQ=.

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

How to use HTTP Basic Auth as separate FastAPI service?

python

authentication

basic-authentication

microservices

fastapi