diff --git a/src/nexus_symbiote/policy.py b/src/nexus_symbiote/policy.py index c4b9b62..86833f1 100644 --- a/src/nexus_symbiote/policy.py +++ b/src/nexus_symbiote/policy.py @@ -159,16 +159,35 @@ def find_policy_path(project_root: Path) -> Path | None: belongs to whoever controls that directory, not to the project. Both file names at once are refused: either could silently shadow the other. """ + from nexus_symbiote.utils.confined import fold_name + current = project_root.resolve() if current.is_file(): current = current.parent + try: + entries = sorted(os.listdir(current)) + except OSError: + return None + folded_targets = {fold_name(name): name for name in POLICY_FILE_NAMES} found: list[Path] = [] - for name in POLICY_FILE_NAMES: - candidate = current / name + variants: list[tuple[str, str]] = [] + for entry in entries: + canonical = folded_targets.get(fold_name(entry)) + if canonical is None: + continue + candidate = current / entry if candidate.is_symlink(): raise PolicyError(f"policy must not be a symlink: {candidate}") - if candidate.exists(): - found.append(candidate) + if entry in POLICY_FILE_NAMES: + if candidate.exists(): + found.append(candidate) + else: + variants.append((entry, canonical)) + if variants: + variant_name, canonical = variants[0] + raise PolicyError( + f"case-variant policy file {variant_name} in {current}; rename it to {canonical}" + ) if len(found) > 1: raise PolicyError( f"both {POLICY_FILE_NAMES[0]} and {POLICY_FILE_NAMES[1]} exist in {current}; " diff --git a/src/nexus_symbiote/scanners/live_mcp.py b/src/nexus_symbiote/scanners/live_mcp.py index 3309f20..3b7b2d9 100644 --- a/src/nexus_symbiote/scanners/live_mcp.py +++ b/src/nexus_symbiote/scanners/live_mcp.py @@ -545,23 +545,60 @@ def request( def _read_sse(self, response: http.client.HTTPResponse, body: bytes, maximum: int) -> bytes: """Read event lines until the requested response arrives. - HTTPResponse.readline decodes chunked transfer coding. A single line may - use the whole response byte limit, since SDK servers put an entire result - on one data line. + HTTPResponse.read1 / readline decodes chunked transfer coding. Lines may + terminate in CRLF, LF or bare CR (WHATWG SSE), and a single line may use + the whole response byte limit when SDK servers emit one large data line. """ request_id = _json(body).get("id") decoder = _SSEDecoder(self._max_sse_events) payload = bytearray() + buffer = bytearray() + saw_bare_cr = False + use_read1 = hasattr(response, "read1") while True: - line = response.readline(maximum + 1 - len(payload)) - if not line: - break - payload.extend(line) - if len(payload) > maximum: + remaining = maximum + 1 - len(payload) - len(buffer) + if remaining <= 0: raise DiscoveryError("SSE response byte limit exceeded") - for message in decoder.feed(line): - if message.get("id") == request_id and ("result" in message or "error" in message): - return bytes(payload) + chunk = ( + response.read1(min(4096, remaining)) if use_read1 else response.readline(remaining) + ) + eof = not chunk + buffer.extend(chunk) + while buffer: + cr = buffer.find(b"\r") + lf = buffer.find(b"\n") + if cr < 0 and lf < 0: + if len(payload) + len(buffer) > maximum: + raise DiscoveryError("SSE response byte limit exceeded") + if not eof: + break + end = len(buffer) + elif lf >= 0 and (cr < 0 or lf < cr): + end = lf + 1 + elif cr + 1 < len(buffer): + if buffer[cr + 1] == 0x0A: + end = cr + 2 + else: + saw_bare_cr = True + end = cr + 1 + elif eof or saw_bare_cr: + end = cr + 1 + else: + if len(payload) + len(buffer) > maximum: + raise DiscoveryError("SSE response byte limit exceeded") + break + line = bytes(buffer[:end]) + del buffer[:end] + payload.extend(line) + if len(payload) > maximum: + raise DiscoveryError("SSE response byte limit exceeded") + for message in decoder.feed(line): + if message.get("id") == request_id and ( + "result" in message or "error" in message + ): + return bytes(payload) + if eof: + break raise DiscoveryError("SSE ended without matching response") diff --git a/src/nexus_symbiote/scanners/mcp.py b/src/nexus_symbiote/scanners/mcp.py index bb8bfc0..1d7556b 100644 --- a/src/nexus_symbiote/scanners/mcp.py +++ b/src/nexus_symbiote/scanners/mcp.py @@ -750,12 +750,17 @@ def _purl(package: Package) -> str | None: " unsafeHttpWhitelist checksumBehavior enableOfflineMode injectEnvironmentFiles" " packageExtensions" ) +_BUN_IDENTITY_SETTINGS = frozenset({"preload", "run.preload"}) +_BUN_CODE_SETTINGS = frozenset({"run.shell"}) +_BUN_SOURCE_SETTINGS = frozenset({"install.ca", "install.cafile"}) +_DENO_SOURCE_SETTINGS = frozenset({"importMap", "imports", "scopes"}) _RESOLVER_FILES = { "npm": (".npmrc",), - "deno": (".npmrc",), + "deno": (".npmrc", "deno.json", "deno.jsonc"), "bun": (".npmrc", "bunfig.toml"), "yarn": (".yarnrc.yml", ".yarnrc", ".env.yarn"), "uv-project": ("uv.toml", "pyproject.toml"), + "uv-tool": ("uv.toml", "pyproject.toml"), } # Setting names exported in clear: fixed vocabularies, never a name taken from input. _KNOWN_SETTINGS = ( @@ -765,6 +770,10 @@ def _purl(package: Package) -> str | None: | _NPM_CODE_SETTINGS | _YARN_CODE_SETTINGS | _YARN_SOURCE_SETTINGS + | _BUN_IDENTITY_SETTINGS + | _BUN_CODE_SETTINGS + | _BUN_SOURCE_SETTINGS + | _DENO_SOURCE_SETTINGS | frozenset(_UV_SETTINGS) | {"registry", "npmRegistryServer", "offline", "prefer-offline", "strict-ssl", "environment"} ) @@ -974,6 +983,14 @@ def _file_setting(path: str, family: str, key: str, value: str, launched: set[st "enableOfflineMode": "false", } return "inert" if relaxed.get(key) == value else "source" + if path == "bunfig.toml": + if key in _BUN_IDENTITY_SETTINGS: + return "identity" + if key in _BUN_CODE_SETTINGS: + return "code" + return "source" if key in _BUN_SOURCE_SETTINGS else "inert" + if path in {"deno.json", "deno.jsonc"}: + return "source" if key in _DENO_SOURCE_SETTINGS else "inert" if path in {"uv.toml", "pyproject.toml"}: return "source" if not key.startswith("sources.") or key[8:] in launched else "inert" return "unresolved" if path == ".env.yarn" else "inert" @@ -1135,14 +1152,67 @@ def _read(self, path: str) -> _ResolverFile: if key in _YARN_CODE_SETTINGS or key in _YARN_SOURCE_SETTINGS ] return _ResolverFile(tuple(entries)) + if path in {"deno.json", "deno.jsonc"}: + text = self._text(path) + if text is None: + return _ResolverFile() + try: + data = _load_jsonc(text) + except (ValueError, RecursionError): + self._result.scan_failures.append(f"{path}: invalid JSON or duplicate keys") + return _ResolverFile() + if not isinstance(data, dict): + return _ResolverFile() + deno_entries: list[tuple[str, str]] = [] + import_map = data.get("importMap") + if isinstance(import_map, str) and import_map: + deno_entries.append(("importMap", import_map)) + imports = data.get("imports") + if isinstance(imports, dict): + for target in imports.values(): + if isinstance(target, str): + deno_entries.append(("imports", target)) + scopes = data.get("scopes") + if isinstance(scopes, dict) and scopes: + deno_entries.append(("scopes", json.dumps(scopes, sort_keys=True, default=str))) + return _ResolverFile(tuple(deno_entries)) settings = self._toml(path) if path == "bunfig.toml": install = settings.get("install") - return _ResolverFile( - tuple( - _registries(install if isinstance(install, dict) else {}, "registry", "scopes") + install_map = install if isinstance(install, dict) else {} + bun_entries = _registries(install_map, "registry", "scopes") + for ca_key in ("ca", "cafile"): + if ca_key in install_map and install_map[ca_key]: + bun_entries.append( + ( + f"install.{ca_key}", + install_map[ca_key] + if isinstance(install_map[ca_key], str) + else json.dumps(install_map[ca_key], default=str), + ) + ) + if settings.get("preload"): + bun_entries.append( + ( + "preload", + settings["preload"] + if isinstance(settings["preload"], str) + else json.dumps(settings["preload"], default=str), + ) ) - ) + run_cfg = settings.get("run") + if isinstance(run_cfg, dict): + for run_key in ("preload", "shell"): + if run_cfg.get(run_key): + bun_entries.append( + ( + f"run.{run_key}", + run_cfg[run_key] + if isinstance(run_cfg[run_key], str) + else json.dumps(run_cfg[run_key], default=str), + ) + ) + return _ResolverFile(tuple(bun_entries)) if path == "pyproject.toml": tool = settings.get("tool") uv = tool.get("uv") if isinstance(tool, dict) else None @@ -1860,18 +1930,19 @@ def _resolver_view( view.names.append(label) kind = _file_setting(path, family, key, value, launched) if kind == "identity": - view.identity.append( - f"npm {key} setting from {path}" - if path == ".npmrc" - else f"{path} {key} replaces or extends yarn" - ) + if path == ".npmrc": + view.identity.append(f"npm {key} setting from {path}") + elif path == "bunfig.toml": + view.identity.append(f"{path} {key} preloads local code") + else: + view.identity.append(f"{path} {key} replaces or extends yarn") elif kind == "code": view.code += 1 elif kind == "source": view.sources.append((label, value)) elif kind == "unresolved": view.unresolved.append(label) - if path in {"uv.toml", "pyproject.toml", ".env.yarn"}: + if path in {"uv.toml", "pyproject.toml", ".env.yarn", "deno.json", "deno.jsonc"}: continue default = "npmRegistryServer" if path == ".yarnrc.yml" else "registry" scope_label, scope = ( diff --git a/src/nexus_symbiote/scanners/mcp_launch.py b/src/nexus_symbiote/scanners/mcp_launch.py index 3ad4e53..befad26 100644 --- a/src/nexus_symbiote/scanners/mcp_launch.py +++ b/src/nexus_symbiote/scanners/mcp_launch.py @@ -18,6 +18,7 @@ from __future__ import annotations import fnmatch +import posixpath import re import shlex from collections.abc import Callable, Iterator @@ -1740,12 +1741,18 @@ def _program_reason(name: str, rest: list[str], launch: Launch) -> str | None: return _embedded_reason(rest) -def _start(words: list[str], launch: Launch) -> None: +def _start(words: list[str], launch: Launch, *, package_entry: bool = False) -> None: """A program a runner starts by name: record it and check what it starts in turn.""" launch.executable = normalize_command(words[0]) family = _interpreter_family(launch.executable) reason = _interpreter_reason(family, words[1:], launch) if family else None + # What the words start in turn is the more specific reason, so it is named first. reason = reason or _embedded_reason(words) + if reason is None and package_entry: + if re.search(r"[\\/]", words[0]) or words[0].startswith("."): + reason = "runner command is a path, not a package entry point" + elif family is not None: + reason = f"runner starts {family} instead of a package entry point" if reason is not None: launch.fail(reason) @@ -2181,7 +2188,7 @@ def _package_runner( launch.packages.append(package) return if position < len(rest): - _start(rest[position:], launch) + _start(rest[position:], launch, package_entry=True) launch.packages.extend(npm_spec(spec) for spec in specs) @@ -2232,6 +2239,15 @@ def _npm(name: str, rest: list[str], launch: Launch) -> None: return if subcommand in {"exec", "x"}: launch.launcher = "npm exec" + scanned = _scan(remainder, _NPM_EXEC) + if scanned is not None: + _, pos = scanned + if pos < len(remainder) and (pos == 0 or remainder[pos - 1] != "--"): + after = remainder[pos + 1 :] + unquoted = after[: after.index("--")] if "--" in after else after + if any(word.startswith("-") and word != "-" for word in unquoted): + launch.fail("npm exec options after the package without '--'") + return _package_runner(remainder, launch, _NPM_EXEC, {"-p", "--package"}, options) elif subcommand in _NPM_LOCAL: launch.launcher = f"npm {subcommand}" @@ -2342,7 +2358,7 @@ def _python_runner( if declared is not None: launch.packages.insert(0, python_spec(declared)) if position < len(rest): - _start(rest[position:], launch) + _start(rest[position:], launch, package_entry=True) else: package = python_spec(rest[position], shorthand=shorthand) launch.packages.insert(0, package) @@ -2492,12 +2508,18 @@ def _host_source(spec: str, *, mount: bool) -> str | None: if fields.get("type", "volume") != "bind": return None return fields.get("source") or fields.get("src") or None + if ":" not in spec: + return None source = spec.split(":", 1)[0] if re.match(r"[A-Za-z]:[\\/]", spec): + if ":" not in spec[2:]: + return None source = spec[:2] + spec[2:].split(":", 1)[0] return ( source - if source.startswith(("/", "~", "$", "%", "\\")) or re.match(r"[A-Za-z]:", source) + if source in {".", ".."} + or source.startswith(("/", "./", "../", ".\\", "..\\", "~", "$", "%", "\\")) + or re.match(r"[A-Za-z]:", source) else None ) @@ -2525,12 +2547,22 @@ def _container_context(options: list[tuple[str, str | None]], launch: Launch) -> if source is None: continue launch.mark("host_mounts") - normalized = source.rstrip("/\\") or "/" - if normalized in {"/", "\\"} or re.fullmatch(r"[A-Za-z]:[\\/]?", source): + normalized = source.rstrip("/\\") or ("/" if source.startswith("/") else "\\") + canonical = ( + posixpath.normpath(normalized) + if normalized.startswith("/") and "\\" not in normalized + else normalized + ) + if canonical in {"/", "\\"} or re.fullmatch(r"[A-Za-z]:[\\/]?", canonical): launch.mark("host_root_mount") - elif source.lower().endswith(_SOCKETS): + elif canonical.lower().endswith(_SOCKETS): launch.mark("runtime_socket_mount") - elif _BROAD.fullmatch(source): + elif ( + _BROAD.fullmatch(canonical) + or canonical == ".." + or canonical.startswith(("../", "..\\")) + or posixpath.normpath(normalized.replace("\\", "/")).startswith("..") + ): launch.mark("broad_host_mount") elif option == "--entrypoint": launch.mark("entrypoint_override") diff --git a/src/nexus_symbiote/utils/confined.py b/src/nexus_symbiote/utils/confined.py index 368ecbc..64dc5c6 100644 --- a/src/nexus_symbiote/utils/confined.py +++ b/src/nexus_symbiote/utils/confined.py @@ -314,6 +314,8 @@ def fold_name(name: str) -> str: Case and Unicode normalisation are folded, as on APFS, HFS+ and NTFS; so are the format characters HFS+ ignores and the trailing dots and spaces Windows drops. """ + if name.isascii(): + return name.lower().rstrip(". ") folded = unicodedata.normalize("NFD", unicodedata.normalize("NFD", name).casefold()) return folded.translate(_IGNORED_IN_NAMES).rstrip(". ") @@ -502,6 +504,10 @@ def __init__( self._root_fd: int | None = None self._links: dict[str, str] = {} self._listings: dict[Path, dict[str, list[tuple[str, str]]] | None] = {} + self._exact_names: dict[tuple[int, int], tuple[tuple[int, int], frozenset[str]]] = {} + self._case_sensitive_dirs: set[tuple[int, int]] = set() + self._root_folds = False + self._folded_excluded: set[tuple[str, ...]] | None = None self._variant_budget = self.max_entries self._excluded: set[tuple[str, ...]] = set() self._invalid_policy = False @@ -526,10 +532,38 @@ def __enter__(self) -> ConfinedCollection: return self try: self._root_fd = open_directory_nofollow(self._requested_root) + self._root_folds = self._detect_root_folding() except OSError: self._fail(".", "root unavailable or contains a symlink") return self + def _detect_root_folding(self) -> bool: + """Whether the root's file system folds case on directory lookups.""" + if self._root_fd is None: + return False + try: + root_stat = os.fstat(self._root_fd) + except OSError: + return False + for candidate in (self.root, *self.root.parents): + name = candidate.name + if not name: + break + swapped = name.swapcase() + if swapped == name: + continue + try: + exact = os.stat(candidate, follow_symlinks=False) + if exact.st_dev != root_stat.st_dev: + break + alt = os.stat(candidate.parent / swapped, follow_symlinks=False) + return (alt.st_dev, alt.st_ino) == (exact.st_dev, exact.st_ino) + except FileNotFoundError: + return False + except OSError: + continue + return False + def __exit__(self, *args: object) -> None: if self._root_fd is not None: os.close(self._root_fd) @@ -622,22 +656,108 @@ def excluded(self, path: str | Path) -> bool: # Dependency caches, virtual environments and Git internals are outside # the documented static scope. The BOM records these patterns. Any # further omission within that scope is a blocking coverage gap. - if any(part in DEFAULT_EXCLUDES for part in parts): + if any(part in DEFAULT_EXCLUDES or fold_name(part) in DEFAULT_EXCLUDES for part in parts): return True denied = bool(self._excluded) and any( parts[:index] in self._excluded for index in range(len(parts) + 1) ) + if not denied and self._excluded and self._root_folds: + if self._folded_excluded is None: + self._folded_excluded = { + tuple(fold_name(item) for item in entry) for entry in self._excluded + } + folded = tuple(fold_name(part) for part in parts) + denied = any( + folded[:index] in self._folded_excluded for index in range(len(folded) + 1) + ) if denied: self._exclusion_log.add(f"{relative}: excluded by collection policy") return denied + def _has_exact_name(self, directory_fd: int, name: str) -> bool: + """Whether ``directory_fd`` holds an entry whose exact spelling is ``name``.""" + info = os.fstat(directory_fd) + dir_key = (info.st_dev, info.st_ino) + simple_ascii = name.isascii() and not name.endswith((".", " ")) + if simple_ascii and dir_key in self._case_sensitive_dirs: + return True + if simple_ascii: + swapped = name.swapcase() + if swapped != name: + try: + os.stat(swapped, dir_fd=directory_fd, follow_symlinks=False) + except FileNotFoundError: + self._case_sensitive_dirs.add(dir_key) + return True + except OSError: + # Any other failure decides nothing; the listing below does. + pass + stamp = (info.st_mtime_ns, info.st_ctime_ns) + cached = self._exact_names.get(dir_key) + if cached is None or cached[0] != stamp: + dup_fd = os.dup(directory_fd) + try: + with os.scandir(dup_fd) as stream: + names = frozenset(entry.name for entry in stream) + finally: + os.close(dup_fd) + cached = (stamp, names) + self._exact_names[dir_key] = cached + return name in cached[1] + + def _open_dir_part(self, directory_fd: int, part: str) -> int: + try: + child_fd = os.open( + part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=directory_fd + ) + except FileNotFoundError: + raise + except OSError as exc: + if exc.errno in _LINK_ERRNOS and not self._has_exact_name(directory_fd, part): + raise FileNotFoundError(errno.ENOENT, "exact entry not found") from exc + raise + try: + exact = self._has_exact_name(directory_fd, part) + except BaseException: + os.close(child_fd) + raise + if not exact: + os.close(child_fd) + raise FileNotFoundError(errno.ENOENT, "exact entry not found") + return child_fd + + def _open_leaf_no_follow(self, directory_fd: int, name: str) -> int: + try: + fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory_fd) + except FileNotFoundError: + raise + except OSError as exc: + if exc.errno in _LINK_ERRNOS and not self._has_exact_name(directory_fd, name): + raise FileNotFoundError(errno.ENOENT, "exact entry not found") from exc + raise + try: + exact = self._has_exact_name(directory_fd, name) + except BaseException: + os.close(fd) + raise + if not exact: + os.close(fd) + raise FileNotFoundError(errno.ENOENT, "exact entry not found") + return fd + + def _stat_exact_no_follow(self, directory_fd: int, name: str) -> os.stat_result: + info = os.stat(name, dir_fd=directory_fd, follow_symlinks=False) + if not self._has_exact_name(directory_fd, name): + raise FileNotFoundError(errno.ENOENT, "exact entry not found") + return info + def _open_parent(self, relative: Path) -> int: if self._root_fd is None: raise CollectionError("collection root unavailable") fd = os.dup(self._root_fd) try: for part in relative.parts[:-1]: - next_fd = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=fd) + next_fd = self._open_dir_part(fd, part) os.close(fd) fd = next_fd return fd @@ -737,9 +857,7 @@ def open_regular( raise CollectionError("file excluded by policy") parent = self._open_parent(relative) try: - fd = os.open( - relative.name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=parent - ) + fd = self._open_leaf_no_follow(parent, relative.name) finally: os.close(parent) try: @@ -784,7 +902,7 @@ def path_identity(self, path: str | Path) -> tuple[int, int] | None: return None parent = self._open_parent(relative) try: - info = os.stat(relative.name, dir_fd=parent, follow_symlinks=False) + info = self._stat_exact_no_follow(parent, relative.name) finally: os.close(parent) except (OSError, CollectionError): @@ -810,9 +928,7 @@ def hash_file(self, path: str | Path, *, max_bytes: int | Bound | None = None) - return None parent = self._open_parent(relative) try: - fd = os.open( - relative.name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=parent - ) + fd = self._open_leaf_no_follow(parent, relative.name) finally: os.close(parent) try: @@ -1090,9 +1206,7 @@ def is_directory(self, path: str | Path) -> bool: def _open_no_follow(self, relative: Path) -> int: parent = self._open_parent(relative) try: - return os.open( - relative.name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=parent - ) + return self._open_leaf_no_follow(parent, relative.name) finally: os.close(parent) @@ -1155,7 +1269,7 @@ def _open_directory_no_follow(self, relative: Path) -> int: fd = os.dup(self._root_fd) try: for part in relative.parts: - next_fd = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=fd) + next_fd = self._open_dir_part(fd, part) os.close(fd) fd = next_fd return fd @@ -1186,7 +1300,7 @@ def _link_target(self, relative: Path) -> tuple[Path, int] | None: fd = os.dup(self._root_fd) try: for index, part in enumerate(relative.parts): - info = os.stat(part, dir_fd=fd, follow_symlinks=False) + info = self._stat_exact_no_follow(fd, part) if stat.S_ISLNK(info.st_mode): text = os.readlink(part, dir_fd=fd) target = _lexical_target(Path(*relative.parts[:index]), text) @@ -1201,7 +1315,7 @@ def _link_target(self, relative: Path) -> tuple[Path, int] | None: return target, len(relative.parts) - index - 1 if index + 1 == len(relative.parts) or not stat.S_ISDIR(info.st_mode): return None - next_fd = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=fd) + next_fd = self._open_dir_part(fd, part) os.close(fd) fd = next_fd return None @@ -1237,7 +1351,7 @@ def _walked_link(self, directory_fd: int, child: Path) -> Path | None: try: parent = self._open_parent(target) try: - info = os.stat(target.name, dir_fd=parent, follow_symlinks=False) + info = self._stat_exact_no_follow(parent, target.name) finally: os.close(parent) except FileNotFoundError as exc: diff --git a/src/nexus_symbiote/utils/jsonc.py b/src/nexus_symbiote/utils/jsonc.py index 87a904c..d2792d5 100644 --- a/src/nexus_symbiote/utils/jsonc.py +++ b/src/nexus_symbiote/utils/jsonc.py @@ -8,7 +8,7 @@ import re -_JSONC_EVENT = re.compile(r'"|//|/\*|[,\]}]') +_JSONC_EVENT = re.compile(r'"|//|/\*|[,\]}\[{:]') _JSON_STRING = re.compile(r'"(?:[^"\\]|\\.)*"', re.DOTALL) @@ -24,10 +24,12 @@ def strip_jsonc(text: str) -> str: parts: list[str] = [] position = 0 comma = -1 + can_trail = False while (event := _JSONC_EVENT.search(text, position)) is not None: start = event.start() if text[position:start].strip(): comma = -1 + can_trail = True parts.append(text[position:start]) mark = event.group() if mark == '"': @@ -35,7 +37,7 @@ def strip_jsonc(text: str) -> str: if string is None: raise ValueError("unterminated string") parts.append(string.group()) - comma, position = -1, string.end() + comma, can_trail, position = -1, True, string.end() elif mark == "//": newline = text.find("\n", start) position = len(text) if newline < 0 else newline @@ -47,13 +49,18 @@ def strip_jsonc(text: str) -> str: raise ValueError("unterminated comment") parts.append(" ") position = close + 2 + elif mark in "[{:": + parts.append(mark) + comma, can_trail, position = -1, False, start + 1 elif mark == ",": parts.append(",") - comma, position = len(parts) - 1, start + 1 + comma = len(parts) - 1 if can_trail else -1 + can_trail = False + position = start + 1 else: if comma >= 0: parts[comma] = "" parts.append(mark) - comma, position = -1, start + 1 + comma, can_trail, position = -1, True, start + 1 parts.append(text[position:]) return "".join(parts) diff --git a/src/nexus_symbiote/utils/model_formats.py b/src/nexus_symbiote/utils/model_formats.py index c37d97d..1ddb838 100644 --- a/src/nexus_symbiote/utils/model_formats.py +++ b/src/nexus_symbiote/utils/model_formats.py @@ -452,9 +452,11 @@ def gguf_metadata(read: Read, budget: InspectionBudget | None = None) -> GGUFMet # Jinja sandbox escapes reach Python internals through dunder attributes: by name, # by a string that a subscript, the attr filter or an attribute argument looks up, # or through the attr filter. Ordinary chat templates do none of these. Jinja skips -# Unicode whitespace between tokens, so the pattern matches text, not bytes. +# Unicode whitespace between tokens, so the pattern matches text, not bytes. A dunder +# name joined from two string literals is matched by either half. _TEMPLATE_ESCAPE = re.compile( r"\.\s*__[A-Za-z0-9_]+__|['\"]__[A-Za-z0-9_]+__['\"]|(? str: return "".join(parts) +def _raw_auth_replacement(match: re.Match[str]) -> str: + prefix, token = match.groups() + if token.lower() in _AUTH_SCHEMES or not _credential_like(token, "token"): + return match.group() + return prefix + REDACTED + + +def _cookie_pair_replacement(match: re.Match[str]) -> str: + name_eq, val = match.groups() + if name_eq[:-1].lower() in _COOKIE_ATTRS: + return match.group() + if len(val) >= MIN_SECRET_LENGTH or _credential_like(val, "token"): + return name_eq + REDACTED + return match.group() + + +def _cookie_header_replacement(match: re.Match[str]) -> str: + prefix, raw_cookies = match.groups() + return prefix + _COOKIE_PAIR.sub(_cookie_pair_replacement, raw_cookies) + + def redact_text( value: str, secrets: Iterable[str] | SecretScrubber = (), @@ -214,7 +256,11 @@ def redact_text( value = _redact_private_keys(value) value = _ASSIGNMENT.sub(lambda match: match.group(1) + REDACTED, value) value = _AUTH.sub(_auth_replacement, value) - return (_NAMED_TOKEN if identifier else TOKEN_SHAPE).sub(REDACTED, value) + if identifier: + return _NAMED_TOKEN.sub(REDACTED, value) + value = _RAW_AUTH_HEADER.sub(_raw_auth_replacement, value) + value = _COOKIE_HEADER.sub(_cookie_header_replacement, value) + return TOKEN_SHAPE.sub(REDACTED, value) def redact_metadata(value: Any, secrets: Iterable[str] | SecretScrubber = ()) -> Any: diff --git a/tests/test_audit_pr2_confinement_scanners.py b/tests/test_audit_pr2_confinement_scanners.py new file mode 100644 index 0000000..e5466a4 --- /dev/null +++ b/tests/test_audit_pr2_confinement_scanners.py @@ -0,0 +1,255 @@ +"""Regression tests for PR 2 audit fixes: confinement, MCP launchers/resolvers, SSE, secrets, JSONC, and templates.""" + +from __future__ import annotations + +import errno +import json +import os +from pathlib import Path + +import pytest + +from nexus_symbiote.core.models import MCPServerComponent +from nexus_symbiote.policy import PolicyError, find_policy_path +from nexus_symbiote.scanners import live_mcp as live +from nexus_symbiote.scanners.mcp import MCPScanner +from nexus_symbiote.utils.confined import ConfinedCollection +from nexus_symbiote.utils.jsonc import strip_jsonc +from nexus_symbiote.utils.model_formats import template_escape_markers +from nexus_symbiote.utils.secrets import REDACTED, redact_text, secret_like + + +def _scan_server( + root: Path, + server_config: dict[str, object], + files: dict[str, str] | None = None, +) -> tuple[MCPServerComponent, list[str]]: + if files: + for rel, content in files.items(): + target = root / rel + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(content, encoding="utf-8") + (root / ".mcp.json").write_text( + json.dumps({"mcpServers": {"srv": server_config}}), + encoding="utf-8", + ) + result = MCPScanner().scan(root, config_only=True) + server = next(c for c in result.components if isinstance(c, MCPServerComponent)) + return server, result.surfaces_excluded + + +def test_confined_exact_name_blocks_case_variant_open(tmp_path: Path) -> None: + (tmp_path / ".MCP.json").write_text('{"mcpServers": {}}', encoding="utf-8") + failures: list[str] = [] + with ConfinedCollection(tmp_path, failures) as collection: + assert collection.read_text(".mcp.json", missing_ok=True) is None + result = MCPScanner().scan(tmp_path, config_only=True) + assert any("case variant" in msg for msg in (*result.scan_failures, *result.surfaces_excluded)) + + +@pytest.mark.parametrize("relative", ["notes.txt", "nested/notes.txt"]) +def test_a_failed_exact_name_check_closes_what_it_opened( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, relative: str +) -> None: + target = tmp_path / relative + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text("text", encoding="utf-8") + opened: list[int] = [] + real_open = os.open + + def recording_open(*args, **kwargs): + fd = real_open(*args, **kwargs) + opened.append(fd) + return fd + + def unlistable(self: ConfinedCollection, directory_fd: int, name: str) -> bool: + raise PermissionError(errno.EACCES, "the directory cannot be listed") + + with ConfinedCollection(tmp_path, []) as collection: + monkeypatch.setattr(os, "open", recording_open) + monkeypatch.setattr(ConfinedCollection, "_has_exact_name", unlistable) + assert collection.hash_file(relative) is None + monkeypatch.undo() + assert opened + for fd in opened: + with pytest.raises(OSError): + os.fstat(fd) + + +def test_confined_default_excludes_and_policy_excludes_match_case_folded(tmp_path: Path) -> None: + git_variant = tmp_path / ".GIT" + git_variant.mkdir() + (git_variant / "config").write_text("secret", encoding="utf-8") + failures: list[str] = [] + with ConfinedCollection(tmp_path, failures, exclude_paths=["BuildOut"]) as collection: + assert collection.excluded(".GIT/config") + if (tmp_path / ".git").exists(): + assert collection.excluded("buildout/artifact.json") + + +def test_find_policy_path_rejects_case_variant_policy_filename(tmp_path: Path) -> None: + (tmp_path / ".Nexus-Symbiote.yaml").write_text("fail_on_severity: critical\n", encoding="utf-8") + with pytest.raises(PolicyError, match="case-variant"): + find_policy_path(tmp_path) + + +@pytest.mark.parametrize( + "config", + [ + {"command": "npx", "args": ["-y", "-p", "@acme/mcp@1.0.0", "./evil.sh"]}, + {"command": "npx", "args": ["-y", "-p", "@acme/mcp@1.0.0", "node", "evil.js"]}, + {"command": "uvx", "args": ["--from", "mcp==1.0.0", "python", "local_script.py"]}, + {"command": "pipx", "args": ["run", "--spec", "mcp==1.0.0", "./run.py"]}, + {"command": "npm", "args": ["exec", "@acme/mcp@1.0.0", "--package=evil@latest"]}, + ], +) +def test_package_runners_refuse_local_scripts_interpreters_and_post_positional_flags( + tmp_path: Path, config: dict[str, object] +) -> None: + case_dir = tmp_path / str(abs(hash(json.dumps(config, sort_keys=True)))) + case_dir.mkdir(parents=True, exist_ok=True) + server, gaps = _scan_server(case_dir, config) + assert server.version_pinned is False + assert server.package_url is None + assert any( + r.category == "launcher_identity" and r.severity.value == "high" for r in server.risks + ) + assert any("launcher identity unknown" in gap for gap in gaps) + + +@pytest.mark.parametrize( + "mount_arg,expected_key,expected_severity", + [ + ("./local_dir:/app", "host_mounts", None), + ("../parent:/app", "broad_host_mount", "medium"), + ("/var/run/docker.sock/:/var/run/docker.sock", "runtime_socket_mount", "high"), + ("/etc/..:/host", "host_root_mount", "high"), + ], +) +def test_container_volume_mount_normalizes_relative_socket_and_dotdot_paths( + tmp_path: Path, + mount_arg: str, + expected_key: str, + expected_severity: str | None, +) -> None: + case_dir = tmp_path / expected_key + case_dir.mkdir(parents=True, exist_ok=True) + server, _ = _scan_server( + case_dir, + {"command": "docker", "args": ["run", "-v", mount_arg, "img@sha256:" + "a" * 64]}, + ) + assert server.properties["launch_context"].get(expected_key) == 1 + if expected_severity is not None: + assert any( + r.category == "launch_context" and r.severity.value == expected_severity + for r in server.risks + ) + + +def test_bunfig_preload_marks_launcher_identity_unknown(tmp_path: Path) -> None: + server, gaps = _scan_server( + tmp_path, + {"command": "bunx", "args": ["@acme/mcp@1.0.0"]}, + {"bunfig.toml": 'preload = ["./hook.ts"]\n'}, + ) + assert server.version_pinned is False + assert server.package_url is None + assert any( + r.category == "launcher_identity" and "bunfig.toml preload" in r.detail + for r in server.risks + ) + assert any("launcher identity unknown" in gap for gap in gaps) + + +def test_uvx_reads_project_uv_toml_redirect(tmp_path: Path) -> None: + server, _ = _scan_server( + tmp_path, + {"command": "uvx", "args": ["mcp-server==1.2.3"]}, + {"uv.toml": 'index-url = "https://pypi.attacker.test/simple"\n'}, + ) + assert server.launcher_package is None + assert server.package_url is None + assert "uv.toml:index-url=https://pypi.attacker.test" in server.properties["package_sources"] + + +def test_deno_reads_deno_json_import_redirects(tmp_path: Path) -> None: + server, _ = _scan_server( + tmp_path, + {"command": "deno", "args": ["run", "npm:@acme/mcp@1.0.0"]}, + {"deno.json": json.dumps({"imports": {"@acme/mcp": "npm:evil@9.9.9"}})}, + ) + assert server.launcher_package is None + assert server.package_url is None + assert any(s.startswith("deno.json:imports=") for s in server.properties["package_sources"]) + + +def test_cr_delimited_sse_completes_without_waiting_for_lf_or_eof() -> None: + class KeepAliveCRStream: + """Delivers one CR-delimited JSON-RPC SSE event, then blocks if polled again.""" + + def __init__(self, first_chunk: bytes) -> None: + self._chunks = [first_chunk] + self.reads = 0 + + def read1(self, size: int = -1) -> bytes: + self.reads += 1 + if not self._chunks: + raise AssertionError("stream polled past completed CR-delimited SSE event") + chunk = self._chunks.pop(0) + return chunk[:size] if size >= 0 else chunk + + event = b'event: message\rdata: {"jsonrpc":"2.0","id":7,"result":{"tools":[]}}\r\r' + stream = KeepAliveCRStream(event) + transport = live.PinnedHTTPTransport() + raw = transport._read_sse( # type: ignore[arg-type] + stream, + b'{"jsonrpc":"2.0","id":7,"method":"tools/list"}', + maximum=4096, + ) + assert raw == event + assert stream.reads == 1 + + +def test_secrets_redact_hf_tokens_raw_authorization_and_cookies() -> None: + hf_tok = "hf_" + "A1b2C3d4E5" * 4 + assert secret_like("UNRELATED", hf_tok) + assert redact_text(f"token is {hf_tok}") == f"token is {REDACTED}" + assert redact_text("Authorization: rawSecretToken_12345") == f"Authorization: {REDACTED}" + assert redact_text("Authorization: required") == "Authorization: required" + assert ( + redact_text("Set-Cookie: sid=mySecretSession99; Path=/; HttpOnly") + == f"Set-Cookie: sid={REDACTED}; Path=/; HttpOnly" + ) + assert redact_text("Cookie: policy") == "Cookie: policy" + + +@pytest.mark.parametrize( + "payload", + [ + b"{{ x['\\x5f\\x5fclass\\x5f\\x5f'] }}", + b"{{ x['\\u005f\\u005fglobals\\u005f\\u005f'] }}", + b"{{ x['\\137\\137subclasses\\137\\137'] }}", + b"{% set k = '__' ~ 'class__' %}{{ x[k] }}", + b"{% set k = '__cla' + 'ss__' %}{{ x[k] }}", + ], +) +def test_template_escape_markers_detect_escaped_and_concatenated_dunders(payload: bytes) -> None: + assert template_escape_markers(payload) is True + + +def test_template_escape_markers_preserve_benign_chat_templates() -> None: + benign = b"{% for m in messages %}{{ '<|im_start|>' + m['role'] + '\\n' + m['content'] }}{% endfor %}" + assert template_escape_markers(benign) is False + + +@pytest.mark.parametrize("malformed", ["[,]", "{,}", "[1,,]", "[1,,2]", '{"a": ,}', ","]) +def test_strip_jsonc_does_not_turn_leading_or_double_commas_into_valid_json( + malformed: str, +) -> None: + with pytest.raises(json.JSONDecodeError): + json.loads(strip_jsonc(malformed)) + + +def test_strip_jsonc_accepts_nested_valid_trailing_commas() -> None: + parsed = json.loads(strip_jsonc('{"items": [1, 2, /* ok */ ], "nested": {"k": "v",},}')) + assert parsed == {"items": [1, 2], "nested": {"k": "v"}} diff --git a/tests/test_confined_links_and_variants.py b/tests/test_confined_links_and_variants.py index a0945d1..d143254 100644 --- a/tests/test_confined_links_and_variants.py +++ b/tests/test_confined_links_and_variants.py @@ -125,7 +125,11 @@ def test_missing_directory_reports_its_case_variants(tmp_path): def test_exact_name_is_read_without_a_variant_report(tmp_path): (tmp_path / "CLAUDE.md").write_text("upper") - (tmp_path / "claude.md").write_text("lower") + lower = tmp_path / "claude.md" + if not lower.exists(): + lower.write_text("lower") + else: + (tmp_path / "CLAUDE.md.").write_text("lower") failures: list[str] = [] with ConfinedCollection(tmp_path, failures) as files: assert files.read_text("CLAUDE.md", missing_ok=True) == "upper" @@ -291,12 +295,12 @@ def _same_folding_names(count: int, base: str = "a") -> list[str]: """Names that fold to ``base`` (case, trailing dots and spaces), without ``base``.""" names: list[str] = [] length = 0 + upper = base.upper() while len(names) < count: for suffix in itertools.product(". ", repeat=length): - for first in (base, base.upper()): - name = first + "".join(suffix) - if name != base and len(names) < count: - names.append(name) + name = upper + "".join(suffix) + if name != base and len(names) < count: + names.append(name) length += 1 return names @@ -435,7 +439,10 @@ def test_a_link_to_a_name_that_is_not_utf8_is_refused_wherever_it_is_met(tmp_pat def test_a_case_variant_is_reported_once_however_many_paths_lead_to_it(tmp_path): (tmp_path / "t").mkdir() (tmp_path / "t" / "Settings.json").write_text("{}") - (tmp_path / "t" / "SETTINGS.JSON").write_text("{}") + second = ( + "SETTINGS.JSON" if not (tmp_path / "t" / "SETTINGS.JSON").exists() else "Settings.json." + ) + (tmp_path / "t" / second).write_text("{}") for index in range(3): (tmp_path / f"p{index}").mkdir() (tmp_path / f"p{index}" / ".claude").symlink_to("../t", target_is_directory=True) @@ -444,15 +451,19 @@ def test_a_case_variant_is_reported_once_however_many_paths_lead_to_it(tmp_path) for index in range(3): path = f"p{index}/.claude/settings.json" assert files.read_bytes(path, missing_ok=True) is None - assert sorted(files.case_variants(path)) == [ - Path(f"p{index}/.claude/SETTINGS.JSON"), - Path(f"p{index}/.claude/Settings.json"), - ] + assert sorted(files.case_variants(path)) == sorted( + [ + Path(f"p{index}/.claude/{second}"), + Path(f"p{index}/.claude/Settings.json"), + ] + ) files.report_case_variant("t/Settings.json", "t/settings.json") files.report_case_variant("q/.claude/Settings.json", "q/.claude/settings.json") variant = "is the active file on case-insensitive file systems and is not inventoried" - assert sorted(failures) == [ - f"p0/.claude/SETTINGS.JSON: case variant of p0/.claude/settings.json {variant}", - f"p0/.claude/Settings.json: case variant of p0/.claude/settings.json {variant}", - f"q/.claude/Settings.json: case variant of q/.claude/settings.json {variant}", - ] + assert sorted(failures) == sorted( + [ + f"p0/.claude/{second}: case variant of p0/.claude/settings.json {variant}", + f"p0/.claude/Settings.json: case variant of p0/.claude/settings.json {variant}", + f"q/.claude/Settings.json: case variant of q/.claude/settings.json {variant}", + ] + ) diff --git a/tests/test_prompt_imports.py b/tests/test_prompt_imports.py index 4b9243c..fac5393 100644 --- a/tests/test_prompt_imports.py +++ b/tests/test_prompt_imports.py @@ -3,6 +3,7 @@ from __future__ import annotations +import contextlib import os import string import subprocess @@ -31,8 +32,27 @@ def _write(root: Path, relative: str, text: str) -> None: path = root / relative - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(text, encoding="utf-8") + if len(os.fspath(path)) < 900: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + return + # macOS kernels enforce PATH_MAX = 1024 on single pathname syscalls, whereas + # ConfinedCollection traverses via dir_fd without a total-pathname cap. + root.mkdir(parents=True, exist_ok=True) + parts = [part for part in relative.split("/") if part and part != "."] + dir_fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY) + try: + for part in parts[:-1]: + with contextlib.suppress(FileExistsError): + os.mkdir(part, dir_fd=dir_fd) + next_fd = os.open(part, os.O_RDONLY | os.O_DIRECTORY, dir_fd=dir_fd) + os.close(dir_fd) + dir_fd = next_fd + file_fd = os.open(parts[-1], os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600, dir_fd=dir_fd) + with os.fdopen(file_fd, "w", encoding="utf-8") as stream: + stream.write(text) + finally: + os.close(dir_fd) def _components(result): diff --git a/tests/test_prompt_surfaces.py b/tests/test_prompt_surfaces.py index 2f32bab..3aca0a2 100644 --- a/tests/test_prompt_surfaces.py +++ b/tests/test_prompt_surfaces.py @@ -66,12 +66,22 @@ def test_case_variant_of_a_fixed_path_is_a_failure_naming_both_paths(tmp_path, v def test_both_codex_spellings_are_read_without_a_case_variant_failure(tmp_path): - _write(tmp_path, "codex.md", "lower") - assert not PromptScanner().scan(tmp_path).scan_failures - _write(tmp_path, "CODEX.md", "upper") - result = PromptScanner().scan(tmp_path) - assert not result.scan_failures - assert {"prompt:CODEX.md", "prompt:codex.md"} <= set(_components(result)) + lower_root = tmp_path / "lower" + _write(lower_root, "codex.md", "lower") + lower_result = PromptScanner().scan(lower_root) + assert not lower_result.scan_failures + assert "prompt:codex.md" in _components(lower_result) + if (lower_root / "CODEX.md").exists(): + upper_root = tmp_path / "upper" + _write(upper_root, "CODEX.md", "upper") + upper_result = PromptScanner().scan(upper_root) + assert not upper_result.scan_failures + assert "prompt:CODEX.md" in _components(upper_result) + else: + _write(lower_root, "CODEX.md", "upper") + result = PromptScanner().scan(lower_root) + assert not result.scan_failures + assert {"prompt:CODEX.md", "prompt:codex.md"} <= set(_components(result)) def test_in_root_symlinked_instruction_file_is_read_through_its_target(tmp_path): diff --git a/tests/test_release_due_diligence.py b/tests/test_release_due_diligence.py index db1ebde..30bdae9 100644 --- a/tests/test_release_due_diligence.py +++ b/tests/test_release_due_diligence.py @@ -417,6 +417,8 @@ def test_the_exact_requirements_txt_is_kept_ahead_of_its_case_variants(tmp_path) # On a case-sensitive system pip and main read requirements.txt itself, so its # declaration is the component; a variant is recorded beside it. (tmp_path / "Requirements.txt").write_text("torch>=2\n") + if (tmp_path / "requirements.txt").exists(): + pytest.skip("file system rejects distinct case spellings") (tmp_path / "requirements.txt").write_text("torch==2.3.0\n") try: (tmp_path / "requirement\u017f.txt").write_text("openai==1.30.1\n")