From f4d7391384d69388f039e870ab90cf96b05b217d Mon Sep 17 00:00:00 2001 From: Zach Wentz <4832+zkwentz@users.noreply.github.com> Date: Sun, 12 Jul 2026 07:47:38 -0400 Subject: [PATCH] feat(validation): add spec-aware policy planner --- src/openenv/validation/__init__.py | 60 +- src/openenv/validation/checks.py | 772 ++++++++++++++++++++ src/openenv/validation/models.py | 132 +++- src/openenv/validation/planner.py | 284 +++++++ src/openenv/validation/serialization.py | 215 ++++++ src/openenv/validation/source_inspection.py | 106 +++ tests/test_validation/_helpers.py | 33 +- tests/test_validation/test_planner.py | 308 ++++++++ 8 files changed, 1854 insertions(+), 56 deletions(-) create mode 100644 src/openenv/validation/checks.py create mode 100644 src/openenv/validation/planner.py create mode 100644 src/openenv/validation/serialization.py create mode 100644 src/openenv/validation/source_inspection.py create mode 100644 tests/test_validation/test_planner.py diff --git a/src/openenv/validation/__init__.py b/src/openenv/validation/__init__.py index c920f6abc..9e3b16e53 100644 --- a/src/openenv/validation/__init__.py +++ b/src/openenv/validation/__init__.py @@ -9,73 +9,35 @@ ValidationCheck, ValidationContext, ValidationPlan, + ValidationPolicy, ValidationProfile, ValidationReport, + ValidationRequirementBinding, 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, +from .planner import ( + build_validation_plan, + OPENENV_VALIDATION_POLICY, + VALIDATION_POLICY_VERSION, ) __all__ = [ - "AdapterIdentity", - "ArtifactRequirement", "CheckOutcome", - "DEFAULT_SPEC_REGISTRY", - "DetectionMode", - "EnvironmentRequirements", - "ExecutionModel", - "HealthcheckRequirement", - "NetworkMode", - "NetworkRequirements", - "OpenEnvSpecAdapter", - "PackageIdentity", - "PhaseRequirements", - "RequirementsLoad", - "RequirementsProvenance", - "RequirementsState", + "OPENENV_VALIDATION_POLICY", "RunnerCapabilities", - "SpecIdentity", - "SpecLoad", - "SpecLoadState", - "StepRequirements", + "VALIDATION_POLICY_VERSION", "ValidationCapability", "ValidationCheck", "ValidationContext", "ValidationPlan", + "ValidationPolicy", "ValidationProfile", "ValidationReport", + "ValidationRequirementBinding", "ValidationResult", "ValidationSeverity", - "ValidationSpecAdapter", - "ValidationSpecRegistry", "ValidationStatus", - "ValidationSubject", - "ValidationRequirements", - "load_harbor_requirements", + "build_validation_plan", ] diff --git a/src/openenv/validation/checks.py b/src/openenv/validation/checks.py new file mode 100644 index 000000000..d177ad7dd --- /dev/null +++ b/src/openenv/validation/checks.py @@ -0,0 +1,772 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Built-in validation check registry for RFC 008 policy version 1.""" + +from __future__ import annotations + +import ast +import re +from functools import lru_cache +from pathlib import Path +from typing import Any, Callable + +import yaml + +from .models import ( + CheckOutcome, + ValidationCapability, + ValidationCheck, + ValidationContext, + ValidationProfile, + ValidationRequirementBinding, + ValidationSeverity, + ValidationStatus, +) +from .source_inspection import ( + _dockerfile_installs_openenv_runtime, + _has_main_guard_call, + _is_safe_regular_file, +) +from .specs import RequirementsState, SpecLoadState + +try: + import tomllib +except ModuleNotFoundError: + import tomli as tomllib + + +_STATIC_PROFILES = frozenset( + {ValidationProfile.STATIC, ValidationProfile.RUNTIME, ValidationProfile.FULL} +) +_RUNTIME_PROFILES = frozenset({ValidationProfile.RUNTIME, ValidationProfile.FULL}) +_FULL_PROFILE = frozenset({ValidationProfile.FULL}) +_OPENENV_DEPENDENCY = re.compile(r"^openenv(?:\s*(?:$|[<>=!~@;])|\[)") +_OPENENV_CORE_DEPENDENCY = re.compile(r"^openenv-core(?:\s*(?:$|[<>=!~@;])|\[)") +_MCP_CONTROL_TOOL_NAMES = frozenset( + {"reset", "reset_async", "step", "step_async", "state"} +) + + +def _target_path(context: ValidationContext) -> Path: + if not isinstance(context.target, Path): + return Path(str(context.target)) + return context.target + + +def _load_pyproject(path: Path) -> tuple[dict[str, Any] | None, str | None]: + pyproject_path = path / "pyproject.toml" + if not _is_safe_regular_file(path, pyproject_path): + return None, "Missing or unsafe pyproject.toml" + try: + with pyproject_path.open("rb") as pyproject_file: + parsed = tomllib.load(pyproject_file) + except (OSError, tomllib.TOMLDecodeError) as exc: + return None, f"Unable to parse pyproject.toml ({type(exc).__name__})" + if not isinstance(parsed, dict): + return None, "pyproject.toml must contain a TOML table" + return parsed, None + + +def _validation_spec(context: ValidationContext) -> CheckOutcome: + loaded = context.spec_load + if loaded.state is SpecLoadState.ABSENT: + return CheckOutcome.fail( + loaded.to_dict(), + message="No supported validation spec was detected", + ) + if loaded.state in { + SpecLoadState.INVALID, + SpecLoadState.UNSUPPORTED, + SpecLoadState.AMBIGUOUS, + }: + return CheckOutcome.fail( + loaded.to_dict(), + message="The validation spec could not be loaded", + ) + if loaded.subject is None: + return CheckOutcome.error( + {"state": loaded.state.value, "reason": "loader returned no subject"} + ) + return CheckOutcome.pass_(loaded.to_dict()) + + +def _validation_requirements(context: ValidationContext) -> CheckOutcome: + subject = context.spec_load.subject + if subject is None: + return CheckOutcome.skip( + {"requirements_present": False, "reason": "spec_unavailable"}, + message="Requirements cannot be loaded until the source spec is valid", + ) + loaded = subject.requirements + if loaded.state is RequirementsState.ABSENT: + return CheckOutcome.skip( + { + **loaded.to_evidence(), + "migration": "Missing requirements are locally non-failing in rfc008-v1", + }, + message="No requirements envelope found; policy defaults apply during migration", + ) + if loaded.state in {RequirementsState.INVALID, RequirementsState.UNSUPPORTED}: + return CheckOutcome.fail( + loaded.to_evidence(), + message="The requirements envelope could not be loaded", + ) + if loaded.requirements is None: + return CheckOutcome.error( + {"state": loaded.state.value, "reason": "loader returned no requirements"} + ) + return CheckOutcome.pass_(loaded.to_evidence()) + + +def _openenv_manifest(context: ValidationContext) -> CheckOutcome: + root = _target_path(context) + path = root / "openenv.yaml" + if not _is_safe_regular_file(root, path): + return CheckOutcome.fail( + {"path": str(path), "missing_or_unsafe": True}, + message="Missing or unsafe openenv.yaml", + ) + try: + manifest = yaml.safe_load(path.read_text(encoding="utf-8")) + except (OSError, yaml.YAMLError) as exc: + return CheckOutcome.fail( + {"path": str(path), "error_type": type(exc).__name__}, + message="Unable to parse openenv.yaml", + ) + if not isinstance(manifest, dict): + return CheckOutcome.fail( + {"path": str(path), "actual_type": type(manifest).__name__}, + message="openenv.yaml must contain a mapping", + ) + required = ("spec_version", "name", "runtime", "app", "port") + missing = [field for field in required if field not in manifest] + if missing: + return CheckOutcome.fail( + {"path": str(path), "missing_fields": missing}, + message="openenv.yaml is missing required fields", + ) + invalid_fields: dict[str, Any] = {} + spec_version = manifest.get("spec_version") + if type(spec_version) is not int or spec_version != 1: + invalid_fields["spec_version"] = manifest.get("spec_version") + if not isinstance(manifest.get("name"), str) or not manifest["name"].strip(): + invalid_fields["name"] = type(manifest.get("name")).__name__ + if manifest.get("runtime") != "fastapi": + invalid_fields["runtime"] = manifest.get("runtime") + app = manifest.get("app") + if not isinstance(app, str) or ":" not in app: + invalid_fields["app"] = app + port = manifest.get("port") + if not isinstance(port, int) or isinstance(port, bool) or not 0 < port < 65536: + invalid_fields["port"] = port + if invalid_fields: + return CheckOutcome.fail( + {"path": str(path), "invalid_fields": invalid_fields}, + message="openenv.yaml contains unsupported or invalid field values", + ) + return CheckOutcome.pass_( + { + "path": str(path), + "spec_version": manifest.get("spec_version"), + "name": manifest.get("name"), + "runtime": manifest.get("runtime"), + "app": manifest.get("app"), + "port": manifest.get("port"), + } + ) + + +def _project_layout(context: ValidationContext) -> CheckOutcome: + path = _target_path(context) + pyproject, error = _load_pyproject(path) + issues: list[str] = [] + if error is not None: + issues.append(error) + server_app = path / "server" / "app.py" + if not _is_safe_regular_file(path, server_app): + issues.append("Missing or unsafe server/app.py") + else: + try: + app_source = server_app.read_text(encoding="utf-8") + except OSError: + issues.append("Unable to read server/app.py") + else: + if "def main(" not in app_source: + issues.append("server/app.py missing main() function") + if not _has_main_guard_call(app_source): + issues.append( + "server/app.py main() function not callable " + "(missing if __name__ == '__main__')" + ) + if pyproject is not None: + scripts = pyproject.get("project", {}).get("scripts", {}) + if not isinstance(scripts, dict) or "server" not in scripts: + issues.append("Missing [project.scripts] server entry point") + else: + server_entry = scripts.get("server") + if not isinstance(server_entry, str) or ":main" not in server_entry: + issues.append( + f"Server entry point should reference main function, got: {server_entry}" + ) + + evidence = { + "environment_root": str(path), + "required_paths": ["pyproject.toml", "server/app.py"], + "issues": issues, + } + if issues: + return CheckOutcome.fail(evidence, message=f"{len(issues)} layout issue(s)") + return CheckOutcome.pass_(evidence) + + +def _dependencies(context: ValidationContext) -> CheckOutcome: + path = _target_path(context) + pyproject, error = _load_pyproject(path) + if error is not None or pyproject is None: + return CheckOutcome.fail( + {"error": error}, message="Cannot validate dependencies" + ) + project = pyproject.get("project", {}) + dependencies = project.get("dependencies", []) if isinstance(project, dict) else [] + if not isinstance(dependencies, list): + return CheckOutcome.fail( + {"actual_type": type(dependencies).__name__}, + message="project.dependencies must be a list", + ) + normalized = [str(dependency).strip().lower() for dependency in dependencies] + declared = any( + _OPENENV_DEPENDENCY.match(dependency) + or _OPENENV_CORE_DEPENDENCY.match(dependency) + for dependency in normalized + ) + dockerfile_declared = _dockerfile_installs_openenv_runtime(path) + if not declared and not dockerfile_declared: + return CheckOutcome.fail( + { + "project_dependency_count": len(normalized), + "dockerfile_installs_openenv": False, + }, + message="Missing required dependency: openenv>=0.2.0", + ) + return CheckOutcome.pass_( + { + "declared_in_pyproject": declared, + "declared_in_dockerfile": dockerfile_declared, + } + ) + + +def _lockfile(context: ValidationContext) -> CheckOutcome: + root = _target_path(context) + lockfile = root / "uv.lock" + if not _is_safe_regular_file(root, lockfile): + return CheckOutcome.fail( + {"path": str(lockfile), "missing_or_unsafe": True}, + message="Missing or unsafe uv.lock - run 'uv lock' to generate it", + ) + try: + with lockfile.open("rb") as lockfile_handle: + lock_data = tomllib.load(lockfile_handle) + except (OSError, tomllib.TOMLDecodeError) as exc: + return CheckOutcome.fail( + {"path": str(lockfile), "error_type": type(exc).__name__}, + message="uv.lock is empty or invalid TOML", + ) + if not isinstance(lock_data.get("version"), int): + return CheckOutcome.fail( + {"path": str(lockfile), "missing_fields": ["version"]}, + message="uv.lock does not contain a lockfile version", + ) + return CheckOutcome.pass_( + {"path": str(lockfile), "lock_version": lock_data["version"]} + ) + + +def _schema_sources(context: ValidationContext) -> CheckOutcome: + path = _target_path(context) + candidates = [path / "server" / "app.py", path / "models.py", path / "client.py"] + checked: list[str] = [] + errors: list[dict[str, Any]] = [] + for candidate in candidates: + if not _is_safe_regular_file(path, candidate): + if candidate.exists() or candidate.is_symlink(): + errors.append( + { + "path": str(candidate.relative_to(path)), + "error_type": "unsafe_file", + "line": None, + } + ) + continue + checked.append(str(candidate.relative_to(path))) + try: + source = candidate.read_text(encoding="utf-8") + ast.parse(source, filename=str(candidate)) + except (OSError, SyntaxError) as exc: + errors.append( + { + "path": str(candidate.relative_to(path)), + "error_type": type(exc).__name__, + "line": getattr(exc, "lineno", None), + } + ) + if errors: + return CheckOutcome.fail( + {"checked": checked, "errors": errors}, + message="Python schema/application sources do not compile", + ) + return CheckOutcome.pass_({"checked": checked}) + + +def _dockerfile(context: ValidationContext) -> CheckOutcome: + path = _target_path(context) + subject = context.spec_load.subject + requirements_load = subject.requirements if subject is not None else None + requirements = ( + requirements_load.requirements if requirements_load is not None else None + ) + declared_image = ( + requirements.environment.container_image if requirements is not None else None + ) + if declared_image: + provenance = ( + requirements_load.provenance.to_dict() + if requirements_load is not None + and requirements_load.provenance is not None + else None + ) + return CheckOutcome.pass_( + {"source": provenance, "declared_image": declared_image} + ) + for candidate in (path / "server" / "Dockerfile", path / "Dockerfile"): + if not _is_safe_regular_file(path, candidate): + continue + try: + content = candidate.read_text(encoding="utf-8") + except OSError: + return CheckOutcome.fail( + {"path": str(candidate)}, message="Unable to read Dockerfile" + ) + has_from = any( + line.lstrip().upper().startswith("FROM ") for line in content.splitlines() + ) + if not has_from: + return CheckOutcome.fail( + {"path": str(candidate), "has_from": False}, + message="Dockerfile has no FROM instruction", + ) + return CheckOutcome.pass_( + {"source": "dockerfile", "path": str(candidate), "has_from": True} + ) + return CheckOutcome.fail( + {"searched": ["server/Dockerfile", "Dockerfile"]}, + message="Missing Dockerfile and normalized container-image declaration", + ) + + +def _custom_verifier_declaration(context: ValidationContext) -> CheckOutcome: + subject = context.spec_load.subject + script = subject.verifier_script if subject is not None else None + if script is None: + return CheckOutcome.skip( + {"declared": False}, + message="No optional isolated verifier is declared by this spec", + severity=ValidationSeverity.ADVISORY, + ) + try: + first_line = script.read_text(encoding="utf-8").splitlines()[:1] + except OSError: + return CheckOutcome.fail( + {"path": str(script)}, message="Unable to read the declared verifier" + ) + return CheckOutcome.pass_( + { + "declared": True, + "path": str(script), + "has_shebang": bool(first_line and first_line[0].startswith("#!")), + "execution": "requires an isolated verifier runner", + } + ) + + +def _legacy_runtime_criterion( + criterion_id: str, +) -> Callable[[ValidationContext], CheckOutcome]: + def evaluate(context: ValidationContext) -> CheckOutcome: + probe_error = context.discovered.get("runtime_probe_error") + if probe_error is not None: + return CheckOutcome.error( + {"error_type": str(probe_error)}, + message="Runtime discovery could not be completed", + ) + report = context.discovered.get("runtime_report") + if not isinstance(report, dict): + return CheckOutcome.skip( + {"runtime_url": context.runtime_url}, + message="No runtime discovery report is available", + ) + criteria = report.get("criteria", []) + criterion = next( + ( + candidate + for candidate in criteria + if isinstance(candidate, dict) and candidate.get("id") == criterion_id + ), + None, + ) + if criterion is None: + return CheckOutcome.error( + {"runtime_url": context.runtime_url, "criterion": criterion_id}, + message="Runtime probe omitted a required criterion", + ) + evidence = { + key: criterion[key] for key in ("expected", "actual") if key in criterion + } + if "details" in criterion: + evidence["probe_details"] = criterion["details"] + if criterion.get("passed") is True: + return CheckOutcome.pass_(evidence) + return CheckOutcome.fail(evidence, message=criterion.get("details")) + + return evaluate + + +def _runtime_declaration( + name: str, *, required: bool = False +) -> Callable[[ValidationContext], CheckOutcome]: + def evaluate(context: ValidationContext) -> CheckOutcome: + declarations = context.discovered.get("runtime_declarations", {}) + if not isinstance(declarations, dict) or name not in declarations: + evidence = {"declaration": name, "available": False} + if required: + return CheckOutcome.error( + evidence, + message=( + f"The capable runtime runner did not provide required {name} " + "validation evidence" + ), + ) + return CheckOutcome.skip( + evidence, + message=f"The running API does not declare {name} validation data", + ) + value = declarations[name] + if not isinstance(value, dict) or "valid" not in value: + return CheckOutcome.error( + {"declaration": name, "actual_type": type(value).__name__}, + message=f"Runtime {name} evidence is malformed", + ) + if value.get("valid") is False: + return CheckOutcome.fail(value) + if value.get("valid") is not True: + return CheckOutcome.error( + {"declaration": name, "valid": value.get("valid")}, + message=f"Runtime {name} evidence must declare valid as a boolean", + ) + return CheckOutcome.pass_({"declaration": name, "value": value}) + + return evaluate + + +def _mcp_control_boundary(context: ValidationContext) -> CheckOutcome: + declarations = context.discovered.get("runtime_declarations", {}) + tools = declarations.get("tools") if isinstance(declarations, dict) else None + names = tools.get("names") if isinstance(tools, dict) else None + if not isinstance(names, list) or not all(isinstance(name, str) for name in names): + return CheckOutcome.error( + {"tools_evidence_available": False}, + message="Runtime discovery did not provide a valid MCP tool-name list", + ) + exposed_controls = sorted( + name for name in names if name.casefold() in _MCP_CONTROL_TOOL_NAMES + ) + if exposed_controls: + return CheckOutcome.fail( + {"exposed_control_tools": exposed_controls}, + message="MCP must not expose infrastructure reset, step, or state controls", + ) + return CheckOutcome.pass_({"tool_count": len(names), "exposed_control_tools": []}) + + +def _external_evidence( + criterion_id: str, +) -> Callable[[ValidationContext], CheckOutcome]: + def evaluate(context: ValidationContext) -> CheckOutcome: + value = context.discovered.get(criterion_id) + if value is None: + return CheckOutcome.error( + {"criterion": criterion_id}, + message="Capable runner did not supply criterion evidence", + ) + if isinstance(value, CheckOutcome): + return value + if isinstance(value, bool): + return ( + CheckOutcome.pass_({"value": value}) + if value + else CheckOutcome.fail({"value": value}) + ) + if isinstance(value, dict): + if "status" not in value: + return CheckOutcome.error( + {"criterion": criterion_id}, + message="Runner evidence must declare an explicit status", + ) + raw_status = value["status"] + try: + status = ValidationStatus(raw_status) + except ValueError: + return CheckOutcome.error( + {"criterion": criterion_id}, + message="Runner supplied an invalid evidence status", + ) + evidence = value.get("evidence", value) + return CheckOutcome( + status=status, + evidence=evidence + if isinstance(evidence, dict) + else {"value": evidence}, + message=value.get("message"), + ) + return CheckOutcome.error( + {"criterion": criterion_id, "actual_type": type(value).__name__}, + message="Runner supplied unsupported criterion evidence", + ) + + return evaluate + + +def _check( + criterion_id: str, + requirement: str, + evaluator: Callable[[ValidationContext], CheckOutcome] | None, + *, + capabilities: frozenset[ValidationCapability], + profiles: frozenset[ValidationProfile], + severity: ValidationSeverity = ValidationSeverity.BLOCKING, + timeout_s: float = 10.0, + built_in: bool = True, + requirement_binding: ValidationRequirementBinding | None = None, +) -> ValidationCheck: + return ValidationCheck( + criterion_id=criterion_id, + requirement=requirement, + capabilities=capabilities, + severity=severity, + timeout_s=timeout_s, + evaluator=evaluator, + profiles=profiles, + built_in=built_in, + requirement_binding=requirement_binding, + ) + + +@lru_cache(maxsize=1) +def get_openenv_checks() -> tuple[ValidationCheck, ...]: + """Return the deterministic served-OpenEnv policy-v1 check catalog.""" + source = frozenset({ValidationCapability.SOURCE}) + runtime = frozenset({ValidationCapability.RUNTIME}) + checks = [ + _check( + "source.validation_spec", + "RFC 008 § Core Abstractions: detect and load the source spec", + _validation_spec, + capabilities=source, + profiles=_STATIC_PROFILES, + ), + _check( + "source.validation_requirements", + "RFC 008 § Core Abstractions: normalize execution requirements", + _validation_requirements, + capabilities=source, + profiles=_STATIC_PROFILES, + ), + _check( + "source.openenv_manifest", + "RFC 008 § Local Execution: validate the OpenEnv manifest", + _openenv_manifest, + capabilities=source, + profiles=_STATIC_PROFILES, + ), + _check( + "source.project_layout", + "RFC 008 § Local Execution: validate project layout and entry point", + _project_layout, + capabilities=source, + profiles=_STATIC_PROFILES, + ), + _check( + "source.dependencies", + "RFC 008 § Local Execution: validate runtime dependencies", + _dependencies, + capabilities=source, + profiles=_STATIC_PROFILES, + ), + _check( + "source.lockfile", + "RFC 008 § Local Execution: validate the dependency lockfile", + _lockfile, + capabilities=source, + profiles=_STATIC_PROFILES, + ), + _check( + "source.schema_sources", + "RFC 008 § Local Execution: validate schema and application sources", + _schema_sources, + capabilities=source, + profiles=_STATIC_PROFILES, + ), + _check( + "source.dockerfile", + "RFC 008 § Local Execution: validate the container image declaration", + _dockerfile, + capabilities=source, + profiles=_STATIC_PROFILES, + ), + _check( + "source.custom_verifier", + "RFC 008 § Validation Model: discover an optional isolated verifier", + _custom_verifier_declaration, + capabilities=source, + profiles=_STATIC_PROFILES, + severity=ValidationSeverity.ADVISORY, + ), + ] + + runtime_descriptions = { + "openapi_version_available": "the runtime publishes a versioned OpenAPI contract", + "health_endpoint": "the runtime health endpoint becomes ready", + "metadata_endpoint": "the runtime publishes environment metadata", + "schema_endpoint": "the runtime publishes action, observation, and state schemas", + "mcp_endpoint": "the agent MCP endpoint is reachable", + "mode_endpoint_consistency": "runtime endpoints match the declared OpenEnv mode", + } + for criterion_id, description in runtime_descriptions.items(): + checks.append( + _check( + criterion_id, + f"RFC 008 § Local Execution: {description}", + _legacy_runtime_criterion(criterion_id), + capabilities=runtime, + profiles=_RUNTIME_PROFILES, + timeout_s=15.0, + ) + ) + for declaration in ( + "websocket", + "tools", + "tasks", + "rewards", + "seeds", + "trajectories", + ): + checks.append( + _check( + f"runtime.{declaration}", + f"RFC 008 § Local Execution: validate runtime {declaration}", + _runtime_declaration(declaration, required=declaration == "websocket"), + capabilities=runtime, + profiles=_RUNTIME_PROFILES, + severity=( + ValidationSeverity.BLOCKING + if declaration == "websocket" + else ValidationSeverity.ADVISORY + ), + ) + ) + + checks.append( + _check( + "runtime.mcp_control_boundary", + "OpenEnv invariant: agent MCP tools cannot expose reset, step, or state controls", + _mcp_control_boundary, + capabilities=runtime, + profiles=_RUNTIME_PROFILES, + ) + ) + + checks.extend( + [ + _check( + "custom.verifier", + "RFC 008 § Validation Model: execute task criteria in an isolated verifier", + _external_evidence("custom.verifier"), + capabilities=frozenset( + { + ValidationCapability.RUNTIME, + ValidationCapability.VERIFIER_ISOLATION, + } + ), + profiles=_RUNTIME_PROFILES, + timeout_s=600.0, + built_in=False, + requirement_binding=ValidationRequirementBinding.VERIFIER, + ), + _check( + "remote.egress_enforcement", + "RFC 008 § Remote Runner: enforce normalized network requirements", + _external_evidence("remote.egress_enforcement"), + capabilities=frozenset({ValidationCapability.NETWORK_ENFORCEMENT}), + profiles=_FULL_PROFILE, + requirement_binding=ValidationRequirementBinding.NETWORK, + ), + _check( + "remote.host_containment", + "RFC 008 § Remote Runner: contain the subject from runner hosts", + _external_evidence("remote.host_containment"), + capabilities=frozenset({ValidationCapability.NETWORK_ENFORCEMENT}), + profiles=_FULL_PROFILE, + ), + _check( + "artifact.image_identity", + "RFC 008 § Security and Results: bind results to an immutable image digest", + _external_evidence("artifact.image_identity"), + capabilities=frozenset({ValidationCapability.CONTAINER_IMAGE}), + profiles=_FULL_PROFILE, + ), + _check( + "artifact.sbom", + "RFC 008 § Remote Runner: inspect the image software bill of materials", + _external_evidence("artifact.sbom"), + capabilities=frozenset({ValidationCapability.CONTAINER_IMAGE}), + profiles=_FULL_PROFILE, + severity=ValidationSeverity.ADVISORY, + ), + _check( + "artifact.signature", + "RFC 008 § Security and Results: verify report and image signatures", + _external_evidence("artifact.signature"), + capabilities=frozenset({ValidationCapability.SIGNATURE_VERIFICATION}), + profiles=_FULL_PROFILE, + ), + _check( + "remote.cross_host_reproducibility", + "RFC 008 § Remote Runner: replay trajectories on an independent host", + _external_evidence("remote.cross_host_reproducibility"), + capabilities=frozenset({ValidationCapability.CROSS_HOST}), + profiles=_FULL_PROFILE, + severity=ValidationSeverity.ADVISORY, + ), + _check( + "remote.reference_model", + "RFC 008 § Remote Runner: run the pinned official reference model", + _external_evidence("remote.reference_model"), + capabilities=frozenset( + { + ValidationCapability.GPU, + ValidationCapability.REFERENCE_MODEL, + } + ), + profiles=_FULL_PROFILE, + severity=ValidationSeverity.ADVISORY, + timeout_s=3600.0, + ), + ] + ) + + criterion_ids = [check.criterion_id for check in checks] + if len(criterion_ids) != len(set(criterion_ids)): + raise RuntimeError( + "The default validation registry has duplicate criterion IDs" + ) + return tuple(checks) diff --git a/src/openenv/validation/models.py b/src/openenv/validation/models.py index ea0be6d96..080a03522 100644 --- a/src/openenv/validation/models.py +++ b/src/openenv/validation/models.py @@ -4,12 +4,40 @@ from __future__ import annotations +import re from dataclasses import dataclass, field from enum import Enum from pathlib import Path from typing import Any, Callable, Mapping -from .specs.base import SpecLoad +from .serialization import json_safe, redact_string +from .specs.base import ExecutionModel, SpecIdentity, SpecLoad, ValidationSubject + + +_STABLE_IDENTIFIER = re.compile(r"^[a-z0-9][a-z0-9._-]{0,127}$") +_TRUSTED_METADATA_STRING_KEYS = frozenset( + { + "id", + "isolation_mode", + "kind", + "policy_version", + } +) + + +def _restore_structural_spec_identity( + serialized: dict[str, Any], raw: Mapping[str, Any] +) -> None: + """Restore adapter-owned identity fields after free-text redaction.""" + for key in ("id", "adapter", "execution_model"): + if key in raw: + serialized[key] = raw[key] + raw_requirements = raw.get("requirements") + safe_requirements = serialized.get("requirements") + if isinstance(raw_requirements, Mapping) and isinstance(safe_requirements, dict): + for key in ("id", "adapter"): + if key in raw_requirements: + safe_requirements[key] = raw_requirements[key] class ValidationStatus(str, Enum): @@ -50,6 +78,13 @@ class ValidationCapability(str, Enum): SIGNATURE_VERIFICATION = "signature_verification" +class ValidationRequirementBinding(str, Enum): + """Normalized requirement that can adapt a policy criterion.""" + + VERIFIER = "verifier" + NETWORK = "network" + + @dataclass(frozen=True) class RunnerCapabilities: """Capabilities and trust properties exposed by a validation runner.""" @@ -59,6 +94,14 @@ class RunnerCapabilities: official: bool = False isolation_mode: str | None = None + def __post_init__(self) -> None: + if not _STABLE_IDENTIFIER.fullmatch(self.runner): + raise ValueError("runner kind must be a stable lowercase identifier") + if self.isolation_mode is not None and not _STABLE_IDENTIFIER.fullmatch( + self.isolation_mode + ): + raise ValueError("isolation mode must be a stable lowercase identifier") + def supports(self, required: frozenset[ValidationCapability]) -> bool: """Return whether all requested capabilities are available.""" return required.issubset(self.available) @@ -164,10 +207,11 @@ class ValidationCheck: default_factory=lambda: frozenset(ValidationProfile) ) built_in: bool = True + requirement_binding: ValidationRequirementBinding | None = None def __post_init__(self) -> None: - if not self.criterion_id.strip(): - raise ValueError("ValidationCheck criterion_id cannot be empty") + if not _STABLE_IDENTIFIER.fullmatch(self.criterion_id): + raise ValueError("ValidationCheck criterion_id must be a stable identifier") if not self.requirement.strip(): raise ValueError("ValidationCheck requirement cannot be empty") if self.timeout_s <= 0: @@ -189,6 +233,38 @@ def default_severity(self) -> ValidationSeverity: return self.severity +@dataclass(frozen=True) +class ValidationPolicy: + """Versioned check catalog applicable to explicit specs and lifecycles.""" + + version: str + supported_subjects: frozenset[tuple[str, ExecutionModel]] + checks: tuple[ValidationCheck, ...] + + def __post_init__(self) -> None: + if not _STABLE_IDENTIFIER.fullmatch(self.version): + raise ValueError("ValidationPolicy version must be a stable identifier") + if not self.supported_subjects or any( + not _STABLE_IDENTIFIER.fullmatch(spec_id) + or not isinstance(model, ExecutionModel) + for spec_id, model in self.supported_subjects + ): + raise ValueError("ValidationPolicy must declare supported subject pairs") + criterion_ids = [check.criterion_id for check in self.checks] + if len(criterion_ids) != len(set(criterion_ids)): + raise ValueError("ValidationPolicy contains duplicate criterion IDs") + + def supports(self, subject: ValidationSubject) -> bool: + """Return whether this policy applies to a loaded subject.""" + return self.supports_identity(subject.spec) + + def supports_identity(self, identity: SpecIdentity) -> bool: + """Return whether this policy applies to a detected spec identity.""" + return bool( + (identity.spec_id, identity.execution_model) in self.supported_subjects + ) + + @dataclass class ValidationContext: """Inputs and discovery data shared by checks in one execution.""" @@ -210,20 +286,42 @@ class ValidationPlan: policy_version: str capabilities: RunnerCapabilities checks: tuple[ValidationCheck, ...] + spec: ValidationSubject | None = None + spec_identity: SpecIdentity | None = None requirements: Mapping[str, Any] = field(default_factory=dict) _policy_attestation: object | None = field(default=None, repr=False, compare=False) + _policy_fingerprint: str | None = field(default=None, repr=False, compare=False) def __post_init__(self) -> None: + if not _STABLE_IDENTIFIER.fullmatch(self.policy_version): + raise ValueError( + "ValidationPlan policy version must be a stable identifier" + ) 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") + if ( + self.spec is not None + and self.spec_identity is not None + and self.spec_identity != self.spec.spec + ): + raise ValueError("ValidationPlan spec identity must match its subject") def to_dict(self) -> dict[str, Any]: """Return the safe, execution-independent plan representation.""" - return { + payload = { "target": self.target, "profile": self.profile.value, "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 + ) + ), "runner": self.capabilities.to_dict(), "requirements": dict(self.requirements), "checks": [ @@ -236,10 +334,32 @@ def to_dict(self) -> dict[str, Any]: "severity": check.severity.value, "timeout_s": check.timeout_s, "built_in": check.built_in, + "requirement_binding": ( + check.requirement_binding.value + if check.requirement_binding is not None + else None + ), } for check in self.checks ], } + 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_checks = serialized.get("checks") + if isinstance(serialized_checks, list): + for safe_check, raw_check in zip( + serialized_checks, payload["checks"], strict=True + ): + if isinstance(safe_check, dict): + safe_check["requirement"] = raw_check["requirement"] + return serialized @dataclass(frozen=True) @@ -276,13 +396,13 @@ def to_dict(self) -> dict[str, Any]: "required_capabilities": sorted( capability.value for capability in self.required_capabilities ), - "evidence": dict(self.evidence), + "evidence": json_safe(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 + payload["details"] = redact_string(self.message) for compatibility_key in ("expected", "actual"): if compatibility_key in self.evidence: payload[compatibility_key] = self.evidence[compatibility_key] diff --git a/src/openenv/validation/planner.py b/src/openenv/validation/planner.py new file mode 100644 index 000000000..a296f0f2c --- /dev/null +++ b/src/openenv/validation/planner.py @@ -0,0 +1,284 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Build versioned validation plans from subjects and runner capabilities.""" + +from __future__ import annotations + +import hashlib +import hmac +import json +import subprocess +from dataclasses import replace +from pathlib import Path +from typing import Any, Iterable + +from .checks import get_openenv_checks +from .models import ( + RunnerCapabilities, + ValidationCapability, + ValidationCheck, + ValidationContext, + ValidationPlan, + ValidationPolicy, + ValidationProfile, + ValidationRequirementBinding, + ValidationSeverity, +) +from .specs import ( + DEFAULT_SPEC_REGISTRY, + ExecutionModel, + NetworkMode, + SpecLoad, + SpecLoadState, + ValidationSpecRegistry, +) + + +VALIDATION_POLICY_VERSION = "rfc008-v1" +OPENENV_VALIDATION_POLICY = ValidationPolicy( + version=VALIDATION_POLICY_VERSION, + supported_subjects=frozenset({("openenv", ExecutionModel.SERVED)}), + checks=get_openenv_checks(), +) +_POLICY_ATTESTATION = object() + + +def _repo_sha(path: Path) -> str | None: + try: + completed = subprocess.run( + ["git", "-C", str(path), "rev-parse", "HEAD"], + check=False, + capture_output=True, + text=True, + timeout=3.0, + ) + except (OSError, subprocess.SubprocessError): + return None + if completed.returncode != 0: + return None + sha = completed.stdout.strip() + return sha or None + + +def _profile(value: ValidationProfile | str) -> ValidationProfile: + if isinstance(value, ValidationProfile): + return value + return ValidationProfile(value) + + +def _adapt_policy( + checks: Iterable[ValidationCheck], spec_load: SpecLoad +) -> tuple[ValidationCheck, ...]: + subject = spec_load.subject + requirements = subject.requirements.requirements if subject is not None else None + adapted: list[ValidationCheck] = [] + for check in checks: + replacement = check + if ( + check.requirement_binding is ValidationRequirementBinding.VERIFIER + and requirements is not None + ): + replacement = replace( + check, timeout_s=requirements.verifier.timeout_s or 600.0 + ) + if check.requirement_binding is ValidationRequirementBinding.NETWORK: + network_mode = ( + requirements.environment.network.mode + if requirements is not None + else NetworkMode.UNSPECIFIED + ) + replacement = replace( + replacement, + severity=( + ValidationSeverity.BLOCKING + if network_mode in {NetworkMode.DENY_ALL, NetworkMode.ALLOWLIST} + else ValidationSeverity.ADVISORY + ), + ) + adapted.append(replacement) + return tuple(adapted) + + +def _select_policy_checks( + checks: Iterable[ValidationCheck], + *, + profile: ValidationProfile, + spec_load: SpecLoad, +) -> tuple[ValidationCheck, ...]: + subject = spec_load.subject + selected = tuple(check for check in checks if profile in check.profiles) + selected = tuple( + check + for check in selected + if check.requirement_binding is not ValidationRequirementBinding.VERIFIER + or (subject is not None and subject.verifier_script is not None) + ) + return _adapt_policy(selected, spec_load) + + +def _policy_requirements(spec_load: SpecLoad) -> dict[str, Any]: + subject = spec_load.subject + if subject is None: + return { + "state": "unavailable", + "policy_defaults": True, + } + evidence = subject.requirements.to_evidence() + evidence["policy_defaults"] = subject.requirements.requirements is None + return evidence + + +def _canonical_fingerprint(plan: ValidationPlan, context: ValidationContext) -> str: + payload = plan.to_dict() + payload["repo_sha"] = context.repo_sha + payload["image_digest"] = context.image_digest + encoded = json.dumps( + payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _policy_for_subject( + policy: ValidationPolicy | None, spec_load: SpecLoad +) -> tuple[ValidationPolicy, bool]: + selected = policy or OPENENV_VALIDATION_POLICY + identity = spec_load.spec + if identity is not None and not selected.supports_identity(identity): + raise ValueError( + f"Validation policy {selected.version!r} does not support " + f"spec {identity.spec_id!r} with execution model " + f"{identity.execution_model.value!r}" + ) + return selected, policy is None + + +def _matches_current_default_subject(context: ValidationContext, subject: Any) -> bool: + if not isinstance(context.target, Path): + return False + try: + current = DEFAULT_SPEC_REGISTRY.resolve(context.target) + except Exception: + return False + return bool(current.state is SpecLoadState.LOADED and current.subject == subject) + + +def is_canonical_policy_plan(plan: ValidationPlan, context: ValidationContext) -> bool: + """Return whether a plan is the unmodified trusted OpenEnv policy output.""" + subject = context.spec_load.subject + if ( + plan._policy_attestation is not _POLICY_ATTESTATION + or plan._policy_fingerprint is None + or subject is None + or not OPENENV_VALIDATION_POLICY.supports(subject) + or not _matches_current_default_subject(context, subject) + ): + return False + expected_checks = _select_policy_checks( + OPENENV_VALIDATION_POLICY.checks, + profile=plan.profile, + spec_load=context.spec_load, + ) + structure_matches = bool( + plan.policy_version == OPENENV_VALIDATION_POLICY.version + and plan.target == str(context.target) + and plan.spec == subject + and plan.spec_identity == subject.spec + and plan.checks == expected_checks + and dict(plan.requirements) == _policy_requirements(context.spec_load) + ) + return bool( + structure_matches + and hmac.compare_digest( + plan._policy_fingerprint, _canonical_fingerprint(plan, context) + ) + ) + + +def build_validation_plan( + target: str | Path, + *, + profile: ValidationProfile | str = ValidationProfile.STATIC, + capabilities: RunnerCapabilities | None = None, + spec_load: SpecLoad | None = None, + spec_id: str | None = None, + spec_registry: ValidationSpecRegistry | None = None, + policy: ValidationPolicy | None = None, + runtime_url: str | None = None, + discovered: dict[str, Any] | None = None, + repo_sha: str | None = None, + image_digest: str | None = None, +) -> tuple[ValidationPlan, ValidationContext]: + """Build a plan without importing or executing submitted code.""" + selected_profile = _profile(profile) + target_path = target if isinstance(target, Path) else Path(target) + is_source = target_path.exists() and target_path.is_dir() + + if spec_load is not None and (spec_id is not None or spec_registry is not None): + raise ValueError("spec_load cannot be combined with spec selection options") + resolved_by_default_registry = spec_load is None and spec_registry is None + registry = spec_registry or DEFAULT_SPEC_REGISTRY + if spec_load is None: + spec_load = ( + registry.resolve(target_path, spec_id=spec_id) + if is_source + else SpecLoad(state=SpecLoadState.ABSENT) + ) + + if policy is None and spec_registry is not None and spec_load.spec is None: + raise ValueError( + "The OpenEnv validation policy cannot be used for an unresolved " + "custom spec registry; provide the matching validation policy" + ) + + selected_policy, is_default_policy = _policy_for_subject(policy, spec_load) + if capabilities is None: + available: set[ValidationCapability] = set() + if is_source: + available.add(ValidationCapability.SOURCE) + if runtime_url is not None: + available.add(ValidationCapability.RUNTIME) + capabilities = RunnerCapabilities( + runner="local", available=frozenset(available), official=False + ) + + checks = _select_policy_checks( + selected_policy.checks, + profile=selected_profile, + spec_load=spec_load, + ) + subject = spec_load.subject + canonical_subject = bool( + subject is not None and OPENENV_VALIDATION_POLICY.supports(subject) + ) + uses_canonical_policy = bool( + is_default_policy + and resolved_by_default_registry + and spec_id in {None, "openenv"} + and canonical_subject + ) + + resolved_repo_sha = repo_sha + if resolved_repo_sha is None and is_source: + resolved_repo_sha = _repo_sha(target_path) + context = ValidationContext( + target=target_path if is_source else str(target), + spec_load=spec_load, + runtime_url=runtime_url, + discovered=dict(discovered or {}), + repo_sha=resolved_repo_sha, + image_digest=image_digest, + ) + plan = ValidationPlan( + target=str(target), + profile=selected_profile, + policy_version=selected_policy.version, + capabilities=capabilities, + checks=checks, + spec=subject, + spec_identity=spec_load.spec, + requirements=_policy_requirements(spec_load), + _policy_attestation=(_POLICY_ATTESTATION if uses_canonical_policy else None), + ) + if uses_canonical_policy: + plan = replace(plan, _policy_fingerprint=_canonical_fingerprint(plan, context)) + return plan, context diff --git a/src/openenv/validation/serialization.py b/src/openenv/validation/serialization.py new file mode 100644 index 000000000..38c89248c --- /dev/null +++ b/src/openenv/validation/serialization.py @@ -0,0 +1,215 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Bounded, credential-safe serialization for validation artifacts.""" + +from __future__ import annotations + +import json +import math +import re +from enum import Enum +from pathlib import Path +from typing import Any, Mapping +from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit + + +_SENSITIVE_MAPPING_KEY = re.compile( + r"(?i)(authorization|cookie|credential|password|passwd|private.?key|secret|" + r"token|api.?key|access.?key|(?:^|[_-])auth(?:$|[_-])|" + r"(?:^|[_-])jwt(?:$|[_-])|(?:^|[_-])pat(?:$|[_-])|askpass)" +) +_SENSITIVE_QUERY_KEYS = frozenset( + { + "access_key", + "apikey", + "api_key", + "auth", + "authorization", + "awsaccesskeyid", + "credential", + "googleaccessid", + "key", + "password", + "secret", + "security_token", + "sig", + "signature", + "token", + "x_amz_credential", + "x_amz_security_token", + "x_amz_signature", + "x_goog_credential", + "x_goog_signature", + } +) +_SECRET_VALUE_PATTERNS = ( + re.compile(r"(?i)(bearer\s+)[^\s,;]+"), + re.compile( + r"\b(?:hf_|sk-|github_pat_|gh[pousr]_)[A-Za-z0-9_-]{8,}\b", + re.IGNORECASE, + ), + re.compile(r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b"), + re.compile( + r"-----BEGIN [A-Z ]*PRIVATE KEY-----.*?" + r"(?:-----END [A-Z ]*PRIVATE KEY-----|\Z)", + re.DOTALL, + ), +) +_URI_VALUE = re.compile(r"(?i)\b[a-z][a-z0-9+.-]*://[^\s<>'\"]+") +_URI_SCHEME = re.compile(r"^[a-z][a-z0-9+.-]*$", re.IGNORECASE) +_MAX_DEPTH = 32 +_MAX_NODES = 10_000 +_MAX_STRING_LENGTH = 1_000_000 + + +def _sensitive_query_key(key: str) -> bool: + normalized = re.sub(r"[^a-z0-9]+", "_", key.lower()).strip("_") + return ( + normalized in _SENSITIVE_QUERY_KEYS + or _SENSITIVE_MAPPING_KEY.search(key) is not None + ) + + +def _redact_uri(value: str) -> str: + try: + parsed = urlsplit(value) + hostname = parsed.hostname + port = parsed.port + except (UnicodeError, ValueError): + return "[REDACTED_INVALID_URL]" + if not _URI_SCHEME.fullmatch(parsed.scheme) or not hostname: + return "[REDACTED_INVALID_URL]" + host = hostname + if ":" in host and not host.startswith("["): + host = f"[{host}]" + if port is not None: + host = f"{host}:{port}" + query = [ + (key, "[REDACTED]" if _sensitive_query_key(key) else item) + for key, item in parse_qsl(parsed.query, keep_blank_values=True) + ] + return urlunsplit((parsed.scheme, host, parsed.path, urlencode(query), "")) + + +def redact_string(value: str) -> str: + """Redact credential-like tokens and URI user information from text.""" + redacted = value + for pattern in _SECRET_VALUE_PATTERNS: + redacted = pattern.sub( + lambda match: ( + f"{match.group(1)}[REDACTED]" if match.lastindex else "[REDACTED]" + ), + redacted, + ) + return _URI_VALUE.sub(lambda match: _redact_uri(match.group(0)), redacted) + + +def json_safe( + value: Any, + *, + key: str | None = None, + seen: set[int] | None = None, + depth: int = 0, + budget: list[int] | None = None, + trusted_string_keys: frozenset[str] = frozenset(), +) -> Any: + """Return a bounded JSON-safe value with credential-bearing fields redacted.""" + if depth > _MAX_DEPTH: + raise ValueError("Value exceeds the maximum nesting depth") + if budget is None: + budget = [_MAX_NODES] + budget[0] -= 1 + if budget[0] < 0: + raise ValueError("Value exceeds the maximum node count") + if key is not None and _SENSITIVE_MAPPING_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("Value contains a non-finite float") + return value + if isinstance(value, str): + if len(value) > _MAX_STRING_LENGTH: + raise ValueError("Value string exceeds the maximum length") + if key in trusted_string_keys: + return value + return redact_string(value) + if isinstance(value, Enum): + return json_safe( + value.value, + seen=seen, + depth=depth + 1, + budget=budget, + trusted_string_keys=trusted_string_keys, + ) + if isinstance(value, Path): + return json_safe( + str(value), + seen=seen, + depth=depth + 1, + budget=budget, + trusted_string_keys=trusted_string_keys, + ) + 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("Value 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_NODES: + raise ValueError("Value mapping exceeds the maximum size") + if not isinstance(item_key, (str, int, float, bool)): + raise ValueError("Value mapping contains an unsupported key") + if isinstance(item_key, float) and not math.isfinite(item_key): + raise ValueError("Value mapping contains a non-finite key") + key_text = str(item_key) + if len(key_text) > _MAX_STRING_LENGTH: + raise ValueError("Value 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("Value mapping contains duplicate JSON keys") + result[item_key] = json_safe( + item, + key=item_key, + seen=seen, + depth=depth + 1, + budget=budget, + trusted_string_keys=trusted_string_keys, + ) + seen.remove(container_id) + return result + if isinstance(value, (list, tuple)): + result = [ + json_safe( + item, + seen=seen, + depth=depth + 1, + budget=budget, + trusted_string_keys=trusted_string_keys, + ) + 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, + trusted_string_keys=trusted_string_keys, + ) + 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__} diff --git a/src/openenv/validation/source_inspection.py b/src/openenv/validation/source_inspection.py new file mode 100644 index 000000000..46073aa85 --- /dev/null +++ b/src/openenv/validation/source_inspection.py @@ -0,0 +1,106 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Static source-inspection helpers shared by validation entry points.""" + +from __future__ import annotations + +import ast +import re +from pathlib import Path + + +_OPENENV_DOCKER_INSTALL_RE = re.compile( + r"(?=!~@;])|\[)" +) + + +def _is_safe_regular_file(root: Path, candidate: Path) -> bool: + """Return whether a regular file is contained without symlink traversal.""" + 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: + resolved_root = root.resolve(strict=True) + resolved_candidate = candidate.resolve(strict=True) + except OSError: + return False + return bool( + candidate.is_file() and resolved_candidate.is_relative_to(resolved_root) + ) + + +def _has_main_guard_call(app_content: str) -> bool: + """Return True when the module calls main() under a __main__ guard.""" + try: + tree = ast.parse(app_content) + except SyntaxError: + return ( + "__name__" in app_content + and "__main__" in app_content + and "main(" in app_content + ) + + for node in ast.iter_child_nodes(tree): + if not isinstance(node, ast.If) or not _is_main_guard(node.test): + continue + + if any(_contains_main_call(guarded_node) for guarded_node in node.body): + return True + + return False + + +def _is_main_guard(test: ast.expr) -> bool: + """Return True for `if __name__ == "__main__"` tests.""" + return ( + isinstance(test, ast.Compare) + and isinstance(test.left, ast.Name) + and test.left.id == "__name__" + and len(test.ops) == 1 + and isinstance(test.ops[0], ast.Eq) + and len(test.comparators) == 1 + and isinstance(test.comparators[0], ast.Constant) + and test.comparators[0].value == "__main__" + ) + + +def _contains_main_call(node: ast.AST) -> bool: + """Return True when an AST node contains a direct `main(...)` call.""" + return any( + isinstance(candidate, ast.Call) + and isinstance(candidate.func, ast.Name) + and candidate.func.id == "main" + for candidate in ast.walk(node) + ) + + +def _dockerfile_installs_openenv_runtime(env_path: Path) -> bool: + """Return True when a Docker deployment installs OpenEnv outside pyproject.""" + for dockerfile_path in ( + env_path / "server" / "Dockerfile", + env_path / "Dockerfile", + ): + if not _is_safe_regular_file(env_path, dockerfile_path): + continue + + try: + dockerfile = dockerfile_path.read_text(encoding="utf-8") + except OSError: + continue + + for line in dockerfile.splitlines(): + stripped = line.strip().lower() + if not stripped or stripped.startswith("#"): + continue + if _OPENENV_DOCKER_INSTALL_RE.search(stripped): + return True + if "openenv-core" in stripped: + return True + + return False diff --git a/tests/test_validation/_helpers.py b/tests/test_validation/_helpers.py index de1b53c2f..f016a76fe 100644 --- a/tests/test_validation/_helpers.py +++ b/tests/test_validation/_helpers.py @@ -5,11 +5,42 @@ from pathlib import Path +def write_valid_env(env_dir: Path) -> None: + """Create a minimal source environment that passes static validation.""" + (env_dir / "server").mkdir(parents=True) + (env_dir / "openenv.yaml").write_text( + "spec_version: 1\n" + "name: test_env\n" + "type: space\n" + "runtime: fastapi\n" + "app: server.app:app\n" + "port: 8000\n" + ) + (env_dir / "uv.lock").write_text( + 'version = 1\nrevision = 1\nrequires-python = ">=3.10"\n' + ) + (env_dir / "pyproject.toml").write_text( + "[project]\n" + 'name = "test-env"\n' + 'version = "0.1.0"\n' + 'dependencies = ["openenv>=0.4.0"]\n' + "\n" + "[project.scripts]\n" + 'server = "server.app:main"\n' + ) + (env_dir / "server" / "app.py").write_text( + "def main():\n return None\n\nif __name__ == '__main__':\n main()\n" + ) + (env_dir / "server" / "Dockerfile").write_text( + 'FROM python:3.12-slim\nCMD ["server"]\n' + ) + + 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) + write_valid_env(environment) tests_dir.mkdir() (tests_dir / "test.sh").write_text("#!/bin/sh\nexit 0\n") (task_root / "task.toml").write_text( diff --git a/tests/test_validation/test_planner.py b/tests/test_validation/test_planner.py new file mode 100644 index 000000000..68f4db100 --- /dev/null +++ b/tests/test_validation/test_planner.py @@ -0,0 +1,308 @@ +# SPDX-License-Identifier: BSD-3-Clause + +"""Tests for validation policy planning and capability selection.""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import replace +from pathlib import Path + +import pytest +from openenv.validation.models import ( + CheckOutcome, + ValidationCapability, + ValidationCheck, + ValidationPolicy, + ValidationProfile, + ValidationRequirementBinding, + ValidationSeverity, +) +from openenv.validation.planner import build_validation_plan, is_canonical_policy_plan +from openenv.validation.specs import ( + AdapterIdentity, + DEFAULT_SPEC_REGISTRY, + DetectionMode, + EnvironmentRequirements, + ExecutionModel, + NetworkMode, + NetworkRequirements, + PhaseRequirements, + RequirementsLoad, + RequirementsState, + SpecIdentity, + SpecLoad, + SpecLoadState, + ValidationRequirements, + ValidationSpecRegistry, + ValidationSubject, +) + +from ._helpers import write_harbor_task, write_valid_env + + +def test_plan_carries_sanitized_harbor_execution_requirements(tmp_path: Path) -> None: + env_dir = tmp_path / "test_env" + write_valid_env(env_dir) + (env_dir / "task.toml").write_text( + 'schema_version = "1.1"\n' + 'artifacts = ["/workspace/result.json"]\n' + "[task]\n" + 'name = "org/task"\n' + 'description = "https://alice:hunter2@example.com/private"\n' + "[agent]\n" + "timeout_sec = 120\n" + "[verifier]\n" + "timeout_sec = 30\n" + 'env = { API_TOKEN = "secret-value" }\n' + "[environment]\n" + "build_timeout_sec = 300\n" + "cpus = 4\n" + "memory_mb = 8192\n" + "gpus = 1\n" + "allow_internet = false\n" + "[environment.healthcheck]\n" + 'command = "curl -f http://127.0.0.1:8000/health"\n' + "retries = 5\n" + ) + + plan, _context = build_validation_plan(env_dir) + payload = plan.to_dict() + serialized = json.dumps(payload) + + assert payload["requirements"]["resources"]["cpus"] == 4 + assert payload["requirements"]["resources"]["memory_mb"] == 8192 + assert payload["requirements"]["resources"]["gpus"] == 1 + assert payload["requirements"]["network"]["mode"] == "deny_all" + assert payload["requirements"]["timeouts"] == { + "build_s": 300.0, + "agent_s": 120.0, + "verifier_s": 30.0, + } + assert payload["requirements"]["healthcheck"]["retries"] == 5 + assert payload["requirements"]["artifacts"][0]["source"] == ( + "/workspace/result.json" + ) + assert "API_TOKEN" in serialized + assert "secret-value" not in serialized + assert "alice:hunter2" not in serialized + assert payload["spec"]["id"] == "openenv" + assert payload["spec"]["adapter"] == {"id": "openenv-yaml", "version": "1"} + assert payload["spec"]["execution_model"] == "served" + assert payload["spec"]["requirements"]["id"] == "harbor" + assert payload["spec"]["requirements"]["document_digest"].startswith("sha256:") + + +def test_plan_includes_harbor_step_requirements_and_sibling_verifier( + tmp_path: Path, +) -> None: + task_root = tmp_path / "harbor-task" + environment = write_harbor_task(task_root) + + plan, _context = build_validation_plan( + environment, + profile=ValidationProfile.RUNTIME, + ) + + assert "custom.verifier" in {check.criterion_id for check in plan.checks} + requirements = plan.to_dict()["requirements"] + assert requirements["steps"][0]["min_reward"] == { + "correctness": 0.8, + "style": 0.5, + } + assert requirements["steps"][0]["artifacts"][0]["source"] == ( + "/workspace/result.json" + ) + + +class _OneShotAdapter: + spec_id = "one-shot-test" + adapter_id = "one-shot-test-adapter" + adapter_version = "1" + execution_model = ExecutionModel.ONE_SHOT + signature_files = ("task.oneshot",) + + def detect(self, root: Path) -> bool: + return (root / "task.oneshot").is_file() + + def inspect(self, root: Path) -> SpecLoad: + identity = SpecIdentity( + spec_id=self.spec_id, + spec_version="1", + 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) + verifier = root / "verify.sh" + verifier_declared = verifier.is_file() + return SpecLoad( + state=SpecLoadState.LOADED, + subject=ValidationSubject( + spec=identity, + signature_path="task.oneshot", + detection_mode=DetectionMode.AUTO, + requirements=RequirementsLoad( + state=RequirementsState.LOADED, + requirements=ValidationRequirements( + verifier=PhaseRequirements(timeout_s=17.0), + environment=EnvironmentRequirements( + network=NetworkRequirements(mode=NetworkMode.DENY_ALL) + ), + ), + ), + verifier_script=(verifier if verifier_declared else None), + verifier_path=("verify.sh" if verifier_declared else None), + verifier_digest=( + f"sha256:{hashlib.sha256(verifier.read_bytes()).hexdigest()}" + if verifier_declared + else None + ), + ), + ) + + +def test_custom_spec_uses_its_policy_without_openenv_checks(tmp_path: Path) -> None: + (tmp_path / "task.oneshot").write_text("signature") + registry = ValidationSpecRegistry((_OneShotAdapter(),)) + check = ValidationCheck( + criterion_id="spec.one_shot.structure", + requirement="The one-shot package has its required structure", + capabilities=frozenset({ValidationCapability.SOURCE}), + severity=ValidationSeverity.BLOCKING, + timeout_s=1.0, + evaluator=lambda _context: CheckOutcome.pass_(), + ) + policy = ValidationPolicy( + version="one-shot-v1", + supported_subjects=frozenset({("one-shot-test", ExecutionModel.ONE_SHOT)}), + checks=(check,), + ) + + plan, context = build_validation_plan( + tmp_path, spec_registry=registry, policy=policy + ) + + assert plan.policy_version == "one-shot-v1" + assert [planned.criterion_id for planned in plan.checks] == [ + "spec.one_shot.structure" + ] + assert plan.spec is context.spec_load.subject + assert "source.openenv_manifest" not in { + planned.criterion_id for planned in plan.checks + } + + +def test_openenv_policy_rejects_an_incompatible_execution_model( + tmp_path: Path, +) -> None: + (tmp_path / "task.oneshot").write_text("signature") + + with pytest.raises(ValueError, match="does not support"): + build_validation_plan( + tmp_path, + spec_registry=ValidationSpecRegistry((_OneShotAdapter(),)), + ) + + +def test_openenv_policy_rejects_an_invalid_incompatible_spec( + tmp_path: Path, +) -> None: + class _InvalidOneShotAdapter(_OneShotAdapter): + def inspect(self, root: Path) -> SpecLoad: + return SpecLoad( + state=SpecLoadState.INVALID, + identity=SpecIdentity( + spec_id=self.spec_id, + spec_version=None, + adapter=AdapterIdentity(self.adapter_id, self.adapter_version), + execution_model=self.execution_model, + ), + error="malformed one-shot task", + ) + + (tmp_path / "task.oneshot").write_text("signature") + + with pytest.raises(ValueError, match="does not support"): + build_validation_plan( + tmp_path, + spec_registry=ValidationSpecRegistry((_InvalidOneShotAdapter(),)), + ) + + +def test_openenv_policy_rejects_an_unresolved_custom_registry( + tmp_path: Path, +) -> None: + with pytest.raises(ValueError, match="matching validation policy"): + build_validation_plan( + tmp_path, + spec_registry=ValidationSpecRegistry((_OneShotAdapter(),)), + ) + + +def test_injected_spec_load_cannot_claim_canonical_policy(tmp_path: Path) -> None: + write_valid_env(tmp_path) + loaded = DEFAULT_SPEC_REGISTRY.resolve(tmp_path) + + plan, context = build_validation_plan(tmp_path, spec_load=loaded) + + assert is_canonical_policy_plan(plan, context) is False + + +def test_policy_adaptation_uses_normalized_requirements_for_any_spec( + tmp_path: Path, +) -> None: + (tmp_path / "task.oneshot").write_text("signature") + (tmp_path / "verify.sh").write_text("#!/bin/sh\n") + checks = ( + ValidationCheck( + criterion_id="spec.one_shot.verifier", + requirement="Run the declared verifier", + capabilities=frozenset(), + severity=ValidationSeverity.BLOCKING, + timeout_s=600.0, + evaluator=lambda _context: CheckOutcome.pass_(), + requirement_binding=ValidationRequirementBinding.VERIFIER, + ), + ValidationCheck( + criterion_id="spec.one_shot.egress", + requirement="Enforce network requirements", + capabilities=frozenset(), + severity=ValidationSeverity.ADVISORY, + timeout_s=10.0, + evaluator=lambda _context: CheckOutcome.pass_(), + requirement_binding=ValidationRequirementBinding.NETWORK, + ), + ) + policy = ValidationPolicy( + version="one-shot-v1", + supported_subjects=frozenset({("one-shot-test", ExecutionModel.ONE_SHOT)}), + checks=checks, + ) + + plan, _context = build_validation_plan( + tmp_path, + profile=ValidationProfile.FULL, + spec_registry=ValidationSpecRegistry((_OneShotAdapter(),)), + policy=policy, + ) + + planned = {check.criterion_id: check for check in plan.checks} + assert planned["spec.one_shot.verifier"].timeout_s == 17.0 + assert planned["spec.one_shot.egress"].severity is ValidationSeverity.BLOCKING + + +def test_canonical_plan_revalidates_bound_verifier_file(tmp_path: Path) -> None: + environment = write_harbor_task(tmp_path / "harbor-task") + plan, context = build_validation_plan(environment) + assert plan.spec is not None + assert plan.spec.verifier_script is not None + outside = tmp_path / "outside.sh" + outside.write_text("#!/bin/sh\nexit 0\n") + forged_subject = replace(plan.spec, verifier_script=outside) + forged_load = replace(context.spec_load, subject=forged_subject) + context.spec_load = forged_load + forged_plan = replace(plan, spec=forged_subject) + + assert is_canonical_policy_plan(forged_plan, context) is False