diff --git a/src/kiro_crew/sandbox.py b/src/kiro_crew/sandbox.py index 359864a2d2b..77a2c099187 100644 --- a/src/kiro_crew/sandbox.py +++ b/src/kiro_crew/sandbox.py @@ -1679,12 +1679,48 @@ def main(): # available — better to function (with the original regression risk) # than to refuse to start. - # Pre-read files that must survive dir hiding + # Pre-read files that must survive dir hiding. + # + # An expose source that cannot be READ degrades to "not exposed" with a + # stderr warning, the same way the Step 7 hardlink scan degrades open. + # This read runs during sandbox SETUP, so letting the OSError propagate + # aborts the child before the command runs at all -- and selective + # exposure is an OPTIMIZATION (keep ~/.aws/config reachable so + # credential_process still resolves inside an otherwise-hidden ~/.aws), + # never a security control. Failing the whole spawn because an optional + # convenience is unreadable trades a working sandbox for no sandbox. + # + # `isfile` already covers ABSENT; this covers UNREADABLE, and the two + # are not the same test: `stat` can succeed on a path whose `open` is + # then denied. Seen in the wild as a filesystem restriction inherited + # from the parent process, denying read on a 0600 file the child's own + # uid owned -- so DAC bits and uid both looked correct while every + # cc-mode spawn on that host died here. + # + # Catching the error is the only guard that HOLDS. Do not "tighten" this + # into a pre-flight `os.access(src_path, os.R_OK)`: measured on the + # affected host, `os.stat()` succeeded and `os.access()` reported BOTH + # X_OK and R_OK as True while the operation was denied anyway. The + # weaker check looks equivalent from the source alone and would + # silently restore the abort. + # + # The warning is not optional. Skipping silently would leave the child + # with no ~/.aws/config and no explanation, turning a loud setup failure + # into a later auth failure that points nowhere near this line. expose_data = {{}} for src_path, filename in EXPOSE_FILES: if os.path.isfile(src_path): - with open(src_path, "rb") as fh: - expose_data[src_path] = fh.read() + try: + with open(src_path, "rb") as fh: + expose_data[src_path] = fh.read() + except OSError as exc: + print( + "sandbox: WARNING — cannot read %s (%s); it will be " + "ABSENT inside the sandbox. Anything depending on it " + "(e.g. credential_process in ~/.aws/config) will fail." + % (src_path, exc), + file=sys.stderr, + ) # Bind-mount empty dirs over credential paths (per-dir tmpdir to # prevent content leaking across mounts via shared backing dir). @@ -1725,8 +1761,38 @@ def main(): if HIDE_SSH and os.path.isdir(SSH_DIR): kh_data = b"" if os.path.isfile(SSH_KNOWN_HOSTS): - with open(SSH_KNOWN_HOSTS, "rb") as fh: - kh_data = fh.read() + # Host trust data FAILS CLOSED. This is deliberately NOT the + # degrade-open treatment the EXPOSE_FILES pre-read above gets, + # and the two sites are NOT symmetric: + # + # - an unreadable ~/.aws/config costs REACHABILITY, so + # skipping it trades a convenience for a working sandbox; + # - an unreadable known_hosts costs VERIFICATION. The launcher + # puts StrictHostKeyChecking=accept-new into + # GIT_SSH_COMMAND, gated ONLY on that variable being unset + # -- never on whether this read succeeded. So continuing + # with an empty kh_data points UserKnownHostsFile at an + # absent file while auto-accept is still on: every host then + # reads as NEW and an interceptor's key is accepted. With + # known_hosts present, accept-new REFUSES a CHANGED key. + # + # A degrade here would therefore convert "refuse a changed key" + # into "accept anything". Aborting is the safe direction: no + # sandbox at all beats one that has quietly stopped verifying + # hosts. Report first so the abort is diagnosable, then re-raise + # and let it kill setup. + try: + with open(SSH_KNOWN_HOSTS, "rb") as fh: + kh_data = fh.read() + except OSError as exc: + print( + "sandbox: FATAL — cannot read %s (%s). Refusing to " + "continue: proceeding without it would leave host-key " + "verification accepting any new key." + % (SSH_KNOWN_HOSTS, exc), + file=sys.stderr, + ) + raise # Cross-fs source for the same kernel-race reason as SENSITIVE_DIRS # (line 371) and SENSITIVE_FILES (line 389). ssh_tmp = tempfile.mkdtemp(dir=_tmpfs_src).encode() diff --git a/test/test_sandbox_cc_mode.py b/test/test_sandbox_cc_mode.py index 09cfcb76366..1a91b5fb128 100644 --- a/test/test_sandbox_cc_mode.py +++ b/test/test_sandbox_cc_mode.py @@ -3,6 +3,9 @@ from __future__ import annotations import os +import runpy +import textwrap +from pathlib import Path from unittest.mock import patch import pytest @@ -394,3 +397,416 @@ def test_scrub_agent_denied_env_preserves_aws_ssh(self): assert cleaned["AWS_SESSION_TOKEN"] == "FAKE-session" assert cleaned["SSH_AUTH_SOCK"] == "/tmp/fake-agent.sock" assert cleaned["PATH"] == "/usr/bin" + + +# ── The cc-mode expose pre-read ── +# +# Selective exposure keeps ~/.aws/config readable inside an otherwise-hidden +# ~/.aws, so credential_process still resolves. It is an optimisation, and the +# pre-read of it happens during sandbox SETUP -- so an OSError there aborts the +# child before the command runs at all. `isfile` covers ABSENT; these tests +# cover UNREADABLE, which is a different condition (an EACCES on open() still +# passes isfile when the path is traversable and stat-able). +# +# Like test_sandbox_hardlink_scan.py, these run the block from the SHIPPED +# launcher source rather than a copy, so they cannot drift from what the child +# actually executes. +_EXPOSE_BLOCK_START = "expose_data = {}" +_EXPOSE_BLOCK_END = "# Bind-mount empty dirs over credential paths" +#: Structural landmarks the slice must contain, so an edit that moves either +#: marker and shrinks the block fails HERE rather than leaving the assertions +#: below vacuously green against a fragment that no longer holds the read. +_EXPOSE_SLICE_LANDMARKS = ( + "for src_path, filename in EXPOSE_FILES:", # the loop + "os.path.isfile(src_path)", # the absent-file guard + 'open(src_path, "rb")', # the read itself +) + + +def _expose_pre_read_source() -> str: + """The expose pre-read, lifted verbatim out of the generated cc launcher. + + Sliced from the START OF THE LINE, not from the marker: ``dedent`` measures + the common prefix across all lines, so a first line already stripped of its + indent leaves the rest indented and the block will not parse. + """ + script = _build_launcher_script("cc") + start = script.rindex("\n", 0, script.index(_EXPOSE_BLOCK_START)) + 1 + end = script.rindex("\n", 0, script.index(_EXPOSE_BLOCK_END, start)) + 1 + block = textwrap.dedent(script[start:end]) + missing = [mark for mark in _EXPOSE_SLICE_LANDMARKS if mark not in block] + assert not missing, f"the extracted expose pre-read is missing {missing}" + return block + + +def _run_expose_pre_read( + *, expose_files: list[tuple[str, str]], tmp_path: Path +) -> tuple[dict[str, bytes], str]: + """Run the pre-read over *expose_files*; return ``(expose_data, stderr)``. + + Via ``runpy.run_path`` rather than ``exec`` for the reason given in + test_sandbox_hardlink_scan.py: ``exec`` trips the SAST gate's + ``exec-detected`` rule, and a suppression would be this repo's first. + """ + written: list[str] = [] + + class _Stderr: + def write(self, text: str) -> int: + written.append(text) + return len(text) + + fake_sys = type("_sys", (), {"stderr": _Stderr()})() + block = tmp_path / "_expose_block.py" + block.write_text(_expose_pre_read_source(), encoding="utf-8") + result = runpy.run_path( + str(block), + init_globals={"os": os, "sys": fake_sys, "EXPOSE_FILES": expose_files}, + ) + return result["expose_data"], "".join(written) + + +def _require_eacces(path: Path) -> None: + """Make *path* unreadable, or skip if this host cannot make it so.""" + path.chmod(0o000) + if os.access(path, os.R_OK): # root, or a filesystem ignoring the mode + pytest.skip("this host can read a 0000 file; EACCES is unreachable") + + +class TestCcExposePreReadIsNonFatal: + def test_an_unreadable_expose_source_does_not_abort_setup(self, tmp_path: Path) -> None: + """The regression. Before the guard this raised PermissionError. + + The read sits in sandbox setup, so the exception killed the spawn + outright. Measured consequence on one host: every cc-mode spawn died, + which is the whole ``command`` cron kind (``run_command_sandboxed`` uses + ``mode="cc"`` while ``run_script_sandboxed`` uses ``mode="standard"``), + and the repeated failures latched three jobs into auto-pause. + """ + src = tmp_path / "config" + src.write_text("[default]\nregion = us-east-1\n", encoding="utf-8") + _require_eacces(src) + + # Not raising IS the assertion; _run_expose_pre_read propagates. + expose_data, _ = _run_expose_pre_read( + expose_files=[(str(src), "config")], tmp_path=tmp_path + ) + + assert str(src) not in expose_data, "an unreadable source must not be exposed" + + def test_an_unreadable_expose_source_is_reported_on_stderr(self, tmp_path: Path) -> None: + """Degrading SILENTLY would be the opposite of the intent. + + Without the exposure the child has no ~/.aws/config, so Bedrock auth + fails later with an error pointing nowhere near the pre-read. The + warning is what connects the two. + """ + src = tmp_path / "config" + src.write_text("[default]\n", encoding="utf-8") + _require_eacces(src) + + _, stderr = _run_expose_pre_read(expose_files=[(str(src), "config")], tmp_path=tmp_path) + + assert stderr, "skipping an exposure silently must not be an option" + assert str(src) in stderr, "the warning must name the path it skipped" + + def test_a_readable_expose_source_is_still_read(self, tmp_path: Path) -> None: + """Positive control: the guard must not swallow the happy path. + + This passes with or without the guard, on purpose -- it is what would + catch a "fix" that skipped every exposure. + """ + src = tmp_path / "config" + src.write_bytes(b"[default]\nregion = eu-west-1\n") + + expose_data, stderr = _run_expose_pre_read( + expose_files=[(str(src), "config")], tmp_path=tmp_path + ) + + assert expose_data[str(src)] == b"[default]\nregion = eu-west-1\n" + assert stderr == "", "a successful read must stay quiet" + + def test_an_absent_expose_source_stays_silent(self, tmp_path: Path) -> None: + """``isfile`` still shorts out first, so absence is not a warning. + + ~/.aws/config does not exist on plenty of hosts. Warning there would put + a line on stderr for every cc-mode spawn on all of them. + """ + expose_data, stderr = _run_expose_pre_read( + expose_files=[(str(tmp_path / "absent"), "config")], tmp_path=tmp_path + ) + + assert expose_data == {} + assert stderr == "", "an absent optional exposure is not a problem" + + def test_the_guard_is_the_exception_not_a_pre_flight_access_check(self, tmp_path: Path) -> None: + """`os.access` is not a valid substitute for catching the error. + + Measured on the affected host: `os.stat()` succeeded and `os.access()` + reported both X_OK and R_OK as True while the operation was denied + anyway. So a reviewer "tightening" the guard into + `os.access(src_path, os.R_OK)` would look equivalent from the source and + silently restore the abort. This pins the read as being attempted and the + failure as being caught, by handing the block an `os` whose `access` + lies exactly the way the real one did. + """ + src = tmp_path / "config" + src.write_text("[default]\n", encoding="utf-8") + _require_eacces(src) + + opened: list[str] = [] + + class _LyingOs: + """The real ``os``, but ``access`` always says yes (as measured).""" + + def access(self, path, mode) -> bool: + return True + + def __getattr__(self, name: str): + return getattr(os, name) + + real_open = open + + def _counting_open(path, *args, **kwargs): + opened.append(str(path)) + return real_open(path, *args, **kwargs) + + written: list[str] = [] + + class _Stderr: + def write(self, text: str) -> int: + written.append(text) + return len(text) + + fake_sys = type("_sys", (), {"stderr": _Stderr()})() + block = tmp_path / "_expose_block_access.py" + block.write_text(_expose_pre_read_source(), encoding="utf-8") + result = runpy.run_path( + str(block), + init_globals={ + "os": _LyingOs(), + "sys": fake_sys, + "open": _counting_open, + "EXPOSE_FILES": [(str(src), "config")], + }, + ) + + # A pre-flight os.access guard would have skipped the open entirely and + # emitted nothing, so BOTH of these fail on that rewrite. + assert opened == [str(src)], "the read must be attempted, not gated on os.access" + assert "".join(written), "the denied read must still be reported" + assert str(src) not in result["expose_data"] + + def test_one_unreadable_source_does_not_block_the_others(self, tmp_path: Path) -> None: + """The skip is per entry, not per loop. + + ``_CC_EXPOSE_FILES`` carries one path today, so without this the + per-entry scope is only implied by where the ``try`` sits. + """ + bad = tmp_path / "unreadable" + bad.write_text("x\n", encoding="utf-8") + _require_eacces(bad) + good = tmp_path / "readable" + good.write_bytes(b"kept\n") + + expose_data, stderr = _run_expose_pre_read( + expose_files=[(str(bad), "unreadable"), (str(good), "readable")], + tmp_path=tmp_path, + ) + + assert str(bad) not in expose_data + assert expose_data[str(good)] == b"kept\n" + assert str(bad) in stderr + assert str(good) not in stderr + + +# ── The known_hosts pre-read: same shape as the expose read, OPPOSITE remedy ── +# +# Same root cause (an unguarded `isfile` -> `open` that can raise during setup), +# but the safe direction is REVERSED, so this site fails CLOSED where the expose +# read degrades open. The asymmetry is not a style choice: +# +# - an unreadable ~/.aws/config costs REACHABILITY; +# - an unreadable known_hosts costs VERIFICATION, because the launcher sets +# StrictHostKeyChecking=accept-new in GIT_SSH_COMMAND gated only on that +# variable being unset -- never on whether this read succeeded. Continuing +# with an empty kh_data therefore leaves auto-accept on with no trust +# anchors, so any host key is accepted as new. +# +# Reach: HIDE_SSH is set at the DEFAULT strict level (`hide_ssh = sandbox_level +# == "strict"`, and `sandbox_level` defaults to "strict"), not just in cc mode. +# +# Sliced from the SHIPPED launcher for the same anti-drift reason as the block +# above. Only the pre-read is sliced, NOT the whole `.ssh` block: the lines +# around it call `_libc.mount()`, which cannot run in-process. +_KH_BLOCK_START = 'kh_data = b""' +_KH_BLOCK_END = "# Cross-fs source for the same kernel-race" +#: Structural landmarks, so an edit that moves a marker and shrinks the slice +#: fails HERE rather than leaving the assertions vacuously green. +_KH_SLICE_LANDMARKS = ( + "os.path.isfile(SSH_KNOWN_HOSTS)", # the absent-file guard + 'open(SSH_KNOWN_HOSTS, "rb")', # the read itself +) + + +def _known_hosts_pre_read_source() -> str: + """The known_hosts pre-read, lifted verbatim out of the strict launcher.""" + script = _build_launcher_script("strict") + start = script.rindex("\n", 0, script.index(_KH_BLOCK_START)) + 1 + end = script.rindex("\n", 0, script.index(_KH_BLOCK_END, start)) + 1 + block = textwrap.dedent(script[start:end]) + missing = [mark for mark in _KH_SLICE_LANDMARKS if mark not in block] + assert not missing, f"the extracted known_hosts pre-read is missing {missing}" + return block + + +def _run_known_hosts_pre_read( + *, known_hosts: str, tmp_path: Path, stderr_sink: list[str] | None = None +) -> tuple[bytes, str]: + """Run the pre-read over *known_hosts*; return ``(kh_data, stderr)``. + + Propagates whatever the block raises -- the refusal is the behaviour under + test. Pass ``stderr_sink`` to keep the collected stderr reachable when it DOES + raise, since the returned tuple is unreachable in exactly that case; the + caller owns the list, so nothing has to be carried in module state. + """ + written: list[str] = stderr_sink if stderr_sink is not None else [] + + class _Stderr: + def write(self, text: str) -> int: + written.append(text) + return len(text) + + fake_sys = type("_sys", (), {"stderr": _Stderr()})() + block = tmp_path / "_known_hosts_block.py" + block.write_text(_known_hosts_pre_read_source(), encoding="utf-8") + result = runpy.run_path( + str(block), + init_globals={"os": os, "sys": fake_sys, "SSH_KNOWN_HOSTS": known_hosts}, + ) + return result["kh_data"], "".join(written) + + +class TestKnownHostsPreReadFailsClosed: + def test_an_unreadable_known_hosts_aborts_setup(self, tmp_path: Path) -> None: + """Unreadable host-trust data must FAIL CLOSED, not degrade. + + This test previously pinned the opposite, and that was a defect. The + launcher injects ``StrictHostKeyChecking=accept-new`` into + ``GIT_SSH_COMMAND`` (built at sandbox.py:1513-1515, applied at + sandbox.py:1786-1793) gated only on that variable being unset -- NOT on + whether known_hosts was restored. So degrading to an empty ``kh_data`` + leaves the sandbox pointing ``UserKnownHostsFile`` at an absent file + while auto-accept is still on: every host reads as NEW, and an + interceptor's key is accepted. With known_hosts PRESENT, ``accept-new`` + REFUSES a CHANGED key. Degrading therefore converts "refuse a changed + key" into "accept anything". + + That is why this site is NOT symmetric with the EXPOSE_FILES pre-read. + Hiding ~/.aws/config only costs reachability; hiding known_hosts removes + a trust anchor while leaving the auto-accept that anchor was gating. + """ + kh = tmp_path / "known_hosts" + kh.write_text("example.com ssh-ed25519 AAAA\n", encoding="utf-8") + _require_eacces(kh) + + stderr: list[str] = [] + with pytest.raises(OSError): + _run_known_hosts_pre_read( + known_hosts=str(kh), tmp_path=tmp_path, stderr_sink=stderr + ) + + # The refusal and its diagnostic are ONE behaviour observed from ONE setup, + # so they are asserted together. Refusing silently would strand the operator + # on a bare OSError out of a pre-read they have no reason to connect to host + # trust, which is why the message is pinned as tightly as the raise. + emitted = "".join(stderr) + assert emitted, "refusing must not be silent" + assert str(kh) in emitted, "the message must name the path" + assert "FATAL" in emitted, "this is a refusal, not a warning" + + def test_a_readable_known_hosts_is_still_read(self, tmp_path: Path) -> None: + """Positive control: the guard must not swallow the happy path. + + Passes with or without the guard, on purpose -- it is what would catch a + "fix" that skipped the exposure unconditionally. + """ + kh = tmp_path / "known_hosts" + kh.write_bytes(b"host.example ssh-rsa BBBB\n") + + kh_data, stderr = _run_known_hosts_pre_read(known_hosts=str(kh), tmp_path=tmp_path) + + assert kh_data == b"host.example ssh-rsa BBBB\n" + assert stderr == "", "a successful read must stay quiet" + + def test_an_absent_known_hosts_stays_silent(self, tmp_path: Path) -> None: + """``isfile`` still shorts out first, so absence is not a warning. + + Plenty of hosts have a .ssh directory and no known_hosts; warning there + would put a line on stderr for every strict-mode spawn on all of them. + """ + kh_data, stderr = _run_known_hosts_pre_read( + known_hosts=str(tmp_path / "absent"), tmp_path=tmp_path + ) + + assert kh_data == b"" + assert stderr == "", "an absent known_hosts is not a problem" + + def test_the_known_hosts_guard_is_the_exception_not_a_pre_flight_access_check( + self, tmp_path: Path + ) -> None: + """`os.access` is not a valid substitute here either. + + Same measurement as the expose site: `os.stat()` succeeded and + `os.access()` reported R_OK True while the read was denied anyway. Pinned + the same way -- hand the block an `os` whose `access` lies, and assert + the read was still ATTEMPTED and the failure CAUGHT. + """ + kh = tmp_path / "known_hosts" + kh.write_text("example.com ssh-ed25519 AAAA\n", encoding="utf-8") + _require_eacces(kh) + + opened: list[str] = [] + + class _LyingOs: + """The real ``os``, but ``access`` always says yes (as measured).""" + + def access(self, path, mode) -> bool: + return True + + def __getattr__(self, name: str): + return getattr(os, name) + + real_open = open + + def _counting_open(path, *args, **kwargs): + opened.append(str(path)) + return real_open(path, *args, **kwargs) + + written: list[str] = [] + + class _Stderr: + def write(self, text: str) -> int: + written.append(text) + return len(text) + + fake_sys = type("_sys", (), {"stderr": _Stderr()})() + block = tmp_path / "_known_hosts_block_access.py" + block.write_text(_known_hosts_pre_read_source(), encoding="utf-8") + with pytest.raises(OSError): + runpy.run_path( + str(block), + init_globals={ + "os": _LyingOs(), + "sys": fake_sys, + "open": _counting_open, + "SSH_KNOWN_HOSTS": str(kh), + }, + ) + + # A pre-flight os.access guard would have skipped the open entirely, + # emitted nothing, and CONTINUED with an empty kh_data -- which is the + # fail-open this site must not do. All three of these fail on that + # rewrite: no open attempted, no message, and no refusal. + assert opened == [str(kh)], "the read must be attempted, not gated on os.access" + assert "".join(written), "the denied read must still be reported" + assert "FATAL" in "".join(written), "this is a refusal, not a warning" diff --git a/test/test_worktree_create.py b/test/test_worktree_create.py index 09945ae337d..59dbcc446af 100644 --- a/test/test_worktree_create.py +++ b/test/test_worktree_create.py @@ -1086,6 +1086,14 @@ class TestLauncherAdvisoryIsNotARefusal: "/tmp; scan incomplete (control degrades open)" ) _FATAL = "sandbox: unshare(NEWNS) failed: errno 1" + #: The launcher's other fatal spelling: the host-trust pre-read that FAILS CLOSED. + #: Hand-typed like ``_WARNING`` above, and ratcheted by the same test below. The + #: path is illustrative; the severity word and the prefix are what classify it. + _FATAL_PRE_READ = ( + "sandbox: FATAL — cannot read /home/u/hosts-file ([Errno 13] Permission " + "denied). Refusing to continue: proceeding without it would leave host-key " + "verification accepting any new key." + ) @pytest.fixture(autouse=True) def _spawn_passthrough(self, monkeypatch): @@ -1158,12 +1166,45 @@ def test_the_launcher_emits_no_severity_the_classifier_does_not_know(self): for token in re.findall(r"sandbox: ([^'\"%\\\n]+)", _build_launcher_script("strict")) } - assert severities == {"unshare(NEWUSER)", "unshare(NEWNS)", "BLOCKED", "WARNING"} + assert severities == { + "unshare(NEWUSER)", + "unshare(NEWNS)", + "BLOCKED", + "WARNING", + "FATAL", + } # And the one advisory is spelled the way the classifier looks for it. assert self._WARNING.startswith(wt_mod._SANDBOX_LAUNCHER_WARNING_PREFIX) assert wt_mod._SANDBOX_LAUNCHER_WARNING_PREFIX.startswith( wt_mod._SANDBOX_LAUNCHER_PREFIX ), "the advisory prefix must be a refinement of the launcher prefix, not a rival" + # FATAL is on the REFUSAL side, and deliberately so: the host-trust pre-read + # it reports from re-raises, so setup really does abort. Both halves are + # pinned, because each fails a different way -- without the launcher prefix + # the classifier would not see the line at all and the abort would surface as + # a bare git failure; with the ADVISORY prefix it would be downgraded to a + # warning and the caller would run on with host verification disarmed. + assert self._FATAL_PRE_READ.startswith(wt_mod._SANDBOX_LAUNCHER_PREFIX) + assert not self._FATAL_PRE_READ.startswith(wt_mod._SANDBOX_LAUNCHER_WARNING_PREFIX) + + def test_the_fail_closed_pre_read_line_still_refuses(self, tmp_path, monkeypatch): + """The OTHER fatal spelling must refuse too, not come back as data. + + ``_FATAL`` above is an ``unshare`` failure, so it shares no wording with this + one; a classifier keyed on anything narrower than the prefix pair would pass + that test and fail this one. Refusing is right because the pre-read re-raises: + the sandbox never came up, so running the git command bare would silently drop + the isolation the caller asked for. + """ + from kiro_crew.dashboard.handlers import worktree as wt + + stderr = self._FATAL_PRE_READ + "\n" + monkeypatch.setattr(wt, "run_limited", lambda *a, **k: self._completed(1, stderr)) + + with pytest.raises(SandboxUnavailable) as excinfo: + _run_git(["rev-parse", "HEAD"], str(tmp_path)) + + assert str(excinfo.value) == self._FATAL_PRE_READ def test_gits_own_error_is_never_mistaken_for_a_refusal(self, tmp_path, monkeypatch): from kiro_crew.dashboard.handlers import worktree as wt