Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 16 additions & 7 deletions flocks/tool/skill/flocks_skills.py
Original file line number Diff line number Diff line change
Expand Up @@ -89,6 +89,9 @@
)
_SUBCOMMAND_ENUM = ["find", "install", "status", "install-deps", "list", "remove"]

# Read-only registry / discovery — no shell side effects; skip bash permission gate.
_READ_ONLY_SUBCOMMANDS = frozenset({"find", "list", "status"})


def _flocks_executable() -> Optional[str]:
"""Locate the `flocks` CLI on PATH."""
Expand Down Expand Up @@ -158,13 +161,19 @@ async def flocks_skills(

log.info("flocks_skills.run", {"cmd": cmd})

# Request permission (uses the same "bash" permission type as bash_tool).
await ctx.ask(
permission="bash",
patterns=[" ".join(cmd)],
always=["flocks skills *"],
metadata={"subcommand": subcommand},
)
# Mutating subcommands need bash approval. Read-only (find/list/status) runs
# without prompting — same trust model as listing skills in the UI.
#
# For install/remove/install-deps, always-patterns must match the *full*
# argv string (e.g. "/opt/flocks/bin/flocks skills install ..."); a bare
# "flocks skills *" fails fnmatch and never auto-approved.
if subcommand not in _READ_ONLY_SUBCOMMANDS:
await ctx.ask(
permission="bash",
patterns=[" ".join(cmd)],
always=["*flocks skills *"],
metadata={"subcommand": subcommand},
)

try:
proc = await asyncio.create_subprocess_exec(
Expand Down
21 changes: 15 additions & 6 deletions tests/tool/test_flocks_skills.py
Original file line number Diff line number Diff line change
Expand Up @@ -77,18 +77,26 @@ async def test_unknown_subcommand_returns_error():
@pytest.mark.asyncio
async def test_all_allowed_subcommands_accepted():
"""Ensure none of the valid subcommands are rejected by the allow-list."""
from flocks.tool.skill.flocks_skills import flocks_skills, _ALLOWED_SUBCOMMANDS
from flocks.tool.skill.flocks_skills import (
flocks_skills,
_ALLOWED_SUBCOMMANDS,
_READ_ONLY_SUBCOMMANDS,
)

proc = make_proc(stdout=b"ok", returncode=0)
ctx = make_ctx()

with (
patch("flocks.tool.skill.flocks_skills._flocks_executable", return_value="/usr/local/bin/flocks"),
patch("flocks.tool.skill.flocks_skills.asyncio.create_subprocess_exec", return_value=proc),
):
for sub in _ALLOWED_SUBCOMMANDS:
ctx = make_ctx()
result = await flocks_skills(ctx, subcommand=sub, args="")
assert result.success is True, f"subcommand {sub!r} should succeed"
if sub in _READ_ONLY_SUBCOMMANDS:
ctx.ask.assert_not_called()
else:
ctx.ask.assert_called_once()


# ---------------------------------------------------------------------------
Expand Down Expand Up @@ -129,10 +137,8 @@ async def test_list_success():
cmd_args = mock_exec.call_args[0]
assert "skills" in cmd_args
assert "list" in cmd_args
# Permission must be requested for every execution
ctx.ask.assert_called_once()
call_kwargs = ctx.ask.call_args[1]
assert call_kwargs.get("permission") == "bash"
# list is read-only — no bash permission prompt
ctx.ask.assert_not_called()


@pytest.mark.asyncio
Expand All @@ -153,6 +159,7 @@ async def test_find_passes_args():
assert "find" in cmd_args
assert "phishing" in cmd_args
assert "analysis" in cmd_args
ctx.ask.assert_not_called()


# ---------------------------------------------------------------------------
Expand All @@ -174,6 +181,7 @@ async def test_nonzero_exit_returns_failure():

assert result.success is False
assert "skill not found" in (result.error or "")
ctx.ask.assert_called_once()


# ---------------------------------------------------------------------------
Expand Down Expand Up @@ -203,6 +211,7 @@ async def test_timeout_kills_process():
assert result.success is False
assert "timed out" in (result.error or "").lower()
proc.kill.assert_called_once()
ctx.ask.assert_called_once()


# ---------------------------------------------------------------------------
Expand Down
Loading