Centralize manifest release loading

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:

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)

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')