Skip to content

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 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@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

    # 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"],
            apikey_config=contents["config"],
        )

    # Forward error
    raise HTTPException(response.status_code, f"UAC manager: {read_response_error(response)}")

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 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
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