proxystore.endpoint.identity

proxystore/endpoint/identity.py

Endpoint identities.

Each endpoint has an ed25519 secret key which is stored in its directory. The public key of the endpoint, its EndpointId, identifies the endpoint to clients and peer endpoints. Peers verify each other's identity when establishing a connection because only the endpoint which has the secret key can prove ownership of the public key.

EndpointId

Bases: str

ID of an endpoint.

The ID is the endpoint's ed25519 public key encoded as 64 lowercase hexadecimal characters. An EndpointId is a str so it can be used anywhere a string is expected (e.g., serialized in configuration files). Every instance is a valid public key (not every 32-byte value is). The constructor only accepts the exact format. Use from_str() to also accept surrounding whitespace and uppercase characters.

Example

endpoint_id = EndpointId.from_str('4C1D...')
assert endpoint_id == '4c1d...'
print(endpoint_id.log_name('my-endpoint'))  # my-endpoint(4c1d...)

Raises:

• ValueError –

  If the value is not a valid endpoint ID.

from_str classmethod

from_str(value: str) -> Self

Parse an endpoint ID.

Parameters:

• value (str) –

  Endpoint ID encoded as 64 hexadecimal characters. Surrounding whitespace is removed and uppercase characters are converted to lowercase.

Raises:

• ValueError –

  If value is not a valid endpoint ID.

Source code in proxystore/endpoint/identity.py

@classmethod
def from_str(cls, value: str) -> Self:
    """Parse an endpoint ID.

    Args:
        value: Endpoint ID encoded as 64 hexadecimal characters.
            Surrounding whitespace is removed and uppercase characters
            are converted to lowercase.

    Raises:
        ValueError: If `value` is not a valid endpoint ID.
    """
    if isinstance(value, cls):
        return value
    if isinstance(value, str):
        value = value.strip().lower()
    return cls(value)

random classmethod

random() -> Self

Generate the ID of a new random endpoint.

The secret key of the endpoint is discarded so this is only useful when an ID is needed but the endpoint is not (e.g., in tests).

Source code in proxystore/endpoint/identity.py

@classmethod
def random(cls) -> Self:
    """Generate the ID of a new random endpoint.

    The secret key of the endpoint is discarded so this is only useful
    when an ID is needed but the endpoint is not (e.g., in tests).
    """
    return cls(SecretKey.generate().endpoint_id)

short

short() -> str

Get a short prefix of the ID for logging.

Source code in proxystore/endpoint/identity.py

def short(self) -> str:
    """Get a short prefix of the ID for logging."""
    return self[:10]

log_name

log_name(name: str) -> str

Format the ID with a name as 'name(id-prefix)'.

Source code in proxystore/endpoint/identity.py

def log_name(self, name: str) -> str:
    """Format the ID with a name as `#!python 'name(id-prefix)'`."""
    return f'{name}({self.short()})'

SecretKey

SecretKey(key: bytes)

Secret key of an endpoint.

The secret key is an ed25519 key which proves the identity of the endpoint to peers. The key is never included in its repr() so it is not accidentally logged.

Example

secret_key = SecretKey.generate()
endpoint_id = secret_key.endpoint_id
same_key = SecretKey(secret_key.to_bytes())

Parameters:

• key (bytes) –

  Raw bytes of the secret key.

Raises:

• ValueError –

  If key is not a valid secret key.

Source code in proxystore/endpoint/identity.py

def __init__(self, key: bytes) -> None:
    try:
        public = iroh.SecretKey.from_bytes(key).public()
    except iroh.IrohError:
        raise ValueError(
            f'Endpoint secret key is not a valid ed25519 secret key '
            f'({len(key)} bytes).',
        ) from None
    self._key = bytes(key)
    self._endpoint_id = EndpointId(str(public))

endpoint_id property

endpoint_id: EndpointId

ID of the endpoint with this secret key (i.e., the public key).

generate classmethod

generate() -> Self

Generate a new random secret key.

Source code in proxystore/endpoint/identity.py

@classmethod
def generate(cls) -> Self:
    """Generate a new random secret key."""
    return cls(iroh.SecretKey.generate().to_bytes())

to_bytes

to_bytes() -> bytes

Get the raw bytes of the secret key.

Source code in proxystore/endpoint/identity.py

def to_bytes(self) -> bytes:
    """Get the raw bytes of the secret key."""
    return self._key