55 lines
2 KiB
Python
55 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
|