Search code examples
pythonauthenticationmicroservicesbasic-authenticationfastapi

How to use HTTP Basic Auth as separate FastAPI service?


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}

Solution

  • 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)):