9b658e3f19
The OpenBao policy matches the reviewed file. Both publisher KV paths are still empty because the bootstrap key cannot create IAM applications. Assistant: grok Assistant-Session: 01a11cdb-f9fe-7011-947c-936e72007e54
138 lines
4.9 KiB
Python
138 lines
4.9 KiB
Python
"""Pair custody stays preview-only until a writer identity is admitted."""
|
|
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
spec = importlib.util.spec_from_file_location(
|
|
"pair_custody", ROOT / "scripts/application_backup_pair_custody.py"
|
|
)
|
|
custody = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(custody)
|
|
|
|
PAIR = {
|
|
"AWS_ACCESS_KEY_ID": "SYNTHETICACCESSKEY123",
|
|
"AWS_SECRET_ACCESS_KEY": "SYNTHETICSECRETKEY4567",
|
|
}
|
|
|
|
|
|
class Transport:
|
|
def __init__(self):
|
|
self.calls = []
|
|
self.metadata_result = None
|
|
self.version = 1
|
|
|
|
def metadata(self, path):
|
|
self.calls.append(("metadata", path))
|
|
return self.metadata_result
|
|
|
|
def put_cas0(self, path, body):
|
|
self.calls.append(("put", path, body["options"]["cas"], tuple(body["data"])))
|
|
return {"version": self.version, "request_id": "synthetic-request"}
|
|
|
|
def revoke(self):
|
|
self.calls.append(("revoke",))
|
|
|
|
|
|
def identity():
|
|
return {
|
|
"policies": [custody.WRITER_POLICY],
|
|
"identity_policies": [],
|
|
"ttl": 300,
|
|
"num_uses": 2,
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize("application", sorted(custody.PROFILES))
|
|
def test_preview_reads_draft_contract_and_writes_nothing(application, capsys, monkeypatch):
|
|
monkeypatch.setattr(custody, "provision", lambda *_: pytest.fail("preview provisioned"))
|
|
assert custody.main(["--application", application]) == 0
|
|
receipt = json.loads(capsys.readouterr().out)
|
|
assert receipt["status"] == "preview"
|
|
assert receipt["ccr_status"] == "draft"
|
|
assert receipt["writer_admitted"] is True
|
|
assert receipt["openbao_write"] is False
|
|
assert receipt["provider_keys"] == "not_created"
|
|
assert receipt["effects"] == []
|
|
assert receipt["fields"] == list(custody.FIELDS)
|
|
rendered = json.dumps(receipt)
|
|
assert "SYNTHETIC" not in rendered
|
|
assert "AKIA" not in rendered
|
|
|
|
|
|
def test_unreviewed_writer_makes_no_transport_call(monkeypatch):
|
|
monkeypatch.setattr(custody, "WRITER_ADMITTED", False)
|
|
transport = Transport()
|
|
with pytest.raises(custody.CustodyError, match="writer_identity_unreviewed"):
|
|
custody.provision("email-connect", PAIR, identity(), transport)
|
|
assert transport.calls == []
|
|
|
|
|
|
def test_admitted_writer_uses_one_cas0_write_and_revokes(monkeypatch):
|
|
monkeypatch.setattr(custody, "WRITER_ADMITTED", True)
|
|
transport = Transport()
|
|
receipt = custody.provision("email-connect", dict(PAIR), identity(), transport)
|
|
assert receipt["kv_version"] == 1
|
|
assert receipt["secret_material"] is False
|
|
assert PAIR["AWS_SECRET_ACCESS_KEY"] not in json.dumps(receipt)
|
|
assert transport.calls == [
|
|
("metadata", "platform/metadata/workloads/email-connect/backup-primary"),
|
|
(
|
|
"put",
|
|
"platform/data/workloads/email-connect/backup-primary",
|
|
0,
|
|
custody.FIELDS,
|
|
),
|
|
("revoke",),
|
|
]
|
|
|
|
|
|
def test_reader_policy_and_existing_version_are_refused(monkeypatch):
|
|
monkeypatch.setattr(custody, "WRITER_ADMITTED", True)
|
|
reader = dict(identity(), policies=["workload-kv-read-email-connect-backup-primary"])
|
|
with pytest.raises(custody.CustodyError, match="writer_identity_unreviewed"):
|
|
custody.provision("email-connect", PAIR, reader, Transport())
|
|
transport = Transport()
|
|
transport.metadata_result = {"current_version": 1}
|
|
with pytest.raises(custody.CustodyError, match="existing_custody_requires_rotation_review"):
|
|
custody.provision("email-connect", PAIR, identity(), transport)
|
|
assert [call[0] for call in transport.calls] == ["metadata"]
|
|
|
|
|
|
def test_writer_policy_is_pair_scoped_and_cannot_read_or_delete():
|
|
text = (ROOT / "openbao/policies/backup-pair-writer.hcl").read_text(encoding="utf-8")
|
|
for application in custody.PROFILES:
|
|
assert f'path "platform/data/workloads/{application}/backup-primary"' in text
|
|
assert f'path "platform/metadata/workloads/{application}/backup-primary"' in text
|
|
capabilities = [
|
|
line.split("=", 1)[1]
|
|
for line in text.splitlines()
|
|
if line.strip().startswith("capabilities")
|
|
]
|
|
assert capabilities
|
|
assert all("delete" not in line and "list" not in line for line in capabilities)
|
|
assert 'capabilities = ["read"]' in text
|
|
data_blocks = [
|
|
block
|
|
for block in text.split("path ")
|
|
if "platform/data/" in block
|
|
]
|
|
assert data_blocks
|
|
assert all('["create", "update"]' in block and "read" not in block for block in data_blocks)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"values",
|
|
[
|
|
{"AWS_ACCESS_KEY_ID": PAIR["AWS_ACCESS_KEY_ID"]},
|
|
dict(PAIR, AWS_SECRET_ACCESS_KEY="short"),
|
|
dict(PAIR, AWS_ACCESS_KEY_ID=PAIR["AWS_SECRET_ACCESS_KEY"]),
|
|
dict(PAIR, EXTRA="SYNTHETICEXTRAVALUE1"),
|
|
],
|
|
)
|
|
def test_partial_or_embedded_pairs_are_refused(values):
|
|
with pytest.raises(custody.CustodyError):
|
|
custody.validate_pair(values)
|