diff --git a/README.md b/README.md index 311b1a3..9a8fdb9 100644 --- a/README.md +++ b/README.md @@ -47,3 +47,14 @@ retirement, and relocation from CoulombCore to railiance01 are complete. and eventual archive (`CORE-WP-0010`). Its dual-run design is recorded and S0 is active against published hub-core revision `7e1ec03`, building the durable backend and internal candidate before any public traffic changes. + +Operational migration bundles are emitted without raw key material: + +```bash +core-hub-ops migration export --source-revision "$(git rev-parse HEAD)" \ + --output core-hub-export.json +``` + +`CORE_HUB_V2_WRITE_GROUPS` controls legacy write authority for `registry`, +`credentials`, `interaction`, and `deferred`. The deployment package must +remove a group here before enabling the matching hub-core writer. diff --git a/STATE.md b/STATE.md index 454f8aa..48f3bcb 100644 --- a/STATE.md +++ b/STATE.md @@ -5,9 +5,10 @@ ## One-line posture -**Core Hub is the verified production `/api/v2` runtime on railiance01; the -joint hub-core absorption plan is recorded, and internal-only S0 is active to -build durable PostgreSQL storage and the candidate runtime.** +**Core Hub remains the verified production `/api/v2` authority on railiance01. +Hub-core S0–S5 implementation is published at `8ab1d0c` with immutable image +digest `sha256:2a8b396c5295476d5ce865927b8309a6e054e50eac4e92175ef32cdeaba962fb`; +the next gate is the private production candidate and data import.** ## Production truth @@ -32,7 +33,7 @@ Deployment packaging and rollout truth lives in the `rapp-core-hub` repository. | --- | --- | --- | | `CORE-WP-0001`–`CORE-WP-0009` | finished / archived | Gen3 runtime, contracts, persistence, consumer gates, and hardening delivered | | `CORE-WP-0011` | finished | Production relocated to railiance01 and active CoulombCore runtime retired | -| `CORE-WP-0010` | active | Dual-run design done; internal-only S0 foundation started | +| `CORE-WP-0010` | active | S0–S5 code done; private candidate/data gates next | ### Open work @@ -40,7 +41,7 @@ Deployment packaging and rollout truth lives in the `rapp-core-hub` repository. | --- | --- | --- | | `CORE-WP-0010-T01` inventory | done | Inventory recorded in `docs/specs/runtime-absorption-inventory.md` | | `CORE-WP-0010-T02` dual-run design | done | Route-group plan recorded in `docs/specs/runtime-absorption-plan.md` | -| `CORE-WP-0010-T03` absorption slices | progress | S0 durable backend, compatibility shell, immutable image, candidate deployment | +| `CORE-WP-0010-T03` absorption slices | progress | Deploy private candidate, import/compare data, run consumers | | `CORE-WP-0010-T04` production cutover | wait | Absorption evidence and operator approval | | `CORE-WP-0010-T05` archive | wait | Cutover complete and residual ownership recorded | @@ -68,11 +69,12 @@ curl -fsS https://hub.coulomb.social/readyz - `CORE-WP-0011` is finished but has no State Hub UUID; the registrar warning is retained rather than inventing an identifier. - `.custodian-brief.md` is generated and may lag file-backed workplan truth. -- Hub-core T01–T06 are published at `7e1ec03`; S0 must produce and pin the - immutable candidate image before any public route movement. -- `make lint` currently reports seven baseline style findings in unchanged - Python files (two import-order findings and five long lines); `make test` - passes all 24 tests. +- Hub-core runtime/compatibility/migration implementation is published at + `8ab1d0c`; its candidate image digest is recorded above. Public authority + stays here until the private data, conformance, consumer, and rollback gates + pass. +- `make lint` has baseline style findings in unchanged Python files; the full + suite passes all 26 tests after adding migration export and write fencing. - The stopped CoulombCore workloads and read-only source database are retained only as controlled fallback evidence. - Do not archive this repository before `CORE-WP-0010` completes. diff --git a/src/core_hub/api/v2.py b/src/core_hub/api/v2.py index 7b6a4bd..8dc3f90 100644 --- a/src/core_hub/api/v2.py +++ b/src/core_hub/api/v2.py @@ -3,12 +3,13 @@ from typing import Annotated, Any from hub_core.utils.slugs import slugify_or_default -from fastapi import APIRouter, Depends, Query, Request, status +from fastapi import APIRouter, Depends, HTTPException, Query, Request, status from fastapi.responses import PlainTextResponse, RedirectResponse from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from core_hub.crypto import new_api_key +from core_hub.config import Settings, get_settings from core_hub.db import session_scope from core_hub.models import ( ApiConsumer, @@ -32,6 +33,22 @@ TokenDependency = Annotated[str, Depends(require_api_token)] SessionDependency = Annotated[AsyncSession, Depends(session_scope)] +def require_write_group(group: str): + def dependency(settings: Annotated[Settings, Depends(get_settings)]) -> None: + if group not in settings.v2_write_groups: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail={"code": "write_group_disabled", "group": group}, + ) + + return dependency + + +RegistryWrite = Annotated[None, Depends(require_write_group("registry"))] +CredentialWrite = Annotated[None, Depends(require_write_group("credentials"))] +InteractionWrite = Annotated[None, Depends(require_write_group("interaction"))] + + def encode_body(payload: dict[str, Any]) -> str: return json.dumps(payload, sort_keys=True) @@ -147,7 +164,7 @@ async def list_hubs(_: TokenDependency, session: SessionDependency) -> dict[str, @router.post("/hubs", status_code=status.HTTP_201_CREATED) async def create_hub( - body: dict[str, Any], _: TokenDependency, session: SessionDependency + body: dict[str, Any], _: TokenDependency, session: SessionDependency, __: RegistryWrite ) -> dict[str, Any]: row = Hub( slug=body["slug"], @@ -182,7 +199,7 @@ async def list_hub_capability_manifests( @router.post("/hub-capability-manifests", status_code=status.HTTP_201_CREATED) async def create_hub_capability_manifest( - body: dict[str, Any], _: TokenDependency, session: SessionDependency + body: dict[str, Any], _: TokenDependency, session: SessionDependency, __: RegistryWrite ) -> dict[str, Any]: hub_id = body.get("hubId") hub_slug = None @@ -204,7 +221,8 @@ async def create_hub_capability_manifest( @router.patch("/hub-capability-manifests/{manifest_id}") async def update_hub_capability_manifest( - manifest_id: str, body: dict[str, Any], _: TokenDependency, session: SessionDependency + manifest_id: str, body: dict[str, Any], _: TokenDependency, session: SessionDependency, + __: RegistryWrite, ) -> dict[str, Any]: row = await session.get(HubCapabilityManifest, manifest_id) if row is None: @@ -220,7 +238,7 @@ async def update_hub_capability_manifest( @router.post("/hub-capability-manifests/{manifest_id}/activate") async def activate_hub_capability_manifest( - manifest_id: str, _: TokenDependency, session: SessionDependency + manifest_id: str, _: TokenDependency, session: SessionDependency, __: RegistryWrite ) -> dict[str, Any]: row = await session.get(HubCapabilityManifest, manifest_id) if row is None: @@ -239,7 +257,7 @@ async def list_api_consumers(_: TokenDependency, session: SessionDependency) -> @router.post("/api-consumers", status_code=status.HTTP_201_CREATED) async def create_api_consumer( - body: dict[str, Any], _: TokenDependency, session: SessionDependency + body: dict[str, Any], _: TokenDependency, session: SessionDependency, __: CredentialWrite ) -> dict[str, Any]: row = ApiConsumer( slug=body.get("slug") or slugify_or_default(body["name"]), @@ -259,7 +277,8 @@ async def create_api_consumer( @router.post("/api-consumers/{consumer_id}/api-keys", status_code=status.HTTP_201_CREATED) async def create_api_key( - consumer_id: str, body: dict[str, Any], _: TokenDependency, session: SessionDependency + consumer_id: str, body: dict[str, Any], _: TokenDependency, session: SessionDependency, + __: CredentialWrite, ) -> dict[str, Any]: full_key, key_prefix, key_hash = new_api_key() row = ApiKey( @@ -294,7 +313,7 @@ async def list_widgets(_: TokenDependency, session: SessionDependency) -> dict[s @router.post("/widgets", status_code=status.HTTP_201_CREATED) async def create_widget( - body: dict[str, Any], _: TokenDependency, session: SessionDependency + body: dict[str, Any], _: TokenDependency, session: SessionDependency, __: InteractionWrite ) -> dict[str, Any]: row = Widget( hub_id=body["hubId"], @@ -320,7 +339,8 @@ async def list_interaction_events(_: TokenDependency, session: SessionDependency @router.post("/interaction-events", status_code=status.HTTP_201_CREATED) async def create_interaction_event( - body: dict[str, Any], _: TokenDependency, session: SessionDependency + body: dict[str, Any], _: TokenDependency, session: SessionDependency, + __: InteractionWrite, ) -> dict[str, Any]: row = InteractionEvent( widget_id=body["widgetId"], @@ -352,12 +372,21 @@ async def hub_registry(_: TokenDependency, session: SessionDependency) -> dict[s @router.post("/deployment-records", status_code=status.HTTP_201_CREATED) @router.get("/outcome-signals") @router.post("/outcome-signals", status_code=status.HTTP_201_CREATED) -async def deferred_collection(_: TokenDependency) -> dict[str, list]: +async def deferred_collection( + request: Request, + _: TokenDependency, + settings: Annotated[Settings, Depends(get_settings)], +) -> dict[str, list]: + if request.method == "POST" and "deferred" not in settings.v2_write_groups: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail={"code": "write_group_disabled", "group": "deferred"}, + ) return {"data": []} @router.post("/token", tags=["api-v2-protected"], operation_id="post_token") -async def issue_token(_: TokenDependency) -> dict[str, str]: +async def issue_token(_: TokenDependency, __: CredentialWrite) -> dict[str, str]: return {"access_token": "already-authenticated", "token_type": "bearer"} diff --git a/src/core_hub/config.py b/src/core_hub/config.py index 9c6e568..c5fd2eb 100644 --- a/src/core_hub/config.py +++ b/src/core_hub/config.py @@ -4,6 +4,12 @@ from functools import lru_cache from pydantic import BaseModel, Field +def _env_set(name: str, default: str) -> frozenset[str]: + return frozenset( + value.strip() for value in os.getenv(name, default).split(",") if value.strip() + ) + + class Settings(BaseModel): app_name: str = "Core Hub" environment: str = Field(default_factory=lambda: os.getenv("CORE_HUB_ENV", "development")) @@ -17,6 +23,11 @@ class Settings(BaseModel): auto_create_tables: bool = Field( default_factory=lambda: os.getenv("CORE_HUB_AUTO_CREATE_TABLES", "0") == "1" ) + v2_write_groups: frozenset[str] = Field( + default_factory=lambda: _env_set( + "CORE_HUB_V2_WRITE_GROUPS", "registry,credentials,interaction,deferred" + ) + ) @property def auth_configured(self) -> bool: diff --git a/src/core_hub/migration.py b/src/core_hub/migration.py index 970cdcd..d981b9a 100644 --- a/src/core_hub/migration.py +++ b/src/core_hub/migration.py @@ -1,6 +1,7 @@ import hashlib import json from collections.abc import Iterable +from datetime import datetime, timezone from typing import Any from sqlalchemy import select @@ -34,10 +35,179 @@ def encode_body(payload: dict[str, Any]) -> str: def bundle_digest(bundle: dict[str, Any]) -> str: - canonical = json.dumps(bundle, sort_keys=True, separators=(",", ":")) + digestable = {key: value for key, value in bundle.items() if key != "bundleSha256"} + canonical = json.dumps(digestable, sort_keys=True, separators=(",", ":")) return hashlib.sha256(canonical.encode("utf-8")).hexdigest() +def _timestamp(value: datetime) -> str: + if value.tzinfo is None: + value = value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") + + +def _decoded(raw: str | None) -> dict[str, Any]: + value = json.loads(raw or "{}") + return value if isinstance(value, dict) else {} + + +def _with_timestamps( + body: str | None, + *, + created_at: datetime, + updated_at: datetime | None = None, +) -> dict[str, Any]: + value = _decoded(body) + value["createdAt"] = _timestamp(created_at) + if updated_at is not None: + value["updatedAt"] = _timestamp(updated_at) + return value + + +async def export_bundle( + session: AsyncSession, *, source_revision: str | None = None +) -> dict[str, Any]: + """Export all seven durable Core Hub tables without raw credential material.""" + hubs = list((await session.execute(select(Hub).order_by(Hub.id))).scalars()) + manifests = list( + (await session.execute(select(HubCapabilityManifest).order_by(HubCapabilityManifest.id))) + .scalars() + ) + consumers = list( + (await session.execute(select(ApiConsumer).order_by(ApiConsumer.id))).scalars() + ) + keys = list((await session.execute(select(ApiKey).order_by(ApiKey.id))).scalars()) + widgets = list((await session.execute(select(Widget).order_by(Widget.id))).scalars()) + events = list( + (await session.execute(select(InteractionEvent).order_by(InteractionEvent.id))).scalars() + ) + runs = list( + (await session.execute(select(MigrationRun).order_by(MigrationRun.id))).scalars() + ) + + records: dict[str, list[dict[str, Any]]] = { + "hubs": [], + "hubCapabilityManifests": [], + "apiConsumers": [], + "apiKeys": [], + "widgets": [], + "interactionEvents": [], + "migrationRuns": [], + } + for row in hubs: + value = _with_timestamps( + row.body_json, created_at=row.created_at, updated_at=row.updated_at + ) + value.update( + id=row.id, + slug=row.slug, + name=row.name, + domain=row.domain, + hubKind=row.hub_kind, + hubFamily=row.hub_family, + vsmFunction=row.vsm_function, + vsmSystem=row.vsm_system, + status=row.status, + description=row.description, + ) + records["hubs"].append({key: child for key, child in value.items() if child is not None}) + for row in manifests: + value = _with_timestamps( + row.body_json, created_at=row.created_at, updated_at=row.updated_at + ) + value.update( + id=row.id, + hubId=row.hub_id, + hubSlug=row.hub_slug, + manifestVersion=row.manifest_version, + status=row.status, + ) + records["hubCapabilityManifests"].append( + {key: child for key, child in value.items() if child is not None} + ) + for row in consumers: + value = _with_timestamps(row.body_json, created_at=row.created_at) + value.update( + id=row.id, + slug=row.slug, + name=row.name, + description=row.description, + hubCapabilityManifestId=row.hub_capability_manifest_id, + rateLimitPerMinute=row.rate_limit_per_minute, + quotaPerDay=row.quota_per_day, + keyPrefix=row.key_prefix, + status=row.status, + ) + records["apiConsumers"].append( + {key: child for key, child in value.items() if child is not None} + ) + records["apiKeys"] = [ + { + "id": row.id, + "apiConsumerId": row.api_consumer_id, + "keyPrefix": row.key_prefix, + "keyHash": row.key_hash, + "scopes": row.scopes, + "status": row.status, + "createdAt": _timestamp(row.created_at), + } + for row in keys + ] + for row in widgets: + value = _with_timestamps(row.body_json, created_at=row.created_at) + value.update( + id=row.id, + hubId=row.hub_id, + name=row.name, + widgetType=row.widget_type, + capabilityRef=row.capability_ref, + viewContext=row.view_context, + policyScope=row.policy_scope, + status=row.status, + ) + records["widgets"].append( + {key: child for key, child in value.items() if child is not None} + ) + for row in events: + value = _with_timestamps(row.body_json, created_at=row.created_at) + value.update( + id=row.id, + widgetId=row.widget_id, + eventType=row.event_type, + viewContext=row.view_context, + metadata=_decoded(row.metadata_json), + ) + records["interactionEvents"].append( + {key: child for key, child in value.items() if child is not None} + ) + records["migrationRuns"] = [ + { + "id": row.id, + "source": row.source, + "schemaVersion": row.schema_version, + "bundleSha256": row.bundle_sha256, + "dryRun": row.dry_run, + "status": row.status, + "counts": _decoded(row.counts_json), + "diagnostics": _decoded(row.diagnostics_json), + "createdAt": _timestamp(row.created_at), + } + for row in runs + ] + high_water = max((_timestamp(row.created_at) for row in events), default=None) + bundle: dict[str, Any] = { + "schemaVersion": MIGRATION_SCHEMA_VERSION, + "source": "core-hub", + "exportedAt": _timestamp(datetime.now(timezone.utc)), + "highWaterMark": high_water, + "records": records, + } + if source_revision: + bundle["sourceRevision"] = source_revision + bundle["bundleSha256"] = bundle_digest(bundle) + return bundle + + def records_for(bundle: dict[str, Any], collection: str) -> list[dict[str, Any]]: records = bundle.get("records", {}) if not isinstance(records, dict): diff --git a/src/core_hub/operator_cli.py b/src/core_hub/operator_cli.py index 455aff7..5792c8d 100644 --- a/src/core_hub/operator_cli.py +++ b/src/core_hub/operator_cli.py @@ -10,7 +10,7 @@ from typing import Any import httpx from core_hub.db import get_sessionmaker -from core_hub.migration import import_bundle, validate_bundle +from core_hub.migration import export_bundle, import_bundle, validate_bundle from core_hub.smoke import ( DEFAULT_OPERATOR_TOKEN_ENV, DEFAULT_RUNTIME_TOKEN_ENV, @@ -292,6 +292,12 @@ async def run_import(path: Path, dry_run: bool) -> dict[str, Any]: return await import_bundle(session, bundle, dry_run=dry_run) +async def run_export(source_revision: str | None) -> dict[str, Any]: + sessionmaker = get_sessionmaker() + async with sessionmaker() as session: + return await export_bundle(session, source_revision=source_revision) + + def add_token_args(parser: argparse.ArgumentParser) -> None: parser.add_argument("--base-url", default=os.environ.get("CORE_HUB_BASE_URL")) parser.add_argument( @@ -338,6 +344,8 @@ def build_parser() -> argparse.ArgumentParser: import_parser = migration_subparsers.add_parser("import") import_parser.add_argument("bundle", type=Path) import_parser.add_argument("--dry-run", action="store_true") + export_parser = migration_subparsers.add_parser("export") + export_parser.add_argument("--source-revision") readiness_parser = subparsers.add_parser("readiness-summary", help="Summarize cutover gates.") readiness_parser.add_argument("--deployed-smoke-report", type=Path) @@ -398,8 +406,10 @@ def main(argv: list[str] | None = None) -> int: if args.command == "migration": if args.migration_command == "validate": report = validate_bundle(load_bundle(args.bundle)) - else: + elif args.migration_command == "import": report = asyncio.run(run_import(args.bundle, dry_run=args.dry_run)) + else: + report = asyncio.run(run_export(args.source_revision)) print_report(report, args.output) return 0 if report.get("ok") else 1 diff --git a/tests/test_api_v2.py b/tests/test_api_v2.py index b33185e..755ad3d 100644 --- a/tests/test_api_v2.py +++ b/tests/test_api_v2.py @@ -1,3 +1,6 @@ +from core_hub.config import get_settings + + CATALOG_ENDPOINTS = [ "/api/v2/widget-types", "/api/v2/event-types", @@ -37,6 +40,20 @@ def test_protected_api_v2_endpoints_fail_before_business_logic(client): assert response.json()["detail"]["code"] == "unauthorized" +def test_disabled_write_group_fails_closed(client, monkeypatch): + monkeypatch.setenv("CORE_HUB_V2_WRITE_GROUPS", "credentials") + get_settings.cache_clear() + response = client.post( + "/api/v2/hubs", + headers=OPERATOR_HEADERS, + json={"slug": "blocked", "name": "Blocked"}, + ) + get_settings.cache_clear() + + assert response.status_code == 503 + assert response.json()["detail"]["group"] == "registry" + + def test_openapi_contains_ops_hub_gate_paths(client): payload = client.get("/api/v2/openapi.json").json() diff --git a/tests/test_migration.py b/tests/test_migration.py index 9e30a43..4cdfda4 100644 --- a/tests/test_migration.py +++ b/tests/test_migration.py @@ -3,7 +3,7 @@ import json from sqlalchemy import select from sqlalchemy.ext.asyncio import async_sessionmaker -from core_hub.migration import import_bundle, validate_bundle +from core_hub.migration import export_bundle, import_bundle, validate_bundle from core_hub.models import Hub, MigrationRun, Widget @@ -114,3 +114,24 @@ async def test_import_bundle_persists_records_and_run_summary(sqlite_engine): second = await import_bundle(session, minimal_bundle()) assert second["counts"]["hubs"]["updated"] == 1 assert second["counts"]["interactionEvents"]["updated"] == 1 + + +async def test_export_bundle_covers_all_source_tables_without_raw_keys(sqlite_engine): + sessionmaker = async_sessionmaker(sqlite_engine, expire_on_commit=False) + async with sessionmaker() as session: + await import_bundle(session, minimal_bundle()) + bundle = await export_bundle(session, source_revision="source-revision") + + assert set(bundle["records"]) == { + "hubs", + "hubCapabilityManifests", + "apiConsumers", + "apiKeys", + "widgets", + "interactionEvents", + "migrationRuns", + } + assert bundle["sourceRevision"] == "source-revision" + assert bundle["bundleSha256"] + assert bundle["records"]["apiKeys"][0]["keyHash"] == "a" * 64 + assert "fullKey" not in json.dumps(bundle) diff --git a/uv.lock b/uv.lock index bad86fd..7f91c56 100644 --- a/uv.lock +++ b/uv.lock @@ -659,17 +659,29 @@ dependencies = [ { name = "fastapi" }, { name = "fastmcp" }, { name = "httpx" }, + { name = "jsonschema" }, { name = "pydantic" }, { name = "sqlalchemy", extra = ["asyncio"] }, ] [package.metadata] requires-dist = [ + { name = "alembic", marker = "extra == 'runtime'", specifier = ">=1.13.0" }, + { name = "asyncpg", marker = "extra == 'runtime'", specifier = ">=0.29.0" }, { name = "fastapi", specifier = ">=0.115.0" }, { name = "fastmcp", specifier = ">=2.0.0" }, { name = "httpx", specifier = ">=0.28.0" }, + { name = "jsonschema", specifier = ">=4.23.0" }, + { name = "psycopg2-binary", marker = "extra == 'runtime'", specifier = ">=2.9.0" }, { name = "pydantic", specifier = ">=2.10.0" }, { name = "sqlalchemy", extras = ["asyncio"], specifier = ">=2.0.0" }, + { name = "uvicorn", extras = ["standard"], marker = "extra == 'runtime'", specifier = ">=0.30.0" }, +] + +[package.metadata.requires-dev] +dev = [ + { name = "aiosqlite", specifier = ">=0.20.0" }, + { name = "pytest", specifier = ">=8.0.0" }, ] [[package]] diff --git a/workplans/CORE-WP-0010-runtime-absorption-and-archive.md b/workplans/CORE-WP-0010-runtime-absorption-and-archive.md index d502789..d4195ff 100644 --- a/workplans/CORE-WP-0010-runtime-absorption-and-archive.md +++ b/workplans/CORE-WP-0010-runtime-absorption-and-archive.md @@ -115,6 +115,15 @@ and publication gate. S0 now owns the durable PostgreSQL backend, legacy auth/health compatibility, immutable image, and internal candidate deployment. Do not change public traffic. +Implementation advancement 2026-08-21: hub-core revision `8ab1d0c` adds the +durable PostgreSQL store, audit ledger, all `/api/v2` compatibility groups, +fail-closed writer controls, seven-table idempotent import/export, and reverse +bundle support. Its Forgejo image is pinned at +`sha256:2a8b396c5295476d5ce865927b8309a6e054e50eac4e92175ef32cdeaba962fb`. +This source now exports all seven tables without raw keys and fences writes by +route group. The remaining T03 work is live candidate migration, comparison, +and named consumer evidence. + ## Cutover production traffic ```task