tegwick
de9e3abe5f
T02. deploy/attest-cronjob.yaml: daily at 03:17 UTC against the 168h window, its own ServiceAccount, and a Role reaching exactly one named ConfigMap — get/update/patch, no create, no list. audit_core/attest_publish.py does the publish in stdlib; the image carries no kubectl, and adding one to an audit receiver's image to write a single file is the worse trade. Three refusals, all deliberate: The producer is not the receiver. A receiver that could rewrite its own attestation could forge it. audit-core-egress is now scoped to component: receiver and a separate audit-core-attest-egress carries the 6443 rule, so the receiver never gains API-server reach. Asserted by test. It refuses to publish over a broken chain. A fresh head written over a break replaces an honest chain_break with a fresh-looking attestation. Stale degrades the claim visibly; false does not. Mounted as a directory, not subPath. Found while writing the manifest: a subPath ConfigMap mount is resolved once at pod start and never updates, so the daily attestation would land in the ConfigMap and never reach the running receiver — tamper_evidence would age out to false while the job reported success every night, silent in both directions. The offsite copy stays an operator step. audit-core holds no Nextcloud credential and should not acquire one to publish a hash, so docs/integrity.md states the bound plainly: until that copy exists the delivered control defends against a database owner, not a cluster owner, and no stronger claim may be made from it. T10. layer.yaml lists four infrastructure contacts — platform-pg, state-hub, kube-apiserver, the container registry — each with its role and whether another layer reads it. tooling_contacts stays [], which is true under §5 as written; the companion's totality request is met by the uncatalogued list rather than by inventing a Tooling row. tests/test_layer_conformance.py derives the egress destinations from the manifests and the registry from the pinned digests, so a new contact appearing in deploy/ without a row fails the test rather than waiting for a reviewer to notice. Applying the manifests remains an operator action; nothing here was applied. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Nb7Q6ZmXppNDkTWytfYqfv Assistant: claude-code Assistant-Model: opus Assistant-Process: 2069992@bnt-lap001 Assistant-Session: 167dd7f8-2a25-4be1-aa46-3b6f1a5f94c6
94 lines
2.9 KiB
Python
94 lines
2.9 KiB
Python
"""AUDIT-WP-0009-T02. Publishing the chain-head attestation."""
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from audit_core import attest_publish
|
|
|
|
|
|
class _Response:
|
|
status = 200
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
return False
|
|
|
|
|
|
@pytest.fixture()
|
|
def service_account(tmp_path):
|
|
(tmp_path / "token").write_text("sa-token\n")
|
|
(tmp_path / "namespace").write_text("audit-core\n")
|
|
(tmp_path / "ca.crt").write_text("")
|
|
return str(tmp_path)
|
|
|
|
|
|
def test_publish_patches_one_key_with_the_projected_token(service_account):
|
|
seen = {}
|
|
|
|
def opener(request):
|
|
seen["url"] = request.full_url
|
|
seen["method"] = request.method
|
|
seen["headers"] = {k.lower(): v for k, v in request.headers.items()}
|
|
seen["body"] = json.loads(request.data.decode())
|
|
return _Response()
|
|
|
|
status = attest_publish.publish(
|
|
{"head": "abc", "observed_at": "2026-09-10T03:17:00+00:00"},
|
|
directory=service_account,
|
|
host="https://api.test",
|
|
opener=opener,
|
|
)
|
|
|
|
assert status == 200
|
|
assert seen["url"] == "https://api.test/api/v1/namespaces/audit-core/configmaps/audit-core-chain-head"
|
|
assert seen["method"] == "PATCH"
|
|
assert seen["headers"]["authorization"] == "Bearer sa-token"
|
|
# A merge patch replaces one key. A full PUT would drop anything else the
|
|
# operator put in the ConfigMap.
|
|
assert seen["headers"]["content-type"] == "application/merge-patch+json"
|
|
assert set(seen["body"]) == {"data"}
|
|
assert set(seen["body"]["data"]) == {"chain-head.json"}
|
|
assert json.loads(seen["body"]["data"]["chain-head.json"])["head"] == "abc"
|
|
|
|
|
|
def test_publish_raises_rather_than_returning_a_failure(service_account):
|
|
"""A silent failure leaves a stale attestation aging out with nobody told."""
|
|
|
|
def opener(request):
|
|
raise OSError("apiserver unreachable")
|
|
|
|
with pytest.raises(OSError):
|
|
attest_publish.publish(
|
|
{"head": "abc"}, directory=service_account,
|
|
host="https://api.test", opener=opener,
|
|
)
|
|
|
|
|
|
def test_a_broken_chain_is_not_published_over(monkeypatch, tmp_path, capsys):
|
|
"""The refusal that matters: a fresh head over a break would hide it."""
|
|
|
|
class _Report:
|
|
intact = False
|
|
first_break = "event-42"
|
|
|
|
class _Backend:
|
|
def verify_chain(self):
|
|
return _Report()
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
published = []
|
|
monkeypatch.setattr(attest_publish, "publish", lambda *a, **k: published.append(a))
|
|
monkeypatch.setenv("AUDIT_CORE_ATTESTATION_OUTPUT", str(tmp_path / "head.json"))
|
|
monkeypatch.setattr("audit_core.cli._postgres_backend", lambda *a, **k: _Backend())
|
|
monkeypatch.setattr(
|
|
"audit_core.integrity.write_attestation", lambda path, report: {"head": "x"}
|
|
)
|
|
|
|
assert attest_publish.main([]) == 1
|
|
assert published == []
|
|
assert "refusing to publish" in capsys.readouterr().err
|