fix(compose): retain mounted secret file authority

pg_llm_batch/compose_bootstrap.py

@@ -12,6 +12,8 @@
 from __future__ import annotations
 
 import argparse
+import os
+import stat
 from pathlib import Path
 from typing import Sequence
 
@@ -23,16 +25,59 @@
 
 _DEFAULT_PASSWORD_FILE = Path("/run/secrets/postgres_password")
 _MAX_PASSWORD_BYTES = 65_536
+_SECRET_UNAVAILABLE = "The mounted PostgreSQL password secret is unavailable."
+
+
+def _secret_file_metadata(secret_stat: os.stat_result) -> tuple[int, ...]:
+    """Return observable metadata used to detect mounted-secret mutation."""
+    return (
+        secret_stat.st_mode,
+        secret_stat.st_size,
+        secret_stat.st_nlink,
+        secret_stat.st_uid,
+        secret_stat.st_gid,
+        secret_stat.st_dev,
+        secret_stat.st_ino,
+        secret_stat.st_mtime_ns,
+        secret_stat.st_ctime_ns,
+    )
 
 
 def _load_database_password(password_file: Path) -> str:
-    """Read one bounded UTF-8 password from an explicitly mounted secret file."""
+    """Read one bounded UTF-8 password from an exact regular secret-file object."""
+    try:
+        secure_flags = os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW | os.O_NONBLOCK
+    except AttributeError:
+        raise ConfigError(_SECRET_UNAVAILABLE) from None
+
     try:
-        with password_file.open("rb") as secret_stream:
-            raw_password = secret_stream.read(_MAX_PASSWORD_BYTES + 1)
+        secret_fd = os.open(password_file, secure_flags)
     except OSError:
-        raise ConfigError("The mounted PostgreSQL password secret is unavailable.") from None
+        raise ConfigError(_SECRET_UNAVAILABLE) from None
 
+    raw_password = b""
+    unavailable = False
+    try:
+        try:
+            secret_stat = os.fstat(secret_fd)
+            if not stat.S_ISREG(secret_stat.st_mode):
+                unavailable = True
+            else:
+                initial_metadata = _secret_file_metadata(secret_stat)
+                with os.fdopen(secret_fd, "rb", closefd=False) as secret_stream:
+                    raw_password = secret_stream.read(_MAX_PASSWORD_BYTES + 1)
+                if _secret_file_metadata(os.fstat(secret_fd)) != initial_metadata:
+                    unavailable = True
+        except OSError:
+            unavailable = True
+    finally:
+        try:
+            os.close(secret_fd)
+        except OSError:
+            unavailable = True
+
+    if unavailable:
+        raise ConfigError(_SECRET_UNAVAILABLE) from None
     if not raw_password:
         raise ConfigError("The mounted PostgreSQL password secret is empty.")
     if len(raw_password) > _MAX_PASSWORD_BYTES:

tests/test_compose_bootstrap_secret_file_authority.py

@@ -0,0 +1,123 @@
+# SPDX-License-Identifier: Apache-2.0
+"""Regression tests for mounted Compose secret file authority."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import pytest
+
+from pg_llm_batch import compose_bootstrap
+from pg_llm_batch.exceptions import ConfigError
+
+
+def test_database_password_loader_rejects_final_symlink(tmp_path: Path) -> None:
+    """A mounted-secret pathname cannot redirect authority through a symlink."""
+    secret_text = "private-compose-password"
+    target = tmp_path / "actual-password"
+    target.write_text(secret_text, encoding="utf-8")
+    mounted_path = tmp_path / "mounted-password"
+    mounted_path.symlink_to(target)
+
+    with pytest.raises(ConfigError, match="unavailable") as caught:
+        compose_bootstrap._load_database_password(mounted_path)
+
+    assert secret_text not in str(caught.value)
+    assert caught.value.__cause__ is None
+
+
+def test_database_password_loader_rejects_fifo_without_blocking(tmp_path: Path) -> None:
+    """A non-regular mounted object fails closed before any secret read."""
+    fifo_path = tmp_path / "mounted-password"
+    os.mkfifo(fifo_path)
+
+    with pytest.raises(ConfigError, match="unavailable") as caught:
+        compose_bootstrap._load_database_password(fifo_path)
+
+    assert caught.value.__cause__ is None
+
+
+def test_database_password_loader_fails_closed_without_secure_open_flag(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A platform without no-follow authority cannot silently downgrade the open."""
+    password_file = tmp_path / "database-password"
+    password_file.write_text("private-compose-password", encoding="utf-8")
+    monkeypatch.delattr(compose_bootstrap.os, "O_NOFOLLOW")
+
+    with pytest.raises(ConfigError, match="unavailable") as caught:
+        compose_bootstrap._load_database_password(password_file)
+
+    assert caught.value.__cause__ is None
+
+
+def test_database_password_loader_normalizes_fstat_failure(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Descriptor metadata failures stay content-free and close the retained fd."""
+    password_file = tmp_path / "database-password"
+    password_file.write_text("private-compose-password", encoding="utf-8")
+
+    def fail_fstat(_fd: int) -> os.stat_result:
+        raise OSError("sensitive stat diagnostic")
+
+    monkeypatch.setattr(compose_bootstrap.os, "fstat", fail_fstat)
+
+    with pytest.raises(ConfigError, match="unavailable") as caught:
+        compose_bootstrap._load_database_password(password_file)
+
+    assert "sensitive stat diagnostic" not in str(caught.value)
+    assert caught.value.__cause__ is None
+
+
+def test_database_password_loader_rejects_content_mutation_after_initial_stat(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Secret bytes cannot change after inspection and still become password authority."""
+    password_file = tmp_path / "database-password"
+    secret_text = "private-compose-password"
+    password_file.write_text(secret_text, encoding="utf-8")
+    real_fstat = compose_bootstrap.os.fstat
+    mutated = False
+
+    def mutate_after_first_fstat(fd: int) -> os.stat_result:
+        nonlocal mutated
+        status = real_fstat(fd)
+        if not mutated:
+            mutated = True
+            with password_file.open("ab") as stream:
+                stream.write(b"-mutated")
+        return status
+
+    monkeypatch.setattr(compose_bootstrap.os, "fstat", mutate_after_first_fstat)
+
+    with pytest.raises(ConfigError, match="unavailable") as caught:
+        compose_bootstrap._load_database_password(password_file)
+
+    assert secret_text not in str(caught.value)
+    assert "mutated" not in str(caught.value)
+    assert caught.value.__cause__ is None
+
+
+def test_database_password_loader_normalizes_close_failure(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A descriptor-close failure invalidates otherwise valid secret evidence."""
+    password_file = tmp_path / "database-password"
+    secret_text = "private-compose-password"
+    password_file.write_text(secret_text, encoding="utf-8")
+    real_close = compose_bootstrap.os.close
+
+    def close_then_fail(fd: int) -> None:
+        real_close(fd)
+        raise OSError("sensitive close diagnostic")
+
+    monkeypatch.setattr(compose_bootstrap.os, "close", close_then_fail)
+
+    with pytest.raises(ConfigError, match="unavailable") as caught:
+        compose_bootstrap._load_database_password(password_file)
+
+    assert secret_text not in str(caught.value)
+    assert "sensitive close diagnostic" not in str(caught.value)
+    assert caught.value.__cause__ is None