From ecbf94d14a5903b734d780b6f0e9f6f89150c157 Mon Sep 17 00:00:00 2001 From: Andraxion Date: Wed, 29 Jul 2026 15:01:11 -0400 Subject: [PATCH] Prove adapter launcher modules are runnable --- schemas/adapter-launcher.schema.json | 6 +- src/docforge/adapter_launcher.py | 136 +++++++++++++++++++++- tests/test_adapter_launcher.py | 161 ++++++++++++++++++++++++++- 3 files changed, 293 insertions(+), 10 deletions(-) diff --git a/schemas/adapter-launcher.schema.json b/schemas/adapter-launcher.schema.json index 04197a5..4428828 100644 --- a/schemas/adapter-launcher.schema.json +++ b/schemas/adapter-launcher.schema.json @@ -36,8 +36,10 @@ "type": "string", "minLength": 1, "maxLength": 255, - "pattern": "^[A-Za-z_][A-Za-z0-9_]*(\\.[A-Za-z_][A-Za-z0-9_]*){0,31}$", - "not": { "const": "docforge.mcp_server" } + "anyOf": [ + { "pattern": "^[A-Za-z_][A-Za-z0-9_]*$" }, + { "const": "docforge.reference_mcp" } + ] } }, "additionalProperties": false diff --git a/src/docforge/adapter_launcher.py b/src/docforge/adapter_launcher.py index 6cd57cc..f86be68 100644 --- a/src/docforge/adapter_launcher.py +++ b/src/docforge/adapter_launcher.py @@ -2,11 +2,14 @@ from __future__ import annotations +import json import re import stat +import subprocess +import sys from dataclasses import dataclass from pathlib import Path -from typing import Literal +from typing import Literal, cast from .changeset_contract import document_hash from .config_validation import ID_PATTERN @@ -15,8 +18,22 @@ from .models import IncrementalStateProject, ProjectService, RuntimeValidatedPro ADAPTER_LAUNCHER_SCHEMA_VERSION = 1 _PYTHON_MODULE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*){0,31}") +_TOP_LEVEL_PYTHON_MODULE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*") _ADAPTER_IDENTITY = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}@[A-Za-z0-9][A-Za-z0-9_.+-]{0,63}") _SHA256 = re.compile(r"[0-9a-f]{64}") +_TRUSTED_DOTTED_MODULE = "docforge.reference_mcp" +_MODULE_PROBE_TIMEOUT_SECONDS = 5 +_MAX_MODULE_PROBE_BYTES = 8_192 +_MODULE_PROBE = ( + "import importlib.util,json,sys;" + "spec=importlib.util.find_spec(sys.argv[1]);" + "result=None if spec is None else {" + "'has_loader':spec.loader is not None," + "'is_package':spec.submodule_search_locations is not None," + "'origin':spec.origin};" + "print(json.dumps(result,sort_keys=True,separators=(',',':')))" +) +_run_module_probe = subprocess.run __all__ = [ "ADAPTER_LAUNCHER_SCHEMA_VERSION", "AdapterLauncherV1", @@ -226,10 +243,125 @@ def _validate_launcher_fields(launcher: AdapterLauncherV1) -> None: or len(module) > 255 or _PYTHON_MODULE.fullmatch(module) is None or module == "docforge.mcp_server" + or (module != _TRUSTED_DOTTED_MODULE and _TOP_LEVEL_PYTHON_MODULE.fullmatch(module) is None) ): raise DocForgeError( "invalid_adapter_launcher", - "Adapter launcher module must be one constrained project-owned Python module", + ( + "Adapter launcher module must be one installed project-owned top-level " + "Python module or the fixed DocForge reference binding" + ), + ) + _validate_isolated_module(project_root, module) + + +def _validate_isolated_module(project_root: Path, module: str) -> None: + """Prove ``python -I -m`` can resolve one module without importing project code.""" + + executable = Path(sys.executable) + try: + executable_status = executable.stat() + except OSError as error: + raise DocForgeError( + "adapter_launcher_unavailable", + "The isolated Python executable is unavailable", + ) from error + if not stat.S_ISREG(executable_status.st_mode): + raise DocForgeError( + "adapter_launcher_unavailable", + "The isolated Python executable is not a regular file", + ) + try: + completed = _run_module_probe( + [str(executable), "-I", "-c", _MODULE_PROBE, module], + capture_output=True, + text=True, + timeout=_MODULE_PROBE_TIMEOUT_SECONDS, + check=False, + ) + except (OSError, subprocess.TimeoutExpired) as error: + raise DocForgeError( + "adapter_launcher_unavailable", + "The isolated adapter launcher probe could not complete", + module=module, + ) from error + if ( + completed.returncode != 0 + or len(completed.stdout.encode("utf-8")) > _MAX_MODULE_PROBE_BYTES + or len(completed.stderr.encode("utf-8")) > _MAX_MODULE_PROBE_BYTES + ): + raise DocForgeError( + "adapter_launcher_unavailable", + "The isolated adapter launcher module could not be resolved", + module=module, + ) + try: + result: object = json.loads(completed.stdout) + except (UnicodeError, json.JSONDecodeError) as error: + raise DocForgeError( + "adapter_launcher_unavailable", + "The isolated adapter launcher probe returned invalid evidence", + module=module, + ) from error + if result is None: + raise DocForgeError( + "adapter_launcher_unavailable", + "The isolated adapter launcher module is not installed", + module=module, + ) + if not isinstance(result, dict): + raise DocForgeError( + "adapter_launcher_unavailable", + "The isolated adapter launcher probe returned invalid evidence", + module=module, + ) + evidence = cast(dict[object, object], result) + if ( + set(evidence) != {"has_loader", "is_package", "origin"} + or evidence["has_loader"] is not True + or evidence["is_package"] is not False + or not isinstance(evidence["origin"], str) + ): + raise DocForgeError( + "invalid_adapter_launcher", + "The isolated adapter launcher must resolve to one executable module file", + module=module, + ) + origin = Path(evidence["origin"]) + try: + origin_status = origin.lstat() + resolved_origin = origin.resolve(strict=True) + except OSError as error: + raise DocForgeError( + "adapter_launcher_unavailable", + "The isolated adapter launcher module origin is unavailable", + module=module, + ) from error + if ( + not origin.is_absolute() + or stat.S_ISLNK(origin_status.st_mode) + or not stat.S_ISREG(origin_status.st_mode) + or resolved_origin != origin + or origin.suffix != ".py" + ): + raise DocForgeError( + "invalid_adapter_launcher", + "The isolated adapter launcher must be one canonical Python module file", + module=module, + ) + if module == _TRUSTED_DOTTED_MODULE: + expected = Path(__file__).with_name("reference_mcp.py").resolve(strict=True) + if origin != expected: + raise DocForgeError( + "adapter_launcher_mismatch", + "The fixed reference launcher resolved outside this DocForge installation", + module=module, + ) + elif not origin.is_relative_to(project_root): + raise DocForgeError( + "invalid_adapter_launcher", + "The installed adapter launcher module is not owned by the selected project", + module=module, ) diff --git a/tests/test_adapter_launcher.py b/tests/test_adapter_launcher.py index 2afa1e4..e03df33 100644 --- a/tests/test_adapter_launcher.py +++ b/tests/test_adapter_launcher.py @@ -1,16 +1,22 @@ from __future__ import annotations +import asyncio import json import os import shutil +import subprocess +import sys import tempfile import tomllib import unittest +import venv from dataclasses import replace from pathlib import Path from unittest import mock from jsonschema import Draft202012Validator +from mcp import ClientSession, StdioServerParameters +from mcp.client.stdio import stdio_client from referencing import Registry, Resource from docforge.adapter_contract import AdapterNode, AdapterProject, AdapterProjection @@ -23,6 +29,10 @@ from docforge.client_config import ( from docforge.errors import DocForgeError from docforge.models import Node from docforge.project import Project +from docforge.reference_mcp import ( + REFERENCE_MCP_MODULE, + create_reference_project, +) ROOT = Path(__file__).resolve().parents[1] FIXTURES = ROOT / "tests" / "fixtures" @@ -83,6 +93,26 @@ class AdapterLauncherTests(unittest.TestCase): shutil.copytree(FIXTURES / "alpha", root) return root + def copy_reference_fixture(self, destination: Path) -> Path: + root = destination / "reference-python" + shutil.copytree(FIXTURES / "reference-python", root) + config = root / ".docforge" / "reference-adapter.toml" + config.parent.mkdir(parents=True) + config.write_text( + "\n".join( + ( + "schema_version = 1", + 'project_id = "reference-python"', + 'title = "Runnable Python reference"', + 'language = "python"', + 'source_roots = ["src"]', + "", + ) + ), + encoding="utf-8", + ) + return root.resolve() + def projection( self, root: Path, @@ -135,7 +165,7 @@ class AdapterLauncherTests(unittest.TestCase): project, _ = self.project(root) launcher = AdapterLauncherV1.for_project( project, - module="fixture_adapter.mcp_server", + module=REFERENCE_MCP_MODULE, ) Draft202012Validator(LAUNCHER_SCHEMA).validate(launcher.as_dict()) previous = os.environ.get("DOCFORGE_ADAPTER_LAUNCHER_SECRET") @@ -167,7 +197,7 @@ class AdapterLauncherTests(unittest.TestCase): [ "-I", "-m", - "fixture_adapter.mcp_server", + REFERENCE_MCP_MODULE, "--project-root", str(root), "--capability-mode", @@ -209,10 +239,11 @@ class AdapterLauncherTests(unittest.TestCase): project, _ = self.project(root) launcher = AdapterLauncherV1.for_project( project, - module="fixture_adapter.mcp_server", + module=REFERENCE_MCP_MODULE, ) for module in ( "-m", + "fixture_adapter.mcp_server", "fixture_adapter.mcp_server --debug", "fixture_adapter:mcp_server", "fixture_adapter/mcp_server", @@ -254,7 +285,7 @@ class AdapterLauncherTests(unittest.TestCase): second, _ = self.project(second_root) launcher = AdapterLauncherV1.for_project( first, - module="fixture_adapter.mcp_server", + module=REFERENCE_MCP_MODULE, ) with self.assertRaises(DocForgeError) as wrong_project: @@ -276,7 +307,7 @@ class AdapterLauncherTests(unittest.TestCase): ) drifting_launcher = AdapterLauncherV1.for_project( drifting, - module="fixture_adapter.mcp_server", + module=REFERENCE_MCP_MODULE, ) with self.assertRaises(DocForgeError) as source_drift: generate_adapter_client_configuration( @@ -297,7 +328,7 @@ class AdapterLauncherTests(unittest.TestCase): with self.assertRaises(DocForgeError) as generic_launcher: AdapterLauncherV1.for_project( generic, - module="fixture_adapter.mcp_server", + module=REFERENCE_MCP_MODULE, ) self.assertEqual("adapter_launcher_unavailable", generic_launcher.exception.code) @@ -308,6 +339,124 @@ class AdapterLauncherTests(unittest.TestCase): generate_client_configuration(custom, "codex") self.assertEqual("missing_config", generic_api.exception.code) + def test_installed_project_owned_top_level_module_launches_without_probe_execution( + self, + ) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory).resolve() + environment = root / ".launcher-venv" + venv.EnvBuilder(with_pip=False).create(environment) + executable = environment / "bin" / "python" + site_packages = Path( + subprocess.run( + [ + str(executable), + "-I", + "-c", + "import site; print(site.getsitepackages()[0])", + ], + check=True, + capture_output=True, + text=True, + ).stdout.strip() + ) + (site_packages / "fixture-adapter.pth").write_text( + f"{root}\n", + encoding="utf-8", + ) + sentinel = root / "fixture-adapter-executed" + (root / "fixture_adapter.py").write_text( + "\n".join( + ( + "import argparse", + "import json", + "from pathlib import Path", + "parser = argparse.ArgumentParser()", + 'parser.add_argument("--project-root", required=True)', + ( + 'parser.add_argument("--capability-mode", ' + 'choices=("read",), required=True)' + ), + "arguments = parser.parse_args()", + f"Path({str(sentinel)!r}).write_text('executed', encoding='utf-8')", + ( + "print(json.dumps({'project_root': arguments.project_root}, " + "sort_keys=True))" + ), + "", + ) + ), + encoding="utf-8", + ) + + project, _ = self.project(root) + with mock.patch.object(sys, "executable", str(executable)): + launcher = AdapterLauncherV1.for_project( + project, + module="fixture_adapter", + ) + result = generate_adapter_client_configuration( + project, + launcher, + "codex", + ) + self.assertFalse(sentinel.exists()) + Draft202012Validator(LAUNCHER_SCHEMA).validate(launcher.as_dict()) + binding = result["binding"] + completed = subprocess.run( + [binding["command"], *binding["args"]], + check=True, + capture_output=True, + text=True, + timeout=10, + ) + self.assertEqual({"project_root": str(root)}, json.loads(completed.stdout)) + self.assertEqual("executed", sentinel.read_text(encoding="utf-8")) + + def test_uninstalled_packages_and_non_project_modules_fail_closed(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory).resolve() + project, _ = self.project(root) + for module, code in ( + ("fixture_adapter", "adapter_launcher_unavailable"), + ("os", "adapter_launcher_unavailable"), + ("json", "invalid_adapter_launcher"), + ): + with self.subTest(module=module), self.assertRaises(DocForgeError) as captured: + AdapterLauncherV1.for_project(project, module=module) + self.assertEqual(code, captured.exception.code) + + def test_generated_reference_binding_starts_a_real_isolated_mcp_process(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = self.copy_reference_fixture(Path(directory)) + project = create_reference_project(root) + launcher = AdapterLauncherV1.for_project( + project, + module=REFERENCE_MCP_MODULE, + ) + result = generate_adapter_client_configuration( + project, + launcher, + "codex", + ) + binding = result["binding"] + + async def inspect() -> tuple[str, ...]: + parameters = StdioServerParameters( + command=binding["command"], + args=binding["args"], + env=binding["environment"], + ) + async with ( + stdio_client(parameters) as streams, + ClientSession(*streams) as session, + ): + await session.initialize() + return tuple(tool.name for tool in (await session.list_tools()).tools) + + tools = asyncio.run(inspect()) + self.assertIn("docforge_bootstrap", tools) + def project_source_availability(result: dict[str, object]): from docforge.adapter_launcher import AdapterSourceAvailabilityV1