Components ⚙️

At first glance, this library might look like a grab bag of random bits and pieces. I swear, they’re all carefully designed to work together to help you enforce authorisation rules in your FastAPI application. You aren’t obliged to use these tools, but they are here to help make it easier to do the right thing and harder to accidentally do the wrong thing.

This page covers all the major public-facing components in this library, with information useful for both consumers looking to use this library and implementation details that may be relevant for developers working on the library itself.

Watchdog 🐕‍🦺

The watchdog() is your faithful friend that will watch to make sure you haven’t left any doors unlocked by mistake and bark if you have.

The first thing you should do is add it to your application like this:

from fastapi import FastAPI

from ska_aaa_authhelpers import watchdog

app = FastAPI(lifespan=watchdog())

This introduces a FastAPI lifespan event manager that runs once, at application startup time, and throws a class:SecurityHoleError exception if you have accidentally forgotten to secure any routes. In most cases, this is all you need and you can proceed from here.

If your app contains routes that absolutely must bypass any authorisation enforcement (this is discouraged, consider using Role.ANY instead), you can pass the names of the route functions to watchdog() with the allow_unsecured parameter.

from time import time

from fastapi import FastAPI
from ska_aaa_authhelpers import watchdog

app = FastAPI(
    # We want to bypass any security on the `get_time` route:
    lifespan=watchdog(allow_unsecured=['get_time'])
)

@app.get('/time')
def get_time():
    return time()

Requires() 🔒

The Requires() utility allows you to specify in broad terms the authorisation needed for each path operation in your application. You’ll need to pass one or more roles, one or more scopes, and the audience parameter expected – the audience should always be your own service’s Application ID, assigned at registration time by Microsoft Entra ID.

from typing import Annotated
from ska_aaa_authhelpers import AuthContext, Requires, Role

from .my_app import app, execute_observation, MY_APP_ID
from .my_models import Observation
from .my_scopes import RUN_OBSERVATION


@app.post("/observation/new")
def new_observation(
    data: Observation,
    auth: Annotated[
        AuthContext,
        Requires(
            audience=MY_APP_ID,
            roles={Role.LOW_TELESCOPE_OPERATOR},
            scopes={RUN_OBSERVATION},
        ),
    ],
):
    results = execute_observation(data, owner=auth.user_id)
    return results

Requires() additionally allows you to define authorisation flow from app to app, where the client is a web service with a token created through the client credentials grant. In this case, the client application has to have the role Role.APP2APP assigned to it, and you will have to pass app_ids to include the ID of the client application. If only specifying app to app requirements, no scopes are required.

from typing import Annotated
from ska_aaa_authhelpers import AuthContext, Requires, Role

from .my_app import app, execute_observation, MY_APP_ID
from .my_models import Observation
from .my_scopes import RUN_OBSERVATION
from .my_client_utils import MY_CLIENT_APP_ID


@app.post("/observation/new")
def new_observation(
    data: Observation,
    auth: Annotated[
        AuthContext,
        Requires(
            audience=MY_APP_ID,
            roles={Role.LOW_TELESCOPE_OPERATOR, Role.APP2APP},
            scopes={RUN_OBSERVATION},
            app_ids={MY_CLIENT_APP_ID}
        ),
    ],
):
    results = execute_observation(data, owner=auth.user_id)
    return results

Under the hood, Requires()is a utility that automatically creates a FastAPI Security() dependency and also the security scheme wrapped inside that dependency.

AuthContext 🪪

The AuthContext provided to your view functions by Requires() is a Pydantic model that represents all the authorisation information we have about this request. Your application should further evaluate these parameters to make a decision about whether to accept or deny the request.

pydantic model AuthContext[source]

field user_id: str [Required]: An opaque (UUID) account identifier associated with a request. This value is stable for one account and can be used as a database key.This may represent a human user or an automated app-user

field principals: frozenset[str] [Required]: A set of security principals associated with this request. This set includes one or more account IDs and zero or more group IDs. Use this field to help make authorization judgements.

field groups: frozenset[str] [Required]: An opaque set of group UUIDs of which the user is a member. Typically you would use principals unless you are specifically interested in group membership.

field scopes: frozenset[str] [Required]: A set of OAuth2 scopes associated with the request, showing the user has granted these permissions to this client. Use this field to help make authorization judgements.

field roles: frozenset[Role] [Required]: A set of Roles granted to this user account by the system administrators. Use this field to help make authorization judgements.

field audience: str [Required]: Identifier by which the authorization server knows this service. Consumers should check this to verify this request has been sent to the right place. This should always be ‘us’.

field token_claims: dict[str, JsonValue] [Required]: All the claims provided in the access token by the Authorsation Provider.

field access_token: str [Required]: The original access token as sent by the client.May be used to revalidate for extra assurance.

field trace: str [Optional]: Used to reconcile together log entries spanning a single user action.

You may notice that many of the fields are are frozensets. It’s often a good approach to reason about authorisation in terms of set operations: is this user a member of a group? Is there an intersection between the scopes this client has been granted and the scopes required to perform an action? Is there an intersection between the principals of this request and the owners of a resource?

Role 👤

Conceptually, Roles are globally-defined attributes assigned to users based on the work they are doing at SKAO. In terms of this library Role is just an enum designed to be used in Requires() and your view functions.

class Role(value)[source]

Enum containing the IDs of special role-granting Groups defined in MS Entra ID.

ANY = 'ANY': Generic role automatically granted to all requests in addition to any specific roles derived from the access token.

APP2APP = 'APP2APP': Role for apps using the OAuth2 client credentials grant to make automated requests, rather than acting on behalf of the user.

LOW_TELESCOPE_OPERATOR = '47a4523e-6fbe-441d-914c-14c6ca01922e': Schedules and executes an observing session on SKA-Low and ensure the correctness of science experiment data by controlling and monitoring telescope subsystems.

MID_TELESCOPE_OPERATOR = '405e6fd0-a361-447f-bb5e-9f6d03b2b474': Schedules and executes an observing session on SKA-Mid and ensure the correctness of science experiment data by controlling and monitoring telescope subsystems.

OPERATIONS_SCIENTIST = 'bdc4b862-882c-461a-a2a4-6f0ac89910c4': Oversees the creation of the Project and SBDefinitions to achieve the Science aim.

OPS_PROPOSAL_ADMIN = 'ce3627de-8ec2-4a35-ab1e-300eec6a0a50': Individual who is able to perform all activities associated with proposals

OPS_REVIEWER_SCIENCE = '05883c37-b723-4b63-9216-0a789a61cb07': Individual who is able to provide a review of the science behind a proposal

OPS_REVIEWER_TECHNICAL = '4c45b2ea-1b56-4b2d-b209-8d970b4e39dc': Individual who is able to validate the feasibility of the technical aspects required for a proposal

PROD_SW_ENGINEER = '69d67c05-f536-481e-98b2-35c7f2254592': Responsible for debugging issues live in a production environment

SW_ENGINEER = '2d650a1e-dc34-4452-ab3b-15175d59e5d0': Develops, maintains, and debugs software applications within SKAO. Member of any of the software development teams.

As a matter of implementation, Roles are managed as Entra Groups: Anyone who is in the software engineers group is assigned the SW_ENGINEER role. The point of this enum is basically to hardcode the special role-granting groups so that devs can work with nice enum names instead of opaque UUIDs. It’s also used internally to populate the AuthContext.roles field after decoding the access token.

AuthFailError() ⛔

Pretty much does what it says on the tin. This is raised by AuthHelpers itself when the access token claims fail to meet the application’s policies as declared in Requires(). Applications may also raise it if the app’s own internal authorisation logic is not satisfied, triggering an HTTP 403: Forbidden response to the client.

raise AuthFailError("Only authorised parties allowed.")

This is a subclass of FastAPI’s HTTPException that adds a log entry to the audit log when requests are rejected.

AuditLogFilter 📝

The AuditLogFilter is a Python logging filter designed to be used as part of the ska-ser-logging infrastructure.

from ska_ser_logging import configure_logging
from ska_aaa_authhelpers import AuditLogFilter

configure_logging(level="WARNING", tags_filter=AuditLogFilter)

Despite its name, the AuditLogFilter doesn’t actually filter out any log entries, instead it adds tags for the user_id and trace fields, allowing us to tie log records back to a specific user request and authorisation flow. This is a standard recognised usage for filter objects in Python logging.

Internally, it relies on starlette-context to retrieve the AuthContext from a request-global context manager.

mint_test_token() 🗝️

This is a utility for generating access tokens signed with built-in test keys provided by the library. Because Microsoft controls the private keys used to sign Entra ID access tokens, we can’t simply generate our own tokens that will pass signature verification using the default MS Entra public keys. Instead, for testing purposes, we use our own private key to sign tokens. You can then pass these tokens in the HTTP Authorization header to your application under test.

from fastapi.testclient import TestClient
from ska_aaa_authhelpers.test_helpers import mint_test_token

token = mint_test_token()

client = TestClient(app, headers={"Authorization": f"Bearer {token}"})

This function can be called without any arguments and it will create a token associated to a fake TEST_USER, issued for default TEST_SCOPES, TEST_ROLES and TEST_GROUPS etc. You can pass specific arguments to override any of these token claims if your test scenarios rely on particular users or groups, and you’ll likely need to include scopes and roles relevant for your own application.

Internally, mint_test_token() directly calls the joserfc library to encode and sign a JWT access token with a similar set of claims to those issued by Microsoft.

See Testing Your Applications for more usage examples.

monkeypatch_pubkeys 🙈

On the flip side, once we’ve been minting tokens signed with our own test private key, we’ll need to verify them using our test public keys instead of Microsoft’s pubkeys. monkeypatch_pubkeys is intended to be used inside a pytest test fixture to monkeypatch all Requires() instances replacing the DEFAULT_PUBLIC_KEYS (i.e. Microsoft’s keys) with TEST_PUBLIC_KEYS. You can put this in your conftest.py:

import pytest
from ska_aaa_authhelpers.test_helpers import monkeypatch_pubkeys

# put this in conftest.py
@pytest.fixture(scope="session", autouse=True)
def patch_pubkeys():
    monkeypatch_pubkeys()

The guts of this implementation is really gnarly: it uses gc.get_objects() to fish every single live object in the Python interpreter out of memory, filters with isinstance() to find internal TokenScheme() instances that were generated by Requires()and any functools.partials that could be used during a test session to create more. Then, it loops over everything replacing the keys attributes. It does not bother trying to restore the original keys, because the assumption is that this fixture will be enabled globally for the whole test session.