Proxy a remote API secured using an OAuth2 client credentials grant, exposing it locally without auth

proxy_with_oauth.py

# Taken partly from https://stackoverflow.com/a/36601467
from datetime import datetime, timedelta
import os

from authlib.integrations.httpx_client import AsyncOAuth2Client
from sanic import Sanic
from sanic.log import logger
from sanic.response import raw

# This should be the full token endpoint URL,
# e.g. for Keycloak that's https://{KEYCLOAK_BASE_URL}/realms/{REALM_NAME}/protocol/openid-connect/token
TOKEN_URL = os.environ.get("TOKEN_URL")

CLIENT_ID = os.environ.get("CLIENT_ID")
CLIENT_SECRET = os.environ.get("CLIENT_SECRET")
API_URL = os.environ.get("API_URL")

app = Sanic("ProxyApp")


@app.before_server_start
async def setup_client(app):
    client = AsyncOAuth2Client(CLIENT_ID, CLIENT_SECRET)
    app.ctx.client = client
    app.ctx.token = await client.fetch_token(TOKEN_URL, grant_type="client_credentials")
    logger.info("Token fetch complete")


async def _refresh_token():
    token_expiry = datetime.fromtimestamp(app.ctx.token["expires_at"])

    if token_expiry - timedelta(seconds=30) < datetime.now():
        logger.info("Token expired or expiring soon, refreshing")
        app.ctx.token = await app.ctx.client.fetch_token(TOKEN_URL, grant_type="client_credentials")

@app.route('/<path:path>', methods=["HEAD", "GET", "POST", "PUT", "DELETE"])
async def proxy_request(request, path):
    if path == "":
        target_url = API_URL
    else:
        target_url = f"{API_URL}/{path}"

    logger.info(f"Proxying {request.method} request to /{path} -> {target_url}")

    del (request.headers["host"])

    # Suppress any HTML render, since the links won't work
    del (request.headers["accept"])

    # Refresh the token
    await _refresh_token()

    res = await app.ctx.client.request(
        method=request.method,
        headers=request.headers,
        url=target_url,
        content=request.body,
        params=request.args
    )

    if res.status_code == 301 or res.status_code == 302:
        # Rewrite the location header to avoid redirecting upstream
        res.headers["Location"] = res.headers["Location"].replace(API_URL, "http://127.0.0.1:8000")

    return raw(res.content, res.status_code, res.headers)


if __name__ == "__main__":
    app.run(access_log=True)

requirements.txt

aiohttp
authlib
sanic