feat(validation): add hardened plan executor

Author: zkwentz

src/openenv/validation/__init__.py

@@ -2,6 +2,7 @@
 
 """Versioned OpenEnv validation contracts and Harbor manifest loading."""
 
+from .executor import execute_validation_plan
 from .models import (
     CheckOutcome,
     RunnerCapabilities,
@@ -52,5 +53,6 @@
     "ValidationSeverity",
     "ValidationStatus",
     "build_validation_plan",
+    "execute_validation_plan",
     "load_task_manifest",
 ]

src/openenv/validation/executor.py

@@ -0,0 +1,370 @@
+# SPDX-License-Identifier: BSD-3-Clause
+
+"""Capability-aware executor for shared validation plans."""
+
+from __future__ import annotations
+
+import json
+import math
+import re
+import signal
+import threading
+import time
+from contextlib import contextmanager
+from datetime import datetime, timezone
+from enum import Enum
+from pathlib import Path
+from typing import Any, Iterator, Mapping
+from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
+
+from .models import (
+    CheckOutcome,
+    ValidationContext,
+    ValidationPlan,
+    ValidationProfile,
+    ValidationReport,
+    ValidationResult,
+    ValidationSeverity,
+    ValidationStatus,
+)
+from .planner import is_canonical_policy_plan
+
+
+_SECRET_KEY = re.compile(
+    r"(?i)(authorization|cookie|credential|password|private.?key|secret|token|api.?key)"
+)
+_SECRET_VALUE_PATTERNS = (
+    re.compile(r"(?i)(bearer\s+)[^\s,;]+"),
+    re.compile(r"\b(?:hf_|sk-|github_pat_)[A-Za-z0-9_-]{8,}\b", re.IGNORECASE),
+)
+_MAX_EVIDENCE_DEPTH = 32
+_MAX_EVIDENCE_NODES = 10_000
+_MAX_EVIDENCE_STRING_LENGTH = 1_000_000
+
+
+class _CheckTimedOut(BaseException):
+    pass
+
+
+class _TimeoutContextUnsupported(BaseException):
+    pass
+
+
+def _redact_string(value: str) -> str:
+    redacted = value
+    for pattern in _SECRET_VALUE_PATTERNS:
+        redacted = pattern.sub(
+            lambda match: (
+                f"{match.group(1)}[REDACTED]" if match.lastindex else "[REDACTED]"
+            ),
+            redacted,
+        )
+    try:
+        parsed = urlsplit(redacted)
+        hostname = parsed.hostname
+        port = parsed.port
+    except (UnicodeError, ValueError):
+        if redacted.lower().startswith(("http://", "https://", "ws://", "wss://")):
+            return "[REDACTED_INVALID_URL]"
+        return redacted
+    if parsed.scheme not in {"http", "https", "ws", "wss"} or not hostname:
+        return redacted
+    host = hostname
+    if ":" in host and not host.startswith("["):
+        host = f"[{host}]"
+    if port is not None:
+        host = f"{host}:{port}"
+    query = []
+    for key, item in parse_qsl(parsed.query, keep_blank_values=True):
+        query.append((key, "[REDACTED]" if _SECRET_KEY.search(key) else item))
+    return urlunsplit((parsed.scheme, host, parsed.path, urlencode(query), ""))
+
+
+def _json_safe(
+    value: Any,
+    *,
+    key: str | None = None,
+    seen: set[int] | None = None,
+    depth: int = 0,
+    budget: list[int] | None = None,
+) -> Any:
+    if depth > _MAX_EVIDENCE_DEPTH:
+        raise ValueError("Evidence exceeds the maximum nesting depth")
+    if budget is None:
+        budget = [_MAX_EVIDENCE_NODES]
+    budget[0] -= 1
+    if budget[0] < 0:
+        raise ValueError("Evidence exceeds the maximum node count")
+    if key is not None and _SECRET_KEY.search(key):
+        return "[REDACTED]"
+    if value is None or isinstance(value, (bool, int)):
+        return value
+    if isinstance(value, float):
+        if not math.isfinite(value):
+            raise ValueError("Evidence contains a non-finite float")
+        return value
+    if isinstance(value, str):
+        if len(value) > _MAX_EVIDENCE_STRING_LENGTH:
+            raise ValueError("Evidence string exceeds the maximum length")
+        return _redact_string(value)
+    if isinstance(value, Enum):
+        return _json_safe(value.value, seen=seen, depth=depth + 1, budget=budget)
+    if isinstance(value, Path):
+        return _json_safe(str(value), seen=seen, depth=depth + 1, budget=budget)
+    if seen is None:
+        seen = set()
+    container_id = id(value)
+    if isinstance(value, (Mapping, list, tuple, set, frozenset)):
+        if container_id in seen:
+            raise ValueError("Evidence contains a reference cycle")
+        seen.add(container_id)
+    if isinstance(value, Mapping):
+        items: list[tuple[str, Any]] = []
+        for index, (item_key, item) in enumerate(value.items()):
+            if index >= _MAX_EVIDENCE_NODES:
+                raise ValueError("Evidence mapping exceeds the maximum size")
+            if not isinstance(item_key, (str, int, float, bool)):
+                raise ValueError("Evidence mapping contains an unsupported key")
+            if isinstance(item_key, float) and not math.isfinite(item_key):
+                raise ValueError("Evidence mapping contains a non-finite key")
+            key_text = str(item_key)
+            if len(key_text) > _MAX_EVIDENCE_STRING_LENGTH:
+                raise ValueError("Evidence key exceeds the maximum length")
+            items.append((key_text, item))
+        items.sort(key=lambda pair: pair[0])
+        result: dict[str, Any] = {}
+        for item_key, item in items:
+            if item_key in result:
+                raise ValueError("Evidence mapping contains duplicate JSON keys")
+            result[item_key] = _json_safe(
+                item,
+                key=item_key,
+                seen=seen,
+                depth=depth + 1,
+                budget=budget,
+            )
+        seen.remove(container_id)
+        return result
+    if isinstance(value, (list, tuple)):
+        result = [
+            _json_safe(item, seen=seen, depth=depth + 1, budget=budget)
+            for item in value
+        ]
+        seen.remove(container_id)
+        return result
+    if isinstance(value, (set, frozenset)):
+        converted = [
+            _json_safe(item, seen=seen, depth=depth + 1, budget=budget)
+            for item in value
+        ]
+        seen.remove(container_id)
+        return sorted(converted, key=lambda item: json.dumps(item, sort_keys=True))
+    return {"type": type(value).__name__}
+
+
+@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(
+        context.repo_sha and re.fullmatch(r"[0-9a-fA-F]{40,64}", context.repo_sha)
+    )
+    image_digest_valid = bool(
+        context.image_digest
+        and re.fullmatch(r"sha256:[0-9a-fA-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 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(),
+        repo_sha=context.repo_sha,
+        image_digest=context.image_digest,
+        # 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,
+    )