From 5359617a4b9aeef1bebc2a3c8cbf04df6b24b961 Mon Sep 17 00:00:00 2001 From: Peter van Tol Date: Wed, 14 Oct 2026 10:33:55 +0200 Subject: [PATCH 1/5] Add bounded Workspace project discovery read contract --- .../WORKSPACE_PROJECT_DISCOVERY_READ_V1.md | 137 +++++ package-lock.json | 4 +- package.json | 2 +- pyproject.toml | 2 +- .../ENGINEERING_PLATFORM_CONFIG.json | 2 +- .../ENGINEERING_PLATFORM_VERSION.json | 8 +- .../central_operational_reset.py | 3 +- src/engineering_platform/platform_version.py | 2 +- src/engineering_platform/project_discovery.py | 385 ++++++++++++++ ...workspace-project-discovery-v1.schema.json | 320 ++++++++++++ src/engineering_platform/server.py | 103 +++- .../templates/workspace-config.json | 2 +- ...erver-http-surface.postman_collection.json | 71 +++ .../http-json-api.postman_collection.json | 71 +++ tests/engineering/test_project_discovery.py | 480 ++++++++++++++++++ tests/engineering/test_server_foundation.py | 4 +- 16 files changed, 1578 insertions(+), 18 deletions(-) create mode 100644 docs/engineering/WORKSPACE_PROJECT_DISCOVERY_READ_V1.md create mode 100644 src/engineering_platform/project_discovery.py create mode 100644 src/engineering_platform/schemas/workspace-project-discovery-v1.schema.json create mode 100644 tests/engineering/test_project_discovery.py diff --git a/docs/engineering/WORKSPACE_PROJECT_DISCOVERY_READ_V1.md b/docs/engineering/WORKSPACE_PROJECT_DISCOVERY_READ_V1.md new file mode 100644 index 00000000..d97bdf4b --- /dev/null +++ b/docs/engineering/WORKSPACE_PROJECT_DISCOVERY_READ_V1.md @@ -0,0 +1,137 @@ +# Workspace project discovery — HOB-EP01 + +Assignment: `L2-WORKSPACE-PROJECT-DISCOVERY-READ-V1-20261012`. +JOIN: `L2-L4-HOB-EP01-VERIFIED-LINK-V1-20261012`. + +**Implementation candidate, not yet a qualified handoff.** Schema75 adds one +separate security ledger. Existing B8R project/repository/attachment rows remain +the sole logical/physical topology; no new project register or identity exists. +Contract version1.0 is additive and does not widen existing credentials. + +## HTTP and explicit consumer mapping + +- `GET /v1/onboarding/projects?limit=25&cursor=OPAQUE`. +- `GET /v1/onboarding/projects/{project_id}/topology?expected_topology_revision=DIGEST`. + +The packaged closed response schema is +`schemas/workspace-project-discovery-v1.schema.json`. `DiscoverySnapshot` +contains only permitted project IDs and their explicit canonical authority +repository IDs. `TopologySnapshot` adds that project's existing authority and +children and, if granted, its bounded physical attachment availability. +Multiple attachments do not create multiple logical repositories. No absolute +paths, remote-derived identities, agent host metadata or secrets are returned. +Unavailable/stale/revoked Agents retain their logical repository identity. + +The Workspace proposal at `a6aad520bed9f1c2f7119b393435b8c739f030b3` was design +input, not an existing EP API. Its route meanings are retained. Its proposed +integer revision becomes the owner's **64-character SHA256 string** in both +`topology_revision` and `expected_topology_revision`; never truncate/cast it to +an integer. `snapshot_id` equals that digest. The owner response's explicit +`instance_id`, `producer`, `scope`, and `observed_at` map to Workspace's own +owner/scope/freshness model. Producer source/tree/wheel binding is supplied by +the eventual immutable qualified handoff; a logical snapshot digest is not a +source-commit or artifact digest. Do not synthesize source-vector evidence. + +## Independent finite READ authority + +Required on **both** routes: + +``` +Authorization: Bearer +EP-Discovery-Contract: 1.0 +EP-Instance-ID: +EP-Consumer-ID: +Workspace-Instance-ID: +EP-Audience: workspace +``` + +All identity headers must match the actual bearer-derived grant. The audience +is `workspace`; consumer and Workspace-instance binding are separate checks. +Project scope is an explicit list of0–100 already registered active B8R +projects, never a wildcard. An explicit empty scope is healthy and grants +nothing. Identity fields are mandatory; `attachment_availability` is optional. +For an ungranted attachment field, `attachments:null` and +`attachment_visibility:NOT_GRANTED` must not be presented as no attachments. +Lifetime is finite1–86400 seconds, default900. The verifier uses a separate +domain/table; Console/admin/submission/installation-pairing bearers are not +accepted. There is no remote owner-issuance HTTP route or forwarded admin token. + +Supported ordinary local installation-owner CLI: + +``` +engineering-platform-server discovery-grant-issue --data-root ROOT \ + --expected-instance-id EP_INSTANCE --discovery-grant-id GRANT \ + --consumer-id WORKSPACE_CONSUMER --workspace-instance-id WORKSPACE_INSTANCE \ + --discovery-project-id EXISTING_PROJECT \ + --discovery-field project_identity --discovery-field repository_identity \ + --discovery-field attachment_availability \ + --discovery-ttl-seconds 900 --operation-id ISSUE_OPERATION +engineering-platform-server discovery-grant-status --data-root ROOT \ + --expected-instance-id EP_INSTANCE --discovery-grant-id GRANT +engineering-platform-server discovery-grant-revoke --data-root ROOT \ + --expected-instance-id EP_INSTANCE --discovery-grant-id GRANT \ + --operation-id REVOKE_OPERATION +``` + +Repeat `--discovery-project-id` only for explicitly permitted projects. The +issue operation discloses plaintext once; replay does not mint or redisclose. +Status/revocation contain no token/verifier. Exact revocation replay is +idempotent; a different operation conflicts. These are owner ledger operations, +not project registration or runtime execution. Never publish their plaintext +output in a register or handoff; store it only in the consumer's protected +server credential channel. No real-user grants are part of qualification. + +## Read boundaries, pagination and typed failures + +GET bodies, duplicate/unknown parameters and unsupported contract versions are +rejected. Page size defaults25, bounded1–100. A cursor is opaque, authenticated +by a separate random server-private key stored in the grant ledger and bound +to that scope, snapshot, offset and page +size. It cannot confer rights or be reused after expiry/revocation. A topology +change between discovery/selection/detail requires rediscovery; no mixed old +identity/new topology is published. Maximum100 repositories and100 physical +attachments per selected project; overflow is a typed conflict, not silent +truncation or a healthy empty projection. + +Reads use the existing noncreating/query-only CENTRAL primitive. A first +snapshot is assembled and a separate fresh snapshot rechecks authorization, +expiry, source identity and data before publication. The final read transaction +holds its shared publication fence: owner revocation cannot commit during the +protected send. Source disappearance, replacement and committed revocation +before that boundary cannot publish the old healthy snapshot. No read calls +initialize/migrate/register/attach/scan/submit or start a Mission/Action/provider. +All responses use `Cache-Control:no-store`. + +The publication fence requires an existing rollback-journal store (DELETE, +TRUNCATE or PERSIST). WAL and other journal modes are rejected with422 +`DISCOVERY_JOURNAL_UNSUPPORTED`; a discovery read never switches journal mode. +Source identity and finite authority are checked again after response encoding, +immediately before HTTP headers, while the protected read transaction remains +open. The check binds that publication decision; no guarantee is made against +external filesystem changes after headers have already been published. + +HTTP401 means absent/invalid/expired/revoked READ authority;403 means wrong +consumer/audience/instance/project or unavailable permitted project;409 means +snapshot/cursor-topology conflict or incomplete/over-limit topology;422 means +unsupported contract/schema. Malformed query/body/cursor is400. Real source +loss/IO/unavailability is503; corruption is500. Neither is successful empty +data. Original errors remain in private server evidence; responses do not +expose paths or raw tracebacks. + +Workspace must validate the closed schema, exact qualified producer and +instance, current own actor/request epoch, scope and expiry before applying a +late response. On conflict rediscover; on revoked/unavailable mark linking +unavailable. Never reconstruct identity or fall back to Console, SQL, CLI, +submission bearer, a fixture or a cached prior positive response. + +## Qualification boundary + +First qualify a small actual installed discovery→selected-topology chain on +own synthetic B8R registrations and finite owner-issued READ grants, including +the frozen acceptance matrix. Then existing owning tests/strict coverage, +independent Quality/Security, protected exact-head source, ordinary nonempty +Finalization, new exact-final-main installed HTTP/readback and safe own cleanup. +The immutable final source/tree/wheel/schema/issuer binding and actual results +will be supplied under the JOIN. R34 remains closed and supplies no discovery +PASS. TDE is reported separately. No L1/L3 or complete Workspace GUI prerequisite +is added, and no live EP, signing, new writer or wider HOB family is selected. diff --git a/package-lock.json b/package-lock.json index d718c1b9..2650acd7 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "engineering-platform-browser-validation", - "version": "2.3.119", + "version": "2.3.120", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "engineering-platform-browser-validation", - "version": "2.3.119", + "version": "2.3.120", "devDependencies": { "@openai/codex": "0.160.1", "@playwright/test": "1.63.0", diff --git a/package.json b/package.json index 90e99bbb..a96256fe 100644 --- a/package.json +++ b/package.json @@ -1,7 +1,7 @@ { "name": "engineering-platform-browser-validation", "private": true, - "version": "2.3.119", + "version": "2.3.120", "scripts": { "test:engineering-dashboard": "PYTHONPATH=src python3 -m engineering_platform.dashboard_browser_validation", "test:engineering-dashboard-logic": "node --test tests/engineering/dashboard_status_store.test.mjs tests/engineering/ui_localization_contract.test.mjs tests/engineering/dashboard_translation.test.mjs tests/engineering/dashboard_js_coverage.test.mjs tests/engineering/dashboard_worker_logs.test.mjs tests/engineering/dashboard_log_read.test.mjs", diff --git a/pyproject.toml b/pyproject.toml index 0262d4ce..4b486675 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "engineering-platform" -version = "2.3.119" +version = "2.3.120" description = "Local-first Engineering Platform execution operations runtime" readme = "README.md" requires-python = ">=3.14,<3.15" diff --git a/src/engineering_platform/ENGINEERING_PLATFORM_CONFIG.json b/src/engineering_platform/ENGINEERING_PLATFORM_CONFIG.json index acc3371e..f4b1cfe2 100644 --- a/src/engineering_platform/ENGINEERING_PLATFORM_CONFIG.json +++ b/src/engineering_platform/ENGINEERING_PLATFORM_CONFIG.json @@ -3,7 +3,7 @@ "platform": { "id": "engineering-platform", "name": "Engineering Platform", - "version": "2.3.119", + "version": "2.3.120", "generation": 2, "documentation_namespace": "engineering-platform", "capability_registry_version": 1 diff --git a/src/engineering_platform/ENGINEERING_PLATFORM_VERSION.json b/src/engineering_platform/ENGINEERING_PLATFORM_VERSION.json index c47e5cb5..9af6ea61 100644 --- a/src/engineering_platform/ENGINEERING_PLATFORM_VERSION.json +++ b/src/engineering_platform/ENGINEERING_PLATFORM_VERSION.json @@ -1,15 +1,15 @@ { "bootstrap_contract": "2026.12", "checkpoint_format": 1, - "dashboard_version": "2.3.119", + "dashboard_version": "2.3.120", "handoff_protocol": 1, "memory_format": 2, "minimum_codex_cli": "0.146.0", "inbox_protocol": 1, - "platform_version": "2.3.119", + "platform_version": "2.3.120", "report_format": 2, - "runner_version": "2.3.119", + "runner_version": "2.3.120", "status_model": 1, "storage_schema": 45, - "watcher_version": "2.3.119" + "watcher_version": "2.3.120" } diff --git a/src/engineering_platform/central_operational_reset.py b/src/engineering_platform/central_operational_reset.py index 41ded484..dab72135 100644 --- a/src/engineering_platform/central_operational_reset.py +++ b/src/engineering_platform/central_operational_reset.py @@ -36,7 +36,7 @@ PROFILE = "EP_CENTRAL_OPERATIONAL_HISTORY_V1" PLAN_VERSION = 2 -SCHEMA_VERSION = 74 +SCHEMA_VERSION = 75 _OPERATION = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{7,127}") _INSTANCE_ID = re.compile( r"[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}" @@ -70,6 +70,7 @@ "execution_migration_provenance", }) SECURITY_AND_AUTHORITY_LEDGER = frozenset({ + "ep_workspace_discovery_grants", "ep_agent_pairing_codes", "ep_consumer_credential_recovery_operations", "ep_consumer_credentials", "ep_consumer_registrations", "ep_installation_pairing_credentials", "ep_control_provenance", "ep_external_producer_binding_audit", diff --git a/src/engineering_platform/platform_version.py b/src/engineering_platform/platform_version.py index ea3f97f8..5b4b4677 100644 --- a/src/engineering_platform/platform_version.py +++ b/src/engineering_platform/platform_version.py @@ -12,7 +12,7 @@ SEMVER = re.compile(r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)$") CONTRACT = re.compile(r"^(\d{4})\.(0[1-9]|1[0-2])$") -CURRENT_PLATFORM_VERSION = "2.3.119" +CURRENT_PLATFORM_VERSION = "2.3.120" MANIFEST_FIELDS = frozenset( { "platform_version", diff --git a/src/engineering_platform/project_discovery.py b/src/engineering_platform/project_discovery.py new file mode 100644 index 00000000..df1395bc --- /dev/null +++ b/src/engineering_platform/project_discovery.py @@ -0,0 +1,385 @@ +"""Finite Workspace discovery reads of the existing B8R/CENTRAL authority. + +Issuance/revocation are installation-owner operations. HTTP reads never open a +creating connection, migrate a store, register topology, or execute work. +""" +from __future__ import annotations + +import base64 +from datetime import datetime, timedelta, timezone +import hashlib +import hmac +import json +from pathlib import Path +import re +import secrets +import sqlite3 +from typing import Callable + +from . import storage +from .platform_version import CURRENT_PLATFORM_VERSION + +CONTRACT_VERSION = "1.0" +AUDIENCE = "workspace" +SCHEMA_VERSION = 75 +TABLE = "ep_workspace_discovery_grants" +FIELDS = frozenset({"project_identity", "repository_identity", "attachment_availability"}) +IDENTITY_FIELDS = frozenset({"project_identity", "repository_identity"}) +_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}\Z") +_DIGEST = re.compile(r"[a-f0-9]{64}\Z") +_DOMAIN = b"engineering-platform.workspace-discovery.verifier.v1\0" + + +class DiscoveryError(ValueError): + def __init__(self, code: str, status: int = 400): + super().__init__(code) + self.code, self.status = code, status + + +def _id(value: object) -> str: + if not isinstance(value, str) or _ID.fullmatch(value) is None: + raise DiscoveryError("DISCOVERY_ID_INVALID") + return value + + +def _canonical(value: object) -> bytes: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("ascii") + + +def _now() -> datetime: + return datetime.now(timezone.utc) + + +def _ids(values: object) -> list[str]: + if not isinstance(values, list) or len(values) > 100: + raise DiscoveryError("DISCOVERY_PROJECT_SCOPE_INVALID") + result = [_id(v) for v in values] + if len(set(result)) != len(result): + raise DiscoveryError("DISCOVERY_PROJECT_SCOPE_INVALID") + return sorted(result) + + +def _fields(values: object) -> list[str]: + if not isinstance(values, list) or any(not isinstance(v, str) for v in values): + raise DiscoveryError("DISCOVERY_FIELD_SCOPE_INVALID") + if len(values) != len(set(values)) or not IDENTITY_FIELDS <= set(values) <= FIELDS: + raise DiscoveryError("DISCOVERY_FIELD_SCOPE_INVALID") + return sorted(values) + + +def install_schema(connection: sqlite3.Connection) -> None: + connection.execute("""CREATE TABLE ep_workspace_discovery_grants ( + grant_id TEXT PRIMARY KEY, instance_id TEXT NOT NULL REFERENCES ep_installations(instance_id), + consumer_id TEXT NOT NULL, workspace_instance_id TEXT NOT NULL, + audience TEXT NOT NULL CHECK(audience='workspace'), project_ids TEXT NOT NULL, + fields TEXT NOT NULL, issuance_operation_id TEXT NOT NULL UNIQUE, + verifier BLOB NOT NULL UNIQUE CHECK(length(verifier)=32), + cursor_key BLOB NOT NULL CHECK(length(cursor_key)=32), + issued_at TEXT NOT NULL, expires_at TEXT NOT NULL, + revoked_at TEXT, revocation_operation_id TEXT UNIQUE)""") + + +def openapi_paths() -> dict[str, object]: + """Describe only the two owner GET routes and their separate authority.""" + headers = [dict(name=name, **{"in": "header"}, required=True, + schema={"type": "string"}) for name in ( + "EP-Discovery-Contract", "EP-Instance-ID", "EP-Consumer-ID", + "Workspace-Instance-ID", "EP-Audience")] + responses = {str(code): {"description": description} for code, description in ( + (200, "Closed packaged DiscoverySnapshot/TopologySnapshot v1.0; no execution permission"), + (400, "Malformed, duplicate or unknown query/body/cursor"), (401, "Invalid, expired or revoked discovery bearer"), + (403, "Consumer/audience/instance/project refusal"), (409, "Snapshot or topology conflict"), + (422, "Unsupported contract or store schema"), (500, "CENTRAL corrupt or invalid"), + (503, "CENTRAL unavailable; never an empty healthy list"))} + return { + "/v1/onboarding/projects": {"get": dict(summary="Read an explicitly scoped B8R project page", + security=[{"discoveryBearer": []}], parameters=headers + [ + {"name": "limit", "in": "query", "schema": {"type": "integer", "minimum": 1, "maximum": 100, "default": 25}}, + {"name": "cursor", "in": "query", "schema": {"type": "string", "maxLength": 1024}}], + responses=responses, **{"x-response-json-schema": "urn:engineering-platform:workspace-project-discovery:1.0#/$defs/DiscoverySnapshot"})}, + "/v1/onboarding/projects/{project_id}/topology": {"get": dict(summary="Read one selected project in the same authorized snapshot", + security=[{"discoveryBearer": []}], parameters=headers + [ + {"name": "project_id", "in": "path", "required": True, "schema": {"type": "string", "maxLength": 128}}, + {"name": "expected_topology_revision", "in": "query", "required": True, "schema": {"type": "string", "pattern": "^[a-f0-9]{64}$"}}], + responses=responses, **{"x-response-json-schema": "urn:engineering-platform:workspace-project-discovery:1.0#/$defs/TopologySnapshot"})}, + } + + +def validate_schema(connection: sqlite3.Connection) -> None: + expected = ( + ("grant_id", "TEXT", 0, 1), ("instance_id", "TEXT", 1, 0), + ("consumer_id", "TEXT", 1, 0), ("workspace_instance_id", "TEXT", 1, 0), + ("audience", "TEXT", 1, 0), ("project_ids", "TEXT", 1, 0), + ("fields", "TEXT", 1, 0), ("issuance_operation_id", "TEXT", 1, 0), + ("verifier", "BLOB", 1, 0), ("cursor_key", "BLOB", 1, 0), ("issued_at", "TEXT", 1, 0), + ("expires_at", "TEXT", 1, 0), ("revoked_at", "TEXT", 0, 0), + ("revocation_operation_id", "TEXT", 0, 0), + ) + observed = tuple((r[1], r[2].upper(), r[3], r[5]) for r in connection.execute( + "SELECT * FROM pragma_table_info(?)", (TABLE,))) + unique = {tuple(r[0] for r in connection.execute( + "SELECT name FROM pragma_index_info(?) ORDER BY seqno", (index[1],))) + for index in connection.execute("SELECT * FROM pragma_index_list(?)", (TABLE,)) if index[2]} + foreign = {(r[2], r[3], r[4]) for r in connection.execute( + "SELECT * FROM pragma_foreign_key_list(?)", (TABLE,))} + row = connection.execute("SELECT sql FROM sqlite_master WHERE name=?", (TABLE,)).fetchone() + sql = "" if row is None else "".join(row[0].lower().split()) + if (observed != expected or not {("grant_id",), ("issuance_operation_id",), + ("verifier",), ("revocation_operation_id",)} <= unique + or foreign != {("ep_installations", "instance_id", "instance_id")} + or "check(length(verifier)=32)" not in sql or "check(length(cursor_key)=32)" not in sql + or "check(audience='workspace')" not in sql): + raise DiscoveryError("DISCOVERY_SCHEMA_INVALID", 500) + + +def issue(connection: sqlite3.Connection, *, grant_id: str, instance_id: str, + consumer_id: str, workspace_instance_id: str, project_ids: list[str], + fields: list[str], operation_id: str, ttl_seconds: int) -> dict[str, object]: + for value in (grant_id, instance_id, consumer_id, workspace_instance_id, operation_id): + _id(value) + projects, selected_fields = _ids(project_ids), _fields(fields) + if type(ttl_seconds) is not int or not 1 <= ttl_seconds <= 86400: + raise DiscoveryError("DISCOVERY_LIFETIME_INVALID") + require_current_instance(connection, instance_id) + for project_id in projects: + if connection.execute("SELECT 1 FROM ep_project_registrations WHERE project_id=? AND status='ACTIVE'", (project_id,)).fetchone() is None: + raise DiscoveryError("DISCOVERY_PROJECT_UNAVAILABLE", 403) + if connection.execute("SELECT 1 FROM ep_workspace_discovery_grants WHERE grant_id=? OR issuance_operation_id=?", (grant_id, operation_id)).fetchone(): + raise DiscoveryError("DISCOVERY_ALREADY_ISSUED_USE_STATUS", 409) + token = secrets.token_urlsafe(32) + verifier = hashlib.sha256(_DOMAIN + token.encode("ascii")).digest() + now = _now() + connection.execute("""INSERT INTO ep_workspace_discovery_grants + (grant_id,instance_id,consumer_id,workspace_instance_id,audience,project_ids,fields, + issuance_operation_id,verifier,cursor_key,issued_at,expires_at) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)""", + (grant_id, instance_id, consumer_id, workspace_instance_id, AUDIENCE, + _canonical(projects).decode(), _canonical(selected_fields).decode(), operation_id, + verifier, secrets.token_bytes(32), now.isoformat(), (now + timedelta(seconds=ttl_seconds)).isoformat())) + return {**status(connection, grant_id=grant_id), "credential": token} + + +def status(connection: sqlite3.Connection, *, grant_id: str) -> dict[str, object]: + _id(grant_id) + row = connection.execute("""SELECT instance_id,consumer_id,workspace_instance_id,audience, + project_ids,fields,issued_at,expires_at,revoked_at FROM ep_workspace_discovery_grants WHERE grant_id=?""", (grant_id,)).fetchone() + if row is None: + raise DiscoveryError("DISCOVERY_GRANT_UNKNOWN", 404) + try: + expires = datetime.fromisoformat(row[7]) + issued = datetime.fromisoformat(row[6]) + projects, selected_fields = _ids(json.loads(row[4])), _fields(json.loads(row[5])) + for value in row[:3]: + _id(value) + if (expires.tzinfo is None or issued.tzinfo is None or row[3] != AUDIENCE + or not timedelta(0) < expires - issued <= timedelta(days=1)): + raise ValueError("Invalid finite grant") + except (ValueError, TypeError) as error: + raise DiscoveryError("DISCOVERY_GRANT_CORRUPT", 500) from error + return dict(grant_id=grant_id, instance_id=row[0], consumer_id=row[1], workspace_instance_id=row[2], + audience=row[3], project_ids=projects, fields=selected_fields, + issued_at=row[6], expires_at=row[7], status="REVOKED" if row[8] else ( + "EXPIRED" if _now() >= expires else "ACTIVE")) + + +def revoke(connection: sqlite3.Connection, *, grant_id: str, operation_id: str) -> dict[str, object]: + _id(grant_id); _id(operation_id) + row = connection.execute("SELECT revoked_at,revocation_operation_id FROM ep_workspace_discovery_grants WHERE grant_id=?", (grant_id,)).fetchone() + if row is None: + raise DiscoveryError("DISCOVERY_GRANT_UNKNOWN", 404) + if row[0] is not None: + if row[1] != operation_id: + raise DiscoveryError("DISCOVERY_REVOCATION_CONFLICT", 409) + else: + connection.execute("UPDATE ep_workspace_discovery_grants SET revoked_at=?,revocation_operation_id=? WHERE grant_id=?", (_now().isoformat(), operation_id, grant_id)) + return status(connection, grant_id=grant_id) + + +def require_current_instance(connection: sqlite3.Connection, expected_instance: str) -> str: + """Noncreating owner/read context; initialization is a separate command.""" + _id(expected_instance) + validate_schema(connection) + metadata = dict(connection.execute("SELECT key,value FROM engineering_metadata WHERE key IN ('installation.instance_id','installation.schema_version')")) + if metadata.get("installation.instance_id") != expected_instance: + raise DiscoveryError("DISCOVERY_INSTANCE_MISMATCH", 403) + schema = connection.execute("SELECT MAX(version) FROM engineering_schema_migrations").fetchone()[0] + installation = connection.execute("SELECT schema_version FROM ep_installations WHERE instance_id=?", (expected_instance,)).fetchone() + if metadata.get("installation.schema_version") != str(SCHEMA_VERSION) or schema != SCHEMA_VERSION or installation != (SCHEMA_VERSION,): + raise DiscoveryError("DISCOVERY_SCHEMA_UNSUPPORTED", 422) + return expected_instance + + +def _authenticate(connection: sqlite3.Connection, token: object, headers: dict[str, str]) -> dict[str, object]: + # Publication relies on rollback-journal readers blocking owner commits. + # Never change a store's journal mode as a side effect of discovery reads. + if connection.execute("PRAGMA journal_mode").fetchone()[0] not in {"delete", "truncate", "persist"}: + raise DiscoveryError("DISCOVERY_JOURNAL_UNSUPPORTED", 422) + if not isinstance(token, str) or re.fullmatch(r"[A-Za-z0-9_-]{32,128}", token) is None: + raise DiscoveryError("UNAUTHENTICATED", 401) + validate_schema(connection) + value = hashlib.sha256(_DOMAIN + token.encode("ascii")).digest() + row = connection.execute("SELECT grant_id,verifier FROM ep_workspace_discovery_grants WHERE verifier=?", (value,)).fetchone() + if row is None or not hmac.compare_digest(bytes(row[1]), value): + raise DiscoveryError("UNAUTHENTICATED", 401) + scope = status(connection, grant_id=row[0]) + if scope["status"] != "ACTIVE": + raise DiscoveryError("DISCOVERY_INACTIVE", 401) + expected = {"EP-Instance-ID": scope["instance_id"], "EP-Consumer-ID": scope["consumer_id"], + "Workspace-Instance-ID": scope["workspace_instance_id"], "EP-Audience": AUDIENCE} + if any(headers.get(k) != v for k, v in expected.items()): + raise DiscoveryError("DISCOVERY_SCOPE_MISMATCH", 403) + require_current_instance(connection, str(scope["instance_id"])) + return scope + + +def _topology(connection: sqlite3.Connection, scope: dict[str, object]) -> list[dict[str, object]]: + projects = [] + now = _now() + cutoff = now - timedelta(seconds=90) + for project_id in scope["project_ids"]: + row = connection.execute("SELECT attachment_contract FROM ep_project_registrations WHERE project_id=? AND status='ACTIVE'", (project_id,)).fetchone() + if row is None: + raise DiscoveryError("DISCOVERY_PROJECT_UNAVAILABLE", 403) + authority = json.loads(row[0])["authority_repository_id"] + _id(authority) + repositories = [] + for repository_id, authority_id, role in connection.execute("SELECT repository_id,authority_repository_id,role FROM ep_repository_registrations WHERE project_id=? ORDER BY repository_id LIMIT 101", (project_id,)): + _id(repository_id) + if authority_id != authority or role not in {"authority", "child"}: + raise DiscoveryError("DISCOVERY_TOPOLOGY_CORRUPT", 500) + repositories.append(dict(repository_id=repository_id, role=role)) + if len(repositories) > 100: + raise DiscoveryError("DISCOVERY_TOPOLOGY_LIMIT", 409) + authorities = [r["repository_id"] for r in repositories if r["role"] == "authority"] + if authorities != [authority]: + raise DiscoveryError("DISCOVERY_TOPOLOGY_INCOMPLETE", 409) + attachments = None + if "attachment_availability" in scope["fields"]: + attachments = [] + rows = connection.execute("""SELECT a.repository_id,a.agent_id,a.availability,g.state,g.last_seen_at + FROM ep_agent_repository_attachments a JOIN ep_repository_registrations r ON r.repository_id=a.repository_id + JOIN ep_agent_registrations g ON g.agent_id=a.agent_id WHERE r.project_id=? + ORDER BY a.repository_id,a.agent_id LIMIT 101""", (project_id,)).fetchall() + if len(rows) > 100: + raise DiscoveryError("DISCOVERY_TOPOLOGY_LIMIT", 409) + for repo, agent, available, state, last_seen in rows: + seen = None + if last_seen is not None: + try: + seen = datetime.fromisoformat(last_seen) + if seen.tzinfo is None: + raise ValueError("Timestamp lacks timezone") + except (ValueError, TypeError) as error: + raise DiscoveryError("DISCOVERY_TOPOLOGY_CORRUPT", 500) from error + attachments.append(dict(repository_id=_id(repo), agent_id=_id(agent), + availability="AVAILABLE" if available == "AVAILABLE" and state != "REVOKED" and seen is not None and cutoff <= seen <= now else "UNAVAILABLE")) + projects.append(dict(project_id=project_id, authority_repository_id=authority, + repositories=repositories, attachments=attachments)) + return projects + + +def _snapshot(connection: sqlite3.Connection, scope: dict[str, object], identity: tuple[int, int]) -> tuple[str, list[dict[str, object]]]: + data = _topology(connection, scope) + digest = hashlib.sha256(_canonical([CONTRACT_VERSION, identity, scope, data])).hexdigest() + return digest, data + + +def _cursor(key: bytes, payload: dict[str, object]) -> str: + data = _canonical(payload) + signature = hmac.new(key, b"discovery.cursor.v1\0" + data, hashlib.sha256).digest() + return base64.urlsafe_b64encode(data + signature).decode("ascii").rstrip("=") + + +def _decode_cursor(key: bytes, value: str) -> dict[str, object]: + try: + if len(value) > 1024 or re.fullmatch(r"[A-Za-z0-9_-]+", value) is None: + raise ValueError + data = base64.b64decode(value + "=" * (-len(value) % 4), altchars=b"-_", validate=True) + payload = json.loads(data[:-32]) + if not isinstance(payload, dict) or not hmac.compare_digest(_cursor(key, payload), value): + raise ValueError + return payload + except (ValueError, UnicodeError): + raise DiscoveryError("DISCOVERY_CURSOR_INVALID", 400) from None + + +def _identity(path: Path) -> tuple[int, int]: + stat = path.stat() + return stat.st_dev, stat.st_ino + + +def read(data_root: Path, *, token: str | None, headers: dict[str, str], query: dict[str, list[str]], + project_id: str | None, publish: Callable[[dict[str, object], str, Callable[[], None]], None]) -> None: + """Two fresh read snapshots; publication is fenced by the final shared lock. + + A committed revocation/source change before publication is observed on the + second connection. Owner revocation cannot commit during the final locked + publication. No state/cursor is written by the read. + """ + if headers.get("EP-Discovery-Contract") != CONTRACT_VERSION: + raise DiscoveryError("DISCOVERY_CONTRACT_UNSUPPORTED", 422) + permitted = {"expected_topology_revision"} if project_id is not None else {"cursor", "limit"} + if not set(query) <= permitted or any(len(v) != 1 for v in query.values()): + raise DiscoveryError("DISCOVERY_QUERY_INVALID") + if project_id is not None: + _id(project_id) + expected = query.get("expected_topology_revision", [""])[0] + if _DIGEST.fullmatch(expected) is None: + raise DiscoveryError("DISCOVERY_REVISION_REQUIRED") + else: + try: + limit = int(query.get("limit", ["25"])[0]) + except ValueError: + raise DiscoveryError("DISCOVERY_LIMIT_INVALID") from None + if not 1 <= limit <= 100: + raise DiscoveryError("DISCOVERY_LIMIT_INVALID") + path = data_root / "epdata.sqlite" + identity = _identity(path) + with storage.sqlite_read_connection(path) as connection: + connection.execute("BEGIN") + scope = _authenticate(connection, token, headers) + cursor_key = bytes(connection.execute("SELECT cursor_key FROM ep_workspace_discovery_grants WHERE grant_id=?", (scope["grant_id"],)).fetchone()[0]) + revision, data = _snapshot(connection, scope, identity) + offset = 0 + if project_id is not None: + if project_id not in scope["project_ids"]: + raise DiscoveryError("DISCOVERY_PROJECT_DENIED", 403) + if expected != revision: + raise DiscoveryError("DISCOVERY_SNAPSHOT_CONFLICT", 409) + elif "cursor" in query: + cursor = _decode_cursor(cursor_key, query["cursor"][0]) + if set(cursor) != {"snapshot", "offset", "limit"} or type(cursor["offset"]) is not int or cursor["limit"] != limit or not 0 < cursor["offset"] < len(data): + raise DiscoveryError("DISCOVERY_CURSOR_INVALID") + if cursor["snapshot"] != revision: + raise DiscoveryError("DISCOVERY_SNAPSHOT_CONFLICT", 409) + offset = cursor["offset"] + with storage.sqlite_read_connection(path) as connection: + connection.execute("BEGIN") + current = _authenticate(connection, token, headers) + current_revision, current_data = _snapshot(connection, current, _identity(path)) + if current_revision != revision or current != scope or _identity(path) != identity: + raise DiscoveryError("DISCOVERY_SNAPSHOT_CONFLICT", 409) + # Recheck finite lifetime after data assembly, immediately before send. + _authenticate(connection, token, headers) + payload = dict(contract_version=CONTRACT_VERSION, + producer={"id": "engineering-platform", "version": CURRENT_PLATFORM_VERSION}, + instance_id=scope["instance_id"], snapshot_id=revision, topology_revision=revision, + scope=scope, observed_at=_now().isoformat()) + if project_id is None: + page = current_data[offset:offset + limit] + end = offset + len(page) + payload.update(projects=[{k: p[k] for k in ("project_id", "authority_repository_id")} for p in page], + next_cursor=_cursor(cursor_key, dict(snapshot=revision, offset=end, limit=limit)) if end < len(data) else None, + completeness="PARTIAL_PAGE" if end < len(data) else "COMPLETE_PAGE") + else: + selected = next(p for p in current_data if p["project_id"] == project_id) + payload.update(project={k: selected[k] for k in ("project_id", "authority_repository_id")}, + repositories=selected["repositories"], attachments=selected["attachments"], + attachment_visibility="GRANTED" if selected["attachments"] is not None else "NOT_GRANTED", + completeness="COMPLETE") + def publication_check() -> None: + if _identity(path) != identity: + raise DiscoveryError("DISCOVERY_SNAPSHOT_CONFLICT", 409) + if _authenticate(connection, token, headers) != scope: + raise DiscoveryError("DISCOVERY_SNAPSHOT_CONFLICT", 409) + publication_check() + publish(payload, str(scope["instance_id"]), publication_check) diff --git a/src/engineering_platform/schemas/workspace-project-discovery-v1.schema.json b/src/engineering_platform/schemas/workspace-project-discovery-v1.schema.json new file mode 100644 index 00000000..e1d8e5b8 --- /dev/null +++ b/src/engineering_platform/schemas/workspace-project-discovery-v1.schema.json @@ -0,0 +1,320 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "urn:engineering-platform:workspace-project-discovery:1.0", + "title": "EP-owned HOB-EP01 discovery/topology response", + "oneOf": [ + { + "$ref": "#/$defs/DiscoverySnapshot" + }, + { + "$ref": "#/$defs/TopologySnapshot" + } + ], + "$defs": { + "ReadScope": { + "type": "object", + "additionalProperties": false, + "properties": { + "grant_id": { + "type": "string", + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$" + }, + "instance_id": { + "type": "string", + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$" + }, + "consumer_id": { + "type": "string", + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$" + }, + "workspace_instance_id": { + "type": "string", + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$" + }, + "audience": { + "const": "workspace" + }, + "project_ids": { + "type": "array", + "items": { + "type": "string", + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$" + }, + "maxItems": 100, + "uniqueItems": true + }, + "fields": { + "type": "array", + "items": { + "enum": [ + "project_identity", + "repository_identity", + "attachment_availability" + ] + }, + "maxItems": 3, + "uniqueItems": true + }, + "issued_at": { + "type": "string", + "format": "date-time" + }, + "expires_at": { + "type": "string", + "format": "date-time" + }, + "status": { + "const": "ACTIVE" + } + }, + "required": [ + "grant_id", + "instance_id", + "consumer_id", + "workspace_instance_id", + "audience", + "project_ids", + "fields", + "issued_at", + "expires_at", + "status" + ] + }, + "ProjectIdentity": { + "type": "object", + "additionalProperties": false, + "properties": { + "project_id": { + "type": "string", + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$" + }, + "authority_repository_id": { + "type": "string", + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$" + } + }, + "required": [ + "project_id", + "authority_repository_id" + ] + }, + "RepositoryIdentity": { + "type": "object", + "additionalProperties": false, + "properties": { + "repository_id": { + "type": "string", + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$" + }, + "role": { + "enum": [ + "authority", + "child" + ] + } + }, + "required": [ + "repository_id", + "role" + ] + }, + "Attachment": { + "type": "object", + "additionalProperties": false, + "properties": { + "repository_id": { + "type": "string", + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$" + }, + "agent_id": { + "type": "string", + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$" + }, + "availability": { + "enum": [ + "AVAILABLE", + "UNAVAILABLE" + ] + } + }, + "required": [ + "repository_id", + "agent_id", + "availability" + ] + }, + "DiscoverySnapshot": { + "type": "object", + "additionalProperties": false, + "properties": { + "contract_version": { + "const": "1.0" + }, + "producer": { + "type": "object", + "additionalProperties": false, + "properties": { + "id": { + "const": "engineering-platform" + }, + "version": { + "type": "string", + "pattern": "^[0-9]+\\.[0-9]+\\.[0-9]+$" + } + }, + "required": [ + "id", + "version" + ] + }, + "instance_id": { + "type": "string", + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$" + }, + "snapshot_id": { + "type": "string", + "pattern": "^[a-f0-9]{64}$" + }, + "topology_revision": { + "type": "string", + "pattern": "^[a-f0-9]{64}$" + }, + "scope": { + "$ref": "#/$defs/ReadScope" + }, + "observed_at": { + "type": "string", + "format": "date-time" + }, + "projects": { + "type": "array", + "items": { + "$ref": "#/$defs/ProjectIdentity" + }, + "maxItems": 100, + "uniqueItems": true + }, + "next_cursor": { + "type": [ + "string", + "null" + ], + "maxLength": 1024 + }, + "completeness": { + "enum": [ + "COMPLETE_PAGE", + "PARTIAL_PAGE" + ] + } + }, + "required": [ + "contract_version", + "producer", + "instance_id", + "snapshot_id", + "topology_revision", + "scope", + "observed_at", + "projects", + "next_cursor", + "completeness" + ] + }, + "TopologySnapshot": { + "type": "object", + "additionalProperties": false, + "properties": { + "contract_version": { + "const": "1.0" + }, + "producer": { + "type": "object", + "additionalProperties": false, + "properties": { + "id": { + "const": "engineering-platform" + }, + "version": { + "type": "string", + "pattern": "^[0-9]+\\.[0-9]+\\.[0-9]+$" + } + }, + "required": [ + "id", + "version" + ] + }, + "instance_id": { + "type": "string", + "pattern": "^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$" + }, + "snapshot_id": { + "type": "string", + "pattern": "^[a-f0-9]{64}$" + }, + "topology_revision": { + "type": "string", + "pattern": "^[a-f0-9]{64}$" + }, + "scope": { + "$ref": "#/$defs/ReadScope" + }, + "observed_at": { + "type": "string", + "format": "date-time" + }, + "project": { + "$ref": "#/$defs/ProjectIdentity" + }, + "repositories": { + "type": "array", + "items": { + "$ref": "#/$defs/RepositoryIdentity" + }, + "maxItems": 100, + "uniqueItems": true, + "minItems": 1 + }, + "attachments": { + "anyOf": [ + { + "type": "array", + "items": { + "$ref": "#/$defs/Attachment" + }, + "maxItems": 100, + "uniqueItems": true + }, + { + "type": "null" + } + ] + }, + "attachment_visibility": { + "enum": [ + "GRANTED", + "NOT_GRANTED" + ] + }, + "completeness": { + "const": "COMPLETE" + } + }, + "required": [ + "contract_version", + "producer", + "instance_id", + "snapshot_id", + "topology_revision", + "scope", + "observed_at", + "project", + "repositories", + "attachments", + "attachment_visibility", + "completeness" + ] + } + } +} diff --git a/src/engineering_platform/server.py b/src/engineering_platform/server.py index c6a406e2..142911ce 100644 --- a/src/engineering_platform/server.py +++ b/src/engineering_platform/server.py @@ -38,6 +38,7 @@ from uuid import uuid4 from . import installation_pairing +from . import project_discovery from . import agent_trust from . import central_database from . import merge_delegation @@ -149,7 +150,7 @@ # bootstrap is deliberately separate from the retired predecessor migration # machinery: it creates a clean installation only and never accepts a source # database path. -SERVER_STORE_SCHEMA_VERSION = 74 +SERVER_STORE_SCHEMA_VERSION = 75 SERVER_ENVIRONMENT_DATA_ROOT = "EP_SERVER_DATA_ROOT" FILE_INBOX_DIRECTORY = "file-inbox" HTTP_JSON_OPENAPI_PATH = "/v1/openapi.json" @@ -210,6 +211,7 @@ def _http_json_openapi_document() -> dict[str, object]: "description": "Canonical authenticated submission ingress and platform health.", }, "paths": { + **project_discovery.openapi_paths(), "/v1/projects/{project_id}/submissions/{submission_id}/effect-result": { "get": { "summary": "Read the producer-bound FME result and exact artifact bytes", @@ -444,6 +446,8 @@ def _http_json_openapi_document() -> dict[str, object]: }, "components": { "securitySchemes": { + "discoveryBearer": {"type": "http", "scheme": "bearer", "bearerFormat": "opaque", + "description": "Separate finite, revocable Workspace discovery READ grant; never a submission/admin bearer."}, "consumerBearer": {"type": "http", "scheme": "bearer", "bearerFormat": "opaque"}, }, }, @@ -730,6 +734,7 @@ class ServerConfigurationError(ValueError): "ep_control_provenance", "ep_installation_pairings", "ep_installation_pairing_credentials", + "ep_workspace_discovery_grants", "ep_consumer_credentials", "ep_consumer_registrations", "ep_consumer_credential_recovery_operations", @@ -996,6 +1001,7 @@ def _install_current_schema(connection: sqlite3.Connection, identity: RuntimeIde _install_current_submission_schema(connection) parallel_action_admission.install_schema(connection) installation_pairing.install_schema(connection) + project_discovery.install_schema(connection) _install_forge_action_context_schema(connection) _install_forge_planning_context_schema(connection) _install_execution_host_evidence_schema(connection) @@ -2293,6 +2299,22 @@ def _migrate_schema_74(connection: sqlite3.Connection) -> None: connection.execute("UPDATE ep_installations SET schema_version=74") +def _migrate_schema_75(connection: sqlite3.Connection) -> None: + """Owner migration only, using the existing guarded rebuild transaction. + + Canonical foreign-key targets and all existing credentials are preserved; + discovery HTTP reads never invoke this migration. + """ + connection.execute("ALTER TABLE ep_installations RENAME TO ep_installations_schema74") + connection.execute("CREATE TABLE ep_installations (instance_id TEXT PRIMARY KEY,created_at TEXT NOT NULL,schema_version INTEGER NOT NULL CHECK(schema_version BETWEEN 41 AND 75))") + connection.execute("INSERT INTO ep_installations SELECT instance_id,created_at,75 FROM ep_installations_schema74") + connection.execute("DROP TABLE ep_installations_schema74") + project_discovery.install_schema(connection) + central_operational_reset.install_writer_fences(connection) + connection.execute("INSERT INTO engineering_schema_migrations(version) VALUES(75)") + connection.execute("UPDATE engineering_metadata SET value='75' WHERE key='installation.schema_version'") + + _SERVER_SCHEMA_UPGRADE_STEPS = ( (42, _migrate_schema_42), (43, _migrate_schema_43), @@ -2327,6 +2349,7 @@ def _migrate_schema_74(connection: sqlite3.Connection) -> None: (72, _migrate_schema_72), (73, _migrate_schema_73), (74, _migrate_schema_74), + (75, _migrate_schema_75), ) _SUPPORTED_SERVER_SCHEMA_VERSIONS = frozenset( range(41, SERVER_STORE_SCHEMA_VERSION + 1) @@ -2348,8 +2371,11 @@ def validate_store(data_root: Path, identity: RuntimeIdentity) -> dict[str, obje tables = _table_names(connection) try: installation_pairing.validate_schema(connection) + project_discovery.validate_schema(connection) except installation_pairing.InstallationPairingError as error: raise ServerConfigurationError("EP installation pairing schema is invalid.") from error + except project_discovery.DiscoveryError as error: + raise ServerConfigurationError("EP discovery grant schema is invalid.") from error indexes = _index_names(connection) views = _view_names(connection) triggers = {str(row[0]) for row in connection.execute("SELECT name FROM sqlite_master WHERE type='trigger'")} @@ -5984,8 +6010,10 @@ def _status(self) -> dict[str, object]: return report def _send(self, status_code: int, payload: dict[str, object], instance_id: str | None = None, - *, cache_control: str | None = None) -> None: + *, cache_control: str | None = None, before_headers=None) -> None: encoded = json.dumps(payload, sort_keys=True).encode("utf-8") + if before_headers is not None: + before_headers() self.send_response(status_code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(encoded))) @@ -7423,6 +7451,38 @@ def do_GET(self) -> None: # noqa: N802 if request.path in {HTTP_JSON_OPENAPI_PATH, "/openapi.json", "/swagger.json"}: self._send(200, _http_json_openapi_document()) return + discovery_match = re.fullmatch(r"/v1/onboarding/projects(?:/([^/]+)/topology)?", request.path) + if discovery_match: + if (self.headers.get("Content-Length") not in (None, "0") + or len(self.headers.get_all("Content-Length", [])) > 1 + or self.headers.get("Transfer-Encoding") is not None): + self._send(400, {"error": "DISCOVERY_BODY_FORBIDDEN"}, cache_control="no-store") + return + authorization = self.headers.get("Authorization", "") + token = authorization[7:] if authorization.startswith("Bearer ") else None + try: + project_discovery.read(self.server.data_root, token=token, + headers={key: self.headers.get(key, "") for key in ( + "EP-Discovery-Contract", "EP-Instance-ID", "EP-Consumer-ID", + "Workspace-Instance-ID", "EP-Audience")}, + query=parse_qs(request.query, keep_blank_values=True), + project_id=unquote(discovery_match.group(1)) if discovery_match.group(1) else None, + publish=lambda body, instance, check: self._send(200, body, instance, + cache_control="no-store", before_headers=check)) + except project_discovery.DiscoveryError as error: + self._send(error.status, {"error": error.code}, cache_control="no-store") + except (json.JSONDecodeError, KeyError, TypeError): + logging.getLogger(__name__).exception("Invalid stored discovery topology") + self._send(500, {"error": "CENTRAL_CORRUPT"}, cache_control="no-store") + except sqlite3.Error as error: + unavailable = getattr(error, "sqlite_errorcode", 0) & 255 in { + sqlite3.SQLITE_CANTOPEN, sqlite3.SQLITE_IOERR, sqlite3.SQLITE_BUSY, sqlite3.SQLITE_LOCKED} + logging.getLogger(__name__).exception("Discovery CENTRAL read failed") + self._send(503 if unavailable else 500, + {"error": "CENTRAL_UNAVAILABLE" if unavailable else "CENTRAL_CORRUPT"}, cache_control="no-store") + except OSError: + self._send(503, {"error": "CENTRAL_UNAVAILABLE"}, cache_control="no-store") + return if request.path == "/v1/installation-compatibility": identity = initialize(self.server.data_root).instance_id if request.query or any(self.headers.get(key) is not None for key in ("EP-Project-ID", "EP-Repository-ID")): @@ -8324,7 +8384,7 @@ def health(data_root: Path) -> dict[str, object]: def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="engineering-platform-server", description="Manage the standalone Engineering Platform Server foundation") - parser.add_argument("command", choices=("init", "start", "serve", "stop", "status", "health", "operational-diagnose", "operational-qualify", "operational-readback", "operational-update-assess", "operational-inventory", "system-service-inventory", "legacy-adoption-inspect", "legacy-adoption-authorize", "installation-update-plan", "installation-update-prepare", "installation-update-admit", "installation-update-apply", "installation-update-resume", "installation-update-status", "owner-consumer-readback", "owner-credential-recover", "owner-credential-recovery-adopt-peer-configuration", "owner-credential-recovery-status", "service-install", "service-uninstall", "relay-install", "relay-uninstall", "pairing-create", "agent-status", "agent-revoke", "agent-reset", "topology", "submission-diagnose", "bootstrap-topology", "register-topology", "installation-pairing-register", "installation-pairing-status", "installation-pairing-issue", "installation-pairing-revoke", "installation-pairing-detach", "provision-declaration", "issue-consumer-credential", "issue-development-consumer-credential", "grant-operator-capability", "revoke-operator-capability", "grant-parallel-action-repository", "revoke-parallel-action-repository", "qualify-parallel-action-artifact", "inspect-assurance-target", "select-assurance-target", "revoke-assurance-target", "reserve-merge-delegation", "activate-merge-delegation", "revoke-merge-delegation", "bind-repository", "rebind-repository", "unbind-repository", "resolve-repository", "recover-managed-workspace", "register-producer-binding", "list-producer-bindings", "deactivate-producer-binding")) + parser.add_argument("command", choices=("init", "start", "serve", "stop", "status", "health", "operational-diagnose", "operational-qualify", "operational-readback", "operational-update-assess", "operational-inventory", "system-service-inventory", "legacy-adoption-inspect", "legacy-adoption-authorize", "installation-update-plan", "installation-update-prepare", "installation-update-admit", "installation-update-apply", "installation-update-resume", "installation-update-status", "owner-consumer-readback", "owner-credential-recover", "owner-credential-recovery-adopt-peer-configuration", "owner-credential-recovery-status", "service-install", "service-uninstall", "relay-install", "relay-uninstall", "pairing-create", "agent-status", "agent-revoke", "agent-reset", "topology", "submission-diagnose", "bootstrap-topology", "register-topology", "installation-pairing-register", "installation-pairing-status", "installation-pairing-issue", "installation-pairing-revoke", "installation-pairing-detach", "discovery-grant-issue", "discovery-grant-status", "discovery-grant-revoke", "provision-declaration", "issue-consumer-credential", "issue-development-consumer-credential", "grant-operator-capability", "revoke-operator-capability", "grant-parallel-action-repository", "revoke-parallel-action-repository", "qualify-parallel-action-artifact", "inspect-assurance-target", "select-assurance-target", "revoke-assurance-target", "reserve-merge-delegation", "activate-merge-delegation", "revoke-merge-delegation", "bind-repository", "rebind-repository", "unbind-repository", "resolve-repository", "recover-managed-workspace", "register-producer-binding", "list-producer-bindings", "deactivate-producer-binding")) parser.add_argument("--data-root", type=Path, default=default_data_root()) parser.add_argument("--runtime-profile", choices=("operational", "development"), default="operational") parser.add_argument("--development-venv", type=Path) @@ -8337,6 +8397,11 @@ def build_parser() -> argparse.ArgumentParser: parser.add_argument("--path", type=Path) parser.add_argument("--declaration", type=Path) parser.add_argument("--consumer-id") + parser.add_argument("--workspace-instance-id") + parser.add_argument("--discovery-grant-id") + parser.add_argument("--discovery-project-id", action="append", default=[]) + parser.add_argument("--discovery-field", action="append") + parser.add_argument("--discovery-ttl-seconds", type=int, default=900) parser.add_argument("--submission-id") parser.add_argument("--intake-id") parser.add_argument("--artifact-id") @@ -9130,6 +9195,35 @@ def main(argv: list[str] | None = None) -> int: receipt_complete = transport != "FILE_INBOX" or (isinstance(receipt_id, str) and bool(receipt_id) and isinstance(received_at, str) and bool(received_at)) scope_complete = run_id is not None and (dispatch_project, dispatch_repository) == (project_id, repository_id) result = {"submission_id": args.submission_id, "project_id": project_id, "repository_id": repository_id, "submission_state": state, "admission": admission, "run_id": run_id, "dispatch_state": dispatch_state, "operator_resolution": resolution, "transport_provenance": "COMPLETE" if receipt_complete else "INCOMPLETE", "admission_audit_provenance": "PRESENT" if admission_audit else "UNAVAILABLE", "receipt_run_provenance": "PRESENT" if run_id else "UNAVAILABLE", "dispatch_scope_provenance": "COMPLETE" if scope_complete else "UNAVAILABLE", "lane_blocker": {"run_id": blocked[0], "state": blocked[1]} if blocked else None, "early_failure": early, "worker_eligible": state == "QUEUED" and admission == "ADMITTED" and run_id is None and blocked is None} + elif args.command.startswith("discovery-grant-"): + from .platform_admin import require_installation_owner + require_installation_owner(args.data_root) + if not args.expected_instance_id or not args.discovery_grant_id: + raise ServerConfigurationError("DISCOVERY_OWNER_INSTANCE_REQUIRED") + database = args.data_root / SERVER_DATABASE_FILENAME + context = (storage.sqlite_read_connection(database) if args.command == "discovery-grant-status" + else storage.sqlite_connection(database.resolve().as_uri() + "?mode=rw", uri=True)) + try: + with context as connection: + connection.execute("PRAGMA foreign_keys=ON") + connection.execute("BEGIN" if args.command == "discovery-grant-status" else "BEGIN IMMEDIATE") + identity = project_discovery.require_current_instance(connection, args.expected_instance_id) + if args.command == "discovery-grant-issue": + result = project_discovery.issue(connection, grant_id=args.discovery_grant_id, + instance_id=identity, consumer_id=args.consumer_id, + workspace_instance_id=args.workspace_instance_id, + project_ids=args.discovery_project_id, + fields=args.discovery_field if args.discovery_field is not None else sorted(project_discovery.IDENTITY_FIELDS), + operation_id=args.operation_id, ttl_seconds=args.discovery_ttl_seconds) + elif args.command == "discovery-grant-revoke": + result = project_discovery.revoke(connection, + grant_id=args.discovery_grant_id, operation_id=args.operation_id) + else: + result = project_discovery.status(connection, grant_id=args.discovery_grant_id) + except sqlite3.Error as error: + unavailable = getattr(error, "sqlite_errorcode", 0) & 255 in { + sqlite3.SQLITE_CANTOPEN, sqlite3.SQLITE_IOERR, sqlite3.SQLITE_BUSY, sqlite3.SQLITE_LOCKED} + raise ServerConfigurationError("DISCOVERY_CENTRAL_UNAVAILABLE" if unavailable else "DISCOVERY_CENTRAL_CORRUPT") from error elif args.command.startswith("installation-pairing-"): from .platform_admin import require_installation_owner require_installation_owner(args.data_root) @@ -9505,7 +9599,8 @@ def main(argv: list[str] | None = None) -> int: local_repository_binding.LocalRepositoryBindingError, managed_workspace_recovery.ManagedWorkspaceRecoveryError, parallel_action_admission.ParallelAdmissionError, - external_producer_binding.ProducerBindingError) as error: + external_producer_binding.ProducerBindingError, + project_discovery.DiscoveryError) as error: print(json.dumps({"error": str(error), "ready": False}, sort_keys=True)) return 2 print(json.dumps(result, sort_keys=True)) diff --git a/src/engineering_platform/templates/workspace-config.json b/src/engineering_platform/templates/workspace-config.json index cf2dd518..a7bc610d 100644 --- a/src/engineering_platform/templates/workspace-config.json +++ b/src/engineering_platform/templates/workspace-config.json @@ -3,7 +3,7 @@ "platform": { "id": "engineering-platform", "name": "Engineering Platform", - "version": "2.3.119", + "version": "2.3.120", "generation": 2, "documentation_namespace": "engineering-platform", "capability_registry_version": 1 diff --git a/tests/engineering/postman/ep-server-http-surface.postman_collection.json b/tests/engineering/postman/ep-server-http-surface.postman_collection.json index a35f81d7..9adec3e1 100644 --- a/tests/engineering/postman/ep-server-http-surface.postman_collection.json +++ b/tests/engineering/postman/ep-server-http-surface.postman_collection.json @@ -1234,6 +1234,77 @@ } } ] + }, + { + "name": "HOB-EP01 bounded project discovery", + "item": [ + { + "name": "DiscoverySnapshot", + "request": { + "method": "GET", + "header": [ + { + "key": "Authorization", + "value": "Bearer {{ep_discovery_bearer}}" + }, + { + "key": "EP-Discovery-Contract", + "value": "1.0" + }, + { + "key": "EP-Instance-ID", + "value": "{{ep_instance_id}}" + }, + { + "key": "EP-Consumer-ID", + "value": "{{workspace_consumer_id}}" + }, + { + "key": "Workspace-Instance-ID", + "value": "{{workspace_instance_id}}" + }, + { + "key": "EP-Audience", + "value": "workspace" + } + ], + "url": "{{base_url}}/v1/onboarding/projects?limit=25" + } + }, + { + "name": "TopologySnapshot selected exact revision", + "request": { + "method": "GET", + "header": [ + { + "key": "Authorization", + "value": "Bearer {{ep_discovery_bearer}}" + }, + { + "key": "EP-Discovery-Contract", + "value": "1.0" + }, + { + "key": "EP-Instance-ID", + "value": "{{ep_instance_id}}" + }, + { + "key": "EP-Consumer-ID", + "value": "{{workspace_consumer_id}}" + }, + { + "key": "Workspace-Instance-ID", + "value": "{{workspace_instance_id}}" + }, + { + "key": "EP-Audience", + "value": "workspace" + } + ], + "url": "{{base_url}}/v1/onboarding/projects/{{project_id}}/topology?expected_topology_revision={{topology_revision}}" + } + } + ] } ] } diff --git a/tests/engineering/postman/http-json-api.postman_collection.json b/tests/engineering/postman/http-json-api.postman_collection.json index 5b2a5a91..2868c13a 100644 --- a/tests/engineering/postman/http-json-api.postman_collection.json +++ b/tests/engineering/postman/http-json-api.postman_collection.json @@ -286,6 +286,77 @@ } } ] + }, + { + "name": "HOB-EP01 bounded project discovery", + "item": [ + { + "name": "DiscoverySnapshot", + "request": { + "method": "GET", + "header": [ + { + "key": "Authorization", + "value": "Bearer {{ep_discovery_bearer}}" + }, + { + "key": "EP-Discovery-Contract", + "value": "1.0" + }, + { + "key": "EP-Instance-ID", + "value": "{{ep_instance_id}}" + }, + { + "key": "EP-Consumer-ID", + "value": "{{workspace_consumer_id}}" + }, + { + "key": "Workspace-Instance-ID", + "value": "{{workspace_instance_id}}" + }, + { + "key": "EP-Audience", + "value": "workspace" + } + ], + "url": "{{base_url}}/v1/onboarding/projects?limit=25" + } + }, + { + "name": "TopologySnapshot selected exact revision", + "request": { + "method": "GET", + "header": [ + { + "key": "Authorization", + "value": "Bearer {{ep_discovery_bearer}}" + }, + { + "key": "EP-Discovery-Contract", + "value": "1.0" + }, + { + "key": "EP-Instance-ID", + "value": "{{ep_instance_id}}" + }, + { + "key": "EP-Consumer-ID", + "value": "{{workspace_consumer_id}}" + }, + { + "key": "Workspace-Instance-ID", + "value": "{{workspace_instance_id}}" + }, + { + "key": "EP-Audience", + "value": "workspace" + } + ], + "url": "{{base_url}}/v1/onboarding/projects/{{project_id}}/topology?expected_topology_revision={{topology_revision}}" + } + } + ] } ] } diff --git a/tests/engineering/test_project_discovery.py b/tests/engineering/test_project_discovery.py new file mode 100644 index 00000000..5f7dd447 --- /dev/null +++ b/tests/engineering/test_project_discovery.py @@ -0,0 +1,480 @@ +from __future__ import annotations + +from contextlib import contextmanager, redirect_stdout +import hashlib +import http.client +import io +import json +import sqlite3 +from pathlib import Path +import tempfile +import threading +import unittest +from unittest.mock import patch +from jsonschema import Draft202012Validator, FormatChecker + +from engineering_platform import agent_trust, project_discovery as discovery, project_topology, server, storage + + +class _DiscoveryFixture: + + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + self.root = Path(self.temporary.name) + self.instance = server.initialize(self.root).instance_id + self.database = self.root / server.SERVER_DATABASE_FILENAME + self.register("project-a", "core-a", "authority") + self.register("project-a", "child-a", "child") + self.register("project-b", "core-b", "authority") + self.token = self.issue("grant-a", ["project-a"])["credential"] + self.headers = {"EP-Discovery-Contract": "1.0", "EP-Instance-ID": self.instance, + "EP-Consumer-ID": "workspace-consumer", "Workspace-Instance-ID": "workspace-instance", + "EP-Audience": "workspace"} + + def register(self, project, repository, role): + with storage.sqlite_connection(self.database) as connection: + return project_topology.register_server_local_topology(connection, declaration={ + "schema_version": "1.0", "project": {"id": project, "authority_repository_id": "core-" + project[-1]}, + "repository": {"id": repository, "role": role}, "validation": {"kind": "none"}}) + + def issue(self, grant, projects, **overrides): + arguments = dict(grant_id=grant, instance_id=self.instance, consumer_id="workspace-consumer", + workspace_instance_id="workspace-instance", project_ids=projects, + fields=sorted(discovery.FIELDS), operation_id="issue-" + grant, ttl_seconds=900) + arguments.update(overrides) + with storage.sqlite_connection(self.database) as connection: + return discovery.issue(connection, **arguments) + +class ProjectDiscoveryTests(_DiscoveryFixture, unittest.TestCase): + def read(self, project=None, query=None, token=None, headers=None): + results = [] + discovery.read(self.root, token=self.token if token is None else token, + headers=self.headers if headers is None else headers, query={} if query is None else query, + project_id=project, publish=lambda body, instance, check: (check(), results.append((body, instance)))) + self.assertEqual(len(results), 1) + self.assertEqual(results[0][1], self.instance) + return results[0][0] + + def assert_refused(self, code, **kwargs): + with self.assertRaises(discovery.DiscoveryError) as context: + self.read(**kwargs) + self.assertEqual(context.exception.code, code) + + def owner_command(self, command, *arguments): + output = io.StringIO() + with redirect_stdout(output): + code = server.main([command, "--data-root", str(self.root), + "--expected-instance-id", self.instance, "--discovery-grant-id", "cli-grant", *arguments]) + return code, json.loads(output.getvalue()) + + def test_owner_cli_issue_status_and_revoke_are_same_finite_authority(self): + code, grant = self.owner_command("discovery-grant-issue", "--consumer-id", "workspace-consumer", + "--workspace-instance-id", "workspace-instance", "--discovery-project-id", "project-a", + "--operation-id", "cli-issue", "--discovery-ttl-seconds", "30") + self.assertEqual(code, 0) + page = self.read(token=grant["credential"]) + self.assertEqual(page["scope"]["grant_id"], "cli-grant") + before = self.database.read_bytes() + code, public = self.owner_command("discovery-grant-status") + self.assertEqual(code, 0) + self.assertNotIn("credential", public) + self.assertEqual(self.database.read_bytes(), before) + code, revoked = self.owner_command("discovery-grant-revoke", "--operation-id", "cli-revoke") + self.assertEqual((code, revoked["status"]), (0, "REVOKED")) + self.assert_refused("DISCOVERY_INACTIVE", token=grant["credential"]) + + def test_malformed_attachment_freshness_is_corruption_not_available(self): + agent = "agent-corrupt-clock" + with storage.sqlite_connection(self.database) as connection: + code = agent_trust.create_pairing_code(connection, agent)["pairing_code"] + token = agent_trust.pair(connection, {"protocol_version": "1.0", "agent_id": agent, + "pairing_code": code})["credential"] + agent_trust.heartbeat(connection, {"protocol_version": "1.0", "agent_id": agent}, token) + agent_trust.register_attachment(connection, {"protocol_version": "1.0", + "agent_id": agent, "availability": "AVAILABLE", "attachment": { + "schema_version": "1.0", "project": {"id": "project-a", "authority_repository_id": "core-a"}, + "repository": {"id": "child-a", "role": "child"}, "validation": {"kind": "none"}}}, token) + for malformed in ("zzz", "2026-10-14T09:00:00", "2026-99-99T09:00:00+00:00"): + with self.subTest(timestamp=malformed): + with storage.sqlite_connection(self.database) as connection: + connection.execute("UPDATE ep_agent_registrations SET last_seen_at=? WHERE agent_id=?", (malformed, agent)) + before = self.database.read_bytes() + self.assert_refused("DISCOVERY_TOPOLOGY_CORRUPT") + self.assertEqual(self.database.read_bytes(), before) + + def test_owner_cli_missing_central_does_not_initialize_or_create_it(self): + self.database.rename(self.database.with_suffix(".saved")) + for command in ("discovery-grant-issue", "discovery-grant-status", "discovery-grant-revoke"): + with self.subTest(command=command): + code, failure = self.owner_command(command, "--operation-id", "absent") + self.assertEqual(code, 2) + self.assertEqual(failure["error"], "DISCOVERY_CENTRAL_UNAVAILABLE") + self.assertFalse(self.database.exists()) + + def after_first_read_close(self, intervention): + original = storage.sqlite_read_connection + count = 0 + + @contextmanager + def intercepted(path): + nonlocal count + count += 1 + first = count == 1 + with original(path) as connection: + yield connection + if first: + intervention() + return patch.object(discovery.storage, "sqlite_read_connection", intercepted) + + def test_real_registered_identity_child_and_read_has_zero_database_writes(self): + before = hashlib.sha256(self.database.read_bytes()).hexdigest() + page = self.read() + self.assertEqual(page["projects"], [{"project_id": "project-a", "authority_repository_id": "core-a"}]) + detail = self.read("project-a", {"expected_topology_revision": [page["topology_revision"]]}) + self.assertEqual(detail["repositories"], [{"repository_id": "child-a", "role": "child"}, {"repository_id": "core-a", "role": "authority"}]) + self.assertEqual(detail["attachments"], []) + self.assertNotIn("project-b", json.dumps(detail)) + self.assertNotIn(str(self.root), json.dumps(detail)) + self.assertNotIn(self.token, json.dumps(detail)) + self.assertEqual(hashlib.sha256(self.database.read_bytes()).hexdigest(), before) + + def test_empty_explicit_scope_is_healthy_empty_not_a_wildcard(self): + issued = self.issue("empty", []) + self.assertEqual(self.read(token=issued["credential"])["projects"], []) + + def test_field_scope_does_not_claim_absent_attachments(self): + issued = self.issue("identity-only", ["project-a"], fields=sorted(discovery.IDENTITY_FIELDS)) + page = self.read(token=issued["credential"]) + detail = self.read("project-a", {"expected_topology_revision": [page["topology_revision"]]}, token=issued["credential"]) + self.assertIsNone(detail["attachments"]) + self.assertEqual(detail["attachment_visibility"], "NOT_GRANTED") + + def test_multiple_owned_attachments_preserve_logical_identity_and_revocation(self): + for agent in ("agent-one", "agent-two"): + with storage.sqlite_connection(self.database) as connection: + code = agent_trust.create_pairing_code(connection, agent)["pairing_code"] + token = agent_trust.pair(connection, {"protocol_version": "1.0", + "agent_id": agent, "pairing_code": code})["credential"] + agent_trust.heartbeat(connection, {"protocol_version": "1.0", "agent_id": agent}, token) + agent_trust.register_attachment(connection, {"protocol_version": "1.0", + "agent_id": agent, "availability": "AVAILABLE", "attachment": { + "schema_version": "1.0", "project": {"id": "project-a", "authority_repository_id": "core-a"}, + "repository": {"id": "child-a", "role": "child"}, "validation": {"kind": "none"}}}, token) + page = self.read() + detail = self.read("project-a", {"expected_topology_revision": [page["topology_revision"]]}) + self.assertEqual(len(detail["repositories"]), 2) + self.assertEqual(len(detail["attachments"]), 2) + self.assertEqual({a["availability"] for a in detail["attachments"]}, {"AVAILABLE"}) + with storage.sqlite_connection(self.database) as connection: + agent_trust.revoke(connection, "agent-one") + self.assert_refused("DISCOVERY_SNAPSHOT_CONFLICT", project="project-a", + query={"expected_topology_revision": [page["topology_revision"]]}) + page = self.read() + detail = self.read("project-a", {"expected_topology_revision": [page["topology_revision"]]}) + self.assertEqual({a["agent_id"]: a["availability"] for a in detail["attachments"]}, + {"agent-one": "UNAVAILABLE", "agent-two": "AVAILABLE"}) + before = self.database.read_bytes() + with patch.object(discovery, "_now", return_value=discovery._now() + discovery.timedelta(seconds=100)): + stale_page = self.read() + stale = self.read("project-a", {"expected_topology_revision": [stale_page["topology_revision"]]}) + self.assertEqual({a["availability"] for a in stale["attachments"]}, {"UNAVAILABLE"}) + self.assertEqual(self.database.read_bytes(), before) + + def test_every_consumer_instance_audience_binding_is_independent(self): + for field in ("EP-Instance-ID", "EP-Consumer-ID", "Workspace-Instance-ID", "EP-Audience"): + with self.subTest(field=field): + self.assert_refused("DISCOVERY_SCOPE_MISMATCH", headers={**self.headers, field: "other"}) + + def test_unknown_or_malformed_bearer_never_becomes_authority(self): + for token in ("x", "x" * 43, "é" * 43, "" , "a/b" * 15): + with self.subTest(token=token): + self.assert_refused("UNAUTHENTICATED", token=token) + + def test_ungranted_project_is_denied_not_empty(self): + revision = self.read()["topology_revision"] + self.assert_refused("DISCOVERY_PROJECT_DENIED", project="project-b", query={"expected_topology_revision": [revision]}) + + def test_issue_replay_does_not_reveal_or_mint_another_secret(self): + self.assertRaises(discovery.DiscoveryError, self.issue, "grant-a", ["project-a"]) + with storage.sqlite_read_connection(self.database) as connection: + self.assertEqual(connection.execute("SELECT COUNT(*) FROM ep_workspace_discovery_grants").fetchone()[0], 1) + self.assertNotIn("credential", discovery.status(connection, grant_id="grant-a")) + + def test_issuer_rejects_unknown_projects_fields_and_nonfinite_lifetimes(self): + for kwargs in ({"projects": ["unknown"]}, {"projects": ["project-a", "project-a"]}, + {"projects": ["project-a"], "fields": ["admin"]}, + {"projects": ["project-a"], "ttl_seconds": 0}, + {"projects": ["project-a"], "ttl_seconds": 86401}, + {"projects": ["project-a"], "ttl_seconds": True}): + arguments = dict(kwargs) + projects = arguments.pop("projects") + with self.subTest(kwargs=kwargs): + self.assertRaises(discovery.DiscoveryError, self.issue, "bad", projects, **arguments) + + def test_revoked_grant_and_idempotent_exact_revocation(self): + with storage.sqlite_connection(self.database) as connection: + discovery.revoke(connection, grant_id="grant-a", operation_id="revoke-a") + self.assertEqual(discovery.revoke(connection, grant_id="grant-a", operation_id="revoke-a")["status"], "REVOKED") + self.assertRaises(discovery.DiscoveryError, discovery.revoke, connection, grant_id="grant-a", operation_id="different") + before = self.database.read_bytes() + self.assert_refused("DISCOVERY_INACTIVE") + self.assertEqual(self.database.read_bytes(), before) + + def test_revocation_between_snapshots_prevents_any_publication(self): + def revoke(): + with storage.sqlite_connection(self.database) as connection: + discovery.revoke(connection, grant_id="grant-a", operation_id="late-revoke") + with self.after_first_read_close(revoke): + self.assert_refused("DISCOVERY_INACTIVE") + + def test_wal_cannot_publish_old_authority_after_owner_revocation_commits(self): + with storage.sqlite_connection(self.database) as connection: + self.assertEqual(connection.execute("PRAGMA journal_mode=WAL").fetchone()[0], "wal") + original = discovery._authenticate + calls = 0 + publications = [] + committed = [] + def late_revocation(connection, token, headers): + nonlocal calls + scope = original(connection, token, headers) + calls += 1 + if calls == 3: + with storage.sqlite_connection(self.database) as writer: + discovery.revoke(writer, grant_id="grant-a", operation_id="wal-late-revoke") + committed.append(True) + return scope + with patch.object(discovery, "_authenticate", side_effect=late_revocation): + with self.assertRaises(discovery.DiscoveryError) as failure: + discovery.read(self.root, token=self.token, headers=self.headers, query={}, project_id=None, + publish=lambda *args: publications.append(args)) + self.assertEqual(failure.exception.code, "DISCOVERY_JOURNAL_UNSUPPORTED") + self.assertEqual(publications, []) + self.assertEqual(committed, []) + with storage.sqlite_read_connection(self.database) as connection: + self.assertEqual(connection.execute("PRAGMA journal_mode").fetchone()[0], "wal") + + def test_expiry_between_snapshots_prevents_any_publication(self): + clock = discovery._now() + with self.after_first_read_close(lambda: None), patch.object(discovery, "_now", side_effect=[clock, clock, clock + discovery.timedelta(days=1)]): + self.assert_refused("DISCOVERY_INACTIVE") + + def test_missing_database_never_creates_or_initializes_one(self): + saved = self.database.with_suffix(".saved") + self.database.rename(saved) + with self.assertRaises(FileNotFoundError): + self.read() + self.assertFalse(self.database.exists()) + self.assertTrue(saved.exists()) + + def test_source_loss_between_snapshots_never_recreates_database(self): + saved = self.database.with_suffix(".saved") + with self.after_first_read_close(lambda: self.database.rename(saved)): + with self.assertRaises(sqlite3.OperationalError) as context: + self.read() + self.assertEqual(context.exception.sqlite_errorcode, sqlite3.SQLITE_CANTOPEN) + self.assertFalse(self.database.exists()) + self.assertTrue(saved.exists()) + + def test_topology_change_after_selection_conflicts(self): + revision = self.read()["topology_revision"] + self.register("project-a", "another-child", "child") + self.assert_refused("DISCOVERY_SNAPSHOT_CONFLICT", project="project-a", query={"expected_topology_revision": [revision]}) + + def test_topology_change_between_read_and_publication_conflicts(self): + with self.after_first_read_close(lambda: self.register("project-a", "later-child", "child")): + self.assert_refused("DISCOVERY_SNAPSHOT_CONFLICT") + + def test_cursor_pages_only_the_exact_authorized_snapshot_and_limit(self): + issued = self.issue("both", ["project-a", "project-b"]) + first = self.read(query={"limit": ["1"]}, token=issued["credential"]) + second = self.read(query={"limit": ["1"], "cursor": [first["next_cursor"]]}, token=issued["credential"]) + self.assertEqual(first["projects"][0]["project_id"], "project-a") + self.assertEqual(second["projects"][0]["project_id"], "project-b") + self.assertIsNone(second["next_cursor"]) + self.assert_refused("DISCOVERY_CURSOR_INVALID", query={"limit": ["2"], "cursor": [first["next_cursor"]]}, token=issued["credential"]) + self.assert_refused("DISCOVERY_CURSOR_INVALID", query={"limit": ["1"], "cursor": [first["next_cursor"] + "a"]}, token=issued["credential"]) + self.assert_refused("DISCOVERY_CURSOR_INVALID", query={"limit": ["1"], "cursor": [first["next_cursor"]]}) + + def test_cursor_cannot_survive_topology_change(self): + issued = self.issue("both", ["project-a", "project-b"]) + first = self.read(query={"limit": ["1"]}, token=issued["credential"]) + self.register("project-a", "changed-child", "child") + self.assert_refused("DISCOVERY_SNAPSHOT_CONFLICT", query={"limit": ["1"], "cursor": [first["next_cursor"]]}, token=issued["credential"]) + + def test_bearer_holder_cannot_forge_cursor_or_read_server_signing_key(self): + issued = self.issue("both", ["project-a", "project-b"]) + page = self.read(query={"limit": ["1"]}, token=issued["credential"]) + self.assertNotIn("cursor_key", issued) + self.assertNotIn("cursor_key", page["scope"]) + forged = discovery._cursor(issued["credential"].encode("ascii"), + dict(snapshot=page["snapshot_id"], offset=1, limit=1)) + self.assert_refused("DISCOVERY_CURSOR_INVALID", + query={"limit": ["1"], "cursor": [forged]}, token=issued["credential"]) + with storage.sqlite_read_connection(self.database) as connection: + key = bytes(connection.execute("SELECT cursor_key FROM ep_workspace_discovery_grants WHERE grant_id='both'").fetchone()[0]) + self.assertEqual(len(key), 32) + self.assertNotIn(key.hex(), json.dumps(page)) + + def test_source_refault_requires_new_snapshot_and_never_uses_cached_success(self): + page = self.read() + saved = self.database.with_suffix(".saved") + self.database.rename(saved) + with self.assertRaises(FileNotFoundError): + self.read() + self.assertFalse(self.database.exists()) + saved.rename(self.database) + restored = self.read("project-a", {"expected_topology_revision": [page["topology_revision"]]}) + self.assertEqual(restored["project"]["project_id"], "project-a") + self.database.rename(saved) + with self.assertRaises(FileNotFoundError): + self.read() + self.assertFalse(self.database.exists()) + + def test_unknown_duplicate_malformed_queries_and_contracts_refuse(self): + for query in ({"limit": ["0"]}, {"limit": ["101"]}, {"limit": ["x"]}, + {"limit": ["1", "2"]}, {"wildcard": ["true"]}): + with self.subTest(query=query): + with self.assertRaises(discovery.DiscoveryError): + self.read(query=query) + self.assert_refused("DISCOVERY_CONTRACT_UNSUPPORTED", headers={**self.headers, "EP-Discovery-Contract": "2.0"}) + self.assert_refused("DISCOVERY_REVISION_REQUIRED", project="project-a") + + +class ProjectDiscoveryMigrationTests(_DiscoveryFixture, unittest.TestCase): + def prepare_schema74(self): + with storage.sqlite_connection(self.database) as connection: + connection.execute("PRAGMA foreign_keys=OFF") + connection.execute("PRAGMA legacy_alter_table=ON") + connection.execute("DROP TABLE ep_workspace_discovery_grants") + connection.execute("ALTER TABLE ep_installations RENAME TO fixture_installations75") + connection.execute("CREATE TABLE ep_installations (instance_id TEXT PRIMARY KEY, created_at TEXT NOT NULL, schema_version INTEGER NOT NULL CHECK(schema_version BETWEEN 41 AND 74))") + connection.execute("INSERT INTO ep_installations SELECT instance_id,created_at,74 FROM fixture_installations75") + connection.execute("DROP TABLE fixture_installations75") + connection.execute("DELETE FROM engineering_schema_migrations WHERE version=75") + connection.execute("INSERT INTO engineering_schema_migrations(version) VALUES(74)") + connection.execute("UPDATE engineering_metadata SET value='74' WHERE key='installation.schema_version'") + + def registered_data(self): + with storage.sqlite_read_connection(self.database) as connection: + return {table: connection.execute("SELECT * FROM " + table + " ORDER BY 1").fetchall() + for table in ("ep_project_registrations", "ep_repository_registrations", + "ep_agent_registrations", "ep_agent_repository_attachments", + "ep_installation_pairings", "ep_installation_pairing_credentials")} + + def test_populated74_upgrade_preserves_registered_data_identity_and_foreign_targets(self): + self.prepare_schema74() + data = self.registered_data() + self.assertEqual(server.initialize(self.root).instance_id, self.instance) + self.assertEqual(self.registered_data(), data) + with storage.sqlite_read_connection(self.database) as connection: + self.assertEqual(connection.execute("PRAGMA foreign_key_check").fetchall(), []) + discovery.validate_schema(connection) + self.assertEqual(connection.execute("SELECT schema_version FROM ep_installations").fetchall(), [(75,)]) + self.assertEqual(connection.execute("SELECT MAX(version) FROM engineering_schema_migrations").fetchone(), (75,)) + before = self.database.read_bytes() + self.assertEqual(server.initialize(self.root).instance_id, self.instance) + self.assertEqual(self.database.read_bytes(), before) + issued = self.issue("after-upgrade", ["project-a"]) + self.assertEqual(issued["status"], "ACTIVE") + + def test_failed75_migration_rolls_back_original_database_bytes_and_data(self): + self.prepare_schema74() + data = self.registered_data() + before = self.database.read_bytes() + with patch.object(discovery, "install_schema", side_effect=sqlite3.OperationalError("synthetic migration boundary failure")): + self.assertRaises(server.ServerConfigurationError, server.initialize, self.root) + self.assertEqual(self.database.read_bytes(), before) + self.assertEqual(self.registered_data(), data) + with storage.sqlite_read_connection(self.database) as connection: + self.assertEqual(connection.execute("SELECT MAX(version) FROM engineering_schema_migrations").fetchone(), (74,)) + self.assertEqual(connection.execute("SELECT schema_version FROM ep_installations").fetchall(), [(74,)]) + self.assertIsNone(connection.execute("SELECT name FROM sqlite_master WHERE name='ep_workspace_discovery_grants'").fetchone()) + + +class ProjectDiscoveryHTTPTests(_DiscoveryFixture, unittest.TestCase): + def setUp(self): + super().setUp() + self.listener = server._ConsoleHTTPServer(("127.0.0.1", 0), server._HealthHandler) + self.listener.daemon_threads = False + self.listener.data_root = self.root + self.thread = threading.Thread(target=self.listener.serve_forever, daemon=True) + self.thread.start() + self.addCleanup(self.stop_listener) + + def stop_listener(self): + self.listener.shutdown() + self.listener.server_close() + self.thread.join(5) + self.assertFalse(self.thread.is_alive()) + + def request(self, path="/v1/onboarding/projects", headers=None, body=None): + connection = http.client.HTTPConnection("127.0.0.1", self.listener.server_address[1], timeout=5) + try: + connection.request("GET", path, body=body, headers={"Authorization": "Bearer " + self.token, + **(self.headers if headers is None else headers)}) + response = connection.getresponse() + payload = json.loads(response.read()) + self.assertEqual(response.getheader("Cache-Control"), "no-store") + return response.status, payload + finally: + connection.close() + + def test_actual_http_discovery_selected_topology_and_no_database_changes(self): + before = self.database.read_bytes() + code, page = self.request() + self.assertEqual(code, 200) + code, detail = self.request("/v1/onboarding/projects/project-a/topology?expected_topology_revision=" + page["topology_revision"]) + self.assertEqual(code, 200) + self.assertEqual(detail["project"]["authority_repository_id"], "core-a") + self.assertEqual(len(detail["repositories"]), 2) + self.assertEqual(self.database.read_bytes(), before) + + def test_actual_http_responses_validate_against_packaged_closed_schema(self): + schema = json.loads((Path(discovery.__file__).parent / "schemas" / + "workspace-project-discovery-v1.schema.json").read_text()) + Draft202012Validator.check_schema(schema) + validator = Draft202012Validator(schema, format_checker=FormatChecker()) + code, page = self.request() + self.assertEqual(code, 200) + validator.validate(page) + code, detail = self.request("/v1/onboarding/projects/project-a/topology?expected_topology_revision=" + page["topology_revision"]) + self.assertEqual(code, 200) + validator.validate(detail) + self.assertTrue(list(validator.iter_errors({**page, "unauthorized_extra_field": "x"}))) + + def test_actual_http_revocation_wrong_audience_body_and_unsupported(self): + self.assertEqual(self.request(headers={**self.headers, "EP-Audience": "other"})[0], 403) + self.assertEqual(self.request(headers={**self.headers, "EP-Discovery-Contract": "2.0"})[0], 422) + self.assertEqual(self.request(body="not-allowed")[0], 400) + with storage.sqlite_connection(self.database) as connection: + discovery.revoke(connection, grant_id="grant-a", operation_id="http-revoke") + self.assertEqual(self.request()[0], 401) + + def test_actual_http_missing_source_is503_and_never_creates_database(self): + self.database.rename(self.database.with_suffix(".saved")) + self.assertEqual(self.request(), (503, {"error": "CENTRAL_UNAVAILABLE"})) + self.assertFalse(self.database.exists()) + + def test_actual_http_corruption_stays_red_not_a_healthy_empty_projection(self): + self.database.rename(self.database.with_suffix(".saved")) + self.database.write_bytes(b"corrupt synthetic qualification database") + corrupted = self.database.read_bytes() + self.assertEqual(self.request(), (500, {"error": "CENTRAL_CORRUPT"})) + self.assertEqual(self.database.read_bytes(), corrupted) + + def test_actual_http_source_loss_during_response_encoding_is503_before_headers(self): + original = json.dumps + saved = self.database.with_suffix(".saved") + def disappear(payload, *args, **kwargs): + if isinstance(payload, dict) and payload.get("contract_version") == "1.0": + self.database.rename(saved) + return original(payload, *args, **kwargs) + with patch.object(server.json, "dumps", side_effect=disappear): + self.assertEqual(self.request(), (503, {"error": "CENTRAL_UNAVAILABLE"})) + self.assertFalse(self.database.exists()) + self.assertTrue(saved.exists()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/engineering/test_server_foundation.py b/tests/engineering/test_server_foundation.py index 4bfc5e5c..98e6c433 100644 --- a/tests/engineering/test_server_foundation.py +++ b/tests/engineering/test_server_foundation.py @@ -52,8 +52,8 @@ def tearDown(self) -> None: self.temporary.cleanup() def _prepare_pre74_schema(self, connection) -> None: - """Historical schemas had no installation-pairing tables or rows.""" - for table in ('ep_installation_pairing_credentials','ep_installation_pairings'): + """Historical schemas had neither pairing nor discovery ledgers.""" + for table in ('ep_workspace_discovery_grants', 'ep_installation_pairing_credentials','ep_installation_pairings'): if connection.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?",(table,)).fetchone(): self.assertEqual(connection.execute('SELECT count(*) FROM '+table).fetchone()[0],0) connection.execute('DROP TABLE '+table) From 39a85674459ad017daf1e8801e3d6ce5bbb643b1 Mon Sep 17 00:00:00 2001 From: Peter van Tol Date: Wed, 14 Oct 2026 10:39:43 +0200 Subject: [PATCH 2/5] Fence final discovery publication and preserve failed-stop evidence --- src/engineering_platform/project_discovery.py | 4 +- tests/engineering/test_project_discovery.py | 104 +++++++++++++++++- 2 files changed, 104 insertions(+), 4 deletions(-) diff --git a/src/engineering_platform/project_discovery.py b/src/engineering_platform/project_discovery.py index df1395bc..3dcd8f50 100644 --- a/src/engineering_platform/project_discovery.py +++ b/src/engineering_platform/project_discovery.py @@ -377,9 +377,9 @@ def read(data_root: Path, *, token: str | None, headers: dict[str, str], query: attachment_visibility="GRANTED" if selected["attachments"] is not None else "NOT_GRANTED", completeness="COMPLETE") def publication_check() -> None: - if _identity(path) != identity: - raise DiscoveryError("DISCOVERY_SNAPSHOT_CONFLICT", 409) if _authenticate(connection, token, headers) != scope: raise DiscoveryError("DISCOVERY_SNAPSHOT_CONFLICT", 409) + if _identity(path) != identity: + raise DiscoveryError("DISCOVERY_SNAPSHOT_CONFLICT", 409) publication_check() publish(payload, str(scope["instance_id"]), publication_check) diff --git a/tests/engineering/test_project_discovery.py b/tests/engineering/test_project_discovery.py index 5f7dd447..74cc8183 100644 --- a/tests/engineering/test_project_discovery.py +++ b/tests/engineering/test_project_discovery.py @@ -19,8 +19,8 @@ class _DiscoveryFixture: def setUp(self): - self.temporary = tempfile.TemporaryDirectory() - self.addCleanup(self.temporary.cleanup) + self.temporary = tempfile.TemporaryDirectory(delete=False) + self.addCleanup(self.cleanup_store) self.root = Path(self.temporary.name) self.instance = server.initialize(self.root).instance_id self.database = self.root / server.SERVER_DATABASE_FILENAME @@ -32,6 +32,10 @@ def setUp(self): "EP-Consumer-ID": "workspace-consumer", "Workspace-Instance-ID": "workspace-instance", "EP-Audience": "workspace"} + def cleanup_store(self): + if getattr(self, "listener_stopped", True): + self.temporary.cleanup() + def register(self, project, repository, role): with storage.sqlite_connection(self.database) as connection: return project_topology.register_server_local_topology(connection, declaration={ @@ -395,6 +399,7 @@ def test_failed75_migration_rolls_back_original_database_bytes_and_data(self): class ProjectDiscoveryHTTPTests(_DiscoveryFixture, unittest.TestCase): def setUp(self): super().setUp() + self.listener_stopped = False self.listener = server._ConsoleHTTPServer(("127.0.0.1", 0), server._HealthHandler) self.listener.daemon_threads = False self.listener.data_root = self.root @@ -407,6 +412,61 @@ def stop_listener(self): self.listener.server_close() self.thread.join(5) self.assertFalse(self.thread.is_alive()) + self.listener_stopped = True + + def test_active_handler_is_joined_before_store_cleanup(self): + entered, release, stopped = threading.Event(), threading.Event(), threading.Event() + original = server._HealthHandler._send + responses, errors = [], [] + def hold(handler, *args, **kwargs): + if args[0] == 200 and isinstance(args[1], dict) and "contract_version" in args[1]: + entered.set() + if not release.wait(5): + raise TimeoutError("bounded own handler hold expired") + return original(handler, *args, **kwargs) + def client(): + try: + responses.append(self.request()) + except BaseException as error: + errors.append(error) + def stopper(): + try: + self.stop_listener() + except BaseException as error: + errors.append(error) + finally: + stopped.set() + with patch.object(server._HealthHandler, "_send", hold): + request = threading.Thread(target=client) + stop = threading.Thread(target=stopper) + request.start() + try: + self.assertTrue(entered.wait(3)) + stop.start() + self.assertFalse(stopped.wait(0.1)) + self.cleanup_store() + self.assertTrue(self.database.exists()) + finally: + release.set() + request.join(5) + if stop.ident is not None: + stop.join(5) + self.assertFalse(request.is_alive()) + self.assertFalse(stop.is_alive()) + self.assertEqual(errors, []) + self.assertEqual(responses[0][0], 200) + self.assertTrue(self.listener_stopped) + + def test_failed_listener_stop_preserves_store_until_real_stop(self): + before = self.database.read_bytes() + with patch.object(self.listener, "server_close", side_effect=RuntimeError("synthetic unknown stop")): + self.assertRaisesRegex(RuntimeError, "synthetic unknown stop", self.stop_listener) + self.assertFalse(self.listener_stopped) + self.cleanup_store() + self.assertTrue(self.database.exists()) + self.assertEqual(self.database.read_bytes(), before) + self.stop_listener() + self.assertTrue(self.listener_stopped) def request(self, path="/v1/onboarding/projects", headers=None, body=None): connection = http.client.HTTPConnection("127.0.0.1", self.listener.server_address[1], timeout=5) @@ -475,6 +535,46 @@ def disappear(payload, *args, **kwargs): self.assertFalse(self.database.exists()) self.assertTrue(saved.exists()) + def test_actual_http_source_loss_during_final_authority_check_is503(self): + original = discovery._authenticate + calls = 0 + saved = self.database.with_suffix(".saved") + def lose_source(connection, token, headers): + nonlocal calls + result = original(connection, token, headers) + calls += 1 + if calls == 5: + self.database.rename(saved) + return result + with patch.object(discovery, "_authenticate", side_effect=lose_source): + self.assertEqual(self.request(), (503, {"error": "CENTRAL_UNAVAILABLE"})) + self.assertEqual(calls, 5) + self.assertFalse(self.database.exists()) + self.assertTrue(saved.exists()) + + def test_actual_http_different_instance_source_replacement_cannot_publish_old_success(self): + replacement = self.root / "replacement" + replacement_instance = server.initialize(replacement).instance_id + self.assertNotEqual(replacement_instance, self.instance) + replacement_database = replacement / server.SERVER_DATABASE_FILENAME + expected = replacement_database.read_bytes() + original = discovery._authenticate + calls = 0 + saved = self.database.with_suffix(".saved") + def replace_source(connection, token, headers): + nonlocal calls + result = original(connection, token, headers) + calls += 1 + if calls == 5: + self.database.rename(saved) + replacement_database.rename(self.database) + return result + with patch.object(discovery, "_authenticate", side_effect=replace_source): + self.assertEqual(self.request(), (409, {"error": "DISCOVERY_SNAPSHOT_CONFLICT"})) + self.assertEqual(calls, 5) + self.assertEqual(self.database.read_bytes(), expected) + self.assertTrue(saved.exists()) + if __name__ == "__main__": unittest.main() From 6a7037e589ee0b9f32758e537fa88a367cedb8ad Mon Sep 17 00:00:00 2001 From: Peter van Tol Date: Wed, 14 Oct 2026 10:47:03 +0200 Subject: [PATCH 3/5] Prove retained credentials and installed revocation commit fence --- tests/engineering/test_project_discovery.py | 71 ++++++++++++++++++++- 1 file changed, 70 insertions(+), 1 deletion(-) diff --git a/tests/engineering/test_project_discovery.py b/tests/engineering/test_project_discovery.py index 74cc8183..4fab503c 100644 --- a/tests/engineering/test_project_discovery.py +++ b/tests/engineering/test_project_discovery.py @@ -9,11 +9,12 @@ from pathlib import Path import tempfile import threading +import time import unittest from unittest.mock import patch from jsonschema import Draft202012Validator, FormatChecker -from engineering_platform import agent_trust, project_discovery as discovery, project_topology, server, storage +from engineering_platform import agent_trust, installation_pairing, project_discovery as discovery, project_topology, server, storage class _DiscoveryFixture: @@ -347,6 +348,21 @@ def test_unknown_duplicate_malformed_queries_and_contracts_refuse(self): class ProjectDiscoveryMigrationTests(_DiscoveryFixture, unittest.TestCase): def prepare_schema74(self): + # Populate retained pre75 ledgers through their owning primitives. + with storage.sqlite_connection(self.database) as connection: + agent = "migration-agent" + code = agent_trust.create_pairing_code(connection, agent)["pairing_code"] + self.agent_credential = agent_trust.pair(connection, {"protocol_version": "1.0", + "agent_id": agent, "pairing_code": code})["credential"] + agent_trust.heartbeat(connection, {"protocol_version": "1.0", "agent_id": agent}, self.agent_credential) + agent_trust.register_attachment(connection, {"protocol_version": "1.0", "agent_id": agent, + "availability": "AVAILABLE", "attachment": {"schema_version": "1.0", + "project": {"id": "project-a", "authority_repository_id": "core-a"}, + "repository": {"id": "child-a", "role": "child"}, "validation": {"kind": "none"}}}, self.agent_credential) + installation_pairing.register(connection, binding_id="migration-pairing", ep_instance_id=self.instance, + forge_instance_id="migration-forge", consumer_id="forge", operation_id="migration-register") + self.pairing_credential = installation_pairing.issue(connection, + binding_id="migration-pairing", operation_id="migration-pairing-issue").credential with storage.sqlite_connection(self.database) as connection: connection.execute("PRAGMA foreign_keys=OFF") connection.execute("PRAGMA legacy_alter_table=ON") @@ -369,6 +385,7 @@ def registered_data(self): def test_populated74_upgrade_preserves_registered_data_identity_and_foreign_targets(self): self.prepare_schema74() data = self.registered_data() + self.assertTrue(all(data.values())) self.assertEqual(server.initialize(self.root).instance_id, self.instance) self.assertEqual(self.registered_data(), data) with storage.sqlite_read_connection(self.database) as connection: @@ -376,6 +393,10 @@ def test_populated74_upgrade_preserves_registered_data_identity_and_foreign_targ discovery.validate_schema(connection) self.assertEqual(connection.execute("SELECT schema_version FROM ep_installations").fetchall(), [(75,)]) self.assertEqual(connection.execute("SELECT MAX(version) FROM engineering_schema_migrations").fetchone(), (75,)) + self.assertIsNotNone(installation_pairing.authenticate(connection, self.pairing_credential)) + agent_trust.authenticate(connection, "migration-agent", self.agent_credential) + self.assertNotIn(self.pairing_credential.encode(), self.database.read_bytes()) + self.assertNotIn(self.agent_credential.encode(), self.database.read_bytes()) before = self.database.read_bytes() self.assertEqual(server.initialize(self.root).instance_id, self.instance) self.assertEqual(self.database.read_bytes(), before) @@ -468,6 +489,54 @@ def test_failed_listener_stop_preserves_store_until_real_stop(self): self.stop_listener() self.assertTrue(self.listener_stopped) + def test_supported_journal_holds_owner_revocation_commit_until_publication_finishes(self): + entered, release, staged, committed = (threading.Event() for _ in range(4)) + original = server._HealthHandler._send + responses, errors, publication_times, commit_times = [], [], [], [] + def hold(handler, *args, **kwargs): + if args[0] == 200 and isinstance(args[1], dict) and "contract_version" in args[1]: + entered.set() + if not release.wait(4): + raise TimeoutError("bounded own publication hold expired") + result = original(handler, *args, **kwargs) + publication_times.append(time.monotonic()) + return result + return original(handler, *args, **kwargs) + def client(): + try: + responses.append(self.request()) + except BaseException as error: + errors.append(error) + def revoke(): + try: + with storage.sqlite_connection(self.database, timeout=3) as writer: + discovery.revoke(writer, grant_id="grant-a", operation_id="held-publication-revoke") + staged.set() + commit_times.append(time.monotonic()) + committed.set() + except BaseException as error: + errors.append(error) + request, writer = threading.Thread(target=client), threading.Thread(target=revoke) + with patch.object(server._HealthHandler, "_send", hold): + request.start() + try: + self.assertTrue(entered.wait(2)) + writer.start() + self.assertTrue(staged.wait(1)) + self.assertFalse(committed.wait(0.1)) + finally: + release.set() + request.join(5) + if writer.ident is not None: + writer.join(5) + self.assertFalse(request.is_alive()) + self.assertFalse(writer.is_alive()) + self.assertEqual(errors, []) + self.assertTrue(committed.is_set()) + self.assertLess(publication_times[0], commit_times[0]) + self.assertEqual(responses[0][0], 200) + self.assertEqual(self.request()[0], 401) + def request(self, path="/v1/onboarding/projects", headers=None, body=None): connection = http.client.HTTPConnection("127.0.0.1", self.listener.server_address[1], timeout=5) try: From f43ff7b7f8e2941bcfd55c53532a498ea352eb0f Mon Sep 17 00:00:00 2001 From: Peter van Tol Date: Wed, 14 Oct 2026 11:55:02 +0200 Subject: [PATCH 4/5] Bind discovery Postman qualification and preserve pre75 fixture shapes --- ...erver-http-surface.postman_collection.json | 111 +++++++----------- .../http-json-api.postman_collection.json | 60 ++++++++-- .../test_central_operational_reset.py | 2 + tests/engineering/test_component_logging.py | 2 +- .../test_owner_credential_recovery.py | 4 +- .../test_receipt_run_provenance.py | 3 + .../ep_server_postman_surface.py | 7 ++ .../http_json_postman_contract.py | 27 ++++- 8 files changed, 131 insertions(+), 85 deletions(-) diff --git a/tests/engineering/postman/ep-server-http-surface.postman_collection.json b/tests/engineering/postman/ep-server-http-surface.postman_collection.json index 9adec3e1..ae151beb 100644 --- a/tests/engineering/postman/ep-server-http-surface.postman_collection.json +++ b/tests/engineering/postman/ep-server-http-surface.postman_collection.json @@ -1218,90 +1218,65 @@ ] }, { - "name": "Unknown route", + "name": "Workspace discovery requires separate credential", "request": { "method": "GET", - "header": [], - "url": "{{baseUrl}}/not-an-ep-route" + "header": [ + { + "key": "EP-Discovery-Contract", + "value": "1.0" + } + ], + "url": "{{baseUrl}}/v1/onboarding/projects" }, "event": [ { "listen": "test", "script": { "exec": [ - "pm.test('expected status', () => pm.response.to.have.status(404));" + "pm.test('expected status', () => pm.response.to.have.status(401));" ] } } ] }, { - "name": "HOB-EP01 bounded project discovery", - "item": [ + "name": "Workspace topology requires separate credential", + "request": { + "method": "GET", + "header": [ + { + "key": "EP-Discovery-Contract", + "value": "1.0" + } + ], + "url": "{{baseUrl}}/v1/onboarding/projects/postman-project/topology?expected_topology_revision=0000000000000000000000000000000000000000000000000000000000000000" + }, + "event": [ { - "name": "DiscoverySnapshot", - "request": { - "method": "GET", - "header": [ - { - "key": "Authorization", - "value": "Bearer {{ep_discovery_bearer}}" - }, - { - "key": "EP-Discovery-Contract", - "value": "1.0" - }, - { - "key": "EP-Instance-ID", - "value": "{{ep_instance_id}}" - }, - { - "key": "EP-Consumer-ID", - "value": "{{workspace_consumer_id}}" - }, - { - "key": "Workspace-Instance-ID", - "value": "{{workspace_instance_id}}" - }, - { - "key": "EP-Audience", - "value": "workspace" - } - ], - "url": "{{base_url}}/v1/onboarding/projects?limit=25" - } - }, + "listen": "test", + "script": { + "exec": [ + "pm.test('expected status', () => pm.response.to.have.status(401));" + ] + } + } + ] + }, + { + "name": "Unknown route", + "request": { + "method": "GET", + "header": [], + "url": "{{baseUrl}}/not-an-ep-route" + }, + "event": [ { - "name": "TopologySnapshot selected exact revision", - "request": { - "method": "GET", - "header": [ - { - "key": "Authorization", - "value": "Bearer {{ep_discovery_bearer}}" - }, - { - "key": "EP-Discovery-Contract", - "value": "1.0" - }, - { - "key": "EP-Instance-ID", - "value": "{{ep_instance_id}}" - }, - { - "key": "EP-Consumer-ID", - "value": "{{workspace_consumer_id}}" - }, - { - "key": "Workspace-Instance-ID", - "value": "{{workspace_instance_id}}" - }, - { - "key": "EP-Audience", - "value": "workspace" - } - ], - "url": "{{base_url}}/v1/onboarding/projects/{{project_id}}/topology?expected_topology_revision={{topology_revision}}" + "listen": "test", + "script": { + "exec": [ + "pm.test('expected status', () => pm.response.to.have.status(404));" + ] } } ] diff --git a/tests/engineering/postman/http-json-api.postman_collection.json b/tests/engineering/postman/http-json-api.postman_collection.json index 2868c13a..508a62ab 100644 --- a/tests/engineering/postman/http-json-api.postman_collection.json +++ b/tests/engineering/postman/http-json-api.postman_collection.json @@ -9,6 +9,26 @@ { "key": "baseUrl", "value": "http://127.0.0.1:8765" + }, + { + "key": "ep_discovery_bearer", + "value": "" + }, + { + "key": "ep_discovery_instance", + "value": "" + }, + { + "key": "ep_discovery_consumer", + "value": "" + }, + { + "key": "ep_discovery_workspace", + "value": "" + }, + { + "key": "ep_discovery_revision", + "value": "" } ], "item": [ @@ -305,23 +325,33 @@ }, { "key": "EP-Instance-ID", - "value": "{{ep_instance_id}}" + "value": "{{ep_discovery_instance}}" }, { "key": "EP-Consumer-ID", - "value": "{{workspace_consumer_id}}" + "value": "{{ep_discovery_consumer}}" }, { "key": "Workspace-Instance-ID", - "value": "{{workspace_instance_id}}" + "value": "{{ep_discovery_workspace}}" }, { "key": "EP-Audience", "value": "workspace" } ], - "url": "{{base_url}}/v1/onboarding/projects?limit=25" - } + "url": "{{baseUrl}}/v1/onboarding/projects" + }, + "event": [ + { + "listen": "test", + "script": { + "exec": [ + "pm.test('expected status', () => pm.response.to.have.status(200));" + ] + } + } + ] }, { "name": "TopologySnapshot selected exact revision", @@ -338,23 +368,33 @@ }, { "key": "EP-Instance-ID", - "value": "{{ep_instance_id}}" + "value": "{{ep_discovery_instance}}" }, { "key": "EP-Consumer-ID", - "value": "{{workspace_consumer_id}}" + "value": "{{ep_discovery_consumer}}" }, { "key": "Workspace-Instance-ID", - "value": "{{workspace_instance_id}}" + "value": "{{ep_discovery_workspace}}" }, { "key": "EP-Audience", "value": "workspace" } ], - "url": "{{base_url}}/v1/onboarding/projects/{{project_id}}/topology?expected_topology_revision={{topology_revision}}" - } + "url": "{{baseUrl}}/v1/onboarding/projects/:project_id/topology?expected_topology_revision={{ep_discovery_revision}}" + }, + "event": [ + { + "listen": "test", + "script": { + "exec": [ + "pm.test('expected status', () => pm.response.to.have.status(200));" + ] + } + } + ] } ] } diff --git a/tests/engineering/test_central_operational_reset.py b/tests/engineering/test_central_operational_reset.py index a505f7aa..1bde8473 100644 --- a/tests/engineering/test_central_operational_reset.py +++ b/tests/engineering/test_central_operational_reset.py @@ -827,6 +827,8 @@ def test_chat_parent_migration_preserves_rows_and_enforces_canonical_run(self) - connection.execute("DELETE FROM engineering_schema_migrations WHERE version>=68") connection.execute("INSERT INTO engineering_schema_migrations(version) VALUES(67)") connection.execute("UPDATE engineering_metadata SET value='67' WHERE key='installation.schema_version'") + self.assertEqual(connection.execute("SELECT count(*) FROM ep_workspace_discovery_grants").fetchone()[0], 0) + connection.execute("DROP TABLE ep_workspace_discovery_grants") connection.execute("DROP TABLE ep_installation_pairing_credentials") connection.execute("DROP TABLE ep_installation_pairings") connection.execute("ALTER TABLE ep_installations RENAME TO installation_schema68") diff --git a/tests/engineering/test_component_logging.py b/tests/engineering/test_component_logging.py index ea8978a3..e548b415 100644 --- a/tests/engineering/test_component_logging.py +++ b/tests/engineering/test_component_logging.py @@ -237,7 +237,7 @@ def test_schema_62_store_upgrades_additively_to_shielded_diagnostics(self) -> No connection.execute("DROP TABLE ep_consumer_credential_recovery_operations") connection.execute("INSERT INTO engineering_schema_migrations(version) VALUES(62)") connection.execute("UPDATE engineering_metadata SET value='62' WHERE key='installation.schema_version'") - for table in ('ep_installation_pairing_credentials','ep_installation_pairings'): + for table in ('ep_workspace_discovery_grants', 'ep_installation_pairing_credentials','ep_installation_pairings'): self.assertEqual(connection.execute('SELECT count(*) FROM '+table).fetchone()[0],0) connection.execute('DROP TABLE '+table) connection.execute("ALTER TABLE ep_installations RENAME TO ep_installations_schema63_fixture") diff --git a/tests/engineering/test_owner_credential_recovery.py b/tests/engineering/test_owner_credential_recovery.py index 2c40f88e..00702e5e 100644 --- a/tests/engineering/test_owner_credential_recovery.py +++ b/tests/engineering/test_owner_credential_recovery.py @@ -1533,7 +1533,7 @@ def test_schema_63_installation_upgrades_to_durable_recovery_journal(self) -> No "UPDATE engineering_metadata SET value='63' " "WHERE key='installation.schema_version'" ) - for table in ('ep_installation_pairing_credentials','ep_installation_pairings'): + for table in ('ep_workspace_discovery_grants', 'ep_installation_pairing_credentials','ep_installation_pairings'): self.assertEqual(connection.execute('SELECT count(*) FROM '+table).fetchone()[0],0) connection.execute('DROP TABLE '+table) connection.execute("ALTER TABLE ep_installations RENAME TO ep_installations_schema65") @@ -1592,7 +1592,7 @@ def test_schema_64_migration_preserves_legacy_uncertain_and_successor_receipts(s with sqlite3.connect(self.root / server.SERVER_DATABASE_FILENAME) as connection: connection.execute("PRAGMA foreign_keys=OFF") connection.execute("PRAGMA legacy_alter_table=ON") - for table in ('ep_installation_pairing_credentials','ep_installation_pairings'): + for table in ('ep_workspace_discovery_grants', 'ep_installation_pairing_credentials','ep_installation_pairings'): self.assertEqual(connection.execute('SELECT count(*) FROM '+table).fetchone()[0],0) connection.execute('DROP TABLE '+table) connection.execute("ALTER TABLE ep_installations RENAME TO ep_installations_schema65") diff --git a/tests/engineering/test_receipt_run_provenance.py b/tests/engineering/test_receipt_run_provenance.py index 05d36f6f..217c8f79 100644 --- a/tests/engineering/test_receipt_run_provenance.py +++ b/tests/engineering/test_receipt_run_provenance.py @@ -66,6 +66,9 @@ def _downgrade_to_schema_51(self) -> None: db.execute("DROP TRIGGER ep_receipt_run_provenance_immutable_update") db.execute("DROP TRIGGER ep_receipt_run_provenance_immutable_delete") db.execute("DROP TABLE ep_receipt_run_provenance") + for table in ("ep_workspace_discovery_grants", "ep_installation_pairing_credentials", "ep_installation_pairings"): + self.assertEqual(db.execute("SELECT count(*) FROM " + table).fetchone()[0], 0) + db.execute("DROP TABLE " + table) db.execute("ALTER TABLE ep_installations RENAME TO ep_installations_schema52") db.execute("CREATE TABLE ep_installations (instance_id TEXT PRIMARY KEY, created_at TEXT NOT NULL, schema_version INTEGER NOT NULL CHECK(schema_version IN (41,42,43,44,45,46,47,48,49,50,51)))") db.execute("INSERT INTO ep_installations(instance_id,created_at,schema_version) SELECT instance_id,created_at,51 FROM ep_installations_schema52") diff --git a/tools/qualification/ep_server_postman_surface.py b/tools/qualification/ep_server_postman_surface.py index a3dd37ba..ef3e5661 100644 --- a/tools/qualification/ep_server_postman_surface.py +++ b/tools/qualification/ep_server_postman_surface.py @@ -20,6 +20,7 @@ class Case: body: str | None = None project: bool = False origin: bool = False + headers: tuple[tuple[str, str], ...] = () # One concrete case for every route family in console_route_ownership, plus @@ -82,6 +83,11 @@ class Case: Case("Evidence artifact requires credential", "GET", "/v1/projects/postman-project/artifacts/terminal-evidence:missing", 401), Case("Agent pairing rejects malformed request", "POST", "/v1/agent/pair", 400, "{}"), Case("Agent registration requires credential", "POST", "/v1/agent/register", 401, "{}"), + Case("Workspace discovery requires separate credential", "GET", "/v1/onboarding/projects", 401, + headers=(("EP-Discovery-Contract", "1.0"),)), + Case("Workspace topology requires separate credential", "GET", + "/v1/onboarding/projects/postman-project/topology?expected_topology_revision=" + "0" * 64, 401, + headers=(("EP-Discovery-Contract", "1.0"),)), Case("Unknown route", "GET", "/not-an-ep-route", 404), ) @@ -92,6 +98,7 @@ def item(case: Case) -> dict[str, object]: if case.body is not None else []) if case.origin: headers.append({"key": "Origin", "value": "{{baseUrl}}"}) + headers.extend({"key": key, "value": value} for key, value in case.headers) suffix = "?project=postman-project" if case.project else "" request: dict[str, object] = {"method": case.method, "header": headers, "url": "{{baseUrl}}" + case.path + suffix} diff --git a/tools/qualification/http_json_postman_contract.py b/tools/qualification/http_json_postman_contract.py index 0c4c11d9..3399789d 100644 --- a/tools/qualification/http_json_postman_contract.py +++ b/tools/qualification/http_json_postman_contract.py @@ -16,7 +16,8 @@ from urllib.error import HTTPError from urllib.request import Request, urlopen -from engineering_platform import server, submission_service +from engineering_platform import project_discovery, server, submission_service +from jsonschema import Draft202012Validator, FormatChecker from ep_server_postman_surface import postman_collection @@ -93,7 +94,7 @@ def _port() -> int: return int(probe.getsockname()[1]) -def _request(base_url: str, item: dict[str, object]) -> int: +def _request(base_url: str, item: dict[str, object], discovery_binding: dict[str, str] | None = None) -> int: request = item["request"] assert isinstance(request, dict) method, path = _collection_operation(item) @@ -105,10 +106,21 @@ def _request(base_url: str, item: dict[str, object]) -> int: url += "?" + raw_url.split("?", 1)[1] headers = {str(header["key"]): str(header["value"]).replace("{{baseUrl}}", base_url) for header in request.get("header", []) if isinstance(header, dict) and "key" in header and "value" in header} + for key, value in (discovery_binding or {}).items(): + placeholder = "{{" + key + "}}" + url = url.replace(placeholder, value) + headers = {name: content.replace(placeholder, value) for name, content in headers.items()} body = request.get("body", {}) data = str(body.get("raw", "")).encode("utf-8") if isinstance(body, dict) and method != "GET" else None try: with urlopen(Request(url, data=data, headers=headers, method=method), timeout=3) as response: + if path.startswith("/v1/onboarding/projects") and response.status == 200: + payload = json.loads(response.read()) + schema = json.loads((Path(project_discovery.__file__).parent / "schemas" / + "workspace-project-discovery-v1.schema.json").read_text()) + Draft202012Validator(schema, format_checker=FormatChecker()).validate(payload) + if discovery_binding is not None and path == "/v1/onboarding/projects": + discovery_binding["ep_discovery_revision"] = payload["topology_revision"] return response.status except HTTPError as error: return error.code @@ -217,12 +229,19 @@ def main(argv: list[str] | None = None) -> int: with tempfile.TemporaryDirectory(prefix="ep-postman-contract-") as temporary: data_root = Path(temporary) / "data" port = _port() - server.initialize(data_root, bind_port=port) + identity = server.initialize(data_root, bind_port=port) with sqlite_connection(data_root / server.SERVER_DATABASE_FILENAME) as connection: server.project_topology.register_server_local_topology(connection, declaration={ "schema_version": "1.0", "project": {"id": "postman-project", "authority_repository_id": "postman-repository"}, "repository": {"id": "postman-repository", "role": "authority"}, "validation": {"kind": "none"}, }) + issued = project_discovery.issue(connection, grant_id="postman-discovery", + instance_id=identity.instance_id, consumer_id="postman-workspace", + workspace_instance_id="postman-workspace-instance", project_ids=["postman-project"], + fields=sorted(project_discovery.FIELDS), operation_id="postman-discovery-issue", ttl_seconds=900) + discovery_binding = {"ep_discovery_bearer": issued["credential"], + "ep_discovery_instance": identity.instance_id, "ep_discovery_consumer": "postman-workspace", + "ep_discovery_workspace": "postman-workspace-instance"} http_server = http.server.ThreadingHTTPServer(("127.0.0.1", port), server._HealthHandler) http_server.data_root = data_root # type: ignore[attr-defined] http_server.central_data_transfer_lock = Lock() # type: ignore[attr-defined] @@ -239,7 +258,7 @@ def main(argv: list[str] | None = None) -> int: if served_openapi != expected_openapi: raise RuntimeError("API_OPENAPI_DRIFT") for item in items: - observed, expected = _request(base_url, item), _expected_status(item) + observed, expected = _request(base_url, item, discovery_binding), _expected_status(item) if observed != expected: raise RuntimeError(f"API_POSTMAN_DRIFT {item.get('name')}: expected {expected}, got {observed}") for item in _collection_items(surface_collection.get("item")): From c218460aac7cf362598c11048f9f09df4706f375 Mon Sep 17 00:00:00 2001 From: Peter van Tol Date: Wed, 14 Oct 2026 12:08:44 +0200 Subject: [PATCH 5/5] Join owning Postman handlers before fixture cleanup --- .../test_postman_helper_lifecycle.py | 112 ++++++++++++++++++ .../http_json_postman_contract.py | 36 +++++- 2 files changed, 145 insertions(+), 3 deletions(-) create mode 100644 tests/engineering/test_postman_helper_lifecycle.py diff --git a/tests/engineering/test_postman_helper_lifecycle.py b/tests/engineering/test_postman_helper_lifecycle.py new file mode 100644 index 00000000..d624914a --- /dev/null +++ b/tests/engineering/test_postman_helper_lifecycle.py @@ -0,0 +1,112 @@ +"""Real handler-stop and failed-cleanup boundaries for the owning HTTP helper.""" +from __future__ import annotations + +import http.server +from pathlib import Path +import socket +import sys +import threading +import unittest +from unittest.mock import patch + + +_TOOLS = Path(__file__).resolve().parents[2] / "tools" / "qualification" +sys.path.insert(0, str(_TOOLS)) +import http_json_postman_contract as helper +sys.path.pop(0) + + +class PostmanHelperLifecycleTest(unittest.TestCase): + def test_held_real_handler_prevents_fixture_cleanup(self): + entered, release, stopped = threading.Event(), threading.Event(), threading.Event() + + class Handler(http.server.BaseHTTPRequestHandler): + def do_GET(self): + entered.set() + release.wait(5) + self.send_response(204) + self.end_headers() + + def log_message(self, *args): + pass + + errors = [] + with helper._qualification_directory() as directory: + fixture = Path(directory) / "database-bytes" + fixture.write_bytes(b"original fixture bytes") + listener = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) + listener.daemon_threads = False + worker = threading.Thread(target=listener.serve_forever) + client = stopper = None + try: + worker.start() + client = socket.create_connection(listener.server_address) + client.sendall(b"GET / HTTP/1.0\r\n\r\n") + self.assertTrue(entered.wait(3)) + + def stop(): + try: + helper._stop_server(listener, worker) + except BaseException as error: + errors.append(error) + finally: + stopped.set() + + stopper = threading.Thread(target=stop) + stopper.start() + self.assertFalse(stopped.wait(0.7), "active handler must block server_close") + self.assertEqual(fixture.read_bytes(), b"original fixture bytes") + finally: + release.set() + try: + if stopper is not None and stopper.ident is not None: + stopper.join(5) + if stopper.is_alive(): + raise RuntimeError("POSTMAN_TEST_STOP_UNCONFIRMED") + if worker.is_alive(): + helper._stop_server(listener, worker) + else: + listener.server_close() + finally: + if client is not None: + client.close() + self.assertTrue(stopped.is_set()) + self.assertFalse(worker.is_alive()) + self.assertEqual(errors, []) + self.assertEqual(listener.socket.fileno(), -1) + self.assertFalse(fixture.parent.exists()) + + def test_unknown_stop_retains_original_bytes_and_both_errors(self): + listener = http.server.ThreadingHTTPServer(("127.0.0.1", 0), http.server.BaseHTTPRequestHandler) + listener.daemon_threads = False + worker = threading.Thread(target=listener.serve_forever) + directory = None + try: + worker.start() + with self.assertRaises(BaseExceptionGroup) as observed: + with helper._qualification_directory() as directory: + fixture = Path(directory) / "database-bytes" + fixture.write_bytes(b"original fixture bytes") + try: + raise ValueError("primary request failure") + finally: + with patch.object(listener, "shutdown", side_effect=RuntimeError("stop unconfirmed")): + helper._stop_server(listener, worker) + self.assertEqual([str(error) for error in observed.exception.exceptions], + ["primary request failure", "stop unconfirmed"]) + self.assertTrue(worker.is_alive()) + self.assertEqual(fixture.read_bytes(), b"original fixture bytes") + finally: + if worker.is_alive(): + listener.shutdown() + worker.join(3) + if worker.is_alive(): + raise RuntimeError("POSTMAN_TEST_STOP_UNCONFIRMED") + listener.server_close() + if directory is not None: + import shutil + shutil.rmtree(directory) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/qualification/http_json_postman_contract.py b/tools/qualification/http_json_postman_contract.py index 3399789d..aaec0996 100644 --- a/tools/qualification/http_json_postman_contract.py +++ b/tools/qualification/http_json_postman_contract.py @@ -5,12 +5,14 @@ from engineering_platform.storage import sqlite_connection import argparse +from contextlib import contextmanager import http.server import json from pathlib import Path import re import socket import sqlite3 +import sys import tempfile from threading import Lock, Thread from urllib.error import HTTPError @@ -26,6 +28,34 @@ _STATUS = re.compile(r"pm\.response\.to\.have\.status\((\d+)\)") +@contextmanager +def _qualification_directory(): + """Keep failed evidence; clean only after the guarded server has stopped.""" + temporary = tempfile.TemporaryDirectory(prefix="ep-postman-contract-", delete=False) + try: + yield temporary.name + except BaseException: + print(f"POSTMAN_FIXTURE_RETAINED={temporary.name}", file=sys.stderr) + raise + else: + temporary.cleanup() + + +def _stop_server(http_server, worker): + """Join the listener and every non-daemon handler before fixture cleanup.""" + primary = sys.exception() + try: + http_server.shutdown() + worker.join(timeout=3) + if worker.is_alive(): + raise RuntimeError("POSTMAN_LISTENER_STOP_UNCONFIRMED") + http_server.server_close() + except BaseException as cleanup_error: + if primary is not None: + raise BaseExceptionGroup("Postman request and cleanup failed", [primary, cleanup_error]) from None + raise + + class _NoopService: """Minimal lifecycle double for isolated HTTP surface qualification.""" def start(self) -> None: pass @@ -226,7 +256,7 @@ def main(argv: list[str] | None = None) -> int: items = _collection_items(collection.get("item")) if _operations_from_openapi(expected_openapi) != {_collection_operation(item) for item in items}: raise RuntimeError("OPENAPI_POSTMAN_OPERATION_DRIFT") - with tempfile.TemporaryDirectory(prefix="ep-postman-contract-") as temporary: + with _qualification_directory() as temporary: data_root = Path(temporary) / "data" port = _port() identity = server.initialize(data_root, bind_port=port) @@ -243,6 +273,7 @@ def main(argv: list[str] | None = None) -> int: "ep_discovery_instance": identity.instance_id, "ep_discovery_consumer": "postman-workspace", "ep_discovery_workspace": "postman-workspace-instance"} http_server = http.server.ThreadingHTTPServer(("127.0.0.1", port), server._HealthHandler) + http_server.daemon_threads = False http_server.data_root = data_root # type: ignore[attr-defined] http_server.central_data_transfer_lock = Lock() # type: ignore[attr-defined] http_server.central_data_transfer_active = False # type: ignore[attr-defined] @@ -267,8 +298,7 @@ def main(argv: list[str] | None = None) -> int: raise RuntimeError(f"SERVER_SURFACE_POSTMAN_DRIFT {item.get('name')}: expected {expected}, got {observed}") _queue_action_contract(data_root, base_url) finally: - http_server.shutdown() - worker.join(timeout=3) + _stop_server(http_server, worker) print("HTTP_JSON_API_OPENAPI_POSTMAN_CONTRACT=PASS") return 0