FastAPI Auth Tasks¤

task api names: bearer_token_store, bearer_token_list, bearer_token_delete, bearer_token_check

FastAPI service supports bearer token REST API authentication. To handle tokens lifecycle a number of FastAPI Service methods created allowing to store, delete, list and check API tokens.

Inputs¤

Parameter	Required	Description
`username`	Required for create and username-scoped delete/list	User name associated with one or more bearer tokens.
`token`	Required for check and token-scoped delete	Bearer token value. If omitted on create, the worker generates one.
`expire`	No	Token expiration time in seconds. If omitted, the token does not expire.
`workers`	No	FastAPI workers to target. Defaults to all workers.

Output¤

Create, delete, and check tasks return booleans per worker. List tasks return stored token records with username, token, age, creation time, and expiration time.

Examples¤

Example

CLIPython

Store authentication token in FastAPI service database:

nf# fastapi auth create-token username foobar token f343ff34r3fg4g5g34gf34g34g3g34g4 expire 3600
{
    "fastapi-worker-1": true
}
nf#

expire is optional and indicates token expiration time in seconds. If no expire argument is provided, the token does not expire. Multiple tokens can be stored for any given user.

List stored tokens for a specific user:

nf# fastapi auth list-tokens username foobar
 worker            username  token                             age             creation                    expires
 fastapi-worker-1  foobar    f343ff34r3fg4g5g34gf34g34g3g34g4  0:01:29.688340  2025-02-16 20:08:51.914919  2025-02-16 21:08:51.914919
 fastapi-worker-1  foobar    888945f96b824bf1b4358de790c452b6  8:08:51.548662  2025-02-16 12:01:30.054597  None
nf#

List all stored tokens:

nf# fastapi auth list-tokens
 worker            username  token                             age             creation                    expires
 fastapi-worker-1  pytest    11111111111111111111111111111111  1:26:18.492274  2025-02-16 18:44:18.124019  None
 fastapi-worker-1  foobar    f343ff34r3fg4g5g34gf34g34g3g34g4  0:01:44.701374  2025-02-16 20:08:51.914919  2025-02-16 21:08:51.914919
nf#

Delete a specific token:

nf# fastapi auth delete-token token 888945f96b824bf1b4358de790c452b6
{
    "fastapi-worker-1": true
}
nf#

Delete all tokens for a given user:

nf# fastapi auth delete-token username foo
{
    "fastapi-worker-1": true
}
nf#

Check if a token is valid:

nf# fastapi auth check-token token 888945f96b824bf1b4358de790c452b6
{
    "fastapi-worker-1": false
}
nf# fastapi auth check-token token f343ff34r3fg4g5g34gf34g34g3g34g4
{
    "fastapi-worker-1": true
}
nf#

Context manager - create and list tokens:

import pprint

from norfab.core.nfapi import NorFab

with NorFab(inventory="inventory.yaml") as nf:
    client = nf.make_client()

    client.run_job(
        service="fastapi",
        task="bearer_token_store",
        kwargs={
            "username": "foobar",
            "token": "secret-token",
            "expire": 3600,
        },
        workers="all",
    )

    result = client.run_job(
        service="fastapi",
        task="bearer_token_list",
        kwargs={"username": "foobar"},
        workers="all",
    )

    pprint.pprint(result)

Direct lifecycle - check and delete a token:

import pprint

from norfab.core.nfapi import NorFab

nf = NorFab(inventory="inventory.yaml")
try:
    nf.start()
    client = nf.make_client()

    check_result = client.run_job(
        service="fastapi",
        task="bearer_token_check",
        kwargs={"token": "secret-token"},
        workers="all",
    )
    delete_result = client.run_job(
        service="fastapi",
        task="bearer_token_delete",
        kwargs={"token": "secret-token"},
        workers="all",
    )

    pprint.pprint(check_result)
    pprint.pprint(delete_result)
finally:
    nf.destroy()

NORFAB FastAPI Service Auth Tasks Command Shell Reference¤

NorFab shell supports these command options for FastAPI auth tasks:

nf# man tree fastapi.auth
root
└── fastapi:    FastAPI service
    └── auth:    Manage auth tokens
        ├── create-token:    Create authentication token
        │   ├── timeout:    Job timeout
        │   ├── workers:    Filter worker to target, default 'all'
        │   ├── token:    Token string to store, autogenerate if not given
        │   ├── *username:    Name of the user to store token for
        │   └── expire:    Seconds before token expire
        ├── list-tokens:    Retrieve authentication tokens
        │   ├── timeout:    Job timeout
        │   ├── workers:    Filter worker to target, default 'all'
        │   └── username:    Name of the user to list tokens for
        ├── delete-token:    Delete existing authentication token
        │   ├── timeout:    Job timeout
        │   ├── workers:    Filter worker to target, default 'all'
        │   ├── username:    Name of the user to delete tokens for
        │   └── token:    Token string to delete
        └── check-token:    Check if given token valid
            ├── timeout:    Job timeout
            ├── workers:    Filter worker to target, default 'all'
            └── *token:    Token string to check
nf#

Python API Reference¤

bearer_token_store¤

Method to store a bearer token in the database.

This method stores a bearer token associated with a username in the cache.

If an expiration time is not provided, it retrieves the default token TTL from the FastAPI inventory configuration.

Parameters:

Name	Type	Description	Default
`username`	`str`	str - The name of the user to store the token for.	required
`token`	`str`	str - The token string to store.	required
`expire`	`int`	int, optional - The number of seconds before the token expires.	`None`

Returns:

Type	Description
`Result`	bool - Returns True if the token is successfully stored.

Source code in norfab\workers\fastapi_worker\fastapi_worker.py

@Task(input=BearerTokenStoreInput, output=BoolResult, fastapi=False, mcp=False)
def bearer_token_store(
    self, job: Job, username: str, token: str, expire: int = None
) -> Result:
    """
    Method to store a bearer token in the database.

    This method stores a bearer token associated with a username in the cache.

    If an expiration time is not provided, it retrieves the default token TTL
    from the FastAPI inventory configuration.

    Args:
        username: str - The name of the user to store the token for.
        token: str - The token string to store.
        expire: int, optional - The number of seconds before the token expires.

    Returns:
        bool - Returns True if the token is successfully stored.
    """
    expire = expire or self.fastapi_inventory.get("auth_bearer", {}).get(
        "token_ttl", expire
    )
    self.cache.expire()
    cache_key = f"bearer_token::{token}"
    if cache_key in self.cache:
        user_token = self.cache.get(cache_key)
    else:
        user_token = {
            "token": token,
            "username": username,
            "created": str(datetime.now()),
        }
    self.cache.set(cache_key, user_token, expire=expire, tag=username)

    return Result(task=f"{self.name}:bearer_token_store", result=True)

bearer_token_delete¤

Deletes a bearer token from the cache. This method removes a bearer token from the cache based on either the token itself or the associated username.

If a token is provided, it will be removed directly. If a username is provided, all tokens associated with that username will be evicted from the cache.

Parameters:

Name	Type	Description	Default
`username`	`str`	The username associated with the token(s) to be removed. Defaults to None.	`None`
`token`	`str`	The bearer token to be removed. Defaults to None.	`None`

Returns:

Name	Type	Description
`bool`	`Result`	True if the operation was successful, otherwise raises an exception.

Raises:

Type	Description
`RuntimeError`	If the token removal from the cache fails.
`Exception`	If neither username nor token is provided.

Source code in norfab\workers\fastapi_worker\fastapi_worker.py

@Task(input=BearerTokenDeleteInput, output=BoolResult, fastapi=False, mcp=False)
def bearer_token_delete(
    self, job: Job, username: str = None, token: str = None
) -> Result:
    """
    Deletes a bearer token from the cache.
    This method removes a bearer token from the cache based on either
    the token itself or the associated username.

    If a token is provided, it will be removed directly. If a username
    is provided, all tokens associated with that username will be evicted
    from the cache.

    Args:
        username (str, optional): The username associated with the token(s) to be removed. Defaults to None.
        token (str, optional): The bearer token to be removed. Defaults to None.

    Returns:
        bool: True if the operation was successful, otherwise raises an exception.

    Raises:
        RuntimeError: If the token removal from the cache fails.
        Exception: If neither username nor token is provided.
    """
    self.cache.expire()
    token_removed_count = 0
    if token:
        cache_key = f"bearer_token::{token}"
        if cache_key in self.cache:
            if self.cache.delete(cache_key, retry=True):
                token_removed_count = 1
            else:
                raise RuntimeError(f"Failed to remove {username} token from cache")
    elif username:
        token_removed_count = self.cache.evict(tag=username, retry=True)
    else:
        raise Exception("Cannot delete, either username or token must be provided")

    log.info(
        f"{self.name} removed {token_removed_count} token(s) for user {username}"
    )

    return Result(task=f"{self.name}:bearer_token_delete", result=True)

bearer_token_list¤

Retrieves a list of bearer tokens from the cache, optionally filtered by username.

Parameters:

Name	Type	Description	Default
`username`	`str`	The username to filter tokens by. Defaults to None.	`None`

Returns:

Name	Type	Description
`list`	`Result`	A list of dictionaries containing token information. Each dictionary contains: "username" (str): The username associated with the token. "token" (str): The bearer token. "age" (str): The age of the token. "creation" (str): The creation time of the token. "expires" (str): The expiration time of the token, if available.

If no tokens are found, a list with a single dictionary containing empty strings for all fields is returned.

Source code in norfab\workers\fastapi_worker\fastapi_worker.py

@Task(
    input=BearerTokenListInput,
    output=BearerTokenListResult,
    fastapi=False,
    mcp=False,
)
def bearer_token_list(self, job: Job, username: str = None) -> Result:
    """
    Retrieves a list of bearer tokens from the cache, optionally filtered by username.

    Args:
        username (str, optional): The username to filter tokens by. Defaults to None.

    Returns:
        list: A list of dictionaries containing token information. Each dictionary contains:

            - "username" (str): The username associated with the token.
            - "token" (str): The bearer token.
            - "age" (str): The age of the token.
            - "creation" (str): The creation time of the token.
            - "expires" (str): The expiration time of the token, if available.

    If no tokens are found, a list with a single dictionary containing
    empty strings for all fields is returned.
    """

    self.cache.expire()
    ret = Result(task=f"{self.name}:bearer_token_list", result=[])

    for cache_key in self.cache:
        token_data, expires, tag = self.cache.get(
            cache_key, expire_time=True, tag=True
        )
        if username and tag != username:
            continue
        if expires is not None:
            expires = datetime.fromtimestamp(expires)
        creation = datetime.fromisoformat(token_data["created"])
        age = datetime.now() - creation
        ret.result.append(
            {
                "username": token_data["username"],
                "token": token_data["token"],
                "age": str(age),
                "creation": str(creation),
                "expires": str(expires),
            }
        )

    # return empty result if no tokens found
    if not ret.result:
        ret.result = [
            {
                "username": "",
                "token": "",
                "age": "",
                "creation": "",
                "expires": "",
            }
        ]

    return ret

bearer_token_check¤

Checks if the provided bearer token is present in the cache and still active.

Parameters:

Name	Type	Description	Default
`token`	`str`	The bearer token to check.	required

Returns:

Name	Type	Description
`bool`	`Result`	True if the token is found in the cache, False otherwise.

Source code in norfab\workers\fastapi_worker\fastapi_worker.py

@Task(input=BearerTokenCheckInput, output=BoolResult, fastapi=False, mcp=False)
def bearer_token_check(self, token: str, job: Job) -> Result:
    """
    Checks if the provided bearer token is present in the cache and still active.

    Args:
        token (str): The bearer token to check.

    Returns:
        bool: True if the token is found in the cache, False otherwise.
    """
    self.cache.expire()
    cache_key = f"bearer_token::{token}"
    return Result(
        task=f"{self.name}:bearer_token_check", result=cache_key in self.cache
    )