🟢 unaiverse.networking.node.tokens

What this module does 🟢

Verifies JWT authorization tokens used to authenticate peers joining worlds and pools, decoding and validating claims for connection authorization.

tokens

█████ █████ ██████ █████ █████ █████ █████ ██████████ ███████████ █████████ ██████████ ░░███ ░░███ ░░██████ ░░███ ░░███ ░░███ ░░███ ░░███░░░░░█░░███░░░░░███ ███░░░░░███░░███░░░░░█ ░███ ░███ ░███░███ ░███ ██████ ░███ ░███ ░███ ░███ █ ░ ░███ ░███ ░███ ░░░ ░███ █ ░ ░███ ░███ ░███░░███░███ ░░░░░███ ░███ ░███ ░███ ░██████ ░██████████ ░░█████████ ░██████
░███ ░███ ░███ ░░██████ ███████ ░███ ░░███ ███ ░███░░█ ░███░░░░░███ ░░░░░░░░███ ░███░░█
░███ ░███ ░███ ░░█████ ███░░███ ░███ ░░░█████░ ░███ ░ █ ░███ ░███ ███ ░███ ░███ ░ █ ░░████████ █████ ░░█████░░████████ █████ ░░███ ██████████ █████ █████░░█████████ ██████████ ░░░░░░░░ ░░░░░ ░░░░░ ░░░░░░░░ ░░░░░ ░░░ ░░░░░░░░░░ ░░░░░ ░░░░░ ░░░░░░░░░ ░░░░░░░░░░ A Collectionless AI Project (https://collectionless.ai) Registration/Login: https://unaiverse.io Code Repositories: https://github.com/collectionlessai/ Main Developers: Stefano Melacci (Project Leader), Christian Di Maio, Tommaso Guidi

TokenVerifier

TokenVerifier(public_key: str | bytes)

JWT verifier for UNaIVERSE node tokens signed with RS256.

Wraps the PyJWT library to decode and validate JSON Web Tokens that are issued by the UNaIVERSE platform as node credentials. Verification uses RS256 (RSA + SHA-256) asymmetric signing: the platform signs tokens with its private key and nodes verify them with the corresponding public key stored in this instance.

The optional keyword arguments of verify_token allow callers to enforce that specific claims in the token payload match known values, making it possible to bind a token to a particular network identity (node ID, IP address, hostname, port, or P2P peer list).

Attributes:

Name	Type	Description
public_key		The RSA public key used to verify token signatures. Stored exactly as supplied at construction time (str or bytes).

Initialize the verifier with an RSA public key.

The supplied key must be the RSA public key that corresponds to the private key used by the UNaIVERSE platform to sign node JWTs. It is stored as-is and passed directly to jwt.decode on every call to verify_token.

Parameters:

Name	Type	Description	Default
public_key	str | bytes	The RSA public key for RS256 signature verification. Accepted as either a PEM-encoded string or raw bytes.	required

Source code in unaiverse/networking/node/tokens.py

def __init__(self, public_key: str | bytes):
    """Initialize the verifier with an RSA public key.

    The supplied key must be the RSA public key that corresponds to the
    private key used by the UNaIVERSE platform to sign node JWTs. It is
    stored as-is and passed directly to ``jwt.decode`` on every call to
    ``verify_token``.

    Args:
        public_key: The RSA public key for RS256 signature verification.
            Accepted as either a PEM-encoded string or raw bytes.
    """
    self.public_key = public_key

public_key instance-attribute

public_key = public_key

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)

Decode and verify a UNaIVERSE node JWT against optional claim constraints.

The token is decoded with RS256 using the public key supplied at construction. If decoding fails or the token has expired, (None, None) is returned immediately. When decoding succeeds, each non-None keyword argument is matched against the corresponding claim in the payload; the first mismatch causes an early (None, None) return. Only when all supplied constraints are satisfied does the method return the identity values extracted from the payload.

The p2p_peer check tests membership in the payload's p2p_peers list rather than equality, so a single token may authorise multiple peers.

Parameters:

Name	Type	Description	Default
token	str | bytes	The JWT to verify, provided as a string or bytes object.	required
node_id	str | None	If not None, the payload's node_id claim must equal this value. Defaults to None.	None
ip	str | None	If not None, the payload's ip claim must equal this value. Defaults to None.	None
hostname	str | None	If not None, the payload's hostname claim must equal this value. Defaults to None.	None
port	int | None	If not None, the payload's port claim must equal this value. Defaults to None.	None
p2p_peer	str | None	If not None, this value must be present in the payload's p2p_peers list. Defaults to None.	None

Returns:

Type	Description
	A two-element tuple (node_id, cv_hash) extracted from the token
	payload when all checks pass. Both elements are None if the token
	cannot be decoded, has expired, or fails any supplied constraint check.

Raises:

Type	Description
KeyError	If the payload is valid but is missing the node_id or cv_hash keys (or any other checked key such as ip, hostname, port, or p2p_peers when the corresponding argument is not None).

Source code in unaiverse/networking/node/tokens.py

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):
    """Decode and verify a UNaIVERSE node JWT against optional claim constraints.

    The token is decoded with RS256 using the public key supplied at
    construction. If decoding fails or the token has expired, ``(None, None)``
    is returned immediately. When decoding succeeds, each non-``None`` keyword
    argument is matched against the corresponding claim in the payload; the
    first mismatch causes an early ``(None, None)`` return. Only when all
    supplied constraints are satisfied does the method return the identity
    values extracted from the payload.

    The ``p2p_peer`` check tests membership in the payload's ``p2p_peers``
    list rather than equality, so a single token may authorise multiple peers.

    Args:
        token: The JWT to verify, provided as a string or bytes object.
        node_id: If not ``None``, the payload's ``node_id`` claim must equal
            this value. Defaults to None.
        ip: If not ``None``, the payload's ``ip`` claim must equal this value.
            Defaults to None.
        hostname: If not ``None``, the payload's ``hostname`` claim must equal
            this value. Defaults to None.
        port: If not ``None``, the payload's ``port`` claim must equal this
            value. Defaults to None.
        p2p_peer: If not ``None``, this value must be present in the payload's
            ``p2p_peers`` list. Defaults to None.

    Returns:
        A two-element tuple ``(node_id, cv_hash)`` extracted from the token
        payload when all checks pass. Both elements are ``None`` if the token
        cannot be decoded, has expired, or fails any supplied constraint check.

    Raises:
        KeyError: If the payload is valid but is missing the ``node_id`` or
            ``cv_hash`` keys (or any other checked key such as ``ip``,
            ``hostname``, ``port``, or ``p2p_peers`` when the corresponding
            argument is not ``None``).
    """

    # Decoding token using the public key
    try:
        payload = jwt.decode(token, self.public_key, algorithms=["RS256"])
    except jwt.DecodeError:
        return None, None
    except jwt.ExpiredSignatureError:  # 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"]