easy-oauth

Small package to add OAuth-based authentication to a Starlette/FastAPI app. Users may also retrieve a token to authenticate themselves.

Install

uv add git+https://github.com/mila-iqia/easy-oauth@v0.0.2

Usage

If you want to authenticate through Google, first you will need to create a project in GCP and get a client_id and client_secret from the console. Then you can do it like this:

from easy_oauth import OAuthManager, CapabilitySet

oauth = OAuthManager(
    # This page describes where the endpoint urls are defined
    server_metadata_url="https://accounts.google.com/.well-known/openid-configuration",
    # A secret key to encrypt the session and tokens, you can generate it yourself
    secret_key=SECRET_KEY,
    # The client id from Google Console
    client_id=CLIENT_ID,
    # The client secret from Google Console
    client_secret=CLIENT_SECRET,
    # Arguments to the auth request, you can just use what's below
    client_kwargs={
        "scope": "openid email",
        "prompt": "select_account",
    }
    # Set of capabilities that can be assigned to users
    capabilities=CapabilitySet(
        graph={
            # Basic capability
            "read": [],
            # write, also implies read
            "write": ["read"],
            # moderate, also implies read and write
            "moderate": ["read", "write"],
            # announce, also implies read and write
            "announce": ["read", "write"],
            # "user_management" is the capability needed to set the capabilities of
            # users.
            "user_management": [],
        },
        # Create the "admin" capability that has every other capability
        auto_admin=True,
        # File where each user's capability is stored
        user_file="caps.yaml",
        # Capabilities granted to all authenticated users
        default_capabilities=["read"],
        # Capabilities granted to unauthenticated users (guests)
        guest_capabilities=[],
    ),
    # If you want routes to be at /api/v1/login etc., put "/api/v1" here
    prefix="",
)

app = FastAPI()

oauth.install(app)

Here is an example of a user capability file:

your-email@you.com:
  - admin
friend@rainbows.com:
  - moderate
  - announce
pierre-jean-jacques@youhou.fr:
  - read
hateful-being@cornhole.co: []

In order to require a certain capability for a given route, you can declare it like this in FastAPI:

@app.get("/shout")
async def route_shout(
    request: Request,
    message: str,
    email: str = Depends(oauth.get_email_capability("announce", redirect=True)),
):
    return PlainTextResponse(f"{email} shouts {message!r}")

If redirect=True in get_email_capability, then the browser will redirect to the login page if the user is not logged in, then it will redirect back to the original page.

Token workflow

First, the user should point their browser to the /token endpoint. This will prompt them to log in and it will display a token. Copy it.

Then you can use use Bearer authentication with the token. That is to say, the Authorization header should be set to Bearer INSERT_TOKEN_HERE. Using httpx, for example (it should work the same with requests):

# Unauthorized access
assert httpx.get(f"{app_url}/something").status_code == 401

# Authorized access
assert (
    httpx.get(f"{app_url}/something", headers={"Authorization": f"Bearer {token}"}).status_code
    == 200
)

Revoking tokens

Tokens are revoked with POST /revoke_token, or the interface at GET /revoke_token, which revokes the underlying refresh token at the OAuth provider (RFC 7009, using its revocation_endpoint). The token to revoke can be passed in the Authorization header or as {"token": "<token>"} in the request body; if no token is passed, the refresh token of the current session is revoked:

# Revoke a specific token
httpx.post(f"{app_url}/revoke_token", json={"token": token})

# Revoke the session's own token (e.g. from the user's browser)
httpx.post(f"{app_url}/revoke_token")

Things to be aware of:

• Revocation happens at the provider, so every token derived from that refresh token is affected (Google generally reuses one refresh token per user and client). Existing browser sessions keep working until logout, since they do not consult the provider, but they will not be able to produce new tokens.
• Service tokens are a local construct and cannot be revoked this way; remove the service account's capabilities instead (that takes effect immediately).
• Users can also revoke the app's access from their Google account page.

Service accounts

Service accounts are non-human accounts intended for automated scripts or bots. Their usernames must end with @service (e.g. bot@service).

Creating a service token

An admin (a user with the user_management capability) first assigns capabilities to the service account, then fetches a token for it by hitting GET /token?user=<service-account> while authenticated:

import httpx

# Assign capabilities
httpx.post(
    f"{app_url}/manage_capabilities/set",
    json={"email": "bot@service", "capabilities": ["baker"]},
    headers=admin_headers,
)

# Get a service token
resp = httpx.get(f"{app_url}/token", params={"user": "bot@service"}, headers=admin_headers)
service_token = resp.json()["service_token"]

The response includes the token, the username, and the capabilities baked into it:

{"service_token": "<encrypted_token>", "user": "bot@service", "capabilities": ["baker"]}

Using a service token

Pass the service token as a Bearer token, exactly like a regular user token:

bot_headers = {"Authorization": f"Bearer {service_token}"}
httpx.get(f"{app_url}/bake", params={"food": "croissant"}, headers=bot_headers)

Capability semantics

Access is granted only when both of the following are true:

1. The requested capability is covered by the capabilities embedded in the token (snapshotted at creation time, including implied capabilities).
2. The service account currently has the requested capability in the database.

This means:

• Adding a new capability to the service account requires generating a fresh token before it takes effect.
• Removing a capability from the database immediately revokes access, regardless of what the token says.

Reading configuration from a file

The configuration for the above OAuthManager can be written in a file, like this:

server_metadata_url: https://accounts.google.com/.well-known/openid-configuration
secret_key: "<SECRET_KEY>"
client_id: "<CLIENT_ID>"
client_secret: "<CLIENT_SECRET>"
client_kwargs:
  scope: openid email
  prompt: select_account
capabilities:
  graph:
    read: []
    write: [read]
    moderate: [read, write]
    announce: [read, write]
    user_management: []
  auto_admin: true
  user_file: caps.yaml
  default_capabilities: [read]
  guest_capabilities: []
prefix: ""

And instantiated like this:

from serieux import deserialize

oauth = deserialize(OAuthManager, Path("config.yaml"))

Of course, you can nest that configuration within a larger configuration.

Encrypting the configuration

The secrets written in the config file can be encrypted using serieux (The -m option must point to the type of the root of the configuration using the syntax module:symbol, in this case it is simply easy_oauth:OAuthManager):

export SERIEUX_PASSWORD="change_me!!1"
serieux patch -m easy_oauth:OAuthManager -f config.yaml

You must then modify the instantiation code like this:

import os
from serieux import deserialize
from serieux.features.encrypt import EncryptionKey

oauth = deserialize(
    OAuthManager, Path("config.yaml"), EncryptionKey(os.getenv("SERIEUX_PASSWORD"))
)

Routes

The OAuthManager automatically adds the following routes when installed on your Starlette/FastAPI application:

Authentication Routes

• GET /login

  • Initiates the OAuth login flow
  • Clears the current session and redirects to the OAuth provider
  • Query parameters:
    • redirect (optional): Name of the auth callback route (default: auth)
    • offline_token=true (optional): Request a refresh token with offline access
  • Stores the original URL in session to redirect back after authentication

• GET /auth

  • OAuth callback route that handles the authorization code
  • Exchanges the authorization code for tokens and stores user information in the session
  • Redirects to the original URL (default: /)

• GET /token

  • Returns an encrypted refresh token for the authenticated user
  • Response: {"refresh_token": "<encrypted_token>"}
  • Query parameters:
    • user (optional): If set to a @service account name, returns a service token for that account instead (requires user management capability). Response: {"service_token": "<encrypted_token>", "user": "<service-account>", "capabilities": [...]}

• GET /logout

  • Clears the user session and redirects to /

Token Revocation Route

• GET/POST /revoke_token
  • Revokes a refresh token at the OAuth provider so that it can no longer be used to authenticate. With GET, a minimal HTML form is displayed; with POST, the token to revoke can be passed in the Authorization header, in a JSON body as {"token": "<token>"}, or in a form field named token. If no (or a blank) token is provided, the refresh token of the current session is revoked (and removed from the session)
  • The revocation request is authenticated with the app's own client credentials, so it only affects this app's grant at the provider
  • Response: {"status": "ok"}

Capability Management Routes

• GET /manage_capabilities/list
  • Lists capabilities for a user
  • Query parameters:
    • email (optional): Email address to query (defaults to current user)
  • Requires user management capability if querying another user's capabilities
  • Response: {"status": "ok", "email": "<email>", "capabilities": [...]}

The following routes are only added if there is a user_management capability:

• POST /manage_capabilities/add

  • Adds a capability to a user
  • Requires user management capability
  • Request body: {"email": "<email>", "capability": "<capability_name>"}
  • Response: {"status": "ok", "email": "<email>", "capabilities": [...]}

• POST /manage_capabilities/remove

  • Removes a capability from a user
  • Requires user management capability
  • Request body: {"email": "<email>", "capability": "<capability_name>"}
  • Response: {"status": "ok", "email": "<email>", "capabilities": [...]}

• POST /manage_capabilities/set

  • Sets the complete capability set for a user (replaces existing capabilities)
  • Requires user management capability
  • Request body: {"email": "<email>", "capabilities": ["<cap1>", "<cap2>", ...]}
  • Response: {"status": "ok", "email": "<email>", "capabilities": [...]}

Testing

For testing, easy_oauth defines a mock OAuth server that always logs you in unconditionally as test@example.com by default. That way you don't need a browser or any secrets to test things.

uvicorn easy_oauth.testing.oauth_mock:app

To set the email address the mock OAuth server with authentify all requests as, send a POST request with JSON data like this:

curl -X POST -H "Content-Type: application/json" -d '{"email": "a@b.c"}' http://127.0.0.1:8000/set_email

To use it with easy_oauth, set server_metadata_url to http://127.0.0.1:8000/.well-known/openid-configuration (depending on the host and port).

Fixtures

easy-oauth provides the OAuthMock and AppTester classes to make testing easier. Here is a very simple example of how to use them:

from easy_oauth.testing.utils import AppTester, OAuthMock


@pytest.fixture(scope="session")
def oauth_mock():
    # Start one mock oauth server for the session. It's important that the
    # OAUTH_PORT conforms to the server_metadata_url you configure the test app
    # with
    with OAuthMock(port=OAUTH_PORT) as oauth:
        yield oauth


@pytest.fixture(scope="session")
def app(oauth_mock):
    # This doesn't have to be session-scoped, but if your app is read-only it may
    # as well be.
    with AppTester(your_app, oauth_mock) as appt:
        yield appt


def test_view_payroll(app):
    # Use app.client to pretend to be various users
    guest = app.client()
    user = app.client("simple.user@website.web")
    accountant = app.client("mr.bean@website.web")
    admin = app.client("admin@website.web")

    # Guests are not authentified (so we expect HTTP error 401)
    guest.get("/payroll/view", expect=401)
    # Normal users are unauthorized to view the payroll
    user.get("/payroll/view", expect=403)
    # Accountants and admins are authorized
    accountant.get("/payroll/view", expect=200)
    admin.get("/payroll/view", expect=200)

TODO

There are a few things that need to be done in the future:

• Users with user_management capability should only be able to add/remove capabilities that they have.