diff --git a/.claude/docs/REPO_WALKTHROUGH.md b/.claude/docs/REPO_WALKTHROUGH.md index 7f9c8d9da..e67f95072 100644 --- a/.claude/docs/REPO_WALKTHROUGH.md +++ b/.claude/docs/REPO_WALKTHROUGH.md @@ -58,6 +58,8 @@ src/ │ │ ├── local_python_executor.py # Python code execution │ │ └── git_server_client.py # Git operations │ │ +│ ├── validation/ # Spec adapters, versioned policy, planning, and reports +│ │ │ └── cli/ # Command-line interface │ ├── __main__.py # Entry point (`python -m openenv.cli`) │ ├── commands/ # CLI subcommands @@ -118,6 +120,7 @@ tests/ │ └── ... │ ├── test_cli/ # CLI command tests +├── test_validation/ # Validation policy and execution tests └── scripts/ # Test utility scripts ``` diff --git a/src/openenv/validation/__init__.py b/src/openenv/validation/__init__.py new file mode 100644 index 000000000..c920f6abc --- /dev/null +++ b/src/openenv/validation/__init__.py @@ -0,0 +1,81 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Versioned, spec-neutral OpenEnv validation contracts.""" + +from .models import ( + CheckOutcome, + RunnerCapabilities, + ValidationCapability, + ValidationCheck, + ValidationContext, + ValidationPlan, + ValidationProfile, + ValidationReport, + ValidationResult, + ValidationSeverity, + ValidationStatus, +) +from .specs import ( + AdapterIdentity, + ArtifactRequirement, + DEFAULT_SPEC_REGISTRY, + DetectionMode, + EnvironmentRequirements, + ExecutionModel, + HealthcheckRequirement, + load_harbor_requirements, + NetworkMode, + NetworkRequirements, + OpenEnvSpecAdapter, + PackageIdentity, + PhaseRequirements, + RequirementsLoad, + RequirementsProvenance, + RequirementsState, + SpecIdentity, + SpecLoad, + SpecLoadState, + StepRequirements, + ValidationRequirements, + ValidationSpecAdapter, + ValidationSpecRegistry, + ValidationSubject, +) + +__all__ = [ + "AdapterIdentity", + "ArtifactRequirement", + "CheckOutcome", + "DEFAULT_SPEC_REGISTRY", + "DetectionMode", + "EnvironmentRequirements", + "ExecutionModel", + "HealthcheckRequirement", + "NetworkMode", + "NetworkRequirements", + "OpenEnvSpecAdapter", + "PackageIdentity", + "PhaseRequirements", + "RequirementsLoad", + "RequirementsProvenance", + "RequirementsState", + "RunnerCapabilities", + "SpecIdentity", + "SpecLoad", + "SpecLoadState", + "StepRequirements", + "ValidationCapability", + "ValidationCheck", + "ValidationContext", + "ValidationPlan", + "ValidationProfile", + "ValidationReport", + "ValidationResult", + "ValidationSeverity", + "ValidationSpecAdapter", + "ValidationSpecRegistry", + "ValidationStatus", + "ValidationSubject", + "ValidationRequirements", + "load_harbor_requirements", +] diff --git a/src/openenv/validation/models.py b/src/openenv/validation/models.py new file mode 100644 index 000000000..ea0be6d96 --- /dev/null +++ b/src/openenv/validation/models.py @@ -0,0 +1,382 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Shared models for local validation and remote certification runners.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum +from pathlib import Path +from typing import Any, Callable, Mapping + +from .specs.base import SpecLoad + + +class ValidationStatus(str, Enum): + """Outcome of one validation criterion.""" + + PASS = "pass" + FAIL = "fail" + SKIP = "skip" + ERROR = "error" + + +class ValidationSeverity(str, Enum): + """Policy effect of a validation result.""" + + BLOCKING = "blocking" + ADVISORY = "advisory" + + +class ValidationProfile(str, Enum): + """Set of validation checks requested by a caller.""" + + STATIC = "static" + RUNTIME = "runtime" + FULL = "full" + + +class ValidationCapability(str, Enum): + """Execution capabilities that checks may require.""" + + SOURCE = "source" + RUNTIME = "runtime" + CONTAINER_IMAGE = "container_image" + NETWORK_ENFORCEMENT = "network_enforcement" + GPU = "gpu" + CROSS_HOST = "cross_host" + VERIFIER_ISOLATION = "verifier_isolation" + REFERENCE_MODEL = "reference_model" + SIGNATURE_VERIFICATION = "signature_verification" + + +@dataclass(frozen=True) +class RunnerCapabilities: + """Capabilities and trust properties exposed by a validation runner.""" + + runner: str + available: frozenset[ValidationCapability] = frozenset() + official: bool = False + isolation_mode: str | None = None + + def supports(self, required: frozenset[ValidationCapability]) -> bool: + """Return whether all requested capabilities are available.""" + return required.issubset(self.available) + + def to_dict(self) -> dict[str, Any]: + """Return a JSON-safe runner description.""" + return { + "kind": self.runner, + "capabilities": sorted(capability.value for capability in self.available), + "official": self.official, + "isolation_mode": self.isolation_mode, + } + + +@dataclass(frozen=True) +class CheckOutcome: + """Raw outcome returned by a validation check implementation.""" + + status: ValidationStatus + evidence: Mapping[str, Any] = field(default_factory=dict) + message: str | None = None + severity: ValidationSeverity | None = None + + @classmethod + def pass_( + cls, + evidence: Mapping[str, Any] | None = None, + *, + message: str | None = None, + severity: ValidationSeverity | None = None, + ) -> "CheckOutcome": + """Build a passing outcome.""" + return cls( + ValidationStatus.PASS, + evidence or {}, + message=message, + severity=severity, + ) + + @classmethod + def fail( + cls, + evidence: Mapping[str, Any] | None = None, + *, + message: str | None = None, + severity: ValidationSeverity | None = None, + ) -> "CheckOutcome": + """Build a failing outcome.""" + return cls( + ValidationStatus.FAIL, + evidence or {}, + message=message, + severity=severity, + ) + + @classmethod + def skip( + cls, + evidence: Mapping[str, Any] | None = None, + *, + message: str | None = None, + severity: ValidationSeverity | None = None, + ) -> "CheckOutcome": + """Build a skipped outcome.""" + return cls( + ValidationStatus.SKIP, + evidence or {}, + message=message, + severity=severity, + ) + + @classmethod + def error( + cls, + evidence: Mapping[str, Any] | None = None, + *, + message: str | None = None, + severity: ValidationSeverity | None = None, + ) -> "CheckOutcome": + """Build an infrastructure-error outcome.""" + return cls( + ValidationStatus.ERROR, + evidence or {}, + message=message, + severity=severity, + ) + + +CheckEvaluator = Callable[["ValidationContext"], CheckOutcome] + + +@dataclass(frozen=True) +class ValidationCheck: + """Policy-independent implementation contract for one criterion.""" + + criterion_id: str + requirement: str + capabilities: frozenset[ValidationCapability] + severity: ValidationSeverity + timeout_s: float + evaluator: CheckEvaluator | None + profiles: frozenset[ValidationProfile] = field( + default_factory=lambda: frozenset(ValidationProfile) + ) + built_in: bool = True + + def __post_init__(self) -> None: + if not self.criterion_id.strip(): + raise ValueError("ValidationCheck criterion_id cannot be empty") + if not self.requirement.strip(): + raise ValueError("ValidationCheck requirement cannot be empty") + if self.timeout_s <= 0: + raise ValueError("ValidationCheck timeout_s must be positive") + + @property + def rfc_requirement(self) -> str: + """Compatibility alias for the normative requirement.""" + return self.requirement + + @property + def required_capabilities(self) -> frozenset[ValidationCapability]: + """Compatibility alias for capabilities required by this check.""" + return self.capabilities + + @property + def default_severity(self) -> ValidationSeverity: + """Compatibility alias for the policy default severity.""" + return self.severity + + +@dataclass +class ValidationContext: + """Inputs and discovery data shared by checks in one execution.""" + + target: Path | str + spec_load: SpecLoad + runtime_url: str | None = None + discovered: dict[str, Any] = field(default_factory=dict) + repo_sha: str | None = None + image_digest: str | None = None + + +@dataclass(frozen=True) +class ValidationPlan: + """Deterministic set of checks selected for a target and runner.""" + + target: str + profile: ValidationProfile + policy_version: str + capabilities: RunnerCapabilities + checks: tuple[ValidationCheck, ...] + requirements: Mapping[str, Any] = field(default_factory=dict) + _policy_attestation: object | None = field(default=None, repr=False, compare=False) + + def __post_init__(self) -> None: + criterion_ids = [check.criterion_id for check in self.checks] + if len(criterion_ids) != len(set(criterion_ids)): + raise ValueError("ValidationPlan contains duplicate criterion IDs") + + def to_dict(self) -> dict[str, Any]: + """Return the safe, execution-independent plan representation.""" + return { + "target": self.target, + "profile": self.profile.value, + "policy_version": self.policy_version, + "runner": self.capabilities.to_dict(), + "requirements": dict(self.requirements), + "checks": [ + { + "id": check.criterion_id, + "requirement": check.requirement, + "required_capabilities": sorted( + capability.value for capability in check.capabilities + ), + "severity": check.severity.value, + "timeout_s": check.timeout_s, + "built_in": check.built_in, + } + for check in self.checks + ], + } + + +@dataclass(frozen=True) +class ValidationResult: + """Structured result of executing one validation criterion.""" + + criterion_id: str + requirement: str + status: ValidationStatus + severity: ValidationSeverity + evidence: Mapping[str, Any] + duration_s: float + timeout_s: float + required_capabilities: frozenset[ValidationCapability] + built_in: bool = True + message: str | None = None + + @property + def passed(self) -> bool: + """Return whether this criterion explicitly passed.""" + return self.status is ValidationStatus.PASS + + def to_dict(self) -> dict[str, Any]: + """Return the shared local/remote criterion schema.""" + payload: dict[str, Any] = { + "id": self.criterion_id, + "description": self.requirement, + "requirement": self.requirement, + "status": self.status.value, + "severity": self.severity.value, + "passed": self.passed, + "required": self.severity is ValidationSeverity.BLOCKING, + "built_in": self.built_in, + "required_capabilities": sorted( + capability.value for capability in self.required_capabilities + ), + "evidence": dict(self.evidence), + "duration_s": round(self.duration_s, 6), + "duration_ms": round(self.duration_s * 1000, 3), + "timeout_s": self.timeout_s, + } + if self.message is not None: + payload["details"] = self.message + for compatibility_key in ("expected", "actual"): + if compatibility_key in self.evidence: + payload[compatibility_key] = self.evidence[compatibility_key] + return payload + + +@dataclass(frozen=True) +class ValidationReport: + """Versioned report shared by local and remote validation executors.""" + + target: str + profile: ValidationProfile + policy_version: str + runner: RunnerCapabilities + results: tuple[ValidationResult, ...] + duration_s: float + started_at: str + finished_at: str + repo_sha: str | None = None + image_digest: str | None = None + certified: bool = False + certification_eligible: bool = False + report_schema_version: str = "1.0" + + @property + def passed(self) -> bool: + """Return whether all executed blocking criteria avoid failure/error.""" + return not any( + result.severity is ValidationSeverity.BLOCKING + and result.status in {ValidationStatus.FAIL, ValidationStatus.ERROR} + for result in self.results + ) + + @property + def status(self) -> ValidationStatus: + """Return the aggregate report status.""" + return ValidationStatus.PASS if self.passed else ValidationStatus.FAIL + + def _summary(self) -> dict[str, Any]: + counts = { + status.value: sum(1 for result in self.results if result.status is status) + for status in ValidationStatus + } + failed = [ + result.criterion_id + for result in self.results + if result.status in {ValidationStatus.FAIL, ValidationStatus.ERROR} + ] + blocking_failed = [ + result.criterion_id + for result in self.results + if result.severity is ValidationSeverity.BLOCKING + and result.status in {ValidationStatus.FAIL, ValidationStatus.ERROR} + ] + return { + "passed_count": counts[ValidationStatus.PASS.value], + "failed_count": counts[ValidationStatus.FAIL.value], + "skipped_count": counts[ValidationStatus.SKIP.value], + "error_count": counts[ValidationStatus.ERROR.value], + "total_count": len(self.results), + "failed_criteria": failed, + "blocking_failed_criteria": blocking_failed, + "required_passed_count": sum( + 1 + for result in self.results + if result.severity is ValidationSeverity.BLOCKING + and result.status is ValidationStatus.PASS + ), + "required_total_count": sum( + 1 + for result in self.results + if result.severity is ValidationSeverity.BLOCKING + ), + "status_counts": counts, + } + + def to_dict(self) -> dict[str, Any]: + """Return the versioned JSON report.""" + return { + "report_schema_version": self.report_schema_version, + "policy_version": self.policy_version, + "target": self.target, + "validation_type": "openenv_validation", + "profile": self.profile.value, + "runner": self.runner.to_dict(), + "status": self.status.value, + "passed": self.passed, + "certified": self.certified, + "certification_eligible": self.certification_eligible, + "repo_sha": self.repo_sha, + "image_digest": self.image_digest, + "started_at": self.started_at, + "finished_at": self.finished_at, + "duration_s": round(self.duration_s, 6), + "summary": self._summary(), + "criteria": [result.to_dict() for result in self.results], + } diff --git a/src/openenv/validation/specs/__init__.py b/src/openenv/validation/specs/__init__.py new file mode 100644 index 000000000..98e3ac8fa --- /dev/null +++ b/src/openenv/validation/specs/__init__.py @@ -0,0 +1,60 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Trusted source-spec adapters and normalized validation requirements.""" + +from .base import ( + AdapterIdentity, + ArtifactRequirement, + DetectionMode, + EnvironmentRequirements, + ExecutionModel, + HealthcheckRequirement, + NetworkMode, + NetworkRequirements, + PackageIdentity, + PhaseRequirements, + RequirementsLoad, + RequirementsProvenance, + RequirementsState, + SpecIdentity, + SpecLoad, + SpecLoadState, + StepRequirements, + ValidationRequirements, + ValidationSpecAdapter, + ValidationSpecRegistry, + ValidationSubject, +) +from .harbor import find_harbor_verifier_script, load_harbor_requirements +from .openenv import OpenEnvSpecAdapter + + +DEFAULT_SPEC_REGISTRY = ValidationSpecRegistry((OpenEnvSpecAdapter(),)) + +__all__ = [ + "AdapterIdentity", + "ArtifactRequirement", + "DEFAULT_SPEC_REGISTRY", + "DetectionMode", + "EnvironmentRequirements", + "ExecutionModel", + "HealthcheckRequirement", + "NetworkMode", + "NetworkRequirements", + "OpenEnvSpecAdapter", + "PackageIdentity", + "PhaseRequirements", + "RequirementsLoad", + "RequirementsProvenance", + "RequirementsState", + "SpecIdentity", + "SpecLoad", + "SpecLoadState", + "StepRequirements", + "ValidationRequirements", + "ValidationSpecAdapter", + "ValidationSpecRegistry", + "ValidationSubject", + "find_harbor_verifier_script", + "load_harbor_requirements", +] diff --git a/src/openenv/validation/specs/base.py b/src/openenv/validation/specs/base.py new file mode 100644 index 000000000..3fbf7f1f1 --- /dev/null +++ b/src/openenv/validation/specs/base.py @@ -0,0 +1,646 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Spec-neutral contracts for validation subjects and requirements.""" + +from __future__ import annotations + +import hashlib +import re +from dataclasses import dataclass, replace +from enum import Enum +from pathlib import Path +from typing import Any, Protocol, Sequence + + +_STABLE_ID = re.compile(r"^[a-z0-9][a-z0-9._-]{0,127}$") +_SHA256_DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$") + + +class ExecutionModel(str, Enum): + """Lifecycle used to execute a validation subject.""" + + SERVED = "served" + ONE_SHOT = "one_shot" + EXTERNAL = "external" + + +class SpecLoadState(str, Enum): + """Outcome of detecting and safely inspecting a source spec.""" + + ABSENT = "absent" + LOADED = "loaded" + INVALID = "invalid" + UNSUPPORTED = "unsupported" + AMBIGUOUS = "ambiguous" + + +class RequirementsState(str, Enum): + """Outcome of loading normalized execution requirements.""" + + ABSENT = "absent" + LOADED = "loaded" + INVALID = "invalid" + UNSUPPORTED = "unsupported" + + +class DetectionMode(str, Enum): + """How a spec adapter was selected.""" + + AUTO = "auto" + EXPLICIT = "explicit" + RUNTIME = "runtime" + + +class NetworkMode(str, Enum): + """Normalized outbound-network requirement.""" + + UNSPECIFIED = "unspecified" + DENY_ALL = "deny_all" + ALLOW_ALL = "allow_all" + ALLOWLIST = "allowlist" + + +@dataclass(frozen=True) +class AdapterIdentity: + """Stable identity of trusted adapter code.""" + + adapter_id: str + adapter_version: str + + def __post_init__(self) -> None: + if not _STABLE_ID.fullmatch(self.adapter_id): + raise ValueError("adapter ID must be a stable lowercase identifier") + if not self.adapter_version.strip() or len(self.adapter_version) > 128: + raise ValueError("adapter version must be a non-empty bounded string") + + def to_dict(self) -> dict[str, str]: + return {"id": self.adapter_id, "version": self.adapter_version} + + +@dataclass(frozen=True) +class SpecIdentity: + """Source-format and execution-model provenance.""" + + spec_id: str + spec_version: str | None + adapter: AdapterIdentity + execution_model: ExecutionModel + + def __post_init__(self) -> None: + if not _STABLE_ID.fullmatch(self.spec_id): + raise ValueError("spec ID must be a stable lowercase identifier") + if self.spec_version is not None and ( + not self.spec_version.strip() or len(self.spec_version) > 128 + ): + raise ValueError("spec version must be a non-empty bounded string") + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.spec_id, + "version": self.spec_version, + "adapter": self.adapter.to_dict(), + "execution_model": self.execution_model.value, + } + + +@dataclass(frozen=True) +class RequirementsProvenance: + """Origin of a normalized requirements envelope.""" + + source_id: str + source_version: str | None + adapter: AdapterIdentity + path: str + document_digest: str | None = None + + def __post_init__(self) -> None: + if not _STABLE_ID.fullmatch(self.source_id): + raise ValueError( + "requirements source ID must be a stable lowercase identifier" + ) + if self.source_version is not None and ( + not self.source_version.strip() or len(self.source_version) > 128 + ): + raise ValueError("requirements source version must be bounded") + if ( + not self.path + or len(self.path) > 4096 + or "\x00" in self.path + or Path(self.path).is_absolute() + ): + raise ValueError("requirements source path must be relative") + if self.document_digest is not None and not _SHA256_DIGEST.fullmatch( + self.document_digest + ): + raise ValueError( + "requirements document digest must be a lowercase SHA-256 digest" + ) + + def to_dict(self) -> dict[str, Any]: + return { + "id": self.source_id, + "version": self.source_version, + "adapter": self.adapter.to_dict(), + "path": self.path, + "document_digest": self.document_digest, + } + + +@dataclass(frozen=True) +class ArtifactRequirement: + """Artifact copied from a subject environment to its verifier.""" + + source: str + destination: str | None = None + exclude: tuple[str, ...] = () + + +@dataclass(frozen=True) +class HealthcheckRequirement: + """Normalized healthcheck settings.""" + + command: str + interval_s: float = 5.0 + timeout_s: float = 30.0 + start_period_s: float = 0.0 + start_interval_s: float = 5.0 + retries: int = 3 + + +@dataclass(frozen=True) +class PackageIdentity: + """Optional package identity declared by the source spec.""" + + name: str | None = None + description: str = "" + authors: tuple[dict[str, str], ...] = () + keywords: tuple[str, ...] = () + + +@dataclass(frozen=True) +class PhaseRequirements: + """Timeout, user, and secret-name declarations for one phase.""" + + timeout_s: float | None = None + user: str | int | None = None + env_names: tuple[str, ...] = () + + +@dataclass(frozen=True) +class NetworkRequirements: + """Normalized egress policy independent of a source spec.""" + + mode: NetworkMode = NetworkMode.UNSPECIFIED + endpoints: tuple[str, ...] = () + + def __post_init__(self) -> None: + if self.mode is not NetworkMode.ALLOWLIST and self.endpoints: + raise ValueError("network endpoints require allowlist mode") + + +@dataclass(frozen=True) +class EnvironmentRequirements: + """Resource and containment requirements for a subject environment.""" + + build_timeout_s: float | None = None + container_image: str | None = None + cpus: int | None = None + memory_mb: int | None = None + storage_mb: int | None = None + gpus: int | None = None + gpu_types: tuple[str, ...] = () + network: NetworkRequirements = NetworkRequirements() + env_names: tuple[str, ...] = () + skills_dir: str | None = None + workdir: str | None = None + mcp_server_count: int = 0 + healthcheck: HealthcheckRequirement | None = None + + +@dataclass(frozen=True) +class StepRequirements: + """Normalized per-step overrides for a multi-step subject.""" + + name: str + agent: PhaseRequirements = PhaseRequirements() + verifier: PhaseRequirements = PhaseRequirements(timeout_s=600.0) + min_reward: float | dict[str, float] | None = None + healthcheck: HealthcheckRequirement | None = None + artifacts: tuple[ArtifactRequirement, ...] = () + + +@dataclass(frozen=True) +class ValidationRequirements: + """Sanitized planner inputs independent of their source format.""" + + identity: PackageIdentity = PackageIdentity() + agent: PhaseRequirements = PhaseRequirements() + verifier: PhaseRequirements = PhaseRequirements(timeout_s=600.0) + environment: EnvironmentRequirements = EnvironmentRequirements() + artifacts: tuple[ArtifactRequirement, ...] = () + steps: tuple[StepRequirements, ...] = () + metadata_keys: tuple[str, ...] = () + solution_env_names: tuple[str, ...] = () + multi_step_reward_strategy: str | None = None + source_declared: bool = False + present_fields: tuple[str, ...] = () + + def to_evidence(self) -> dict[str, Any]: + """Return report-safe requirements without environment values.""" + + def health_evidence( + value: HealthcheckRequirement | None, + ) -> dict[str, Any] | None: + if value is None: + return None + return { + "command_declared": True, + "command_digest": ( + f"sha256:{hashlib.sha256(value.command.encode('utf-8')).hexdigest()}" + ), + "interval_s": value.interval_s, + "timeout_s": value.timeout_s, + "start_period_s": value.start_period_s, + "start_interval_s": value.start_interval_s, + "retries": value.retries, + } + + def artifact_evidence(value: ArtifactRequirement) -> dict[str, Any]: + return { + "source": value.source, + "destination": value.destination, + "exclude": list(value.exclude), + } + + environment = self.environment + return { + "identity": { + "name": self.identity.name, + "description": self.identity.description, + "author_count": len(self.identity.authors), + "keywords": list(self.identity.keywords), + }, + "timeouts": { + "build_s": environment.build_timeout_s, + "agent_s": self.agent.timeout_s, + "verifier_s": self.verifier.timeout_s, + }, + "resources": { + "cpus": environment.cpus, + "memory_mb": environment.memory_mb, + "storage_mb": environment.storage_mb, + "gpus": environment.gpus, + "gpu_types": list(environment.gpu_types), + }, + "network": { + "mode": environment.network.mode.value, + "endpoints": list(environment.network.endpoints), + }, + "healthcheck": health_evidence(environment.healthcheck), + "container_image": environment.container_image, + "execution": { + "agent_user": self.agent.user, + "verifier_user": self.verifier.user, + "workdir": environment.workdir, + "skills_dir": environment.skills_dir, + "mcp_server_count": environment.mcp_server_count, + "source_declared": self.source_declared, + }, + "artifacts": [artifact_evidence(value) for value in self.artifacts], + "step_names": [step.name for step in self.steps], + "multi_step_reward_strategy": self.multi_step_reward_strategy, + "steps": [ + { + "name": step.name, + "agent_timeout_s": step.agent.timeout_s, + "verifier_timeout_s": step.verifier.timeout_s, + "agent_user": step.agent.user, + "verifier_user": step.verifier.user, + "min_reward": step.min_reward, + "healthcheck": health_evidence(step.healthcheck), + "artifacts": [artifact_evidence(value) for value in step.artifacts], + "environment_variable_names": { + "agent": list(step.agent.env_names), + "verifier": list(step.verifier.env_names), + }, + } + for step in self.steps + ], + "environment_variable_names": { + "environment": list(environment.env_names), + "agent": list(self.agent.env_names), + "verifier": list(self.verifier.env_names), + "solution": list(self.solution_env_names), + }, + "metadata_keys": list(self.metadata_keys), + "present_fields": list(self.present_fields), + } + + +@dataclass(frozen=True) +class RequirementsLoad: + """Structured result of adapting an auxiliary requirements document.""" + + state: RequirementsState + provenance: RequirementsProvenance | None = None + requirements: ValidationRequirements | None = None + error: str | None = None + + def __post_init__(self) -> None: + if self.state is RequirementsState.LOADED and self.requirements is None: + raise ValueError("loaded requirements must include normalized values") + if self.state is not RequirementsState.LOADED and self.requirements is not None: + raise ValueError("only loaded requirements may include normalized values") + if self.state in {RequirementsState.INVALID, RequirementsState.UNSUPPORTED}: + if not self.error: + raise ValueError("invalid or unsupported requirements need an error") + + def to_evidence(self) -> dict[str, Any]: + evidence: dict[str, Any] = { + "state": self.state.value, + "present": self.state is not RequirementsState.ABSENT, + } + if self.provenance is not None: + evidence["source"] = self.provenance.to_dict() + if self.requirements is not None: + evidence.update(self.requirements.to_evidence()) + if self.error is not None: + evidence["error"] = self.error + return evidence + + +@dataclass(frozen=True) +class ValidationSubject: + """Safely inspected subject supplied to planners and checks.""" + + spec: SpecIdentity + signature_path: str | None + detection_mode: DetectionMode + requirements: RequirementsLoad + verifier_script: Path | None = None + verifier_path: str | None = None + verifier_digest: str | None = None + document_digest: str | None = None + + def __post_init__(self) -> None: + if self.signature_path is not None: + signature = Path(self.signature_path) + if signature.is_absolute() or ".." in signature.parts: + raise ValueError("spec signature path must stay within the target root") + if self.document_digest is not None and not _SHA256_DIGEST.fullmatch( + self.document_digest + ): + raise ValueError("spec document digest must be a lowercase SHA-256 digest") + verifier_fields = ( + self.verifier_script, + self.verifier_path, + self.verifier_digest, + ) + if any(value is not None for value in verifier_fields) and any( + value is None for value in verifier_fields + ): + raise ValueError( + "verifier path, file, and digest must be declared together" + ) + if self.verifier_path is not None and ( + not self.verifier_path + or len(self.verifier_path) > 4096 + or "\x00" in self.verifier_path + or Path(self.verifier_path).is_absolute() + ): + raise ValueError("verifier path must be a bounded relative path") + if self.verifier_digest is not None and not _SHA256_DIGEST.fullmatch( + self.verifier_digest + ): + raise ValueError("verifier digest must be a lowercase SHA-256 digest") + + def to_dict(self) -> dict[str, Any]: + payload = self.spec.to_dict() + payload.update( + { + "detection_mode": self.detection_mode.value, + "signature_path": self.signature_path, + "document_digest": self.document_digest, + "requirements_state": self.requirements.state.value, + "requirements": ( + self.requirements.provenance.to_dict() + if self.requirements.provenance is not None + else None + ), + "verifier": ( + { + "path": self.verifier_path, + "document_digest": self.verifier_digest, + } + if self.verifier_path is not None + else None + ), + } + ) + return payload + + +@dataclass(frozen=True) +class SpecLoad: + """Result of selecting and safely inspecting a validation spec.""" + + state: SpecLoadState + subject: ValidationSubject | None = None + identity: SpecIdentity | None = None + error: str | None = None + matches: tuple[str, ...] = () + + def __post_init__(self) -> None: + if self.state is SpecLoadState.LOADED and self.subject is None: + raise ValueError("loaded spec must include a validation subject") + if self.state is not SpecLoadState.LOADED and self.subject is not None: + raise ValueError("only a loaded spec may include a validation subject") + if ( + self.subject is not None + and self.identity is not None + and self.identity != self.subject.spec + ): + raise ValueError("spec load identity must match its loaded subject") + if ( + self.state + in { + SpecLoadState.INVALID, + SpecLoadState.UNSUPPORTED, + SpecLoadState.AMBIGUOUS, + } + and not self.error + ): + raise ValueError("invalid, unsupported, or ambiguous specs need an error") + + @property + def spec(self) -> SpecIdentity | None: + return self.subject.spec if self.subject is not None else self.identity + + def to_dict(self) -> dict[str, Any]: + return { + "state": self.state.value, + "spec": self.subject.to_dict() + if self.subject is not None + else (self.identity.to_dict() if self.identity is not None else None), + "matches": list(self.matches), + "error": self.error, + } + + +class ValidationSpecAdapter(Protocol): + """Trusted structural adapter for one validation source format.""" + + spec_id: str + adapter_id: str + adapter_version: str + execution_model: ExecutionModel + signature_files: tuple[str, ...] + + def detect(self, root: Path) -> bool: + """Detect only by trusted signature paths; never import source code.""" + ... + + def inspect(self, root: Path) -> SpecLoad: + """Return a sanitized subject without executing submitted code.""" + ... + + +class ValidationSpecRegistry: + """Deterministic in-memory registry of installed spec adapters.""" + + def __init__( + self, + adapters: Sequence[ValidationSpecAdapter], + ) -> None: + self._adapters = tuple(adapters) + ids = [adapter.spec_id for adapter in self._adapters] + if any(not _STABLE_ID.fullmatch(spec_id) for spec_id in ids): + raise ValueError("spec adapter IDs must be stable lowercase identifiers") + if len(ids) != len(set(ids)): + raise ValueError("spec adapter IDs must be unique") + for adapter in self._adapters: + AdapterIdentity(adapter.adapter_id, adapter.adapter_version) + if not isinstance(adapter.execution_model, ExecutionModel): + raise ValueError("spec adapters must declare an execution model") + if not adapter.signature_files or any( + not signature + or Path(signature).is_absolute() + or ".." in Path(signature).parts + for signature in adapter.signature_files + ): + raise ValueError( + "spec adapter signatures must be non-empty relative paths" + ) + + @property + def supported_specs(self) -> tuple[str, ...]: + return tuple(adapter.spec_id for adapter in self._adapters) + + def get(self, spec_id: str) -> ValidationSpecAdapter: + for adapter in self._adapters: + if adapter.spec_id == spec_id: + return adapter + supported = ", ".join(self.supported_specs) or "none" + raise ValueError( + f"Unsupported validation spec {spec_id!r}. Supported: {supported}" + ) + + @staticmethod + def _unknown_identity(adapter: ValidationSpecAdapter) -> SpecIdentity: + return SpecIdentity( + spec_id=adapter.spec_id, + spec_version=None, + adapter=AdapterIdentity(adapter.adapter_id, adapter.adapter_version), + execution_model=adapter.execution_model, + ) + + def _inspect( + self, + adapter: ValidationSpecAdapter, + root: Path, + mode: DetectionMode, + ) -> SpecLoad: + try: + loaded = adapter.inspect(root) + except Exception as exc: + return SpecLoad( + state=SpecLoadState.INVALID, + identity=self._unknown_identity(adapter), + error=f"Spec adapter inspection failed ({type(exc).__name__})", + matches=(adapter.spec_id,), + ) + if not isinstance(loaded, SpecLoad): + return SpecLoad( + state=SpecLoadState.INVALID, + identity=self._unknown_identity(adapter), + error="Spec adapter returned an invalid inspection result", + matches=(adapter.spec_id,), + ) + identity = loaded.spec + expected_adapter = AdapterIdentity(adapter.adapter_id, adapter.adapter_version) + if ( + identity is None + or identity.spec_id != adapter.spec_id + or identity.adapter != expected_adapter + or identity.execution_model is not adapter.execution_model + ): + return SpecLoad( + state=SpecLoadState.INVALID, + identity=self._unknown_identity(adapter), + error="Spec adapter returned inconsistent provenance", + matches=(adapter.spec_id,), + ) + if ( + loaded.subject is not None + and loaded.subject.signature_path not in adapter.signature_files + ): + return SpecLoad( + state=SpecLoadState.INVALID, + identity=identity, + error="Spec adapter returned an undeclared signature path", + matches=(adapter.spec_id,), + ) + if loaded.subject is None: + return replace(loaded, matches=(adapter.spec_id,)) + subject = replace( + loaded.subject, + detection_mode=mode, + ) + return replace(loaded, subject=subject, matches=(adapter.spec_id,)) + + def resolve(self, root: str | Path, spec_id: str | None = None) -> SpecLoad: + """Resolve an explicit spec or reject absent/ambiguous auto-detection.""" + path = Path(root) + if spec_id is not None: + return self._inspect(self.get(spec_id), path, DetectionMode.EXPLICIT) + + matches: list[ValidationSpecAdapter] = [] + for adapter in self._adapters: + try: + detected = adapter.detect(path) + except Exception as exc: + return SpecLoad( + state=SpecLoadState.INVALID, + identity=self._unknown_identity(adapter), + error=f"Spec adapter detection failed ({type(exc).__name__})", + matches=(adapter.spec_id,), + ) + if not isinstance(detected, bool): + return SpecLoad( + state=SpecLoadState.INVALID, + identity=self._unknown_identity(adapter), + error="Spec adapter detection must return a boolean", + matches=(adapter.spec_id,), + ) + if detected: + matches.append(adapter) + if not matches: + return SpecLoad(state=SpecLoadState.ABSENT) + if len(matches) > 1: + ids = tuple(sorted(adapter.spec_id for adapter in matches)) + return SpecLoad( + state=SpecLoadState.AMBIGUOUS, + error=f"Multiple validation specs matched: {', '.join(ids)}", + matches=ids, + ) + return self._inspect(matches[0], path, DetectionMode.AUTO) diff --git a/src/openenv/validation/specs/harbor.py b/src/openenv/validation/specs/harbor.py new file mode 100644 index 000000000..0d51814a2 --- /dev/null +++ b/src/openenv/validation/specs/harbor.py @@ -0,0 +1,529 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Harbor v0.5 task.toml adapter for normalized validation requirements.""" + +from __future__ import annotations + +import hashlib +import os +import re +from decimal import Decimal, DecimalException +from pathlib import Path +from typing import Any, Literal + +from pydantic import ( + BaseModel, + ConfigDict, + Field, + field_validator, + model_validator, + ValidationError, +) + +try: + import tomllib +except ModuleNotFoundError: + import tomli as tomllib + +from .base import ( + AdapterIdentity, + ArtifactRequirement, + EnvironmentRequirements, + HealthcheckRequirement, + NetworkMode, + NetworkRequirements, + PackageIdentity, + PhaseRequirements, + RequirementsLoad, + RequirementsProvenance, + RequirementsState, + StepRequirements, + ValidationRequirements, +) + + +_MAX_RESOURCE_MB = 2**31 - 1 +_MAX_RESOURCE_COUNT = 1024 +_MAX_MANIFEST_BYTES = 1024 * 1024 +_MAX_MANIFEST_DEPTH = 64 +_MAX_MANIFEST_NODES = 100_000 + +_HARBOR_ADAPTER = AdapterIdentity("harbor-task-toml", "1") + + +class _StrictModel(BaseModel): + model_config = ConfigDict(extra="forbid", allow_inf_nan=False) + + +class _TaskIdentityModel(_StrictModel): + name: str + description: str = "" + authors: list["_AuthorModel"] = Field(default_factory=list) + keywords: list[str] = Field(default_factory=list) + + @field_validator("name") + @classmethod + def _valid_name(cls, value: str) -> str: + pattern = r"^[A-Za-z0-9_-][A-Za-z0-9._-]*/[A-Za-z0-9_-][A-Za-z0-9._-]*$" + if not re.match(pattern, value) or ".." in value: + raise ValueError("task name must use the org/name format") + return value + + +class _AuthorModel(_StrictModel): + name: str = Field(min_length=1) + email: str | None = None + + +class _AgentModel(_StrictModel): + timeout_sec: float | None = Field(default=None, gt=0) + user: str | int | None = None + + +class _VerifierModel(_StrictModel): + timeout_sec: float = Field(default=600.0, gt=0) + env: dict[str, str] = Field(default_factory=dict) + user: str | int | None = None + + +class _SolutionModel(_StrictModel): + env: dict[str, str] = Field(default_factory=dict) + + +class _HealthcheckModel(_StrictModel): + command: str = Field(min_length=1) + interval_sec: float = Field(default=5.0, gt=0) + timeout_sec: float = Field(default=30.0, gt=0) + start_period_sec: float = Field(default=0.0, ge=0) + start_interval_sec: float = Field(default=5.0, gt=0) + retries: int = Field(default=3, gt=0) + + +class _EnvironmentModel(_StrictModel): + build_timeout_sec: float = Field(default=600.0, gt=0) + docker_image: str | None = None + cpus: int = Field(default=1, gt=0, le=_MAX_RESOURCE_COUNT) + memory_mb: int = Field(default=2048, gt=0, le=_MAX_RESOURCE_MB) + storage_mb: int = Field(default=10240, gt=0, le=_MAX_RESOURCE_MB) + gpus: int = Field(default=0, ge=0, le=_MAX_RESOURCE_COUNT) + gpu_types: list[str] | None = None + allow_internet: bool = True + mcp_servers: list["_MCPServerModel"] = Field(default_factory=list) + env: dict[str, str] = Field(default_factory=dict) + skills_dir: str | None = None + healthcheck: _HealthcheckModel | None = None + workdir: str | None = None + memory: str | None = None + storage: str | None = None + + @field_validator("docker_image") + @classmethod + def _nonempty_image(cls, value: str | None) -> str | None: + if value is not None and not value.strip(): + raise ValueError("docker_image cannot be empty") + return value + + +class _ArtifactModel(_StrictModel): + source: str = Field(min_length=1) + destination: str | None = None + exclude: list[str] = Field(default_factory=list) + + +class _MCPServerModel(_StrictModel): + name: str = Field(min_length=1) + transport: str = "sse" + url: str | None = None + command: str | None = None + args: list[str] = Field(default_factory=list) + + @model_validator(mode="after") + def _validate_transport_fields(self) -> "_MCPServerModel": + if self.transport in {"sse", "streamable-http"} and not self.url: + raise ValueError(f"url is required for transport {self.transport}") + if self.transport == "stdio" and not self.command: + raise ValueError("command is required for stdio transport") + return self + + +class _StepModel(_StrictModel): + name: str = Field(min_length=1) + agent: _AgentModel = Field(default_factory=_AgentModel) + verifier: _VerifierModel = Field(default_factory=_VerifierModel) + min_reward: float | dict[str, float] | None = None + healthcheck: _HealthcheckModel | None = None + artifacts: list[str | _ArtifactModel] = Field(default_factory=list) + + +class _TaskManifestModel(_StrictModel): + schema_version: str = "1.1" + task: _TaskIdentityModel | None = None + metadata: dict[str, Any] = Field(default_factory=dict) + verifier: _VerifierModel = Field(default_factory=_VerifierModel) + agent: _AgentModel = Field(default_factory=_AgentModel) + solution: _SolutionModel = Field(default_factory=_SolutionModel) + environment: _EnvironmentModel = Field(default_factory=_EnvironmentModel) + source: str | None = None + artifacts: list[str | _ArtifactModel] = Field(default_factory=list) + steps: list[_StepModel] | None = None + multi_step_reward_strategy: Literal["mean", "final"] | None = None + + +def _size_to_mb(value: str) -> int: + normalized = value.strip().upper() + factors = {"G": Decimal(1024), "M": Decimal(1), "K": Decimal(1) / 1024} + suffix = normalized[-1:] if normalized else "" + if suffix not in factors: + raise ValueError("size must use K, M, or G units") + try: + amount = Decimal(normalized[:-1]) + size_mb = amount * factors[suffix] + if not size_mb.is_finite() or size_mb <= 0 or size_mb > _MAX_RESOURCE_MB: + raise ValueError("size is outside the supported resource range") + return max(1, int(size_mb)) + except DecimalException as exc: + raise ValueError("size must contain a finite supported number") from exc + + +def _normalize_legacy_resources(raw: dict[str, Any]) -> dict[str, Any]: + environment = raw.get("environment") + if not isinstance(environment, dict): + return raw + normalized = dict(raw) + normalized_environment = dict(environment) + for legacy_name, current_name in ( + ("memory", "memory_mb"), + ("storage", "storage_mb"), + ): + legacy_value = normalized_environment.get(legacy_name) + if legacy_value is None: + continue + if not isinstance(legacy_value, str): + continue + converted = _size_to_mb(legacy_value) + if ( + current_name in normalized_environment + and normalized_environment[current_name] != converted + ): + raise ValueError(f"conflicting {legacy_name} and {current_name} fields") + normalized_environment[current_name] = converted + normalized["environment"] = normalized_environment + return normalized + + +def _validate_manifest_structure(raw: dict[str, Any]) -> None: + stack: list[tuple[Any, int]] = [(raw, 0)] + visited = 0 + while stack: + value, depth = stack.pop() + visited += 1 + if visited > _MAX_MANIFEST_NODES: + raise ValueError("task.toml exceeds the maximum structural size") + if depth > _MAX_MANIFEST_DEPTH: + raise ValueError("task.toml exceeds the maximum nesting depth") + if isinstance(value, dict): + stack.extend((item, depth + 1) for item in value.values()) + elif isinstance(value, list): + stack.extend((item, depth + 1) for item in value) + + +def _artifact(model: str | _ArtifactModel) -> ArtifactRequirement: + if isinstance(model, str): + return ArtifactRequirement(source=model) + return ArtifactRequirement( + source=model.source, + destination=model.destination, + exclude=tuple(model.exclude), + ) + + +def _phase( + model: _AgentModel | _VerifierModel, +) -> PhaseRequirements: + env = model.env if isinstance(model, _VerifierModel) else {} + return PhaseRequirements( + timeout_s=model.timeout_sec, + user=model.user, + env_names=tuple(sorted(env)), + ) + + +def _convert( + model: _TaskManifestModel, present_fields: tuple[str, ...] +) -> ValidationRequirements: + healthcheck_model = model.environment.healthcheck + healthcheck = ( + HealthcheckRequirement( + command=healthcheck_model.command, + interval_s=healthcheck_model.interval_sec, + timeout_s=healthcheck_model.timeout_sec, + start_period_s=healthcheck_model.start_period_sec, + start_interval_s=healthcheck_model.start_interval_sec, + retries=healthcheck_model.retries, + ) + if healthcheck_model is not None + else None + ) + environment = EnvironmentRequirements( + build_timeout_s=model.environment.build_timeout_sec, + container_image=model.environment.docker_image, + cpus=model.environment.cpus, + memory_mb=model.environment.memory_mb, + storage_mb=model.environment.storage_mb, + gpus=model.environment.gpus, + gpu_types=tuple(model.environment.gpu_types or ()), + network=NetworkRequirements( + mode=( + NetworkMode.ALLOW_ALL + if model.environment.allow_internet + else NetworkMode.DENY_ALL + ) + ), + env_names=tuple(sorted(model.environment.env)), + skills_dir=model.environment.skills_dir, + workdir=model.environment.workdir, + mcp_server_count=len(model.environment.mcp_servers), + healthcheck=healthcheck, + ) + steps = tuple( + StepRequirements( + name=step.name, + agent=_phase(step.agent), + verifier=_phase(step.verifier), + min_reward=step.min_reward, + healthcheck=( + HealthcheckRequirement( + command=step.healthcheck.command, + interval_s=step.healthcheck.interval_sec, + timeout_s=step.healthcheck.timeout_sec, + start_period_s=step.healthcheck.start_period_sec, + start_interval_s=step.healthcheck.start_interval_sec, + retries=step.healthcheck.retries, + ) + if step.healthcheck is not None + else None + ), + artifacts=tuple(_artifact(artifact) for artifact in step.artifacts), + ) + for step in model.steps or () + ) + task = model.task + return ValidationRequirements( + identity=PackageIdentity( + name=task.name if task is not None else None, + description=task.description if task is not None else "", + authors=( + tuple(author.model_dump() for author in task.authors) + if task is not None + else () + ), + keywords=tuple(task.keywords) if task is not None else (), + ), + agent=_phase(model.agent), + verifier=_phase(model.verifier), + environment=environment, + artifacts=tuple(_artifact(artifact) for artifact in model.artifacts), + steps=steps, + metadata_keys=tuple(sorted(model.metadata)), + solution_env_names=tuple(sorted(model.solution.env)), + multi_step_reward_strategy=model.multi_step_reward_strategy, + source_declared=model.source is not None, + present_fields=present_fields, + ) + + +def _manifest_path(root: Path) -> Path: + direct = root / "task.toml" + if direct.exists() or direct.is_symlink() or root.name != "environment": + return direct + return root.parent / "task.toml" + + +def _contained_regular_file(root: Path, candidate: Path) -> bool: + try: + relative = candidate.relative_to(root) + except ValueError: + return False + current = root + for part in relative.parts: + current /= part + if current.is_symlink(): + return False + try: + return bool( + candidate.is_file() + and candidate.resolve(strict=True).is_relative_to(root.resolve(strict=True)) + ) + except OSError: + return False + + +def find_harbor_verifier_script(root: str | Path) -> Path | None: + """Find tests/test.sh for an OpenEnv root or Harbor environment directory.""" + path = Path(root) + direct = path / "tests" / "test.sh" + if _contained_regular_file(path, direct): + return direct + if path.name == "environment" and (path.parent / "task.toml").is_file(): + sibling = path.parent / "tests" / "test.sh" + if _contained_regular_file(path.parent, sibling): + return sibling + return None + + +def _validation_error(exc: ValidationError) -> str: + issues = [] + for error in exc.errors(include_input=False, include_url=False): + location = ".".join(str(part) for part in error.get("loc", ())) or "task.toml" + issues.append(f"{location}: {error.get('msg', error.get('type', 'invalid'))}") + return "; ".join(issues) + + +def _provenance( + root: Path, + path: Path, + version: str | None, + document_digest: str | None = None, +) -> RequirementsProvenance: + return RequirementsProvenance( + source_id="harbor", + source_version=version, + adapter=_HARBOR_ADAPTER, + path=os.path.relpath(path, start=root), + document_digest=document_digest, + ) + + +def load_harbor_requirements(root: str | Path) -> RequirementsLoad: + """Load Harbor v0.5/schema 1.1 into normalized validation requirements.""" + root_path = Path(root) + path = _manifest_path(root_path) + if not path.exists() and not path.is_symlink(): + return RequirementsLoad(state=RequirementsState.ABSENT) + if path.is_symlink(): + return RequirementsLoad( + state=RequirementsState.INVALID, + provenance=_provenance(root_path, path, None), + error="task.toml must be a regular file, not a symbolic link", + ) + if not path.is_file(): + return RequirementsLoad( + state=RequirementsState.INVALID, + provenance=_provenance(root_path, path, None), + error="task.toml must be a regular file", + ) + + document_digest: str | None = None + try: + with path.open("rb") as manifest_file: + manifest_bytes = manifest_file.read(_MAX_MANIFEST_BYTES + 1) + if len(manifest_bytes) > _MAX_MANIFEST_BYTES: + return RequirementsLoad( + state=RequirementsState.INVALID, + provenance=_provenance(root_path, path, None), + error="task.toml exceeds the 1 MiB validation limit", + ) + document_digest = f"sha256:{hashlib.sha256(manifest_bytes).hexdigest()}" + raw = tomllib.loads(manifest_bytes.decode("utf-8")) + except (OSError, RecursionError, UnicodeError, tomllib.TOMLDecodeError) as exc: + return RequirementsLoad( + state=RequirementsState.INVALID, + provenance=_provenance(root_path, path, None, document_digest), + error=f"Unable to parse task.toml ({type(exc).__name__})", + ) + + if not isinstance(raw, dict): + return RequirementsLoad( + state=RequirementsState.INVALID, + provenance=_provenance(root_path, path, None, document_digest), + error="task.toml must contain a TOML table", + ) + + try: + _validate_manifest_structure(raw) + except (RecursionError, ValueError) as exc: + return RequirementsLoad( + state=RequirementsState.INVALID, + provenance=_provenance(root_path, path, None, document_digest), + error=f"Invalid task.toml structure ({type(exc).__name__})", + ) + + normalized = dict(raw) + legacy_version = normalized.pop("version", None) + if legacy_version is not None: + current_version = normalized.get("schema_version") + if current_version is not None and current_version != legacy_version: + return RequirementsLoad( + state=RequirementsState.INVALID, + provenance=_provenance( + root_path, path, str(current_version), document_digest + ), + error="task.toml has conflicting version and schema_version fields", + ) + normalized["schema_version"] = legacy_version + + try: + normalized = _normalize_legacy_resources(normalized) + model = _TaskManifestModel.model_validate(normalized) + except ValidationError as exc: + return RequirementsLoad( + state=RequirementsState.INVALID, + provenance=_provenance( + root_path, + path, + str(normalized.get("schema_version", "1.1")), + document_digest, + ), + error=_validation_error(exc), + ) + except ( + DecimalException, + OverflowError, + RecursionError, + TypeError, + ValueError, + ) as exc: + return RequirementsLoad( + state=RequirementsState.INVALID, + provenance=_provenance( + root_path, + path, + str(normalized.get("schema_version", "1.1")), + document_digest, + ), + error=f"Invalid task.toml configuration ({type(exc).__name__})", + ) + + provenance = _provenance(root_path, path, model.schema_version, document_digest) + if model.schema_version != "1.1": + return RequirementsLoad( + state=RequirementsState.UNSUPPORTED, + provenance=provenance, + error=( + "Unsupported Harbor schema version " + f"{model.schema_version!r}; expected '1.1'" + ), + ) + present_fields = tuple(sorted(str(key) for key in raw)) + requirements = _convert(model, present_fields) + artifacts = list(requirements.artifacts) + for step in requirements.steps: + artifacts.extend(step.artifacts) + if any(not artifact.source.startswith("/") for artifact in artifacts): + return RequirementsLoad( + state=RequirementsState.INVALID, + provenance=provenance, + error="Harbor artifact sources must use absolute container paths", + ) + environment = requirements.environment + if environment.gpus == 0 and environment.gpu_types: + return RequirementsLoad( + state=RequirementsState.INVALID, + provenance=provenance, + error="Harbor gpu_types requires a positive gpus value", + ) + return RequirementsLoad( + state=RequirementsState.LOADED, + provenance=provenance, + requirements=requirements, + ) diff --git a/src/openenv/validation/specs/openenv.py b/src/openenv/validation/specs/openenv.py new file mode 100644 index 000000000..af5730c9a --- /dev/null +++ b/src/openenv/validation/specs/openenv.py @@ -0,0 +1,140 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Trusted adapter for served OpenEnv environment source trees.""" + +from __future__ import annotations + +import hashlib +import os +from pathlib import Path +from typing import Any + +import yaml + +from .base import ( + AdapterIdentity, + DetectionMode, + ExecutionModel, + SpecIdentity, + SpecLoad, + SpecLoadState, + ValidationSubject, +) +from .harbor import find_harbor_verifier_script, load_harbor_requirements + + +_MAX_SPEC_BYTES = 1024 * 1024 +_MAX_VERIFIER_BYTES = 16 * 1024 * 1024 + + +def _verifier_digest(path: Path) -> str: + digest = hashlib.sha256() + size = 0 + with path.open("rb") as verifier_file: + while chunk := verifier_file.read(1024 * 1024): + size += len(chunk) + if size > _MAX_VERIFIER_BYTES: + raise ValueError( + "declared verifier exceeds the 16 MiB validation limit" + ) + digest.update(chunk) + return f"sha256:{digest.hexdigest()}" + + +class OpenEnvSpecAdapter: + """Inspect `openenv.yaml` without importing submitted environment code.""" + + spec_id = "openenv" + adapter_id = "openenv-yaml" + adapter_version = "1" + execution_model = ExecutionModel.SERVED + signature_files = ("openenv.yaml",) + + @property + def _adapter(self) -> AdapterIdentity: + return AdapterIdentity(self.adapter_id, self.adapter_version) + + def _identity(self, version: str | None) -> SpecIdentity: + return SpecIdentity( + spec_id=self.spec_id, + spec_version=version, + adapter=self._adapter, + execution_model=self.execution_model, + ) + + def detect(self, root: Path) -> bool: + """Detect the OpenEnv signature at the requested root only.""" + signature = root / "openenv.yaml" + return signature.exists() or signature.is_symlink() + + def inspect(self, root: Path) -> SpecLoad: + """Safely inspect OpenEnv identity and normalized auxiliary requirements.""" + signature = root / "openenv.yaml" + unknown_identity = self._identity(None) + if not signature.exists() and not signature.is_symlink(): + return SpecLoad(state=SpecLoadState.ABSENT, identity=unknown_identity) + if signature.is_symlink() or not signature.is_file(): + return SpecLoad( + state=SpecLoadState.INVALID, + identity=unknown_identity, + error="openenv.yaml must be a regular file, not a symbolic link", + ) + + try: + with signature.open("rb") as spec_file: + document = spec_file.read(_MAX_SPEC_BYTES + 1) + if len(document) > _MAX_SPEC_BYTES: + return SpecLoad( + state=SpecLoadState.INVALID, + identity=unknown_identity, + error="openenv.yaml exceeds the 1 MiB validation limit", + ) + parsed: Any = yaml.safe_load(document) + except (OSError, RecursionError, UnicodeError, yaml.YAMLError) as exc: + return SpecLoad( + state=SpecLoadState.INVALID, + identity=unknown_identity, + error=f"Unable to parse openenv.yaml ({type(exc).__name__})", + ) + if not isinstance(parsed, dict): + return SpecLoad( + state=SpecLoadState.INVALID, + identity=unknown_identity, + error="openenv.yaml must contain a mapping", + ) + + raw_version = parsed.get("spec_version") + version = ( + str(raw_version) + if isinstance(raw_version, (int, str)) and not isinstance(raw_version, bool) + else None + ) + identity = self._identity(version) + if type(raw_version) is not int or raw_version != 1: + return SpecLoad( + state=SpecLoadState.UNSUPPORTED, + identity=identity, + error=f"Unsupported OpenEnv spec version {raw_version!r}; expected 1", + ) + + requirements = load_harbor_requirements(root) + verifier_script = find_harbor_verifier_script(root) + subject = ValidationSubject( + spec=identity, + signature_path="openenv.yaml", + detection_mode=DetectionMode.AUTO, + requirements=requirements, + verifier_script=verifier_script, + verifier_path=( + os.path.relpath(verifier_script, start=root) + if verifier_script is not None + else None + ), + verifier_digest=( + _verifier_digest(verifier_script) + if verifier_script is not None + else None + ), + document_digest=f"sha256:{hashlib.sha256(document).hexdigest()}", + ) + return SpecLoad(state=SpecLoadState.LOADED, subject=subject) diff --git a/tests/test_validation/__init__.py b/tests/test_validation/__init__.py new file mode 100644 index 000000000..cc427103b --- /dev/null +++ b/tests/test_validation/__init__.py @@ -0,0 +1,3 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Tests for RFC 008 validation architecture.""" diff --git a/tests/test_validation/_helpers.py b/tests/test_validation/_helpers.py new file mode 100644 index 000000000..de1b53c2f --- /dev/null +++ b/tests/test_validation/_helpers.py @@ -0,0 +1,22 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Shared fixtures for validation architecture tests.""" + +from pathlib import Path + + +def write_harbor_task(task_root: Path) -> Path: + """Create a multi-step Harbor task and return its environment directory.""" + environment = task_root / "environment" + tests_dir = task_root / "tests" + environment.mkdir(parents=True) + tests_dir.mkdir() + (tests_dir / "test.sh").write_text("#!/bin/sh\nexit 0\n") + (task_root / "task.toml").write_text( + 'schema_version = "1.1"\n' + "[[steps]]\n" + 'name = "grade"\n' + "min_reward = { correctness = 0.8, style = 0.5 }\n" + 'artifacts = ["/workspace/result.json"]\n' + ) + return environment diff --git a/tests/test_validation/test_harbor_spec.py b/tests/test_validation/test_harbor_spec.py new file mode 100644 index 000000000..2822bd236 --- /dev/null +++ b/tests/test_validation/test_harbor_spec.py @@ -0,0 +1,262 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Tests for adapting Harbor task manifests into validation requirements.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +from openenv.validation.specs import ( + find_harbor_verifier_script, + load_harbor_requirements, + NetworkMode, + RequirementsState, +) + +from ._helpers import write_harbor_task + + +def test_task_manifest_loader_normalizes_harbor_1_1_envelope( + tmp_path: Path, +) -> None: + task_dir = tmp_path / "task" + task_dir.mkdir() + (task_dir / "task.toml").write_text( + 'schema_version = "1.1"\n' + 'artifacts = ["/workspace/result.json", ' + '{ source = "/workspace/logs", destination = "/logs", exclude = ["*.tmp"] }]\n' + "\n" + "[task]\n" + 'name = "openenv/example"\n' + 'description = "Example environment"\n' + "\n" + "[agent]\n" + "timeout_sec = 120.0\n" + "\n" + "[verifier]\n" + "timeout_sec = 45.0\n" + 'env = { VERIFY_TOKEN = "super-secret" }\n' + 'user = "grader"\n' + "\n" + "[environment]\n" + "build_timeout_sec = 300.0\n" + 'docker_image = "ghcr.io/openenv/example@sha256:abc"\n' + "cpus = 4\n" + "memory_mb = 8192\n" + "storage_mb = 20480\n" + "gpus = 1\n" + 'gpu_types = ["A100", "H100"]\n' + "allow_internet = false\n" + 'env = { API_TOKEN = "do-not-report" }\n' + "\n" + "[environment.healthcheck]\n" + 'command = "curl -f http://127.0.0.1:8000/health"\n' + "interval_sec = 2.0\n" + "timeout_sec = 3.0\n" + "start_period_sec = 4.0\n" + "start_interval_sec = 1.0\n" + "retries = 5\n" + ) + + loaded = load_harbor_requirements(task_dir) + + assert loaded.state is RequirementsState.LOADED + assert loaded.error is None + assert loaded.provenance is not None + assert loaded.provenance.source_id == "harbor" + assert loaded.provenance.source_version == "1.1" + assert loaded.provenance.document_digest is not None + assert loaded.provenance.document_digest.startswith("sha256:") + assert loaded.requirements is not None + requirements = loaded.requirements + assert requirements.identity.name == "openenv/example" + assert requirements.agent.timeout_s == 120.0 + assert requirements.verifier.timeout_s == 45.0 + assert requirements.environment.cpus == 4 + assert requirements.environment.memory_mb == 8192 + assert requirements.environment.storage_mb == 20480 + assert requirements.environment.gpus == 1 + assert requirements.environment.gpu_types == ("A100", "H100") + assert requirements.environment.network.mode is NetworkMode.DENY_ALL + assert requirements.environment.healthcheck is not None + assert requirements.environment.healthcheck.retries == 5 + assert requirements.artifacts[0].source == "/workspace/result.json" + assert requirements.artifacts[1].destination == "/logs" + assert requirements.artifacts[1].exclude == ("*.tmp",) + + evidence = json.dumps(loaded.to_evidence(), sort_keys=True) + assert "API_TOKEN" in evidence + assert "VERIFY_TOKEN" in evidence + assert "do-not-report" not in evidence + assert "super-secret" not in evidence + + +def test_task_manifest_loader_distinguishes_absent_and_invalid( + tmp_path: Path, +) -> None: + absent = load_harbor_requirements(tmp_path) + assert absent.state is RequirementsState.ABSENT + assert absent.requirements is None + + (tmp_path / "task.toml").write_text("[environment\ncpus = 2\n") + invalid = load_harbor_requirements(tmp_path) + assert invalid.state is RequirementsState.INVALID + assert invalid.requirements is None + assert invalid.error + + +def test_task_manifest_legacy_version_alias_is_supported(tmp_path: Path) -> None: + (tmp_path / "task.toml").write_text( + 'version = "1.1"\n[environment]\nallow_internet = true\n' + ) + + loaded = load_harbor_requirements(tmp_path) + + assert loaded.state is RequirementsState.LOADED + assert loaded.requirements is not None + assert loaded.provenance is not None + assert loaded.provenance.source_version == "1.1" + assert loaded.requirements.environment.network.mode is NetworkMode.ALLOW_ALL + + +def test_task_manifest_unknown_fields_are_structured_invalid(tmp_path: Path) -> None: + (tmp_path / "task.toml").write_text( + 'schema_version = "1.1"\n[environment]\ncpus = 2\ninvented = true\n' + ) + + loaded = load_harbor_requirements(tmp_path) + + assert loaded.state is RequirementsState.INVALID + assert loaded.error is not None + assert "invented" in loaded.error + + +def test_task_manifest_reports_unsupported_schema(tmp_path: Path) -> None: + (tmp_path / "task.toml").write_text('schema_version = "2.0"\n') + + loaded = load_harbor_requirements(tmp_path) + + assert loaded.state is RequirementsState.UNSUPPORTED + assert loaded.provenance is not None + assert loaded.provenance.source_version == "2.0" + assert loaded.error is not None + + +@pytest.mark.parametrize( + "body, expected", + [ + ('artifacts = ["relative/result.json"]\n', "absolute container paths"), + ( + '[environment]\ngpus = 0\ngpu_types = ["A100"]\n', + "positive gpus", + ), + ], +) +def test_harbor_specific_invariants_stay_in_adapter( + tmp_path: Path, body: str, expected: str +) -> None: + (tmp_path / "task.toml").write_text(f'schema_version = "1.1"\n{body}') + + loaded = load_harbor_requirements(tmp_path) + + assert loaded.state is RequirementsState.INVALID + assert loaded.error is not None + assert expected in loaded.error + + +@pytest.mark.parametrize( + "memory", + ["1e999G", "sNaNG", "1e9999999999999999G"], +) +def test_task_manifest_rejects_nonfinite_legacy_resources( + tmp_path: Path, memory: str +) -> None: + (tmp_path / "task.toml").write_text( + f'schema_version = "1.1"\n[environment]\nmemory = "{memory}"\n' + ) + + loaded = load_harbor_requirements(tmp_path) + + assert loaded.state is RequirementsState.INVALID + assert loaded.error is not None + + +def test_task_manifest_rejects_excessive_nesting(tmp_path: Path) -> None: + nested = "0" + for _ in range(500): + nested = f"[{nested}]" + (tmp_path / "task.toml").write_text( + f'schema_version = "1.1"\nmetadata = {{ nested = {nested} }}\n' + ) + + loaded = load_harbor_requirements(tmp_path) + + assert loaded.state is RequirementsState.INVALID + assert loaded.error is not None + + +def test_task_manifest_rejects_symbolic_link(tmp_path: Path) -> None: + source = tmp_path / "submitted.toml" + source.write_text('schema_version = "1.1"\n') + try: + (tmp_path / "task.toml").symlink_to(source) + except OSError: + pytest.skip("symbolic links are unavailable on this platform") + + loaded = load_harbor_requirements(tmp_path) + + assert loaded.state is RequirementsState.INVALID + assert loaded.error is not None + assert "symbolic link" in loaded.error + + +def test_task_manifest_rejects_dangling_link_and_directory(tmp_path: Path) -> None: + try: + (tmp_path / "task.toml").symlink_to(tmp_path / "missing.toml") + except OSError: + pytest.skip("symbolic links are unavailable on this platform") + + dangling = load_harbor_requirements(tmp_path) + assert dangling.state is RequirementsState.INVALID + + (tmp_path / "task.toml").unlink() + (tmp_path / "task.toml").mkdir() + directory = load_harbor_requirements(tmp_path) + assert directory.state is RequirementsState.INVALID + + +def test_verifier_discovery_rejects_symlinked_tests_directory(tmp_path: Path) -> None: + outside = tmp_path / "outside" + outside.mkdir() + (outside / "test.sh").write_text("#!/bin/sh\n") + environment = tmp_path / "environment" + environment.mkdir() + (tmp_path / "task.toml").write_text('schema_version = "1.1"\n') + try: + (tmp_path / "tests").symlink_to(outside, target_is_directory=True) + except OSError: + pytest.skip("symbolic links are unavailable on this platform") + + assert find_harbor_verifier_script(environment) is None + + +def test_harbor_step_reward_map_and_sibling_verifier_are_loaded( + tmp_path: Path, +) -> None: + task_root = tmp_path / "harbor-task" + environment = write_harbor_task(task_root) + + loaded = load_harbor_requirements(environment) + + assert loaded.state is RequirementsState.LOADED + assert loaded.requirements is not None + assert loaded.requirements.steps[0].min_reward == { + "correctness": 0.8, + "style": 0.5, + } + assert loaded.requirements.steps[0].artifacts[0].source == ( + "/workspace/result.json" + ) + assert find_harbor_verifier_script(environment) == (task_root / "tests" / "test.sh") diff --git a/tests/test_validation/test_spec_registry.py b/tests/test_validation/test_spec_registry.py new file mode 100644 index 000000000..70369c239 --- /dev/null +++ b/tests/test_validation/test_spec_registry.py @@ -0,0 +1,216 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Tests for spec detection and the normalized subject boundary.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +from openenv.validation.specs import ( + AdapterIdentity, + DetectionMode, + ExecutionModel, + NetworkMode, + RequirementsLoad, + RequirementsState, + SpecIdentity, + SpecLoad, + SpecLoadState, + ValidationRequirements, + ValidationSpecRegistry, + ValidationSubject, +) + + +class _FakeSpecAdapter: + adapter_id = "fake-adapter" + adapter_version = "7" + execution_model = ExecutionModel.ONE_SHOT + signature_files = ("task.fake",) + + def __init__(self, spec_id: str = "fake") -> None: + self.spec_id = spec_id + + def detect(self, root: Path) -> bool: + return (root / "task.fake").exists() + + def inspect(self, root: Path) -> SpecLoad: + identity = SpecIdentity( + spec_id=self.spec_id, + spec_version="2", + adapter=AdapterIdentity(self.adapter_id, self.adapter_version), + execution_model=self.execution_model, + ) + if not self.detect(root): + return SpecLoad(state=SpecLoadState.ABSENT, identity=identity) + subject = ValidationSubject( + spec=identity, + signature_path="task.fake", + detection_mode=DetectionMode.AUTO, + requirements=RequirementsLoad( + state=RequirementsState.LOADED, + requirements=ValidationRequirements(), + ), + ) + return SpecLoad(state=SpecLoadState.LOADED, subject=subject) + + +def _write_openenv_spec(root: Path) -> None: + root.mkdir(exist_ok=True) + (root / "openenv.yaml").write_text( + "spec_version: 1\n" + "name: example\n" + "runtime: fastapi\n" + "app: server.app:app\n" + "port: 8000\n" + ) + + +def test_default_registry_detects_openenv_and_adapts_harbor_requirements( + tmp_path: Path, +) -> None: + from openenv.validation.specs import DEFAULT_SPEC_REGISTRY + + _write_openenv_spec(tmp_path) + (tmp_path / "task.toml").write_text( + 'schema_version = "1.1"\n[environment]\nallow_internet = false\n' + ) + + loaded = DEFAULT_SPEC_REGISTRY.resolve(tmp_path) + + assert loaded.state is SpecLoadState.LOADED + assert loaded.subject is not None + assert loaded.subject.spec.spec_id == "openenv" + assert loaded.subject.spec.execution_model is ExecutionModel.SERVED + assert loaded.subject.detection_mode is DetectionMode.AUTO + assert loaded.subject.requirements.state is RequirementsState.LOADED + requirements = loaded.subject.requirements.requirements + assert requirements is not None + assert requirements.environment.network.mode is NetworkMode.DENY_ALL + assert loaded.subject.document_digest is not None + + +def test_registry_supports_explicit_selection_and_structured_absence( + tmp_path: Path, +) -> None: + registry = ValidationSpecRegistry((_FakeSpecAdapter(),)) + + absent = registry.resolve(tmp_path) + explicit_absent = registry.resolve(tmp_path, spec_id="fake") + + assert absent.state is SpecLoadState.ABSENT + assert absent.matches == () + assert explicit_absent.state is SpecLoadState.ABSENT + assert explicit_absent.spec is not None + assert explicit_absent.spec.spec_id == "fake" + + (tmp_path / "task.fake").write_text("signature") + loaded = registry.resolve(tmp_path, spec_id="fake") + assert loaded.state is SpecLoadState.LOADED + assert loaded.subject is not None + assert loaded.subject.detection_mode is DetectionMode.EXPLICIT + + +def test_registry_rejects_duplicate_and_ambiguous_specs(tmp_path: Path) -> None: + with pytest.raises(ValueError, match="unique"): + ValidationSpecRegistry((_FakeSpecAdapter(), _FakeSpecAdapter())) + + first = _FakeSpecAdapter("first") + second = _FakeSpecAdapter("second") + registry = ValidationSpecRegistry((first, second)) + (tmp_path / "task.fake").write_text("signature") + + loaded = registry.resolve(tmp_path) + + assert loaded.state is SpecLoadState.AMBIGUOUS + assert loaded.matches == ("first", "second") + assert loaded.error == "Multiple validation specs matched: first, second" + + +def test_malformed_openenv_signature_is_not_treated_as_absent(tmp_path: Path) -> None: + from openenv.validation.specs import DEFAULT_SPEC_REGISTRY + + (tmp_path / "openenv.yaml").write_text("spec_version: [") + + loaded = DEFAULT_SPEC_REGISTRY.resolve(tmp_path) + + assert loaded.state is SpecLoadState.INVALID + assert loaded.matches == ("openenv",) + assert loaded.error is not None + assert loaded.spec is not None + assert loaded.spec.adapter == AdapterIdentity("openenv-yaml", "1") + assert loaded.spec.execution_model is ExecutionModel.SERVED + + +def test_registry_generated_failures_retain_adapter_identity(tmp_path: Path) -> None: + class _ExplodingAdapter(_FakeSpecAdapter): + def inspect(self, root: Path) -> SpecLoad: + raise RuntimeError("boom") + + (tmp_path / "task.fake").write_text("signature") + loaded = ValidationSpecRegistry((_ExplodingAdapter(),)).resolve(tmp_path) + + assert loaded.state is SpecLoadState.INVALID + assert loaded.spec == SpecIdentity( + spec_id="fake", + spec_version=None, + adapter=AdapterIdentity("fake-adapter", "7"), + execution_model=ExecutionModel.ONE_SHOT, + ) + + +def test_spec_load_rejects_conflicting_subject_identity() -> None: + identity = SpecIdentity( + spec_id="fake", + spec_version="2", + adapter=AdapterIdentity("fake-adapter", "7"), + execution_model=ExecutionModel.ONE_SHOT, + ) + subject = ValidationSubject( + spec=identity, + signature_path="task.fake", + detection_mode=DetectionMode.AUTO, + requirements=RequirementsLoad(state=RequirementsState.ABSENT), + ) + conflicting = SpecIdentity( + spec_id="other", + spec_version="2", + adapter=AdapterIdentity("fake-adapter", "7"), + execution_model=ExecutionModel.ONE_SHOT, + ) + + with pytest.raises(ValueError, match="identity"): + SpecLoad( + state=SpecLoadState.LOADED, + subject=subject, + identity=conflicting, + ) + + +@pytest.mark.parametrize("version", ["true", "1.0"]) +def test_openenv_spec_version_must_be_the_strict_integer_one( + tmp_path: Path, version: str +) -> None: + from openenv.validation.specs import DEFAULT_SPEC_REGISTRY + + (tmp_path / "openenv.yaml").write_text(f"spec_version: {version}\n") + + loaded = DEFAULT_SPEC_REGISTRY.resolve(tmp_path) + + assert loaded.state is SpecLoadState.UNSUPPORTED + assert loaded.error is not None + + +def test_dangling_openenv_signature_is_invalid_not_absent(tmp_path: Path) -> None: + from openenv.validation.specs import DEFAULT_SPEC_REGISTRY + + try: + (tmp_path / "openenv.yaml").symlink_to(tmp_path / "missing.yaml") + except OSError: + pytest.skip("symbolic links are unavailable on this platform") + + loaded = DEFAULT_SPEC_REGISTRY.resolve(tmp_path) + + assert loaded.state is SpecLoadState.INVALID + assert "symbolic link" in (loaded.error or "")