291 lines
10 KiB
Python
291 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Result model, aggregation, and rendering for the handoff acceptance harness.
|
|
|
|
The classification is deliberately four-valued. ``NOT_APPLICABLE`` is the only
|
|
way a check can be excused, and a check can only reach it by matching a scope
|
|
rule that a reviewer wrote down; anything the harness merely failed to observe
|
|
is ``NOT_RUN``. Because a mandatory ``NOT_RUN`` is a NO_GO exactly like a
|
|
``FAIL``, an unreachable cluster, an exhausted deadline, or a probe nobody
|
|
maintained can never round up to a release.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
from collections.abc import Mapping, Sequence
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
from hermes_handoff_exec import Outcome
|
|
from hermes_handoff_redaction import contains_credential_shape, safe_text, scrub
|
|
|
|
PASS = "PASS"
|
|
FAIL = "FAIL"
|
|
NOT_RUN = "NOT_RUN"
|
|
NOT_APPLICABLE = "NOT_APPLICABLE"
|
|
STATUSES = (PASS, FAIL, NOT_RUN, NOT_APPLICABLE)
|
|
|
|
GO = "GO"
|
|
NO_GO = "NO_GO"
|
|
|
|
ALWAYS = "always"
|
|
EPHEMERAL = "ephemeral"
|
|
|
|
HARNESS = "hermes-handoff-acceptance"
|
|
SCHEMA_VERSION = 1
|
|
|
|
|
|
REVIEW = "review"
|
|
ATTEMPT = "attempt"
|
|
READ = "read"
|
|
STEP_KINDS = (REVIEW, ATTEMPT, READ)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Step:
|
|
"""One command a check needs, and the vantage it must be observed from.
|
|
|
|
``kind`` separates a self access review (``review``) from a real request the
|
|
identity is expected to be refused (``attempt``). A deny check needs both:
|
|
an authorization review states policy, only an attempt proves the API server
|
|
enforces it.
|
|
|
|
``record`` is cleared for bulk evidence — a routing log tail, a large object
|
|
listing — which the harness parses but must not paste into the artifact.
|
|
``max_bytes`` raises the capture budget for that same bulk evidence, where a
|
|
default-sized truncation would silently narrow the window being examined.
|
|
"""
|
|
|
|
key: str
|
|
vantage: str
|
|
argv: tuple[str, ...]
|
|
kind: str = READ
|
|
optional: bool = False
|
|
record: bool = True
|
|
max_bytes: int | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CheckSpec:
|
|
"""A declarative acceptance check.
|
|
|
|
``rule`` names an evaluator; ``expect`` carries its parameters. Keeping the
|
|
catalog declarative means a reviewer reads what is asserted, not how it is
|
|
plumbed, and the harness can prove structural properties over every check
|
|
(no impersonation, no mutation, both vantages present) before running any.
|
|
"""
|
|
|
|
id: str
|
|
title: str
|
|
group: str
|
|
rule: str
|
|
steps: tuple[Step, ...] = ()
|
|
expect: Mapping[str, Any] = field(default_factory=dict)
|
|
mandatory: bool = True
|
|
scope: str = ALWAYS
|
|
rationale: str = ""
|
|
|
|
@property
|
|
def vantages(self) -> tuple[str, ...]:
|
|
"""Return the distinct vantages this check draws evidence from."""
|
|
seen: list[str] = []
|
|
for step in self.steps:
|
|
if step.vantage not in seen:
|
|
seen.append(step.vantage)
|
|
return tuple(seen)
|
|
|
|
|
|
@dataclass
|
|
class CheckResult:
|
|
"""The classified outcome of one acceptance check."""
|
|
|
|
spec: CheckSpec
|
|
status: str
|
|
reason: str = ""
|
|
evidence: dict[str, Any] = field(default_factory=dict)
|
|
outcomes: list[Outcome] = field(default_factory=list)
|
|
|
|
@property
|
|
def blocking(self) -> bool:
|
|
"""Report whether this result alone forces NO_GO."""
|
|
if self.status not in STATUSES:
|
|
return True
|
|
return self.spec.mandatory and self.status in {FAIL, NOT_RUN}
|
|
|
|
def as_dict(self, include_outcomes: bool = True) -> dict[str, Any]:
|
|
"""Return a JSON-serialisable, screened view of the result."""
|
|
payload: dict[str, Any] = {
|
|
"id": self.spec.id,
|
|
"title": self.spec.title,
|
|
"group": self.spec.group,
|
|
"rule": self.spec.rule,
|
|
"scope": self.spec.scope,
|
|
"mandatory": self.spec.mandatory,
|
|
"vantages": list(self.spec.vantages),
|
|
"status": self.status,
|
|
"reason": self.reason,
|
|
"blocking": self.blocking,
|
|
"evidence": self.evidence,
|
|
}
|
|
if self.spec.rationale:
|
|
payload["rationale"] = self.spec.rationale
|
|
if include_outcomes:
|
|
payload["commands"] = [outcome.as_dict() for outcome in self.outcomes]
|
|
return scrub(payload)
|
|
|
|
|
|
@dataclass
|
|
class VantageRecord:
|
|
"""What the harness proved about one vantage before trusting its evidence."""
|
|
|
|
name: str
|
|
description: str
|
|
identity: str = ""
|
|
available: bool = False
|
|
detail: str = ""
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
"""Return a JSON-serialisable view of the vantage."""
|
|
return scrub(
|
|
{
|
|
"name": self.name,
|
|
"description": self.description,
|
|
"identity": self.identity,
|
|
"available": self.available,
|
|
"detail": self.detail,
|
|
}
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Report:
|
|
"""The whole acceptance run: metadata, per-check results, and a verdict."""
|
|
|
|
mode: str
|
|
started_at: str
|
|
baseline: dict[str, Any] = field(default_factory=dict)
|
|
vantages: list[VantageRecord] = field(default_factory=list)
|
|
results: list[CheckResult] = field(default_factory=list)
|
|
harness_errors: list[str] = field(default_factory=list)
|
|
finished_at: str = ""
|
|
|
|
@property
|
|
def counts(self) -> dict[str, int]:
|
|
"""Return the number of results in each status."""
|
|
tally = dict.fromkeys(STATUSES, 0)
|
|
for result in self.results:
|
|
if result.status in tally:
|
|
tally[result.status] += 1
|
|
return tally
|
|
|
|
@property
|
|
def invalid_statuses(self) -> list[str]:
|
|
"""Return check ids whose status is outside the report schema."""
|
|
return [
|
|
result.spec.id for result in self.results if result.status not in STATUSES
|
|
]
|
|
|
|
@property
|
|
def blocking(self) -> list[CheckResult]:
|
|
"""Return every result that forces NO_GO, most severe first."""
|
|
order = {FAIL: 0, NOT_RUN: 1}
|
|
return sorted(
|
|
(result for result in self.results if result.blocking),
|
|
key=lambda result: (order.get(result.status, 2), result.spec.id),
|
|
)
|
|
|
|
@property
|
|
def decision(self) -> str:
|
|
"""Return GO only when nothing mandatory failed, skipped, or errored."""
|
|
if (
|
|
not self.results
|
|
or self.invalid_statuses
|
|
or self.harness_errors
|
|
or self.blocking
|
|
):
|
|
return NO_GO
|
|
return GO
|
|
|
|
def group_summary(self) -> dict[str, dict[str, int]]:
|
|
"""Return per-group status counts, for the human summary."""
|
|
groups: dict[str, dict[str, int]] = {}
|
|
for result in self.results:
|
|
bucket = groups.setdefault(result.spec.group, dict.fromkeys(STATUSES, 0))
|
|
bucket[result.status] = bucket.get(result.status, 0) + 1
|
|
return groups
|
|
|
|
def as_dict(self, include_outcomes: bool = True) -> dict[str, Any]:
|
|
"""Return the machine-readable report."""
|
|
return scrub(
|
|
{
|
|
"harness": HARNESS,
|
|
"schema_version": SCHEMA_VERSION,
|
|
"mode": self.mode,
|
|
"started_at": self.started_at,
|
|
"finished_at": self.finished_at,
|
|
"decision": self.decision,
|
|
"counts": self.counts,
|
|
"baseline": scrub(self.baseline),
|
|
"vantages": [vantage.as_dict() for vantage in self.vantages],
|
|
"harness_errors": [scrub(error) for error in self.harness_errors],
|
|
"invalid_statuses": self.invalid_statuses,
|
|
"blocking": [result.spec.id for result in self.blocking],
|
|
"checks": [result.as_dict(include_outcomes) for result in self.results],
|
|
}
|
|
)
|
|
|
|
def render_summary(self, width: int = 88) -> str:
|
|
"""Return the concise human summary printed alongside the JSON."""
|
|
counts = self.counts
|
|
lines = [
|
|
f"{HARNESS}: {self.decision}",
|
|
f" mode={self.mode} started={self.started_at} checks={len(self.results)}",
|
|
" " + " ".join(f"{status}={counts[status]}" for status in STATUSES),
|
|
]
|
|
for vantage in self.vantages:
|
|
state = "available" if vantage.available else "unavailable"
|
|
identity = vantage.identity or "unknown"
|
|
lines.append(f" vantage {vantage.name}: {state} identity={identity}")
|
|
for error in self.harness_errors:
|
|
lines.append(f" harness error: {error}")
|
|
if self.blocking:
|
|
lines.append(" blocking:")
|
|
for result in self.blocking:
|
|
lines.append(f" {result.status:<15}{result.spec.id}")
|
|
lines.append(f" {'':<15}{result.reason[:width]}")
|
|
groups = self.group_summary()
|
|
lines.append(" groups:")
|
|
for name in sorted(groups):
|
|
bucket = groups[name]
|
|
detail = " ".join(
|
|
f"{status}={bucket[status]}" for status in STATUSES if bucket[status]
|
|
)
|
|
lines.append(f" {name:<18}{detail}")
|
|
return safe_text("\n".join(lines), 32 * 1024)[0]
|
|
|
|
|
|
def utc_now(clock: Any = None) -> str:
|
|
"""Return an ISO-8601 UTC timestamp, injectable for deterministic tests."""
|
|
now = clock() if clock else dt.datetime.now(dt.timezone.utc)
|
|
return now.replace(microsecond=0).isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
def unscreened_fields(payload: Any, path: str = "$") -> list[str]:
|
|
"""Return JSON pointers whose text still looks credential-shaped.
|
|
|
|
The report is already screened field by field; this is the invariant check
|
|
that runs before anything is written, so a screening gap fails the run
|
|
instead of publishing an artifact nobody re-read.
|
|
"""
|
|
offenders: list[str] = []
|
|
if isinstance(payload, str):
|
|
if contains_credential_shape(payload):
|
|
offenders.append(path)
|
|
elif isinstance(payload, Mapping):
|
|
for key, value in payload.items():
|
|
offenders.extend(unscreened_fields(value, f"{path}.{key}"))
|
|
elif isinstance(payload, Sequence) and not isinstance(payload, (str, bytes)):
|
|
for index, value in enumerate(payload):
|
|
offenders.extend(unscreened_fields(value, f"{path}[{index}]"))
|
|
return offenders
|