Deliver database credentials as a rotatable mounted directory

AUDIT-WP-0005-T02 (progress). rapp-postgres has landed platform-pg with the
audit_core database, roles, and dynamic credential provisioning, so
audit-core's side is now built against it.

In-cluster delivery is a mounted directory rather than environment variables.
A dynamic lease rotates while the pod runs and an env var is fixed at process
start, so env delivery would force a restart on every rotation - and every
restart is a delivery gap, which is what this task forbids.
CredentialDirectory is re-read on every connection attempt via psycopg_pool's
callable kwargs, so a rotated lease takes effect with no restart. Rotation is
logged by password fingerprint, never by value.

deploy/externalsecrets.yaml follows the ClusterSecretStore -> ExternalSecret ->
Secret pattern already used by activity-core and rapp-qonto, at a 15m refresh
rather than the default 1h since the interval bounds how long a revoked lease
can stay mounted. All manifests validated --dry-run=server --validate=strict.

The rotation test was initially vacuous: it passed against a deliberately naive
implementation that read credentials once at startup, because pooled sessions
stay authenticated after a password change and nothing forced a reconnect. It
now terminates the role's sessions first, and is verified to fail against the
naive implementation and pass against the real one. Tests 82 -> 84.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

audit_core/postgres_backend.py

@@ -32,6 +32,7 @@ from audit_core.interface import (
     RetentionPolicy,
     validate_event,
 )
+from audit_core.credentials import CredentialDirectory
 from audit_core.redaction import Finding
 
 try:  # pragma: no cover - import guard
@@ -139,14 +140,21 @@ class PostgresAuditBackend:
         max_size: int = 8,
         statement_timeout_ms: int = 30_000,
         migrate: bool = True,
+        credential_dir: str | None = None,
     ) -> None:
+        # A mounted credential directory takes precedence: it is the only
+        # source that can change while the process runs, which is what dynamic
+        # leases require.
+        self.credentials = (
+            CredentialDirectory(credential_dir) if credential_dir else None
+        )
         # An empty conninfo is valid: libpq then reads PGHOST/PGUSER/PGPASSWORD/
         # PGPORT/PGDATABASE from the environment. That is exactly the shape the
         # railiance-platform credential broker injects into a child process, so
         # a brokered lease needs no DSN assembled by hand — and no credential
         # ever passes through audit-core's own configuration.
         self.dsn = dsn if dsn is not None else os.environ.get("AUDIT_CORE_DATABASE_URL", "")
-        if not self.dsn and not _libpq_env_present():
+        if not self.dsn and not self.credentials and not _libpq_env_present():
             raise ValueError(
                 "no connection information: set AUDIT_CORE_DATABASE_URL, or supply "
                 "PGHOST/PGUSER/PGDATABASE (as the credential broker does)"
@@ -155,17 +163,32 @@ class PostgresAuditBackend:
             raise ValueError(f"unsafe schema name: {schema!r}")
         self.schema = schema
         self.retention_days = retention_days
+        base_kwargs = {
+            "autocommit": True,
+            # A stalled write must surface as unavailable rather than hold a
+            # request open indefinitely.
+            "options": f"-c statement_timeout={int(statement_timeout_ms)}",
+        }
+        # psycopg_pool resolves a callable on every connection attempt, so
+        # passing one is what makes a rotated lease take effect without a
+        # restart. A fixed dict would freeze the credential at startup.
+        kwargs: Any = base_kwargs
+        if self.credentials:
+            def kwargs() -> dict:  # type: ignore[misc]
+                params = self.credentials.read()
+                if "dsn" in params:
+                    raise ValueError(
+                        "a 'dsn' file is not supported with pooled connections; "
+                        "supply username/password/host/port/dbname files"
+                    )
+                return {**base_kwargs, **params}
+
         try:
             self.pool = ConnectionPool(
                 self.dsn,
                 min_size=min_size,
                 max_size=max_size,
-                kwargs={
-                    "autocommit": True,
-                    # A stalled write must surface as unavailable rather than
-                    # hold a request open indefinitely.
-                    "options": f"-c statement_timeout={int(statement_timeout_ms)}",
-                },
+                kwargs=kwargs,
                 open=True,
                 timeout=10,
             )