audit-core/audit_core/credentials.py

"""Database credentials read from a mounted directory (AUDIT-WP-0005-T02).

The credential broker issues *dynamic* PostgreSQL leases, so the password the
pod should use changes while the pod is running. Environment variables cannot
express that: they are fixed at process start, so every rotation would need a
restart and every restart is a delivery gap.

A mounted Secret can. Kubernetes updates the files in place when the
ExternalSecret refreshes, so reading them at connection time — rather than at
startup — makes rotation invisible to senders.

Expected layout, one value per file, matching what an ExternalSecret template
produces::

    /etc/audit-core/db/username
    /etc/audit-core/db/password
    /etc/audit-core/db/host
    /etc/audit-core/db/port
    /etc/audit-core/db/dbname

A single ``dsn`` file is also accepted for stores that vend one string.
"""

from __future__ import annotations

import logging
import pathlib

log = logging.getLogger("audit_core.credentials")

# Maps a file name to its libpq connection keyword.
_FIELDS = {
    "username": "user",
    "user": "user",
    "password": "password",
    "host": "host",
    "port": "port",
    "dbname": "dbname",
    "database": "dbname",
    "sslmode": "sslmode",
}


class CredentialDirectory:
    """Connection parameters re-read from disk on every use."""

    def __init__(self, path: str | pathlib.Path) -> None:
        self.path = pathlib.Path(path)
        if not self.path.is_dir():
            raise ValueError(f"credential directory does not exist: {self.path}")
        self._last_fingerprint: str | None = None

    def read(self) -> dict[str, str]:
        """Return the current connection parameters.

        Called per connection attempt, so a rotated password is picked up by
        the next connection the pool opens without any restart.

        Kubernetes Secret volumes update by swapping the ``..data`` directory.
        Reading ``username`` then ``password`` from the mount point can tear
        across that swap and pair a new user with an old password. Resolve
        ``..data`` once and read every file from that snapshot.
        """
        last_error: Exception | None = None
        for _ in range(5):
            try:
                params = self._read_snapshot()
            except (FileNotFoundError, ValueError) as exc:
                last_error = exc
                continue
            if "dsn" in params:
                self._note_rotation(params["dsn"])
                return params
            if "user" not in params:
                last_error = ValueError(f"no username in credential directory {self.path}")
                continue
            self._note_rotation(params.get("password", ""))
            return params
        if last_error is not None:
            raise last_error
        raise ValueError(f"no username in credential directory {self.path}")

    def _snapshot_dir(self) -> pathlib.Path:
        data = self.path / "..data"
        if data.exists():
            return data.resolve()
        return self.path

    def _read_snapshot(self) -> dict[str, str]:
        root = self._snapshot_dir()
        params: dict[str, str] = {}
        dsn_file = root / "dsn"
        if dsn_file.exists():
            params["dsn"] = dsn_file.read_text().strip()
            return params
        for name, keyword in _FIELDS.items():
            candidate = root / name
            if candidate.exists():
                value = candidate.read_text().strip()
                if value:
                    params.setdefault(keyword, value)
        return params

    def _note_rotation(self, secret: str) -> None:
        import hashlib

        fingerprint = hashlib.sha256(secret.encode()).hexdigest()[:12]
        if self._last_fingerprint is None:
            self._last_fingerprint = fingerprint
        elif fingerprint != self._last_fingerprint:
            log.info("database credential rotated (fingerprint %s)", fingerprint)
            self._last_fingerprint = fingerprint