From db2c299fcbfab17dc9c7e37ef4ae55fd93bb438e Mon Sep 17 00:00:00 2001 From: Coden Date: Mon, 22 Jun 2026 02:08:23 +0900 Subject: [PATCH] Centralize manifest loading to prevent release drift Keep the command manifest as a literal-only data file while sharing the bounded no-follow AST loader across runtime, release gates, smoke checks, and tests. Deduplicate release-smoke dispatcher launch orchestration without changing scenarios. Constraint: context_guard_commands.py must remain literal-only and must not be imported or executed as code. Rejected: import context_guard_commands directly | would execute mutable manifest code and weaken the runtime trust boundary. Confidence: high Scope-risk: moderate Directive: Add future command/package fields through context_guard_commands.py and the shared literal loader, then sync plugin copies. Tested: python3 -m py_compile changed scripts/helpers; 7 targeted manifest/runtime tests; PYTHONDONTWRITEBYTECODE=1 python3 -m unittest discover -s tests -p 'test_*.py' (737 tests); python3 scripts/prepublish_check.py --skip-tests; python3 scripts/sync_plugin_copies.py --check; git diff --check; python3 scripts/release_smoke.py --timeout 20. Not-tested: CI matrix before PR creation. --- context-guard-kit/context_guard_cli.py | 88 ++++---- .../context_guard_command_manifest_loader.py | 123 +++++++++++ context-guard-kit/context_guard_commands.py | 5 +- plugins/context-guard/bin/context-guard | 88 ++++---- .../context_guard_command_manifest_loader.py | 123 +++++++++++ .../lib/context_guard_commands.py | 5 +- scripts/prepublish_check.py | 94 ++++---- scripts/release_smoke.py | 205 ++++++++++-------- tests/test_context_guard_kit.py | 72 +++--- 9 files changed, 540 insertions(+), 263 deletions(-) create mode 100644 context-guard-kit/context_guard_command_manifest_loader.py create mode 100644 plugins/context-guard/lib/context_guard_command_manifest_loader.py diff --git a/context-guard-kit/context_guard_cli.py b/context-guard-kit/context_guard_cli.py index 9cfc691..4665252 100755 --- a/context-guard-kit/context_guard_cli.py +++ b/context-guard-kit/context_guard_cli.py @@ -9,7 +9,6 @@ import json import os -import ast from pathlib import Path import subprocess import stat @@ -19,25 +18,13 @@ COMMAND_NAME = "context-guard" PACKAGE_NAME = "@ictechgy/context-guard" MAX_VERSION_METADATA_BYTES = 64 * 1024 -MAX_COMMAND_MANIFEST_BYTES = 128 * 1024 +MAX_MANIFEST_HELPER_BYTES = 128 * 1024 ALLOWED_FIRST_ABSOLUTE_SYMLINKS = { "tmp": Path("/private/tmp"), "var": Path("/private/var"), } MANIFEST_LOAD_ERROR: str | None = None -COMMAND_MANIFEST_LITERAL_NAMES = { - "IMPLEMENTATION_PAIRS", - "HELPER_PAIRS", - "NPM_BINS", - "NPM_BIN_PATHS", - "DISPATCHER_SUBCOMMANDS", - "LEGACY_WRAPPERS", - "ENTRYPOINT_SMOKE_CASES", - "PLUGIN_ENTRYPOINTS", - "DISPATCHER_SMOKE_CASES", - "EXPECTED_COMMAND_PACK_FILES", -} def _manifest_candidates(script_dir: Path) -> tuple[Path, ...]: @@ -52,7 +39,15 @@ def _manifest_candidates(script_dir: Path) -> tuple[Path, ...]: return () -def _manifest_open_flags() -> int | None: +def _manifest_helper_candidates(script_dir: Path) -> tuple[Path, ...]: + if script_dir.name == "context-guard-kit": + return (script_dir / "context_guard_command_manifest_loader.py",) + if script_dir.name == "bin": + return (script_dir.parent / "lib" / "context_guard_command_manifest_loader.py",) + return () + + +def _trusted_source_open_flags() -> int | None: if not hasattr(os, "O_NOFOLLOW"): return None flags = os.O_RDONLY | os.O_NOFOLLOW @@ -65,25 +60,25 @@ def _manifest_open_flags() -> int | None: return flags -def _read_manifest_source(path: Path) -> str | None: - flags = _manifest_open_flags() +def _read_trusted_helper_source(path: Path) -> str | None: + flags = _trusted_source_open_flags() if flags is None: return None fd = -1 try: fd = os.open(path, flags) st = os.fstat(fd) - if not stat.S_ISREG(st.st_mode) or st.st_size > MAX_COMMAND_MANIFEST_BYTES: + if not stat.S_ISREG(st.st_mode) or st.st_size > MAX_MANIFEST_HELPER_BYTES: return None chunks: list[bytes] = [] total = 0 while True: - chunk = os.read(fd, min(64 * 1024, MAX_COMMAND_MANIFEST_BYTES + 1 - total)) + chunk = os.read(fd, min(64 * 1024, MAX_MANIFEST_HELPER_BYTES + 1 - total)) if not chunk: break chunks.append(chunk) total += len(chunk) - if total > MAX_COMMAND_MANIFEST_BYTES: + if total > MAX_MANIFEST_HELPER_BYTES: return None return b"".join(chunks).decode("utf-8") except (OSError, UnicodeDecodeError): @@ -96,39 +91,40 @@ def _read_manifest_source(path: Path) -> str | None: pass -def _literal_manifest_assignments(source: str) -> dict[str, Any] | None: - try: - tree = ast.parse(source) - except SyntaxError: - return None - values: dict[str, Any] = {} - for node in tree.body: - if isinstance(node, ast.Expr) and isinstance(node.value, ast.Constant) and isinstance(node.value.value, str): +def _load_manifest_helper() -> dict[str, Any] | None: + script_dir = Path(__file__).resolve().parent + for candidate in _manifest_helper_candidates(script_dir): + source = _read_trusted_helper_source(candidate) + if source is None: continue - target: str | None = None - value: ast.expr | None = None - if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): - target = node.target.id - value = node.value - elif isinstance(node, ast.Assign) and len(node.targets) == 1 and isinstance(node.targets[0], ast.Name): - target = node.targets[0].id - value = node.value - if target is None: - return None - if target not in COMMAND_MANIFEST_LITERAL_NAMES or value is None: - return None + namespace: dict[str, Any] = { + "__builtins__": __builtins__, + "__file__": str(candidate), + "__name__": "_context_guard_command_manifest_loader", + } try: - values[target] = ast.literal_eval(value) - except (SyntaxError, ValueError): - return None - return values + exec(compile(source, str(candidate), "exec"), namespace) + except Exception: + continue + if callable(namespace.get("read_manifest_source")) and callable(namespace.get("literal_command_manifest_from_source")): + return namespace + return None + + +COMMAND_MANIFEST_LOADER = _load_manifest_helper() def _load_manifest_from_path(path: Path) -> dict[str, Any] | None: - source = _read_manifest_source(path) + if COMMAND_MANIFEST_LOADER is None: + return None + source = COMMAND_MANIFEST_LOADER["read_manifest_source"](path) if source is None: return None - return _literal_manifest_assignments(source) + try: + values = COMMAND_MANIFEST_LOADER["literal_command_manifest_from_source"](source) + except ValueError: + return None + return values def _coerce_helper_subcommands(value: Any) -> dict[str, tuple[str, ...]] | None: diff --git a/context-guard-kit/context_guard_command_manifest_loader.py b/context-guard-kit/context_guard_command_manifest_loader.py new file mode 100644 index 0000000..ea337fc --- /dev/null +++ b/context-guard-kit/context_guard_command_manifest_loader.py @@ -0,0 +1,123 @@ +"""Trusted literal loader for the ContextGuard command manifest. + +The command manifest is intentionally a literal-only Python data file so release +gates and runtime dispatchers can inspect it without executing manifest code. +This helper centralizes the bounded no-follow read and AST-literal parsing logic +used by the runtime dispatcher, release gates, and tests. +""" +from __future__ import annotations + +import ast +import os +from pathlib import Path +import stat +from typing import Any, Iterable, Mapping + +MAX_COMMAND_MANIFEST_BYTES = 128 * 1024 + +COMMAND_MANIFEST_LITERAL_NAMES = frozenset( + { + "IMPLEMENTATION_PAIRS", + "HELPER_PAIRS", + "NPM_BINS", + "NPM_BIN_PATHS", + "DISPATCHER_SUBCOMMANDS", + "LEGACY_WRAPPERS", + "ENTRYPOINT_SMOKE_CASES", + "PLUGIN_ENTRYPOINTS", + "DISPATCHER_SMOKE_CASES", + "EXPECTED_COMMAND_PACK_FILES", + } +) + + +def manifest_open_flags() -> int | None: + if not hasattr(os, "O_NOFOLLOW"): + return None + flags = os.O_RDONLY | os.O_NOFOLLOW + if hasattr(os, "O_CLOEXEC"): + flags |= os.O_CLOEXEC + if hasattr(os, "O_NONBLOCK"): + flags |= os.O_NONBLOCK + if hasattr(os, "O_NOCTTY"): + flags |= os.O_NOCTTY + return flags + + +def read_manifest_source(path: Path, *, max_bytes: int = MAX_COMMAND_MANIFEST_BYTES) -> str | None: + flags = manifest_open_flags() + if flags is None: + return None + fd = -1 + try: + fd = os.open(path, flags) + st = os.fstat(fd) + if not stat.S_ISREG(st.st_mode) or st.st_size > max_bytes: + return None + chunks: list[bytes] = [] + total = 0 + while True: + chunk = os.read(fd, min(64 * 1024, max_bytes + 1 - total)) + if not chunk: + break + chunks.append(chunk) + total += len(chunk) + if total > max_bytes: + return None + return b"".join(chunks).decode("utf-8") + except (OSError, UnicodeDecodeError): + return None + finally: + if fd >= 0: + try: + os.close(fd) + except OSError: + pass + + +def literal_command_manifest_from_source( + source: str, + *, + allowed_names: Iterable[str] = COMMAND_MANIFEST_LITERAL_NAMES, +) -> dict[str, Any]: + try: + tree = ast.parse(source) + except SyntaxError as exc: + raise ValueError(f"invalid Python manifest syntax: line {exc.lineno}: {exc.msg}") from exc + allowed = set(allowed_names) + values: dict[str, Any] = {} + for node in tree.body: + if isinstance(node, ast.Expr) and isinstance(node.value, ast.Constant) and isinstance(node.value.value, str): + continue + target: str | None = None + value: ast.expr | None = None + if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): + target = node.target.id + value = node.value + elif isinstance(node, ast.Assign) and len(node.targets) == 1 and isinstance(node.targets[0], ast.Name): + target = node.targets[0].id + value = node.value + if target is None: + raise ValueError(f"unsupported executable manifest statement: {type(node).__name__}") + if target not in allowed or value is None: + raise ValueError(f"unsupported manifest assignment: {target}") + try: + values[target] = ast.literal_eval(value) + except (SyntaxError, ValueError) as exc: + raise ValueError(f"manifest assignment must be a literal: {target}") from exc + return values + + +def command_manifest_namespace(values: Mapping[str, Any], *, required: Iterable[str] = ()) -> type: + missing = sorted(set(required) - set(values)) + if missing: + raise ValueError(f"trusted command manifest missing required literals: {', '.join(missing)}") + return type("CommandManifest", (), dict(values)) + + +def load_command_manifest(path: Path, *, required: Iterable[str] = ()) -> type: + source = read_manifest_source(path) + if source is None: + raise ValueError(f"could not load trusted command manifest source: {path}") + values = literal_command_manifest_from_source(source) + return command_manifest_namespace(values, required=required) diff --git a/context-guard-kit/context_guard_commands.py b/context-guard-kit/context_guard_commands.py index 913676a..dafaa44 100644 --- a/context-guard-kit/context_guard_commands.py +++ b/context-guard-kit/context_guard_commands.py @@ -28,7 +28,9 @@ ('trim_command_output.py', 'context-guard-trim-output')) HELPER_PAIRS = (('hook_secret_patterns.py', 'lib/hook_secret_patterns.py'), - ('context_guard_commands.py', 'lib/context_guard_commands.py')) + ('context_guard_commands.py', 'lib/context_guard_commands.py'), + ('context_guard_command_manifest_loader.py', + 'lib/context_guard_command_manifest_loader.py')) NPM_BINS = ('context-guard', 'context-guard-cost', @@ -226,5 +228,6 @@ 'plugins/context-guard/bin/context-guard-statusline-merged', 'plugins/context-guard/bin/context-guard-tool-prune', 'plugins/context-guard/bin/context-guard-trim-output', + 'plugins/context-guard/lib/context_guard_command_manifest_loader.py', 'plugins/context-guard/lib/context_guard_commands.py', 'plugins/context-guard/lib/hook_secret_patterns.py') diff --git a/plugins/context-guard/bin/context-guard b/plugins/context-guard/bin/context-guard index 9cfc691..4665252 100755 --- a/plugins/context-guard/bin/context-guard +++ b/plugins/context-guard/bin/context-guard @@ -9,7 +9,6 @@ from __future__ import annotations import json import os -import ast from pathlib import Path import subprocess import stat @@ -19,25 +18,13 @@ from typing import Any, NoReturn COMMAND_NAME = "context-guard" PACKAGE_NAME = "@ictechgy/context-guard" MAX_VERSION_METADATA_BYTES = 64 * 1024 -MAX_COMMAND_MANIFEST_BYTES = 128 * 1024 +MAX_MANIFEST_HELPER_BYTES = 128 * 1024 ALLOWED_FIRST_ABSOLUTE_SYMLINKS = { "tmp": Path("/private/tmp"), "var": Path("/private/var"), } MANIFEST_LOAD_ERROR: str | None = None -COMMAND_MANIFEST_LITERAL_NAMES = { - "IMPLEMENTATION_PAIRS", - "HELPER_PAIRS", - "NPM_BINS", - "NPM_BIN_PATHS", - "DISPATCHER_SUBCOMMANDS", - "LEGACY_WRAPPERS", - "ENTRYPOINT_SMOKE_CASES", - "PLUGIN_ENTRYPOINTS", - "DISPATCHER_SMOKE_CASES", - "EXPECTED_COMMAND_PACK_FILES", -} def _manifest_candidates(script_dir: Path) -> tuple[Path, ...]: @@ -52,7 +39,15 @@ def _manifest_candidates(script_dir: Path) -> tuple[Path, ...]: return () -def _manifest_open_flags() -> int | None: +def _manifest_helper_candidates(script_dir: Path) -> tuple[Path, ...]: + if script_dir.name == "context-guard-kit": + return (script_dir / "context_guard_command_manifest_loader.py",) + if script_dir.name == "bin": + return (script_dir.parent / "lib" / "context_guard_command_manifest_loader.py",) + return () + + +def _trusted_source_open_flags() -> int | None: if not hasattr(os, "O_NOFOLLOW"): return None flags = os.O_RDONLY | os.O_NOFOLLOW @@ -65,25 +60,25 @@ def _manifest_open_flags() -> int | None: return flags -def _read_manifest_source(path: Path) -> str | None: - flags = _manifest_open_flags() +def _read_trusted_helper_source(path: Path) -> str | None: + flags = _trusted_source_open_flags() if flags is None: return None fd = -1 try: fd = os.open(path, flags) st = os.fstat(fd) - if not stat.S_ISREG(st.st_mode) or st.st_size > MAX_COMMAND_MANIFEST_BYTES: + if not stat.S_ISREG(st.st_mode) or st.st_size > MAX_MANIFEST_HELPER_BYTES: return None chunks: list[bytes] = [] total = 0 while True: - chunk = os.read(fd, min(64 * 1024, MAX_COMMAND_MANIFEST_BYTES + 1 - total)) + chunk = os.read(fd, min(64 * 1024, MAX_MANIFEST_HELPER_BYTES + 1 - total)) if not chunk: break chunks.append(chunk) total += len(chunk) - if total > MAX_COMMAND_MANIFEST_BYTES: + if total > MAX_MANIFEST_HELPER_BYTES: return None return b"".join(chunks).decode("utf-8") except (OSError, UnicodeDecodeError): @@ -96,39 +91,40 @@ def _read_manifest_source(path: Path) -> str | None: pass -def _literal_manifest_assignments(source: str) -> dict[str, Any] | None: - try: - tree = ast.parse(source) - except SyntaxError: - return None - values: dict[str, Any] = {} - for node in tree.body: - if isinstance(node, ast.Expr) and isinstance(node.value, ast.Constant) and isinstance(node.value.value, str): +def _load_manifest_helper() -> dict[str, Any] | None: + script_dir = Path(__file__).resolve().parent + for candidate in _manifest_helper_candidates(script_dir): + source = _read_trusted_helper_source(candidate) + if source is None: continue - target: str | None = None - value: ast.expr | None = None - if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): - target = node.target.id - value = node.value - elif isinstance(node, ast.Assign) and len(node.targets) == 1 and isinstance(node.targets[0], ast.Name): - target = node.targets[0].id - value = node.value - if target is None: - return None - if target not in COMMAND_MANIFEST_LITERAL_NAMES or value is None: - return None + namespace: dict[str, Any] = { + "__builtins__": __builtins__, + "__file__": str(candidate), + "__name__": "_context_guard_command_manifest_loader", + } try: - values[target] = ast.literal_eval(value) - except (SyntaxError, ValueError): - return None - return values + exec(compile(source, str(candidate), "exec"), namespace) + except Exception: + continue + if callable(namespace.get("read_manifest_source")) and callable(namespace.get("literal_command_manifest_from_source")): + return namespace + return None + + +COMMAND_MANIFEST_LOADER = _load_manifest_helper() def _load_manifest_from_path(path: Path) -> dict[str, Any] | None: - source = _read_manifest_source(path) + if COMMAND_MANIFEST_LOADER is None: + return None + source = COMMAND_MANIFEST_LOADER["read_manifest_source"](path) if source is None: return None - return _literal_manifest_assignments(source) + try: + values = COMMAND_MANIFEST_LOADER["literal_command_manifest_from_source"](source) + except ValueError: + return None + return values def _coerce_helper_subcommands(value: Any) -> dict[str, tuple[str, ...]] | None: diff --git a/plugins/context-guard/lib/context_guard_command_manifest_loader.py b/plugins/context-guard/lib/context_guard_command_manifest_loader.py new file mode 100644 index 0000000..ea337fc --- /dev/null +++ b/plugins/context-guard/lib/context_guard_command_manifest_loader.py @@ -0,0 +1,123 @@ +"""Trusted literal loader for the ContextGuard command manifest. + +The command manifest is intentionally a literal-only Python data file so release +gates and runtime dispatchers can inspect it without executing manifest code. +This helper centralizes the bounded no-follow read and AST-literal parsing logic +used by the runtime dispatcher, release gates, and tests. +""" +from __future__ import annotations + +import ast +import os +from pathlib import Path +import stat +from typing import Any, Iterable, Mapping + +MAX_COMMAND_MANIFEST_BYTES = 128 * 1024 + +COMMAND_MANIFEST_LITERAL_NAMES = frozenset( + { + "IMPLEMENTATION_PAIRS", + "HELPER_PAIRS", + "NPM_BINS", + "NPM_BIN_PATHS", + "DISPATCHER_SUBCOMMANDS", + "LEGACY_WRAPPERS", + "ENTRYPOINT_SMOKE_CASES", + "PLUGIN_ENTRYPOINTS", + "DISPATCHER_SMOKE_CASES", + "EXPECTED_COMMAND_PACK_FILES", + } +) + + +def manifest_open_flags() -> int | None: + if not hasattr(os, "O_NOFOLLOW"): + return None + flags = os.O_RDONLY | os.O_NOFOLLOW + if hasattr(os, "O_CLOEXEC"): + flags |= os.O_CLOEXEC + if hasattr(os, "O_NONBLOCK"): + flags |= os.O_NONBLOCK + if hasattr(os, "O_NOCTTY"): + flags |= os.O_NOCTTY + return flags + + +def read_manifest_source(path: Path, *, max_bytes: int = MAX_COMMAND_MANIFEST_BYTES) -> str | None: + flags = manifest_open_flags() + if flags is None: + return None + fd = -1 + try: + fd = os.open(path, flags) + st = os.fstat(fd) + if not stat.S_ISREG(st.st_mode) or st.st_size > max_bytes: + return None + chunks: list[bytes] = [] + total = 0 + while True: + chunk = os.read(fd, min(64 * 1024, max_bytes + 1 - total)) + if not chunk: + break + chunks.append(chunk) + total += len(chunk) + if total > max_bytes: + return None + return b"".join(chunks).decode("utf-8") + except (OSError, UnicodeDecodeError): + return None + finally: + if fd >= 0: + try: + os.close(fd) + except OSError: + pass + + +def literal_command_manifest_from_source( + source: str, + *, + allowed_names: Iterable[str] = COMMAND_MANIFEST_LITERAL_NAMES, +) -> dict[str, Any]: + try: + tree = ast.parse(source) + except SyntaxError as exc: + raise ValueError(f"invalid Python manifest syntax: line {exc.lineno}: {exc.msg}") from exc + allowed = set(allowed_names) + values: dict[str, Any] = {} + for node in tree.body: + if isinstance(node, ast.Expr) and isinstance(node.value, ast.Constant) and isinstance(node.value.value, str): + continue + target: str | None = None + value: ast.expr | None = None + if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): + target = node.target.id + value = node.value + elif isinstance(node, ast.Assign) and len(node.targets) == 1 and isinstance(node.targets[0], ast.Name): + target = node.targets[0].id + value = node.value + if target is None: + raise ValueError(f"unsupported executable manifest statement: {type(node).__name__}") + if target not in allowed or value is None: + raise ValueError(f"unsupported manifest assignment: {target}") + try: + values[target] = ast.literal_eval(value) + except (SyntaxError, ValueError) as exc: + raise ValueError(f"manifest assignment must be a literal: {target}") from exc + return values + + +def command_manifest_namespace(values: Mapping[str, Any], *, required: Iterable[str] = ()) -> type: + missing = sorted(set(required) - set(values)) + if missing: + raise ValueError(f"trusted command manifest missing required literals: {', '.join(missing)}") + return type("CommandManifest", (), dict(values)) + + +def load_command_manifest(path: Path, *, required: Iterable[str] = ()) -> type: + source = read_manifest_source(path) + if source is None: + raise ValueError(f"could not load trusted command manifest source: {path}") + values = literal_command_manifest_from_source(source) + return command_manifest_namespace(values, required=required) diff --git a/plugins/context-guard/lib/context_guard_commands.py b/plugins/context-guard/lib/context_guard_commands.py index 913676a..dafaa44 100644 --- a/plugins/context-guard/lib/context_guard_commands.py +++ b/plugins/context-guard/lib/context_guard_commands.py @@ -28,7 +28,9 @@ ('trim_command_output.py', 'context-guard-trim-output')) HELPER_PAIRS = (('hook_secret_patterns.py', 'lib/hook_secret_patterns.py'), - ('context_guard_commands.py', 'lib/context_guard_commands.py')) + ('context_guard_commands.py', 'lib/context_guard_commands.py'), + ('context_guard_command_manifest_loader.py', + 'lib/context_guard_command_manifest_loader.py')) NPM_BINS = ('context-guard', 'context-guard-cost', @@ -226,5 +228,6 @@ 'plugins/context-guard/bin/context-guard-statusline-merged', 'plugins/context-guard/bin/context-guard-tool-prune', 'plugins/context-guard/bin/context-guard-trim-output', + 'plugins/context-guard/lib/context_guard_command_manifest_loader.py', 'plugins/context-guard/lib/context_guard_commands.py', 'plugins/context-guard/lib/hook_secret_patterns.py') diff --git a/scripts/prepublish_check.py b/scripts/prepublish_check.py index 6343095..75ed0b1 100755 --- a/scripts/prepublish_check.py +++ b/scripts/prepublish_check.py @@ -7,7 +7,6 @@ from __future__ import annotations import argparse -import ast import json import os import py_compile @@ -69,25 +68,13 @@ r")" ) PATH_LABEL_MAX_CHARS = 160 -MAX_COMMAND_MANIFEST_BYTES = 128 * 1024 +MAX_MANIFEST_HELPER_BYTES = 128 * 1024 ALLOWED_FIRST_ABSOLUTE_SYMLINKS = { "tmp": Path("/private/tmp"), "var": Path("/private/var"), } -COMMAND_MANIFEST_LITERAL_NAMES = { - "IMPLEMENTATION_PAIRS", - "HELPER_PAIRS", - "NPM_BINS", - "NPM_BIN_PATHS", - "DISPATCHER_SUBCOMMANDS", - "LEGACY_WRAPPERS", - "ENTRYPOINT_SMOKE_CASES", - "PLUGIN_ENTRYPOINTS", - "DISPATCHER_SMOKE_CASES", - "EXPECTED_COMMAND_PACK_FILES", -} -def manifest_open_flags() -> int | None: +def trusted_source_open_flags() -> int | None: if not hasattr(os, "O_NOFOLLOW"): return None flags = os.O_RDONLY | os.O_NOFOLLOW @@ -100,25 +87,25 @@ def manifest_open_flags() -> int | None: return flags -def read_manifest_source(path: Path) -> str | None: - flags = manifest_open_flags() +def read_manifest_helper_source(path: Path) -> str | None: + flags = trusted_source_open_flags() if flags is None: return None fd = -1 try: fd = os.open(path, flags) st = os.fstat(fd) - if not stat.S_ISREG(st.st_mode) or st.st_size > MAX_COMMAND_MANIFEST_BYTES: + if not stat.S_ISREG(st.st_mode) or st.st_size > MAX_MANIFEST_HELPER_BYTES: return None chunks: list[bytes] = [] total = 0 while True: - chunk = os.read(fd, min(64 * 1024, MAX_COMMAND_MANIFEST_BYTES + 1 - total)) + chunk = os.read(fd, min(64 * 1024, MAX_MANIFEST_HELPER_BYTES + 1 - total)) if not chunk: break chunks.append(chunk) total += len(chunk) - if total > MAX_COMMAND_MANIFEST_BYTES: + if total > MAX_MANIFEST_HELPER_BYTES: return None return b"".join(chunks).decode("utf-8") except (OSError, UnicodeDecodeError): @@ -131,32 +118,41 @@ def read_manifest_source(path: Path) -> str | None: pass -def literal_command_manifest_from_source(source: str) -> dict[str, object]: +def load_manifest_helper() -> dict[str, object]: + helper_path = KIT_DIR / "context_guard_command_manifest_loader.py" + source = read_manifest_helper_source(helper_path) + if source is None: + raise SystemExit(f"could not load trusted command manifest helper source: {helper_path}") + namespace: dict[str, object] = { + "__builtins__": __builtins__, + "__file__": str(helper_path), + "__name__": "_context_guard_command_manifest_loader", + } try: - tree = ast.parse(source) - except SyntaxError as exc: - raise ValueError(f"invalid Python manifest syntax: line {exc.lineno}: {exc.msg}") from exc - values: dict[str, object] = {} - for node in tree.body: - if isinstance(node, ast.Expr) and isinstance(node.value, ast.Constant) and isinstance(node.value.value, str): - continue - target: str | None = None - value: ast.expr | None = None - if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): - target = node.target.id - value = node.value - elif isinstance(node, ast.Assign) and len(node.targets) == 1 and isinstance(node.targets[0], ast.Name): - target = node.targets[0].id - value = node.value - if target is None: - raise ValueError(f"unsupported executable manifest statement: {type(node).__name__}") - if target not in COMMAND_MANIFEST_LITERAL_NAMES or value is None: - raise ValueError(f"unsupported manifest assignment: {target}") - try: - values[target] = ast.literal_eval(value) - except (SyntaxError, ValueError) as exc: - raise ValueError(f"manifest assignment must be a literal: {target}") from exc - return values + exec(compile(source, str(helper_path), "exec"), namespace) + except Exception as exc: + raise SystemExit(f"could not load trusted command manifest helper: {helper_path}: {exc}") from exc + required = ( + "COMMAND_MANIFEST_LITERAL_NAMES", + "MAX_COMMAND_MANIFEST_BYTES", + "command_manifest_namespace", + "literal_command_manifest_from_source", + "manifest_open_flags", + "read_manifest_source", + ) + missing = [name for name in required if name not in namespace] + if missing: + raise SystemExit(f"trusted command manifest helper missing required API: {', '.join(missing)}") + return namespace + + +COMMAND_MANIFEST_HELPER = load_manifest_helper() +MAX_COMMAND_MANIFEST_BYTES = COMMAND_MANIFEST_HELPER["MAX_COMMAND_MANIFEST_BYTES"] +COMMAND_MANIFEST_LITERAL_NAMES = COMMAND_MANIFEST_HELPER["COMMAND_MANIFEST_LITERAL_NAMES"] +manifest_open_flags = COMMAND_MANIFEST_HELPER["manifest_open_flags"] +read_manifest_source = COMMAND_MANIFEST_HELPER["read_manifest_source"] +literal_command_manifest_from_source = COMMAND_MANIFEST_HELPER["literal_command_manifest_from_source"] +command_manifest_namespace = COMMAND_MANIFEST_HELPER["command_manifest_namespace"] def load_command_manifest(): @@ -169,10 +165,10 @@ def load_command_manifest(): except ValueError as exc: raise SystemExit(f"could not parse trusted command manifest literals: {manifest_path}: {exc}") from exc required = {"IMPLEMENTATION_PAIRS", "HELPER_PAIRS", "NPM_BINS", "LEGACY_WRAPPERS", "EXPECTED_COMMAND_PACK_FILES"} - missing = sorted(required - values.keys()) - if missing: - raise SystemExit(f"trusted command manifest missing required literals: {', '.join(missing)}") - return type("CommandManifest", (), values) + try: + return command_manifest_namespace(values, required=required) + except ValueError as exc: + raise SystemExit(str(exc)) from exc COMMAND_MANIFEST = load_command_manifest() diff --git a/scripts/release_smoke.py b/scripts/release_smoke.py index 169fc50..81c60dc 100755 --- a/scripts/release_smoke.py +++ b/scripts/release_smoke.py @@ -3,7 +3,6 @@ from __future__ import annotations import argparse -import ast import json import os import queue @@ -22,7 +21,8 @@ ROOT = Path(__file__).resolve().parents[1] PLUGIN_DIR = ROOT / "plugins" / "context-guard" PLUGIN_BIN = ROOT / "plugins" / "context-guard" / "bin" -MAX_COMMAND_MANIFEST_BYTES = 128 * 1024 +KIT_DIR = ROOT / "context-guard-kit" +MAX_MANIFEST_HELPER_BYTES = 128 * 1024 PACKAGE_REQUIRED_FILES = (".claude-plugin/plugin.json",) PACKAGE_REQUIRED_DIRS = ("bin", "lib", "skills") PACKAGE_COPY_IGNORE_NAMES = { @@ -37,21 +37,9 @@ "context-guard-diet", "context-guard-audit", ) -COMMAND_MANIFEST_LITERAL_NAMES = { - "IMPLEMENTATION_PAIRS", - "HELPER_PAIRS", - "NPM_BINS", - "NPM_BIN_PATHS", - "DISPATCHER_SUBCOMMANDS", - "LEGACY_WRAPPERS", - "ENTRYPOINT_SMOKE_CASES", - "PLUGIN_ENTRYPOINTS", - "DISPATCHER_SMOKE_CASES", - "EXPECTED_COMMAND_PACK_FILES", -} -def manifest_open_flags() -> int | None: +def trusted_source_open_flags() -> int | None: if not hasattr(os, "O_NOFOLLOW"): return None flags = os.O_RDONLY | os.O_NOFOLLOW @@ -64,25 +52,25 @@ def manifest_open_flags() -> int | None: return flags -def read_manifest_source(path: Path) -> str | None: - flags = manifest_open_flags() +def read_manifest_helper_source(path: Path) -> str | None: + flags = trusted_source_open_flags() if flags is None: return None fd = -1 try: fd = os.open(path, flags) st = os.fstat(fd) - if not stat.S_ISREG(st.st_mode) or st.st_size > MAX_COMMAND_MANIFEST_BYTES: + if not stat.S_ISREG(st.st_mode) or st.st_size > MAX_MANIFEST_HELPER_BYTES: return None chunks: list[bytes] = [] total = 0 while True: - chunk = os.read(fd, min(64 * 1024, MAX_COMMAND_MANIFEST_BYTES + 1 - total)) + chunk = os.read(fd, min(64 * 1024, MAX_MANIFEST_HELPER_BYTES + 1 - total)) if not chunk: break chunks.append(chunk) total += len(chunk) - if total > MAX_COMMAND_MANIFEST_BYTES: + if total > MAX_MANIFEST_HELPER_BYTES: return None return b"".join(chunks).decode("utf-8") except (OSError, UnicodeDecodeError): @@ -95,32 +83,41 @@ def read_manifest_source(path: Path) -> str | None: pass -def literal_command_manifest_from_source(source: str) -> dict[str, Any]: +def load_manifest_helper() -> dict[str, Any]: + helper_path = KIT_DIR / "context_guard_command_manifest_loader.py" + source = read_manifest_helper_source(helper_path) + if source is None: + raise SystemExit(f"could not load trusted command manifest helper source: {helper_path}") + namespace: dict[str, Any] = { + "__builtins__": __builtins__, + "__file__": str(helper_path), + "__name__": "_context_guard_command_manifest_loader", + } try: - tree = ast.parse(source) - except SyntaxError as exc: - raise ValueError(f"invalid Python manifest syntax: line {exc.lineno}: {exc.msg}") from exc - values: dict[str, Any] = {} - for node in tree.body: - if isinstance(node, ast.Expr) and isinstance(node.value, ast.Constant) and isinstance(node.value.value, str): - continue - target: str | None = None - value: ast.expr | None = None - if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): - target = node.target.id - value = node.value - elif isinstance(node, ast.Assign) and len(node.targets) == 1 and isinstance(node.targets[0], ast.Name): - target = node.targets[0].id - value = node.value - if target is None: - raise ValueError(f"unsupported executable manifest statement: {type(node).__name__}") - if target not in COMMAND_MANIFEST_LITERAL_NAMES or value is None: - raise ValueError(f"unsupported manifest assignment: {target}") - try: - values[target] = ast.literal_eval(value) - except (SyntaxError, ValueError) as exc: - raise ValueError(f"manifest assignment must be a literal: {target}") from exc - return values + exec(compile(source, str(helper_path), "exec"), namespace) + except Exception as exc: + raise SystemExit(f"could not load trusted command manifest helper: {helper_path}: {exc}") from exc + required = ( + "COMMAND_MANIFEST_LITERAL_NAMES", + "MAX_COMMAND_MANIFEST_BYTES", + "command_manifest_namespace", + "literal_command_manifest_from_source", + "manifest_open_flags", + "read_manifest_source", + ) + missing = [name for name in required if name not in namespace] + if missing: + raise SystemExit(f"trusted command manifest helper missing required API: {', '.join(missing)}") + return namespace + + +COMMAND_MANIFEST_HELPER = load_manifest_helper() +MAX_COMMAND_MANIFEST_BYTES = COMMAND_MANIFEST_HELPER["MAX_COMMAND_MANIFEST_BYTES"] +COMMAND_MANIFEST_LITERAL_NAMES = COMMAND_MANIFEST_HELPER["COMMAND_MANIFEST_LITERAL_NAMES"] +manifest_open_flags = COMMAND_MANIFEST_HELPER["manifest_open_flags"] +read_manifest_source = COMMAND_MANIFEST_HELPER["read_manifest_source"] +literal_command_manifest_from_source = COMMAND_MANIFEST_HELPER["literal_command_manifest_from_source"] +command_manifest_namespace = COMMAND_MANIFEST_HELPER["command_manifest_namespace"] def load_command_manifest(): @@ -133,10 +130,10 @@ def load_command_manifest(): except ValueError as exc: raise SystemExit(f"could not parse trusted command manifest literals: {manifest_path}: {exc}") from exc required = {"ENTRYPOINT_SMOKE_CASES", "DISPATCHER_SMOKE_CASES"} - missing = sorted(required - values.keys()) - if missing: - raise SystemExit(f"trusted command manifest missing required literals: {', '.join(missing)}") - return type("CommandManifest", (), values) + try: + return command_manifest_namespace(values, required=required) + except ValueError as exc: + raise SystemExit(str(exc)) from exc COMMAND_MANIFEST = load_command_manifest() @@ -757,6 +754,59 @@ def check_auto_explain_smoke(proc: subprocess.CompletedProcess[str], command: st fail(f"{command} adaptive_k missing source verification safeguard") +def run_entrypoint_launch_smokes( + *, + plugin_bin: Path, + launch_plan: dict[str, dict[str, Any]], + cwd: Path, + env: dict[str, str], + timeout: float, +) -> None: + for name, plan in launch_plan.items(): + mode = str(plan["mode"]) + run_command( + entrypoint_launch_argv(command_path(plugin_bin, name), list(plan["args"])), + cwd=cwd, + env=env, + timeout=timeout, + input_text=launch_stdin(mode), + expect=lambda proc, command=name, launch_mode=mode: check_launch_smoke(proc, command, launch_mode), + ) + + +def run_dispatcher_launch_smokes( + *, + bin_dir: Path, + plans: tuple[dict[str, Any], ...], + cwd: Path, + env: dict[str, str], + timeout: float, + trusted_root: Path | None = None, + label_prefix: str = "", +) -> None: + for plan in plans: + entrypoint = str(plan["entrypoint"]) + mode = str(plan["mode"]) + args = [str(arg) for arg in plan["args"]] + entrypoint_path = bin_dir / entrypoint + if not entrypoint_path.is_file(): + fail(f"{label_prefix}{entrypoint} dispatcher bin missing: {entrypoint_path}") + if trusted_root is None: + argv = entrypoint_launch_argv(command_path(bin_dir, entrypoint), args) + else: + require_path_inside(entrypoint_path, trusted_root, label=f"{entrypoint} npm bin") + argv = entrypoint_launch_argv(entrypoint_path, args, trusted_root=trusted_root) + command_label = " ".join([label_prefix.rstrip(), entrypoint, *args]).strip() + run_command( + argv, + cwd=cwd, + env=env, + timeout=timeout, + input_text=launch_stdin(mode), + expect=lambda proc, command=command_label, launch_mode=mode: check_launch_smoke(proc, command, launch_mode), + ) + + def run_smoke(plugin_bin: Path, timeout: float) -> None: plugin_bin = plugin_bin.resolve() commands = {name: command_path(plugin_bin, name) for name in REQUIRED_COMMANDS} @@ -890,29 +940,14 @@ def run_smoke(plugin_bin: Path, timeout: float) -> None: check_json_field(load_json(proc.stdout, "context-guard-audit"), "records", 1, "context-guard-audit") ), ) - for name, plan in launch_plan.items(): - mode = str(plan["mode"]) - run_command( - entrypoint_launch_argv(command_path(plugin_bin, name), list(plan["args"])), - cwd=project, - env=env, - timeout=timeout, - input_text=launch_stdin(mode), - expect=lambda proc, command=name, launch_mode=mode: check_launch_smoke(proc, command, launch_mode), - ) - for plan in DISPATCHER_SMOKE_COMMANDS: - entrypoint = str(plan["entrypoint"]) - mode = str(plan["mode"]) - args = [str(arg) for arg in plan["args"]] - command_label = " ".join([entrypoint, *args]) - run_command( - entrypoint_launch_argv(command_path(plugin_bin, entrypoint), args), - cwd=project, - env=env, - timeout=timeout, - input_text=launch_stdin(mode), - expect=lambda proc, command=command_label, launch_mode=mode: check_launch_smoke(proc, command, launch_mode), - ) + run_entrypoint_launch_smokes(plugin_bin=plugin_bin, launch_plan=launch_plan, cwd=project, env=env, timeout=timeout) + run_dispatcher_launch_smokes( + bin_dir=plugin_bin, + plans=DISPATCHER_SMOKE_COMMANDS, + cwd=project, + env=env, + timeout=timeout, + ) def run_npm_package_smoke(timeout: float) -> None: @@ -1082,23 +1117,15 @@ def run_npm_package_smoke(timeout: float) -> None: "isolated npm context-guard setup brief-mode apply", ), ) - for plan in npm_dispatcher_smoke_plan(): - entrypoint = str(plan["entrypoint"]) - mode = str(plan["mode"]) - args = [str(arg) for arg in plan["args"]] - entrypoint_path = isolated_bin / entrypoint - if not entrypoint_path.is_file(): - fail(f"isolated npm install missing dispatcher bin: {entrypoint_path}") - require_path_inside(entrypoint_path, install_prefix, label=f"{entrypoint} npm bin") - command_label = " ".join(["isolated npm", entrypoint, *args]) - run_command( - entrypoint_launch_argv(entrypoint_path, args, trusted_root=install_prefix), - cwd=project, - env=env, - timeout=timeout, - input_text=launch_stdin(mode), - expect=lambda proc, command=command_label, launch_mode=mode: check_launch_smoke(proc, command, launch_mode), - ) + run_dispatcher_launch_smokes( + bin_dir=isolated_bin, + plans=npm_dispatcher_smoke_plan(), + cwd=project, + env=env, + timeout=timeout, + trusted_root=install_prefix, + label_prefix="isolated npm ", + ) def npm_dispatcher_smoke_plan() -> tuple[dict[str, Any], ...]: diff --git a/tests/test_context_guard_kit.py b/tests/test_context_guard_kit.py index e844e0d..ad14745 100644 --- a/tests/test_context_guard_kit.py +++ b/tests/test_context_guard_kit.py @@ -1,5 +1,4 @@ import argparse -import ast import base64 import csv import contextlib @@ -34,39 +33,33 @@ KIT_REWRITE = KIT_DIR / "rewrite_bash_for_token_budget.py" PLUGIN_REWRITE = PLUGIN_BIN / "context-guard-rewrite-bash" SAFE_SHELL = shutil.which("sh") or "/bin/sh" -COMMAND_MANIFEST_LITERAL_NAMES = { - "IMPLEMENTATION_PAIRS", - "HELPER_PAIRS", - "NPM_BINS", - "NPM_BIN_PATHS", - "DISPATCHER_SUBCOMMANDS", - "LEGACY_WRAPPERS", - "ENTRYPOINT_SMOKE_CASES", - "PLUGIN_ENTRYPOINTS", - "DISPATCHER_SMOKE_CASES", - "EXPECTED_COMMAND_PACK_FILES", -} + + +def load_manifest_helper_for_tests(): + helper_path = KIT_DIR / "context_guard_command_manifest_loader.py" + namespace = { + "__builtins__": __builtins__, + "__file__": str(helper_path), + "__name__": "_context_guard_command_manifest_loader_for_tests", + } + exec(compile(helper_path.read_text(encoding="utf-8"), str(helper_path), "exec"), namespace) + return type("CommandManifestLoaderForTests", (), namespace) + + +COMMAND_MANIFEST_HELPER = load_manifest_helper_for_tests() +COMMAND_MANIFEST_LITERAL_NAMES = COMMAND_MANIFEST_HELPER.COMMAND_MANIFEST_LITERAL_NAMES def load_command_manifest_for_tests(): manifest_path = KIT_DIR / "context_guard_commands.py" - tree = ast.parse(manifest_path.read_text(encoding="utf-8")) - values = {} - for node in tree.body: - if isinstance(node, ast.Expr) and isinstance(node.value, ast.Constant) and isinstance(node.value.value, str): - continue - target = None - value = None - if isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): - target = node.target.id - value = node.value - elif isinstance(node, ast.Assign) and len(node.targets) == 1 and isinstance(node.targets[0], ast.Name): - target = node.targets[0].id - value = node.value - if target is None or target not in COMMAND_MANIFEST_LITERAL_NAMES or value is None: - raise RuntimeError(f"command manifest must contain only literal assignments: {manifest_path}") - values[target] = ast.literal_eval(value) - return type("CommandManifestForTests", (), values) + try: + values = COMMAND_MANIFEST_HELPER.literal_command_manifest_from_source( + manifest_path.read_text(encoding="utf-8"), + allowed_names=COMMAND_MANIFEST_LITERAL_NAMES, + ) + return COMMAND_MANIFEST_HELPER.command_manifest_namespace(values) + except ValueError as exc: + raise RuntimeError(f"command manifest must contain only literal assignments: {manifest_path}: {exc}") from exc COMMAND_MANIFEST = load_command_manifest_for_tests() @@ -413,6 +406,10 @@ def test_command_manifest_covers_release_and_runtime_surfaces(self): self.assertTrue(set(COMMAND_MANIFEST.LEGACY_WRAPPERS).issubset(set(COMMAND_MANIFEST.PLUGIN_ENTRYPOINTS))) prepublish = load_module_from_path(ROOT / "scripts" / "prepublish_check.py", "prepublish_manifest_test") + self.assertIs( + prepublish.literal_command_manifest_from_source, + prepublish.COMMAND_MANIFEST_HELPER["literal_command_manifest_from_source"], + ) self.assertEqual(prepublish.IMPLEMENTATION_PAIRS, COMMAND_MANIFEST.IMPLEMENTATION_PAIRS) self.assertEqual(prepublish.HELPER_PAIRS, COMMAND_MANIFEST.HELPER_PAIRS) self.assertEqual(prepublish.REQUIRED_NPM_BINS, set(COMMAND_MANIFEST.NPM_BINS)) @@ -436,6 +433,10 @@ def test_command_manifest_covers_release_and_runtime_surfaces(self): ) smoke = load_module_from_path(ROOT / "scripts" / "release_smoke.py", "release_smoke_manifest_test") + self.assertIs( + smoke.literal_command_manifest_from_source, + smoke.COMMAND_MANIFEST_HELPER["literal_command_manifest_from_source"], + ) self.assertEqual(smoke.ENTRYPOINT_SMOKE_COMMANDS, COMMAND_MANIFEST.ENTRYPOINT_SMOKE_CASES) self.assertEqual(smoke.DISPATCHER_SMOKE_COMMANDS, COMMAND_MANIFEST.DISPATCHER_SMOKE_CASES) self.assertEqual(set(smoke.ENTRYPOINT_SMOKE_COMMANDS), set(COMMAND_MANIFEST.PLUGIN_ENTRYPOINTS)) @@ -448,7 +449,11 @@ def test_command_manifest_loaders_reject_executable_python(self): non_literal = "DISPATCHER_SUBCOMMANDS = dict(setup=('context-guard-setup',))\n" import_stmt = "from typing import Any\nDISPATCHER_SUBCOMMANDS = {'setup': ('context-guard-setup',)}\n" function_stmt = "DISPATCHER_SUBCOMMANDS = {'setup': ('context-guard-setup',)}\ndef expected_command_pack_files():\n return ()\n" - for loader in (smoke.literal_command_manifest_from_source, prepublish.literal_command_manifest_from_source): + for loader in ( + COMMAND_MANIFEST_HELPER.literal_command_manifest_from_source, + smoke.literal_command_manifest_from_source, + prepublish.literal_command_manifest_from_source, + ): with self.subTest(loader=loader.__module__, case="executable"): with self.assertRaises(ValueError): loader(malicious) @@ -470,6 +475,10 @@ def test_context_guard_dispatcher_ignores_pythonpath_command_manifest_shadow(sel "DISPATCHER_SUBCOMMANDS = {'pwned': ('context-guard-pwned',)}\n", encoding="utf-8", ) + (shadow / "context_guard_command_manifest_loader.py").write_text( + "raise RuntimeError('helper shadow executed')\n", + encoding="utf-8", + ) env = os.environ.copy() env["PYTHONPATH"] = str(shadow) for script in (KIT_DIR / "context_guard_cli.py", PLUGIN_BIN / "context-guard"): @@ -485,6 +494,7 @@ def test_context_guard_dispatcher_ignores_pythonpath_command_manifest_shadow(sel self.assertIn(" experiments", proc.stdout) self.assertNotIn("pwned", proc.stdout) self.assertNotIn("context-guard-pwned", proc.stdout) + self.assertNotIn("helper shadow executed", proc.stderr) def test_staged_plugin_dispatcher_runs_with_packaged_command_manifest(self): smoke = load_module_from_path(ROOT / "scripts" / "release_smoke.py", "release_smoke_manifest_stage_test")