Jon Chery
0d2cbdb423
Genericize forge-detection code: gitea→forge/generic_forge, GITEA_ACTOR→FORGE_ACTOR. Drop .gitea byte-identity test assertions (keep GitHub-side + contract conformance). Add test_no_forge_mentions.py guard test (REQ-230). Delete completed migration docs (NOVA_MIGRATION.md, NOVA_AWS_MIGRATION.md). Move NO_HUMANS_THESIS.md to .ciagent/ (internal artifact). Strip ciagent-internal provenance from synced docs (REQ-/D-/P-/CAP- IDs, milestone headers, .ciagent/PROJECT.md citations). Trim README.md (reusable deploy section, local key rotation paragraph). Fix version-tag drift (@v1.13→@v1.19, acdl/→nova/). ---ci--- project: acdl phase: 1 milestone: v1.20 status: execute requirements: [REQ-230, REQ-231, REQ-232] ---/ci---
106 lines
4.1 KiB
Python
106 lines
4.1 KiB
Python
"""Check that qaApprover != prodApprover for a contract (ARCHITECTURE.md
|
|
§10.3, D-042). Reads `approver_qa` from the DynamoDB outbox for the
|
|
contractId, compares to the prod-dispatch the CI actor.
|
|
Blocks on equality, emits `SEPARATION_OF_DUTIES_VIOLATION`, routes a halt
|
|
artifact to SRE on-call.
|
|
|
|
v1.9 (REQ-107, D-085): route_halt_artifact is a real implementation —
|
|
publishes to SNS topic `acdl-sod-halt` (ARN from NOVA_SOD_HALT_TOPIC_ARN)
|
|
when set; falls back to a structured stderr emission + a
|
|
SEPARATION_OF_DUTIES_VIOLATION event write to the DynamoDB outbox when
|
|
unset. No silent print-only stub. (Dual-read via core/env.py: NOVA_*
|
|
preferred, ACDL_* fallback until P5; the SNS topic ARN is the AWS
|
|
resource `acdl-sod-halt` → renamed `nova-sod-halt` in P4.)
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
from typing import Optional, Tuple
|
|
|
|
# Repo root on sys.path so `from core import env` resolves to THIS package
|
|
# when imported/run in a context where an editable-installed third-party
|
|
# `core` package would otherwise shadow it.
|
|
_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
if _REPO_ROOT not in sys.path:
|
|
sys.path.insert(0, _REPO_ROOT)
|
|
|
|
from core import env
|
|
|
|
|
|
def check(outbox_client, contract_id: str,
|
|
current_prod_approver: Optional[str]) -> Tuple[bool, str]:
|
|
"""Return (ok, reason). ok=False means block the prod promotion."""
|
|
if outbox_client is None:
|
|
return (True, "no outbox client (dev-only spike)")
|
|
item = outbox_client.get(contract_id)
|
|
if item is None:
|
|
return (True, "no prior approver (first promotion)")
|
|
qa_approver = item.get("approver_qa")
|
|
if not qa_approver:
|
|
return (True, "no QA approver recorded (dev-only spike)")
|
|
if current_prod_approver is None:
|
|
return (True, "no prod approver supplied (dev-only spike)")
|
|
if qa_approver == current_prod_approver:
|
|
return (False,
|
|
f"SEPARATION_OF_DUTIES_VIOLATION: "
|
|
f"qaApprover==prodApprover=={qa_approver}")
|
|
return (True, "distinct")
|
|
|
|
|
|
def route_halt_artifact(contract_id: str, violation_reason: str,
|
|
oncall_client=None) -> None:
|
|
"""Route a halt artifact to SRE on-call (REQ-107, D-085).
|
|
|
|
When NOVA_SOD_HALT_TOPIC_ARN is set, publish to the SNS topic via
|
|
boto3. When unset (dev/CI), fall back to a structured stderr emission
|
|
+ a SEPARATION_OF_DUTIES_VIOLATION event write to the DynamoDB outbox
|
|
via outbox_writer.write_event (so the halt is in the audit chain).
|
|
The oncall_client, when provided, is the SNS client (test injection).
|
|
"""
|
|
topic_arn = env.get_env("SOD_HALT_TOPIC_ARN", "") or ""
|
|
halt_payload = {
|
|
"contractId": contract_id,
|
|
"reason": violation_reason,
|
|
"action": "HALT_PROMOTION",
|
|
}
|
|
if topic_arn:
|
|
import json
|
|
try:
|
|
import boto3
|
|
if oncall_client is not None:
|
|
sns = oncall_client
|
|
else:
|
|
sns = boto3.client("sns")
|
|
sns.publish(
|
|
TopicArn=topic_arn,
|
|
Message=json.dumps(halt_payload),
|
|
Subject="Nova SoD halt",
|
|
)
|
|
print(f"[halt-artifact] SNS published contract={contract_id} "
|
|
f"topic={topic_arn}", flush=True)
|
|
return
|
|
except Exception as exc:
|
|
sys.stderr.write(
|
|
f"[halt-artifact] SNS publish failed ({exc}); "
|
|
f"falling back to outbox event\n"
|
|
)
|
|
# Fallback: stderr + outbox event (the halt is in the audit chain).
|
|
sys.stderr.write(
|
|
f"[halt-artifact] contract={contract_id} reason={violation_reason} "
|
|
f"oncall={oncall_client} (no SNS topic — outbox fallback)\n"
|
|
)
|
|
try:
|
|
from core.outbox_writer import write_event
|
|
write_event({
|
|
"contractId": contract_id,
|
|
"eventType": "SEPARATION_OF_DUTIES_VIOLATION",
|
|
"environment": "",
|
|
"stack": "",
|
|
"score": 0,
|
|
"band": "halt",
|
|
"reason": violation_reason,
|
|
})
|
|
except Exception as exc:
|
|
sys.stderr.write(
|
|
f"[halt-artifact] outbox fallback write failed ({exc})\n"
|
|
) |