auth

Functions and classes used for ZeroMQ authentication.

load_certificate

load_certificate(path: PathLike) -> tuple[bytes, bytes | None]

Load public and secret keys from a ZeroMQ certificate file.

Parameters:

Name	Type	Description	Default
path	PathLike	The path to a certificate file. Can include ~ in the path, which will expand the user's home directory.	required

Returns:

Type	Description
tuple[bytes, bytes | None]	The (public, secret) keys. The secret key might be None if it is not defined in the file.

Source code in src/msl/network/auth.py

def load_certificate(path: PathLike) -> tuple[bytes, bytes | None]:
    """Load public and secret keys from a ZeroMQ certificate file.

    Args:
        path: The path to a certificate file. Can include `~` in the path, which will expand the
            user's home directory.

    Returns:
        The `(public, secret)` keys. The `secret` key might be `None` if it is not defined in the file.
    """
    path = Path(os.fsdecode(path)).expanduser()
    if not path.is_file():
        msg = f"Cannot load certificate at {path}"  # prefer FileNotFoundError instead of pyzmq exception
        raise FileNotFoundError(msg)

    return certs.load_certificate(path)  # pyright: ignore[reportUnknownMemberType]

AuthCurve

AuthCurve(public_key: bytes, secret_key: bytes, broker_key: bytes)

CURVE authentication credentials to connect to a Broker.

Parameters:

Name	Type	Description	Default
public_key	bytes	The public key of the Client or Service.	required
secret_key	bytes	The secret key of the Client or Service.	required
broker_key	bytes	The public key of the Broker.	required

Source code in src/msl/network/auth.py

def __init__(self, public_key: bytes, secret_key: bytes, broker_key: bytes) -> None:
    """[CURVE](https://rfc.zeromq.org/spec/25/) authentication credentials to connect to a [Broker][].

    Args:
        public_key: The public key of the [Client][] or [Service][].
        secret_key: The secret key of the [Client][] or [Service][].
        broker_key: The public key of the [Broker][].
    """
    self.public_key: bytes = public_key
    self.secret_key: bytes = secret_key
    self.broker_key: bytes = broker_key

load staticmethod

load(broker: PathLike, own: PathLike | None = None) -> AuthCurve

Load CURVE authentication credentials from files.

Parameters:

Name	Type	Description	Default
broker	PathLike	The path to the file that contains the Broker's public key. Can include ~ in the path, which will expand the user's home directory.	required
own	PathLike | None	The path to the file that contains the public and secret keys of the Client or Service that connects to the Broker. If None, loads the default credentials that were created by running the msl-network curve command. Can include ~ in the path, which will expand the user's home directory.	None

Returns:

Type	Description
AuthCurve	The CURVE credentials.

Source code in src/msl/network/auth.py

@staticmethod
def load(broker: PathLike, own: PathLike | None = None) -> AuthCurve:
    """Load [CURVE](https://rfc.zeromq.org/spec/25/) authentication credentials from files.

    Args:
        broker: The path to the file that contains the [Broker][]'s public key.
            Can include `~` in the path, which will expand the user's home directory.
        own: The path to the file that contains the public and secret keys of the
            [Client][] or [Service][] that connects to the [Broker][]. If `None`,
            loads the default credentials that were created by running the
            `msl-network curve` command. Can include `~` in the path, which will
            expand the user's home directory.

    Returns:
        The [CURVE](https://rfc.zeromq.org/spec/25/) credentials.
    """
    broker_public, _ = load_certificate(broker)

    default = MSL_NETWORK_HOME / f"{socket.gethostname()}.key_secret"
    file = default if own is None else os.fsdecode(own)
    own_public, own_secret = load_certificate(file)
    if own_secret is None:
        msg = f"No secret key found in {file}"
        raise ValueError(msg)

    return AuthCurve(public_key=own_public, secret_key=own_secret, broker_key=broker_public)

AuthPlain

AuthPlain(username: str | bytes, password: str | bytes)

PLAIN authentication credentials to connect to a Broker.

Parameters:

Name	Type	Description	Default
username	str | bytes	The username registered with the Broker.	required
password	str | bytes	The password registered with the Broker.	required

Source code in src/msl/network/auth.py

def __init__(self, username: str | bytes, password: str | bytes) -> None:
    """[PLAIN](https://rfc.zeromq.org/spec/24/) authentication credentials to connect to a [Broker][].

    Args:
        username: The username registered with the [Broker][].
        password: The password registered with the [Broker][].
    """
    self.username: bytes = username.encode() if isinstance(username, str) else username
    self.password: bytes = password.encode() if isinstance(password, str) else password

load staticmethod

load(path: PathLike | None = None, sep: str | None = None) -> AuthPlain

Load PLAIN authentication credentials from a file.

Parameters:

Name	Type	Description	Default
path	PathLike | None	The path to load the credentials from. If the path has the .json extension, the file must contain a single username to password mapping. If None, loads the credentials that were created by running the msl-network plain add command. Can include ~ in the path, which will expand the user's home directory.	None
sep	str | None	If the path extension is not .json, the separator to use to split the contents of the file to get the username and password values. The value is passed to str.split.	None

Returns:

Type	Description
AuthPlain	The PLAIN credentials.

Source code in src/msl/network/auth.py

@staticmethod
def load(path: PathLike | None = None, sep: str | None = None) -> AuthPlain:
    """Load [PLAIN](https://rfc.zeromq.org/spec/24/) authentication credentials from a file.

    Args:
        path: The path to load the credentials from. If the path has the `.json`
            extension, the file must contain a single *username* to *password*
            mapping. If `None`, loads the credentials that were created by running
            the `msl-network plain add` command. Can include `~` in the path, which
            will expand the user's home directory.
        sep: If the `path` extension is not `.json`, the separator to use to split the
            contents of the file to get the *username* and *password* values. The value
            is passed to [str.split][].

    Returns:
        The [PLAIN](https://rfc.zeromq.org/spec/24/) credentials.
    """
    default = MSL_NETWORK_HOME / "plain.json"
    file = default if path is None else Path(os.fsdecode(path)).expanduser()
    if not file.exists():
        msg = f"No such file: '{file}'"
        if file == default:
            msg += ", run `msl-network plain add` in a terminal"
        raise FileNotFoundError(msg)

    if file.suffix == ".json":
        _, auth = load_plain(file)
        if not auth:
            msg = "No PLAIN credentials found"
            if file == default:
                msg += ", run `msl-network plain add` in a terminal"
            raise ValueError(msg)

        if len(auth) > 1:
            msg = (
                "Too many PLAIN credentials found, run `msl-network plain remove` in a "
                "terminal or specify a different file"
            )
            raise ValueError(msg)

        username = next(iter(auth))
        password = auth[username]
    else:
        u, p = file.read_text().split(sep, maxsplit=1)
        username = u.strip()
        password = p.strip()

    return AuthPlain(username=username, password=password)