Source code for edri.protocol.keys

"""Pre-shared keys, from the environment or a file.

A file is supported because environment variables are readable through
``/proc/<pid>/environ``, ``docker inspect`` and pod specs, and because a file is
what Kubernetes and Docker secrets mount. It holds one key per line so a
rotation can have both the old and the new key present at once.
"""
import os
from collections.abc import Iterator
from pathlib import Path

__all__ = ["MINIMUM_SECRET_LENGTH", "KeySet"]

MINIMUM_SECRET_LENGTH = 32




class KeySet:
    """One or more accepted keys. ``primary`` is what this process sends with."""

    def __init__(self, keys: list[bytes]) -> None:
        if not keys:
            raise ValueError("at least one key is required")
        for key in keys:
            if len(key) < MINIMUM_SECRET_LENGTH:
                raise ValueError(
                    f"secrets must be at least {MINIMUM_SECRET_LENGTH} bytes; "
                    "generate one with "
                    "python -c \"import secrets; print(secrets.token_urlsafe(32))\""
                )
        self.keys = keys

    @property
    def primary(self) -> bytes:
        return self.keys[0]

    def __len__(self) -> int:
        return len(self.keys)

    def __iter__(self) -> Iterator[bytes]:
        return iter(self.keys)



    @classmethod
    def from_file(cls, path: str) -> "KeySet":
        try:
            content = Path(path).read_text()
        except OSError as e:
            raise ValueError(f"cannot read the secret file {path}: {e}") from e
        keys = [
            line.strip().encode()
            for line in content.splitlines()
            if line.strip() and not line.lstrip().startswith("#")
        ]
        if not keys:
            raise ValueError(f"the secret file {path} contains no keys")
        return cls(keys)




    @classmethod
    def from_settings(cls) -> "KeySet | None":
        """Build from configuration, or ``None`` when no secret is configured."""
        secret_file = os.getenv("EDRI_SWITCH_SECRET_FILE")
        if secret_file:
            return cls.from_file(secret_file)
        secret = os.getenv("EDRI_SWITCH_SECRET")
        if secret:
            return cls([secret.encode()])
        return None