1900e24f80
Record founder acceptance on CCR-2026-0036 and CCR-2026-0037, install the reader roles, and admit only the email-connect node delivery. Treat a missing OpenBao token period as non-periodic. Assistant: grok Assistant-Session: 01a11cdb-f9fe-7011-947c-936e72007e54
351 lines
13 KiB
Python
351 lines
13 KiB
Python
"""Synthetic credential boundaries and interrupted-delivery recovery, no live calls."""
|
|
|
|
import base64
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
spec = importlib.util.spec_from_file_location(
|
|
"delivery", ROOT / "scripts/application_backup_credentials.py"
|
|
)
|
|
d = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(d)
|
|
|
|
KEY = "SYNTHETIC_ACCESS_12345"
|
|
SECRET = "SYNTHETIC_SECRET_987654321"
|
|
TOKEN = "SYNTHETIC_BAO_READER_12345"
|
|
JWT = "SYNTHETIC_KUBERNETES_JWT_12345"
|
|
|
|
|
|
def source(application="coulomb-social"):
|
|
return (
|
|
ROOT
|
|
/ "credential-change-requests"
|
|
/ f"{d.PROFILES[application]['ccr']}-{application}-backup-primary.yaml"
|
|
)
|
|
|
|
|
|
def approved(application="coulomb-social"):
|
|
ccr = json.loads(source(application).read_bytes())
|
|
ccr["status"] = "applied"
|
|
ccr["openbao"]["auth"]["bound_claims_confirmed"] = True
|
|
ccr["review"]["comments"] = [
|
|
{
|
|
"reviewer": f"User ({owner})",
|
|
"decision": "approved",
|
|
"source_ref": "synthetic-owner-proof",
|
|
}
|
|
for owner in ("platform-owner", f"{application}-owner")
|
|
]
|
|
ccr["delivery"]["admission"] = {
|
|
"node_executor_approved": True,
|
|
"provider_scope_verified": True,
|
|
"reader_boundary_verified": True,
|
|
"source_ref": "synthetic-proof",
|
|
}
|
|
return ccr
|
|
|
|
|
|
class FakeNative:
|
|
def __init__(self, application="coulomb-social"):
|
|
self.application = application
|
|
self.calls = []
|
|
policy = d.expected_openbao(application)["policy_name"]
|
|
self.login = {"auth": {"client_token": TOKEN, "policies": [policy], "lease_duration": 300}}
|
|
self.info = {
|
|
"data": {
|
|
"policies": [policy],
|
|
"identity_policies": [],
|
|
"ttl": 299,
|
|
"period": 0,
|
|
"explicit_max_ttl": 300,
|
|
"num_uses": 3,
|
|
}
|
|
}
|
|
self.record = {
|
|
"data": {
|
|
"data": {d.FIELDS[0]: KEY, d.FIELDS[1]: SECRET},
|
|
"metadata": {"version": 1, "destroyed": False, "deletion_time": ""},
|
|
}
|
|
}
|
|
self.fail = None
|
|
|
|
def identity(self, application):
|
|
assert application == self.application
|
|
self.calls.append(("identity", application, None))
|
|
return JWT
|
|
|
|
def request(self, method, path, payload=None, token=None):
|
|
self.calls.append((method, path, token))
|
|
if path == self.fail:
|
|
raise RuntimeError(SECRET + TOKEN + JWT)
|
|
if path == "auth/kubernetes/login":
|
|
assert payload == {
|
|
"role": d.expected_openbao(self.application)["auth"]["role"],
|
|
"jwt": JWT,
|
|
}
|
|
return self.login
|
|
assert token == TOKEN
|
|
if path == "auth/token/lookup-self":
|
|
return self.info
|
|
if path.startswith("platform/data/"):
|
|
assert path == f"platform/data/workloads/{self.application}/backup-primary"
|
|
return self.record
|
|
assert path == "auth/token/revoke-self"
|
|
return {}
|
|
|
|
|
|
@pytest.fixture
|
|
def directory(tmp_path):
|
|
tmp_path.chmod(0o700)
|
|
fd = d.private_directory(tmp_path, uid=os.getuid())
|
|
yield tmp_path, fd
|
|
os.close(fd)
|
|
|
|
|
|
@pytest.mark.parametrize("application", d.PROFILES)
|
|
def test_draft_preview_has_no_effects(application, capsys, monkeypatch):
|
|
monkeypatch.setattr(d, "Native", lambda: pytest.fail("preview made a native client"))
|
|
assert d.main(["--application", application, "--ccr", str(source(application))]) == 0
|
|
receipt = json.loads(capsys.readouterr().out)
|
|
assert receipt["status"] == "preview" and receipt["effects"] == []
|
|
assert receipt["target"] == d.PROFILES[application]["directory"] + "/primary.env"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"change", ["draft", "unconfirmed", "hash", "owner", "admission", "sibling", "default"]
|
|
)
|
|
def test_live_contract_refuses_missing_authority_before_identity(change):
|
|
ccr = approved()
|
|
if change == "draft":
|
|
ccr["status"] = "draft"
|
|
elif change == "unconfirmed":
|
|
ccr["openbao"]["auth"]["bound_claims_confirmed"] = False
|
|
elif change == "owner":
|
|
ccr["review"]["comments"].pop()
|
|
elif change == "admission":
|
|
ccr["delivery"]["admission"]["provider_scope_verified"] = False
|
|
elif change == "sibling":
|
|
ccr["openbao"]["kv_path"] = "platform/workloads/email-connect/backup-primary"
|
|
elif change == "default":
|
|
ccr["openbao"]["auth"]["policies"].append("default")
|
|
raw = json.dumps(ccr).encode()
|
|
digest = "0" * 64 if change == "hash" else hashlib.sha256(raw).hexdigest()
|
|
with pytest.raises(d.DeliveryError):
|
|
d.contract(raw, "coulomb-social", True, digest)
|
|
|
|
|
|
@pytest.mark.parametrize("application", d.PROFILES)
|
|
def test_atomic_private_delivery_and_identical_resume(application, directory, capsys):
|
|
path, fd = directory
|
|
ccr = approved(application)
|
|
native = FakeNative(application)
|
|
receipts = [
|
|
d.deliver(ccr, "source-only-hash", application, 1, native, fd, os.getuid())
|
|
for _ in range(2)
|
|
]
|
|
assert [r["outcome"] for r in receipts] == ["created", "unchanged"]
|
|
assert (
|
|
path / "primary.env"
|
|
).read_bytes() == f"{d.FIELDS[0]}={KEY}\n{d.FIELDS[1]}={SECRET}\n".encode()
|
|
assert (path / "primary.env").stat().st_mode & 0o777 == 0o600
|
|
assert len(list(path.glob("delivery-v1-*.json"))) == 2
|
|
for receipt in receipts:
|
|
serialized = json.dumps(receipt)
|
|
assert all(secret not in serialized for secret in (KEY, SECRET, TOKEN, JWT))
|
|
assert receipt["readback_equal"] and receipt["token_revoked"]
|
|
assert all(secret not in capsys.readouterr().out for secret in (KEY, SECRET, TOKEN, JWT))
|
|
assert native.calls[4][1] == "auth/token/revoke-self"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"change",
|
|
["default", "identity", "unbounded", "uses", "period", "version", "destroyed", "fields", "injection"],
|
|
)
|
|
def test_scope_and_payload_failures_revoke_without_publishing(change, directory):
|
|
path, fd = directory
|
|
native = FakeNative()
|
|
if change == "default":
|
|
native.login["auth"]["policies"].append("default")
|
|
elif change == "identity":
|
|
native.info["data"]["identity_policies"] = ["platform-admin"]
|
|
elif change == "unbounded":
|
|
native.info["data"]["explicit_max_ttl"] = 0
|
|
elif change == "uses":
|
|
native.info["data"]["num_uses"] = 0
|
|
elif change == "period":
|
|
native.info["data"]["period"] = 60
|
|
elif change == "version":
|
|
native.record["data"]["metadata"]["version"] = 2
|
|
elif change == "destroyed":
|
|
native.record["data"]["metadata"]["destroyed"] = True
|
|
elif change == "fields":
|
|
native.record["data"]["data"]["ESCROW"] = SECRET
|
|
else:
|
|
native.record["data"]["data"][d.FIELDS[1]] = SECRET + "\nOTHER=INJECTED"
|
|
with pytest.raises(d.DeliveryError):
|
|
d.deliver(approved(), "source-hash", "coulomb-social", 1, native, fd, os.getuid())
|
|
assert native.calls[-1][1] == "auth/token/revoke-self"
|
|
assert not (path / "primary.env").exists() and not list(path.glob("delivery-v*.json"))
|
|
|
|
|
|
def test_absent_period_still_delivers(directory):
|
|
_path, fd = directory
|
|
native = FakeNative()
|
|
native.info["data"]["period"] = None
|
|
receipt = d.deliver(approved(), "source-hash", "coulomb-social", 1, native, fd, os.getuid())
|
|
assert receipt["outcome"] == "created"
|
|
|
|
|
|
def test_revocation_failure_does_not_publish_or_disclose(directory, capsys, monkeypatch, tmp_path):
|
|
path, fd = directory
|
|
ccr = approved()
|
|
candidate = tmp_path / "contract.json"
|
|
raw = json.dumps(ccr).encode()
|
|
candidate.write_bytes(raw)
|
|
native = FakeNative()
|
|
native.fail = "auth/token/revoke-self"
|
|
monkeypatch.setattr(d.os, "geteuid", lambda: 0)
|
|
monkeypatch.setattr(d, "private_directory", lambda *_: os.dup(fd))
|
|
monkeypatch.setattr(d, "Native", lambda: native)
|
|
original_deliver = d.deliver
|
|
monkeypatch.setattr(d, "deliver", lambda *a: original_deliver(*a, uid=os.getuid()))
|
|
assert (
|
|
d.main(
|
|
[
|
|
"--application",
|
|
"coulomb-social",
|
|
"--ccr",
|
|
str(candidate),
|
|
"--execute",
|
|
"--expected-ccr-sha256",
|
|
hashlib.sha256(raw).hexdigest(),
|
|
"--expected-kv-version",
|
|
"1",
|
|
]
|
|
)
|
|
== 1
|
|
)
|
|
output = capsys.readouterr()
|
|
assert "token_revocation_unconfirmed" in output.err
|
|
assert all(secret not in output.out + output.err for secret in (KEY, SECRET, TOKEN, JWT))
|
|
assert not (path / "primary.env").exists()
|
|
|
|
|
|
@pytest.mark.parametrize("existing", ["different", "insecure", "symlink", "hardlink", "fifo"])
|
|
def test_existing_target_is_preserved_and_unsafe_files_refused(existing, directory):
|
|
path, fd = directory
|
|
target = path / "primary.env"
|
|
if existing == "symlink":
|
|
victim = path / "victim"
|
|
victim.write_bytes(b"previous")
|
|
target.symlink_to(victim)
|
|
elif existing == "fifo":
|
|
os.mkfifo(target, 0o600)
|
|
else:
|
|
target.write_bytes(b"previous")
|
|
target.chmod(0o644 if existing == "insecure" else 0o600)
|
|
if existing == "hardlink":
|
|
os.link(target, path / "alias")
|
|
with pytest.raises((d.DeliveryError, OSError)):
|
|
d.deliver(approved(), "source-hash", "coulomb-social", 1, FakeNative(), fd, os.getuid())
|
|
assert not list(path.glob("delivery-v*.json"))
|
|
if existing != "fifo":
|
|
assert target.read_bytes() == b"previous"
|
|
|
|
|
|
def test_secret_write_failure_removes_temporary_and_is_resumable(directory, monkeypatch):
|
|
path, fd = directory
|
|
original = d.os.link
|
|
monkeypatch.setattr(d.os, "link", lambda *a, **k: (_ for _ in ()).throw(OSError(SECRET)))
|
|
with pytest.raises(OSError):
|
|
d.deliver(approved(), "source-hash", "coulomb-social", 1, FakeNative(), fd, os.getuid())
|
|
assert not list(path.glob(".primary.env.*")) and not (path / "primary.env").exists()
|
|
monkeypatch.setattr(d.os, "link", original)
|
|
assert (
|
|
d.deliver(approved(), "source-hash", "coulomb-social", 1, FakeNative(), fd, os.getuid())[
|
|
"status"
|
|
]
|
|
== "delivered"
|
|
)
|
|
|
|
|
|
def test_receipt_failure_resume_keeps_identical_secret(directory, monkeypatch):
|
|
path, fd = directory
|
|
original = d.publish
|
|
|
|
def fail_receipt(fd, name, body, uid):
|
|
if name.startswith("delivery-v"):
|
|
raise OSError(SECRET)
|
|
return original(fd, name, body, uid)
|
|
|
|
monkeypatch.setattr(d, "publish", fail_receipt)
|
|
with pytest.raises(OSError):
|
|
d.deliver(approved(), "source-hash", "coulomb-social", 1, FakeNative(), fd, os.getuid())
|
|
before = (path / "primary.env").read_bytes()
|
|
monkeypatch.setattr(d, "publish", original)
|
|
assert (
|
|
d.deliver(approved(), "source-hash", "coulomb-social", 1, FakeNative(), fd, os.getuid())[
|
|
"outcome"
|
|
]
|
|
== "unchanged"
|
|
)
|
|
assert (path / "primary.env").read_bytes() == before
|
|
|
|
|
|
def test_native_wrong_cluster_stops_before_token_request(monkeypatch):
|
|
calls = []
|
|
|
|
def command(argv):
|
|
calls.append(argv)
|
|
return json.dumps({"metadata": {"uid": "wrong-cluster"}}).encode()
|
|
|
|
monkeypatch.setattr(d, "command", command)
|
|
with pytest.raises(d.DeliveryError, match="cluster_identity_mismatch"):
|
|
d.Native().identity("coulomb-social")
|
|
assert len(calls) == 1 and "create" not in calls[0]
|
|
|
|
|
|
def test_native_tokenrequest_has_no_secret_argv_or_secret_get(monkeypatch):
|
|
calls = []
|
|
claims = {
|
|
"sub": f"system:serviceaccount:coulomb-social:{d.SA}",
|
|
"aud": ["openbao"],
|
|
"exp": time.time() + 600,
|
|
"kubernetes.io": {"serviceaccount": {"uid": "account-uid"}},
|
|
}
|
|
jwt = (
|
|
"header."
|
|
+ base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip("=")
|
|
+ ".signature"
|
|
)
|
|
responses = [
|
|
{"metadata": {"uid": d.CLUSTER_UID}},
|
|
{"spec": {"clusterIP": "10.43.12.34", "ports": [{"port": 8200}]}},
|
|
{"metadata": {"uid": "account-uid"}, "automountServiceAccountToken": False},
|
|
]
|
|
|
|
def command(argv):
|
|
calls.append(argv)
|
|
return json.dumps(responses.pop(0)).encode() if responses else jwt.encode()
|
|
|
|
monkeypatch.setattr(d, "command", command)
|
|
native = d.Native()
|
|
assert native.identity("coulomb-social") == jwt
|
|
assert native.address == "http://10.43.12.34:8200/v1/"
|
|
assert calls[-1][-3:] == [d.SA, "--audience=openbao", "--duration=10m"]
|
|
assert all("secret" not in arg.lower() and jwt not in arg for call in calls for arg in call)
|
|
|
|
|
|
def test_private_directory_rejects_symlink_ancestor(tmp_path):
|
|
real = tmp_path / "real"
|
|
real.mkdir(mode=0o700)
|
|
(tmp_path / "alias").symlink_to(real)
|
|
with pytest.raises(OSError):
|
|
d.private_directory(tmp_path / "alias", os.getuid())
|