FastAPI Auth Tasks¤

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.

Auth Tasks Sample Usage¤

To store authentication token in FastAPI service database:

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

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

To list stored tokens for 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
 fastapi-worker-1  foobar    04d685d203274a089c1a7df1395ed7e1  7:58:23.498734  2025-02-16 12:11:58.104525  None
 fastapi-worker-1  foobar    dfea48a8e412451cb2918fd526ab6c99  7:58:22.485560  2025-02-16 12:11:59.117699  None
nf#

To 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
 fastapi-worker-1  foobar    888945f96b824bf1b4358de790c452b6  8:09:06.561696  2025-02-16 12:01:30.054597  None
 fastapi-worker-1  foobar    04d685d203274a089c1a7df1395ed7e1  7:58:38.511768  2025-02-16 12:11:58.104525  None
 fastapi-worker-1  foobar    dfea48a8e412451cb2918fd526ab6c99  7:58:37.498594  2025-02-16 12:11:59.117699  None
 fastapi-worker-1  foo       4565t4yjn56h534gh35h543h5h45h4h4  0:00:27.641482  2025-02-16 20:10:08.974811  2025-02-16 21:10:08.974812
nf#

To delete specific token:

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

To delete all tokens for given user:

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

To check if 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#

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
`bool`	bool - Returns True if the token is successfully stored.

Source code in norfab\workers\fastapi_worker.py

def bearer_token_store(self, username: str, token: str, expire: int = None) -> bool:
    """
    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`	`bool`	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.py

def bearer_token_delete(self, username: str = None, token: str = None) -> bool:
    """
    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`	`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.

Source code in norfab\workers\fastapi_worker.py

def bearer_token_list(self, username: str = None) -> list:
    """
    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`	`bool`	True if the token is found in the cache, False otherwise.

Source code in norfab\workers\fastapi_worker.py

def bearer_token_check(self, token: str) -> bool:
    """
    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
    )