Validation executor: hardened shared reports and eligibility

src/openenv/validation/__init__.py

@@ -2,6 +2,7 @@
 
 """Versioned, spec-neutral OpenEnv validation contracts."""
 
+from .executor import execute_validation_plan
 from .models import (
     CheckOutcome,
     RunnerCapabilities,
@@ -40,4 +41,5 @@
     "ValidationSeverity",
     "ValidationStatus",
     "build_validation_plan",
+    "execute_validation_plan",
 ]

src/openenv/validation/executor.py

@@ -0,0 +1,254 @@
+# SPDX-License-Identifier: BSD-3-Clause
+
+"""Capability-aware executor for shared validation plans."""
+
+from __future__ import annotations
+
+import re
+import signal
+import threading
+import time
+from contextlib import contextmanager
+from datetime import datetime, timezone
+from typing import Any, Iterator, Mapping
+
+from .models import (
+    CheckOutcome,
+    ValidationContext,
+    ValidationPlan,
+    ValidationProfile,
+    ValidationReport,
+    ValidationResult,
+    ValidationSeverity,
+    ValidationStatus,
+)
+from .planner import is_canonical_policy_plan
+from .serialization import json_safe, redact_string
+
+
+class _CheckTimedOut(BaseException):
+    pass
+
+
+class _TimeoutContextUnsupported(BaseException):
+    pass
+
+
+@contextmanager
+def _timeout(seconds: float) -> Iterator[None]:
+    supported = (
+        hasattr(signal, "SIGALRM")
+        and threading.current_thread() is threading.main_thread()
+    )
+    if not supported:
+        raise _TimeoutContextUnsupported
+
+    previous_handler = signal.getsignal(signal.SIGALRM)
+    previous_timer = signal.getitimer(signal.ITIMER_REAL)
+
+    def handle_timeout(_signum: int, _frame: Any) -> None:
+        raise _CheckTimedOut
+
+    signal.signal(signal.SIGALRM, handle_timeout)
+    signal.setitimer(signal.ITIMER_REAL, seconds)
+    try:
+        yield
+    finally:
+        signal.setitimer(signal.ITIMER_REAL, *previous_timer)
+        signal.signal(signal.SIGALRM, previous_handler)
+
+
+def _execute_check(
+    plan: ValidationPlan,
+    context: ValidationContext,
+    index: int,
+) -> ValidationResult:
+    check = plan.checks[index]
+    missing = check.capabilities - plan.capabilities.available
+    if missing:
+        return ValidationResult(
+            criterion_id=check.criterion_id,
+            requirement=check.requirement,
+            status=ValidationStatus.SKIP,
+            severity=check.severity,
+            evidence={
+                "reason": "runner_capability_unavailable",
+                "required": sorted(
+                    capability.value for capability in check.capabilities
+                ),
+                "missing": sorted(capability.value for capability in missing),
+                "available": sorted(
+                    capability.value for capability in plan.capabilities.available
+                ),
+            },
+            duration_s=0.0,
+            timeout_s=check.timeout_s,
+            required_capabilities=check.capabilities,
+            built_in=check.built_in,
+            message="Runner does not provide the capabilities required by this check",
+        )
+
+    started = time.perf_counter()
+    if check.evaluator is None:
+        return ValidationResult(
+            criterion_id=check.criterion_id,
+            requirement=check.requirement,
+            status=ValidationStatus.ERROR,
+            severity=check.severity,
+            evidence={"reason": "missing_check_implementation"},
+            duration_s=time.perf_counter() - started,
+            timeout_s=check.timeout_s,
+            required_capabilities=check.capabilities,
+            built_in=check.built_in,
+            message="No check implementation is registered",
+        )
+
+    try:
+        with _timeout(check.timeout_s):
+            outcome = check.evaluator(context)
+        if not isinstance(outcome, CheckOutcome):
+            outcome = CheckOutcome.error(
+                {"actual_type": type(outcome).__name__},
+                message="Check implementation returned an invalid outcome",
+            )
+        elif (
+            not isinstance(outcome.status, ValidationStatus)
+            or not isinstance(outcome.evidence, Mapping)
+            or (outcome.message is not None and not isinstance(outcome.message, str))
+        ):
+            outcome = CheckOutcome.error(
+                {"reason": "invalid_outcome_shape"},
+                message="Check implementation returned malformed status, evidence, or message",
+            )
+        elif time.perf_counter() - started > check.timeout_s:
+            outcome = CheckOutcome.error(
+                {"reason": "timeout", "timeout_s": check.timeout_s},
+                message=f"Check exceeded its {check.timeout_s:g}s timeout",
+            )
+    except _CheckTimedOut:
+        outcome = CheckOutcome.error(
+            {"reason": "timeout", "timeout_s": check.timeout_s},
+            message=f"Check exceeded its {check.timeout_s:g}s timeout",
+        )
+    except _TimeoutContextUnsupported:
+        outcome = CheckOutcome.error(
+            {
+                "reason": "timeout_context_unsupported",
+                "timeout_s": check.timeout_s,
+            },
+            message=(
+                "Timed checks must run on the POSIX main thread or in a "
+                "runner-managed worker process"
+            ),
+        )
+    except Exception as exc:
+        outcome = CheckOutcome.error(
+            {"reason": "uncaught_exception", "error_type": type(exc).__name__},
+            message="Check implementation raised an unexpected exception",
+        )
+
+    try:
+        evidence = json_safe(outcome.evidence)
+    except Exception as exc:
+        outcome = CheckOutcome.error(
+            {
+                "reason": "invalid_evidence",
+                "error_type": type(exc).__name__,
+            },
+            message="Check evidence is not safe to serialize",
+        )
+        evidence = json_safe(outcome.evidence)
+
+    return ValidationResult(
+        criterion_id=check.criterion_id,
+        requirement=check.requirement,
+        status=outcome.status,
+        # Severity is policy, not runner evidence. A subject or lane must not
+        # downgrade a canonical blocking criterion through CheckOutcome.
+        severity=check.severity,
+        evidence=evidence,
+        duration_s=time.perf_counter() - started,
+        timeout_s=check.timeout_s,
+        required_capabilities=check.capabilities,
+        built_in=check.built_in,
+        message=(
+            redact_string(outcome.message) if outcome.message is not None else None
+        ),
+    )
+
+
+def execute_validation_plan(
+    plan: ValidationPlan, context: ValidationContext
+) -> ValidationReport:
+    """Execute every planned check and preserve partial results on errors."""
+    started_at = datetime.now(timezone.utc)
+    started = time.perf_counter()
+    results = tuple(
+        _execute_check(plan, context, index) for index in range(len(plan.checks))
+    )
+    finished_at = datetime.now(timezone.utc)
+
+    blocking_skips = any(
+        result.severity is ValidationSeverity.BLOCKING
+        and result.status is ValidationStatus.SKIP
+        for result in results
+    )
+    all_blocking_pass = not any(
+        result.severity is ValidationSeverity.BLOCKING
+        and result.status is not ValidationStatus.PASS
+        for result in results
+    )
+    policy_complete = bool(
+        plan.profile is ValidationProfile.FULL
+        and is_canonical_policy_plan(plan, context)
+    )
+    repo_sha_valid = bool(
+        isinstance(context.repo_sha, str)
+        and re.fullmatch(r"(?:[0-9a-f]{40}|[0-9a-f]{64})", context.repo_sha)
+    )
+    image_digest_valid = bool(
+        isinstance(context.image_digest, str)
+        and re.fullmatch(r"sha256:[0-9a-f]{64}", context.image_digest)
+    )
+    certification_eligible = bool(
+        policy_complete
+        and plan.capabilities.official
+        and plan.capabilities.isolation_mode == "dedicated"
+        and context.discovered.get("runner_attestation_verified") is True
+        and context.discovered.get("provenance_binding_verified") is True
+        and repo_sha_valid
+        and image_digest_valid
+        and not blocking_skips
+        and all_blocking_pass
+    )
+    return ValidationReport(
+        target=(
+            redact_string(plan.target)
+            if isinstance(plan.target, str)
+            else "[INVALID_TARGET]"
+        ),
+        profile=plan.profile,
+        policy_version=plan.policy_version,
+        runner=plan.capabilities,
+        results=results,
+        duration_s=time.perf_counter() - started,
+        started_at=started_at.isoformat(),
+        finished_at=finished_at.isoformat(),
+        spec=plan.spec,
+        spec_identity=plan.spec_identity,
+        repo_sha=(
+            redact_string(context.repo_sha)
+            if isinstance(context.repo_sha, str)
+            else None
+        ),
+        image_digest=(
+            redact_string(context.image_digest)
+            if isinstance(context.image_digest, str)
+            else None
+        ),
+        # Certification is a property of the signed registry envelope produced
+        # after this immutable payload exists. An executor cannot verify a
+        # signature over a report it has not produced yet.
+        certified=False,
+        certification_eligible=certification_eligible,
+    )

src/openenv/validation/models.py

@@ -421,12 +421,26 @@ class ValidationReport:
     duration_s: float
     started_at: str
     finished_at: str
+    spec: ValidationSubject | None = None
+    spec_identity: SpecIdentity | None = None
     repo_sha: str | None = None
     image_digest: str | None = None
     certified: bool = False
     certification_eligible: bool = False
     report_schema_version: str = "1.0"
 
+    def __post_init__(self) -> None:
+        if not _STABLE_IDENTIFIER.fullmatch(self.policy_version):
+            raise ValueError(
+                "ValidationReport policy version must be a stable identifier"
+            )
+        if (
+            self.spec is not None
+            and self.spec_identity is not None
+            and self.spec_identity != self.spec.spec
+        ):
+            raise ValueError("ValidationReport spec identity must match its subject")
+
     @property
     def passed(self) -> bool:
         """Return whether all executed blocking criteria avoid failure/error."""
@@ -481,9 +495,18 @@ def _summary(self) -> dict[str, Any]:
 
     def to_dict(self) -> dict[str, Any]:
         """Return the versioned JSON report."""
-        return {
+        payload = {
             "report_schema_version": self.report_schema_version,
             "policy_version": self.policy_version,
+            "spec": (
+                self.spec.to_dict()
+                if self.spec is not None
+                else (
+                    self.spec_identity.to_dict()
+                    if self.spec_identity is not None
+                    else None
+                )
+            ),
             "target": self.target,
             "validation_type": "openenv_validation",
             "profile": self.profile.value,
@@ -500,3 +523,29 @@ def to_dict(self) -> dict[str, Any]:
             "summary": self._summary(),
             "criteria": [result.to_dict() for result in self.results],
         }
+        serialized = json_safe(
+            payload,
+            trusted_string_keys=_TRUSTED_METADATA_STRING_KEYS,
+        )
+        assert isinstance(serialized, dict)
+        if isinstance(payload["spec"], Mapping) and isinstance(
+            serialized.get("spec"), dict
+        ):
+            _restore_structural_spec_identity(serialized["spec"], payload["spec"])
+        serialized_criteria = serialized.get("criteria")
+        if isinstance(serialized_criteria, list):
+            for safe_result, raw_result in zip(
+                serialized_criteria, payload["criteria"], strict=True
+            ):
+                if isinstance(safe_result, dict):
+                    safe_result["description"] = raw_result["description"]
+                    safe_result["requirement"] = raw_result["requirement"]
+        serialized_summary = serialized.get("summary")
+        if isinstance(serialized_summary, dict):
+            serialized_summary["failed_criteria"] = payload["summary"][
+                "failed_criteria"
+            ]
+            serialized_summary["blocking_failed_criteria"] = payload["summary"][
+                "blocking_failed_criteria"
+            ]
+        return serialized