88 lines
3 KiB
Python
88 lines
3 KiB
Python
|
"""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.
|
||
|
|
"""
|
||
|
|
params: dict[str, str] = {}
|
||
|
|
dsn_file = self.path / "dsn"
|
||
|
|
if dsn_file.exists():
|
||
|
|
params["dsn"] = dsn_file.read_text().strip()
|
||
|
|
self._note_rotation(params["dsn"])
|
||
|
|
return params
|
||
|
|
|
||
|
|
for name, keyword in _FIELDS.items():
|
||
|
|
candidate = self.path / name
|
||
|
|
if candidate.exists():
|
||
|
|
value = candidate.read_text().strip()
|
||
|
|
if value:
|
||
|
|
params.setdefault(keyword, value)
|
||
|
|
if "user" not in params:
|
||
|
|
raise ValueError(f"no username in credential directory {self.path}")
|
||
|
|
# Fingerprint the password so rotation is visible in logs without the
|
||
|
|
# password itself ever being logged.
|
||
|
|
self._note_rotation(params.get("password", ""))
|
||
|
|
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
|