How to use HTTP Basic Auth as separate FastAPI service?

Question

What I want to achieve? Have one service responsible for HTTP Basic Auth (access) and two services (a, b) where some endpoints are protected by access service.

Why? In scenario where there will be much more services with protected endpoints to not duplicate authorize function in each service. Also to do modification in one place in case of changing to OAuth2 (maybe in future).

What I did? I followed guide on official website and created example service which works totally fine.

Problem occurs when I try to move authorization to separate service and then use it within few other services with protected endpoints. I can't figure out how to do it. Could you please help me out?

I have tried different functions setup. Nothing helped, so far my code looks like this:

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

a-service

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

Thanks to Soumojit Ghosh answer and FastAPI Issue 1037 I figured out how should I modify my code. a-service after changes:

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}

Now header can be sent through SwaggerUI. Click Authorize and then enter it in Value field. To generate your header from login and password you can use for example this tool. It will look like: Basic YWRtaW46cGFzc3dvcmQ=.

Answer 2

The issue here is that, when you are calling Service_A with credentials it's making a call to the Access_Service in the has_access() function.

If you look closely,

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

You are simply making a GET call without forwarding the credentials as headers for this request to the Access_Service.

Rewrite your has_access() in all the services to

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.',
        )

Inside your access service you have mistakenly typed True as true,

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

I have cloned your repo and tested it, it's working now. Please check and confirm.

[EDIT] Swagger does not allow authorization header for basic auth (https://github.com/tiangolo/fastapi/issues/612)

Work-Around (not recommended)

from fastapi.security import HTTPBasic, HTTPBasicCredentials

security = HTTPBasic()

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