rs_server_common/authentication/apikey.md

<< Back to index

API key authentication functions implementation.

Note: calls https://github.com/csgroup-oss/apikey-manager

__apikey_security_cached(apikey_value) async

Cached version of apikey_security. Cache an infinite (sys.maxsize) number of results for 120 seconds.

This function serves as a cached version of apikey_security. It retrieves user access control information from the User Authentication and Authorization Control (UAC) manager and caches the result for performance optimization.

Parameters:

Name	Type	Description	Default
apikey_value	str	The API key value.	required

Returns:

Name	Type	Description
AuthInfo	AuthInfo	Authentication information from the keycloak account, associated to the api key.

Raises:

Type	Description
HTTPException	If there is an error connecting to the UAC manager or if the UAC manager returns an error.

Source code in docs/rs-server/services/common/rs_server_common/authentication/apikey.py

@cached(cache=ttl_cache)
async def __apikey_security_cached(apikey_value) -> AuthInfo:
    """
    Cached version of apikey_security. Cache an infinite (sys.maxsize) number of results for 120 seconds.

    This function serves as a cached version of apikey_security. It retrieves user access control information
    from the User Authentication and Authorization Control (UAC) manager and caches the result for performance
    optimization.

    Args:
        apikey_value (str): The API key value.

    Returns:
        AuthInfo: Authentication information from the keycloak account, associated to the api key.

    Raises:
        HTTPException: If there is an error connecting to the UAC manager or if the UAC manager returns an error.
    """

    # The uac manager check url is passed as an environment variable
    try:
        check_url = env["RSPY_UAC_CHECK_URL"]
    except KeyError:
        raise HTTPException(HTTP_400_BAD_REQUEST, "UAC manager URL is undefined")  # pylint: disable=raise-missing-from

    # Remove the " and ' around the api key
    apikey_value = (apikey_value or "").strip("\"'")

    # Request the uac, pass user-defined api key in http header
    try:
        response = await settings.http_client().get(check_url, headers={APIKEY_HEADER: apikey_value or ""})
    except httpx.HTTPError as error:
        message = "Error connecting to the UAC manager"
        logger.error(f"{message}\n{traceback.format_exc()}")
        raise HTTPException(HTTP_500_INTERNAL_SERVER_ERROR, message) from error

    # Read the api key info
    if response.is_success:
        contents = response.json()
        # Note: for now, config is an empty dict
        return AuthInfo(
            user_login=contents["user_login"],
            iam_roles=contents["iam_roles"],
            attributes=contents["config"],
        )

    # In case of error, check if the api key has the right format.
    # It should be an uuid.

    message = ""
    try:
        uuid.UUID(apikey_value)  # try to parse the uuid
    except ValueError:
        masked_key = re.sub("[0-9a-zA-Z]", "*", apikey_value)
        masked_ok = re.sub("[0-9a-zA-Z]", "*", str(uuid.uuid4()))
        message = (
            f"Your API key does not have the right format: {masked_key!r}. "
            + f"It should be an UUID like: {masked_ok!r}. "
        )

    # Forward error from the UAC manager
    raise HTTPException(response.status_code, f"{message}UAC manager response: {read_response_error(response)!r}")

apikey_security(apikey_value='') async

Check the api key validity, passed as an HTTP header.

Parameters:

Name	Type	Description	Default
apikey_value	Security	API key passed in HTTP header	''

Returns:

Type	Description
AuthInfo | None	Authentication information from the keycloak account, associated to the api key.
AuthInfo | None	Or None if no api key is provided.

Source code in docs/rs-server/services/common/rs_server_common/authentication/apikey.py

async def apikey_security(
    apikey_value: Annotated[str, Security(APIKEY_AUTH_HEADER)] = "",
) -> AuthInfo | None:
    """
    Check the api key validity, passed as an HTTP header.

    Args:
        apikey_value (Security): API key passed in HTTP header

    Returns:
        Authentication information from the keycloak account, associated to the api key.
        Or None if no api key is provided.
    """

    if not apikey_value:
        return None

    # Call the cached function (fastapi Depends doesn't work with @cached)
    ret = await __apikey_security_cached(str(apikey_value))
    logger.debug(f"API key information: {ret}")
    return ret