tegwick
c404c910cd
Secret volume rotation swaps ..data. Sequential reads of username then password can tear across two leases. Resolve the snapshot once. Also document why ESO AppRole login cannot parent database/creds leases: the token discard DROP ROLEs the role ESO just stored.
54 lines
2 KiB
Python
54 lines
2 KiB
Python
"""Mounted-directory credential reads (AUDIT-WP-0005-T02)."""
|
|
|
|
from audit_core.credentials import CredentialDirectory
|
|
|
|
|
|
def _write(directory, **fields):
|
|
for name, value in fields.items():
|
|
(directory / name).write_text(value)
|
|
|
|
|
|
def test_reads_a_flat_directory(tmp_path):
|
|
_write(tmp_path, username="u1", password="p1", host="db", port="5432", dbname="audit_core")
|
|
params = CredentialDirectory(tmp_path).read()
|
|
assert params["user"] == "u1"
|
|
assert params["password"] == "p1"
|
|
assert params["host"] == "db"
|
|
|
|
|
|
def test_kubernetes_snapshot_is_not_torn(tmp_path):
|
|
"""A Secret volume swap must not pair lease A's user with lease B's password."""
|
|
first = tmp_path / "..ts1"
|
|
second = tmp_path / "..ts2"
|
|
first.mkdir()
|
|
second.mkdir()
|
|
_write(first, username="lease-a", password="secret-a", host="db", port="5432", dbname="audit")
|
|
_write(second, username="lease-b", password="secret-b", host="db", port="5432", dbname="audit")
|
|
data = tmp_path / "..data"
|
|
data.symlink_to(first.name)
|
|
for name in ("username", "password", "host", "port", "dbname"):
|
|
(tmp_path / name).symlink_to(f"..data/{name}")
|
|
|
|
creds = CredentialDirectory(tmp_path)
|
|
# Flip the snapshot after the directory is opened the way kubelet does:
|
|
# replace ..data to point at the new timestamped directory.
|
|
seen = creds.read()
|
|
data.unlink()
|
|
data.symlink_to(second.name)
|
|
after = creds.read()
|
|
|
|
assert seen["user"] == "lease-a" and seen["password"] == "secret-a"
|
|
assert after["user"] == "lease-b" and after["password"] == "secret-b"
|
|
|
|
|
|
def test_rotation_is_logged_by_fingerprint_not_value(tmp_path, caplog):
|
|
caplog.set_level("INFO", logger="audit_core.credentials")
|
|
_write(tmp_path, username="u", password="one", host="db", port="5432", dbname="audit")
|
|
creds = CredentialDirectory(tmp_path)
|
|
creds.read()
|
|
(tmp_path / "password").write_text("two")
|
|
creds.read()
|
|
text = caplog.text
|
|
assert "rotated" in text
|
|
assert "one" not in text
|
|
assert "two" not in text
|