Files
acdl/tests/test_hitl_gates.py
T
Jon Chery e74a8c2f5d feat(P42): stub implementation — SoD, HITL gates, attestation matrix, Wiz, Kyverno
---ci---
project: acdl
phase: 42
milestone: v1.9
status: execute
---/ci---

Phase 42 — stub-implementation (REQ-107..111, D-084):

route_halt_artifact (REQ-107):
- core/separation_of_duties.py: real SNS publish (ACDL_SOD_HALT_TOPIC_ARN)
  + outbox fallback (SEPARATION_OF_DUTIES_VIOLATION event via
  outbox_writer) + stderr emission. No silent print-only stub.
- terraform/platform/main.tf: aws_sns_topic.acdl-sod-halt + output.

HITL attestation gates (REQ-108):
- core/hitl_gates.py: attest(contract_id, env, approver, evidence,
  outbox_client) records approver_qa/approver_prod/approver_dr to
  outbox, runs SoD check on prod, invokes attestation matrix, returns
  (ok, reason). Dev skips (autonomous). approver_from_env() reads
  GITHUB_ACTOR/GITEA_ACTOR.
- scripts/run_platform.sh: Step 7b HITL gate before apply for qa/prod/dr.

8-concern attestation matrix (REQ-109, D-084):
- core/attestation_matrix.py: check(env, evidence) runs the 8 concerns
  from hitl_matrix_design.md §10.4. Offline-testable (contract_nfrs,
  schema_validity, policy_pass) run for real. Operator-supplied accept
  signed artifacts validated for freshness (FRESHNESS_DAYS table) +
  schema. Signature skip when ACDL_ATTESTATION_SIGNING_KEY_ID unset
  (D-089). Fail loud if missing/expired for prod/dr.

Wiz real client (REQ-110):
- adapters/wiz/wiz_adapter.py: WizClient (GraphQL API, Bearer auth,
  pagination via pageInfo.hasNextPage + endCursor). fetch_and_adapt
  translates issues → PolicyCheckResult; graceful degrade when
  WIZ_API_TOKEN/WIZ_API_URL unset.

Kyverno fleshed out (REQ-111):
- adapters/kyverno/kyverno_adapter.py: full PolicyReport →
  PolicyCheckResult mapping (pass/fail/skip/warn + severity + skip-with-
  reason + resource ref construction from kind/name/namespace).
  adapt_inactive() emits KYVERNO_INACTIVE_TF_STACK guard. --kube-version
  stub parsed for future GitOps.

Tests: +47 (test_route_halt_artifact.py, test_hitl_gates.py,
test_attestation_matrix.py, test_wiz_adapter_real_client.py, expanded
test_kyverno_adapter.py). Existing wiz_adapter tests updated for the
real client's control.name ruleId. 493 passed; run_ci.sh green;
run_platform.sh --check-only green.
2026-07-23 04:40:44 +00:00

126 lines
4.1 KiB
Python

"""REQ-108: HITL qa/prod/dr attestation gates."""
import sys
from pathlib import Path
from unittest import mock
import pytest
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
from core.hitl_gates import attest, approver_from_env
from core.attestation_matrix import FRESHNESS_DAYS, ENV_CONCERNS
import datetime
def _fresh_evidence_for(env):
"""Build a valid evidence bundle with fresh artifacts for every concern in env."""
evidence = {"contract_nfrs": {"valid": True}}
for concern in ENV_CONCERNS.get(env, []):
if concern != "contract_nfrs":
ts = datetime.datetime.now(datetime.timezone.utc)
evidence[concern] = {"timestamp": ts.isoformat(), "type": concern,
"payload": {}, "signature": "sig"}
return evidence
class FakeOutbox:
"""Minimal outbox client for tests: stores approver attrs per contract."""
def __init__(self):
self.records = {}
def put_approver(self, contract_id, attr, value):
self.records.setdefault(contract_id, {})[attr] = value
def get(self, contract_id):
return self.records.get(contract_id)
def test_dev_skips_gate():
ok, reason = attest("c1", "dev", "alice")
assert ok is True
assert "autonomous" in reason
def test_qa_records_approver():
outbox = FakeOutbox()
ok, reason = attest("c2", "qa", "bob", evidence=_fresh_evidence_for("qa"),
outbox_client=outbox)
assert ok is True
assert outbox.records["c2"]["approver_qa"] == "bob"
def test_prod_records_approver():
outbox = FakeOutbox()
ok, reason = attest("c3", "prod", "carol", evidence=_fresh_evidence_for("prod"),
outbox_client=outbox)
assert ok is True
assert outbox.records["c3"]["approver_prod"] == "carol"
def test_dr_records_approver():
outbox = FakeOutbox()
ok, reason = attest("c4", "dr", "dave", evidence=_fresh_evidence_for("dr"),
outbox_client=outbox)
assert ok is True
assert outbox.records["c4"]["approver_dr"] == "dave"
def test_prod_sod_blocks_on_identity_equality():
"""When approver_qa == approver_prod, prod promotion is blocked."""
outbox = FakeOutbox()
outbox.put_approver("c5", "approver_qa", "eve")
with mock.patch("core.separation_of_duties.route_halt_artifact"):
ok, reason = attest("c5", "prod", "eve", outbox_client=outbox)
assert ok is False
assert "SEPARATION_OF_DUTIES_VIOLATION" in reason
def test_prod_sod_passes_when_approvers_differ():
outbox = FakeOutbox()
outbox.put_approver("c6", "approver_qa", "alice")
ok, reason = attest("c6", "prod", "bob", evidence=_fresh_evidence_for("prod"),
outbox_client=outbox)
assert ok is True
def test_no_approver_blocks_non_dev():
ok, reason = attest("c7", "qa", "", outbox_client=FakeOutbox())
assert ok is False
assert "no approver" in reason
def test_unknown_env_blocks():
ok, reason = attest("c8", "staging", "alice")
assert ok is False
assert "unknown environment" in reason
def test_approver_from_env_github(monkeypatch):
monkeypatch.setenv("GITHUB_ACTOR", "gh-user")
monkeypatch.delenv("GITEA_ACTOR", raising=False)
assert approver_from_env() == "gh-user"
def test_approver_from_env_gitea(monkeypatch):
monkeypatch.delenv("GITHUB_ACTOR", raising=False)
monkeypatch.setenv("GITEA_ACTOR", "gitea-user")
assert approver_from_env() == "gitea-user"
def test_attest_invokes_attestation_matrix_for_prod():
"""attest calls the attestation matrix for prod."""
outbox = FakeOutbox()
outbox.put_approver("c9", "approver_qa", "alice")
with mock.patch("core.attestation_matrix.check", return_value=(False, "missing evidence")) as m:
ok, reason = attest("c9", "prod", "bob", outbox_client=outbox)
m.assert_called_once()
assert ok is False
assert "missing evidence" in reason
def test_run_platform_sh_has_hitl_gate_step():
text = (ROOT / "scripts" / "run_platform.sh").read_text()
assert "HITL attestation gate" in text
assert "hitl_gates" in text
assert "RESOLVED_ENV" in text