diff --git a/src/kiro_crew/dashboard/handlers/agents.py b/src/kiro_crew/dashboard/handlers/agents.py index c42310e7ecc..71d9a8f711c 100644 --- a/src/kiro_crew/dashboard/handlers/agents.py +++ b/src/kiro_crew/dashboard/handlers/agents.py @@ -2033,7 +2033,11 @@ async def api_models(request: web.Request) -> web.Response: try: env = {**os.environ} env["PATH"] = augmented_path(env.get("PATH", "")) - _resolve_ssh_auth_sock(env) + # OFF the loop: the resolver globs /tmp/ssh-*/agent.* and stats + # every hit, so its latency scales with the /tmp entry count. Its + # sibling wrapper's contract states it must never run on the event + # loop, and this endpoint is polled every 8s while degraded. + await asyncio.to_thread(_resolve_ssh_auth_sock, env) # The Docker entrypoint removes credentials from the long-lived # gateway environment. This fixed-argv child is the official # kiro-cli and KIRO_API_KEY is its own model credential, so settle diff --git a/src/kiro_crew/dashboard/handlers/terminal.py b/src/kiro_crew/dashboard/handlers/terminal.py index 7e29117db2c..1c87b62950a 100644 --- a/src/kiro_crew/dashboard/handlers/terminal.py +++ b/src/kiro_crew/dashboard/handlers/terminal.py @@ -24,6 +24,7 @@ from kiro_crew.dashboard.origin import check_origin, mark_audit_claimed from kiro_crew.executors import discovery_executor, subprocess_executor from kiro_crew.hooks import validate_file_path +from kiro_crew.sandbox import _PYTHON_ENV_PREFIXES from kiro_crew.security import ( is_sensitive_path, redact_credentials, @@ -468,6 +469,28 @@ def _is_bash_shell(shell: str) -> bool: _READY_PREV_VAR = "KIROCREW_TERMINAL_READY_PREV" +def _pty_child_env(extra: dict[str, str]) -> dict[str, str]: + """Build the interactive shell's environment: the gateway environment plus + *extra*, minus Kiro Crew's own Python startup variables. + + ``PYTHONPATH``/``PYTHONHOME``/``PYTHONPYCACHEPREFIX`` are searched BEFORE a + venv's own site-packages, so leaking the gateway's copies makes a user's + Python 3.13 venv import Kiro Crew's 3.12 site-packages and its C extensions + fail to load. The agent surface strips them via + ``sandbox.scrub_agent_subprocess_env``; this brings the terminal into line. + + Only the Python prefixes are dropped. Unlike the agent spawn, this is the + user's own unsandboxed shell (see the spawn comment below), so + ``SSH_AUTH_SOCK``, the AWS vars and the rest of the credential-bearing + environment must survive or git-over-SSH and the AWS CLI break in it. + """ + env = {**os.environ, **extra} + for key in list(env): + if any(key.startswith(prefix) for prefix in _PYTHON_ENV_PREFIXES): + del env[key] + return env + + def _bash_ready_env(token: str) -> dict[str, str]: """Environment that makes a real login Bash report readiness at its prompt. @@ -860,7 +883,7 @@ async def api_terminal_ws(request: web.Request) -> web.WebSocketResponse | web.R cwd = _resolve_cwd(cfg, request.query.get("cwd")) if not os.path.isdir(cwd): cwd = os.path.expanduser("~") - env = {**os.environ, "KIROCREW_TERMINAL": "1"} + env = _pty_child_env({"KIROCREW_TERMINAL": "1"}) argv = [shell, "-NoLogo"] if "powershell" in shell.lower() else [shell] try: wp = WindowsPty(argv, cwd=cwd, env=env, cols=80, rows=24) @@ -907,8 +930,7 @@ async def api_terminal_ws(request: web.Request) -> web.WebSocketResponse | web.R struct.pack("HHHH", 24, 80, 0, 0), ) cwd = _resolve_cwd(cfg, request.query.get("cwd")) - env = { - **os.environ, + env = _pty_child_env({ "TERM": "xterm-256color", "KIROCREW_TERMINAL": "1", # Export the shell actually being spawned (already resolved to @@ -918,7 +940,7 @@ async def api_terminal_ws(request: web.Request) -> web.WebSocketResponse | web.R # (vim's :sh, tmux default-shell) open the wrong one. POSIX # branch only: PowerShell does not consult $SHELL. "SHELL": shell, - } + }) # Security: intentionally unsandboxed — this is the user's own # interactive terminal (like SSH), not agent-executed code. # Auth is enforced at WS handshake via token_auth_middleware. diff --git a/src/kiro_crew/llm_helpers.py b/src/kiro_crew/llm_helpers.py index 4ac5dcd663c..1e9b6f0753f 100644 --- a/src/kiro_crew/llm_helpers.py +++ b/src/kiro_crew/llm_helpers.py @@ -886,6 +886,84 @@ def _collect(obj: object) -> None: return results +# Longest single string the tool_input scan will attempt. +# +# CPython's ``re`` does not release the GIL for the duration of one match call, +# so the worker-thread hop below yields BETWEEN per-string scans but not WITHIN +# one: a single huge string holds the GIL inside one ``.search`` and the event +# loop cannot run its watchdog heartbeat for that whole time. +# +# The anchor rewrite in this change cut the constant but NOT the growth -- the +# scan is still superlinear in the length of one line. Measured on one dev box: +# +# 4 KiB 0.16s | 8 KiB 0.31s | 16 KiB 1.14s | 20 KiB 1.52s | 32 KiB 3.79s +# +# 16->32 KiB is 3.3x for 2x the input (~n^1.7), so extrapolation is the wrong +# instinct here and a ceiling has to be set from the measured curve. A CI runner +# under parallel load came in roughly an order of magnitude slower than this box, +# which is what sets the margin: 20 KiB is ~1.5s here and ~15s there, under the +# 25s loop watchdog on both. +# +# Exceeding it is FAIL-CLOSED: the call is denied, never skipped. A skip would +# convert a liveness bug into a security hole by letting unscanned input through +# the deny surface; a denial only refuses input we cannot prove safe, and is +# strictly better than the alternative it replaces, which was crashing the +# gateway and losing the whole turn. +# +# Known cost of that trade, and why the ceiling is an interim: a permission-gated +# write of a benign file larger than this lands its whole content in +# ``tool_input`` and is now refused. Chunked scanning cannot lift the ceiling on +# its own, because one branch of the pattern puts an unbounded ``.*`` between a +# verb and the path, so no chunk overlap preserves that distance. The durable fix +# is to stop running the SHELL-COMMAND matcher over fields that never carry a +# command. Tracked in https://github.com/kirodotdev/KiroCrew/issues/8053. +_MAX_SCANNABLE_TOOL_INPUT_CHARS = 20 * 1024 + + +def _first_tool_input_denial( + strings: list[str], + denied_regexes: list[str] | None, +) -> tuple[str, str, str] | None: + """Return the first tool_input denial among *strings*, or ``None``. + + Pure, synchronous, and blocking: the three predicates are regex-heavy and + ``_extract_tool_input_strings`` hands over EVERY string in the payload, so a + single long document body can occupy this loop for seconds. It therefore + runs on a worker thread (one hop for the whole loop, not one per string), + which keeps the event loop free BETWEEN strings -- not within one, because + ``re`` holds the GIL for a whole match call. That is why each string is + length-checked against :data:`_MAX_SCANNABLE_TOOL_INPUT_CHARS` first, and an + oversized one is denied rather than scanned or skipped. + + The tuple is ``(kind, reason, matched_string)`` where *kind* is + ``"path"`` / ``"bash"`` / ``"regex"`` / ``"oversize"``. Mechanism + classification stays with the caller on the event loop, because it consults + the HookManager. + """ + for s in strings: + if len(s) > _MAX_SCANNABLE_TOOL_INPUT_CHARS: + # Fail closed: too long to scan inside the loop's liveness budget, + # so it cannot be shown safe and is refused. + return ( + "oversize", + ( + "Blocked: a tool_input string is too large to security-scan " + f"({len(s)} chars > {_MAX_SCANNABLE_TOOL_INPUT_CHARS} limit); " + "refused rather than left unscanned" + ), + s[:64], + ) + if is_sensitive_path(s): + return ("path", f"Blocked: sensitive path in tool_input: {s}", s) + _input_bash = is_sensitive_bash_command(s) + if _input_bash: + return ("bash", _input_bash, s) + _input_deny = is_denied(s, denied_regexes=denied_regexes) + if _input_deny: + return ("regex", _input_deny, s) + return None + + # ── Tool Approval Policies ── @@ -2046,29 +2124,25 @@ def _regex_deny_mechanism(probe: str, unconditional: str) -> str: if _tool_input: # Extract string values from JSON tool_input for path/command checking. _input_strings = _extract_tool_input_strings(_tool_input) - for s in _input_strings: - if is_sensitive_path(s): - await provider.reject_tool(event.request_id) - _log( - "denied", - error=f"Blocked: sensitive path in tool_input: {s}", - metadata={"mechanism": "always_deny_input"}, - ) - return False - _input_bash = is_sensitive_bash_command(s) - if _input_bash: - await provider.reject_tool(event.request_id) - _log("denied", error=_input_bash, metadata={"mechanism": "always_deny_input"}) - return False - _input_deny = is_denied(s, denied_regexes=_denied_regexes) - if _input_deny: - await provider.reject_tool(event.request_id) - _log( - "denied", - error=_input_deny, - metadata={"mechanism": _regex_deny_mechanism(s, "always_deny_input")}, - ) - return False + # Offloaded: the scan is regex-heavy over every string in the payload, + # so a large document body would block the event loop past its watchdog + # and take the gateway down. One hop for the whole loop. + _hit = await asyncio.to_thread(_first_tool_input_denial, _input_strings, _denied_regexes) + if _hit is not None: + _kind, _reason, _matched = _hit + await provider.reject_tool(event.request_id) + _log( + "denied", + error=_reason, + metadata={ + "mechanism": ( + _regex_deny_mechanism(_matched, "always_deny_input") + if _kind == "regex" + else "always_deny_input" + ) + }, + ) + return False if policy == ToolApprovalPolicy.HOOK_BASED and hooks: tool_result = hooks.on_tool_call( diff --git a/src/kiro_crew/security.py b/src/kiro_crew/security.py index d932010b2bc..f23ea69ac7b 100644 --- a/src/kiro_crew/security.py +++ b/src/kiro_crew/security.py @@ -7439,18 +7439,26 @@ def _build_sensitive_regex() -> re.Pattern[str]: # colon lists, comma/semicolon-joined args) — without the latter a # ``FOO=bar:~/.aws/credentials`` or ``PATH=/x:~/.ssh/id_rsa`` token slips # past the backstop while no verb branch fires either. + # + # The anchor is written ``(?:^|[\s'\"=:,;])`` with NO leading ``.*``: this + # pattern is only ever used via ``.search`` (see ``_get_sensitive_re`` + # callers), which already retries at every offset, so a leading ``.*`` + # matched nothing extra while making the scan quadratic in the longest + # line. Note ``\n`` is in the class, so a path at the start of a later + # line still matches even though ``.`` never crossed a newline anyway. + # Do NOT reintroduce ``.*`` here. # (3) write-protected leaf: matched verb-INDEPENDENTLY too (same token # anchor), so a quoted redirect (``> "$HOME/.../marker"``), ``cp``, # ``python -c "open(...,'w')"`` or any novel write verb is still caught. rf"(?:(?:{_READ_CMDS}.*|{_WRITE_CMDS}.*|{_SCRIPT_OPEN}.*|.*[<>|]\s*)" rf"{sensitive_path}" - rf"|(?:^|.*[\s'\"=:,;]){sensitive_path}" - rf"|(?:^|.*[\s'\"=:,;]){write_protected_path}" + rf"|(?:^|[\s'\"=:,;]){sensitive_path}" + rf"|(?:^|[\s'\"=:,;]){write_protected_path}" # (3b) publish artifacts of a keystone leaf -- the atomic-write temp and the lock # sibling -- in both the POSIX and the Windows-native spelling. Verb-independent # like (2)/(3): naming the artifact is the signal, so a redirect, a ``cp``, or an # embedded ``open(...,'w')`` is caught without enumerating write verbs. - rf"|(?:^|.*[\s'\"=:,;]){artifact_path}" + rf"|(?:^|[\s'\"=:,;]){artifact_path}" # (4) Windows-native spelling, verb-independent (same token anchor): # covers quoted backslash paths AND embedded-script literals that the # tokenizing passes cannot see. (5) the %APPDATA% / %LOCALAPPDATA% @@ -7459,12 +7467,12 @@ def _build_sensitive_regex() -> re.Pattern[str]: # (7) the distinctive leaves as a bare path SEGMENT, with no anchor at # all, because branches (3) and (6) both fall to a ``cd`` plus a # relative name. - rf"|(?:^|.*[\s'\"=:,;]){win_sensitive_path}" - rf"|(?:^|.*[\s'\"=:,;]){win_artifact_path}" - rf"|(?:^|.*[\s'\"=:,;]){appdata_sensitive_path}" - rf"|(?:^|.*[\s'\"=:,;]){localappdata_sensitive_path}" - rf"|(?:^|.*[\s'\"=:,;]){win_write_protected_path}" - rf"|(?:^|.*[\s'\"=:,;]){win_crew_var_leaf_path}" + rf"|(?:^|[\s'\"=:,;]){win_sensitive_path}" + rf"|(?:^|[\s'\"=:,;]){win_artifact_path}" + rf"|(?:^|[\s'\"=:,;]){appdata_sensitive_path}" + rf"|(?:^|[\s'\"=:,;]){localappdata_sensitive_path}" + rf"|(?:^|[\s'\"=:,;]){win_write_protected_path}" + rf"|(?:^|[\s'\"=:,;]){win_crew_var_leaf_path}" # (8) ~/.kiro/agents (POSIX and Windows-native spelling, plus the # ``$KIRO_HOME`` override), matched verb-INDEPENDENTLY with the same token # anchor as (2)/(3): naming the dir is the signal, so ``curl -o``/``wget @@ -7472,8 +7480,8 @@ def _build_sensitive_regex() -> re.Pattern[str]: # write verb are caught, not just an enumerated allowlist. Bash reads of # the dir are blocked incidentally (harmless — no secret, Python readers # only); tool-path reads stay allowed. - rf"|(?:^|.*[\s'\"=:,;]){agents_write_path}" - rf"|(?:^|.*[\s'\"=:,;]){win_agents_write_path}" + rf"|(?:^|[\s'\"=:,;]){agents_write_path}" + rf"|(?:^|[\s'\"=:,;]){win_agents_write_path}" # (10) whisper weight FILENAMES, also with no anchor, because the digest the # model store checks only binds the bytes if the name it then loads cannot be # rewritten by a ``cd``-relative command. @@ -12562,19 +12570,29 @@ def redact_credentials(text: str) -> tuple[str, list[str]]: # This is the hot path — the alternation is 23 branches retried at nearly # every position, and real text almost never contains a credential. if _might_contain_credential(result): - for m in _CREDENTIAL_PATTERNS.finditer(result): - matched = m.group() - tag = _REDACTED_CREDENTIAL_TAG - result = result.replace(matched, tag, 1) + + def _redact_one(m: re.Match[str]) -> str: # Emit ONLY non-sensitive metadata (length). Do NOT slice any part of - # `matched` into the warning: `_CREDENTIAL_PATTERNS` matches the raw + # the match into the warning: `_CREDENTIAL_PATTERNS` matches the raw # secret value itself (e.g. `ghp_…`, `sk-ant-…`), so even a short prefix # is genuine plaintext key material — a fixed-length token prefix leaves # ~12-16 secret chars in a 20-char slice. The warnings list is a # redaction-subsystem output expected to be safe to log/surface, so it # must carry no secret bytes. Mirrors the base64 / bare-secret branches # below, which already log length only. - warnings.append(f"Redacted credential pattern ({len(matched)} chars)") + warnings.append(f"Redacted credential pattern ({len(m.group())} chars)") + return _REDACTED_CREDENTIAL_TAG + + # ONE pass. `sub` walks the matches left-to-right exactly as `finditer` + # did and calls the replacer in that same order, so `warnings` is + # appended in an identical order with identical contents. The previous + # shape rebuilt the entire string per match via + # `result.replace(matched, tag, 1)` — O(n) per match, O(n²) overall on + # credential-dense text — and replaced the FIRST occurrence of the + # matched text rather than the span that actually matched. `sub` splices + # each matched span in place, which is both linear and positionally + # exact. + result = _CREDENTIAL_PATTERNS.sub(_redact_one, result) # Passes 2 and 3 both scan the ORIGINAL `text` for runs of the base64 # alphabet, and they select the SAME spans: `[A-Za-z0-9+/]{40,}` is greedy and diff --git a/test/test_api_models_retry.py b/test/test_api_models_retry.py index ae0ccdf32f5..24c415c5a2a 100644 --- a/test/test_api_models_retry.py +++ b/test/test_api_models_retry.py @@ -16,6 +16,7 @@ import asyncio import json import logging +import threading from pathlib import Path from types import SimpleNamespace from typing import Any @@ -430,3 +431,45 @@ def _refuse(argv, **kwargs): } # The remedy reaches the operator rather than a bare traceback. assert any("not Linux" in r.getMessage() for r in caplog.records), caplog.text + + +def test_ssh_auth_sock_resolver_runs_off_the_event_loop(tmp_path): + """``_resolve_ssh_auth_sock`` globs ``/tmp/ssh-*/agent.*`` and ``os.stat``s + every hit, so its latency scales with the ``/tmp`` entry count. The frontend + polls this endpoint every 8s while the model list is degraded, so an on-loop + call stalls chat, cron and the liveness heartbeat on exactly the host where + the probe is slowest. Its sibling wrapper's docstring states the rule: never + on the event loop. Same shape as + ``test_acp_spawn_offload.py``'s resolver assertions. + """ + payload = json.dumps({"models": [{"model_name": "claude-opus-4.8"}]}).encode() + seen: list[threading.Thread] = [] + + def _probe(env): + seen.append(threading.current_thread()) + return None + + async def _drive(): + loop_thread = threading.current_thread() + resp = await agents.api_models(_kiro_request(tmp_path)) + return loop_thread, resp + + with patch.object(agents.KiroCrewConfig, "load", return_value=_kiro_cfg()), patch( + "kiro_crew.acp.client._resolve_kiro_bin_for_spawn", return_value="/usr/bin/kiro-cli" + ), patch("kiro_crew.acp.client._resolve_ssh_auth_sock", _probe), patch( + "kiro_crew.env.augmented_path", lambda p: p + ), patch( + "kiro_crew.dashboard.handlers.agents.wrap_argv", _stub_wrap_argv + ), patch( + "kiro_crew.dashboard.handlers.agents.cgroup_scope_argv", lambda argv: argv + ), patch( + "kiro_crew.sandbox.resource_limit_preexec", lambda: None + ), patch.object( + agents.asyncio, "create_subprocess_exec", return_value=_FakeProc(payload) + ): + loop_thread, resp = _run(_drive()) + + assert resp.status == 200 + assert seen, "_resolve_ssh_auth_sock was never called by api_models" + for thread in seen: + assert thread is not loop_thread, "ssh resolver ran on the event loop thread" diff --git a/test/test_credential_prefilter.py b/test/test_credential_prefilter.py index bb7ae39e296..642e01ff5c5 100644 --- a/test/test_credential_prefilter.py +++ b/test/test_credential_prefilter.py @@ -44,15 +44,32 @@ def _reference_redact_credentials(text: str) -> tuple[str, list[str]]: - """The original three-pass body, verbatim. Do not "optimise" this.""" + """The original three-pass body. Do not "optimise" this. + + Pass 1 is the ONE deliberate divergence from the pre-optimisation source. + The original wrote ``result.replace(matched, tag, 1)``, which replaces the + first occurrence of the matched *text* rather than the span the regex + actually matched. When an earlier, NON-matching lookalike contains the + matched text as a substring (``xM`` before a boundary-anchored + ``M``), the original redacted the innocent lookalike and left the real + credential in the output in plaintext -- a genuine leak, not a cosmetic + difference. See ``test_matched_span_is_redacted_not_an_earlier_lookalike``. + + Passes 2 and 3 keep the ``.replace(..., 1)`` shape because they scan the + ORIGINAL ``text`` and select spans by value, and because production keeps + them as separate passes deliberately (see the comment in + ``security.redact_credentials``); their latent equivalent is out of scope + here and is NOT fixed by this oracle. + """ warnings: list[str] = [] result = text - # 1. plaintext credential patterns — ungated full scan - for m in _CREDENTIAL_PATTERNS.finditer(result): - matched = m.group() - result = result.replace(matched, _REDACTED_CREDENTIAL_TAG, 1) - warnings.append(f"Redacted credential pattern ({len(matched)} chars)") + # 1. plaintext credential patterns — ungated full scan, positionally exact + def _redact_one(m: "re.Match[str]") -> str: + warnings.append(f"Redacted credential pattern ({len(m.group())} chars)") + return _REDACTED_CREDENTIAL_TAG + + result = _CREDENTIAL_PATTERNS.sub(_redact_one, result) # 2. base64-encoded credentials — own scan, decode via the generic helper for m in _B64_CHUNK_RE.finditer(text): @@ -330,6 +347,26 @@ def test_output_is_byte_identical_to_reference(text: str) -> None: assert redact_credentials(text) == _reference_redact_credentials(text) +def test_matched_span_is_redacted_not_an_earlier_lookalike() -> None: + """Pass 1 must redact the span that matched, not an earlier substring. + + Regression for the shape ``result.replace(matched, tag, 1)``. Here the + boundary-anchored pattern matches only the SECOND token; the first is an + ``x``-prefixed lookalike that happens to contain the matched text. The old + shape redacted the lookalike and emitted the real credential verbatim. + """ + token = "M" + "a" * 24 + ".abc123." + "b" * 27 + text = f"x{token} {token}" + assert _CREDENTIAL_PATTERNS.search(text), "corpus assumption: the pattern fires" + + redacted, warnings = redact_credentials(text) + + # The real credential -- the standalone second token -- must be gone. + assert f" {token}" not in redacted, "the matched credential survived redaction" + assert redacted == f"x{token} {_REDACTED_CREDENTIAL_TAG}" + assert warnings == [f"Redacted credential pattern ({len(token)} chars)"] + + def test_corpus_actually_exercises_every_pass() -> None: """A differential corpus that never triggers a pass proves nothing about it.""" kinds = {w.split("(")[0].strip() for text in CORPUS for w in redact_credentials(text)[1]} diff --git a/test/test_llm_helpers_tool_input_offload.py b/test/test_llm_helpers_tool_input_offload.py new file mode 100644 index 00000000000..4c1da91ac21 --- /dev/null +++ b/test/test_llm_helpers_tool_input_offload.py @@ -0,0 +1,325 @@ +"""The tool_input security scan must not run on the asyncio event loop. + +``_resolve_permission`` inspects EVERY string ``_extract_tool_input_strings`` +pulls out of the parsed tool input — including whole document bodies, which are +scanned as shell commands. The three predicates +(``is_sensitive_path`` / ``is_sensitive_bash_command`` / ``is_denied``) are +regex-heavy, so one long newline-free line held the loop past the 25s watchdog +and killed the gateway (Mesh-3693). The scan now happens in ONE +``asyncio.to_thread`` hop for the whole loop. + +These tests pin three contracts: + +1. Equivalence — the decision AND the reason/mechanism reaching the SEL row are + byte-identical to the pre-offload behaviour, for a denied path, a denied bash + command, a benign payload, a nested structure, and an empty input. +2. Off-loop — the predicates run on a thread that is not the loop's when they + are applied to tool_input strings (the title-tier checks above still run on + the loop, so calls are matched by their argument). +3. Liveness — a concurrent asyncio task keeps ticking while a 20 KB + newline-free non-shell document body is scanned. +""" + +from __future__ import annotations + +import asyncio +import json +import threading +import time +from unittest.mock import MagicMock, patch + +import pytest + +import kiro_crew.sel as sel_mod +from kiro_crew import llm_helpers +from kiro_crew.llm_helpers import ToolApprovalPolicy, _resolve_permission +from kiro_crew.providers.base import EVENT_PERMISSION_REQUEST, LLMEvent + +# A title that survives every title-tier check, so the decision under test is +# decided by tool_input alone. +_BENIGN_TITLE = "Read" + + +class _RecordingProvider: + def __init__(self) -> None: + self.approved: list[str] = [] + self.rejected: list[str] = [] + + async def approve_tool(self, request_id: str) -> None: + self.approved.append(request_id) + + async def reject_tool(self, request_id: str) -> None: + self.rejected.append(request_id) + + +def _event(tool_input: str, title: str = _BENIGN_TITLE) -> LLMEvent: + return LLMEvent( + kind=EVENT_PERMISSION_REQUEST, + title=title, + request_id="r1", + tool_input=tool_input, + ) + + +async def _resolve(tool_input: str) -> tuple[bool, _RecordingProvider, list[dict]]: + """Drive ``_resolve_permission`` and capture every SEL row it logged.""" + provider = _RecordingProvider() + rows: list[dict] = [] + sel_stub = MagicMock() + sel_stub.log_tool_invocation.side_effect = lambda **kw: rows.append(kw) + with patch.object(sel_mod, "sel", lambda: sel_stub): + approved = await _resolve_permission( + provider, # type: ignore[arg-type] + _event(tool_input), + ToolApprovalPolicy.AUTO_APPROVE, + None, + ) + return approved, provider, rows + + +def _decision(rows: list[dict]) -> tuple[str, str, str]: + """(outcome, error, mechanism) of the single decision row.""" + assert len(rows) == 1, rows + row = rows[0] + return ( + row.get("outcome", ""), + str(row.get("error") or ""), + str((row.get("metadata") or {}).get("mechanism") or ""), + ) + + +class TestEquivalence: + """The offloaded scan must decide exactly what the on-loop loop decided.""" + + @pytest.mark.asyncio + async def test_denied_sensitive_path_in_tool_input(self) -> None: + payload = json.dumps({"path": "~/.aws/credentials"}) + approved, provider, rows = await _resolve(payload) + assert approved is False + assert provider.rejected == ["r1"] + assert provider.approved == [] + outcome, error, mechanism = _decision(rows) + assert outcome == "denied" + assert error == "Blocked: sensitive path in tool_input: ~/.aws/credentials" + assert mechanism == "always_deny_input" + + @pytest.mark.asyncio + async def test_denied_bash_command_in_tool_input(self) -> None: + payload = json.dumps({"command": "rm -rf /"}) + approved, provider, rows = await _resolve(payload) + assert approved is False + assert provider.rejected == ["r1"] + outcome, error, mechanism = _decision(rows) + assert outcome == "denied" + assert error, "the bash reason text must still reach the SEL row" + assert mechanism == "always_deny_input" + + @pytest.mark.asyncio + async def test_benign_payload_is_approved(self) -> None: + payload = json.dumps({"path": "README.md"}) + approved, provider, rows = await _resolve(payload) + assert approved is True + assert provider.approved == ["r1"] + assert provider.rejected == [] + outcome, _error, _mechanism = _decision(rows) + assert outcome == "auto_approved" + + @pytest.mark.asyncio + async def test_nested_structure_is_still_scanned(self) -> None: + """The recursive walk's reach must not narrow: a deny nested three + levels down inside a list still denies, and still names that string.""" + payload = json.dumps( + {"args": {"targets": ["README.md", {"file": "~/.ssh/id_rsa"}]}, "mode": "read"} + ) + approved, provider, rows = await _resolve(payload) + assert approved is False + assert provider.rejected == ["r1"] + outcome, error, mechanism = _decision(rows) + assert outcome == "denied" + assert error == "Blocked: sensitive path in tool_input: ~/.ssh/id_rsa" + assert mechanism == "always_deny_input" + + @pytest.mark.asyncio + async def test_first_denial_short_circuits_in_order(self) -> None: + """Order is preserved: the FIRST denying string decides, and a later + denying string never reaches the reason.""" + payload = json.dumps(["~/.aws/credentials", "~/.ssh/id_rsa"]) + _approved, _provider, rows = await _resolve(payload) + _outcome, error, _mechanism = _decision(rows) + assert error == "Blocked: sensitive path in tool_input: ~/.aws/credentials" + + @pytest.mark.asyncio + async def test_empty_tool_input_skips_the_scan_entirely(self) -> None: + with patch.object( + llm_helpers, "_first_tool_input_denial", side_effect=AssertionError("scanned") + ): + approved, provider, rows = await _resolve("") + assert approved is True + assert provider.approved == ["r1"] + outcome, _error, _mechanism = _decision(rows) + assert outcome == "auto_approved" + + +class TestOffLoop: + """The scan's predicates must not execute on the event loop's thread.""" + + @pytest.mark.asyncio + async def test_scan_predicates_run_on_a_worker_thread(self) -> None: + loop_ident = threading.get_ident() + probe_target = "kirocrew-offload-probe.md" + seen: list[int] = [] + real = llm_helpers.is_sensitive_path + + def _probe(s: str, *a, **kw): + # The title-tier check calls this on the loop by design; only the + # tool_input string's call is under test here. + if s == probe_target: + seen.append(threading.get_ident()) + return real(s, *a, **kw) + + with patch.object(llm_helpers, "is_sensitive_path", _probe): + approved, _provider, _rows = await _resolve(json.dumps({"path": probe_target})) + + assert approved is True + assert seen, "the tool_input scan never ran" + assert loop_ident not in seen, ( + "the tool_input scan ran on the event loop thread; a long payload " + "there stalls the watchdog and takes the gateway down" + ) + + @pytest.mark.asyncio + async def test_whole_loop_is_one_thread_hop(self) -> None: + """One ``to_thread`` for the whole loop, not one per string.""" + loop_ident = threading.get_ident() + strings = [f"offload-file-{i}.txt" for i in range(25)] + seen: list[int] = [] + real = llm_helpers.is_sensitive_path + + def _probe(s: str, *a, **kw): + if s.startswith("offload-file-"): + seen.append(threading.get_ident()) + return real(s, *a, **kw) + + with patch.object(llm_helpers, "is_sensitive_path", _probe): + await _resolve(json.dumps(strings)) + + assert len(seen) == 25, seen + assert len(set(seen)) == 1, "the scan hopped threads per string instead of once" + assert loop_ident not in seen, "the single hop landed on the event loop thread" + + +class TestOversizeFailClosed: + """A string too large to scan is DENIED, never skipped. + + The thread hop yields between strings but not within one -- ``re`` holds the + GIL for a whole match call -- so one huge string could still hold the loop + past the 25s watchdog and kill the gateway. The guard refuses such input. + + Fail-closed is the load-bearing property: skipping an unscannable string + would let it past the deny surface, which trades a crash for a security + hole. These tests pin the denial, not merely the absence of a stall. + """ + + def test_oversized_string_is_denied_not_skipped(self) -> None: + cap = llm_helpers._MAX_SCANNABLE_TOOL_INPUT_CHARS + # Deliberately BENIGN content: nothing here matches any predicate, so a + # skip would return None and the call would be approved. Only the guard + # can produce a denial. + blob = "a" * (cap + 1) + hit = llm_helpers._first_tool_input_denial([blob], None) + assert hit is not None, "oversized input must be refused, not skipped" + kind, reason, matched = hit + assert kind == "oversize" + assert str(cap) in reason and str(len(blob)) in reason + assert len(matched) <= 64, "the echoed sample must stay bounded" + + def test_string_at_the_cap_is_still_scanned(self, monkeypatch) -> None: + """The guard must not fire one character early. + + The cap is patched DOWN rather than scanning a real full-size string: + the scan is superlinear, so a genuine at-the-cap scan costs seconds here + and timed out at 120s on a loaded CI runner. The off-by-one this pins is + a property of the comparison, not of the length. + """ + monkeypatch.setattr(llm_helpers, "_MAX_SCANNABLE_TOOL_INPUT_CHARS", 64) + assert llm_helpers._first_tool_input_denial(["a" * 64], None) is None + assert llm_helpers._first_tool_input_denial(["a" * 65], None) is not None + + def test_a_real_denial_still_wins_over_size(self) -> None: + """An earlier scannable string keeps its own, more specific reason.""" + cap = llm_helpers._MAX_SCANNABLE_TOOL_INPUT_CHARS + hit = llm_helpers._first_tool_input_denial(["~/.aws/credentials", "a" * (cap + 1)], None) + assert hit is not None + assert hit[0] == "path" + + @pytest.mark.asyncio + async def test_oversized_tool_input_is_rejected_end_to_end(self) -> None: + cap = llm_helpers._MAX_SCANNABLE_TOOL_INPUT_CHARS + payload = json.dumps({"content": "a" * (cap + 1)}) + approved, provider, rows = await _resolve(payload) + assert approved is False + assert provider.rejected == ["r1"] + assert provider.approved == [] + outcome, error, mechanism = _decision(rows) + assert outcome == "denied" + assert "too large to security-scan" in error + assert mechanism == "always_deny_input" + + @pytest.mark.asyncio + async def test_the_guard_bounds_the_scan_far_under_the_watchdog(self) -> None: + """A 1 MB body returns promptly instead of scanning for minutes.""" + payload = json.dumps({"content": "a" * (1024 * 1024)}) + started = time.perf_counter() + approved, _provider, _rows = await _resolve(payload) + elapsed = time.perf_counter() - started + assert approved is False + assert elapsed < 5.0, ( + f"oversized input took {elapsed:.2f}s -- the size guard is not " + "short-circuiting before the regex predicates" + ) + + +class TestLiveness: + """A 20 KB newline-free document body must not stall the loop.""" + + @pytest.mark.timeout(300) + @pytest.mark.asyncio + async def test_loop_keeps_ticking_during_a_large_document_scan(self) -> None: + # Newline-free, no shell metacharacters: a plain prose body, which is + # exactly the payload that used to be scanned as one giant command. + body = ("the quick brown fox jumps over the lazy dog " * 500)[:20_000] + assert "\n" not in body and len(body) >= 20_000 + + ticks = 0 + stop = False + + async def _ticker() -> None: + nonlocal ticks + while not stop: + ticks += 1 + await asyncio.sleep(0.005) + + task = asyncio.create_task(_ticker()) + try: + await asyncio.sleep(0.02) + before = ticks + started = time.monotonic() + approved, _provider, _rows = await _resolve(json.dumps({"content": body})) + elapsed = time.monotonic() - started + finally: + stop = True + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + assert approved is True + # Generous: the scan itself legitimately costs tens of seconds on this + # payload (is_sensitive_bash_command dominates). What the watchdog cares + # about is that the LOOP stayed live throughout, asserted below. + assert elapsed < 120.0, f"scan took {elapsed:.1f}s — far past any sane bound" + assert ticks > before, ( + f"the event loop made no progress during the scan (ticks stuck at " + f"{before}); the scan is still blocking the loop" + ) diff --git a/test/test_security_regex_linearity.py b/test/test_security_regex_linearity.py new file mode 100644 index 00000000000..d90a4a0ad2d --- /dev/null +++ b/test/test_security_regex_linearity.py @@ -0,0 +1,327 @@ +"""Differential + complexity guards for the two ``security.py`` linearity fixes. + +Both fixes are performance-only and MUST be behaviour-preserving, so the tests +here are written as *differentials*: the expected values were captured from the +implementation as it stood immediately BEFORE each change (origin/main +``760d8f570``) and are pinned as literals. A verdict or byte that moves in either +direction fails. + +Covered: + +* Mesh-3654 -- ``redact_credentials`` pass 1 was ``for m in + _CREDENTIAL_PATTERNS.finditer(result): result = result.replace(...)``, which + rebuilt the whole string per match (O(n^2) on credential-dense text). It is now + a single ``_CREDENTIAL_PATTERNS.sub(...)``. The redacted text AND the + ``warnings`` list (content *and* order) must be unchanged. +* Mesh-3693 -- eleven branches of the sensitive-path regex were anchored + ``(?:^|.*[\\s'\\"=:,;])``. The leading ``.*`` is redundant under ``re.search`` + (which retries at every offset) and made matching quadratic in the longest + line. The anchor is now ``(?:^|[\\s'\\"=:,;])``. This is a DENY surface, so the + verdict tests below replay positives and negatives to make it obvious that + nothing became more permissive. +""" + +from __future__ import annotations + +import re +import time + +import pytest + +from kiro_crew.security import ( + is_sensitive_bash_command, + is_sensitive_path, + redact_credentials, +) + +# ───────────────────────────────────────────────────────────────────────────── +# Mesh-3654: redact_credentials pass 1 -- single sub() must be byte-identical +# ───────────────────────────────────────────────────────────────────────────── + +# (input, expected_redacted_text, expected_warnings) captured from the +# pre-change loop implementation. Secret-shaped fixtures are written as adjacent +# literals so no single source line is a complete provider token (matches the +# convention in test_security.py, which keeps secret scanners quiet). +_AKIA = "AKIAIOSFODNN7EXAMPLE" +_GHP = "ghp_" "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdef12" +_ANT = "sk-ant-api03-" "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOP" +_GLPAT = "glpat-" "xxxx1234xxxx5678xxxx" +_XOXB = "xoxb-" "1234567890-abcdefghij" +_TAG = "[REDACTED: credential]" + +REDACTION_GOLDEN: list[tuple[str, str, list[str]]] = [ + ( + f"Found key {_AKIA} in output", + f"Found key {_TAG} in output", + ["Redacted credential pattern (20 chars)"], + ), + # Two occurrences of the SAME credential: both spans replaced, two warnings. + # This is the case the old `str.replace(matched, tag, 1)` shape depended on + # positional luck for -- sub() splices each matched span in place. + ( + f"a {_AKIA} b {_AKIA} c", + f"a {_TAG} b {_TAG} c", + [ + "Redacted credential pattern (20 chars)", + "Redacted credential pattern (20 chars)", + ], + ), + # Three DIFFERENT credentials -- pins warning ORDER (20, 38, 26 chars), + # which is the ordering guarantee sub() has to preserve. + ( + f"first {_AKIA} then {_GHP} and {_XOXB} tail", + f"first {_TAG} then {_TAG} and {_TAG} tail", + [ + "Redacted credential pattern (20 chars)", + "Redacted credential pattern (38 chars)", + "Redacted credential pattern (26 chars)", + ], + ), + ( + "SecretAccessKey=wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", + _TAG, + ["Redacted credential pattern (56 chars)"], + ), + ( + "aws_secret_access_key = wJalrXUtnFEMI/K7MDENG", + _TAG, + ["Redacted credential pattern (45 chars)"], + ), + ( + f"Token is {_XOXB}", + f"Token is {_TAG}", + ["Redacted credential pattern (26 chars)"], + ), + (f"KEY={_GHP}", f"KEY={_TAG}", ["Redacted credential pattern (38 chars)"]), + (f"KEY={_ANT}", f"KEY={_TAG}", ["Redacted credential pattern (55 chars)"]), + (f"KEY={_GLPAT}", f"KEY={_TAG}", ["Redacted credential pattern (26 chars)"]), + ( + "mongodb://user:supersecretpassword@cluster0.example.net/db", + f"{_TAG}cluster0.example.net/db", + ["Redacted credential pattern (35 chars)"], + ), + ( + "-----BEGIN OPENSSH PRIVATE KEY-----\nb3BlbnNzaC1r", + _TAG, + ["Redacted credential pattern (48 chars)"], + ), + # Negatives: the cheap superset gate must still short-circuit to identity. + ( + "See the PRIVATE KEY handling section of the runbook.", + "See the PRIVATE KEY handling section of the runbook.", + [], + ), + ( + "just some ordinary log line with no secrets at all", + "just some ordinary log line with no secrets at all", + [], + ), + ("", "", []), +] + + +@pytest.mark.parametrize( + ("text", "expected_text", "expected_warnings"), + REDACTION_GOLDEN, + ids=[f"case-{i}" for i in range(len(REDACTION_GOLDEN))], +) +def test_pass1_single_sub_is_byte_identical_to_pre_change_loop( + text: str, expected_text: str, expected_warnings: list[str] +) -> None: + """Pass 1 as one ``sub()`` reproduces the old loop's bytes and warnings. + + Differential for Mesh-3654. ``expected_warnings`` is compared with ``==`` on + the list, so both the CONTENT and the ORDER are pinned -- appending in the + replacement callback has to keep the left-to-right match order the old + ``finditer`` loop had. + """ + result, warnings = redact_credentials(text) + assert result == expected_text + assert warnings == expected_warnings + + +def test_pass1_warning_order_tracks_match_order_not_length() -> None: + """Warnings come out in match order, not sorted or grouped. + + A replacement callback that batched or reordered its appends would still + produce identical TEXT, so this asserts the ordering separately. + """ + text = f"{_ANT} {_AKIA} {_GHP}" + _, warnings = redact_credentials(text) + assert warnings == [ + f"Redacted credential pattern ({len(_ANT)} chars)", + f"Redacted credential pattern ({len(_AKIA)} chars)", + f"Redacted credential pattern ({len(_GHP)} chars)", + ] + + +def test_pass1_warnings_still_carry_no_secret_bytes() -> None: + """The replacement callback must not slice the match into the warning.""" + text = f"KEY={_ANT}" + _, warnings = redact_credentials(text) + joined = " ".join(warnings) + assert _ANT not in joined + assert _ANT[:20] not in joined + assert "Redacted credential pattern" in joined + + +def test_pass1_is_linear_on_credential_dense_text() -> None: + """Complexity guard for Mesh-3654. + + The old shape rebuilt the whole string per match, so redacting N credentials + in an N-credential string was O(N^2). 4000 credentials (~84 KB) is + sub-second as one ``sub()`` pass; the generous ceiling keeps this off slow + CI's flake list while still failing hard if the per-match rebuild returns. + """ + dense = f"{_AKIA} " * 4000 + started = time.perf_counter() + result, warnings = redact_credentials(dense) + elapsed = time.perf_counter() - started + assert len(warnings) == 4000 + assert _AKIA not in result + assert elapsed < 5.0, f"pass 1 took {elapsed:.2f}s -- per-match string rebuild is back" + + +# ───────────────────────────────────────────────────────────────────────────── +# Mesh-3693: sensitive-path anchor -- zero verdict change (DENY surface) +# ───────────────────────────────────────────────────────────────────────────── + +# (command, expected_verdict) captured from the pre-change regex. Ordered so the +# separator-boundary cases the character class exists for are explicit: the path +# preceded by a space, a single quote, a double quote, `=`, `:`, `,`, `;`, and at +# string start -- plus mid-token cases that must stay NEGATIVE. +SENSITIVE_COMMAND_GOLDEN: list[tuple[str, bool]] = [ + # ── separator boundaries: each must stay a HIT ── + ("cat ~/.aws/credentials", True), # space + ("cat '~/.aws/credentials'", True), # single quote + ('cat "~/.ssh/id_rsa"', True), # double quote + ("FOO=~/.aws/credentials", True), # `=` (VAR=path) + ("PATH=/x:~/.ssh/id_rsa", True), # `:` (PATH-style list) + ("cmd --a=1,~/.aws/credentials", True), # `,` + ("run;~/.aws/credentials", True), # `;` + ("~/.aws/credentials", True), # start-of-string (`^`) + ("echo x\n~/.aws/credentials", True), # newline is in the class + # ── the same hits further into the line: `.*` was never what found these, + # `re.search` retrying at every offset was ── + ("a b c d e f g h ~/.aws/credentials", True), + ("prefix text then FOO=bar:~/.aws/credentials suffix", True), + ("deploy --flag ~/.ssh/id_rsa", True), + ("a,~/.gnupg/secring.gpg", True), + # ── other spellings that route through the rewritten branches ── + ("cat $HOME/.aws/credentials", True), + ("type %USERPROFILE%\\.aws\\credentials", True), + ("type $env:USERPROFILE\\.ssh\\id_rsa", True), + ("cp ~/.kiro/agents/x.json /tmp/y", True), + # ── embedded MID-TOKEN: no separator immediately before the path, so the + # anchor must NOT fire. These are the cases that would flip to True if + # the character class were dropped along with the `.*`. ── + ("xyz~/.aws/credentials", False), + ("FOO=bar~/.aws/credentials", False), + ("VAR=x~/.gnupg/secring.gpg", False), + ("printf q~/.aws/credentials", False), + # ── ordinary commands: must stay allowed ── + ("ls -la", False), + ("echo hello world", False), + ("cat myfile.txt", False), + ("notaws/credentials", False), + ("python -c 'print(1)'", False), + ("grep -r pattern src/", False), + ("cat ./relative/notsensitive.json", False), + ("git status", False), + ("make build", False), +] + + +@pytest.mark.parametrize(("command", "expected"), SENSITIVE_COMMAND_GOLDEN) +def test_sensitive_bash_verdicts_unchanged_by_anchor_rewrite(command: str, expected: bool) -> None: + """Differential for Mesh-3693 on ``is_sensitive_bash_command``. + + Every verdict is pinned to what the pre-change regex returned. Dropping the + redundant ``.*`` cannot change any of them: the alternative is still ``^`` or + a single separator character, and ``re.search`` already retried at every + offset. A regression in EITHER direction fails here -- the negatives are what + make it obvious the gate did not become more permissive. + """ + assert bool(is_sensitive_bash_command(command)) is expected + + +SENSITIVE_PATH_GOLDEN: list[tuple[str, bool]] = [ + ("~/.aws/credentials", True), + ("~/.ssh/id_rsa", True), + ("~/.gnupg/secring.gpg", True), + ("/tmp/harmless.txt", False), + ("./README.md", False), + ("src/kiro_crew/security.py", False), + ("notes.md", False), +] + + +@pytest.mark.parametrize(("path", "expected"), SENSITIVE_PATH_GOLDEN) +def test_sensitive_path_verdicts_unchanged_by_anchor_rewrite(path: str, expected: bool) -> None: + """Differential for Mesh-3693 on ``is_sensitive_path``.""" + assert bool(is_sensitive_path(path)) is expected + + +def test_sensitive_anchor_has_no_leading_wildcard() -> None: + """Source guard: the redundant ``.*`` must not come back. + + ``_build_sensitive_regex`` is the only place these anchors are written. The + check is on the source text rather than the compiled pattern because the + compiled form interpolates the path alternations and is impractical to + assert against. + """ + from kiro_crew import security as security_mod + + source = inspect_source(security_mod._build_sensitive_regex) + assert r"""(?:^|.*[\s'\"=:,;])""" not in source, ( + "a leading `.*` is back in the sensitive-path anchor -- it is redundant " + "under re.search and makes matching quadratic in the longest line" + ) + # And the fixed form is still there, on every branch it was applied to. + # Count the BRANCH spelling (``rf"|`` prefix) so the explanatory comment in + # `_build_sensitive_regex`, which quotes the anchor in prose, is not counted. + branch_anchor = r"""rf"|(?:^|[\s'\"=:,;])""" + assert source.count(branch_anchor) == 11 + + +def inspect_source(func: object) -> str: + """``inspect.getsource`` indirection kept local so the test module has one import.""" + import inspect + + return inspect.getsource(func) # type: ignore[arg-type] + + +def test_long_nonshell_line_does_not_blow_up() -> None: + """Complexity guard for Mesh-3693. + + A ~20 KB newline-free non-shell string is the worst case for the old anchor: + eleven branches each retried a greedy ``.*`` from every offset. Measured on + the dev box this took ~27 s before the rewrite and ~1.5 s after, so a 6 s + ceiling clears the fixed path by ~4x while the quadratic form overshoots by + ~4.5x. Deliberately generous -- this test exists to catch a complexity + regression, not to benchmark CI. + """ + blob = "abcdefgh " * 2500 + assert len(blob) > 20_000 + started = time.perf_counter() + verdict = is_sensitive_bash_command(blob) + elapsed = time.perf_counter() - started + assert bool(verdict) is False + assert elapsed < 6.0, ( + f"is_sensitive_bash_command took {elapsed:.2f}s on a 20 KB line -- " + "a leading `.*` in the sensitive-path anchor is quadratic" + ) + + +def test_credential_pattern_module_still_compiles_one_alternation() -> None: + """Invariant: the rewritten pass 1 still uses the shared compiled pattern. + + Guards against a future refactor swapping in a locally compiled regex, which + would silently drop the ``_might_contain_credential`` pre-filter pairing. + """ + from kiro_crew import security as security_mod + + assert isinstance(security_mod._CREDENTIAL_PATTERNS, re.Pattern) + body = inspect_source(security_mod.redact_credentials) + assert "_CREDENTIAL_PATTERNS.sub(" in body + assert "_might_contain_credential(result)" in body diff --git a/test/test_terminal_handler.py b/test/test_terminal_handler.py index 3d40dd72538..4f23ce9de67 100644 --- a/test/test_terminal_handler.py +++ b/test/test_terminal_handler.py @@ -4325,3 +4325,144 @@ def test_session_dataclass_has_write_lock(self): names = {f.name for f in dataclasses.fields(terminal._TerminalSession)} assert "write_lock" in names + + +class TestPtyChildEnvStripsPythonStartupVars: + """``PYTHONPATH``/``PYTHONHOME``/``PYTHONPYCACHEPREFIX`` are searched BEFORE a + venv's own site-packages, so leaking the gateway's copies into an interactive + shell makes a user's Python 3.13 venv import Kiro Crew's 3.12 site-packages + and its C extensions fail to load. The agent surface already strips them + (``sandbox.scrub_agent_subprocess_env``); these pin the terminal surface, + which was never brought into line. + """ + + def test_python_startup_vars_are_dropped(self, monkeypatch): + monkeypatch.setenv("PYTHONPATH", "/gateway/site-packages") + monkeypatch.setenv("PYTHONHOME", "/gateway/python3.12") + monkeypatch.setenv("PYTHONPYCACHEPREFIX", "/gateway/pycache") + monkeypatch.setenv("KIROCREW_UNRELATED_KEEPME", "keep-this-value") + + env = terminal._pty_child_env( + {"TERM": "xterm-256color", "KIROCREW_TERMINAL": "1"} + ) + + assert "PYTHONPATH" not in env + assert "PYTHONHOME" not in env + assert "PYTHONPYCACHEPREFIX" not in env + assert env["KIROCREW_TERMINAL"] == "1" + assert env["TERM"] == "xterm-256color" + assert env["KIROCREW_UNRELATED_KEEPME"] == "keep-this-value" + + def test_credential_bearing_vars_survive(self, monkeypatch): + """Only the Python prefixes are dropped. This is the user's own + unsandboxed shell, so borrowing the AGENT spawn's credential scrub would + break git-over-SSH and the AWS CLI inside the panel.""" + monkeypatch.setenv("SSH_AUTH_SOCK", "/tmp/ssh-abc/agent.1") + monkeypatch.setenv("AWS_SESSION_TOKEN", "FAKE-token") + monkeypatch.setenv("PYTHONPATH", "/gateway/site-packages") + + env = terminal._pty_child_env({"KIROCREW_TERMINAL": "1"}) + + assert env["SSH_AUTH_SOCK"] == "/tmp/ssh-abc/agent.1" + assert env["AWS_SESSION_TOKEN"] == "FAKE-token" + assert "PYTHONPATH" not in env + + @pytest.mark.asyncio + async def test_posix_pty_spawn_env_has_no_python_vars(self, monkeypatch): + """End-to-end through the POSIX branch: assert on the env actually handed + to the spawn, so rebuilding the dict in place is caught. The spawn is + made to fail AFTER the call is recorded so no read loop starts.""" + monkeypatch.setenv("PYTHONPATH", "/gateway/site-packages") + monkeypatch.setenv("PYTHONHOME", "/gateway/python3.12") + monkeypatch.setenv("SHELL", "/bin/bash") + monkeypatch.setattr( + terminal.shutil, "which", lambda c: c if c == "/bin/zsh" else None + ) + + registry: dict = {} + req = _make_request(registry=registry, session_id="posix-pyenv") + req.query = MagicMock() + req.query.get = lambda *a, **k: None + + ws = AsyncMock() + ws.closed = False + + fds = os.pipe() # real fds so the cleanup os.close() calls succeed + spawn = AsyncMock(side_effect=RuntimeError("stop before read loop")) + cfg = {"enabled": True, "shell": "/bin/zsh"} + with patch.object(terminal.platform_compat, "IS_POSIX", True), \ + patch.object(terminal.platform_compat, "IS_WINDOWS", False), \ + patch.object(terminal._pty, "openpty", return_value=fds), \ + patch.object(terminal.fcntl, "ioctl", lambda *a: None), \ + patch.object(terminal.asyncio, "create_subprocess_exec", spawn), \ + patch.object(terminal, "_get_config", return_value=cfg), \ + patch.object(terminal.web, "WebSocketResponse", return_value=ws), \ + patch.object(terminal, "_sel") as mock_sel: + mock_sel.return_value.log_api_access = MagicMock() + await terminal.api_terminal_ws(req) + + spawn.assert_awaited_once() + env = spawn.call_args.kwargs["env"] + assert "PYTHONPATH" not in env + assert "PYTHONHOME" not in env + assert env["KIROCREW_TERMINAL"] == "1" + assert env["TERM"] == "xterm-256color" + assert env["SHELL"] == "/bin/zsh" + + @pytest.mark.asyncio + async def test_conpty_spawn_env_has_no_python_vars(self, monkeypatch, tmp_path): + """The Windows ConPTY branch IS reachable on Linux: ``IS_WINDOWS`` is a + module attribute and ``WindowsPty`` is a thin pywinpty wrapper the suite + already fakes, so the same code path runs here.""" + monkeypatch.setenv("PYTHONPATH", "/gateway/site-packages") + monkeypatch.setenv("PYTHONHOME", "/gateway/python3.12") + + cfg_file = tmp_path / "config.json" + cfg_file.write_text(json.dumps({"dashboard": {"terminal": {"enabled": True}}})) + monkeypatch.setattr(terminal, "config_path", lambda: cfg_file) + monkeypatch.setattr(terminal, "_sel", lambda: MagicMock()) + monkeypatch.setattr(terminal.platform_compat, "IS_POSIX", False) + monkeypatch.setattr(terminal.platform_compat, "IS_WINDOWS", True) + + captured: dict = {} + + class _FakeWinPty: + def __init__(self, argv, cwd=None, env=None, cols=80, rows=24): + captured["env"] = env + self.pid = 4321 + self._reads = iter((b"PS> ", b"")) + + def read(self, size=4096): + return next(self._reads) + + def write(self, data): + return len(data) + + def resize(self, cols, rows): + pass + + def isalive(self): + return True + + def terminate(self, force=True): + pass + + monkeypatch.setattr("kiro_crew.conpty.WindowsPty", _FakeWinPty) + + registry: dict = {} + app = _make_app(registry=registry) + + from aiohttp.test_utils import TestClient, TestServer + + async with TestClient(TestServer(app)) as client: + async with client.ws_connect("/api/ws/terminal/win-pyenv") as ws: + await ws.receive(timeout=3) + await ws.close() + + if "win-pyenv" in registry: + await terminal._kill_session(registry["win-pyenv"]) + + env = captured["env"] + assert "PYTHONPATH" not in env + assert "PYTHONHOME" not in env + assert env["KIROCREW_TERMINAL"] == "1"