ITWorx_MCP_Hub-Public/scripts/contextforge_runtime_contract.py

#!/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())