Skip to content

networking.node.tokens

tokens

Classes:

Name Description
TokenVerifier

Classes

TokenVerifier 

TokenVerifier(public_key: str | bytes)

Initializes the TokenVerifier with a public key.

This key is essential for securely decoding and verifying JSON Web Tokens (JWTs) issued by a corresponding private key. The public key can be provided as either a string or a bytes object.

Parameters:

Name Type Description Default
public_key str | bytes

The public key used for decoding and verification.

 required

Methods:

Name Description
verify_token

Verifies a JSON Web Token (JWT) against a set of criteria.
Source code in unaiverse/networking/node/tokens.py 
19
20
21
22
23
24
25
26
27
28
def __init__(self, public_key: str | bytes):
    """Initializes the `TokenVerifier` with a public key.

    This key is essential for securely decoding and verifying JSON Web Tokens (JWTs) issued by a corresponding
    private key. The public key can be provided as either a string or a bytes object.

    Args:
        public_key: The public key used for decoding and verification.
    """
    self.public_key = public_key
Methods:
verify_token 
verify_token(token: str | bytes, node_id: str | None = None, ip: str | None = None, hostname: str | None = None, port: int | None = None, p2p_peer: str | None = None)

Verifies a JSON Web Token (JWT) against a set of criteria.

The method first attempts to decode the token using the provided public key and the RS256 algorithm, handling DecodeError and ExpiredSignatureError. It then performs optional checks to ensure that the token's payload matches specific network identifiers, such as node_id, ip, hostname, and port. It can also verify if a specific peer is present in the token's list of p2p_peers.

Parameters:

Name Type Description Default
token str | bytes

The JWT to verify, as a string or bytes object.

 required
node_id str | None

Optional node_id to check against the token's payload.

 None
ip str | None

Optional IP address to check.

 None
hostname str | None

Optional hostname to check.

 None
port int | None

Optional port number to check.

 None
p2p_peer str | None

Optional peer identifier to check within the p2p_peers list.

 None

Returns:

Type Description

A tuple containing the node_id and cv_hash from the token's payload if all checks pass. Otherwise,

it returns a tuple of (None, None).
Source code in unaiverse/networking/node/tokens.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def verify_token(self, token: str | bytes,
                 node_id: str | None = None, ip: str | None = None,
                 hostname: str | None = None,
                 port: int | None = None,
                 p2p_peer: str | None = None):
    """Verifies a JSON Web Token (JWT) against a set of criteria.

    The method first attempts to decode the token using the provided public key and the RS256 algorithm,
    handling `DecodeError` and `ExpiredSignatureError`. It then performs optional checks to ensure that
    the token's payload matches specific network identifiers, such as `node_id`, `ip`, `hostname`, and `port`.
    It can also verify if a specific peer is present in the token's list of `p2p_peers`.

    Args:
        token: The JWT to verify, as a string or bytes object.
        node_id: Optional `node_id` to check against the token's payload.
        ip: Optional IP address to check.
        hostname: Optional hostname to check.
        port: Optional port number to check.
        p2p_peer: Optional peer identifier to check within the `p2p_peers` list.

    Returns:
        A tuple containing the `node_id` and `cv_hash` from the token's payload if all checks pass. Otherwise,
        it returns a tuple of `(None, None)`.
    """

    # Decoding token using the public key
    try:
        payload = jwt.decode(token, self.public_key, algorithms=["RS256"])
    except jwt.DecodeError as e:
        return None, None
    except jwt.ExpiredSignatureError as e:  # This checks expiration time (required)
        return None, None

    # Checking optional information
    if node_id is not None and payload["node_id"] != node_id:
        return None, None
    if ip is not None and payload["ip"] != ip:
        return None, None
    if hostname is not None and payload["hostname"] != hostname:
        return None, None
    if port is not None and payload["port"] != port:
        return None, None
    if p2p_peer is not None and p2p_peer not in payload["p2p_peers"]:
        return None, None

    # All ok
    return payload["node_id"], payload["cv_hash"]