From 3fe97bd3e38ba17931ec040566d0698e3b78adc6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=F0=9D=90=8E=F0=9D=90=A7=F0=9D=90=9E=20=F0=9D=90=85?= =?UTF-8?q?=F0=9D=90=A2=F0=9D=90=A7=F0=9D=90=9E=20=F0=9D=90=92=F0=9D=90=AD?= =?UTF-8?q?=F0=9D=90=9A=F0=9D=90=AB=F0=9D=90=AC=F0=9D=90=AD=F0=9D=90=AE?= =?UTF-8?q?=F0=9D=90=9F=F0=9D=90=9F?= Date: Mon, 6 Jul 2026 15:06:50 +0630 Subject: [PATCH] fix: harden daily sentinel evidence checks --- .../DAILY_SENTINEL_V24_DEVSECOPS_RUNBOOK.md | 95 ++++ pyproject.toml | 1 + scripts/daily_sentinel_operational_check.py | 506 ++++++++++++++++++ .../test_daily_sentinel_operational_check.py | 260 +++++++++ 4 files changed, 862 insertions(+) create mode 100644 docs/reports/DAILY_SENTINEL_V24_DEVSECOPS_RUNBOOK.md create mode 100644 scripts/daily_sentinel_operational_check.py create mode 100644 tests/test_daily_sentinel_operational_check.py diff --git a/docs/reports/DAILY_SENTINEL_V24_DEVSECOPS_RUNBOOK.md b/docs/reports/DAILY_SENTINEL_V24_DEVSECOPS_RUNBOOK.md new file mode 100644 index 00000000..c0d632e8 --- /dev/null +++ b/docs/reports/DAILY_SENTINEL_V24_DEVSECOPS_RUNBOOK.md @@ -0,0 +1,95 @@ +# Daily Sentinel AI v2.4 DevSecOps Operational Check Runbook + +This runbook defines the daily evidence pack used to validate the Omni-Sentinel Cognitive Execution Environment for G-SIFI and Fortune 500 financial-services deployments. + +## Required checks + +1. **Sentinel telemetry/dashboard health** — verify the Sentinel v2.4 dashboard or API endpoint returns a 2xx health result. The default endpoint is `https://sentinel.internal.g-sifi.local`. +2. **Global Systemic Risk Index (G-SRI)** — verify a signed, fresh G-SRI evidence bundle reports `current_value < threshold` under the approved risk-appetite policy version. +3. **PQC WORM audit batches** — verify `pqc_worm_logger.py` committed the latest audit batch to the designated S3 Object Lock bucket, including object version, Object Lock mode, retention-until timestamp, Merkle root, PQC signature algorithm, and commit-lag SLO. +4. **TEE/TPM attestation** — verify fresh attestation evidence reports `PCR_MATCH=true`, trusted TEE status, node identity, PCR policy hash, TPM quote ID, and verifier identity. + +## Evidence JSON fields + +### G-SRI evidence + +```json +{ + "timestamp_utc": "2026-05-29T12:00:00Z", + "current_value": 0.41, + "threshold": 0.70, + "policy_version": "gsri-risk-appetite-2026.05", + "signed": true +} +``` + +### WORM evidence + +```json +{ + "timestamp_utc": "2026-05-29T12:00:00Z", + "logger_name": "pqc_worm_logger.py", + "batch_status": "committed", + "bucket": "sentinel-object-lock-audit", + "object_key": "audit/2026/05/29/batch-0001.jsonl", + "object_version_id": "3HL4kqtJlcpXrof3vjVBH40Nrjfkd", + "object_lock_mode": "COMPLIANCE", + "retention_until_utc": "2033-05-29T12:00:00Z", + "merkle_root": "sha256:abc123", + "pqc_signature_alg": "ML-DSA-65", + "pqc_signature_verified": true, + "commit_lag_seconds": 42 +} +``` + +### TEE/TPM attestation evidence + +```json +{ + "timestamp_utc": "2026-05-29T12:00:00Z", + "PCR_MATCH": true, + "tee_status": "trusted", + "node_id": "sentinel-prod-01", + "pcr_policy_hash": "sha256:def456", + "tpm_quote_id": "quote-20260529-0001", + "verifier_id": "attestor-v2.4.3", + "attestation_signature_verified": true +} +``` + +## CLI usage + +Run a live check with dashboard/API reachability and the designated Object Lock bucket: + +```bash +python -m scripts.daily_sentinel_operational_check \ + --gsri-evidence /secure/evidence/gsri.json \ + --worm-evidence /secure/evidence/worm.json \ + --expected-worm-bucket sentinel-object-lock-audit \ + --attestation-evidence /secure/evidence/attestation.json +``` + +Run an offline evidence-pack check when the dashboard network is intentionally unavailable: + +```bash +python -m scripts.daily_sentinel_operational_check \ + --skip-dashboard \ + --gsri-evidence /secure/evidence/gsri.json \ + --worm-evidence /secure/evidence/worm.json \ + --attestation-evidence /secure/evidence/attestation.json +``` + +Use `--json` for machine-readable CI output. The JSON payload includes `generated_at_utc`, `overall_status`, and a `results` array so CI can route each failed check to the correct control owner. Use `--require-compliance-object-lock` when governance policy forbids S3 Object Lock GOVERNANCE mode for the audit bucket. + +## Fail-closed rules + +Treat the daily control state as **RED** when any of the following conditions occurs: + +- Sentinel dashboard/API health is not 2xx and the outage is not covered by an approved maintenance window. +- G-SRI evidence is missing, stale, unsigned, or at/above the policy threshold. +- `pqc_worm_logger.py` has not committed the latest WORM batch, the bucket does not match the designated Object Lock bucket, Object Lock metadata is missing, retention is expired, the PQC signature is not verified, or commit lag exceeds the SLO. +- TEE/TPM attestation is stale, `PCR_MATCH` is not true, TEE status is not trusted/verified, or the attestation signature is not verified. + +## Regulatory mapping + +The daily evidence pack supports EU AI Act technical documentation and systemic-risk GPAI monitoring, NIST AI RMF 1.0 and AI 600-1 operational risk management, ISO/IEC 42001 AIMS monitoring evidence, SR 11-7/SR 26-2 model-risk governance, DORA and NIS2 operational-resilience evidence, GDPR accountability, and Basel operational-risk oversight. diff --git a/pyproject.toml b/pyproject.toml index 0229b49f..a77fdced 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,3 +31,4 @@ governance = ["jsonschema"] [project.scripts] validate-gsifi-governance-assets = "scripts.validate_gsifi_governance_assets:main" +daily-sentinel-operational-check = "scripts.daily_sentinel_operational_check:main" diff --git a/scripts/daily_sentinel_operational_check.py b/scripts/daily_sentinel_operational_check.py new file mode 100644 index 00000000..2967f4dc --- /dev/null +++ b/scripts/daily_sentinel_operational_check.py @@ -0,0 +1,506 @@ +#!/usr/bin/env python3 +"""Daily Sentinel AI v2.4 DevSecOps operational checks. + +The checker is intentionally dependency-light so it can run from CI, a jump host, +or an auditor workstation. It validates four evidence planes used by the +Omni-Sentinel Cognitive Execution Environment: + +* Sentinel telemetry/dashboard reachability +* Global Systemic Risk Index (G-SRI) threshold evidence +* PQC WORM logger S3 Object Lock commit evidence +* TEE/TPM attestation evidence, including PCR_MATCH + +The command exits non-zero whenever a required check is RED. Optional dashboard +reachability can be skipped for offline evidence-pack validation. +""" + +from __future__ import annotations + +import argparse +import datetime as dt +import json +import ssl +import sys +import urllib.error +import urllib.request +from collections.abc import Callable +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any + +UTC = dt.timezone.utc +DEFAULT_DASHBOARD_URL = "https://sentinel.internal.g-sifi.local" +DEFAULT_STALENESS_MINUTES = 30 +DEFAULT_WORM_MAX_LAG_SECONDS = 900 +PASS = "PASS" +FAIL = "FAIL" +WARN = "WARN" + + +class CheckError(RuntimeError): + """Raised when operational evidence cannot be validated.""" + + +@dataclass(frozen=True) +class CheckResult: + """Single operational check result.""" + + name: str + status: str + summary: str + remediation: str = "" + + @property + def is_failure(self) -> bool: + return self.status == FAIL + + def to_dict(self) -> dict[str, str]: + """Return a stable machine-readable representation.""" + return asdict(self) + + +def load_json(path: Path) -> dict[str, Any]: + """Load a JSON object from *path*.""" + try: + data = json.loads(path.read_text(encoding="utf-8")) + except OSError as exc: + raise CheckError(f"unable to read {path}: {exc}") from exc + except json.JSONDecodeError as exc: + raise CheckError(f"invalid JSON in {path}: {exc}") from exc + if not isinstance(data, dict): + raise CheckError(f"{path} must contain a JSON object") + return data + + +def parse_utc_timestamp(value: object, field_name: str) -> dt.datetime: + """Parse a strict UTC timestamp ending in Z.""" + if not isinstance(value, str) or not value.endswith("Z"): + raise CheckError( + f"{field_name} must be an ISO-8601 UTC timestamp ending in 'Z'" + ) + try: + parsed = dt.datetime.fromisoformat(value.removesuffix("Z") + "+00:00") + except ValueError as exc: + raise CheckError( + f"{field_name} is not a valid ISO-8601 timestamp: {value}" + ) from exc + return parsed.astimezone(UTC) + + +def require_fresh( + timestamp: dt.datetime, max_age_minutes: int, field_name: str +) -> None: + """Require *timestamp* to be no older than *max_age_minutes*.""" + now = dt.datetime.now(UTC) + age = now - timestamp + if age < dt.timedelta(seconds=-60): + raise CheckError(f"{field_name} is from the future: {timestamp.isoformat()}") + if age > dt.timedelta(minutes=max_age_minutes): + age_minutes = int(age.total_seconds() // 60) + raise CheckError( + f"{field_name} is stale: {age_minutes} minutes old; " + f"maximum allowed is {max_age_minutes} minutes" + ) + + +def _ssl_context(insecure_tls: bool) -> ssl.SSLContext | None: + """Build the TLS context for dashboard checks.""" + if not insecure_tls: + return None + context = ssl.create_default_context() + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + return context + + +def _open_dashboard( + url: str, + method: str, + timeout_seconds: float, + context: ssl.SSLContext | None, +) -> int: + request = urllib.request.Request(url, method=method) + with urllib.request.urlopen( + request, timeout=timeout_seconds, context=context + ) as resp: + return resp.getcode() + + +def check_dashboard( + url: str, timeout_seconds: float, insecure_tls: bool +) -> CheckResult: + """Check Sentinel telemetry/dashboard reachability. + + Some dashboard endpoints reject HEAD even when their health route is usable, so + HTTP 405/501 automatically falls back to GET before declaring the check RED. + """ + context = _ssl_context(insecure_tls) + try: + status = _open_dashboard(url, "HEAD", timeout_seconds, context) + except urllib.error.HTTPError as exc: + if exc.code not in {405, 501}: + return CheckResult( + "sentinel_dashboard", + FAIL, + f"{url} returned HTTP {exc.code}", + "Restore the Sentinel dashboard/API upstream and capture Envoy/service health evidence.", + ) + try: + status = _open_dashboard(url, "GET", timeout_seconds, context) + except urllib.error.HTTPError as get_exc: + return CheckResult( + "sentinel_dashboard", + FAIL, + f"{url} returned HTTP {get_exc.code} after HEAD fallback", + "Restore the Sentinel dashboard/API upstream and capture Envoy/service health evidence.", + ) + except (urllib.error.URLError, TimeoutError, OSError) as get_exc: + return CheckResult( + "sentinel_dashboard", + FAIL, + f"{url} is unreachable after HEAD fallback: {get_exc}", + "Validate DNS, mTLS, Envoy upstream health, service discovery, and Sentinel pod readiness.", + ) + except (urllib.error.URLError, TimeoutError, OSError) as exc: + return CheckResult( + "sentinel_dashboard", + FAIL, + f"{url} is unreachable: {exc}", + "Validate DNS, mTLS, Envoy upstream health, service discovery, and Sentinel pod readiness.", + ) + if 200 <= status < 300: + return CheckResult("sentinel_dashboard", PASS, f"{url} returned HTTP {status}") + return CheckResult( + "sentinel_dashboard", + FAIL, + f"{url} returned non-success HTTP {status}", + "Treat telemetry as degraded until a 2xx dashboard/API health result is restored.", + ) + + +def check_gsri(evidence: dict[str, Any], max_age_minutes: int) -> CheckResult: + """Validate Global Systemic Risk Index evidence.""" + timestamp = parse_utc_timestamp(evidence.get("timestamp_utc"), "gsri.timestamp_utc") + require_fresh(timestamp, max_age_minutes, "gsri.timestamp_utc") + value = evidence.get("current_value") + threshold = evidence.get("threshold") + if not isinstance(value, (int, float)) or isinstance(value, bool): + raise CheckError("gsri.current_value must be numeric") + if not isinstance(threshold, (int, float)) or isinstance(threshold, bool): + raise CheckError("gsri.threshold must be numeric") + policy_version = evidence.get("policy_version") + if not isinstance(policy_version, str) or not policy_version.strip(): + raise CheckError("gsri.policy_version must be a non-empty string") + signed = ( + evidence.get("signed") is True or evidence.get("signature_verified") is True + ) + if value < threshold and signed: + return CheckResult( + "gsri_threshold", + PASS, + f"G-SRI {value:.4f} is below threshold {threshold:.4f} under {policy_version}", + ) + if not signed: + return CheckResult( + "gsri_threshold", + FAIL, + "G-SRI evidence signature is not verified", + "Regenerate the G-SRI bundle with signer identity, policy version, and WORM pointer.", + ) + return CheckResult( + "gsri_threshold", + FAIL, + f"G-SRI {value:.4f} is at or above threshold {threshold:.4f}", + "Open the systemic-risk escalation path and apply containment/reduction controls before green reporting.", + ) + + +def check_worm( + evidence: dict[str, Any], + max_age_minutes: int, + max_lag_seconds: int, + expected_bucket: str | None = None, + require_compliance_mode: bool = False, +) -> CheckResult: + """Validate PQC WORM logger S3 Object Lock commit evidence.""" + timestamp = parse_utc_timestamp(evidence.get("timestamp_utc"), "worm.timestamp_utc") + require_fresh(timestamp, max_age_minutes, "worm.timestamp_utc") + required_strings = [ + "logger_name", + "bucket", + "object_key", + "object_version_id", + "object_lock_mode", + "retention_until_utc", + "merkle_root", + "pqc_signature_alg", + ] + missing = [ + key + for key in required_strings + if not isinstance(evidence.get(key), str) or not evidence.get(key) + ] + if missing: + raise CheckError(f"worm evidence missing non-empty string fields: {missing}") + if evidence["logger_name"] != "pqc_worm_logger.py": + raise CheckError("worm.logger_name must be pqc_worm_logger.py") + if expected_bucket and evidence["bucket"] != expected_bucket: + return CheckResult( + "pqc_worm_logger", + FAIL, + f"WORM bucket {evidence['bucket']!r} does not match expected bucket {expected_bucket!r}", + "Route WORM batches to the designated S3 Object Lock bucket and regenerate commit evidence.", + ) + if evidence.get("batch_status") != "committed": + return CheckResult( + "pqc_worm_logger", + FAIL, + f"WORM batch status is {evidence.get('batch_status')!r}", + "Drain/replay the WORM queue and verify the next committed object version under Object Lock.", + ) + mode = str(evidence["object_lock_mode"]).upper() + allowed_modes = ( + {"COMPLIANCE"} if require_compliance_mode else {"COMPLIANCE", "GOVERNANCE"} + ) + if mode not in allowed_modes: + required_mode = ( + "COMPLIANCE" if require_compliance_mode else "COMPLIANCE or GOVERNANCE" + ) + raise CheckError(f"worm.object_lock_mode must be {required_mode}") + if evidence.get("pqc_signature_verified") is not True: + return CheckResult( + "pqc_worm_logger", + FAIL, + "WORM PQC signature has not been verified", + "Verify the ML-DSA signature over the batch manifest before accepting the WORM evidence.", + ) + retention_until = parse_utc_timestamp( + evidence["retention_until_utc"], "worm.retention_until_utc" + ) + if retention_until <= dt.datetime.now(UTC): + raise CheckError("worm.retention_until_utc must be in the future") + lag = evidence.get("commit_lag_seconds") + if not isinstance(lag, int) or isinstance(lag, bool) or lag < 0: + raise CheckError("worm.commit_lag_seconds must be a non-negative integer") + if lag > max_lag_seconds: + return CheckResult( + "pqc_worm_logger", + FAIL, + f"WORM commit lag {lag}s exceeds {max_lag_seconds}s SLO", + "Investigate queue backlog, S3/KMS errors, network path, and CloudTrail put-object evidence.", + ) + return CheckResult( + "pqc_worm_logger", + PASS, + f"{evidence['logger_name']} committed {evidence['object_key']} version {evidence['object_version_id']} with {mode} Object Lock", + ) + + +def check_attestation(evidence: dict[str, Any], max_age_minutes: int) -> CheckResult: + """Validate TEE/TPM attestation evidence.""" + timestamp = parse_utc_timestamp( + evidence.get("timestamp_utc"), "attestation.timestamp_utc" + ) + require_fresh(timestamp, max_age_minutes, "attestation.timestamp_utc") + if evidence.get("PCR_MATCH") is not True: + return CheckResult( + "tee_tpm_attestation", + FAIL, + f"PCR_MATCH is {evidence.get('PCR_MATCH')!r}", + "Quarantine the node/workload, collect TPM quote details, and compare against the approved PCR policy.", + ) + tee_status = evidence.get("tee_status") + if tee_status not in {"trusted", "verified"}: + return CheckResult( + "tee_tpm_attestation", + FAIL, + f"TEE status is {tee_status!r}", + "Deny production routing until enclave/workload measurement is verified by the attestation service.", + ) + required_strings = ["node_id", "pcr_policy_hash", "tpm_quote_id", "verifier_id"] + missing = [ + key + for key in required_strings + if not isinstance(evidence.get(key), str) or not evidence.get(key) + ] + if missing: + raise CheckError( + f"attestation evidence missing non-empty string fields: {missing}" + ) + if evidence.get("attestation_signature_verified") is not True: + return CheckResult( + "tee_tpm_attestation", + FAIL, + "TEE/TPM attestation signature has not been verified", + "Verify the TPM quote signature and attestation-service signature before accepting PCR_MATCH.", + ) + return CheckResult( + "tee_tpm_attestation", + PASS, + f"PCR_MATCH=TRUE for node {evidence['node_id']} using verifier {evidence['verifier_id']}", + ) + + +def overall_status(results: list[CheckResult]) -> str: + """Return the aggregate RED/AMBER/GREEN status for *results*.""" + if any(result.is_failure for result in results): + return "RED" + if any(result.status == WARN for result in results): + return "AMBER" + return "GREEN" + + +def render_markdown(results: list[CheckResult]) -> str: + """Render check results as a concise Markdown report.""" + lines = [ + "# Daily Sentinel AI v2.4 Operational Check", + "", + f"Overall status: **{overall_status(results)}**", + "", + "| Check | Status | Summary |", + "|---|---:|---|", + ] + for result in results: + lines.append(f"| {result.name} | {result.status} | {result.summary} |") + remediations = [result for result in results if result.remediation] + if remediations: + lines.extend(["", "## Required remediation", ""]) + for result in remediations: + lines.append(f"- **{result.name}:** {result.remediation}") + return "\n".join(lines) + "\n" + + +def render_json(results: list[CheckResult]) -> str: + """Render check results as machine-readable JSON.""" + payload = { + "generated_at_utc": dt.datetime.now(UTC) + .replace(microsecond=0) + .isoformat() + .replace("+00:00", "Z"), + "overall_status": overall_status(results), + "results": [result.to_dict() for result in results], + } + return json.dumps(payload, indent=2) + + +def result_from_check(name: str, check: Callable[[], CheckResult]) -> CheckResult: + """Convert an individual check exception into its own RED result.""" + try: + return check() + except CheckError as exc: + return CheckResult( + name, + FAIL, + str(exc), + "Regenerate the evidence item and rerun the daily check.", + ) + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Run Sentinel AI v2.4 daily operational checks" + ) + parser.add_argument("--dashboard-url", default=DEFAULT_DASHBOARD_URL) + parser.add_argument("--dashboard-timeout", type=float, default=5.0) + parser.add_argument( + "--insecure-tls", + action="store_true", + help="Disable TLS certificate validation for dashboard check", + ) + parser.add_argument( + "--skip-dashboard", + action="store_true", + help="Skip live dashboard reachability check", + ) + parser.add_argument( + "--gsri-evidence", + type=Path, + required=True, + help="JSON evidence for G-SRI threshold state", + ) + parser.add_argument( + "--worm-evidence", + type=Path, + required=True, + help="JSON evidence for pqc_worm_logger.py S3 Object Lock commit", + ) + parser.add_argument( + "--expected-worm-bucket", + help="Designated S3 Object Lock bucket expected for WORM commits", + ) + parser.add_argument( + "--require-compliance-object-lock", + action="store_true", + help="Reject GOVERNANCE mode and require COMPLIANCE Object Lock", + ) + parser.add_argument( + "--attestation-evidence", + type=Path, + required=True, + help="JSON evidence for TEE/TPM attestation", + ) + parser.add_argument( + "--max-age-minutes", type=int, default=DEFAULT_STALENESS_MINUTES + ) + parser.add_argument( + "--worm-max-lag-seconds", type=int, default=DEFAULT_WORM_MAX_LAG_SECONDS + ) + parser.add_argument( + "--json", + action="store_true", + help="Emit machine-readable JSON instead of Markdown", + ) + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + results: list[CheckResult] = [] + if not args.skip_dashboard: + results.append( + check_dashboard( + args.dashboard_url, args.dashboard_timeout, args.insecure_tls + ) + ) + else: + results.append( + CheckResult( + "sentinel_dashboard", WARN, "dashboard reachability skipped by operator" + ) + ) + results.append( + result_from_check( + "gsri_threshold", + lambda: check_gsri(load_json(args.gsri_evidence), args.max_age_minutes), + ) + ) + results.append( + result_from_check( + "pqc_worm_logger", + lambda: check_worm( + load_json(args.worm_evidence), + args.max_age_minutes, + args.worm_max_lag_seconds, + args.expected_worm_bucket, + args.require_compliance_object_lock, + ), + ) + ) + results.append( + result_from_check( + "tee_tpm_attestation", + lambda: check_attestation( + load_json(args.attestation_evidence), args.max_age_minutes + ), + ) + ) + + if args.json: + print(render_json(results)) + else: + print(render_markdown(results)) + return 1 if any(result.is_failure for result in results) else 0 + + +if __name__ == "__main__": # pragma: no cover + sys.exit(main()) diff --git a/tests/test_daily_sentinel_operational_check.py b/tests/test_daily_sentinel_operational_check.py new file mode 100644 index 00000000..662b232e --- /dev/null +++ b/tests/test_daily_sentinel_operational_check.py @@ -0,0 +1,260 @@ +import datetime as dt +import json +from pathlib import Path + +import pytest + +import scripts.daily_sentinel_operational_check as checker + + +def utc_now_z() -> str: + return ( + dt.datetime.now(dt.timezone.utc) + .replace(microsecond=0) + .isoformat() + .replace("+00:00", "Z") + ) + + +def future_z(days: int = 365) -> str: + return ( + (dt.datetime.now(dt.timezone.utc) + dt.timedelta(days=days)) + .replace(microsecond=0) + .isoformat() + .replace("+00:00", "Z") + ) + + +def write_json(path: Path, payload: dict) -> Path: + path.write_text(json.dumps(payload), encoding="utf-8") + return path + + +def valid_gsri() -> dict: + return { + "timestamp_utc": utc_now_z(), + "current_value": 0.41, + "threshold": 0.7, + "policy_version": "gsri-risk-appetite-2026.05", + "signed": True, + } + + +def valid_worm() -> dict: + return { + "timestamp_utc": utc_now_z(), + "logger_name": "pqc_worm_logger.py", + "batch_status": "committed", + "bucket": "sentinel-object-lock-audit", + "object_key": "audit/2026/05/29/batch-0001.jsonl", + "object_version_id": "3HL4kqtJlcpXrof3vjVBH40Nrjfkd", + "object_lock_mode": "COMPLIANCE", + "retention_until_utc": future_z(), + "merkle_root": "sha256:abc123", + "pqc_signature_alg": "ML-DSA-65", + "pqc_signature_verified": True, + "commit_lag_seconds": 42, + } + + +def valid_attestation() -> dict: + return { + "timestamp_utc": utc_now_z(), + "PCR_MATCH": True, + "tee_status": "trusted", + "node_id": "sentinel-prod-01", + "pcr_policy_hash": "sha256:def456", + "tpm_quote_id": "quote-20260529-0001", + "verifier_id": "attestor-v2.4.3", + "attestation_signature_verified": True, + } + + +def test_check_gsri_passes_when_signed_and_below_threshold() -> None: + result = checker.check_gsri(valid_gsri(), max_age_minutes=30) + assert result.status == checker.PASS + assert "below threshold" in result.summary + + +def test_check_gsri_fails_when_unsigned() -> None: + evidence = valid_gsri() + evidence["signed"] = False + result = checker.check_gsri(evidence, max_age_minutes=30) + assert result.status == checker.FAIL + assert "signature is not verified" in result.summary + + +def test_check_gsri_fails_when_above_threshold() -> None: + evidence = valid_gsri() + evidence["current_value"] = 0.99 + result = checker.check_gsri(evidence, max_age_minutes=30) + assert result.status == checker.FAIL + assert "at or above" in result.summary + + +def test_check_worm_passes_for_committed_object_lock_batch() -> None: + result = checker.check_worm(valid_worm(), max_age_minutes=30, max_lag_seconds=900) + assert result.status == checker.PASS + assert "Object Lock" in result.summary + + +def test_check_worm_fails_for_lag_breach() -> None: + evidence = valid_worm() + evidence["commit_lag_seconds"] = 901 + result = checker.check_worm(evidence, max_age_minutes=30, max_lag_seconds=900) + assert result.status == checker.FAIL + assert "exceeds" in result.summary + + +def test_check_worm_rejects_wrong_logger_name() -> None: + evidence = valid_worm() + evidence["logger_name"] = "other.py" + with pytest.raises(checker.CheckError, match="pqc_worm_logger.py"): + checker.check_worm(evidence, max_age_minutes=30, max_lag_seconds=900) + + +def test_check_worm_fails_for_expected_bucket_mismatch() -> None: + evidence = valid_worm() + result = checker.check_worm( + evidence, + max_age_minutes=30, + max_lag_seconds=900, + expected_bucket="different-object-lock-bucket", + ) + assert result.status == checker.FAIL + assert "does not match expected bucket" in result.summary + + +def test_check_worm_fails_when_pqc_signature_not_verified() -> None: + evidence = valid_worm() + evidence["pqc_signature_verified"] = False + result = checker.check_worm(evidence, max_age_minutes=30, max_lag_seconds=900) + assert result.status == checker.FAIL + assert "PQC signature" in result.summary + + +def test_check_attestation_passes_for_pcr_match() -> None: + result = checker.check_attestation(valid_attestation(), max_age_minutes=30) + assert result.status == checker.PASS + assert "PCR_MATCH=TRUE" in result.summary + + +def test_check_attestation_fails_for_pcr_mismatch() -> None: + evidence = valid_attestation() + evidence["PCR_MATCH"] = False + result = checker.check_attestation(evidence, max_age_minutes=30) + assert result.status == checker.FAIL + assert "PCR_MATCH" in result.summary + + +def test_check_attestation_fails_when_signature_not_verified() -> None: + evidence = valid_attestation() + evidence["attestation_signature_verified"] = False + result = checker.check_attestation(evidence, max_age_minutes=30) + assert result.status == checker.FAIL + assert "signature" in result.summary + + +def test_parse_utc_timestamp_requires_z_suffix() -> None: + with pytest.raises(checker.CheckError, match="ending in 'Z'"): + checker.parse_utc_timestamp("2026-05-29T12:00:00+00:00", "timestamp_utc") + + +def test_main_returns_zero_for_valid_offline_evidence( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + gsri = write_json(tmp_path / "gsri.json", valid_gsri()) + worm = write_json(tmp_path / "worm.json", valid_worm()) + attestation = write_json(tmp_path / "attestation.json", valid_attestation()) + + rc = checker.main( + [ + "--skip-dashboard", + "--gsri-evidence", + str(gsri), + "--worm-evidence", + str(worm), + "--attestation-evidence", + str(attestation), + ] + ) + + assert rc == 0 + assert "Overall status: **AMBER**" in capsys.readouterr().out + + +def test_main_returns_nonzero_when_evidence_is_stale( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + stale = valid_gsri() + stale["timestamp_utc"] = "2026-01-01T00:00:00Z" + gsri = write_json(tmp_path / "gsri.json", stale) + worm = write_json(tmp_path / "worm.json", valid_worm()) + attestation = write_json(tmp_path / "attestation.json", valid_attestation()) + + rc = checker.main( + [ + "--skip-dashboard", + "--gsri-evidence", + str(gsri), + "--worm-evidence", + str(worm), + "--attestation-evidence", + str(attestation), + ] + ) + + assert rc == 1 + assert "gsri_threshold" in capsys.readouterr().out + + +def test_main_continues_after_individual_evidence_error( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + bad_gsri = write_json(tmp_path / "bad-gsri.json", {"timestamp_utc": "not-z"}) + worm = write_json(tmp_path / "worm.json", valid_worm()) + attestation = write_json(tmp_path / "attestation.json", valid_attestation()) + + rc = checker.main( + [ + "--skip-dashboard", + "--gsri-evidence", + str(bad_gsri), + "--worm-evidence", + str(worm), + "--attestation-evidence", + str(attestation), + ] + ) + + output = capsys.readouterr().out + assert rc == 1 + assert "gsri_threshold" in output + assert "pqc_worm_logger | PASS" in output + assert "tee_tpm_attestation | PASS" in output + + +def test_json_output_contains_overall_status_and_results( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + gsri = write_json(tmp_path / "gsri.json", valid_gsri()) + worm = write_json(tmp_path / "worm.json", valid_worm()) + attestation = write_json(tmp_path / "attestation.json", valid_attestation()) + + rc = checker.main( + [ + "--skip-dashboard", + "--json", + "--gsri-evidence", + str(gsri), + "--worm-evidence", + str(worm), + "--attestation-evidence", + str(attestation), + ] + ) + + payload = json.loads(capsys.readouterr().out) + assert rc == 0 + assert payload["overall_status"] == "AMBER" + assert len(payload["results"]) == 4