From 95bfed62905543c061414a5149b471605f51c65b Mon Sep 17 00:00:00 2001 From: ECHO OMEGA PRIME Date: Sun, 1 Nov 2026 23:23:57 -0600 Subject: [PATCH] fix: reconcile exact-source project runners with governed execution --- contracts/certforge-sdk-capabilities.v1.json | 101 +++++++++ docs/RUNTIME_IDENTITY.md | 51 ++++- scripts/register_certforge_sdk_schemas.sql | 2 +- src/echo_certification_forge/app.py | 2 +- .../operational_telemetry.py | 43 ++-- src/echo_certification_forge/run_worker.py | 27 ++- .../runtime_admission.py | 103 +++++++++ .../runtime_identity.py | 35 ++- src/echo_certification_forge/service.py | 14 +- src/echo_certification_forge/subscriber.py | 2 +- tests/test_intake_recovery_isolation.py | 19 ++ .../test_operational_telemetry_connections.py | 73 +++++++ tests/test_runtime_admission.py | 199 ++++++++++++++++++ 13 files changed, 642 insertions(+), 29 deletions(-) create mode 100644 src/echo_certification_forge/runtime_admission.py create mode 100644 tests/test_operational_telemetry_connections.py create mode 100644 tests/test_runtime_admission.py diff --git a/contracts/certforge-sdk-capabilities.v1.json b/contracts/certforge-sdk-capabilities.v1.json index 7e009b6..de984fb 100644 --- a/contracts/certforge-sdk-capabilities.v1.json +++ b/contracts/certforge-sdk-capabilities.v1.json @@ -1770,6 +1770,107 @@ "type": "null" } ] + }, + "project_id": { + "anyOf": [ + { + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$", + "type": "string" + }, + { + "type": "null" + } + ] + }, + "target": { + "anyOf": [ + { + "additionalProperties": false, + "properties": { + "artifact_sha256": { + "anyOf": [ + { + "pattern": "^[0-9a-f]{64}$", + "type": "string" + }, + { + "type": "null" + } + ] + }, + "identity_digest": { + "pattern": "^[0-9a-f]{64}$", + "type": "string" + }, + "path": { + "anyOf": [ + { + "maxLength": 2048, + "minLength": 1, + "type": "string" + }, + { + "type": "null" + } + ] + }, + "ref": { + "anyOf": [ + { + "maxLength": 256, + "minLength": 1, + "type": "string" + }, + { + "type": "null" + } + ] + }, + "reference": { + "maxLength": 2048, + "minLength": 1, + "type": "string" + }, + "source_commit": { + "anyOf": [ + { + "maxLength": 64, + "minLength": 7, + "type": "string" + }, + { + "type": "null" + } + ] + }, + "target_type": { + "pattern": "^(local|git|archive|container|package|deployment|mcp|sdk|cli)$", + "type": "string" + }, + "url": { + "anyOf": [ + { + "maxLength": 2048, + "minLength": 1, + "type": "string" + }, + { + "type": "null" + } + ] + } + }, + "required": [ + "target_type", + "identity_digest", + "reference" + ], + "type": "object" + }, + { + "type": "null" + } + ] } }, "required": [ diff --git a/docs/RUNTIME_IDENTITY.md b/docs/RUNTIME_IDENTITY.md index c6f2f58..0982564 100644 --- a/docs/RUNTIME_IDENTITY.md +++ b/docs/RUNTIME_IDENTITY.md @@ -1,5 +1,48 @@ # Measured worker runtime identity +## Source-bound project profiles + +An operator may select `certforge.runtime.v2` in the same public profile file and +add `project_bindings`. Each binding requires a unique `binding_id`, exact +`tenant_id`, `project_id`, HTTPS `repository`, 40-character `source_commit`, a +nonempty `dependencies` map of repository-relative paths to SHA-256 hashes, +an exact `journey`, and `sandbox` settings using the existing restricted fields. +Image pins, non-root execution and all existing isolation checks remain required. +The operator file is not accepted from target source or the subscriber request. +Installing it still requires independent image qualification and release admission. + +The runtime resolver accepts optional `project_id` and `target` together, checks +live project access, and resolves the operator binding for that tenant, project, +repository and revision. It returns the measured environment using the existing +`certforge.runtime.v1` response contract. Clients cannot choose a profile or image. +Other tenants, projects and repositories keep the unchanged default sandbox; +unadmitted revisions of a bound repository fail closed. This also preserves other +repositories when an older App installation shares one subscriber project. + +The worker derives its project from the persisted run, selects the same binding, +and verifies the declared environment before acquisition. After acquiring the +exact revision it verifies the actual source commit and each dependency hash, +rejecting missing, oversized, changed, escaping or symlinked files before execution. +The complete operator binding digest is included in the measured environment; +changing dependency expectations therefore invalidates the old commitment. +All reservation, worker, lease, signer and final environment fences remain active. +Selection logs include binding ID, source SHA and binding digest, with no secrets. + +Deploy the paired GitHub App resolver change and regenerated SDK schemas only +through the governed release path. An old queued environment must not be silently +rewritten or replayed as a new certification. No service, trust, queue, signing key, +readiness report or production attestation is changed by this source implementation. + +Stage the matching API, dispatcher, worker, generated SDK contract and App caller +before activating a source-bound profile. Preserve the previous admitted release, +profile and caller configuration as one rollback unit: a v1-only binary rejects a +v2 profile. Retain queued runs and submission checkpoints without rewriting their +environment or replaying uncertain intents. A source, image or profile change +requires fresh exact-identity evidence; prior readiness and image admission do not +transfer to the new candidate. Connection cleanup retains transaction commit and +rollback while explicitly closing the worker guard and telemetry handles; failure +to remove an owned acceptance workspace still fails its cleanup check. + The `certforge.runtime.v1` contract replaces fixed `certforge-worker-env:*` labels. An environment commits to the selected manifest digest, exact journey argv and mandatory rules, executable CertForge package bytes, dependency declarations, @@ -116,7 +159,13 @@ secret-field rejection; service/dispatcher/worker identity agreement; and reject before acquisition, authenticated per-job resolution, caller-image rejection, and equivalent source/real-built-wheel harness commitments. Subscriber regression tests additionally verify that changed -journey commitments cannot execute or receive a signed verdict. Full repository +journey commitments cannot execute or receive a signed verdict. Exact-source +project bindings retain the operator-input circuit fence before tenant selection. +An unadmitted source uses the same bounded tenant retry delay as a journey or +environment mismatch, without halting unrelated tenants. The project-binding +tests verify API/worker agreement, altered or missing dependency rejection before +execution, unchanged default isolation and the retained operator classification. +Full repository and acceptance checks remain required before a release claim. Host Python and kernel measurements are intentional identity inputs. A host update diff --git a/scripts/register_certforge_sdk_schemas.sql b/scripts/register_certforge_sdk_schemas.sql index 7074bce..0ffbfbb 100644 --- a/scripts/register_certforge_sdk_schemas.sql +++ b/scripts/register_certforge_sdk_schemas.sql @@ -54,7 +54,7 @@ INSERT INTO certforge_sdk_contract(id, input_schema_json, output_schema_json) VA ('echo.certforge.r5.status', '{"additionalProperties":false,"properties":{"command":{"maxLength":64,"minLength":1,"type":"string"},"run_id":{"pattern":"^[A-Za-z0-9._-]{1,64}$","type":"string"}},"required":["command","run_id"],"type":"object"}'::jsonb, '{"additionalProperties":false,"properties":{"note":{"type":"string"},"result":{"anyOf":[{"additionalProperties":false,"properties":{"completion_marker":{"type":["string","null"]},"deployment_authorized":{"type":"boolean"},"evidence_run_id":{"type":"string"},"forge_verification":{"additionalProperties":true,"type":"object"},"mode":{"enum":["preflight","full"],"type":"string"},"operator_exit_code":{"type":"integer"},"operator_report":{"additionalProperties":true,"type":"object"},"r5_gate":{"enum":["PASS","BLOCK"],"type":"string"},"release_verdict":{"enum":["NOT_READY"],"type":"string"},"run_outcome":{"type":"string"},"schema":{"type":"string"},"stderr_tail":{"type":"string"}},"required":["schema","mode","evidence_run_id","run_outcome","release_verdict","r5_gate","deployment_authorized","completion_marker","operator_report","forge_verification","operator_exit_code"],"type":"object"},{"type":"null"}]},"run_id":{"type":"string"},"status":{"enum":["RUNNING","COMPLETE","FAILED"],"type":"string"}},"required":["run_id","status"],"type":"object"}'::jsonb), ('echo.certforge.r5.submit_async', '{"additionalProperties":false,"properties":{"command":{"maxLength":64,"minLength":1,"type":"string"},"dry_run":{"type":"boolean"},"evidence_run_id":{"pattern":"^[A-Za-z0-9._-]{1,64}$","type":"string"},"evidence_run_nonce":{"maxLength":128,"minLength":16,"type":"string"},"expected_base_model_digest":{"pattern":"^[0-9a-f]{64}$","type":"string"},"expected_base_model_revision":{"maxLength":128,"minLength":6,"type":"string"},"expected_gs343_digest":{"pattern":"^[0-9a-f]{64}$","type":"string"},"expected_r2d2_digest":{"pattern":"^[0-9a-f]{64}$","type":"string"},"expected_registry_revision":{"maxLength":128,"minLength":6,"type":"string"},"expected_registry_snapshot_digest":{"pattern":"^[0-9a-f]{64}$","type":"string"},"expected_server_build_digest":{"pattern":"^[0-9a-f]{64}$","type":"string"},"expected_signing_key_id":{"maxLength":128,"minLength":8,"type":"string"},"target_family":{"enum":["gs343","r2d2"],"type":"string"}},"required":["command","target_family","expected_server_build_digest","expected_registry_snapshot_digest","expected_registry_revision","expected_signing_key_id","expected_base_model_digest","expected_base_model_revision","expected_gs343_digest","expected_r2d2_digest","evidence_run_id","evidence_run_nonce"],"type":"object"}'::jsonb, '{"additionalProperties":false,"properties":{"idempotent":{"type":"boolean"},"mode":{"enum":["preflight","full"],"type":"string"},"poll_capability":{"enum":["echo.certforge.r5.status"],"type":"string"},"run_id":{"type":"string"},"status":{"enum":["RESERVED","RUNNING"],"type":"string"}},"required":["run_id","status","mode"],"type":"object"}'::jsonb), ('echo.certforge.run', '{"additionalProperties":false,"properties":{"command":{"maxLength":64,"minLength":1,"type":"string"},"run_id":{"pattern":"^[A-Za-z0-9._-]{1,128}$","type":"string"},"tenant":{"pattern":"^[A-Za-z0-9._-]{1,128}$","type":"string"}},"required":["command","run_id","tenant"],"type":"object"}'::jsonb, '{"additionalProperties":false,"properties":{"poll_capability":{"enum":["echo.certforge.status"],"type":"string"},"run_id":{"type":"string"},"sandboxed":{"type":"boolean"},"status":{"enum":["PENDING","CLAIMED"],"type":"string"},"tenant":{"type":"string"}},"required":["run_id","status","tenant","sandboxed","poll_capability"],"type":"object"}'::jsonb), - ('echo.certforge.runtime_environment', '{"additionalProperties":false,"properties":{"command":{"maxLength":64,"minLength":1,"type":"string"},"journey":{"anyOf":[{"items":{"maxLength":2048,"minLength":1,"pattern":"^[^\\x00]*$","type":"string"},"maxItems":32,"minItems":1,"type":"array"},{"type":"null"}]}},"required":["command"],"type":"object"}'::jsonb, '{"additionalProperties":false,"properties":{"environment":{"additionalProperties":false,"properties":{"adapter_set_sha256":{"pattern":"^[0-9a-f]{64}$","type":"string"},"egress_policy_sha256":{"pattern":"^[0-9a-f]{64}$","type":"string"},"harness_sha256":{"pattern":"^[0-9a-f]{64}$","type":"string"},"model_route_sha256":{"pattern":"^[0-9a-f]{64}$","type":"string"},"os_runtime_sha256":{"pattern":"^[0-9a-f]{64}$","type":"string"},"policy_sha256":{"pattern":"^[0-9a-f]{64}$","type":"string"},"prompt_set_sha256":{"pattern":"^[0-9a-f]{64}$","type":"string"},"runner_image_sha256":{"pattern":"^[0-9a-f]{64}$","type":"string"},"test_plan_sha256":{"pattern":"^[0-9a-f]{64}$","type":"string"}},"required":["runner_image_sha256","adapter_set_sha256","test_plan_sha256","policy_sha256","harness_sha256","prompt_set_sha256","model_route_sha256","os_runtime_sha256","egress_policy_sha256"],"type":"object"},"journey":{"anyOf":[{"items":{"maxLength":2048,"minLength":1,"pattern":"^[^\\x00]*$","type":"string"},"type":"array"},{"type":"null"}]},"schema_version":{"const":"certforge.runtime.v1","type":"string"},"submit_environment":{"additionalProperties":false,"properties":{"identity_digest":{"pattern":"^[0-9a-f]{64}$","type":"string"},"runner_image_digest":{"anyOf":[{"pattern":"^sha256:[0-9a-f]{64}$","type":"string"},{"type":"null"}]}},"required":["identity_digest"],"type":"object"}},"required":["journey","environment","submit_environment"],"type":"object"}'::jsonb), + ('echo.certforge.runtime_environment', '{"additionalProperties":false,"properties":{"command":{"maxLength":64,"minLength":1,"type":"string"},"journey":{"anyOf":[{"items":{"maxLength":2048,"minLength":1,"pattern":"^[^\\x00]*$","type":"string"},"maxItems":32,"minItems":1,"type":"array"},{"type":"null"}]},"project_id":{"anyOf":[{"pattern":"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$","type":"string"},{"type":"null"}]},"target":{"anyOf":[{"additionalProperties":false,"properties":{"artifact_sha256":{"anyOf":[{"pattern":"^[0-9a-f]{64}$","type":"string"},{"type":"null"}]},"identity_digest":{"pattern":"^[0-9a-f]{64}$","type":"string"},"path":{"anyOf":[{"maxLength":2048,"minLength":1,"type":"string"},{"type":"null"}]},"ref":{"anyOf":[{"maxLength":256,"minLength":1,"type":"string"},{"type":"null"}]},"reference":{"maxLength":2048,"minLength":1,"type":"string"},"source_commit":{"anyOf":[{"maxLength":64,"minLength":7,"type":"string"},{"type":"null"}]},"target_type":{"pattern":"^(local|git|archive|container|package|deployment|mcp|sdk|cli)$","type":"string"},"url":{"anyOf":[{"maxLength":2048,"minLength":1,"type":"string"},{"type":"null"}]}},"required":["target_type","identity_digest","reference"],"type":"object"},{"type":"null"}]}},"required":["command"],"type":"object"}'::jsonb, '{"additionalProperties":false,"properties":{"environment":{"additionalProperties":false,"properties":{"adapter_set_sha256":{"pattern":"^[0-9a-f]{64}$","type":"string"},"egress_policy_sha256":{"pattern":"^[0-9a-f]{64}$","type":"string"},"harness_sha256":{"pattern":"^[0-9a-f]{64}$","type":"string"},"model_route_sha256":{"pattern":"^[0-9a-f]{64}$","type":"string"},"os_runtime_sha256":{"pattern":"^[0-9a-f]{64}$","type":"string"},"policy_sha256":{"pattern":"^[0-9a-f]{64}$","type":"string"},"prompt_set_sha256":{"pattern":"^[0-9a-f]{64}$","type":"string"},"runner_image_sha256":{"pattern":"^[0-9a-f]{64}$","type":"string"},"test_plan_sha256":{"pattern":"^[0-9a-f]{64}$","type":"string"}},"required":["runner_image_sha256","adapter_set_sha256","test_plan_sha256","policy_sha256","harness_sha256","prompt_set_sha256","model_route_sha256","os_runtime_sha256","egress_policy_sha256"],"type":"object"},"journey":{"anyOf":[{"items":{"maxLength":2048,"minLength":1,"pattern":"^[^\\x00]*$","type":"string"},"type":"array"},{"type":"null"}]},"schema_version":{"const":"certforge.runtime.v1","type":"string"},"submit_environment":{"additionalProperties":false,"properties":{"identity_digest":{"pattern":"^[0-9a-f]{64}$","type":"string"},"runner_image_digest":{"anyOf":[{"pattern":"^sha256:[0-9a-f]{64}$","type":"string"},{"type":"null"}]}},"required":["identity_digest"],"type":"object"}},"required":["journey","environment","submit_environment"],"type":"object"}'::jsonb), ('echo.certforge.service_status', '{"additionalProperties":false,"properties":{"command":{"maxLength":64,"minLength":1,"type":"string"}},"required":["command"],"type":"object"}'::jsonb, '{"additionalProperties":true,"type":"object"}'::jsonb), ('echo.certforge.status', '{"additionalProperties":false,"properties":{"command":{"maxLength":64,"minLength":1,"type":"string"},"run_id":{"type":"string"}},"required":["command","run_id"],"type":"object"}'::jsonb, '{"additionalProperties":true,"type":"object"}'::jsonb), ('echo.certforge.submit', '{"additionalProperties":false,"properties":{"command":{"maxLength":64,"minLength":1,"type":"string"},"environment":{"additionalProperties":false,"properties":{"identity_digest":{"pattern":"^[0-9a-f]{64}$","type":"string"},"runner_image_digest":{"anyOf":[{"pattern":"^sha256:[0-9a-f]{64}$","type":"string"},{"type":"null"}]}},"required":["identity_digest"],"type":"object"},"idempotency_key":{"maxLength":128,"minLength":8,"type":"string"},"journey":{"anyOf":[{"items":{"type":"string"},"maxItems":32,"type":"array"},{"type":"null"}]},"policy_version":{"maxLength":128,"minLength":1,"type":"string"},"project_id":{"anyOf":[{"pattern":"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$","type":"string"},{"type":"null"}]},"target":{"additionalProperties":false,"properties":{"artifact_sha256":{"anyOf":[{"pattern":"^[0-9a-f]{64}$","type":"string"},{"type":"null"}]},"identity_digest":{"pattern":"^[0-9a-f]{64}$","type":"string"},"path":{"anyOf":[{"maxLength":2048,"minLength":1,"type":"string"},{"type":"null"}]},"ref":{"anyOf":[{"maxLength":256,"minLength":1,"type":"string"},{"type":"null"}]},"reference":{"maxLength":2048,"minLength":1,"type":"string"},"source_commit":{"anyOf":[{"maxLength":64,"minLength":7,"type":"string"},{"type":"null"}]},"target_type":{"pattern":"^(local|git|archive|container|package|deployment|mcp|sdk|cli)$","type":"string"},"url":{"anyOf":[{"maxLength":2048,"minLength":1,"type":"string"},{"type":"null"}]}},"required":["target_type","identity_digest","reference"],"type":"object"},"tenant_id":{"maxLength":128,"minLength":1,"type":"string"}},"required":["command","environment","idempotency_key","policy_version","target","tenant_id"],"type":"object"}'::jsonb, '{"additionalProperties":true,"type":"object"}'::jsonb), diff --git a/src/echo_certification_forge/app.py b/src/echo_certification_forge/app.py index a5dc8c2..5648b67 100644 --- a/src/echo_certification_forge/app.py +++ b/src/echo_certification_forge/app.py @@ -98,7 +98,7 @@ def _load_certification_environment(runtime: RuntimeProfile | None = None) -> di "runner_image_digest": "sha256:" + environment.runner_image_sha256, "adapter_set_sha256": adapter_sha256, "adapter_execution_profile_sha256": profile_sha256, - "runtime_profile_schema": "certforge.runtime.v1", + "runtime_profile_schema": "certforge.runtime.v2" if runtime.bindings else "certforge.runtime.v1", "runtime_journey_json": json.dumps(runtime.journey), "runtime_journey_mode": "operator_pinned" if runtime.journey is not None else "per_job", } diff --git a/src/echo_certification_forge/operational_telemetry.py b/src/echo_certification_forge/operational_telemetry.py index 6931e3f..05ecd2f 100644 --- a/src/echo_certification_forge/operational_telemetry.py +++ b/src/echo_certification_forge/operational_telemetry.py @@ -10,6 +10,8 @@ import json import sqlite3 +from collections.abc import Iterator +from contextlib import contextmanager from datetime import datetime, timedelta from pathlib import Path from typing import Any, Literal @@ -70,8 +72,17 @@ def _connect(self) -> sqlite3.Connection: connection.execute("PRAGMA synchronous = FULL") return connection + @contextmanager + def _connection(self) -> Iterator[sqlite3.Connection]: + connection = self._connect() + try: + with connection: + yield connection + finally: + connection.close() + def _initialize(self) -> None: - with self._connect() as connection: + with self._connection() as connection: connection.executescript( """ CREATE TABLE IF NOT EXISTS authenticated_operational_reports ( @@ -231,7 +242,7 @@ def _ensure_column( def hydrate_transport_registry(self, trusted: TrustedTransportRegistry) -> None: """Apply durable tenant trust state after static bootstrap keys are loaded.""" - with self._connect() as connection: + with self._connection() as connection: rows = connection.execute( """ SELECT tenant_id, key_id, public_key_pem, state @@ -287,7 +298,7 @@ def begin_key_rotation( } )[:24] old_public_key_pem = trusted.public_key_pem(old_key_id) - with self._connect() as connection: + with self._connection() as connection: active = connection.execute( "SELECT 1 FROM operational_key_rotations WHERE tenant_id = ? AND state = 'OVERLAP'", (tenant_id,), @@ -356,7 +367,7 @@ def begin_key_rotation( } def key_rotation_status(self, tenant_id: str) -> list[dict[str, Any]]: - with self._connect() as connection: + with self._connection() as connection: rows = connection.execute( """ SELECT rotation_id, old_key_id, new_key_id, state, reason, @@ -422,7 +433,7 @@ def finalize_key_rotation( ) -> dict[str, Any]: """Retire the old key only after a new-key-authenticated operational report.""" current = to_utc_iso(now or utc_now()) - with self._connect() as connection: + with self._connection() as connection: row = connection.execute( """ SELECT old_key_id, new_key_id, state, started_at, overlap_expires_at @@ -494,7 +505,7 @@ def reconcile_key_rotations( current = to_utc_iso(now) blocked: list[tuple[str, str]] = [] finalized: list[str] = [] - with self._connect() as connection: + with self._connection() as connection: rows = connection.execute( """ SELECT tenant_id, rotation_id, old_key_id, new_key_id, started_at @@ -557,7 +568,7 @@ def enroll_runner( ) -> dict[str, Any]: """Pin a known authenticated runner to its signing identity and image digest.""" current = to_utc_iso(now or utc_now()) - with self._connect() as connection: + with self._connection() as connection: source = connection.execute( """ SELECT runner_key_id, worker_image_sha256 @@ -617,7 +628,7 @@ def revoke_runner_enrollment( now: datetime | None = None, ) -> dict[str, Any]: current = to_utc_iso(now or utc_now()) - with self._connect() as connection: + with self._connection() as connection: changed = connection.execute( """ UPDATE operational_runner_enrollments @@ -645,7 +656,7 @@ def remediate_adapter_maturity( ) -> dict[str, Any]: """Atomically quarantine every authenticated adapter that is not STABLE.""" current = to_utc_iso(now or utc_now()) - with self._connect() as connection: + with self._connection() as connection: rows = connection.execute( """ SELECT adapter_id, maturity FROM authenticated_adapter_inventory @@ -699,7 +710,7 @@ def quarantine( if subject_type == "runner" else ("authenticated_adapter_inventory", "adapter_id") ) - with self._connect() as connection: + with self._connection() as connection: known = connection.execute( f"SELECT 1 FROM {source_table} WHERE tenant_id = ? AND {source_column} = ?", (tenant_id, subject_id), @@ -763,7 +774,7 @@ def release_quarantine( ) -> dict[str, Any]: """Restore eligibility while retaining the quarantine record and audit history.""" current = to_utc_iso(now or utc_now()) - with self._connect() as connection: + with self._connection() as connection: existing = connection.execute( """ SELECT 1 FROM operational_quarantines @@ -888,7 +899,7 @@ def ingest_worker_heartbeat( received_at = to_utc_iso(current) observed_at = to_utc_iso(response.issued_at) - with self._connect() as connection: + with self._connection() as connection: policy = connection.execute( "SELECT enforced FROM operational_enrollment_policy WHERE tenant_id=?", (response.tenant_id,), @@ -994,7 +1005,7 @@ def ingest_adapter_inventory( received_at = to_utc_iso(current) observed_at = to_utc_iso(response.issued_at) - with self._connect() as connection: + with self._connection() as connection: policy = connection.execute( "SELECT enforced FROM operational_enrollment_policy WHERE tenant_id=?", (claims.tenant_id,), @@ -1094,7 +1105,7 @@ def ingest_windows_package_result( raise OperationalTelemetryError("windows_package_result_body_invalid", 422) from exc received_at = to_utc_iso(current) observed_at = to_utc_iso(response.issued_at) - with self._connect() as connection: + with self._connection() as connection: binding = connection.execute( """ SELECT r.state AS reservation_state, r.target_type, @@ -1207,7 +1218,7 @@ def ingest_windows_package_result( def windows_package_result(self, run_id: str, tenant_id: str) -> dict[str, Any] | None: """Return a verified stored result for the key-holding finalizer.""" - with self._connect() as connection: + with self._connection() as connection: row = connection.execute( """ SELECT runner_id, response_id, body_json, body_sha256, @@ -1229,7 +1240,7 @@ def snapshot( heartbeat_freshness: timedelta = timedelta(minutes=5), ) -> dict[str, Any]: current = now or utc_now() - with self._connect() as connection: + with self._connection() as connection: heartbeats = connection.execute( """ SELECT runner_id, health, capacity_total, capacity_available, diff --git a/src/echo_certification_forge/run_worker.py b/src/echo_certification_forge/run_worker.py index 59edece..eb1a05a 100644 --- a/src/echo_certification_forge/run_worker.py +++ b/src/echo_certification_forge/run_worker.py @@ -19,6 +19,7 @@ import tempfile import threading from collections.abc import Callable +from contextlib import closing from pathlib import Path from cryptography.hazmat.primitives import serialization @@ -104,6 +105,7 @@ def _worker_environment( manifest: RuleManifest | None = None, sandbox: DockerSandbox | None = None, journey: list[str] | None = None, + runtime_binding_sha256: str | None = None, ) -> EnvironmentIdentity: """Compatibility entrypoint for the measured public runtime contract.""" return measured_environment( @@ -111,6 +113,7 @@ def _worker_environment( sandbox=sandbox, journey=journey, adapter_set_sha256=adapter_set_sha256, adapter_execution_profile_sha256=adapter_execution_profile_sha256, + runtime_binding_sha256=runtime_binding_sha256, ) @@ -173,16 +176,27 @@ def run( except ValueError: return {"run_id": run_id, "error": "runtime_operator_profile_mismatch"} try: + if runtime_profile.bindings: + if subscribers is None: + return {"run_id": run_id, "error": "subscriber_governance_required"} + declared = store.get_run(run_id, tenant) + # The launcher supplies only the operator default. Selection comes + # from the persisted tenant/project, never caller image fields. + runtime_profile = runtime_profile.select(tenant, declared.get("project_id"), target_spec) + sandbox = runtime_profile.sandbox runtime_profile.check_execution(sandbox, journey) - except ValueError: - _LOGGER.warning("runtime_profile_rejected", extra={"run_id": run_id, "error_code": "runtime_profile_mismatch"}) - return {"run_id": run_id, "error": "runtime_profile_mismatch"} + except (ValueError, KeyError) as exc: + code = "runtime_binding_target_mismatch" if str(exc) == "runtime_binding_target_mismatch" else "runtime_profile_mismatch" + _LOGGER.warning("runtime_profile_rejected", extra={"run_id": run_id, "error_code": code}) + return {"run_id": run_id, "error": code} + binding = runtime_profile.binding if runtime_profile else None try: # Reject unmeasurable inputs before claiming work or acquiring a target. prepared = _worker_environment( adapter_set_digest(adapter_records) if adapter_records is not None else None, sha256_json(adapter_bundle_response.body) if adapter_bundle_response is not None else None, manifest=manifest, sandbox=sandbox, journey=journey, + runtime_binding_sha256=binding.digest if binding else None, ) if runtime_profile is not None: try: @@ -206,7 +220,7 @@ def run( return {"run_id": run_id, "error": "runtime_identity_unavailable"} if subscribers is None: try: - with sqlite3.connect(store.db_path) as connection: + with closing(sqlite3.connect(store.db_path)) as connection: subscriber_tenant = bool( connection.execute( """ @@ -392,10 +406,15 @@ def run( # not merely the launcher's original selection, before any customer code. if runtime_profile is not None: runtime_profile.check_execution(sandbox_effective, journey) + if binding is not None: + runtime_error = "runtime_binding_source_mismatch" + binding.check_target(target_spec) + binding.check_source(acquired.source_root, acquired_source_commit) runtime_error = "runtime_identity_unavailable" environment = _worker_environment( adapter_digest, adapter_execution_profile_sha256, manifest=manifest, sandbox=sandbox_effective, journey=journey, + runtime_binding_sha256=binding.digest if binding else None, ) except (ValueError, OSError): if heartbeat is not None: diff --git a/src/echo_certification_forge/runtime_admission.py b/src/echo_certification_forge/runtime_admission.py new file mode 100644 index 0000000..a0d9c48 --- /dev/null +++ b/src/echo_certification_forge/runtime_admission.py @@ -0,0 +1,103 @@ +"""Public operator bindings for dependency-complete, source-specific sandboxes. + +This is configuration selection, not image qualification or a signing authority. +Only an independently admitted operator release may install this configuration. +""" +from __future__ import annotations + +import re +from dataclasses import dataclass +from pathlib import Path, PurePosixPath +from urllib.parse import urlsplit + +from .canonical import sha256_bytes, sha256_json +from .sandbox import DockerSandbox + + +@dataclass(frozen=True) +class RuntimeBinding: + binding_id: str + tenant_id: str + project_id: str + repository: str + source_commit: str + dependencies: tuple[tuple[str, str], ...] + sandbox: DockerSandbox + journey: tuple[str, ...] + digest: str + + def check_target(self, target: dict) -> None: + if (target.get("type") != "git" or target.get("url") != self.repository + or target.get("ref") != self.source_commit + or target.get("source_commit", self.source_commit) != self.source_commit): + raise ValueError("runtime_binding_target_mismatch") + + def check_source(self, root: Path, source_commit: str | None) -> None: + if source_commit != self.source_commit: + raise ValueError("runtime_binding_source_mismatch") + root = root.resolve(strict=True) + for name, expected in self.dependencies: + path = root.joinpath(*PurePosixPath(name).parts) + # Reject links even when they resolve within the acquired tree. + if any(p.is_symlink() for p in (path, *path.parents) if p != root and root in p.parents): + raise ValueError("runtime_binding_dependency_mismatch") + try: + if not path.resolve(strict=True).is_relative_to(root) or not path.is_file(): + raise ValueError("runtime_binding_dependency_mismatch") + with path.open("rb") as stream: + content = stream.read(2 * 1024 * 1024 + 1) + if len(content) > 2 * 1024 * 1024 or sha256_bytes(content) != expected: + raise ValueError("runtime_binding_dependency_mismatch") + except OSError as exc: + raise ValueError("runtime_binding_dependency_mismatch") from exc + + +def parse_bindings(raw: object, *, docker: tuple[str, ...], sandbox_fields: frozenset[str], + validate_sandbox, validate_journey) -> tuple[RuntimeBinding, ...]: + if not isinstance(raw, list) or len(raw) > 128: + raise ValueError("runtime_bindings_invalid") + result = [] + identities = set() + names = set() + fields = {"binding_id", "tenant_id", "project_id", "repository", "source_commit", + "dependencies", "sandbox", "journey"} + for item in raw: + if not isinstance(item, dict) or set(item) != fields: + raise ValueError("runtime_binding_fields_invalid") + for key in ("binding_id", "tenant_id", "project_id"): + if not isinstance(item[key], str) or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}", item[key]): + raise ValueError("runtime_binding_identity_invalid") + repo = item["repository"] + if not isinstance(repo, str) or len(repo) > 2048: + raise ValueError("runtime_binding_repository_invalid") + uri = urlsplit(repo) + if uri.scheme != "https" or not uri.hostname or uri.username or uri.password or uri.query or uri.fragment: + raise ValueError("runtime_binding_repository_invalid") + if not isinstance(item["source_commit"], str) or not re.fullmatch(r"[0-9a-f]{40}", item["source_commit"]): + raise ValueError("runtime_binding_commit_invalid") + deps = item["dependencies"] + if not isinstance(deps, dict) or not 1 <= len(deps) <= 32: + raise ValueError("runtime_binding_dependencies_invalid") + for name, digest in deps.items(): + if (not isinstance(name, str) or not name or len(name) > 256 or "\\" in name or ":" in name + or any(ord(c) < 32 for c in name) + or PurePosixPath(name).is_absolute() or any(p in {"", ".", ".."} for p in name.split("/")) + or not isinstance(digest, str) or not re.fullmatch(r"[0-9a-f]{64}", digest)): + raise ValueError("runtime_binding_dependencies_invalid") + config = item["sandbox"] + if not isinstance(config, dict) or "image" not in config or set(config) - sandbox_fields: + raise ValueError("runtime_binding_sandbox_invalid") + sandbox = DockerSandbox(**config, docker=docker) + validate_sandbox(sandbox) + journey = validate_journey(item["journey"]) + if journey is None: + raise ValueError("runtime_binding_journey_required") + identity = (item["tenant_id"], item["project_id"], repo, item["source_commit"]) + if identity in identities or item["binding_id"] in names: + raise ValueError("runtime_binding_duplicate") + identities.add(identity) + names.add(item["binding_id"]) + result.append(RuntimeBinding(item["binding_id"], item["tenant_id"], item["project_id"], repo, + item["source_commit"], tuple(sorted(deps.items())), sandbox, journey, + sha256_json(item))) + return tuple(result) diff --git a/src/echo_certification_forge/runtime_identity.py b/src/echo_certification_forge/runtime_identity.py index 49a0a9d..a5f0ec6 100644 --- a/src/echo_certification_forge/runtime_identity.py +++ b/src/echo_certification_forge/runtime_identity.py @@ -22,6 +22,7 @@ from .models import EnvironmentIdentity from .policy import RuleManifest from .sandbox import DEFAULT_IMAGE, DockerSandbox +from .runtime_admission import RuntimeBinding, parse_bindings _PACKAGE = Path(__file__).resolve().parent _LOG = logging.getLogger(__name__) @@ -80,6 +81,23 @@ def _journey(value: object) -> tuple[str, ...] | None: class RuntimeProfile: sandbox: DockerSandbox journey: tuple[str, ...] | None = None + bindings: tuple[RuntimeBinding, ...] = () + binding: RuntimeBinding | None = None + + def select(self, tenant: str, project: str | None, target: dict) -> RuntimeProfile: + candidates = [b for b in self.bindings if b.tenant_id == tenant and b.project_id == project + and b.repository == target.get("url")] + if not candidates: + return self + for binding in candidates: + try: + binding.check_target(target) + except ValueError: + continue + _LOG.info("runtime_binding_selected", extra={"binding_id": binding.binding_id, + "binding_sha256": binding.digest, "source_commit": binding.source_commit}) + return RuntimeProfile(binding.sandbox, binding.journey, binding=binding) + raise ValueError("runtime_binding_target_mismatch") def check_execution(self, sandbox: DockerSandbox | None, journey: list[str] | None) -> None: selected = _journey(journey) @@ -100,9 +118,12 @@ def load_runtime_profile( raw = json.loads(Path(selected).read_text(encoding="utf-8")) except (OSError, ValueError) as exc: raise ValueError("runtime_profile_unreadable") from exc - if not isinstance(raw, dict) or raw.get("schema_version") != _SCHEMA: + if not isinstance(raw, dict) or raw.get("schema_version") not in {_SCHEMA, "certforge.runtime.v2"}: raise ValueError("runtime_profile_schema_invalid") - if set(raw) - {"schema_version", "sandbox", "journey"}: + allowed = {"schema_version", "sandbox", "journey"} + if raw.get("schema_version") == "certforge.runtime.v2": + allowed.add("project_bindings") + if set(raw) - allowed: raise ValueError("runtime_profile_fields_invalid") config = raw.get("sandbox", {}) if not isinstance(config, dict) or set(config) - _FIELDS: @@ -115,7 +136,10 @@ def load_runtime_profile( command = docker or tuple(os.environ.get("ECHO_CERTFORGE_SANDBOX_DOCKER", "docker").split()) sandbox = DockerSandbox(**config, docker=command) _sandbox_inputs(sandbox) - return RuntimeProfile(sandbox, _journey(raw.get("journey"))) + bindings = parse_bindings(raw.get("project_bindings", []), docker=command, + sandbox_fields=_FIELDS, validate_sandbox=_sandbox_inputs, + validate_journey=_journey) + return RuntimeProfile(sandbox, _journey(raw.get("journey")), bindings) def _harness_digest(source_root: Path | None = None) -> str: @@ -144,8 +168,13 @@ def measured_environment( adapter_set_sha256: str | None = None, adapter_execution_profile_sha256: str | None = None, source_root: Path | None = None, + runtime_binding_sha256: str | None = None, ) -> EnvironmentIdentity: config = _sandbox_inputs(sandbox) + if runtime_binding_sha256 is not None: + if not re.fullmatch(r"[0-9a-f]{64}", runtime_binding_sha256): + raise ValueError("runtime_binding_digest_invalid") + config = dict(config, operator_binding_sha256=runtime_binding_sha256) plan = _journey(journey) try: executable_sha256 = sha256_bytes(Path(sys.executable).read_bytes()) diff --git a/src/echo_certification_forge/service.py b/src/echo_certification_forge/service.py index 425536b..8805605 100644 --- a/src/echo_certification_forge/service.py +++ b/src/echo_certification_forge/service.py @@ -116,6 +116,8 @@ class DeployGateRequest(BaseModel): class RuntimeEnvironmentRequest(BaseModel): model_config = ConfigDict(extra="forbid") journey: list[JourneyArgument] | None = Field(default=None, min_length=1, max_length=32) + project_id: str | None = Field(default=None, pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") + target: SubmitTarget | None = None @field_validator("journey") @classmethod @@ -541,21 +543,29 @@ def resolve_runtime_environment( x_tenant_id: str | None = Header(default=None), authorization: str | None = Header(default=None), ) -> RuntimeEnvironmentResponse: - subscriber_principal(x_tenant_id, authorization, Permission.RUN_READ, "runtime.environment.resolve") + principal = subscriber_principal(x_tenant_id, authorization, Permission.RUN_READ, "runtime.environment.resolve") profile = context.runtime_profile public = context.certification_environment if profile is None or public is None: raise HTTPException(status_code=503, detail="runtime_profile_unavailable") + if (request.project_id is None) != (request.target is None): + raise HTTPException(status_code=422, detail="runtime_project_target_pair_required") try: + if request.project_id is not None: + context.subscribers.get_project(principal, request.project_id, required_status="ACTIVE") + profile = profile.select(principal.organization_id, request.project_id, request.target.worker_spec()) profile.check_execution(profile.sandbox, request.journey) environment = measured_environment( manifest=context.manifest, sandbox=profile.sandbox, journey=request.journey, adapter_set_sha256=public.get("adapter_set_sha256"), adapter_execution_profile_sha256=public.get("adapter_execution_profile_sha256"), + runtime_binding_sha256=profile.binding.digest if profile.binding else None, ) + except SubscriberError as exc: + raise HTTPException(status_code=exc.status_code, detail=exc.code) from exc except ValueError as exc: code = str(exc) - if code in {"runtime_profile_mismatch", "runtime_journey_invalid"}: + if code in {"runtime_profile_mismatch", "runtime_journey_invalid", "runtime_binding_target_mismatch"}: raise HTTPException(status_code=422, detail=code) from exc raise HTTPException(status_code=503, detail="runtime_identity_unavailable") from exc return RuntimeEnvironmentResponse( diff --git a/src/echo_certification_forge/subscriber.py b/src/echo_certification_forge/subscriber.py index 4fb28ba..31aeefb 100644 --- a/src/echo_certification_forge/subscriber.py +++ b/src/echo_certification_forge/subscriber.py @@ -4131,7 +4131,7 @@ def claim_dispatch( self._recover_expired_claims(connection, organization_id, now_value) self._recover_expired_dispatches(connection, now_value) filters = ["d.state = 'PENDING'", "r.state = 'BOUND'", "runs.state = 'QUEUED'", - "(d.last_error IS NULL OR d.last_error NOT IN ('runtime_profile_mismatch', 'runtime_environment_mismatch') OR d.updated_at <= ?)"] + "(d.last_error IS NULL OR d.last_error NOT IN ('runtime_profile_mismatch', 'runtime_environment_mismatch', 'runtime_binding_target_mismatch') OR d.updated_at <= ?)"] parameters: list[Any] = [to_utc_iso(now_value - timedelta(seconds=60))] if run_id is not None: filters.append("d.run_id = ?") diff --git a/tests/test_intake_recovery_isolation.py b/tests/test_intake_recovery_isolation.py index 1668d1a..1129968 100644 --- a/tests/test_intake_recovery_isolation.py +++ b/tests/test_intake_recovery_isolation.py @@ -196,6 +196,25 @@ def test_tenant_runtime_mismatch_delays_retries_and_other_tenant_progresses(tmp_ assert db.execute('SELECT count(*) FROM subscriber_usage_events WHERE organization_id=? AND quantity<0', (org.organization_id,)).fetchone() == (1,) +def test_unadmitted_project_source_delays_retry_and_preserves_other_tenant(tmp_path, manifest): + from echo_certification_forge.dispatch_worker import dispatch_once + from echo_certification_forge.signing import Ed25519VerdictSigner + + store, governance, client = _stack(tmp_path, manifest) + org, request, _ = reserve(governance, client, manifest, tmp_path, 'unadmitted-source') + other, _, _ = reserve(governance, client, manifest, tmp_path, 'unaffected-tenant') + _, result = dispatch_once(governance, + lambda *args, **kwargs: {'error': 'runtime_binding_target_mismatch'}, + dispatcher_id='binding-rejection', organization_id=org.organization_id, + worker_options={'store': store, 'manifest': manifest, 'subscribers': governance, + 'signer': Ed25519VerdictSigner.generate()}) + assert not result.get('dispatcher_halted') + assert governance.dispatch_status(request.run_id(manifest.digest), org.organization_id)['attempts'] == 1 + if governance.claim_dispatch('immediate-binding-retry', organization_id=org.organization_id) is not None: + pytest.fail('unadmitted source was reclaimed without the existing tenant retry delay') + assert governance.claim_dispatch('unaffected-progress', organization_id=other.organization_id) is not None + + def test_post_claim_measurement_failure_halts_cleanly_without_reopening_terminal_run(tmp_path, manifest, monkeypatch): from echo_certification_forge import run_worker from echo_certification_forge.dispatch_worker import dispatch_once diff --git a/tests/test_operational_telemetry_connections.py b/tests/test_operational_telemetry_connections.py new file mode 100644 index 0000000..0b99330 --- /dev/null +++ b/tests/test_operational_telemetry_connections.py @@ -0,0 +1,73 @@ +"""Connection ownership controls use real temporary SQLite files, never live telemetry.""" +import sqlite3 +from contextlib import closing + +import pytest + +from echo_certification_forge.operational_telemetry import OperationalTelemetryRegistry + + +@pytest.mark.parametrize("fail_initialization", [False, True]) +def test_registry_initialization_closes_owned_connection(tmp_path, monkeypatch, fail_initialization): + connect = sqlite3.connect + opened = [] + + class ObservedConnection(sqlite3.Connection): + def executescript(self, script, /): + if fail_initialization: + raise sqlite3.OperationalError("synthetic initialization failure") + return super().executescript(script) + + def tracked_connect(*args, **kwargs): + connection = connect(*args, **kwargs, factory=ObservedConnection) + opened.append(connection) + return connection + + monkeypatch.setattr(sqlite3, "connect", tracked_connect) + try: + if fail_initialization: + with pytest.raises(sqlite3.OperationalError, match="synthetic"): + OperationalTelemetryRegistry(tmp_path / "owned.sqlite3") + else: + OperationalTelemetryRegistry(tmp_path / "owned.sqlite3") + assert len(opened) == 1 + with pytest.raises(sqlite3.ProgrammingError, match="closed"): + opened[0].execute("SELECT 1") + finally: + for connection in opened: + connection.close() + + +@pytest.mark.parametrize("fail_transaction", [False, True]) +def test_registry_connection_preserves_commit_and_rollback(tmp_path, monkeypatch, fail_transaction): + path = tmp_path / "transaction.sqlite3" + registry = OperationalTelemetryRegistry(path) + with closing(sqlite3.connect(path)) as connection: + connection.execute("CREATE TABLE owned_connection_control (value INTEGER)") + connection.commit() + connect = registry._connect + opened = [] + + def tracked_connect(): + connection = connect() + opened.append(connection) + return connection + + monkeypatch.setattr(registry, "_connect", tracked_connect) + try: + if fail_transaction: + with pytest.raises(RuntimeError, match="synthetic rollback"): + with registry._connection() as connection: + connection.execute("INSERT INTO owned_connection_control VALUES (1)") + raise RuntimeError("synthetic rollback") + else: + with registry._connection() as connection: + connection.execute("INSERT INTO owned_connection_control VALUES (1)") + assert len(opened) == 1 + with pytest.raises(sqlite3.ProgrammingError, match="closed"): + opened[0].execute("SELECT 1") + with closing(sqlite3.connect(path)) as connection: + assert connection.execute("SELECT COUNT(*) FROM owned_connection_control").fetchone()[0] == (0 if fail_transaction else 1) + finally: + for connection in opened: + connection.close() diff --git a/tests/test_runtime_admission.py b/tests/test_runtime_admission.py new file mode 100644 index 0000000..21a82f2 --- /dev/null +++ b/tests/test_runtime_admission.py @@ -0,0 +1,199 @@ +import copy +import dataclasses +import json + +import pytest + +from echo_certification_forge.canonical import sha256_bytes +from echo_certification_forge.runtime_identity import load_runtime_profile, measured_environment +from echo_certification_forge.sandbox import DEFAULT_IMAGE, DockerSandbox, SandboxResult +from echo_certification_forge.signing import Ed25519VerdictSigner +from test_runtime_job_binding import runtime_api + +SHA = "a" * 40 +REPO = "https://github.com/example/fielddesk.git" +IMAGE = "sha256:" + "e" * 64 +JOURNEY = ["python3", "-B", "scripts/certforge_journey.py"] + + +def config(tenant="tenant-a", project="project-a"): + return {"schema_version": "certforge.runtime.v2", "project_bindings": [{ + "binding_id": "fielddesk-exact", "tenant_id": tenant, "project_id": project, + "repository": REPO, "source_commit": SHA, + "dependencies": {"package-lock.json": sha256_bytes(b"locked dependencies\n")}, + "sandbox": {"image": IMAGE, "memory": "2g", "timeout_s": 120}, "journey": JOURNEY, + }]} + + +def profile(tmp_path, value): + path = tmp_path / "operator.json" + path.write_text(json.dumps(value), encoding="utf-8") + return load_runtime_profile(path) + + +def spec(**changes): + return dict({"type": "git", "url": REPO, "ref": SHA, "source_commit": SHA}, **changes) + + +def test_operator_selects_exact_source_and_keeps_default_isolation(tmp_path, manifest): + catalog = profile(tmp_path, config()) + selected = catalog.select("tenant-a", "project-a", spec()) + assert selected.sandbox.image == IMAGE + assert selected.sandbox.memory == "2g" + selected.check_execution(selected.sandbox, JOURNEY) + assert catalog.sandbox == DockerSandbox() + assert catalog.sandbox.image == DEFAULT_IMAGE + assert catalog.select("tenant-b", "project-a", spec()) is catalog + assert catalog.select("tenant-a", "project-b", spec()) is catalog + assert catalog.select("tenant-a", "project-a", spec(url="https://github.com/example/other.git")) is catalog + base = measured_environment(manifest=manifest, sandbox=selected.sandbox, journey=JOURNEY) + bound = measured_environment(manifest=manifest, sandbox=selected.sandbox, journey=JOURNEY, + runtime_binding_sha256=selected.binding.digest) + assert base.identity_digest != bound.identity_digest + changed = config() + changed["project_bindings"][0]["dependencies"]["package-lock.json"] = "f" * 64 + replacement = profile(tmp_path, changed).select("tenant-a", "project-a", spec()) + assert selected.binding.digest != replacement.binding.digest + + +@pytest.mark.parametrize("change", [{"ref": "main"}, {"source_commit": "b" * 40}, {"type": "local"}]) +def test_configured_project_rejects_unadmitted_source(tmp_path, change): + with pytest.raises(ValueError, match="runtime_binding_target_mismatch"): + profile(tmp_path, config()).select("tenant-a", "project-a", spec(**change)) + + +@pytest.mark.parametrize("change", [ + {"sandbox": {"image": "python:latest"}}, {"sandbox": {"image": IMAGE, "user": "0:0"}}, + {"sandbox": {"image": IMAGE, "network": "host"}}, {"sandbox": {"image": IMAGE, "extra_env": {}}}, + {"dependencies": {"../outside": "a" * 64}}, {"dependencies": {"C:/outside": "a" * 64}}, + {"dependencies": {}}, {"journey": None}, {"source_commit": "main"}, + {"dependencies": {"bad\u0000path": "a" * 64}}, + {"repository": "https://user:password@example.test/repo"}, {"unexpected": True}, +]) +def test_operator_binding_rejects_unsafe_or_incomplete_input(tmp_path, change): + value = config() + value["project_bindings"][0].update(change) + with pytest.raises(ValueError): + profile(tmp_path, value) + + +def test_duplicate_admission_and_v1_extension_rejected(tmp_path): + value = config() + value["project_bindings"].append(copy.deepcopy(value["project_bindings"][0])) + with pytest.raises(ValueError, match="runtime_binding_duplicate"): + profile(tmp_path, value) + value = config() + value["schema_version"] = "certforge.runtime.v1" + with pytest.raises(ValueError, match="runtime_profile_fields_invalid"): + profile(tmp_path, value) + + +def test_bound_catalog_preserves_operator_mismatch_before_tenant_selection(tmp_path, store, manifest, monkeypatch): + from echo_certification_forge import run_worker + + catalog = profile(tmp_path, config()) + monkeypatch.setattr(run_worker, "acquire_target", lambda *args: pytest.fail("operator drift must not acquire")) + result = run_worker.run("not-created", "tenant-a", spec(), store=store, + manifest=manifest, signer=Ed25519VerdictSigner.generate(), + sandbox=dataclasses.replace(catalog.sandbox, memory="7g"), + runtime_profile=catalog, journey=JOURNEY) + assert result == {"run_id": "not-created", "error": "runtime_operator_profile_mismatch"} + + +def test_subscriber_governance_guard_closes_its_real_sqlite_read(runtime_api, monkeypatch): + import sqlite3 + from echo_certification_forge import run_worker + + _client, context, headers = runtime_api + connect = sqlite3.connect + opened = [] + + def tracked_connect(*args, **kwargs): + connection = connect(*args, **kwargs) + opened.append(connection) + return connection + + monkeypatch.setattr(run_worker.sqlite3, "connect", tracked_connect) + try: + result = run_worker.run("governance-close-guard", headers["X-Tenant-ID"], spec(), + store=context.store, manifest=context.manifest, signer=Ed25519VerdictSigner.generate()) + assert result["error"] == "subscriber_governance_required" + assert len(opened) == 1 + with pytest.raises(sqlite3.ProgrammingError, match="closed"): + opened[0].execute("SELECT 1") + finally: + for connection in opened: + connection.close() + + +@pytest.mark.parametrize("content", [b"changed", b"x" * (2 * 1024 * 1024 + 1), None], ids=["changed", "too-large", "missing"]) +def test_dependency_mismatch_or_missing_file_denied(tmp_path, content): + selected = profile(tmp_path, config()).select("tenant-a", "project-a", spec()) + if content is not None: + (tmp_path / "package-lock.json").write_bytes(content) + with pytest.raises(ValueError, match="runtime_binding_dependency_mismatch"): + selected.binding.check_source(tmp_path, SHA) + + +@pytest.mark.parametrize("tampered", [False, True]) +def test_api_and_worker_select_same_profile_and_fence_dependencies(runtime_api, tmp_path, monkeypatch, tampered): + from echo_certification_forge import run_worker + from echo_certification_forge.acquisition import acquire_target + from echo_certification_forge.models import TargetIdentity, declared_target_identity_digest + + client, context, headers = runtime_api + context.certification_environment = {} + source = tmp_path / "source" + source.mkdir() + (source / "package-lock.json").write_bytes(b"changed" if tampered else b"locked dependencies\n") + acquired = dataclasses.replace(acquire_target({"type": "local", "path": str(source)}, tmp_path / "unused"), + target_type="git", canonical_ref=REPO + "@" + SHA, source_commit=SHA) + target = TargetIdentity(headers["X-Tenant-ID"], "git", acquired.canonical_ref, acquired.artifact_sha256, source_commit=SHA) + project = client.post("/v1/subscriber/projects", headers=headers, json={ + "slug": "fielddesk", "name": "FieldDesk fixture", "target_reference": target.canonical_ref, + }).json()["project_id"] + context.runtime_profile = profile(tmp_path, config(headers["X-Tenant-ID"], project)) + target_body = {"target_type": "git", "reference": target.canonical_ref, "url": REPO, "ref": SHA, + "source_commit": SHA, "identity_digest": declared_target_identity_digest( + headers["X-Tenant-ID"], "git", None, SHA, target.canonical_ref)} + body = {"journey": JOURNEY, "project_id": project, "target": target_body} + response = client.post("/v1/subscriber/runtime/environment", headers=headers, json=body) + assert response.status_code == 200, response.text + declaration = response.json()["submit_environment"] + assert declaration["runner_image_digest"] == IMAGE + wrong = copy.deepcopy(body) + wrong["target"]["ref"] = "b" * 40 + assert client.post("/v1/subscriber/runtime/environment", headers=headers, json=wrong).status_code == 422 + wrong = dict(body, project_id="other-project") + assert client.post("/v1/subscriber/runtime/environment", headers=headers, json=wrong).status_code == 404 + assert not context.store.list_runs(headers["X-Tenant-ID"]) + submitted = client.post("/v1/certifications", headers=headers, json={ + "tenant_id": headers["X-Tenant-ID"], "project_id": project, "target": target_body, + "environment": declaration, "policy_version": context.manifest.manifest_id, + "idempotency_key": "bound-fielddesk-fixture", "journey": JOURNEY, + }) + assert submitted.status_code == 201, submitted.text + monkeypatch.setattr(run_worker, "acquire_target", lambda *args: acquired) + executions = [] + monkeypatch.setattr(DockerSandbox, "run", lambda self, argv, *args: + executions.append((self.image, self.memory, argv)) or SandboxResult(0, "fixture", "", False)) + result = run_worker.run(submitted.json()["run_id"], headers["X-Tenant-ID"], spec(), + store=context.store, manifest=context.manifest, signer=Ed25519VerdictSigner.generate(), + subscribers=context.subscribers, sandbox=context.runtime_profile.sandbox, + runtime_profile=context.runtime_profile, journey=JOURNEY) + if tampered: + assert result["error"] == "runtime_binding_source_mismatch" + assert not executions + else: + assert "error" not in result, result + assert result["environment_identity_digest"] == declaration["identity_digest"] + assert result["release_verdict"] == "NOT_READY" # Fixture transport cannot certify production. + assert executions == [(IMAGE, "2g", JOURNEY)] + + +@pytest.mark.parametrize("extra", [{"project_id": "project-a"}, {"target": { + "target_type": "git", "reference": REPO, "identity_digest": "a" * 64}}, {"profile": "fielddesk"}]) +def test_resolution_rejects_partial_binding_or_caller_profile(runtime_api, extra): + client, _context, headers = runtime_api + assert client.post("/v1/subscriber/runtime/environment", headers=headers, + json=dict({"journey": JOURNEY}, **extra)).status_code == 422