wehub-resource-sync
343 lines
13 KiB
Python
343 lines
13 KiB
Python
import importlib
|
|
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")
|
|
|
|
|
|
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_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()
|