langchain-ai/deepagents
1
0
Fork 0
mirror of https://github.com/langchain-ai/deepagents.git synced 2026-07-22 01:35:28 -04:00
Code Issues 169 Packages Projects Releases 16 Wiki Activity

feat(code): in-the-moment trust prompt for symlinked skills (#4200)

Browse Source 
Skills whose `SKILL.md` resolves outside trusted directories (e.g. via a
symlink) can now be approved in the moment via a trust prompt, instead
of requiring a quit-and-relaunch to edit config. Approved directories
are persisted to `~/.deepagents/.state/skill_trust.json` and can be
audited or revoked with `dcode skills trust list|revoke|clear`.

---

Today, invoking a skill whose `SKILL.md` resolves (via symlink) outside
every trusted skill root fails hard with a `PermissionError`, and the
only way to proceed is to quit, set `DEEPAGENTS_CODE_EXTRA_SKILLS_DIRS`
/ `[skills].extra_allowed_dirs`, and relaunch. This adds an
in-the-moment approval flow modeled on the existing project-level MCP
trust screen.

When containment fails, the TUI shows a `SkillTrustScreen` naming the
skill and the resolved target directory. Allowing reads the skill
immediately, persists the resolved target directory to a new trust store
at `~/.deepagents/.state/skill_trust.json` (managed by
`skills/trust.py`, mirroring `mcp_trust.py`), and extends the in-session
containment allowlist. Denying is safe and is remembered for the rest of
the session so the user is not re-nagged. The non-interactive (`-x`)
path keeps the existing hard-fail + hint.

Trust is keyed by the resolved target directory — the canonical path
shown to the user at approval time, stored as-is and never re-resolved.
Two distinct post-approval symlink swaps are caught by two distinct
layers, so neither grants access the user never approved:

- Re-pointing the *discovery* symlink (the `SKILL.md` path) at a new
target is caught by containment enforcement in `load_skill_content`: the
new target is not on the allowlist, so the read is refused and the user
is re-prompted. The stored trust entry — the original resolved target —
is untouched.
- Replacing the *stored* directory itself (or one of its parents) with a
symlink is caught by the `resolve()`-to-self re-verification in
`load_trusted_skill_dirs`, which drops the stale entry rather than
following the injected symlink to a directory the user never approved.

Persisted approvals join the containment roots in
`discover_skills_and_roots`, so they apply on subsequent launches.
`skills trust list|revoke|clear` lets users audit and revoke approved
directories; the declarative `extra_allowed_dirs` / env allowlist still
works unchanged as the permanent path.

---------

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
This commit is contained in:
Mason Daugherty
2026-07-06 10:17:58 -04:00
committed by GitHub
parent d73f36ebae
commit a4431e4339
12 changed files with 2402 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/code/deepagents_code/app.py
+154 -2
View File
@@ -279,6 +279,11 @@ def _create_model_with_deepagents_import_lock(
)
def _resolve_parent_dir(path: str | Path) -> str:
"""Return the resolved parent directory for a path."""
return str(Path(path).resolve().parent)
def _extra_is_ready(extra: str) -> bool | None:
"""Return whether all dependencies for `extra` are installed.
@@ -2786,6 +2791,13 @@ class DeepAgentsApp(App):
Built alongside `_discovered_skills`.
"""
self._skill_trust_denied: set[str] = set()
"""Resolved skill directories the user declined to trust this session.
Prevents re-prompting for the same untrusted location after a deny
within a single run.
"""
# Media
# Lazily imported here to avoid pulling image dependencies into
# argument parsing paths.
@@ -9791,8 +9803,15 @@ class DeepAgentsApp(App):
normalized_name,
exc_info=True,
)
await _mount_error(str(exc))
return
content = await self._prompt_skill_trust_and_retry(
normalized_name,
skill_path,
allowed_roots,
fallback_error=str(exc),
mount_error=_mount_error,
)
if content is None:
return
except OSError as exc:
logger.warning(
"Filesystem error loading skill %r",
@@ -9842,6 +9861,139 @@ class DeepAgentsApp(App):
message_kwargs=envelope.message_kwargs,
)
async def _prompt_skill_trust_and_retry(
self,
skill_name: str,
skill_path: str | Path,
allowed_roots: list[Path],
*,
fallback_error: str,
mount_error: Callable[[str], Awaitable[None]],
) -> str | None:
"""Prompt to trust an out-of-bounds skill directory, then reload it.
Mirrors the MCP project-trust flow: when containment fails, the user is
asked once to allow the resolved target directory. Allowing persists the
decision, extends the in-session containment allowlist, and retries the
read. Denying (or a prior deny this session) shows the original error.
Args:
skill_name: Normalized skill name, for messaging.
skill_path: Path to the skill's `SKILL.md`.
allowed_roots: Containment roots to extend on approval.
fallback_error: Original `PermissionError` message to show on deny.
mount_error: Callback to surface an error in the chat log.
Returns:
The skill content on approval and successful reload, or `None` when
the user declined or the retry failed (an error was mounted).
"""
from deepagents_code.skills.load import load_skill_content
from deepagents_code.skills.trust import trust_skill_dir
from deepagents_code.widgets.skill_trust import SkillTrustScreen
try:
target_dir = await asyncio.to_thread(_resolve_parent_dir, skill_path)
except (OSError, RuntimeError):
# Resolving the skill path can fail (e.g. a symlink loop introduced
# in the window after the first containment check raised). Fail
# closed with the original error rather than letting the worker
# exception escape unhandled — this mirrors the retry block below,
# which already guards its own resolve.
logger.warning(
"Could not resolve skill path %r for the trust prompt; refusing",
skill_path,
exc_info=True,
)
await mount_error(fallback_error)
return None
if target_dir in self._skill_trust_denied:
await mount_error(fallback_error)
return None
try:
allowed = await self._push_screen_wait(
SkillTrustScreen(skill_name, target_dir)
)
except Exception:
# This runs inside `_invoke_skill`'s `except PermissionError` block,
# so an error escaping the modal push would not be caught by that
# method's sibling handlers and would surface as an unhandled worker
# crash. Fail closed: treat a modal failure as a deny.
logger.warning(
"Skill trust prompt failed to display for %s; refusing",
target_dir,
exc_info=True,
)
self._skill_trust_denied.add(target_dir)
await mount_error(fallback_error)
return None
if not allowed:
self._skill_trust_denied.add(target_dir)
await mount_error(fallback_error)
return None
if not await asyncio.to_thread(trust_skill_dir, target_dir):
# The modal told the user this location would be remembered for
# future sessions. Persisting failed (read-only/full `.state`), so
# surface that: we honor the approval this session, but they will be
# asked again next launch. Logging alone would silently break that
# promise.
logger.warning("Could not persist skill trust for %s", target_dir)
self.notify(
"Approved for this session, but the decision could not be "
"saved — you may be asked again next time.",
severity="warning",
markup=False,
)
target_path = Path(target_dir)
for roots in (allowed_roots, self._skill_allowed_roots):
if target_path not in roots:
roots.append(target_path)
def _retry() -> str | None:
return load_skill_content(str(skill_path), allowed_roots=allowed_roots)
try:
content = await asyncio.to_thread(_retry)
except PermissionError:
# Containment failed *after* the target dir was allowlisted, so the
# skill path must now resolve somewhere else — a symlink swap during
# the prompt window. Refuse and flag it distinctly from a plain read
# error. (PermissionError is an OSError subclass, so this must come
# first.)
logger.warning(
"Skill %r resolved outside the approved directory on retry "
"(target may have changed since approval); refusing",
skill_name,
exc_info=True,
)
await mount_error(
f"Could not load skill: {skill_name}. Its location changed "
"after you approved it, so it was not read.",
)
return None
except OSError:
logger.warning(
"Retry load failed for skill %r after trust",
skill_name,
exc_info=True,
)
await mount_error(
f"Could not load skill: {skill_name} after granting trust.",
)
return None
if content is None:
await mount_error(
f"Could not read content for skill: {skill_name}. "
"Check that the SKILL.md file exists, is readable, "
"and is saved as UTF-8.",
)
return content
async def _handle_skill_command(self, command: str) -> None:
"""Handle a `/skill:<name>` command by loading and invoking a skill.
libs/code/deepagents_code/non_interactive.py
+6 -1
View File
@@ -1441,7 +1441,12 @@ async def run_non_interactive(
normalized_skill = initial_skill.strip().lower()
try:
skills, allowed_roots = discover_skills_and_roots(assistant_id)
# Offloaded to a thread: discovery does blocking filesystem I/O
# (a JSON trust-store read plus `Path.resolve()` calls) that must
# not block the event loop.
skills, allowed_roots = await asyncio.to_thread(
discover_skills_and_roots, assistant_id
)
skill = next((s for s in skills if s["name"] == normalized_skill), None)
except OSError as e:
console.print(
libs/code/deepagents_code/skills/commands.py
+172 -2
View File
@@ -5,7 +5,7 @@ from __future__ import annotations
import argparse
import shutil
from pathlib import Path
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Any, assert_never
if TYPE_CHECKING:
from collections.abc import Callable
@@ -853,6 +853,131 @@ def _delete(
)
def _trust(args: argparse.Namespace) -> None:
"""Handle `skills trust list|revoke|clear`.
Args:
args: Parsed arguments with a `trust_command` attribute.
Raises:
SystemExit: If the trust store cannot be read, or trust entries cannot
be revoked or cleared.
"""
from rich.markup import escape
from deepagents_code.config import console, get_glyphs
from deepagents_code.skills.trust import (
RevokeResult,
clear_trusted_skill_dirs,
list_trusted_skill_dir_entries,
revoke_skill_dir_trust,
)
command = getattr(args, "trust_command", None)
output_format = getattr(args, "output_format", "text")
checkmark = get_glyphs().checkmark
if command in {"list", "ls"}:
# Read strictly so an unreadable store surfaces as an error instead of
# falsely reporting "No trusted skill directories" — the whole point of
# the audit command is to show what is trusted so it can be revoked.
try:
entries = list_trusted_skill_dir_entries(strict=True)
except (OSError, ValueError) as exc:
console.print(
f"[bold red]Error:[/bold red] Could not read the skill trust "
f"store: {escape(str(exc))}"
)
raise SystemExit(1) from exc
if output_format == "json":
from deepagents_code.output import write_json
write_json(
"skills trust list",
[
{"dir": path, "trusted_at": trusted_at}
for path, trusted_at in entries
],
)
return
if not entries:
console.print()
console.print("[yellow]No trusted skill directories.[/yellow]")
console.print(
"[dim]Directories are trusted when you approve a skill that "
"resolves outside the standard skill roots.[/dim]",
style=theme.MUTED,
)
console.print()
return
console.print(
"\n[bold]Trusted skill directories:[/bold]\n", style=theme.PRIMARY
)
for path, trusted_at in entries:
console.print(f" {escape(str(path))}")
if trusted_at:
console.print(
f" [dim]trusted {escape(trusted_at)}[/dim]", style=theme.MUTED
)
console.print()
elif command == "revoke":
target = args.dir
result = revoke_skill_dir_trust(target)
# An I/O/read failure is a hard error regardless of output format
# (matching `list`): print red and exit non-zero without emitting a
# success envelope a script might misread.
if result is RevokeResult.ERROR:
console.print(
"[bold red]Error:[/bold red] Could not revoke trust for: "
f"{escape(str(target))}"
)
raise SystemExit(1)
if output_format == "json":
from deepagents_code.output import write_json
write_json(
"skills trust revoke",
{"dir": str(target), "result": result.value},
)
return
# `ERROR` was handled above (early exit), so only `REMOVED`/`NOT_FOUND`
# remain. Match exhaustively with `assert_never` so adding a future
# `RevokeResult` member is a static error here rather than a silent
# success that prints nothing yet exits 0.
match result:
case RevokeResult.REMOVED:
console.print(
f"{checkmark} Revoked trust for: {escape(str(target))}",
style=theme.PRIMARY,
)
case RevokeResult.NOT_FOUND:
# Report honestly, not a false success.
console.print(
f"[yellow]No trust entry found for:[/yellow] {escape(str(target))}"
)
case _: # pragma: no cover - exhaustiveness guard
assert_never(result)
elif command == "clear":
if not clear_trusted_skill_dirs():
console.print(
"[bold red]Error:[/bold red] Could not clear trusted directories."
)
raise SystemExit(1)
if output_format == "json":
from deepagents_code.output import write_json
write_json("skills trust clear", {"cleared": True})
return
console.print(
f"{checkmark} Cleared all trusted skill directories.",
style=theme.PRIMARY,
)
else:
from deepagents_code.ui import show_skills_trust_help
show_skills_trust_help()
def setup_skills_parser(
subparsers: Any, # noqa: ANN401 # argparse subparsers uses dynamic typing
*,
@@ -1012,6 +1137,45 @@ def setup_skills_parser(
action="store_true",
help="Show what would happen without making changes",
)
# Skills trust — manage directories approved to be read outside the
# standard skill roots (the persistent counterpart to the in-TUI prompt).
trust_parser = skills_subparsers.add_parser(
"trust",
help="Manage trusted skill directories",
description=(
"List, revoke, or clear skill directories that have been trusted "
"to be read even though they resolve outside the standard skill "
"roots (for example, symlink targets approved at invocation time)."
),
add_help=False,
parents=help_parent(_lazy_help("show_skills_trust_help")),
)
if add_output_args is not None:
add_output_args(trust_parser)
trust_subparsers = trust_parser.add_subparsers(
dest="trust_command", help="Trust command"
)
trust_list_parser = trust_subparsers.add_parser(
"list",
aliases=["ls"],
help="List trusted skill directories",
)
if add_output_args is not None:
add_output_args(trust_list_parser)
revoke_parser = trust_subparsers.add_parser(
"revoke",
help="Revoke trust for a directory",
)
revoke_parser.add_argument("dir", help="Directory path to revoke")
if add_output_args is not None:
add_output_args(revoke_parser)
clear_parser = trust_subparsers.add_parser(
"clear",
help="Remove all trusted skill directories",
)
if add_output_args is not None:
add_output_args(clear_parser)
return skills_parser
@@ -1026,8 +1190,14 @@ def execute_skills_command(args: argparse.Namespace) -> None:
"""
from deepagents_code.config import console
# The `trust` subcommand manages directory paths, not agent-scoped skills,
# so it has no `--agent` and is dispatched before agent validation.
if args.skills_command == "trust":
_trust(args)
return
# validate agent argument
if args.agent:
if getattr(args, "agent", None):
is_valid, error_msg = _validate_name(args.agent)
if not is_valid:
console.print(
libs/code/deepagents_code/skills/invocation.py
+9
View File
@@ -38,6 +38,7 @@ def discover_skills_and_roots(
"""
from deepagents_code.config import settings
from deepagents_code.skills.load import list_skills
from deepagents_code.skills.trust import load_trusted_skill_dirs
skills = list_skills(
built_in_skills_dir=settings.get_built_in_skills_dir(),
@@ -62,6 +63,14 @@ def discover_skills_and_roots(
if path is not None
]
roots.extend(path.resolve() for path in settings.get_extra_skills_dirs())
# Persisted in-the-moment approvals extend the containment allowlist just
# like the declarative `extra_allowed_dirs`, but are managed by the trust
# store rather than hand-edited config. These entries are already the
# canonical approved directories and are verified against post-approval
# symlink swaps by `load_trusted_skill_dirs`, so they are added as-is
# rather than re-resolved (re-resolving would follow an injected symlink to
# a directory the user never approved).
roots.extend(load_trusted_skill_dirs())
return skills, roots
libs/code/deepagents_code/skills/trust.py
+547
View File
@@ -0,0 +1,547 @@
"""Trust store for skill directories that resolve outside trusted roots.
`load_skill_content` refuses to read a `SKILL.md` whose resolved path falls
outside every trusted skill root — this stops a symlink inside a skill
directory from reading arbitrary files. The static escape hatch is the
`DEEPAGENTS_CODE_EXTRA_SKILLS_DIRS` env var / `[skills].extra_allowed_dirs`
config allowlist.
This module adds an in-the-moment, persistent approval path (mirroring
`deepagents_code.mcp_trust`): when a skill resolves outside the trusted roots,
the user is asked once to allow the resolved target directory, and the decision
is remembered. Trust is keyed by the approved target directory — the canonical
path resolved and shown to the user at approval time, stored as-is and never
re-resolved.
Two distinct post-approval swaps are caught by two distinct layers, so neither
grants access the user never approved:
* Re-pointing the *discovery* symlink (the `SKILL.md` path) at a new target is
caught by containment enforcement in `load_skill_content`: the new target is
not on the allowlist, so the read is refused and the user is re-prompted.
The stored trust entry — the original resolved target — is untouched.
* Replacing the *stored* directory itself (or one of its parents) with a symlink
is caught by the `resolve()`-to-self re-verification in
`load_trusted_skill_dirs`, which drops the stale entry rather than following
the injected symlink to a directory the user never approved.
Trust entries are app-managed bookkeeping (a set of approved directories), not
user-facing configuration, so they live alongside the other state files under
`~/.deepagents/.state/skill_trust.json` rather than in the hand-editable
`config.toml`.
"""
from __future__ import annotations
import contextlib
import json
import logging
import os
import tempfile
from datetime import UTC, datetime
from enum import Enum
from pathlib import Path
from typing import TYPE_CHECKING, Any, TypedDict
if TYPE_CHECKING:
from collections.abc import Mapping
logger = logging.getLogger(__name__)
_STORAGE_VERSION = 1
"""Schema version stamped into `skill_trust.json`; bump on incompatible changes."""
class _TrustEntry(TypedDict):
"""One trusted-directory record in the store's `dirs` map."""
trusted_at: str
"""ISO-8601 UTC timestamp of when the directory was approved."""
class _TrustStore(TypedDict):
"""On-disk shape of `skill_trust.json`."""
version: int
"""Schema version written to the store file."""
dirs: dict[str, _TrustEntry]
"""Trusted directories keyed by their approved absolute path."""
class RevokeResult(Enum):
"""Outcome of a `revoke_skill_dir_trust` call.
Distinguishing `NOT_FOUND` from `REMOVED` lets the CLI print an honest
message instead of a false success when the target was never trusted (a
plain bool collapsed the two).
"""
REMOVED = "removed"
"""An entry existed and was removed from the store."""
NOT_FOUND = "not_found"
"""No matching entry existed; the store was left unchanged."""
ERROR = "error"
"""The store could not be read or the removal could not be persisted."""
def _default_store_path() -> Path:
"""Return `~/.deepagents/.state/skill_trust.json`.
Resolved at call time (not import time) so tests can redirect storage by
monkeypatching `deepagents_code.model_config.DEFAULT_STATE_DIR` — the same
pattern `deepagents_code.mcp_trust._default_store_path` uses.
"""
from deepagents_code.model_config import DEFAULT_STATE_DIR
return DEFAULT_STATE_DIR / "skill_trust.json"
def _normalize(target_dir: Path | str) -> str:
"""Return the resolved absolute string form of a directory key."""
return str(Path(target_dir).expanduser().resolve())
def _approved_key(target_dir: Path | str) -> str:
"""Return the already-approved directory key without resolving again."""
return str(Path(target_dir).expanduser())
def _load_store(store_path: Path, *, strict: bool = False) -> dict[str, Any]:
"""Read the JSON trust store file.
Args:
store_path: Path to the trust store file.
strict: When `True`, a store that exists but cannot be read or parsed
re-raises instead of degrading to `{}`. Read/modify/write callers
pass `strict=True` so a transient read error aborts the write
rather than silently rebuilding the store from an empty dict (which
would clobber every prior approval). The audit path passes it too so
it can report an unreadable store instead of claiming nothing is
trusted. Enforcement callers leave it `False` to stay fail-closed.
Returns:
Parsed JSON data, or an empty dict when the file is missing, or (only
when `strict` is `False`) when it is unreadable or corrupt.
A corrupt store degrades to "nothing trusted" so a bad file can't
crash startup. It is *not* self-healed on the next approval:
ordinary writes read with `strict=True` and refuse rather than
clobber a store they can't parse, so recovery from a corrupt file
requires `skills trust clear` (or `clear_trusted_skill_dirs`,
the only writer that overwrites blindly).
Raises:
OSError: When `strict` and an existing store cannot be read.
json.JSONDecodeError: When `strict` and an existing store is not valid
JSON.
ValueError: When `strict` and the store's top-level value is not a JSON
object, or its `version` is unrecognized (non-integer, or newer than
this build understands).
"""
# A missing store is a normal first-run state, never an error — return
# empty even under `strict` so callers don't have to special-case it.
if not store_path.exists():
return {}
try:
data = json.loads(store_path.read_text(encoding="utf-8"))
except json.JSONDecodeError as exc:
if strict:
raise
# A corrupt store silently drops every prior approval and forces a
# re-prompt, so log at WARNING (not DEBUG) to leave a breadcrumb for
# the otherwise-unexplained re-prompt.
logger.warning(
"Skill trust store %s is corrupt; treating as empty: %s", store_path, exc
)
return {}
except OSError as exc:
if strict:
raise
logger.warning(
"Could not read skill trust store %s; treating as empty: %s",
store_path,
exc,
)
return {}
if not isinstance(data, dict):
if strict:
msg = f"Skill trust store {store_path} is not a JSON object"
raise ValueError(msg)
logger.warning(
"Skill trust store %s is not a JSON object; ignoring", store_path
)
return {}
# A store written by a newer build may carry an incompatible schema. Reading
# its `dirs` regardless could misinterpret entries, so refuse: fail-closed
# (treat as nothing trusted) for enforcement, and surface the error for the
# audit path. A present-but-non-integer `version` is unrecognized in the same
# way (only tampering or a corrupt write produces it, since every writer
# stamps an int), so it is refused too rather than falling through and
# trusting `dirs`. A missing `version` stays tolerated: an empty `{}` file
# has no `dirs` to trust anyway. Together this makes the `_STORAGE_VERSION`
# "bump on incompatible changes" contract enforceable rather than
# aspirational.
version = data.get("version")
if version is not None and (
not isinstance(version, int) or version > _STORAGE_VERSION
):
if strict:
msg = (
f"Skill trust store {store_path} has an unrecognized schema "
f"version {version!r} (this build understands <= {_STORAGE_VERSION}); "
f"refusing to read it"
)
raise ValueError(msg)
logger.warning(
"Skill trust store %s has an unrecognized schema version %r "
"(this build understands <= %s); treating as empty",
store_path,
version,
_STORAGE_VERSION,
)
return {}
return data
def _save_store(data: Mapping[str, Any], store_path: Path) -> bool:
"""Atomic write of JSON trust data to `store_path`.
Uses `tempfile.mkstemp` + `Path.replace` for crash safety.
Args:
data: Full store dict to write.
store_path: Destination path.
Returns:
`True` on success, `False` on I/O failure.
"""
try:
store_path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp_path = tempfile.mkstemp(dir=store_path.parent, suffix=".tmp")
# Wrap the raw fd in a file object in its own stage: if `os.fdopen`
# raises, it did not take ownership, so the fd is still open and must be
# closed explicitly (otherwise it leaks). Only close it here — once
# `fdopen` succeeds the `with` below owns and closes it exactly once, and
# a bare `os.close(fd)` in the outer handler could race a recycled fd
# (this runs under `asyncio.to_thread`).
try:
handle = os.fdopen(fd, "w", encoding="utf-8")
except BaseException:
with contextlib.suppress(OSError):
os.close(fd)
with contextlib.suppress(OSError):
Path(tmp_path).unlink()
raise
try:
with handle as f:
json.dump(data, f, indent=2)
Path(tmp_path).replace(store_path)
except BaseException:
with contextlib.suppress(OSError):
Path(tmp_path).unlink()
raise
except (OSError, ValueError):
logger.exception("Failed to save skill trust store to %s", store_path)
return False
return True
def _read_dirs(store_path: Path, *, strict: bool = False) -> dict[str, Any]:
"""Return the `dirs` mapping from the store, or an empty dict.
Args:
store_path: Path to the trust store file.
strict: Propagated to `_load_store`; see its docstring.
"""
dirs = _load_store(store_path, strict=strict).get("dirs", {})
return dirs if isinstance(dirs, dict) else {}
def is_skill_dir_trusted(
target_dir: Path | str,
*,
store_path: Path | None = None,
) -> bool:
"""Check whether a resolved skill directory has been trusted.
Warning:
This resolves `target_dir` and checks raw membership; it does NOT do
the `resolve()`-to-self re-verification that `load_trusted_skill_dirs`
performs on each stored entry. It is therefore **not** the
containment-enforcement primitive — enforcement builds the allowlist
from `load_trusted_skill_dirs`, which drops post-approval symlink swaps.
Use this only for informational "is this exact resolved dir on record?"
checks.
Note that this check happens to fail *closed*, not open: because it
resolves the query `target_dir`, a stored directory later swapped for a
symlink is reported **not** trusted (the query resolves to the swap
target, which is not the stored key), forcing a re-prompt — the safe
direction. It is excluded from enforcement for being an exact-membership
test that skips the resolve-to-self recheck, not because it could grant
access the user never approved.
The lookup resolves `target_dir` (via `_normalize`), but `trust_skill_dir`
stores the expanduser-only `_approved_key`. In the live flow the two
coincide because callers approve an already-resolved path, so the keys
are identical. A caller that trusted a *non-canonical* path would see a
false negative here (the only failure direction, and the safe one). Pass
an already-resolved directory to keep the check meaningful.
Args:
target_dir: Directory to check; resolved before lookup.
store_path: Path to the trust store file. Defaults to
`~/.deepagents/.state/skill_trust.json`.
Returns:
`True` if the resolved directory is present in the store.
"""
if store_path is None:
store_path = _default_store_path()
return _normalize(target_dir) in _read_dirs(store_path)
def trust_skill_dir(
target_dir: Path | str,
*,
store_path: Path | None = None,
) -> bool:
"""Persist trust for a resolved skill directory.
Args:
target_dir: Canonical directory to trust. This is expected to be the
already-resolved path shown to the user, and is not resolved again
before storing so a post-approval symlink swap cannot change what
gets persisted.
store_path: Path to the trust store file. Defaults to
`~/.deepagents/.state/skill_trust.json`.
Returns:
`True` if the entry was saved successfully.
"""
if store_path is None:
store_path = _default_store_path()
# Read strictly: if an existing store can't be read, abort rather than
# rebuild it from `{}` and overwrite (which would drop every prior
# approval). A transient read error should re-prompt next time, not
# silently erase the store.
try:
data = _load_store(store_path, strict=True)
except (OSError, ValueError):
logger.exception(
"Refusing to persist skill trust: could not read existing store %s",
store_path,
)
return False
# The key is stored expanduser-only and never re-resolved (that is the
# anti-symlink-swap property). That only holds the invariant "the stored key
# is the canonical dir the user approved" if the caller already passed a
# canonical path. If it did not, `load_trusted_skill_dirs` will later drop
# the entry at its resolve()-to-self check and the approval silently never
# persists (re-prompt every session). Warn at the write boundary so that
# caller bug surfaces here instead of as a mysterious never-remembered trust.
key = _approved_key(target_dir)
try:
is_canonical = key == _normalize(target_dir)
except OSError:
# Resolving for the diagnostic failed; skip the warning rather than
# abort the write. The read-time resolve()-to-self check is the actual
# safety net, not this best-effort boundary hint.
is_canonical = True
if not is_canonical:
logger.warning(
"trust_skill_dir called with a non-canonical path %r; the stored "
"entry will be dropped at read time. Pass an already-resolved "
"directory.",
target_dir,
)
dirs = data.get("dirs")
if not isinstance(dirs, dict):
dirs = {}
dirs[key] = _TrustEntry(trusted_at=datetime.now(UTC).isoformat())
return _save_store(_TrustStore(version=_STORAGE_VERSION, dirs=dirs), store_path)
def revoke_skill_dir_trust(
target_dir: Path | str,
*,
store_path: Path | None = None,
) -> RevokeResult:
"""Remove trust for a skill directory.
Matches on both the approved (expanduser-only) key form that
`trust_skill_dir` stores and the fully-resolved form, so a caller can
revoke either by the path they see in `skills trust list` or by the
original symlink path.
Args:
target_dir: Directory to revoke.
store_path: Path to the trust store file. Defaults to
`~/.deepagents/.state/skill_trust.json`.
Returns:
`RevokeResult.REMOVED` if a matching entry was removed and persisted,
`RevokeResult.NOT_FOUND` if no entry matched (store left unchanged), or
`RevokeResult.ERROR` if the store could not be read or the write failed.
"""
if store_path is None:
store_path = _default_store_path()
# Read strictly so a transient read error aborts rather than rebuilding
# from `{}` and dropping the other entries on the next save.
try:
data = _load_store(store_path, strict=True)
except (OSError, ValueError):
logger.exception(
"Refusing to revoke skill trust: could not read existing store %s",
store_path,
)
return RevokeResult.ERROR
dirs = data.get("dirs")
if not isinstance(dirs, dict):
return RevokeResult.NOT_FOUND
keys = {_approved_key(target_dir), _normalize(target_dir)}
removed = False
for key in keys:
if key in dirs:
del dirs[key]
removed = True
if not removed:
return RevokeResult.NOT_FOUND
data["version"] = _STORAGE_VERSION
data["dirs"] = dirs
return RevokeResult.REMOVED if _save_store(data, store_path) else RevokeResult.ERROR
def clear_trusted_skill_dirs(*, store_path: Path | None = None) -> bool:
"""Remove all trusted skill directories.
Args:
store_path: Path to the trust store file. Defaults to
`~/.deepagents/.state/skill_trust.json`.
Returns:
`True` if the store was cleared (or was already empty).
"""
if store_path is None:
store_path = _default_store_path()
if not store_path.exists():
return True
return _save_store(_TrustStore(version=_STORAGE_VERSION, dirs={}), store_path)
def list_trusted_skill_dirs(
*,
store_path: Path | None = None,
strict: bool = False,
) -> list[str]:
"""Return the sorted list of trusted skill directory paths.
Args:
store_path: Path to the trust store file. Defaults to
`~/.deepagents/.state/skill_trust.json`.
strict: When `True`, an existing-but-unreadable store re-raises instead
of degrading to an empty list. The audit command (`skills trust
list`) passes `strict=True` so it can report an error rather than
falsely printing "No trusted skill directories" while entries the
user cannot then see or revoke sit in an unreadable file.
Returns:
Sorted absolute directory paths previously trusted.
When `strict`, an existing-but-unreadable or corrupt store propagates
the underlying error (`OSError` / `json.JSONDecodeError` / `ValueError`)
from `_load_store` instead of returning a list.
"""
if store_path is None:
store_path = _default_store_path()
return sorted(_read_dirs(store_path, strict=strict))
def list_trusted_skill_dir_entries(
*,
store_path: Path | None = None,
strict: bool = False,
) -> list[tuple[str, str]]:
"""Return trusted directories paired with their approval timestamps.
The audit surface for the `trusted_at` metadata that `trust_skill_dir`
records: `list_trusted_skill_dirs` returns only paths (all enforcement
needs), so this is the one reader of the timestamp, used by `skills trust
list` to show *when* each directory was approved.
Args:
store_path: Path to the trust store file. Defaults to
`~/.deepagents/.state/skill_trust.json`.
strict: Propagated to `_load_store`; see `list_trusted_skill_dirs`.
Returns:
`(path, trusted_at)` tuples sorted by path. `trusted_at` is the stored
ISO-8601 string, or `""` when a hand-edited entry omitted
or malformed it (the path is still listed so it remains
visible and revocable).
"""
if store_path is None:
store_path = _default_store_path()
entries: list[tuple[str, str]] = []
for path, entry in _read_dirs(store_path, strict=strict).items():
trusted_at = entry.get("trusted_at", "") if isinstance(entry, dict) else ""
entries.append((path, trusted_at if isinstance(trusted_at, str) else ""))
return sorted(entries)
def load_trusted_skill_dirs(*, store_path: Path | None = None) -> list[Path]:
"""Return verified trusted skill directories as canonical `Path` objects.
Used to extend the containment allowlist passed to `load_skill_content`.
Stored entries are the exact canonical directory the user approved (already
resolved at trust time). Each entry is re-verified here rather than blindly
re-resolved: if a stored path no longer resolves to itself — because it, or
a parent component, was replaced with a symlink after approval — the current
resolution would point somewhere the user never approved. Such entries are
dropped (and logged) instead of silently allowlisting the swapped target, so
a post-approval symlink swap re-prompts rather than granting access.
Args:
store_path: Path to the trust store file. Defaults to
`~/.deepagents/.state/skill_trust.json`.
Returns:
Canonical directory paths that still resolve to themselves; empty when
nothing is trusted.
"""
verified: list[Path] = []
for entry in list_trusted_skill_dirs(store_path=store_path):
stored = Path(entry)
try:
resolves_to_self = stored.resolve() == stored
except (OSError, RuntimeError):
# A single unresolvable entry (e.g. a symlink cycle introduced under
# the stored path) must not abort discovery of every other skill.
# Drop it like the swap case below. `RuntimeError` is caught
# alongside `OSError` to match the resolve guard in
# `app._prompt_skill_trust_and_retry` (some Python builds surface a
# symlink loop as `RuntimeError`).
logger.warning(
"Trusted skill directory %s could not be resolved; "
"ignoring the trust entry.",
entry,
exc_info=True,
)
continue
if resolves_to_self:
verified.append(stored)
else:
logger.warning(
"Trusted skill directory %s no longer resolves to itself "
"(a symlink may have been introduced since approval); "
"ignoring the stale trust entry.",
entry,
)
return verified
libs/code/deepagents_code/ui.py
+25
View File
@@ -350,6 +350,7 @@ def show_skills_help() -> None:
console.print(" create <name> Create a new skill")
console.print(" info <name> Show detailed information about a skill")
console.print(" delete <name> Delete a skill")
console.print(" trust Manage trusted skill directories")
console.print()
_print_option_section(
" --agent <name> Specify agent identifier (default: agent)",
@@ -455,6 +456,30 @@ def show_skills_delete_help() -> None:
console.print()
def show_skills_trust_help() -> None:
"""Show help information for the `skills trust` subcommand."""
console.print()
console.print("[bold]Usage:[/bold]", style=theme.PRIMARY)
console.print(" dcode skills trust <command>")
console.print()
console.print("[bold]Commands:[/bold]", style=theme.PRIMARY)
console.print(" list|ls List trusted skill directories")
console.print(" revoke <dir> Revoke trust for a directory")
console.print(" clear Remove all trusted skill directories")
console.print()
console.print(
"Directories are trusted when you approve a skill that resolves "
"outside the standard skill roots (for example, a symlink target). "
"Trust is stored in ~/.deepagents/.state/skill_trust.json."
)
console.print()
console.print("[bold]Examples:[/bold]", style=theme.PRIMARY)
console.print(" dcode skills trust list")
console.print(" dcode skills trust revoke /shared/skills/my-skill")
console.print(" dcode skills trust clear")
console.print()
def show_update_help() -> None:
"""Show help information for the `update` subcommand."""
console.print()
libs/code/deepagents_code/widgets/skill_trust.py
+131
View File
@@ -0,0 +1,131 @@
"""Trust prompt for skills that resolve outside trusted skill directories.
When a `/skill:<name>` invocation reads a `SKILL.md` whose resolved path (via
symlink) falls outside every trusted skill root, `load_skill_content` refuses
the read. Rather than forcing the user to quit, edit env/config, and relaunch,
this non-blocking modal asks for an in-the-moment decision. Allowing persists
the resolved target directory to the skill trust store.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, ClassVar
from textual.binding import Binding, BindingType
from textual.containers import Vertical
from textual.content import Content
from textual.screen import ModalScreen
from textual.widgets import Static
if TYPE_CHECKING:
from textual.app import ComposeResult
class SkillTrustScreen(ModalScreen[bool | None]):
"""Approval overlay for a skill resolving outside trusted directories.
Dismisses with `True` when the user allows the resolved target and `False`
when the user declines. Esc is treated as deny so the user is never forced
into reading from an untrusted location they did not explicitly choose.
Typed `bool | None` rather than `bool`: a programmatic pop, or the app's
priority Esc binding falling through to `dismiss(None)` (see
`action_cancel`), can yield `None`. The caller collapses `None` and `False`
to deny (`if not allowed`), so both dismiss values fail closed.
"""
BINDINGS: ClassVar[list[BindingType]] = [
Binding("enter", "confirm", "Allow", show=False, priority=True),
Binding("escape", "cancel", "Deny", show=False, priority=True),
]
CSS = """
SkillTrustScreen {
align: center middle;
}
SkillTrustScreen > Vertical {
width: 72;
max-width: 90%;
height: auto;
background: $surface;
border: solid $warning;
padding: 1 2;
}
SkillTrustScreen .skill-trust-title {
text-style: bold;
color: $warning;
text-align: center;
margin-bottom: 1;
}
SkillTrustScreen .skill-trust-body {
height: auto;
color: $text;
margin-bottom: 1;
}
SkillTrustScreen .skill-trust-help {
height: 1;
color: $text-muted;
text-style: italic;
text-align: center;
}
"""
def __init__(self, skill_name: str, target_dir: str) -> None:
"""Initialize the prompt.
Args:
skill_name: Name of the skill being invoked.
target_dir: Resolved directory the skill's `SKILL.md` lives in.
"""
super().__init__()
self._skill_name = skill_name
self._target_dir = target_dir
def compose(self) -> ComposeResult:
"""Compose the skill trust dialog.
Yields:
Title, body, and help-row widgets parented inside a `Vertical`.
"""
with Vertical():
yield Static(
"Allow skill from outside trusted directories?",
classes="skill-trust-title",
markup=False,
)
yield Static(
Content.from_markup(
"Skill [bold]$name[/bold] resolves to [bold]$dir[/bold], "
"outside your trusted skill directories. Allowing reads "
"instructions from there and remembers this location for "
"future sessions.",
name=self._skill_name,
dir=self._target_dir,
),
classes="skill-trust-body",
markup=False,
)
yield Static(
"Enter to allow, Esc to deny",
classes="skill-trust-help",
markup=False,
)
def action_confirm(self) -> None:
"""Dismiss with `True`."""
self.dismiss(True)
def action_cancel(self) -> None:
"""Dismiss with `False`.
The method name must stay `cancel`: the app owns a priority `escape`
binding that, for an active `ModalScreen`, dispatches to
`action_cancel` if present and otherwise falls through to
`dismiss(None)`. Renaming this would silently regress Esc to a
`None` dismiss instead of an explicit deny.
"""
self.dismiss(False)
libs/code/tests/unit_tests/skills/test_commands.py
+287
View File
@@ -1336,3 +1336,290 @@ class TestDeleteArgparsing:
# rather than falling through to show_skills_help()
joined = "\n".join(output)
assert "not found" in joined.lower()
class TestSkillsTrustCommand:
"""Unit tests for `_trust` and its dispatch in `execute_skills_command`."""
@staticmethod
def _capture() -> tuple[list[str], "object"]:
output: list[str] = []
def capture_print(*args_p: object, **_: object) -> None:
output.append(" ".join(str(a) for a in args_p))
return output, capture_print
def test_list_populated(self) -> None:
"""`list` prints each trusted directory."""
from deepagents_code.skills.commands import _trust
args = argparse.Namespace(trust_command="list")
output, capture_print = self._capture()
with (
patch(
"deepagents_code.skills.trust.list_trusted_skill_dir_entries",
return_value=[
("/shared/a", "2026-01-01T00:00:00+00:00"),
("/shared/b", ""),
],
),
patch("deepagents_code.config.console") as mock_console,
):
mock_console.print = capture_print
_trust(args)
joined = "\n".join(output)
assert "/shared/a" in joined
assert "/shared/b" in joined
# The `trusted_at` timestamp is surfaced when present.
assert "2026-01-01T00:00:00+00:00" in joined
def test_list_empty(self) -> None:
"""`list` on an empty store reports that nothing is trusted."""
from deepagents_code.skills.commands import _trust
args = argparse.Namespace(trust_command="list")
output, capture_print = self._capture()
with (
patch(
"deepagents_code.skills.trust.list_trusted_skill_dir_entries",
return_value=[],
),
patch("deepagents_code.config.console") as mock_console,
):
mock_console.print = capture_print
_trust(args)
assert "no trusted skill directories" in "\n".join(output).lower()
def test_ls_alias_behaves_like_list(self) -> None:
"""`ls` is an alias for `list`."""
from deepagents_code.skills.commands import _trust
args = argparse.Namespace(trust_command="ls")
output, capture_print = self._capture()
with (
patch(
"deepagents_code.skills.trust.list_trusted_skill_dir_entries",
return_value=[("/shared/a", "")],
),
patch("deepagents_code.config.console") as mock_console,
):
mock_console.print = capture_print
_trust(args)
assert "/shared/a" in "\n".join(output)
def test_list_json_output(self) -> None:
"""`list --output json` emits the standard envelope shape."""
import json
from io import StringIO
from deepagents_code.skills.commands import _trust
args = argparse.Namespace(trust_command="list", output_format="json")
buf = StringIO()
with (
patch(
"deepagents_code.skills.trust.list_trusted_skill_dir_entries",
return_value=[("/shared/a", "2026-01-01T00:00:00+00:00")],
),
patch("sys.stdout", buf),
):
_trust(args)
result = json.loads(buf.getvalue())
assert result["command"] == "skills trust list"
assert result["data"] == [
{"dir": "/shared/a", "trusted_at": "2026-01-01T00:00:00+00:00"}
]
def test_list_unreadable_store_errors_and_exits(self) -> None:
"""An unreadable store surfaces an error and exits non-zero.
It must never silently print "No trusted skill directories" while
entries the user cannot see or revoke sit in the file.
"""
from deepagents_code.skills.commands import _trust
args = argparse.Namespace(trust_command="list")
output, capture_print = self._capture()
with (
patch(
"deepagents_code.skills.trust.list_trusted_skill_dir_entries",
side_effect=OSError("permission denied"),
),
patch("deepagents_code.config.console") as mock_console,
):
mock_console.print = capture_print
with pytest.raises(SystemExit) as exc_info:
_trust(args)
assert exc_info.value.code == 1
joined = "\n".join(output).lower()
assert "could not read" in joined
assert "no trusted skill directories" not in joined
def test_revoke_success(self) -> None:
"""A successful revoke confirms the removed directory."""
from deepagents_code.skills.commands import _trust
from deepagents_code.skills.trust import RevokeResult
args = argparse.Namespace(trust_command="revoke", dir="/shared/a")
output, capture_print = self._capture()
with (
patch(
"deepagents_code.skills.trust.revoke_skill_dir_trust",
return_value=RevokeResult.REMOVED,
),
patch("deepagents_code.config.console") as mock_console,
):
mock_console.print = capture_print
_trust(args)
assert "revoked" in "\n".join(output).lower()
def test_revoke_not_found_reports_honestly(self) -> None:
"""Revoking a dir that was never trusted must not print a false success."""
from deepagents_code.skills.commands import _trust
from deepagents_code.skills.trust import RevokeResult
args = argparse.Namespace(trust_command="revoke", dir="/shared/nope")
output, capture_print = self._capture()
with (
patch(
"deepagents_code.skills.trust.revoke_skill_dir_trust",
return_value=RevokeResult.NOT_FOUND,
),
patch("deepagents_code.config.console") as mock_console,
):
mock_console.print = capture_print
_trust(args)
joined = "\n".join(output).lower()
assert "no trust entry found" in joined
assert "revoked" not in joined
def test_revoke_json_output(self) -> None:
"""`revoke --output json` emits the standard envelope with the result."""
import json
from io import StringIO
from deepagents_code.skills.commands import _trust
from deepagents_code.skills.trust import RevokeResult
args = argparse.Namespace(
trust_command="revoke", dir="/shared/a", output_format="json"
)
buf = StringIO()
with (
patch(
"deepagents_code.skills.trust.revoke_skill_dir_trust",
return_value=RevokeResult.NOT_FOUND,
),
patch("sys.stdout", buf),
):
_trust(args)
result = json.loads(buf.getvalue())
assert result["command"] == "skills trust revoke"
assert result["data"] == {"dir": "/shared/a", "result": "not_found"}
def test_revoke_failure_exits(self) -> None:
"""A store/IO error reports an error and exits non-zero."""
from deepagents_code.skills.commands import _trust
from deepagents_code.skills.trust import RevokeResult
args = argparse.Namespace(trust_command="revoke", dir="/shared/a")
output, capture_print = self._capture()
with (
patch(
"deepagents_code.skills.trust.revoke_skill_dir_trust",
return_value=RevokeResult.ERROR,
),
patch("deepagents_code.config.console") as mock_console,
):
mock_console.print = capture_print
with pytest.raises(SystemExit) as exc_info:
_trust(args)
assert exc_info.value.code == 1
assert "could not revoke" in "\n".join(output).lower()
def test_clear_success(self) -> None:
"""A successful clear confirms removal."""
from deepagents_code.skills.commands import _trust
args = argparse.Namespace(trust_command="clear")
output, capture_print = self._capture()
with (
patch(
"deepagents_code.skills.trust.clear_trusted_skill_dirs",
return_value=True,
),
patch("deepagents_code.config.console") as mock_console,
):
mock_console.print = capture_print
_trust(args)
assert "cleared" in "\n".join(output).lower()
def test_clear_json_output(self) -> None:
"""`clear --output json` emits the standard envelope."""
import json
from io import StringIO
from deepagents_code.skills.commands import _trust
args = argparse.Namespace(trust_command="clear", output_format="json")
buf = StringIO()
with (
patch(
"deepagents_code.skills.trust.clear_trusted_skill_dirs",
return_value=True,
),
patch("sys.stdout", buf),
):
_trust(args)
result = json.loads(buf.getvalue())
assert result["command"] == "skills trust clear"
assert result["data"] == {"cleared": True}
def test_clear_failure_exits(self) -> None:
"""A failed clear reports an error and exits non-zero."""
from deepagents_code.skills.commands import _trust
args = argparse.Namespace(trust_command="clear")
_output, capture_print = self._capture()
with (
patch(
"deepagents_code.skills.trust.clear_trusted_skill_dirs",
return_value=False,
),
patch("deepagents_code.config.console") as mock_console,
):
mock_console.print = capture_print
with pytest.raises(SystemExit) as exc_info:
_trust(args)
assert exc_info.value.code == 1
def test_no_subcommand_shows_help(self) -> None:
"""`trust` with no subcommand shows the trust help."""
from deepagents_code.skills.commands import _trust
args = argparse.Namespace(trust_command=None)
with patch("deepagents_code.ui.show_skills_trust_help") as mock_help:
_trust(args)
mock_help.assert_called_once()
def test_execute_dispatches_trust_before_agent_validation(self) -> None:
"""`trust` is routed without an `--agent`, ahead of agent validation."""
args = argparse.Namespace(skills_command="trust", trust_command="list")
with patch("deepagents_code.skills.commands._trust") as mock_trust:
execute_skills_command(args)
mock_trust.assert_called_once_with(args)
libs/code/tests/unit_tests/skills/test_skill_trust.py
+528
View File
@@ -0,0 +1,528 @@
"""Tests for the skill directory trust store."""
from pathlib import Path
from unittest.mock import MagicMock
import pytest
from deepagents_code.skills.trust import (
RevokeResult,
clear_trusted_skill_dirs,
is_skill_dir_trusted,
list_trusted_skill_dir_entries,
list_trusted_skill_dirs,
load_trusted_skill_dirs,
revoke_skill_dir_trust,
trust_skill_dir,
)
class TestSkillTrustStore:
"""CRUD behavior for the persistent skill trust store."""
def test_untrusted_by_default(self, tmp_path: Path) -> None:
"""A directory is untrusted when the store file does not exist."""
store = tmp_path / "skill_trust.json"
assert not is_skill_dir_trusted(tmp_path / "a", store_path=store)
def test_trust_and_verify(self, tmp_path: Path) -> None:
"""Trusting a directory then checking returns True."""
store = tmp_path / "skill_trust.json"
target = tmp_path / "shared"
target.mkdir()
assert trust_skill_dir(target, store_path=store)
assert is_skill_dir_trusted(target, store_path=store)
def test_trust_persists_approved_path_without_resolving_again(
self, tmp_path: Path
) -> None:
"""Trust stores the approved path even after a symlink swap."""
store = tmp_path / "skill_trust.json"
approved = tmp_path / "approved"
approved.mkdir()
approved_key = approved.resolve()
attacker = tmp_path / "attacker"
attacker.mkdir()
approved.rmdir()
approved.symlink_to(attacker)
assert trust_skill_dir(approved_key, store_path=store)
assert list_trusted_skill_dirs(store_path=store) == [str(approved_key)]
assert not is_skill_dir_trusted(attacker, store_path=store)
assert load_trusted_skill_dirs(store_path=store) == []
def test_revoke(self, tmp_path: Path) -> None:
"""Revoking trust makes the directory untrusted again."""
store = tmp_path / "skill_trust.json"
target = tmp_path / "shared"
target.mkdir()
trust_skill_dir(target, store_path=store)
assert revoke_skill_dir_trust(target, store_path=store) is RevokeResult.REMOVED
assert not is_skill_dir_trusted(target, store_path=store)
def test_revoke_nonexistent(self, tmp_path: Path) -> None:
"""Revoking an untrusted directory reports NOT_FOUND, not a false success."""
store = tmp_path / "skill_trust.json"
assert (
revoke_skill_dir_trust(tmp_path / "nope", store_path=store)
is RevokeResult.NOT_FOUND
)
def test_revoke_stale_entry_after_symlink_swap(self, tmp_path: Path) -> None:
"""Revoking the listed path removes stale trust after a symlink swap."""
store = tmp_path / "skill_trust.json"
approved = tmp_path / "approved"
approved.mkdir()
trust_skill_dir(approved, store_path=store)
listed = list_trusted_skill_dirs(store_path=store)
attacker = tmp_path / "attacker"
attacker.mkdir()
approved.rmdir()
approved.symlink_to(attacker)
assert (
revoke_skill_dir_trust(listed[0], store_path=store) is RevokeResult.REMOVED
)
assert list_trusted_skill_dirs(store_path=store) == []
def test_list_sorted(self, tmp_path: Path) -> None:
"""Listing returns resolved paths in sorted order."""
store = tmp_path / "skill_trust.json"
a = tmp_path / "a"
a.mkdir()
b = tmp_path / "b"
b.mkdir()
trust_skill_dir(b, store_path=store)
trust_skill_dir(a, store_path=store)
assert list_trusted_skill_dirs(store_path=store) == sorted(
[str(a.resolve()), str(b.resolve())]
)
def test_load_returns_paths(self, tmp_path: Path) -> None:
"""load_trusted_skill_dirs returns resolved Path objects."""
store = tmp_path / "skill_trust.json"
target = tmp_path / "shared"
target.mkdir()
trust_skill_dir(target, store_path=store)
assert load_trusted_skill_dirs(store_path=store) == [target.resolve()]
def test_load_skips_post_approval_symlink_swap(self, tmp_path: Path) -> None:
"""A stored dir swapped for a symlink after approval is not loaded.
The stored entry is the canonical directory that was approved. If that
path is later replaced by a symlink to a different directory, loading it
must not follow the symlink and allowlist the swapped target.
"""
store = tmp_path / "skill_trust.json"
approved = tmp_path / "approved"
approved.mkdir()
trust_skill_dir(approved, store_path=store)
assert load_trusted_skill_dirs(store_path=store) == [approved.resolve()]
# Attacker replaces the approved directory with a symlink elsewhere.
attacker = tmp_path / "attacker"
attacker.mkdir()
approved.rmdir()
approved.symlink_to(attacker)
assert load_trusted_skill_dirs(store_path=store) == []
def test_load_keeps_unchanged_dir(self, tmp_path: Path) -> None:
"""An unchanged stored dir is still returned as its canonical path."""
store = tmp_path / "skill_trust.json"
target = tmp_path / "shared"
target.mkdir()
trust_skill_dir(target, store_path=store)
assert load_trusted_skill_dirs(store_path=store) == [target.resolve()]
def test_clear(self, tmp_path: Path) -> None:
"""Clearing removes every trusted directory."""
store = tmp_path / "skill_trust.json"
target = tmp_path / "shared"
target.mkdir()
trust_skill_dir(target, store_path=store)
assert clear_trusted_skill_dirs(store_path=store)
assert list_trusted_skill_dirs(store_path=store) == []
def test_clear_replaces_corrupt_store(self, tmp_path: Path) -> None:
"""Clearing resets an existing corrupt store."""
store = tmp_path / "skill_trust.json"
store.write_text("{not valid json")
assert clear_trusted_skill_dirs(store_path=store)
assert list_trusted_skill_dirs(store_path=store, strict=True) == []
def test_corrupt_store_degrades_to_empty(self, tmp_path: Path) -> None:
"""A corrupt store file is treated as nothing trusted."""
store = tmp_path / "skill_trust.json"
store.write_text("{not valid json")
assert list_trusted_skill_dirs(store_path=store) == []
assert not is_skill_dir_trusted(tmp_path, store_path=store)
def test_default_store_path_uses_state_dir(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""The default store path lives under DEFAULT_STATE_DIR."""
import deepagents_code.model_config as mc
monkeypatch.setattr(mc, "DEFAULT_STATE_DIR", tmp_path)
target = tmp_path / "shared"
target.mkdir()
assert trust_skill_dir(target)
assert (tmp_path / "skill_trust.json").exists()
assert is_skill_dir_trusted(target)
class TestSkillTrustStoreRobustness:
"""Durability and honesty guarantees, mirroring `test_mcp_trust.py`."""
def test_save_failure_returns_false(self, tmp_path: Path) -> None:
"""An unwritable store path returns False instead of raising."""
# Parent is a regular file, so mkdir(parents=True) fails with an OSError
# subclass that _save_store must catch and report as a failed write.
blocker = tmp_path / "blocker"
blocker.write_text("x")
store = blocker / "skill_trust.json"
assert trust_skill_dir(tmp_path, store_path=store) is False
def test_trust_heals_malformed_dirs_value(self, tmp_path: Path) -> None:
"""A non-dict `dirs` value is replaced, not crashed on or appended to."""
import json
store = tmp_path / "skill_trust.json"
store.write_text(json.dumps({"version": 1, "dirs": []}))
target = tmp_path / "shared"
target.mkdir()
assert trust_skill_dir(target, store_path=store)
assert is_skill_dir_trusted(target, store_path=store)
def test_revoke_preserves_other_entries_and_version(self, tmp_path: Path) -> None:
"""Revoking one dir leaves siblings intact and re-stamps the version."""
import json
store = tmp_path / "skill_trust.json"
a = tmp_path / "a"
a.mkdir()
b = tmp_path / "b"
b.mkdir()
trust_skill_dir(a, store_path=store)
trust_skill_dir(b, store_path=store)
assert revoke_skill_dir_trust(a, store_path=store) is RevokeResult.REMOVED
assert not is_skill_dir_trusted(a, store_path=store)
assert is_skill_dir_trusted(b, store_path=store)
assert json.loads(store.read_text(encoding="utf-8"))["version"] == 1
def test_on_disk_shape(self, tmp_path: Path) -> None:
"""The store is a versioned JSON object mapping dirs to metadata."""
import json
store = tmp_path / "skill_trust.json"
target = tmp_path / "shared"
target.mkdir()
trust_skill_dir(target, store_path=store)
data = json.loads(store.read_text(encoding="utf-8"))
assert data["version"] == 1
assert str(target.resolve()) in data["dirs"]
assert "trusted_at" in data["dirs"][str(target.resolve())]
def test_trust_does_not_clobber_on_unreadable_store(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A transient read error aborts the write instead of erasing entries.
The read/modify/write must not rebuild the store from `{}` on a
transient `OSError`; doing so would drop every prior approval.
"""
import deepagents_code.skills.trust as trust_mod
store = tmp_path / "skill_trust.json"
first = tmp_path / "first"
first.mkdir()
trust_skill_dir(first, store_path=store)
before = store.read_text(encoding="utf-8")
monkeypatch.setattr(
trust_mod, "_load_store", MagicMock(side_effect=OSError("transient"))
)
second = tmp_path / "second"
second.mkdir()
assert trust_skill_dir(second, store_path=store) is False
# The original store is untouched — the existing approval survives.
assert store.read_text(encoding="utf-8") == before
def test_list_strict_surfaces_unreadable_store(self, tmp_path: Path) -> None:
"""`strict=True` re-raises on a corrupt store; the default degrades."""
import json
store = tmp_path / "skill_trust.json"
store.write_text("{not valid json")
# Enforcement/default path stays fail-closed (empty).
assert list_trusted_skill_dirs(store_path=store) == []
# Audit path opts into surfacing the error.
with pytest.raises(json.JSONDecodeError):
list_trusted_skill_dirs(store_path=store, strict=True)
def test_list_strict_missing_store_is_empty_not_error(self, tmp_path: Path) -> None:
"""A missing store is first-run state, not an error, even under strict."""
store = tmp_path / "skill_trust.json"
assert list_trusted_skill_dirs(store_path=store, strict=True) == []
def test_newer_schema_version_is_refused(self, tmp_path: Path) -> None:
"""A store written by a newer build is not partially read.
Enforcement/default stays fail-closed (empty) so an unknown schema can't
grant access by being misread; the audit path surfaces the error.
"""
import json
store = tmp_path / "skill_trust.json"
store.write_text(
json.dumps({"version": 999, "dirs": {"/shared/a": {}}}),
encoding="utf-8",
)
assert list_trusted_skill_dirs(store_path=store) == []
with pytest.raises(ValueError, match="unrecognized schema version"):
list_trusted_skill_dirs(store_path=store, strict=True)
def test_load_survives_unresolvable_entry(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""One entry that fails to resolve is dropped, not fatal to discovery."""
import json
store = tmp_path / "skill_trust.json"
good = tmp_path / "good"
good.mkdir()
boom = tmp_path / "boom"
store.write_text(
json.dumps(
{
"version": 1,
"dirs": {
str(good): {"trusted_at": "t"},
str(boom): {"trusted_at": "t"},
},
}
)
)
real_resolve = Path.resolve
def flaky_resolve(self: Path, strict: bool = False) -> Path:
if self == boom:
msg = "ELOOP"
raise OSError(msg)
return real_resolve(self, strict)
monkeypatch.setattr(Path, "resolve", flaky_resolve)
assert load_trusted_skill_dirs(store_path=store) == [good]
def test_load_survives_entry_that_raises_runtime_error(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A `RuntimeError` from `resolve()` drops one entry, not all discovery.
Some Python builds surface a symlink loop as `RuntimeError` rather than
`OSError`; the per-entry guard must catch it so one bad stored entry
can't abort discovery of every other skill.
"""
import json
store = tmp_path / "skill_trust.json"
good = tmp_path / "good"
good.mkdir()
boom = tmp_path / "boom"
store.write_text(
json.dumps(
{
"version": 1,
"dirs": {
str(good): {"trusted_at": "t"},
str(boom): {"trusted_at": "t"},
},
}
)
)
real_resolve = Path.resolve
def flaky_resolve(self: Path, strict: bool = False) -> Path:
if self == boom:
msg = "symlink loop"
raise RuntimeError(msg)
return real_resolve(self, strict)
monkeypatch.setattr(Path, "resolve", flaky_resolve)
assert load_trusted_skill_dirs(store_path=store) == [good]
def test_revoke_does_not_clobber_on_unreadable_store(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A transient read error aborts the revoke instead of erasing entries.
Mirrors `test_trust_does_not_clobber_on_unreadable_store` for the revoke
path: a strict-read failure must map to `ERROR` and leave the store
byte-for-byte unchanged, never rebuild it from `{}` and drop siblings.
"""
import deepagents_code.skills.trust as trust_mod
store = tmp_path / "skill_trust.json"
a = tmp_path / "a"
a.mkdir()
b = tmp_path / "b"
b.mkdir()
trust_skill_dir(a, store_path=store)
trust_skill_dir(b, store_path=store)
before = store.read_text(encoding="utf-8")
monkeypatch.setattr(
trust_mod, "_load_store", MagicMock(side_effect=OSError("transient"))
)
assert revoke_skill_dir_trust(a, store_path=store) is RevokeResult.ERROR
# Both approvals survive: the store was not rebuilt from an empty dict.
assert store.read_text(encoding="utf-8") == before
def test_revoke_save_failure_maps_to_error(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A failed write returns `ERROR`, not a false `REMOVED`."""
import deepagents_code.skills.trust as trust_mod
store = tmp_path / "skill_trust.json"
a = tmp_path / "a"
a.mkdir()
trust_skill_dir(a, store_path=store)
monkeypatch.setattr(trust_mod, "_save_store", MagicMock(return_value=False))
assert revoke_skill_dir_trust(a, store_path=store) is RevokeResult.ERROR
def test_top_level_not_a_dict(self, tmp_path: Path) -> None:
"""A store whose top-level JSON is not an object is refused/degraded.
The prior coverage only exercised a non-dict *nested* `dirs`; this pins
the top-level branch: enforcement degrades to empty, audit surfaces it.
"""
store = tmp_path / "skill_trust.json"
store.write_text("[]", encoding="utf-8")
# Enforcement/default path stays fail-closed (empty).
assert list_trusted_skill_dirs(store_path=store) == []
# Audit path opts into surfacing the error.
with pytest.raises(ValueError, match="not a JSON object"):
list_trusted_skill_dirs(store_path=store, strict=True)
def test_non_integer_schema_version_is_refused(self, tmp_path: Path) -> None:
"""A present-but-non-int `version` is unrecognized, not silently trusted.
Only tampering or a corrupt write produces a non-int version (every
writer stamps an int), so it must fail closed like a too-new version
rather than falling through and reading `dirs`.
"""
import json
store = tmp_path / "skill_trust.json"
store.write_text(
json.dumps({"version": "1", "dirs": {"/shared/a": {}}}),
encoding="utf-8",
)
assert list_trusted_skill_dirs(store_path=store) == []
with pytest.raises(ValueError, match="unrecognized schema version"):
list_trusted_skill_dirs(store_path=store, strict=True)
def test_trust_warns_on_non_canonical_path(
self, tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
"""Trusting a non-canonical path warns at the write boundary.
Such a key is dropped later by `load_trusted_skill_dirs`' resolve-to-self
check, so the warning surfaces the caller bug here rather than as a
silently never-remembered trust.
"""
import logging
store = tmp_path / "skill_trust.json"
real = tmp_path / "real"
real.mkdir()
link = tmp_path / "link"
link.symlink_to(real, target_is_directory=True)
with caplog.at_level(logging.WARNING, logger="deepagents_code.skills.trust"):
# `link` expanduser()s to itself but resolve()s to `real`, so it is
# non-canonical and should trip the boundary warning.
assert trust_skill_dir(link, store_path=store)
assert any("non-canonical" in r.message for r in caplog.records)
def test_list_entries_surfaces_trusted_at(self, tmp_path: Path) -> None:
"""`list_trusted_skill_dir_entries` pairs each path with its timestamp."""
store = tmp_path / "skill_trust.json"
target = tmp_path / "shared"
target.mkdir()
trust_skill_dir(target, store_path=store)
entries = list_trusted_skill_dir_entries(store_path=store)
assert len(entries) == 1
path, trusted_at = entries[0]
assert path == str(target.resolve())
# A real ISO-8601 timestamp was recorded, not an empty placeholder.
assert trusted_at
from datetime import datetime
datetime.fromisoformat(trusted_at) # parses without raising
def test_load_skips_parent_component_symlink_swap(self, tmp_path: Path) -> None:
"""A swapped *parent* of a stored dir drops the entry, like a leaf swap.
Both the module docstring and `load_trusted_skill_dirs` claim the
`resolve()`-to-self check catches a symlink introduced at *any* path
component, not just the leaf. Replace a parent directory with a symlink
so the stored path still exists but resolves elsewhere, and confirm the
entry is dropped rather than followed to the swapped target.
"""
import shutil
store = tmp_path / "skill_trust.json"
skill = tmp_path / "a" / "b" / "skill"
skill.mkdir(parents=True)
stored = skill.resolve()
trust_skill_dir(stored, store_path=store)
assert load_trusted_skill_dirs(store_path=store) == [stored]
# Replace the parent component `a/b` with a symlink to a sibling that
# also contains `skill`, so `stored` remains reachable but canonicalizes
# to a directory the user never approved.
evil_parent = tmp_path / "evil"
(evil_parent / "skill").mkdir(parents=True)
parent = tmp_path / "a" / "b"
shutil.rmtree(parent)
parent.symlink_to(evil_parent, target_is_directory=True)
assert skill.exists() # still reachable through the swapped parent
assert load_trusted_skill_dirs(store_path=store) == []
def test_load_corrupt_store_fails_closed(self, tmp_path: Path) -> None:
"""`load_trusted_skill_dirs` degrades to empty on a corrupt store.
The existing corrupt-store tests assert on `list_trusted_skill_dirs`;
this pins fail-closed at the actual allowlist builder that
`discover_skills_and_roots` consumes.
"""
store = tmp_path / "skill_trust.json"
store.write_text("{not valid json", encoding="utf-8")
assert load_trusted_skill_dirs(store_path=store) == []
def test_load_newer_version_fails_closed(self, tmp_path: Path) -> None:
"""A newer-schema store yields no trusted dirs at the enforcement entry.
A store written by a newer build must not be partially read into the
containment allowlist; `load_trusted_skill_dirs` (non-strict) returns
empty rather than trusting `dirs` it may misinterpret.
"""
import json
store = tmp_path / "skill_trust.json"
store.write_text(
json.dumps({"version": 999, "dirs": {str(tmp_path): {"trusted_at": "t"}}}),
encoding="utf-8",
)
assert load_trusted_skill_dirs(store_path=store) == []
libs/code/tests/unit_tests/test_non_interactive.py
+58
View File
@@ -1135,6 +1135,64 @@ class TestNonInteractivePrompt:
assert result == 1
mock_start_server.assert_not_awaited()
async def test_initial_skill_containment_failure_hard_fails(self) -> None:
"""A skill resolving outside trusted roots hard-fails headlessly.
The non-interactive path has no TUI to prompt on, so it must never try
to grant trust: it fails closed with the containment error and never
starts the server.
"""
skill = {
"name": "evil-skill",
"description": "Resolves outside trusted roots",
"path": "/skills/evil-skill/SKILL.md",
"license": None,
"compatibility": None,
"metadata": {},
"allowed_tools": [],
"source": "user",
}
with (
patch(
"deepagents_code.non_interactive.create_model",
return_value=ModelResult(
model=MagicMock(),
model_name="test-model",
provider="test",
),
),
patch(
"deepagents_code.non_interactive.settings",
) as mock_settings,
patch(
"deepagents_code.skills.invocation.discover_skills_and_roots",
return_value=([skill], []),
),
patch(
"deepagents_code.skills.load.load_skill_content",
side_effect=PermissionError(
"Skill path /tmp/evil resolves outside all allowed skill "
"directories."
),
),
patch(
"deepagents_code.server_manager.start_server_and_get_agent",
new_callable=AsyncMock,
) as mock_start_server,
):
mock_settings.shell_allow_list = None
mock_settings.has_tavily = False
mock_settings.model_name = None
result = await run_non_interactive(
message="review this patch",
initial_skill="evil-skill",
quiet=True,
)
assert result == 1
mock_start_server.assert_not_awaited()
def _make_interrupt_chunk(interrupt_id: str = "i1") -> tuple:
"""Return a stream chunk that triggers one HITL interrupt.
libs/code/tests/unit_tests/test_skill_invocation.py
+407 -4
View File
@@ -2,7 +2,7 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@@ -15,9 +15,6 @@ from deepagents_code.command_registry import (
)
from deepagents_code.skills.load import load_skill_content
if TYPE_CHECKING:
from pathlib import Path
class TestLoadSkillContent:
"""Test load_skill_content() reads SKILL.md files correctly."""
@@ -269,6 +266,7 @@ def _make_app() -> MagicMock:
app._assistant_id = "agent"
app._discovered_skills = []
app._skill_allowed_roots = []
app._skill_trust_denied = set()
mounted_messages: list[object] = []
app._mounted_messages = mounted_messages
@@ -278,7 +276,14 @@ def _make_app() -> MagicMock:
app._mount_message = AsyncMock(side_effect=capture_mount)
app._handle_user_message = AsyncMock()
app._send_to_agent = AsyncMock()
# Default to deny so the containment error surfaces unless a test opts into
# approval by overriding the return value.
app._push_screen_wait = AsyncMock(return_value=False)
app.notify = MagicMock()
app._invoke_skill = DeepAgentsApp._invoke_skill.__get__(app)
app._prompt_skill_trust_and_retry = (
DeepAgentsApp._prompt_skill_trust_and_retry.__get__(app)
)
app._handle_skill_command = DeepAgentsApp._handle_skill_command.__get__(app)
app._discover_skills_and_roots = DeepAgentsApp._discover_skills_and_roots.__get__(
app
@@ -598,3 +603,401 @@ class TestHandleSkillCommand:
# Cache should be backfilled with fresh discovery results
assert len(app._discovered_skills) == 1
assert app._discovered_skills[0]["name"] == "new-skill"
class TestPromptSkillTrustAndRetry:
"""Cover the in-the-moment trust prompt routed to on containment failure.
All tests use a cache-hit (`_discovered_skills` pre-populated) so only the
load + trust-prompt path runs, and drive it end-to-end through
`_invoke_skill`. `_push_screen_wait` defaults to deny in `_make_app`; tests
that approve override its return value. The first `load_skill_content` call
raises the containment `PermissionError`; the second is the post-approval
retry.
"""
_CONTAINMENT_ERROR = PermissionError(
"Skill path /tmp/evil resolves outside all allowed skill directories."
)
@staticmethod
def _target_dir() -> str:
"""Resolved parent dir of the fake skill's SKILL.md (the trust key)."""
return str(Path(_fake_skill()["path"]).resolve().parent) # ty: ignore
def _cache_hit_app(self) -> MagicMock:
app = _make_app()
app._discovered_skills = [_fake_skill()]
return app
async def test_allow_persists_and_retries(self) -> None:
"""Approving trusts the dir, extends the allowlist, and reads the skill."""
app = self._cache_hit_app()
app._push_screen_wait = AsyncMock(return_value=True)
with (
patch(
"deepagents_code.skills.load.load_skill_content",
side_effect=[self._CONTAINMENT_ERROR, "# Instructions\nDo stuff"],
),
patch(
"deepagents_code.skills.trust.trust_skill_dir", return_value=True
) as mock_trust,
):
await app._handle_skill_command("/skill:test-skill")
mock_trust.assert_called_once_with(self._target_dir())
app._send_to_agent.assert_awaited_once()
assert "# Instructions" in app._send_to_agent.call_args[0][0]
# The approved directory joins the in-session containment allowlist.
assert Path(self._target_dir()) in app._skill_allowed_roots
async def test_allow_extends_allowlist_exactly_once(self) -> None:
"""The approved dir is appended once, not duplicated across the two lists."""
app = self._cache_hit_app()
app._push_screen_wait = AsyncMock(return_value=True)
with (
patch(
"deepagents_code.skills.load.load_skill_content",
side_effect=[self._CONTAINMENT_ERROR, "# Body"],
),
patch("deepagents_code.skills.trust.trust_skill_dir", return_value=True),
):
await app._handle_skill_command("/skill:test-skill")
assert app._skill_allowed_roots.count(Path(self._target_dir())) == 1
async def test_deny_shows_error_and_remembers(self) -> None:
"""Denying surfaces the original error and suppresses re-prompts."""
app = self._cache_hit_app()
app._push_screen_wait = AsyncMock(return_value=False)
with patch(
"deepagents_code.skills.load.load_skill_content",
side_effect=self._CONTAINMENT_ERROR,
):
await app._handle_skill_command("/skill:test-skill")
assert any("resolves outside" in t for t in _app_message_texts(app))
assert self._target_dir() in app._skill_trust_denied
app._send_to_agent.assert_not_awaited()
async def test_prior_deny_does_not_reprompt(self) -> None:
"""A dir denied earlier this session errors without a second prompt."""
app = self._cache_hit_app()
app._skill_trust_denied.add(self._target_dir())
app._push_screen_wait = AsyncMock(return_value=True)
with patch(
"deepagents_code.skills.load.load_skill_content",
side_effect=self._CONTAINMENT_ERROR,
):
await app._handle_skill_command("/skill:test-skill")
app._push_screen_wait.assert_not_awaited()
assert any("resolves outside" in t for t in _app_message_texts(app))
app._send_to_agent.assert_not_awaited()
async def test_retry_permission_error_flags_location_change(self) -> None:
"""A retry containment failure (symlink swap) shows a distinct message."""
app = self._cache_hit_app()
app._push_screen_wait = AsyncMock(return_value=True)
with (
patch(
"deepagents_code.skills.load.load_skill_content",
side_effect=[self._CONTAINMENT_ERROR, self._CONTAINMENT_ERROR],
),
patch("deepagents_code.skills.trust.trust_skill_dir", return_value=True),
):
await app._handle_skill_command("/skill:test-skill")
assert any("location changed" in t.lower() for t in _app_message_texts(app))
app._send_to_agent.assert_not_awaited()
async def test_retry_returns_none_shows_read_error(self) -> None:
"""A readable-but-empty retry result shows the read-failure message."""
app = self._cache_hit_app()
app._push_screen_wait = AsyncMock(return_value=True)
with (
patch(
"deepagents_code.skills.load.load_skill_content",
side_effect=[self._CONTAINMENT_ERROR, None],
),
patch("deepagents_code.skills.trust.trust_skill_dir", return_value=True),
):
await app._handle_skill_command("/skill:test-skill")
assert any("could not read" in t.lower() for t in _app_message_texts(app))
app._send_to_agent.assert_not_awaited()
async def test_persist_failure_still_allows_but_notifies(self) -> None:
"""When the trust store can't be written, we read this session and warn."""
app = self._cache_hit_app()
app._push_screen_wait = AsyncMock(return_value=True)
with (
patch(
"deepagents_code.skills.load.load_skill_content",
side_effect=[self._CONTAINMENT_ERROR, "# Body"],
),
patch("deepagents_code.skills.trust.trust_skill_dir", return_value=False),
):
await app._handle_skill_command("/skill:test-skill")
app.notify.assert_called_once()
assert app.notify.call_args.kwargs.get("severity") == "warning"
app._send_to_agent.assert_awaited_once()
async def test_retry_os_error_shows_generic_read_failure(self) -> None:
"""A transient FS error on the post-approval retry surfaces distinctly."""
app = self._cache_hit_app()
app._push_screen_wait = AsyncMock(return_value=True)
with (
patch(
"deepagents_code.skills.load.load_skill_content",
side_effect=[self._CONTAINMENT_ERROR, OSError("disk gone")],
),
patch("deepagents_code.skills.trust.trust_skill_dir", return_value=True),
):
await app._handle_skill_command("/skill:test-skill")
assert any("after granting trust" in t.lower() for t in _app_message_texts(app))
app._send_to_agent.assert_not_awaited()
async def test_resolve_failure_fails_closed_without_prompting(self) -> None:
"""If resolving the skill path errors, refuse without prompting.
A symlink loop introduced after the first containment check makes
`_resolve_parent_dir` raise; the flow must mount the original error and
never reach the trust prompt rather than crashing the worker.
"""
app = self._cache_hit_app()
app._push_screen_wait = AsyncMock(return_value=True)
with (
patch(
"deepagents_code.skills.load.load_skill_content",
side_effect=self._CONTAINMENT_ERROR,
),
patch(
"deepagents_code.app._resolve_parent_dir",
side_effect=OSError("ELOOP"),
),
):
await app._handle_skill_command("/skill:test-skill")
assert any("resolves outside" in t for t in _app_message_texts(app))
app._push_screen_wait.assert_not_awaited()
app._send_to_agent.assert_not_awaited()
async def test_modal_push_failure_fails_closed(self) -> None:
"""If the trust modal itself fails to display, treat it as a deny.
The prompt runs inside `_invoke_skill`'s `except PermissionError` block,
so an error escaping the push would not be caught by that method's own
handlers and would surface as an unhandled worker crash. It must fail
closed: mount the original error, remember the deny, and not proceed.
"""
app = self._cache_hit_app()
app._push_screen_wait = AsyncMock(side_effect=RuntimeError("screen stack"))
with patch(
"deepagents_code.skills.load.load_skill_content",
side_effect=self._CONTAINMENT_ERROR,
):
await app._handle_skill_command("/skill:test-skill")
assert any("resolves outside" in t for t in _app_message_texts(app))
assert self._target_dir() in app._skill_trust_denied
app._send_to_agent.assert_not_awaited()
async def test_allowlisted_dir_not_reprompted_same_session(self) -> None:
"""Once approved, a later invocation loads without a second prompt.
The approved directory joins the in-session containment allowlist, so a
subsequent read that no longer trips containment must not re-nag.
"""
app = self._cache_hit_app()
app._push_screen_wait = AsyncMock(return_value=True)
with (
patch(
"deepagents_code.skills.load.load_skill_content",
# 1st invoke: containment fails then the retry succeeds.
# 2nd invoke: loads directly (dir already allowlisted).
side_effect=[self._CONTAINMENT_ERROR, "# Body", "# Body"],
),
patch("deepagents_code.skills.trust.trust_skill_dir", return_value=True),
):
await app._handle_skill_command("/skill:test-skill")
await app._handle_skill_command("/skill:test-skill")
assert app._push_screen_wait.await_count == 1
assert app._send_to_agent.await_count == 2
class TestDiscoverSkillsAndRoots:
"""The containment roots must include persisted trust and extra dirs.
This is the join that makes an in-session approval survive a relaunch. The
real `discover_skills_and_roots` is exercised (not mocked wholesale) so a
regression that drops the trust join, or re-resolves it instead of adding it
as-is, is caught.
"""
def _settings_with_no_builtin_roots(self, extra: list[Path]) -> MagicMock:
settings = MagicMock()
for getter in (
"get_built_in_skills_dir",
"get_user_skills_dir",
"get_project_skills_dir",
"get_user_agent_skills_dir",
"get_project_agent_skills_dir",
"get_user_claude_skills_dir",
"get_project_claude_skills_dir",
):
getattr(settings, getter).return_value = None
settings.get_extra_skills_dirs.return_value = extra
return settings
def test_persisted_trust_added_as_is_while_extra_dirs_resolved(
self, tmp_path: Path
) -> None:
"""Trusted dirs join roots verbatim; `extra_allowed_dirs` are resolved.
A symlinked entry makes the two code paths distinguishable: re-resolving
the trusted entry (the regression this guards) would follow the symlink
to a directory the user never approved, so it must be added as-is. The
declarative `extra_allowed_dirs` allowlist is intentionally resolved. A
real (non-symlink) dir could not tell these apart because `resolve()`
would be idempotent.
"""
from deepagents_code.skills.invocation import discover_skills_and_roots
real_trusted = tmp_path / "real_trusted"
real_trusted.mkdir()
link_trusted = tmp_path / "link_trusted"
link_trusted.symlink_to(real_trusted, target_is_directory=True)
real_extra = tmp_path / "real_extra"
real_extra.mkdir()
link_extra = tmp_path / "link_extra"
link_extra.symlink_to(real_extra, target_is_directory=True)
with (
patch(
"deepagents_code.config.settings",
self._settings_with_no_builtin_roots([link_extra]),
),
patch("deepagents_code.skills.load.list_skills", return_value=[]),
patch(
"deepagents_code.skills.trust.load_trusted_skill_dirs",
return_value=[link_trusted],
),
):
_skills, roots = discover_skills_and_roots("agent")
# Trusted entry added verbatim — not followed to its symlink target.
assert link_trusted in roots
assert real_trusted.resolve() not in roots
# `extra_allowed_dirs` is the declarative allowlist and is resolved.
assert real_extra.resolve() in roots
class TestResolveParentDir:
"""`_resolve_parent_dir` keys trust on the resolved target, not the link."""
def test_resolves_then_takes_parent(self, tmp_path: Path) -> None:
"""A symlinked SKILL.md keys on the real target's parent directory.
`resolve().parent` (what the code does) and `parent.resolve()` diverge
when the SKILL.md file is itself a symlink and its discovery directory
is a *real* directory: the former yields the symlink target's parent,
the latter the discovery directory. The trust key must match what
`load_skill_content` enforces (it resolves the file), i.e. the resolved
target's parent — so the retry's containment check can pass.
"""
from deepagents_code.app import _resolve_parent_dir
disc = tmp_path / "disc"
disc.mkdir()
real = tmp_path / "real"
real.mkdir()
(real / "SKILL.md").write_text("# body", encoding="utf-8")
link = disc / "SKILL.md"
link.symlink_to(real / "SKILL.md")
assert _resolve_parent_dir(link) == str(real.resolve())
# Not the (real) discovery directory the link lexically sits in.
assert _resolve_parent_dir(link) != str(disc.resolve())
class TestSkillTrustRealContainment:
"""Drive `_invoke_skill` through the REAL `load_skill_content` gate.
Unlike `TestPromptSkillTrustAndRetry` (which mocks `load_skill_content`),
these build real directories and symlinks so the actual containment check
runs on both the initial load and the post-approval retry — the path a mock
cannot exercise, where a wrong/empty `allowed_roots` would silently read a
swapped target. `trust_skill_dir` is patched so no real state dir is written.
"""
@staticmethod
def _outside_skill(tmp_path: Path) -> tuple[MagicMock, Path, Path]:
"""Return (app, allowed root, outside target) for an escaping skill.
The discovered skill's SKILL.md lives under `root` (the sole allowed
root) but symlinks out to `outside/SKILL.md`, so the initial real
containment check refuses it.
"""
root = tmp_path / "skills"
root.mkdir()
outside = tmp_path / "outside"
outside.mkdir()
(outside / "SKILL.md").write_text(
"# Real body\nDo real stuff", encoding="utf-8"
)
link = root / "linked-skill"
link.symlink_to(outside, target_is_directory=True)
app = _make_app()
app._skill_allowed_roots = [root.resolve()]
app._discovered_skills = [
_fake_skill(name="linked-skill", path=str(link / "SKILL.md"))
]
return app, root, outside
async def test_real_retry_reads_after_approval(self, tmp_path: Path) -> None:
"""Approving an outside skill reads it through the real gate on retry."""
app, _root, outside = self._outside_skill(tmp_path)
app._push_screen_wait = AsyncMock(return_value=True)
with patch("deepagents_code.skills.trust.trust_skill_dir", return_value=True):
await app._handle_skill_command("/skill:linked-skill")
app._send_to_agent.assert_awaited_once()
assert "Do real stuff" in app._send_to_agent.call_args[0][0]
# The resolved outside dir was actually admitted to the allowlist by the
# real retry, not just asserted via a mock.
assert outside.resolve() in app._skill_allowed_roots
async def test_real_retry_refuses_symlink_swap_during_prompt(
self, tmp_path: Path
) -> None:
"""Re-pointing the discovery symlink during the prompt is refused.
The user approves the originally-resolved target dir, but the skill's
SKILL.md is re-pointed to an unapproved location before the retry runs.
The real containment check must refuse the swapped target rather than
read it, even though a directory *was* just approved.
"""
app, root, _outside = self._outside_skill(tmp_path)
link = root / "linked-skill"
evil = tmp_path / "evil"
evil.mkdir()
(evil / "SKILL.md").write_text("# stolen", encoding="utf-8")
def approve_then_swap(_screen: object) -> bool:
# Swap the discovery symlink to an unapproved target inside the
# prompt window, then approve the (original) resolved dir.
link.unlink()
link.symlink_to(evil, target_is_directory=True)
return True
app._push_screen_wait = AsyncMock(side_effect=approve_then_swap)
with patch("deepagents_code.skills.trust.trust_skill_dir", return_value=True):
await app._handle_skill_command("/skill:linked-skill")
assert any("location changed" in t.lower() for t in _app_message_texts(app))
app._send_to_agent.assert_not_awaited()
libs/code/tests/unit_tests/test_skill_trust_screen.py
+78
View File
@@ -0,0 +1,78 @@
"""Tests for the `SkillTrustScreen` out-of-bounds-skill approval modal."""
from __future__ import annotations
from textual.app import App, ComposeResult
from textual.widgets import Static
from deepagents_code.widgets.skill_trust import SkillTrustScreen
class _SkillTrustTestApp(App[None]):
def compose(self) -> ComposeResult:
yield Static("base")
class TestSkillTrustScreen:
"""Behavior tests for `SkillTrustScreen`.
The Enter=allow / Esc=deny mapping is a security default (Esc must never
grant read access), so each route is pinned explicitly.
"""
async def test_enter_dismisses_with_true(self) -> None:
"""Pressing Enter approves the out-of-bounds directory."""
app = _SkillTrustTestApp()
async with app.run_test() as pilot:
outcomes: list[bool | None] = []
app.push_screen(
SkillTrustScreen("demo", "/shared/skills/demo"), outcomes.append
)
await pilot.pause()
await pilot.press("enter")
await pilot.pause()
assert outcomes == [True]
async def test_escape_dismisses_with_false(self) -> None:
"""Pressing Esc denies (never an implicit approval)."""
app = _SkillTrustTestApp()
async with app.run_test() as pilot:
outcomes: list[bool | None] = []
app.push_screen(
SkillTrustScreen("demo", "/shared/skills/demo"), outcomes.append
)
await pilot.pause()
await pilot.press("escape")
await pilot.pause()
assert outcomes == [False]
async def test_action_cancel_dismisses_with_false(self) -> None:
"""`action_cancel` denies — the path the app's priority Esc binding takes.
The app owns a priority `escape` binding that, for an active
`ModalScreen`, dispatches to `action_cancel` if present and otherwise
falls through to `dismiss(None)`. Without an `action_cancel` returning
`False`, real-app Esc would silently None-dismiss instead of an explicit
deny.
"""
app = _SkillTrustTestApp()
async with app.run_test() as pilot:
outcomes: list[bool | None] = []
screen = SkillTrustScreen("demo", "/shared/skills/demo")
app.push_screen(screen, outcomes.append)
await pilot.pause()
screen.action_cancel()
await pilot.pause()
assert outcomes == [False]
async def test_renders_skill_name_and_target(self) -> None:
"""The skill name and resolved target directory are surfaced in the body."""
app = _SkillTrustTestApp()
async with app.run_test() as pilot:
app.push_screen(SkillTrustScreen("demo", "/shared/skills/demo"))
await pilot.pause()
bodies = app.screen.query(".skill-trust-body")
assert len(bodies) == 1
rendered = str(bodies.first().render())
assert "demo" in rendered
assert "/shared/skills/demo" in rendered