diff --git a/CLAUDE.md b/CLAUDE.md index 2647817..93414f9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -80,8 +80,8 @@ Module-level `@tool` handlers. Call `set_workspace_root()` before use. - **`workspace`**: path resolve under primary root **or** temporary allowlist roots; path substring denylist; command basename denylist; clearer deny messages. - **`file`**: `read_file` (`with_lineno`), `write_file`, `listdir`, `tree` (VCS + default noise dirs + optional `skip_dirs` / denylist walk), `glob_paths`, `grep_files` (regex, skip VCS/binary), `edit_replace` (`max_replaces`, reports remaining matches), `edit_lineno` (1-based range), `copy_path` / `move_path` / `delete_path`, `new_temporary_workspace` (system temp allowlist; cleanup on chat exit; no session rebind). -- **`process`**: `run_command` (argv, no shell, timeout, optional stdin/env); `run_command_batch` (serial steps, `pipe_out` on provider, `mix_stderr`, `stop_on_error`); PTY `open_pty` / `read_pty` / `write_pty` (literal) / `write_pty_keys` (escapes) / `close_pty` (POSIX: `pty`+`fork`; Windows: ConPTY via `pywinpty` env marker dep). -- PTY: backend in `pty_backend.py`; structured status (`alive`/`exit_code`/`data`); `read_pty(..., until=)` (**CSI sanitized** so host TTY is never reset); keys via `write_pty_keys` only (`\xHH`, `ctrl+x`, `key=esc`); session limit/idle TTL/output budget; close terminate→kill. +- **`process`**: `run_command` (argv, no shell, timeout, optional stdin/env); `run_command_batch` (serial steps, `pipe_out` on provider, `mix_stderr`, `stop_on_error`); PTY `open_pty` / `read_pty` / `write_pty` (literal) / `write_pty_keys` (escapes) / `ask_into_pty` (human→PTY only; `secret` no-echo; answer never in tool result) / `close_pty` (POSIX: `pty`+`fork`; Windows: ConPTY via `pywinpty` env marker dep). +- PTY: backend in `pty_backend.py`; structured status (`alive`/`exit_code`/`data`); `read_pty(..., until=)` (**CSI sanitized** so host TTY is never reset); keys via `write_pty_keys` only (`\xHH`, `ctrl+x`, `key=esc`); secrets via `ask_into_pty` (not `write_pty` data); session limit/idle TTL/output budget; close terminate→kill. - CLI limit hooks: interactive confirm to raise tool-loop rounds, PTY session cap, or PTY output budget. - Destructive confirms: `classify_danger` + `ToolRegistry(on_confirm=…)`; CLI default deny; config `confirm_destructive` / `path_denylist`. Soft confirm also for shells/REPLs and `python|bash -c` one-liners (argv + code preview); deny may include a user comment for the model. Session YOLO: `/yolo on|off|once` and `--yes` (skip soft confirms; hard denylists unchanged; `once` expires after the next user turn). - **`vcs`**: read-only VCS tools (`vcs_kind` / `vcs_status` / `vcs_diff` / `vcs_log` / `vcs_branch`) via `VcsBackend` protocol; **git** implemented; detectors are pluggable for other systems. diff --git a/README.md b/README.md index 6a74c2a..29407b5 100644 --- a/README.md +++ b/README.md @@ -235,6 +235,7 @@ Safety defaults: Prefer file tools over full-screen editors (`vim`/`nano`) for edits. `read_pty` sanitizes CSI/controls so tool results cannot reprogram the host TTY (no host terminal reset on exit). `write_pty` is literal text only; use `write_pty_keys` for `\xHH`, `ctrl+x`, `key=esc|enter|…`. + For passwords/sudo/ssh prompts use `ask_into_pty` (human types locally; answer never returns to the model). ## Usage / context (CLI) diff --git a/src/plyngent/prompting.py b/src/plyngent/prompting.py index de53fe4..9cac2fb 100644 --- a/src/plyngent/prompting.py +++ b/src/plyngent/prompting.py @@ -55,6 +55,10 @@ class PromptBackend(Protocol): completions: Sequence[str] | None = None, ) -> str: ... + def read_secret_line(self, prompt: str) -> str: + """Read a line without echo (passwords). Cancel raises NonInteractiveError.""" + ... + def confirm(self, prompt: str, *, default: bool = False) -> bool: ... def echo(self, message: str = "", *, err: bool = False) -> None: ... @@ -117,6 +121,16 @@ class ClickPromptBackend: return default return raw + def read_secret_line(self, prompt: str) -> str: + import getpass + + try: + display = f"{prompt}: " if not prompt.endswith(": ") else prompt + return getpass.getpass(display) + except (KeyboardInterrupt, EOFError) as exc: + msg = "prompt cancelled" + raise NonInteractiveError(msg) from exc + def confirm(self, prompt: str, *, default: bool = False) -> bool: try: return bool(click.confirm(prompt, default=default)) @@ -150,6 +164,10 @@ class NonInteractiveBackend: msg = f"non-interactive: cannot prompt for {prompt!r}" raise NonInteractiveError(msg) + def read_secret_line(self, prompt: str) -> str: + msg = f"non-interactive: cannot prompt for secret {prompt!r}" + raise NonInteractiveError(msg) + def confirm(self, prompt: str, *, default: bool = False) -> bool: del prompt return default @@ -255,6 +273,19 @@ def ask( return backend.read_line("Answer", default=default, completions=completions).strip() +def ask_secret(prompt: str) -> str: + """Prompt for a secret line (no echo). Never use for values that return to the model. + + Cancel (Ctrl+C / EOF) raises :class:`NonInteractiveError`. + """ + backend = get_prompt_backend() + if not backend.is_interactive(): + msg = f"non-interactive: cannot prompt for secret {prompt!r}" + raise NonInteractiveError(msg) + backend.secho(prompt, fg="yellow") + return backend.read_secret_line("Secret") + + def choose( prompt: str, options: Sequence[str] | Sequence[ChoiceOption], @@ -363,6 +394,10 @@ async def ask_async(prompt: str, *, default: str | None = None) -> str: return await run_prompt_async(ask, prompt, default=default) +async def ask_secret_async(prompt: str) -> str: + return await run_prompt_async(ask_secret, prompt) + + async def choose_async( prompt: str, options: Sequence[str] | Sequence[ChoiceOption], diff --git a/src/plyngent/tools/__init__.py b/src/plyngent/tools/__init__.py index 1858009..609436d 100644 --- a/src/plyngent/tools/__init__.py +++ b/src/plyngent/tools/__init__.py @@ -16,6 +16,7 @@ from .file import read_file as read_file from .file import tree as tree from .file import write_file as write_file from .process import PROCESS_TOOLS as PROCESS_TOOLS +from .process import ask_into_pty as ask_into_pty from .process import close_pty as close_pty from .process import open_pty as open_pty from .process import read_pty as read_pty diff --git a/src/plyngent/tools/process/__init__.py b/src/plyngent/tools/process/__init__.py index 35988a7..faa85d4 100644 --- a/src/plyngent/tools/process/__init__.py +++ b/src/plyngent/tools/process/__init__.py @@ -1,3 +1,4 @@ +from .ask_into_pty import ask_into_pty as ask_into_pty from .close_pty import close_pty as close_pty from .open_pty import open_pty as open_pty from .pty_terminal import decode_write_data as decode_write_data @@ -15,5 +16,6 @@ PROCESS_TOOLS = [ read_pty, write_pty, write_pty_keys, + ask_into_pty, close_pty, ] diff --git a/src/plyngent/tools/process/ask_into_pty.py b/src/plyngent/tools/process/ask_into_pty.py new file mode 100644 index 0000000..ccfe9f3 --- /dev/null +++ b/src/plyngent/tools/process/ask_into_pty.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +from typing import Literal + +from plyngent.agent import tool +from plyngent.prompting import NonInteractiveError, ask_async, ask_secret_async +from plyngent.tools.workspace import WorkspaceError + +from .pty_session import PtyManager +from .write_pty import write_pty_payload + +type _PromptResult = tuple[Literal["ok"], str] | tuple[Literal["err"], str] + + +def _status_without_payload( + status: str, + *, + secret: bool, + submit: bool, +) -> str: + lines = [line for line in status.splitlines() if not line.startswith("wrote=")] + lines.append("wrote=true") + lines.append(f"secret={'true' if secret else 'false'}") + lines.append(f"submit={'true' if submit else 'false'}") + lines.append("source=human") + return "\n".join(lines) + + +async def _prompt_answer(label: str, *, secret: bool) -> _PromptResult: + try: + answer = await ask_secret_async(label) if secret else await ask_async(label) + except NonInteractiveError as exc: + return ("err", f"error: {exc}") + except KeyboardInterrupt, EOFError: + return ("err", "error: prompt cancelled") + if answer == "": + return ("err", "error: empty input cancelled (nothing written to PTY)") + return ("ok", answer) + + +@tool +async def ask_into_pty( + session_id: int, + message: str, + *, + secret: bool = False, + submit: bool = True, +) -> str: + """Prompt the **human** and write their answer into a PTY (local only). + + Use for interactive prompts (e.g. sudo/ssh password). The answer is written + to the PTY master on this machine and is **never** included in the tool + result (so it is not sent to external model APIs). + + ``message`` is shown to the human (what to enter), not the secret itself. + ``secret=true`` uses no-echo input. ``submit=true`` (default) appends a + newline after the answer. Empty input and Ctrl+C cancel without writing. + """ + label = message.strip() or ("Secret" if secret else "Input") + try: + # Validate session before blocking the human. + _ = PtyManager.refresh(session_id) + except WorkspaceError as exc: + return f"error: {exc}" + + kind, value = await _prompt_answer(label, secret=secret) + if kind == "err": + return value + + payload = value + ("\n" if submit else "") + try: + status = write_pty_payload(session_id, payload) + except WorkspaceError as exc: + return f"error: {exc}" + except OSError as exc: + return f"error: failed to write PTY: {exc}" + + return _status_without_payload(status, secret=secret, submit=submit) diff --git a/tests/test_prompting.py b/tests/test_prompting.py index 14e3d00..b5886e3 100644 --- a/tests/test_prompting.py +++ b/tests/test_prompting.py @@ -18,9 +18,16 @@ from plyngent.prompting import ( class ScriptedBackend: """Deterministic backend for unit tests.""" - def __init__(self, lines: list[str], *, confirms: list[bool] | None = None) -> None: + def __init__( + self, + lines: list[str], + *, + confirms: list[bool] | None = None, + secrets: list[str] | None = None, + ) -> None: self.lines = list(lines) self.confirms = list(confirms or []) + self.secrets = list(secrets or []) self.interactive = True self.echoes: list[str] = [] @@ -42,6 +49,13 @@ class ScriptedBackend: msg = "no scripted lines left" raise NonInteractiveError(msg) + def read_secret_line(self, prompt: str) -> str: + del prompt + if self.secrets: + return self.secrets.pop(0) + msg = "no scripted secrets left" + raise NonInteractiveError(msg) + def confirm(self, prompt: str, *, default: bool = False) -> bool: del prompt if self.confirms: @@ -63,6 +77,14 @@ def test_ask_free_text() -> None: assert ask("Name?") == "hello world" +def test_ask_secret() -> None: + from plyngent.prompting import ask_secret + + backend = ScriptedBackend([], secrets=["s3cret"]) + with temporary_backend(backend): + assert ask_secret("Password?") == "s3cret" + + def test_ask_default_when_empty_uses_backend_default() -> None: backend = ScriptedBackend([]) with temporary_backend(backend): diff --git a/tests/test_tools/test_process.py b/tests/test_tools/test_process.py index 005de38..b1f1faa 100644 --- a/tests/test_tools/test_process.py +++ b/tests/test_tools/test_process.py @@ -5,6 +5,7 @@ import sys from pathlib import Path from plyngent.tools.process import ( + ask_into_pty, close_pty, open_pty, read_pty, @@ -248,6 +249,92 @@ def test_write_pty_unknown_session(workspace: object) -> None: assert "error" in call_sync(write_pty, 999_999, "x") +async def test_ask_into_pty_writes_without_leaking_secret(workspace: object) -> None: + """Human answer goes to PTY only; tool result must not contain the secret.""" + del workspace + from plyngent.prompting import temporary_backend + from tests.test_prompting import ScriptedBackend + + secret = "super-secret-password-xyz" + backend = ScriptedBackend([], secrets=[secret]) + try: + opened = call_sync( + open_pty, + _py( + "import sys\n" + "while True:\n" + " line = sys.stdin.readline()\n" + " if not line:\n" + " break\n" + " sys.stdout.write(line)\n" + " sys.stdout.flush()\n" + ), + ) + session_id = _session_id(opened) + with temporary_backend(backend): + result = await call_async( + ask_into_pty, + session_id, + "Enter password", + secret=True, + ) + assert "error:" not in result + assert "wrote=true" in result + assert "secret=true" in result + assert "source=human" in result + assert secret not in result + # Length of secret must not appear as wrote=N either. + assert f"wrote={len(secret)}" not in result + assert f"wrote={len(secret) + 1}" not in result + + echoed = await call_async(read_pty, session_id, timeout=2.0, until=secret) + assert secret in echoed + closed = await call_async(close_pty, session_id) + assert "session_id=" in closed + finally: + PtyManager.close_all() + + +async def test_ask_into_pty_empty_cancels(workspace: object) -> None: + del workspace + from plyngent.prompting import temporary_backend + from tests.test_prompting import ScriptedBackend + + backend = ScriptedBackend([], secrets=[""]) + try: + opened = call_sync(open_pty, _py("import time; time.sleep(30)")) + session_id = _session_id(opened) + with temporary_backend(backend): + result = await call_async(ask_into_pty, session_id, "Pass?", secret=True) + assert "error:" in result + assert "empty" in result.lower() or "cancel" in result.lower() + finally: + PtyManager.close_all() + + +async def test_ask_into_pty_nonsecret_line(workspace: object) -> None: + del workspace + from plyngent.prompting import temporary_backend + from tests.test_prompting import ScriptedBackend + + backend = ScriptedBackend(["yes"]) + try: + opened = call_sync( + open_pty, + _py("import sys\nline = sys.stdin.readline()\nsys.stdout.write(line)\nsys.stdout.flush()\n"), + ) + session_id = _session_id(opened) + with temporary_backend(backend): + result = await call_async(ask_into_pty, session_id, "Continue?", secret=False) + assert "wrote=true" in result + assert "secret=false" in result + assert "yes" not in result # answer not in tool result even for non-secret + text = await call_async(read_pty, session_id, timeout=2.0, until="yes") + assert "yes" in text + finally: + PtyManager.close_all() + + async def test_pty_exec_failure_surfaces(workspace: object) -> None: del workspace try: