The problem is the following: We need to invoke a Fast API app running on GC Cloud Run from some other GC service (eg PubSub in our own case). However we have setup OAuth protection of our API routes (eg using Clerk in our own case) and backend services cannot signin to our OAuth provider.

The solution is to allow for multiple authentication methods at the level of our REST API, something that FastAPI supports.

Initially our app is setup like this:

app = FastAPI()

protected = APIRouter(
    tags=["Protected API"],
    dependencies=[Depends(clerk_auth)]
)

@protected.post("/private/func")
def func(msg: str):
    """
    Example:
        msg = 'a dummy payload'
    """
    return {"message": msg}

app.include_router(protected)
    
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=PORT, timeout_keep_alive=60)

Where clerk_oauthi s defined as follow:

from fastapi_clerk_auth import (
    ClerkConfig, ClerkHTTPBearer,
    HTTPAuthorizationCredentials as ClerkAuthCredentials
)

clerk_config = ClerkConfig(jwks_url=f"https://{settings.CLERK_DOMAIN}/.well-known/jwks.json")

clerk_auth_guard = ClerkHTTPBearer(config=clerk_config, auto_error=False)

async def clerk_auth(credentials: ClerkAuthCredentials | None = Depends(clerk_auth_guard)):
    token = credentials.credentials
    logging.info(f"Received Bearer token: {token}")
    return credentials

We are going to replace the dependencies in the APIRoute by a function that supports both Clerk and GCP OAuth.

First we need to configure our service (PubSub here).

  • We use IAM rules to grant our service (PubSub here) the right to invoke our Cloud Run instance.
  • We configure our service so that it sends credential tokens to the target (ie our Fast API app) url and we specify the service account that will be associated to the token.
  • We also specify an audience value that will be added to the token (cf aud field) which will be used as extra validation.

These settings can be see on the following screenshot with fields Push authentication, Push endpoint and Service account:

Using the web console we can check that our Cloud Run app is allowed to receive internal traffic.

Now we are ready to implement the FastAPI validation of the tokens being sent by PubSub. To do this we use the python package google-auth.

from fastapi import Request, HTTPException, Depends
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
import google.oauth2.id_token
import google.auth.transport.requests

http_bearer = HTTPBearer(auto_error=False)

async def gcp_auth(
	request: Request,
	gcp_credentials: HTTPAuthorizationCredentials = Depends(http_bearer)
) -> Optional[Dict[str, str]]:
    if not gcp_credentials:
        return None  # Let Clerk auth handle it
    
    try:
        # Verify Google-issued ID token
        token = gcp_credentials.credentials
        id_info = google.oauth2.id_token.verify_token(
            token,
            google.auth.transport.requests.Request(),
            audience=settings.GC_OAUTH_AUDIENCE,
        )
        
        # Validate service account details
        if id_info.get('iss') not in ['accounts.google.com', 'https://accounts.google.com']:
            raise ValueError('gcp_auth: invalid issuer')
            
        # Ensure service account belongs to your project
        sa_email = id_info.get('email', '')
        if not sa_email.endswith(f'@{settings.GC_PROJECT}.iam.gserviceaccount.com'):
            raise ValueError('gcp_auth: service account not in project')

        return {'service_account': sa_email, 'auth_type': 'gcp'}
    except Exception as _err:
        return None

Finally we can combine together both the Clerk auth and GCP auth methods into a single function to use in our APIRouter:

from fastapi_clerk_auth import HTTPAuthorizationCredentials as ClerkAuthCredentials

async def combined_auth(
    gcp_result: Optional[Dict[str, str]] = Depends(gcp_auth),
    clerk_credentials: Optional[ClerkAuthCredentials] = Depends(clerk_auth_guard)
):
    """
    Combined authentication that accepts either GCP service or Clerk credentials
    """
    if gcp_result:
        return gcp_result
    elif clerk_credentials:
        logging.info(f'combined_auth: Clerk block')
        # The ClerkHTTPBearer dependency already verified the token
        return clerk_credentials
    else:
        raise HTTPException(status_code=401, detail='combined_auth: invalid authentication credentials')

Final remark, the full code is available in this github repo: https://github.com/aleascope/fastapi-clerk-auth