#!/usr/bin/env python3 """Read-only fail-closed ContextForge runtime compatibility verifier.""" from __future__ import annotations import argparse import hashlib import importlib import importlib.metadata import json import os import re import subprocess import sys from collections.abc import Callable, Iterable, Mapping from pathlib import Path from typing import Any from sqlalchemy import create_engine, text from sqlalchemy.engine import URL DEFAULT_CONTRACT = Path("/opt/itworx/contextforge-runtime-contract.json") EXPECTED_SCHEMA = "itworx.contextforge-runtime-contract/v1" ALEMBIC_BIN = "/app/.venv/bin/alembic" ALEMBIC_CONFIG = "mcpgateway/alembic.ini" APP_ROOT = "/app" SHA256 = re.compile(r"^[0-9a-f]{64}$") CONTEXTFORGE_DB_PASSWORD_FILE = Path("/run/secrets/contextforge_database_password") class RuntimeContractError(RuntimeError): """Bounded compatibility failure with no secret-bearing detail.""" def _fail(code: str) -> None: raise RuntimeContractError(code) def contract_sha256(path: Path) -> str: try: return hashlib.sha256(path.read_bytes()).hexdigest() except OSError as exc: raise RuntimeContractError("contract-unreadable") from exc def validate_contract_digest(path: Path, environment: Mapping[str, str]) -> None: expected = environment.get("ITWORX_CONTEXTFORGE_EXPECTED_CONTRACT_SHA256", "") if not SHA256.fullmatch(expected): _fail("contract-digest-config") if contract_sha256(path) != expected: _fail("contract-digest") def load_contract(path: Path) -> dict[str, Any]: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError, UnicodeError) as exc: raise RuntimeContractError("contract-unreadable") from exc if not isinstance(payload, dict): _fail("contract-shape") validate_contract_shape(payload) return payload def validate_contract_shape(contract: Mapping[str, Any]) -> None: if contract.get("schema") != EXPECTED_SCHEMA: _fail("contract-schema") for key in ("sourceCommit", "alembicHead", "mcpSdk", "requiredSymbol"): value = contract.get(key) if not isinstance(value, str) or not value: _fail(f"contract:{key}") if re.fullmatch(r"[0-9a-f]{40}", str(contract["sourceCommit"])) is None: _fail("contract:sourceCommit") production = contract.get("productionEnvironment") if not isinstance(production, dict) or not production: _fail("contract:productionEnvironment") if any( not isinstance(key, str) or not isinstance(value, str) for key, value in production.items() ): _fail("contract:productionEnvironment") policy = contract.get("migrationPolicy") if not isinstance(policy, dict): _fail("migration-policy") if ( policy.get("startupMigrations") != "forbidden" or policy.get("skipEnv") != "MCPGATEWAY_SKIP_MIGRATIONS" or policy.get("skipValue") != "true" or production.get("MCPGATEWAY_SKIP_MIGRATIONS") != "true" ): _fail("migration-policy") def validate_environment(contract: Mapping[str, Any], environment: Mapping[str, str]) -> None: production = contract["productionEnvironment"] assert isinstance(production, dict) for name, expected in production.items(): if environment.get(str(name)) != expected: _fail(f"environment:{name}") def installed_mcp_version() -> str: return importlib.metadata.version("mcp") def load_mcp_symbol(module_name: str, symbol: str) -> object: module = importlib.import_module(module_name) return getattr(module, symbol) def validate_mcp_runtime( contract: Mapping[str, Any], *, version_getter: Callable[[str], str] = importlib.metadata.version, symbol_loader: Callable[[str, str], object] = load_mcp_symbol, ) -> None: try: actual_version = version_getter("mcp") except Exception as exc: raise RuntimeContractError("mcp-version") from exc if actual_version != contract["mcpSdk"]: _fail("mcp-version") try: symbol_loader("mcp", str(contract["requiredSymbol"])) except Exception as exc: raise RuntimeContractError("mcp-symbol") from exc def query_image_alembic_heads() -> list[str]: try: completed = subprocess.run( # noqa: S603 [ALEMBIC_BIN, "-c", ALEMBIC_CONFIG, "heads"], cwd=APP_ROOT, capture_output=True, text=True, timeout=15, check=False, ) except (OSError, subprocess.SubprocessError) as exc: raise RuntimeContractError("alembic-head") from exc if completed.returncode != 0: _fail("alembic-head") heads: list[str] = [] for raw_line in completed.stdout.splitlines(): token = raw_line.strip().split(maxsplit=1)[0] if raw_line.strip() else "" if token: heads.append(token) return sorted(set(heads)) def validate_alembic_head(contract: Mapping[str, Any], heads: Iterable[str]) -> None: actual = sorted(set(str(item) for item in heads if str(item))) if actual != [str(contract["alembicHead"])]: _fail("alembic-head") def resolve_database_url(environment: Mapping[str, str]) -> str: configured = environment.get("DATABASE_URL", "") if configured: return configured try: password = ( CONTEXTFORGE_DB_PASSWORD_FILE.read_text(encoding="utf-8") .replace("\r", "") .replace("\n", "") ) except (OSError, UnicodeError) as exc: raise RuntimeContractError("database-url") from exc if not password: _fail("database-url") return URL.create( "postgresql+psycopg", username="contextforge", password=password, host="postgres", port=5432, database="contextforge", ).render_as_string(hide_password=False) def query_database_revisions(database_url: str) -> list[str]: if not database_url: _fail("database-url") engine = create_engine(database_url, pool_pre_ping=True) try: with engine.connect() as connection: rows = connection.execute( text("SELECT version_num FROM alembic_version") ).scalars().all() except Exception as exc: raise RuntimeContractError("database-revision") from exc finally: engine.dispose() return sorted(set(str(item) for item in rows if str(item))) def validate_database_revision(contract: Mapping[str, Any], revisions: Iterable[str]) -> None: actual = sorted(set(str(item) for item in revisions if str(item))) if actual != [str(contract["alembicHead"])]: _fail("database-revision") def verify_runtime(contract: Mapping[str, Any], environment: Mapping[str, str]) -> None: validate_contract_shape(contract) validate_environment(contract, environment) validate_mcp_runtime( contract, version_getter=lambda _name: installed_mcp_version(), symbol_loader=load_mcp_symbol, ) validate_alembic_head(contract, query_image_alembic_heads()) validate_database_revision( contract, query_database_revisions(resolve_database_url(environment)), ) def main() -> int: parser = argparse.ArgumentParser(description=__doc__) mode = parser.add_mutually_exclusive_group(required=True) mode.add_argument("--startup", action="store_true") mode.add_argument("--health", action="store_true") parser.add_argument( "--contract", type=Path, default=Path(os.environ.get("ITWORX_CONTEXTFORGE_CONTRACT_FILE", DEFAULT_CONTRACT)), ) args = parser.parse_args() if os.environ.get("ITWORX_CONTEXTFORGE_CONTRACT_MODE", "relaxed") != "strict": print("ContextForge runtime contract: inactive", flush=True) return 0 try: validate_contract_digest(args.contract, os.environ) contract = load_contract(args.contract) verify_runtime(contract, os.environ) except RuntimeContractError as exc: print(f"ContextForge runtime contract FAILED [{exc}]", file=sys.stderr, flush=True) return 78 print("ContextForge runtime contract: verified", flush=True) return 0 if __name__ == "__main__": raise SystemExit(main())