bedrock.websockets.authentication 

 1import time  # pragma: unit
 2from collections.abc import Callable  # pragma: unit
 3
 4from bedrock._helpers.crypto import decode_access_token  # pragma: unit
 5
 6
 7def _get_ttl(decoded_token, now):  # pragma: unit
 8    expiry = decoded_token["exp"]
 9    ttl = max(0, expiry - now)
10    return ttl
11
12
13def authenticate_token(token):  # pragma: unit
14    """
15    Authenticate a JWT token by decoding it and calculating its TTL.
16    :param token: The JWT token to authenticate
17    :return: A tuple of (decoded_token, now, ttl) where:
18        - decoded_token: the decoded JWT token,
19        - now: the current time in seconds, and
20        - ttl: the time to live of the token in seconds.
21    """
22    decoded_token = decode_access_token(token)
23    now = int(time.time())
24    ttl = _get_ttl(decoded_token, now)
25    return decoded_token, now, ttl
26
27
28def default_authoriser(token):  # pragma: unit
29    """
30    Default authoriser function that returns an empty dictionary. To override this, see `set_authoriser`.
31    :param token: The JWT token to authorise
32    :return: A dictionary of authorisation filters to apply to the websocket connection
33    """
34    return {}
35
36
37WEBSOCKET_AUTHORISATION = default_authoriser  # pragma: unit
38"""
39The function that is called to authorise a websocket connection. By default, it is `default_authoriser`, but can be set to a custom function
40(see `set_authoriser`).
41"""
42
43
44def set_authoriser(fn: Callable[[str], dict]):  # pragma: unit
45    global WEBSOCKET_AUTHORISATION
46    WEBSOCKET_AUTHORISATION = fn
47
48
49def authorise_token(token) -> dict:  # pragma: unit
50    return WEBSOCKET_AUTHORISATION(token)
def authenticate_token(token): 
14def authenticate_token(token):  # pragma: unit
15    """
16    Authenticate a JWT token by decoding it and calculating its TTL.
17    :param token: The JWT token to authenticate
18    :return: A tuple of (decoded_token, now, ttl) where:
19        - decoded_token: the decoded JWT token,
20        - now: the current time in seconds, and
21        - ttl: the time to live of the token in seconds.
22    """
23    decoded_token = decode_access_token(token)
24    now = int(time.time())
25    ttl = _get_ttl(decoded_token, now)
26    return decoded_token, now, ttl

Authenticate a JWT token by decoding it and calculating its TTL.

Parameters
  • token: The JWT token to authenticate
Returns

A tuple of (decoded_token, now, ttl) where: - decoded_token: the decoded JWT token, - now: the current time in seconds, and - ttl: the time to live of the token in seconds.

def default_authoriser(token): 
29def default_authoriser(token):  # pragma: unit
30    """
31    Default authoriser function that returns an empty dictionary. To override this, see `set_authoriser`.
32    :param token: The JWT token to authorise
33    :return: A dictionary of authorisation filters to apply to the websocket connection
34    """
35    return {}

Default authoriser function that returns an empty dictionary. To override this, see set_authoriser.

Parameters
  • token: The JWT token to authorise
Returns

A dictionary of authorisation filters to apply to the websocket connection

def WEBSOCKET_AUTHORISATION(token): 
29def default_authoriser(token):  # pragma: unit
30    """
31    Default authoriser function that returns an empty dictionary. To override this, see `set_authoriser`.
32    :param token: The JWT token to authorise
33    :return: A dictionary of authorisation filters to apply to the websocket connection
34    """
35    return {}

The function that is called to authorise a websocket connection. By default, it is default_authoriser, but can be set to a custom function (see set_authoriser).

def set_authoriser(fn: Callable[[str], dict]): 
45def set_authoriser(fn: Callable[[str], dict]):  # pragma: unit
46    global WEBSOCKET_AUTHORISATION
47    WEBSOCKET_AUTHORISATION = fn
def authorise_token(token) -> dict: 
50def authorise_token(token) -> dict:  # pragma: unit
51    return WEBSOCKET_AUTHORISATION(token)