mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-11-09 04:20:34 +00:00
7d805cd1b8
* fix(skills): reject Windows device names when creating a skill
A skill name is a `custom/<name>` directory component, but the hyphen-case
grammar admits Windows reserved device names (`nul`, `con`, `com1`, ...).
`skill_manage` create and `.skill` archive install accepted them: on Windows
`mkdir("nul")` reports success without producing a directory and the first
write inside it fails with FileNotFoundError; on Linux the install succeeds
but the tree can never be copied, mounted or exported to a Windows host.
Creation now rejects those names through `windows_incompatible_segment()`, the
check uploads, sandbox writes, skill support files, thread IDs and agent names
already use. Archive extraction also rejects member segments that cannot be
created on every host. Reads and deletes stay permissive so a skill an older
release created on a Linux host can still be inspected and removed.
Fixes #7945
* test(skills): pin the extraction guard and the tool-level create wiring
Address review findings on the skill-name portability change:
- Isolate the extraction guard: an archive with a portable folder and a
portable frontmatter name but a reserved support-file segment
(`my-skill/references/nul.md`) now fails with the member-path error, so
deleting the guard turns this suite red again.
- Pin the `skill_manage` wiring: `create` of a device name raises through the
tool, while `edit`/`delete` keep reaching the existing "not found" path.
- Reword the `safe_extract_skill_archive` docstring: members before the
offending one are already written into staging, so what the check
guarantees is that the install never commits under `custom/`.
---------
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
811 lines
33 KiB
Python
811 lines
33 KiB
Python
import asyncio
|
|
import gc
|
|
import importlib
|
|
import threading
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import anyio
|
|
import pytest
|
|
|
|
from deerflow.skills.security_static_scanner import StaticScannerError
|
|
|
|
skill_manage_module = importlib.import_module("deerflow.tools.skill_manage_tool")
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _record_skills_cache_reset_publications(monkeypatch):
|
|
"""Keep the cross-replica marker out of whatever extensions config this checkout resolves.
|
|
|
|
Every mutation publishes it; tests/test_skills_cache_shared_reset.py covers
|
|
the real publication against a temporary shared config.
|
|
"""
|
|
published: list[str | None] = []
|
|
monkeypatch.setattr(skill_manage_module, "publish_skills_cache_reset", lambda *, user_id=None: published.append(user_id))
|
|
return published
|
|
|
|
|
|
def _skill_content(name: str, description: str = "Demo skill") -> str:
|
|
return f"---\nname: {name}\ndescription: {description}\n---\n\n# {name}\n"
|
|
|
|
|
|
async def _async_result(decision: str, reason: str):
|
|
from deerflow.skills.security_scanner import ScanResult
|
|
|
|
return ScanResult(decision=decision, reason=reason)
|
|
|
|
|
|
def _make_config(skills_root: Path):
|
|
return SimpleNamespace(
|
|
skills=SimpleNamespace(
|
|
get_skills_path=lambda: skills_root,
|
|
container_path="/mnt/skills",
|
|
use="deerflow.skills.storage.local_skill_storage:LocalSkillStorage",
|
|
),
|
|
skill_evolution=SimpleNamespace(enabled=True, moderation_model_name=None),
|
|
)
|
|
|
|
|
|
def _make_runtime(*, thread_id: str = "thread-1", user_id: str = "default"):
|
|
return SimpleNamespace(
|
|
context={"thread_id": thread_id, "user_id": user_id},
|
|
config={"configurable": {"thread_id": thread_id, "user_id": user_id}},
|
|
)
|
|
|
|
|
|
def test_skill_manage_create_and_patch(monkeypatch, tmp_path):
|
|
skills_root = tmp_path / "skills"
|
|
config = _make_config(skills_root)
|
|
monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
|
|
monkeypatch.setattr("deerflow.skills.security_scanner.get_app_config", lambda: config)
|
|
# Patch get_paths so UserScopedSkillStorage resolves user dirs under tmp_path
|
|
from deerflow.config.paths import Paths
|
|
|
|
monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path))
|
|
monkeypatch.setattr("deerflow.config.paths._paths", None)
|
|
|
|
refresh_calls = []
|
|
|
|
async def _refresh(user_id: str):
|
|
refresh_calls.append(("refresh", user_id))
|
|
|
|
monkeypatch.setattr(skill_manage_module, "refresh_user_skills_system_prompt_cache_async", _refresh)
|
|
monkeypatch.setattr(skill_manage_module, "scan_skill_content", lambda *args, **kwargs: _async_result("allow", "ok"))
|
|
|
|
runtime = _make_runtime(user_id="default")
|
|
|
|
result = anyio.run(
|
|
skill_manage_module.skill_manage_tool.coroutine,
|
|
runtime,
|
|
"create",
|
|
"demo-skill",
|
|
_skill_content("demo-skill"),
|
|
)
|
|
assert "Created custom skill" in result
|
|
|
|
patch_result = anyio.run(
|
|
skill_manage_module.skill_manage_tool.coroutine,
|
|
runtime,
|
|
"patch",
|
|
"demo-skill",
|
|
None,
|
|
None,
|
|
"Demo skill",
|
|
"Patched skill",
|
|
1,
|
|
)
|
|
assert "Patched custom skill" in patch_result
|
|
# User-scoped: custom skills written under users/default/skills/custom/
|
|
user_custom = tmp_path / "users" / "default" / "skills" / "custom"
|
|
assert "Patched skill" in (user_custom / "demo-skill" / "SKILL.md").read_text(encoding="utf-8")
|
|
assert refresh_calls == [("refresh", "default"), ("refresh", "default")]
|
|
|
|
|
|
def test_skill_manage_patch_replaces_single_occurrence_by_default(monkeypatch, tmp_path):
|
|
skills_root = tmp_path / "skills"
|
|
config = _make_config(skills_root)
|
|
monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
|
|
monkeypatch.setattr("deerflow.skills.security_scanner.get_app_config", lambda: config)
|
|
from deerflow.config.paths import Paths
|
|
|
|
monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path))
|
|
monkeypatch.setattr("deerflow.config.paths._paths", None)
|
|
|
|
async def _refresh(user_id: str):
|
|
return None
|
|
|
|
monkeypatch.setattr(skill_manage_module, "refresh_user_skills_system_prompt_cache_async", _refresh)
|
|
monkeypatch.setattr(skill_manage_module, "scan_skill_content", lambda *args, **kwargs: _async_result("allow", "ok"))
|
|
|
|
runtime = _make_runtime(user_id="default")
|
|
content = _skill_content("demo-skill", "Demo skill") + "\nRepeated: Demo skill\n"
|
|
|
|
anyio.run(skill_manage_module.skill_manage_tool.coroutine, runtime, "create", "demo-skill", content)
|
|
patch_result = anyio.run(
|
|
skill_manage_module.skill_manage_tool.coroutine,
|
|
runtime,
|
|
"patch",
|
|
"demo-skill",
|
|
None,
|
|
None,
|
|
"Demo skill",
|
|
"Patched skill",
|
|
)
|
|
|
|
user_custom = tmp_path / "users" / "default" / "skills" / "custom"
|
|
skill_text = (user_custom / "demo-skill" / "SKILL.md").read_text(encoding="utf-8")
|
|
assert "1 replacement(s) applied, 2 match(es) found" in patch_result
|
|
assert skill_text.count("Patched skill") == 1
|
|
assert skill_text.count("Demo skill") == 1
|
|
|
|
|
|
def test_skill_manage_rejects_public_skill_patch(monkeypatch, tmp_path):
|
|
skills_root = tmp_path / "skills"
|
|
public_dir = skills_root / "public" / "deep-research"
|
|
public_dir.mkdir(parents=True, exist_ok=True)
|
|
(public_dir / "SKILL.md").write_text(_skill_content("deep-research"), encoding="utf-8")
|
|
config = _make_config(skills_root)
|
|
monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
|
|
from deerflow.config.paths import Paths
|
|
|
|
monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path))
|
|
monkeypatch.setattr("deerflow.config.paths._paths", None)
|
|
|
|
runtime = _make_runtime(user_id="default")
|
|
|
|
with pytest.raises(ValueError, match="built-in skill"):
|
|
anyio.run(
|
|
skill_manage_module.skill_manage_tool.coroutine,
|
|
runtime,
|
|
"patch",
|
|
"deep-research",
|
|
None,
|
|
None,
|
|
"Demo skill",
|
|
"Patched",
|
|
)
|
|
|
|
|
|
def test_skill_manage_sync_wrapper_supported(monkeypatch, tmp_path):
|
|
skills_root = tmp_path / "skills"
|
|
config = _make_config(skills_root)
|
|
monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
|
|
from deerflow.config.paths import Paths
|
|
|
|
monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path))
|
|
monkeypatch.setattr("deerflow.config.paths._paths", None)
|
|
|
|
refresh_calls = []
|
|
|
|
async def _refresh(user_id: str):
|
|
refresh_calls.append(("refresh", user_id))
|
|
|
|
monkeypatch.setattr(skill_manage_module, "refresh_user_skills_system_prompt_cache_async", _refresh)
|
|
monkeypatch.setattr(skill_manage_module, "scan_skill_content", lambda *args, **kwargs: _async_result("allow", "ok"))
|
|
|
|
runtime = _make_runtime(thread_id="thread-sync", user_id="default")
|
|
result = skill_manage_module.skill_manage_tool.func(
|
|
runtime=runtime,
|
|
action="create",
|
|
name="sync-skill",
|
|
content=_skill_content("sync-skill"),
|
|
)
|
|
|
|
assert "Created custom skill" in result
|
|
assert refresh_calls == [("refresh", "default")]
|
|
|
|
|
|
def test_skill_manage_rejects_support_path_traversal(monkeypatch, tmp_path):
|
|
skills_root = tmp_path / "skills"
|
|
config = _make_config(skills_root)
|
|
monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
|
|
monkeypatch.setattr("deerflow.skills.security_scanner.get_app_config", lambda: config)
|
|
from deerflow.config.paths import Paths
|
|
|
|
monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path))
|
|
monkeypatch.setattr("deerflow.config.paths._paths", None)
|
|
|
|
async def _refresh(user_id: str):
|
|
return None
|
|
|
|
monkeypatch.setattr(skill_manage_module, "refresh_user_skills_system_prompt_cache_async", _refresh)
|
|
monkeypatch.setattr(skill_manage_module, "scan_skill_content", lambda *args, **kwargs: _async_result("allow", "ok"))
|
|
|
|
runtime = _make_runtime(user_id="default")
|
|
anyio.run(skill_manage_module.skill_manage_tool.coroutine, runtime, "create", "demo-skill", _skill_content("demo-skill"))
|
|
|
|
with pytest.raises(ValueError, match="parent-directory traversal|selected support directory"):
|
|
anyio.run(
|
|
skill_manage_module.skill_manage_tool.coroutine,
|
|
runtime,
|
|
"write_file",
|
|
"demo-skill",
|
|
"malicious overwrite",
|
|
"references/../SKILL.md",
|
|
)
|
|
|
|
|
|
def test_skill_manage_remove_file_updates_sandbox_projection_before_return(monkeypatch, tmp_path):
|
|
skills_root = tmp_path / "skills"
|
|
config = _make_config(skills_root)
|
|
monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
|
|
monkeypatch.setattr("deerflow.skills.security_scanner.get_app_config", lambda: config)
|
|
from deerflow.config.paths import Paths
|
|
|
|
monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path))
|
|
monkeypatch.setattr("deerflow.config.paths._paths", None)
|
|
|
|
async def _refresh(user_id: str):
|
|
return None
|
|
|
|
monkeypatch.setattr(skill_manage_module, "refresh_user_skills_system_prompt_cache_async", _refresh)
|
|
monkeypatch.setattr(skill_manage_module, "scan_skill_content", lambda *args, **kwargs: _async_result("allow", "ok"))
|
|
|
|
runtime = _make_runtime(user_id="default")
|
|
anyio.run(skill_manage_module.skill_manage_tool.coroutine, runtime, "create", "demo-skill", _skill_content("demo-skill"))
|
|
anyio.run(
|
|
skill_manage_module.skill_manage_tool.coroutine,
|
|
runtime,
|
|
"write_file",
|
|
"demo-skill",
|
|
"supporting content",
|
|
"references/guide.md",
|
|
)
|
|
projected_file = tmp_path / "users" / "default" / "skills_view" / "custom" / "demo-skill" / "references" / "guide.md"
|
|
assert projected_file.read_text(encoding="utf-8") == "supporting content"
|
|
|
|
result = anyio.run(
|
|
skill_manage_module.skill_manage_tool.coroutine,
|
|
runtime,
|
|
"remove_file",
|
|
"demo-skill",
|
|
None,
|
|
"references/guide.md",
|
|
)
|
|
|
|
assert result == "Removed 'references/guide.md' from custom skill 'demo-skill'."
|
|
assert not projected_file.exists()
|
|
|
|
|
|
def _setup_demo_skill_with_binary_asset(monkeypatch, tmp_path):
|
|
skills_root = tmp_path / "skills"
|
|
config = _make_config(skills_root)
|
|
monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
|
|
monkeypatch.setattr("deerflow.skills.security_scanner.get_app_config", lambda: config)
|
|
from deerflow.config.paths import Paths
|
|
|
|
monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path))
|
|
monkeypatch.setattr("deerflow.config.paths._paths", None)
|
|
|
|
async def _refresh(user_id: str):
|
|
return None
|
|
|
|
monkeypatch.setattr(skill_manage_module, "refresh_user_skills_system_prompt_cache_async", _refresh)
|
|
monkeypatch.setattr(skill_manage_module, "scan_skill_content", lambda *args, **kwargs: _async_result("allow", "ok"))
|
|
runtime = _make_runtime(user_id="default")
|
|
anyio.run(skill_manage_module.skill_manage_tool.coroutine, runtime, "create", "demo-skill", _skill_content("demo-skill"))
|
|
# A binary support file as a .skill archive install would leave it (the
|
|
# installer only rejects *executable* binaries).
|
|
skill_dir = tmp_path / "users" / "default" / "skills" / "custom" / "demo-skill"
|
|
(skill_dir / "assets" / "nested").mkdir(parents=True)
|
|
(skill_dir / "assets" / "logo.png").write_bytes(b"\x89PNG\r\n\x1a\n\xff\xfe\x00")
|
|
(skill_dir / "assets" / "nested" / "inner.txt").write_text("inner", encoding="utf-8")
|
|
return runtime, skill_dir
|
|
|
|
|
|
def _history_records(skill_dir: Path) -> list[dict]:
|
|
import json
|
|
|
|
history_file = skill_dir.parent.parent / "history" / "demo-skill.jsonl"
|
|
if not history_file.exists(): # storage layouts differ; fall back to a search
|
|
history_file = next(p for p in skill_dir.parent.parent.rglob("*.jsonl") if "demo-skill" in p.name)
|
|
return [json.loads(line) for line in history_file.read_text(encoding="utf-8").splitlines() if line.strip()]
|
|
|
|
|
|
def test_skill_manage_remove_file_handles_a_binary_support_file(monkeypatch, tmp_path):
|
|
runtime, skill_dir = _setup_demo_skill_with_binary_asset(monkeypatch, tmp_path)
|
|
|
|
result = anyio.run(skill_manage_module.skill_manage_tool.coroutine, runtime, "remove_file", "demo-skill", None, "assets/logo.png")
|
|
|
|
assert result == "Removed 'assets/logo.png' from custom skill 'demo-skill'."
|
|
assert not (skill_dir / "assets" / "logo.png").exists()
|
|
record = _history_records(skill_dir)[-1]
|
|
assert record["action"] == "remove_file"
|
|
assert record["file_path"] == "assets/logo.png"
|
|
assert record["prev_content"] is None
|
|
|
|
|
|
def test_skill_manage_write_file_can_overwrite_a_binary_support_file(monkeypatch, tmp_path):
|
|
runtime, skill_dir = _setup_demo_skill_with_binary_asset(monkeypatch, tmp_path)
|
|
|
|
result = anyio.run(skill_manage_module.skill_manage_tool.coroutine, runtime, "write_file", "demo-skill", "now text", "assets/logo.png")
|
|
|
|
assert result == "Wrote 'assets/logo.png' for custom skill 'demo-skill'."
|
|
assert (skill_dir / "assets" / "logo.png").read_text(encoding="utf-8") == "now text"
|
|
record = _history_records(skill_dir)[-1]
|
|
assert record["action"] == "write_file"
|
|
assert record["prev_content"] is None
|
|
assert record["new_content"] == "now text"
|
|
|
|
|
|
@pytest.mark.parametrize("action, content", [("remove_file", None), ("write_file", "x")])
|
|
@pytest.mark.parametrize("path", ["assets", "assets/nested"], ids=["bare-subdir", "nested-dir"])
|
|
def test_skill_manage_rejects_a_support_directory_path(monkeypatch, tmp_path, action, content, path):
|
|
runtime, skill_dir = _setup_demo_skill_with_binary_asset(monkeypatch, tmp_path)
|
|
|
|
with pytest.raises(ValueError, match="file"):
|
|
anyio.run(skill_manage_module.skill_manage_tool.coroutine, runtime, action, "demo-skill", content, path)
|
|
|
|
assert (skill_dir / "assets" / "logo.png").exists()
|
|
assert (skill_dir / "assets" / "nested" / "inner.txt").read_text(encoding="utf-8") == "inner"
|
|
|
|
|
|
def test_skill_manage_static_critical_blocks_create_before_llm(monkeypatch, tmp_path):
|
|
skills_root = tmp_path / "skills"
|
|
config = _make_config(skills_root)
|
|
monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
|
|
monkeypatch.setattr("deerflow.skills.security_scanner.get_app_config", lambda: config)
|
|
from deerflow.config.paths import Paths
|
|
|
|
monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path))
|
|
monkeypatch.setattr("deerflow.config.paths._paths", None)
|
|
refresh_calls = []
|
|
llm_calls = []
|
|
|
|
async def _refresh(user_id: str):
|
|
refresh_calls.append(("refresh", user_id))
|
|
|
|
async def _scan(*args, **kwargs):
|
|
llm_calls.append({"args": args, "kwargs": kwargs})
|
|
return await _async_result("allow", "ok")
|
|
|
|
monkeypatch.setattr(skill_manage_module, "refresh_user_skills_system_prompt_cache_async", _refresh)
|
|
monkeypatch.setattr(skill_manage_module, "scan_skill_content", _scan)
|
|
|
|
runtime = _make_runtime(user_id="default")
|
|
content = _skill_content("blocked-skill") + "\n-----BEGIN RSA PRIVATE KEY-----\nabc\n-----END RSA PRIVATE KEY-----\n"
|
|
|
|
with pytest.raises(ValueError) as excinfo:
|
|
anyio.run(
|
|
skill_manage_module.skill_manage_tool.coroutine,
|
|
runtime,
|
|
"create",
|
|
"blocked-skill",
|
|
content,
|
|
)
|
|
|
|
assert "Static security scan blocked" in str(excinfo.value)
|
|
assert "secret-private-key" in str(excinfo.value)
|
|
assert llm_calls == []
|
|
assert refresh_calls == []
|
|
assert not (tmp_path / "users" / "default" / "skills" / "custom" / "blocked-skill" / "SKILL.md").exists()
|
|
|
|
|
|
def test_skill_manage_static_scan_failure_blocks_create_before_llm(monkeypatch, tmp_path):
|
|
skills_root = tmp_path / "skills"
|
|
config = _make_config(skills_root)
|
|
monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
|
|
monkeypatch.setattr("deerflow.skills.security_scanner.get_app_config", lambda: config)
|
|
from deerflow.config.paths import Paths
|
|
|
|
monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path))
|
|
monkeypatch.setattr("deerflow.config.paths._paths", None)
|
|
refresh_calls = []
|
|
llm_calls = []
|
|
|
|
async def _refresh(user_id: str):
|
|
refresh_calls.append(("refresh", user_id))
|
|
|
|
async def _scan(*args, **kwargs):
|
|
llm_calls.append({"args": args, "kwargs": kwargs})
|
|
return await _async_result("allow", "ok")
|
|
|
|
def _broken_static_scan(skill_dir, *, skill_name=None, app_config=None):
|
|
raise StaticScannerError("native scanner unavailable")
|
|
|
|
monkeypatch.setattr(skill_manage_module, "refresh_user_skills_system_prompt_cache_async", _refresh)
|
|
monkeypatch.setattr(skill_manage_module, "scan_skill_content", _scan)
|
|
monkeypatch.setattr(skill_manage_module, "enforce_static_scan", _broken_static_scan)
|
|
|
|
runtime = _make_runtime(user_id="default")
|
|
|
|
with pytest.raises(ValueError, match="Static security scan failed.*native scanner unavailable"):
|
|
anyio.run(
|
|
skill_manage_module.skill_manage_tool.coroutine,
|
|
runtime,
|
|
"create",
|
|
"scanner-failure-skill",
|
|
_skill_content("scanner-failure-skill"),
|
|
)
|
|
|
|
assert llm_calls == []
|
|
assert refresh_calls == []
|
|
assert not (tmp_path / "users" / "default" / "skills" / "custom" / "scanner-failure-skill" / "SKILL.md").exists()
|
|
|
|
|
|
def test_skill_manage_per_user_isolation(monkeypatch, tmp_path):
|
|
"""Two different users must get separate custom skill directories."""
|
|
skills_root = tmp_path / "skills"
|
|
config = _make_config(skills_root)
|
|
monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
|
|
monkeypatch.setattr("deerflow.skills.security_scanner.get_app_config", lambda: config)
|
|
from deerflow.config.paths import Paths
|
|
|
|
monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path))
|
|
monkeypatch.setattr("deerflow.config.paths._paths", None)
|
|
|
|
async def _refresh(user_id: str):
|
|
return None
|
|
|
|
monkeypatch.setattr(skill_manage_module, "refresh_user_skills_system_prompt_cache_async", _refresh)
|
|
monkeypatch.setattr(skill_manage_module, "scan_skill_content", lambda *args, **kwargs: _async_result("allow", "ok"))
|
|
|
|
# Alice creates a skill
|
|
runtime_alice = _make_runtime(user_id="alice")
|
|
result_a = anyio.run(
|
|
skill_manage_module.skill_manage_tool.coroutine,
|
|
runtime_alice,
|
|
"create",
|
|
"alice-skill",
|
|
_skill_content("alice-skill"),
|
|
)
|
|
assert "Created custom skill" in result_a
|
|
|
|
# Bob creates a different skill
|
|
runtime_bob = _make_runtime(user_id="bob")
|
|
result_b = anyio.run(
|
|
skill_manage_module.skill_manage_tool.coroutine,
|
|
runtime_bob,
|
|
"create",
|
|
"bob-skill",
|
|
_skill_content("bob-skill"),
|
|
)
|
|
assert "Created custom skill" in result_b
|
|
|
|
# Verify separate directories
|
|
alice_dir = tmp_path / "users" / "alice" / "skills" / "custom" / "alice-skill"
|
|
bob_dir = tmp_path / "users" / "bob" / "skills" / "custom" / "bob-skill"
|
|
assert alice_dir.exists()
|
|
assert bob_dir.exists()
|
|
# No cross-contamination
|
|
assert not (tmp_path / "users" / "alice" / "skills" / "custom" / "bob-skill").exists()
|
|
assert not (tmp_path / "users" / "bob" / "skills" / "custom" / "alice-skill").exists()
|
|
|
|
|
|
# --- tracing wiring: the in-graph choke point (see the INVARIANT in
|
|
# packages/harness/deerflow/agents/lead_agent/agent.py) ---
|
|
|
|
|
|
def test_scan_or_raise_does_not_attach_model_tracing(monkeypatch, tmp_path):
|
|
"""``_scan_or_raise`` is the in-graph choke point for the skill security scan.
|
|
|
|
The graph root already attached the tracing callbacks, so the scan model must
|
|
not attach them again: double-attaching emits duplicate spans and blocks the
|
|
Langfuse handler's ``propagate_attributes`` path, so session_id/user_id never
|
|
reach the trace. Drives the real ``scan_skill_content`` rather than stubbing it,
|
|
so the flag is pinned all the way to the model factory.
|
|
"""
|
|
config = _make_config(tmp_path / "skills")
|
|
monkeypatch.setattr("deerflow.skills.security_scanner.get_app_config", lambda: config)
|
|
|
|
create_kwargs = {}
|
|
|
|
class FakeModel:
|
|
async def ainvoke(self, *args, **kwargs):
|
|
return SimpleNamespace(content='{"decision":"allow","reason":"ok"}')
|
|
|
|
def _fake_create_chat_model(**kwargs):
|
|
create_kwargs.update(kwargs)
|
|
return FakeModel()
|
|
|
|
monkeypatch.setattr("deerflow.skills.security_scanner.create_chat_model", _fake_create_chat_model)
|
|
|
|
result = anyio.run(
|
|
lambda: skill_manage_module._scan_or_raise(
|
|
_skill_content("demo-skill"),
|
|
executable=False,
|
|
location="demo-skill/SKILL.md",
|
|
)
|
|
)
|
|
|
|
assert result["decision"] == "allow"
|
|
assert create_kwargs["attach_tracing"] is False
|
|
|
|
|
|
# --- per-(user, skill) lock across event loops (#5846) ---
|
|
|
|
|
|
class _ParkingStorage:
|
|
"""Skill storage whose first delete parks inside the skill lock until released."""
|
|
|
|
def __init__(self) -> None:
|
|
self.entered = threading.Event()
|
|
self.release = threading.Event()
|
|
self.deleted: list[str] = []
|
|
self.callers: list[str] = []
|
|
self.max_active = 0
|
|
self._active = 0
|
|
self._guard = threading.Lock()
|
|
|
|
def delete_custom_skill(self, name: str, *, history_meta: dict) -> None:
|
|
with self._guard:
|
|
self._active += 1
|
|
self.max_active = max(self.max_active, self._active)
|
|
try:
|
|
if not self.entered.is_set():
|
|
self.entered.set()
|
|
# Bounded so a broken lock cannot hang pytest teardown.
|
|
assert self.release.wait(5), "test never released the parked holder"
|
|
self.deleted.append(name)
|
|
self.callers.append(history_meta["thread_id"])
|
|
finally:
|
|
with self._guard:
|
|
self._active -= 1
|
|
|
|
|
|
class _ObservedSkillLock:
|
|
"""Report waits on a held skill lock without replacing its locking behavior."""
|
|
|
|
def __init__(self, lock, arrived) -> None:
|
|
self.lock = lock
|
|
self.arrived = arrived
|
|
|
|
async def __aenter__(self):
|
|
if self.lock.locked():
|
|
self.arrived()
|
|
return await self.lock.__aenter__()
|
|
|
|
async def __aexit__(self, *exc):
|
|
return await self.lock.__aexit__(*exc)
|
|
|
|
|
|
def _park_skill_mutations(monkeypatch, arrived) -> _ParkingStorage:
|
|
storage = _ParkingStorage()
|
|
monkeypatch.setattr(skill_manage_module, "get_or_new_user_skill_storage", lambda _user_id: storage)
|
|
|
|
async def _refresh(_user_id: str) -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr(skill_manage_module, "refresh_user_skills_system_prompt_cache_async", _refresh)
|
|
get_lock = skill_manage_module._get_lock
|
|
monkeypatch.setattr(skill_manage_module, "_get_lock", lambda user_id, name: _ObservedSkillLock(get_lock(user_id, name), arrived))
|
|
return storage
|
|
|
|
|
|
def _delete(user_id: str, name: str, *, thread_id: str = "thread-1"):
|
|
return skill_manage_module.skill_manage_tool.coroutine(_make_runtime(thread_id=thread_id, user_id=user_id), "delete", name)
|
|
|
|
|
|
def test_skill_manage_same_skill_sync_calls_do_not_hang_across_event_loops(monkeypatch):
|
|
# ToolNode runs same-turn sync calls on separate threads; the sync wrapper
|
|
# gives each one its own asyncio.run loop.
|
|
arrivals = threading.Semaphore(0)
|
|
storage = _park_skill_mutations(monkeypatch, arrivals.release)
|
|
results = {}
|
|
|
|
def call(tag):
|
|
try:
|
|
results[tag] = skill_manage_module.skill_manage_tool.func(runtime=_make_runtime(user_id="alice"), action="delete", name="cross-loop-skill")
|
|
except Exception as exc:
|
|
results[tag] = exc
|
|
|
|
threads = [threading.Thread(target=call, args=("holder",), daemon=True)]
|
|
threads[0].start()
|
|
try:
|
|
assert storage.entered.wait(5)
|
|
for tag in ("waiter-1", "waiter-2"):
|
|
threads.append(threading.Thread(target=call, args=(tag,), daemon=True))
|
|
threads[-1].start()
|
|
assert arrivals.acquire(timeout=5), f"{tag} did not wait for the held skill lock"
|
|
finally:
|
|
storage.release.set()
|
|
for thread in threads:
|
|
thread.join(5)
|
|
|
|
assert not any(thread.is_alive() for thread in threads), "a same-skill sync call never returned"
|
|
assert results == dict.fromkeys(("holder", "waiter-1", "waiter-2"), "Deleted custom skill 'cross-loop-skill'.")
|
|
assert storage.max_active == 1
|
|
gc.collect()
|
|
assert ("alice", "cross-loop-skill") not in skill_manage_module._skill_locks
|
|
|
|
|
|
def test_skill_manage_lock_waiters_leave_workers_for_other_skills(monkeypatch):
|
|
async def scenario():
|
|
asyncio.get_running_loop().set_default_executor(ThreadPoolExecutor(max_workers=2))
|
|
waiters_parked = asyncio.Event()
|
|
arrived = 0
|
|
|
|
def on_arrival():
|
|
nonlocal arrived
|
|
arrived += 1
|
|
if arrived == 2:
|
|
waiters_parked.set()
|
|
|
|
storage = _park_skill_mutations(monkeypatch, on_arrival)
|
|
holder = asyncio.create_task(_delete("alice", "busy-skill"))
|
|
tasks = [holder]
|
|
try:
|
|
assert await asyncio.to_thread(storage.entered.wait, 5)
|
|
tasks += [asyncio.create_task(_delete("alice", "busy-skill")) for _ in range(2)]
|
|
await asyncio.wait_for(waiters_parked.wait(), 5)
|
|
# The holder parks one worker; the waiters must leave the other free
|
|
# for another skill and for the same skill of another user.
|
|
assert await asyncio.wait_for(_delete("alice", "other-skill"), 5) == "Deleted custom skill 'other-skill'."
|
|
assert await asyncio.wait_for(_delete("bob", "busy-skill"), 5) == "Deleted custom skill 'busy-skill'."
|
|
assert not holder.done()
|
|
finally:
|
|
storage.release.set()
|
|
results = await asyncio.wait_for(asyncio.gather(*tasks, return_exceptions=True), 5)
|
|
assert results == ["Deleted custom skill 'busy-skill'."] * 3
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_skill_manage_cancelled_lock_waiter_neither_waits_nor_strands_the_skill(monkeypatch):
|
|
async def scenario():
|
|
lock = skill_manage_module._get_lock("alice", "cancel-skill")
|
|
await lock.__aenter__()
|
|
held = True
|
|
all_parked = asyncio.Event()
|
|
arrived = 0
|
|
|
|
def on_arrival():
|
|
nonlocal arrived
|
|
arrived += 1
|
|
if arrived == 3:
|
|
all_parked.set()
|
|
|
|
storage = _park_skill_mutations(monkeypatch, on_arrival)
|
|
storage.entered.set()
|
|
waiters = [asyncio.create_task(_delete("alice", "cancel-skill")) for _ in range(3)]
|
|
try:
|
|
await asyncio.wait_for(all_parked.wait(), 5)
|
|
waiters[0].cancel()
|
|
# asyncio.wait, not wait_for: a waiter that shields its wait from
|
|
# cancellation must fail this test, not deadlock against the lock we hold.
|
|
await asyncio.wait({waiters[0]}, timeout=5)
|
|
assert waiters[0].cancelled(), "cancellation waited for the holder"
|
|
# Cancel the next waiter in the same step as the release, before it can re-acquire.
|
|
held = False
|
|
await lock.__aexit__(None, None, None)
|
|
waiters[1].cancel()
|
|
await asyncio.wait({waiters[1]}, timeout=5)
|
|
assert waiters[1].cancelled()
|
|
assert await asyncio.wait_for(waiters[2], 5) == "Deleted custom skill 'cancel-skill'."
|
|
finally:
|
|
if held:
|
|
await lock.__aexit__(None, None, None)
|
|
for task in waiters:
|
|
task.cancel()
|
|
await asyncio.wait_for(asyncio.gather(*waiters, return_exceptions=True), 5)
|
|
assert storage.deleted == ["cancel-skill"]
|
|
|
|
asyncio.run(scenario())
|
|
gc.collect()
|
|
assert ("alice", "cancel-skill") not in skill_manage_module._skill_locks
|
|
|
|
|
|
def test_skill_manage_lock_rejects_release_when_unlocked():
|
|
# Like asyncio.Lock: releasing a lock nobody holds is an error, not a no-op.
|
|
lock = skill_manage_module._get_lock("alice", "unlocked-skill")
|
|
with pytest.raises(RuntimeError):
|
|
asyncio.run(lock.__aexit__(None, None, None))
|
|
|
|
|
|
def test_skill_manage_same_skill_waiters_keep_arrival_order(monkeypatch):
|
|
# Same-turn calls on one skill can depend on each other (patch a->b, then
|
|
# b->c). The gateway runs them on one loop, where they must take the lock in
|
|
# the order they reached it, as asyncio.Lock did.
|
|
async def scenario():
|
|
lock = skill_manage_module._get_lock("alice", "ordered-skill")
|
|
await lock.__aenter__()
|
|
arrived = asyncio.Event()
|
|
storage = _park_skill_mutations(monkeypatch, arrived.set)
|
|
storage.entered.set()
|
|
# The thread id only tags each call in the storage record.
|
|
calls = [f"call-{i}" for i in range(8)]
|
|
waiters = []
|
|
try:
|
|
for call in calls:
|
|
arrived.clear()
|
|
waiters.append(asyncio.create_task(_delete("alice", "ordered-skill", thread_id=call)))
|
|
await asyncio.wait_for(arrived.wait(), 5)
|
|
finally:
|
|
await lock.__aexit__(None, None, None)
|
|
await asyncio.wait_for(asyncio.gather(*waiters, return_exceptions=True), 5)
|
|
assert storage.callers == calls
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def test_skill_manage_later_call_does_not_overtake_a_parked_waiter(monkeypatch):
|
|
# The storage lookup runs before the lock, so a later call can reach the
|
|
# lock in the same loop step as a release. As with asyncio.Lock, it must
|
|
# queue behind the call already waiting instead of taking the lock first.
|
|
async def scenario():
|
|
lock = skill_manage_module._get_lock("alice", "fair-skill")
|
|
await lock.__aenter__()
|
|
arrived = asyncio.Event()
|
|
storage = _park_skill_mutations(monkeypatch, arrived.set)
|
|
storage.entered.set()
|
|
to_thread = skill_manage_module._to_thread
|
|
lookup_parked = asyncio.Event()
|
|
finish_lookup = asyncio.Event()
|
|
late = None
|
|
|
|
async def gated_to_thread(func, /, *args, **kwargs):
|
|
if func is skill_manage_module.get_or_new_user_skill_storage and asyncio.current_task() is late:
|
|
lookup_parked.set()
|
|
await finish_lookup.wait()
|
|
# Resume straight into the lock, with no thread hop in between.
|
|
return func(*args, **kwargs)
|
|
return await to_thread(func, *args, **kwargs)
|
|
|
|
monkeypatch.setattr(skill_manage_module, "_to_thread", gated_to_thread)
|
|
waiter = asyncio.create_task(_delete("alice", "fair-skill", thread_id="queued"))
|
|
tasks = [waiter]
|
|
held = True
|
|
try:
|
|
await asyncio.wait_for(arrived.wait(), 5)
|
|
late = asyncio.create_task(_delete("alice", "fair-skill", thread_id="late"))
|
|
tasks.append(late)
|
|
await asyncio.wait_for(lookup_parked.wait(), 5)
|
|
# Release while the later call's lookup completes in the same step.
|
|
finish_lookup.set()
|
|
held = False
|
|
await lock.__aexit__(None, None, None)
|
|
await asyncio.wait_for(asyncio.gather(*tasks), 5)
|
|
finally:
|
|
if held:
|
|
await lock.__aexit__(None, None, None)
|
|
finish_lookup.set()
|
|
await asyncio.wait_for(asyncio.gather(*tasks, return_exceptions=True), 5)
|
|
assert storage.callers == ["queued", "late"]
|
|
|
|
asyncio.run(scenario())
|
|
|
|
|
|
def _patch_skill_manage_env(monkeypatch, tmp_path):
|
|
"""Config, user paths and the scanner stubs every tool-level case needs."""
|
|
skills_root = tmp_path / "skills"
|
|
config = _make_config(skills_root)
|
|
monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
|
|
monkeypatch.setattr("deerflow.skills.security_scanner.get_app_config", lambda: config)
|
|
from deerflow.config.paths import Paths
|
|
|
|
monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path))
|
|
monkeypatch.setattr("deerflow.config.paths._paths", None)
|
|
|
|
async def _refresh(user_id: str):
|
|
return None
|
|
|
|
monkeypatch.setattr(skill_manage_module, "refresh_user_skills_system_prompt_cache_async", _refresh)
|
|
monkeypatch.setattr(skill_manage_module, "scan_skill_content", lambda *args, **kwargs: _async_result("allow", "ok"))
|
|
return tmp_path / "users" / "default" / "skills" / "custom"
|
|
|
|
|
|
def test_skill_manage_create_rejects_windows_device_names(monkeypatch, tmp_path):
|
|
"""Creation is the only action that must refuse a host-incompatible name."""
|
|
user_custom = _patch_skill_manage_env(monkeypatch, tmp_path)
|
|
runtime = _make_runtime(user_id="default")
|
|
|
|
with pytest.raises(ValueError, match="not portable to Windows"):
|
|
anyio.run(skill_manage_module.skill_manage_tool.coroutine, runtime, "create", "nul", _skill_content("nul"))
|
|
|
|
# ``nul`` resolves to the NUL device on Windows, so assert on ``is_dir``.
|
|
assert not (user_custom / "nul").is_dir()
|
|
|
|
|
|
def test_skill_manage_edits_stay_permissive_for_device_named_skills(monkeypatch, tmp_path):
|
|
"""An existing skill an older release created on Linux stays editable/deletable."""
|
|
_patch_skill_manage_env(monkeypatch, tmp_path)
|
|
runtime = _make_runtime(user_id="default")
|
|
|
|
# The name passes validation; the skill simply does not exist, which is a
|
|
# different error than the creation-time portability rejection.
|
|
with pytest.raises(FileNotFoundError, match="not found"):
|
|
anyio.run(skill_manage_module.skill_manage_tool.coroutine, runtime, "edit", "nul", _skill_content("nul"))
|
|
with pytest.raises(FileNotFoundError, match="not found"):
|
|
anyio.run(skill_manage_module.skill_manage_tool.coroutine, runtime, "delete", "nul")
|