diff --git a/src/plyngent/agent/__init__.py b/src/plyngent/agent/__init__.py index a115625..db6d2a7 100644 --- a/src/plyngent/agent/__init__.py +++ b/src/plyngent/agent/__init__.py @@ -30,6 +30,7 @@ from .notices import skill_tools_notice as skill_tools_notice from .notices import truncated_turn_notice as truncated_turn_notice from .tools import BatchConfirmDecision as BatchConfirmDecision from .tools import DangerClassifier as DangerClassifier +from .tools import DangerReason as DangerReason from .tools import ToolBatchConfirmHook as ToolBatchConfirmHook from .tools import ToolConfirmHook as ToolConfirmHook from .tools import ToolConfirmRequest as ToolConfirmRequest diff --git a/src/plyngent/agent/tools.py b/src/plyngent/agent/tools.py index 0ffdb2c..40fff2b 100644 --- a/src/plyngent/agent/tools.py +++ b/src/plyngent/agent/tools.py @@ -16,8 +16,22 @@ from msgspec import Struct from plyngent.lmproto.openai_compatible.model import ChatImagePart, ToolFunction, ToolFunctionItem from plyngent.typedef import JSONSchema # noqa: TC001 + +@dataclass(frozen=True, slots=True) +class DangerReason: + """Why a call needs a soft confirm, and whether YOLO mode may drop that prompt. + + ``yolo_skip`` is false for a call that reaches rights the session does not + have (``sudo`` / ``doas`` / ``pkexec`` and friends): the human answers it + however YOLO mode is set. + """ + + text: str + yolo_skip: bool = True + + type ToolHandler = Callable[..., Any | Awaitable[Any]] -type DangerClassifier = Callable[[str, Mapping[str, object]], str | None] +type DangerClassifier = Callable[[str, Mapping[str, object]], str | DangerReason | None] # Returns True to allow, False to deny, or a non-empty str as denial reason for the model. type ToolConfirmResult = bool | str type ToolConfirmHook = Callable[ @@ -313,6 +327,13 @@ def _denied_error(name: str, reason: str, comment: str | None) -> str: return f"error: tool {name!r} denied by user confirm ({reason})" +def _as_danger_reason(raw: str | DangerReason | None) -> DangerReason | None: + """Normalize a classifier answer; a bare string is a reason YOLO mode may drop.""" + if raw is None: + return None + return raw if isinstance(raw, DangerReason) else DangerReason(text=raw) + + def _write_path_key(value: str) -> str: """Comparable key for one path argument of a writing tool. @@ -559,7 +580,9 @@ class ToolRegistry: Mirrors the front half of :meth:`execute`: a call whose arguments fail to decode or miss a required argument is answered with an error there and - must not be put in front of the user now. + must not be put in front of the user now, and a call whose reason YOLO + mode covers runs unprompted, so it must not take a slot in the round's + aggregated prompt either. """ definition = self._tools.get(name) if self._danger is None or definition is None: @@ -567,8 +590,10 @@ class ToolRegistry: args = _decode_tool_args(arguments_json) if isinstance(args, str) or _missing_required_error(definition, args) is not None: return None - reason = self._danger(name, args) - return None if reason is None else ToolConfirmRequest(name=name, args=args, reason=reason) + reason = _as_danger_reason(self._danger(name, args)) + if reason is None or (self._yolo and (definition.tags & ToolTag.YOLO) and reason.yolo_skip): + return None + return ToolConfirmRequest(name=name, args=args, reason=reason.text) def _batch_requests( self, @@ -633,29 +658,32 @@ class ToolRegistry: tags: ToolTag, decision: BatchConfirmDecision | None = None, ) -> str | None: - """Soft-confirm gate driven by danger reason + tags + grants. + """Soft-confirm gate driven by danger reason + tags + grants + YOLO. Pipeline (soft gate only; hard denylists stay in tool handlers):: no soft reason → run + YOLO mode, (tags & YOLO) and a reason it may drop → allow batch decision for this call → apply it (tags & TRUSTABLE) and grant exists → allow - else → on_confirm (decorator YOLO tag does not override - dangerous invocations — internal danger classification - always takes precedence) + else → on_confirm (a tool without the YOLO tag, or a reason YOLO + may not drop — a privilege escalation — still prompts + under YOLO mode) """ if self._danger is None: return None - reason = self._danger(name, args) + reason = _as_danger_reason(self._danger(name, args)) if reason is None: return None + if self._yolo and (tags & ToolTag.YOLO) and reason.yolo_skip: + return None async with self._confirm_lock: if decision is not None: - return await self._apply_batch_decision(name, reason, decision, tags=tags) + return await self._apply_batch_decision(name, reason.text, decision, tags=tags) if self._grant_allows(name, tags=tags): return None - return await self._prompt_soft_confirm(name, args, reason, tags=tags) + return await self._prompt_soft_confirm(name, args, reason.text, tags=tags) async def execute( self, diff --git a/src/plyngent/tools/command_scan.py b/src/plyngent/tools/command_scan.py index c9cda28..0817f8d 100644 --- a/src/plyngent/tools/command_scan.py +++ b/src/plyngent/tools/command_scan.py @@ -16,7 +16,16 @@ from typing import TYPE_CHECKING if TYPE_CHECKING: from collections.abc import Sequence -__all__ = ["CommandString", "UnwrappedCommand", "basename", "is_interpreter", "unwrap_command"] +__all__ = [ + "ESCALATION_PROGRAMS", + "CommandString", + "UnwrappedCommand", + "basename", + "dash_c_code", + "escalators_in_chain", + "is_interpreter", + "unwrap_command", +] # Wrappers that run the program named on their own command line. _ENV_WRAPPERS: frozenset[str] = frozenset( @@ -99,6 +108,10 @@ _INTERPRETERS: frozenset[str] = frozenset( # Trailing version of an interpreter: ``python3.12`` / ``node20`` / ``php8.2``. _VERSION_SUFFIX = re.compile(r"[0-9][0-9.]*$") +# Privilege escalation: these run with rights the session itself does not have, +# so a call that reaches one is never approved on the caller's behalf +# (``danger._command_reason`` keeps the confirm whatever YOLO mode says). +ESCALATION_PROGRAMS: frozenset[str] = frozenset({"doas", "pkexec", "run0", "su", "sudo", "systemd-run"}) # ``FOO=bar`` in front of a program (env/sudo style). _ASSIGNMENT = re.compile(r"[A-Za-z_][A-Za-z0-9_]*=") @@ -245,6 +258,31 @@ def is_interpreter(base: str) -> bool: return base in _INTERPRETERS or _VERSION_SUFFIX.sub("", base) in _INTERPRETERS +def dash_c_code(command: Sequence[str]) -> tuple[int, str] | None: + """Index and text of the ``-c`` code in an interpreter argv, if present. + + Only interpreters hand ``-c`` a block of code; for anything else it is a + plain flag (``grep -c``, ``od -c``) with nothing to review. + """ + for index, part in enumerate(command[1:], start=1): + if part == "-c": + return index, command[index + 1] if index + 1 < len(command) else "" + # Combined short forms are uncommon; only exact -c is supported. + return None + + +def escalators_in_chain(argv: Sequence[str]) -> tuple[str, ...]: + """Escalation programs *argv* runs: the program itself or a wrapper around it. + + ``sudo -u root python …`` and ``doas bash`` both count, whatever program they + finally run. Code an interpreter only carries as text is out of scope here — + this reads the command line, not what the text would do. + """ + scan = unwrap_command(argv) + names = (scan.base, *scan.wrappers) + return tuple(dict.fromkeys(name for name in names if name in ESCALATION_PROGRAMS)) + + def unwrap_command(argv: Sequence[str]) -> UnwrappedCommand: """Resolve *argv* to the program behind any wrappers and ``run`` launchers. diff --git a/src/plyngent/tools/danger.py b/src/plyngent/tools/danger.py index e70947a..293538f 100644 --- a/src/plyngent/tools/danger.py +++ b/src/plyngent/tools/danger.py @@ -4,7 +4,8 @@ import json import shlex from typing import TYPE_CHECKING, cast -from plyngent.tools.command_scan import is_interpreter, unwrap_command +from plyngent.agent.tools import DangerReason +from plyngent.tools.command_scan import dash_c_code, escalators_in_chain, is_interpreter, unwrap_command from plyngent.tools.workspace import AccessMode, WorkspaceError, resolve_path if TYPE_CHECKING: @@ -23,19 +24,6 @@ def _as_argv(args: Mapping[str, object], *, key: str = "command") -> list[str] | return out -def _dash_c_code(command: Sequence[str]) -> tuple[int, str] | None: - """Index and text of the ``-c`` code in an interpreter argv, if present. - - Only interpreters hand ``-c`` a block of code; for anything else it is a - plain flag (``grep -c``, ``od -c``) with nothing to review. - """ - for index, part in enumerate(command[1:], start=1): - if part == "-c": - return index, command[index + 1] if index + 1 < len(command) else "" - # Combined short forms are uncommon; only exact -c is supported. - return None - - def _indent_block(text: str, *, prefix: str = " ") -> str: """Indent every line of *text* (for multi-line -c code in confirm prompts).""" if not text: @@ -51,7 +39,7 @@ def _format_argv_for_confirm(argv: Sequence[str], *, code_index: int | None) -> return shlex.join(parts) -def _command_reason(argv: Sequence[str], *, via: str) -> str | None: +def _command_reason(argv: Sequence[str], *, via: str) -> DangerReason | None: """Confirm a command that runs code or detaches, whatever wraps it. Multi-line reason (shown inside the CLI confirm box). ``via`` is a short @@ -65,6 +53,11 @@ def _command_reason(argv: Sequence[str], *, via: str) -> str | None: (``env -S "…"``, ``su -c "…"``) hides code the same way, so it is confirmed too, with the text it carries printed under its own ``