tegwick
3cd9955ac9
Load pep-stance.yaml as the live unreachable-engine gate and record named stance fields on privileged evidence. Classify evidence, queue load-bearing records in a local outbox, and add heartbeat/drain commands that never sit on a mutation path. Publish proposed SSH-CA and secret-use evidence contracts without adding an OpenBao SSH-CA write. T02 (access-engine decision records) and T06 (no standing credential) stay wait on external endpoints. Assistant: grok Assistant-Session: 01a04cea-cb33-7c63-bad7-c1b0f9f0076b
102 lines
4.2 KiB
Python
102 lines
4.2 KiB
Python
"""§9.6 classification, local outbox, heartbeat, and drain."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import urllib.error
|
|
from types import SimpleNamespace
|
|
|
|
from secrets_engine.evidence import EvidenceWriter, drain_outbox, write_heartbeat
|
|
from secrets_engine.evidence_class import classify
|
|
|
|
|
|
def test_load_bearing_record_is_queued_locally_before_jsonl(tmp_path):
|
|
writer = EvidenceWriter(evidence_dir=tmp_path, hub_url="http://hub.invalid", topic_id="t")
|
|
record = writer.record(
|
|
"lifecycle-destroy",
|
|
result="attempt",
|
|
catalog_id="lane",
|
|
stage="prod",
|
|
detail={"value": "should-not-leak"},
|
|
hub=True,
|
|
)
|
|
assert record["evidence_kind"] == "load-bearing"
|
|
assert record["completeness_claimed"] is False
|
|
assert record["outbox_queued"] is True
|
|
assert record["hub_delivery_requested"] is False
|
|
queued = list((tmp_path / "outbox").glob("*.json"))
|
|
assert len(queued) == 1
|
|
queued_payload = json.loads(queued[0].read_text())
|
|
assert queued_payload["record_id"] == record["record_id"]
|
|
assert "should-not-leak" not in queued[0].read_text()
|
|
jsonl = next(tmp_path.glob("evidence-*.jsonl")).read_text()
|
|
assert record["record_id"] in jsonl
|
|
|
|
|
|
def test_attributive_record_is_not_queued_and_may_post_hub(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(
|
|
"urllib.request.urlopen",
|
|
lambda *_args, **_kwargs: SimpleNamespace(status=200, read=lambda: b"{}"),
|
|
)
|
|
writer = EvidenceWriter(evidence_dir=tmp_path, hub_url="http://hub.invalid", topic_id="t")
|
|
record = writer.record("apply", result="applied", catalog_id="lane", stage="prod")
|
|
assert record["evidence_kind"] == "attributive"
|
|
assert record["hub_delivery_requested"] is True
|
|
assert not list((tmp_path / "outbox").glob("*.json"))
|
|
|
|
|
|
def test_heartbeat_is_a_positive_claim_not_a_permission(tmp_path):
|
|
writer = EvidenceWriter(evidence_dir=tmp_path, hub_url="")
|
|
record = write_heartbeat(writer, stage="prod")
|
|
assert record["action"] == "evidence-heartbeat"
|
|
assert record["result"] == "nothing-to-report"
|
|
assert record["evidence_kind"] == "heartbeat"
|
|
assert record["completeness_claimed"] is False
|
|
assert record["outbox_queued"] is True
|
|
# Heartbeat must not be used as "no jsonl means nothing happened".
|
|
assert classify("evidence-heartbeat", "prod").kind == "heartbeat"
|
|
|
|
|
|
def test_drain_without_audit_core_keeps_files(tmp_path):
|
|
writer = EvidenceWriter(evidence_dir=tmp_path, hub_url="")
|
|
writer.record("provision", result="from-file", catalog_id="lane", stage="prod")
|
|
result = drain_outbox(writer, audit_core_url="")
|
|
assert result["queued"] == 1
|
|
assert result["skipped"] == 1
|
|
assert result["delivered"] == 0
|
|
assert result["completeness_claimed"] is False
|
|
assert list((tmp_path / "outbox").glob("*.json"))
|
|
|
|
|
|
def test_drain_audit_core_outage_does_not_raise_or_delete(tmp_path, monkeypatch):
|
|
writer = EvidenceWriter(evidence_dir=tmp_path, hub_url="")
|
|
writer.record("revoke", result="native-access-deactivated", catalog_id="lane", stage="prod")
|
|
|
|
def offline(*_args, **_kwargs):
|
|
raise urllib.error.URLError("audit-core down")
|
|
|
|
monkeypatch.setattr("urllib.request.urlopen", offline)
|
|
result = drain_outbox(writer, audit_core_url="http://audit-core.invalid")
|
|
assert result["failed"] == 1
|
|
assert result["delivered"] == 0
|
|
assert list((tmp_path / "outbox").glob("*.json"))
|
|
|
|
|
|
def test_successful_drain_removes_only_delivered_files(tmp_path, monkeypatch):
|
|
writer = EvidenceWriter(evidence_dir=tmp_path, hub_url="")
|
|
writer.record("provision", result="from-file", catalog_id="lane", stage="prod")
|
|
monkeypatch.setattr(
|
|
"urllib.request.urlopen",
|
|
lambda *_args, **_kwargs: SimpleNamespace(status=202, read=lambda: b"{}"),
|
|
)
|
|
result = drain_outbox(writer, audit_core_url="http://audit-core.invalid")
|
|
assert result["delivered"] == 1
|
|
assert not list((tmp_path / "outbox").glob("*.json"))
|
|
|
|
|
|
def test_classify_is_not_a_permission_api():
|
|
"""Missing or unknown actions are attributive labels, never a deny."""
|
|
unknown = classify("not-a-control", "prod")
|
|
assert unknown.kind == "attributive"
|
|
assert unknown.completeness_claimed is False
|
|
assert "permit" not in dir(classify)
|
|
assert "deny" not in dir(classify)
|