diff --git a/skillopt_sleep/skill_resolver.py b/skillopt_sleep/skill_resolver.py new file mode 100644 index 00000000..aedf92e5 --- /dev/null +++ b/skillopt_sleep/skill_resolver.py @@ -0,0 +1,137 @@ +"""SkillOpt-Sleep — resolve a discovered skill name to a local ``SKILL.md``. + +Skill names observed in transcripts are untrusted strings, so resolution is +deliberately narrow: a name is normalized, matched only inside documented local +skill roots, and reported. Nothing here writes, edits, or creates files. + +Resolution outcomes are distinguishable on purpose — ``missing`` (no root has +the skill) is a different signal from ``ambiguous`` (several roots do) and from +``rejected`` (the name itself is unusable), so callers can fall back to the +existing managed-skill behavior instead of guessing. +""" +from __future__ import annotations + +import os +from dataclasses import dataclass, field +from typing import List, Sequence + +SKILL_FILENAME = "SKILL.md" + +FOUND = "found" +MISSING = "missing" +AMBIGUOUS = "ambiguous" +REJECTED = "rejected" + + +@dataclass(frozen=True) +class SkillResolution: + """The outcome of resolving one skill name. Never a partial success.""" + + name: str + status: str + path: str = "" + candidates: List[str] = field(default_factory=list) + reason: str = "" + + @property + def ok(self) -> bool: + return self.status == FOUND + + +def normalize_skill_name(name: object) -> str: + """Return a usable skill directory name, or "" when the name is unusable. + + Only a single path segment is accepted: no separators, no parent traversal, + no absolute or home-relative paths, no control characters. The name is + whitespace-trimmed but otherwise preserved, since skill directories are + case- and punctuation-sensitive. + """ + if not isinstance(name, str): + return "" + candidate = name.strip() + if not candidate or candidate in {os.curdir, os.pardir}: + return "" + if candidate.startswith("~"): + return "" + if os.path.isabs(candidate) or os.path.splitdrive(candidate)[0]: + return "" + if "/" in candidate or "\\" in candidate or os.sep in candidate: + return "" + if os.altsep and os.altsep in candidate: + return "" + if any(ord(ch) < 32 or ord(ch) == 127 for ch in candidate): + return "" + return candidate + + +def skill_search_roots(cfg: object) -> List[str]: + """Documented local skill roots for a config: user skills, then plugin cache. + + ``/skills`` holds hand-written skills; installed Claude Code + plugins expose theirs under ``/plugins/cache/*/*/skills``. + Only existing directories are returned, in that fixed precedence order. + """ + claude_home = os.path.abspath(os.path.expanduser(str(getattr(cfg, "claude_home", "")))) + if not claude_home: + return [] + roots = [os.path.join(claude_home, "skills")] + + cache = os.path.join(claude_home, "plugins", "cache") + if os.path.isdir(cache): + for marketplace in sorted(os.listdir(cache)): + plugins_dir = os.path.join(cache, marketplace) + if not os.path.isdir(plugins_dir): + continue + for plugin in sorted(os.listdir(plugins_dir)): + roots.append(os.path.join(plugins_dir, plugin, "skills")) + return [r for r in roots if os.path.isdir(r)] + + +def _contained_skill_file(root: str, name: str) -> str: + """Return the real ``SKILL.md`` path under ``root`` for ``name``, else "". + + Symlinks are followed and then re-checked against the real root, so a skill + directory or file that points outside the root is refused rather than read. + """ + try: + real_root = os.path.realpath(root) + skill_file = os.path.realpath(os.path.join(real_root, name, SKILL_FILENAME)) + except OSError: + return "" + if not os.path.isfile(skill_file): + return "" + if os.path.commonpath([real_root, skill_file]) != real_root: + return "" + return skill_file + + +def resolve_skill(name: object, roots: Sequence[str]) -> SkillResolution: + """Resolve ``name`` against ``roots`` without touching any skill content.""" + normalized = normalize_skill_name(name) + if not normalized: + return SkillResolution( + name=name if isinstance(name, str) else "", + status=REJECTED, + reason="skill name is empty or not a single safe path segment", + ) + + matches: List[str] = [] + for root in roots: + found = _contained_skill_file(root, normalized) + if found and found not in matches: + matches.append(found) + + if not matches: + return SkillResolution( + name=normalized, + status=MISSING, + reason=f"no {SKILL_FILENAME} for this skill in the configured skill roots", + ) + if len(matches) > 1: + return SkillResolution( + name=normalized, + status=AMBIGUOUS, + candidates=matches, + reason="several skill roots define this skill", + ) + return SkillResolution(name=normalized, status=FOUND, path=matches[0], candidates=matches) diff --git a/tests/test_sleep_skill_resolver.py b/tests/test_sleep_skill_resolver.py new file mode 100644 index 00000000..89772d20 --- /dev/null +++ b/tests/test_sleep_skill_resolver.py @@ -0,0 +1,166 @@ +"""Tests for bounded skill-name resolution (issue #120). + +Pure-stdlib (unittest), hermetic (tmpdir only), no API key, no network. +Run: python -m pytest tests/test_sleep_skill_resolver.py +""" +from __future__ import annotations + +import os +import tempfile +import unittest + +from skillopt_sleep.config import load_config +from skillopt_sleep.skill_resolver import ( + AMBIGUOUS, + FOUND, + MISSING, + REJECTED, + normalize_skill_name, + resolve_skill, + skill_search_roots, +) + + +def _write_skill(root, name, body="# skill\n"): + path = os.path.join(root, name, "SKILL.md") + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as f: + f.write(body) + return path + + +class TestNormalizeSkillName(unittest.TestCase): + def test_trims_but_preserves_case_and_punctuation(self): + self.assertEqual(normalize_skill_name(" Brand-Voice.v2 "), "Brand-Voice.v2") + + def test_rejects_unusable_names(self): + for bad in ["", " ", ".", "..", "../escape", "a/b", "a\\b", "/abs/skill", + "~/skill", "bad\nname", "bad\x00name", None, 3]: + self.assertEqual(normalize_skill_name(bad), "", repr(bad)) + + +class TestResolveSkill(unittest.TestCase): + def test_resolves_a_single_local_skill(self): + with tempfile.TemporaryDirectory() as tmp: + expected = _write_skill(tmp, "example-skill") + res = resolve_skill(" example-skill ", [tmp]) + self.assertEqual(res.status, FOUND) + self.assertTrue(res.ok) + self.assertEqual(res.path, os.path.realpath(expected)) + self.assertEqual(res.name, "example-skill") + + def test_missing_skill_is_distinct_from_ambiguous(self): + with tempfile.TemporaryDirectory() as tmp: + _write_skill(tmp, "other-skill") + res = resolve_skill("example-skill", [tmp]) + self.assertEqual(res.status, MISSING) + self.assertEqual(res.path, "") + self.assertEqual(res.candidates, []) + + def test_directory_without_skill_file_is_missing(self): + with tempfile.TemporaryDirectory() as tmp: + os.makedirs(os.path.join(tmp, "example-skill")) + self.assertEqual(resolve_skill("example-skill", [tmp]).status, MISSING) + + def test_same_skill_in_two_roots_is_ambiguous(self): + with tempfile.TemporaryDirectory() as tmp: + local, cache = os.path.join(tmp, "local"), os.path.join(tmp, "cache") + first = _write_skill(local, "example-skill") + second = _write_skill(cache, "example-skill") + res = resolve_skill("example-skill", [local, cache]) + self.assertEqual(res.status, AMBIGUOUS) + self.assertEqual(res.path, "") + self.assertEqual(res.candidates, + [os.path.realpath(first), os.path.realpath(second)]) + + def test_repeated_root_is_not_ambiguous(self): + with tempfile.TemporaryDirectory() as tmp: + _write_skill(tmp, "example-skill") + self.assertEqual(resolve_skill("example-skill", [tmp, tmp]).status, FOUND) + + def test_traversal_is_rejected_without_touching_the_filesystem(self): + with tempfile.TemporaryDirectory() as tmp: + root = os.path.join(tmp, "roots") + _write_skill(tmp, "outside-skill") + os.makedirs(root, exist_ok=True) + res = resolve_skill("../outside-skill", [root]) + self.assertEqual(res.status, REJECTED) + self.assertEqual(res.path, "") + + def test_symlinked_skill_dir_escaping_the_root_is_refused(self): + with tempfile.TemporaryDirectory() as tmp: + root = os.path.join(tmp, "roots") + os.makedirs(root) + outside = os.path.join(tmp, "outside") + _write_skill(outside, "example-skill") + os.symlink(os.path.join(outside, "example-skill"), + os.path.join(root, "example-skill")) + self.assertEqual(resolve_skill("example-skill", [root]).status, MISSING) + + def test_symlinked_skill_file_escaping_the_root_is_refused(self): + with tempfile.TemporaryDirectory() as tmp: + root = os.path.join(tmp, "roots") + os.makedirs(os.path.join(root, "example-skill")) + elsewhere = os.path.join(tmp, "elsewhere.md") + with open(elsewhere, "w", encoding="utf-8") as f: + f.write("# not in the root\n") + os.symlink(elsewhere, os.path.join(root, "example-skill", "SKILL.md")) + self.assertEqual(resolve_skill("example-skill", [root]).status, MISSING) + + def test_symlinked_root_itself_still_resolves(self): + with tempfile.TemporaryDirectory() as tmp: + real = os.path.join(tmp, "real") + _write_skill(real, "example-skill") + link = os.path.join(tmp, "link") + os.symlink(real, link) + self.assertEqual(resolve_skill("example-skill", [link]).status, FOUND) + + def test_resolution_never_modifies_the_skill(self): + with tempfile.TemporaryDirectory() as tmp: + path = _write_skill(tmp, "example-skill", "# original\n") + before = (os.stat(path).st_size, open(path, encoding="utf-8").read()) + resolve_skill("example-skill", [tmp]) + self.assertEqual((os.stat(path).st_size, + open(path, encoding="utf-8").read()), before) + + def test_no_roots_is_missing(self): + self.assertEqual(resolve_skill("example-skill", []).status, MISSING) + + +class TestSkillSearchRoots(unittest.TestCase): + def test_user_skills_root_comes_first_then_plugin_cache(self): + with tempfile.TemporaryDirectory() as tmp: + claude_home = os.path.join(tmp, ".claude") + skills = os.path.join(claude_home, "skills") + os.makedirs(skills) + plugin_skills = os.path.join( + claude_home, "plugins", "cache", "marketplace", "plugin", "skills" + ) + os.makedirs(plugin_skills) + cfg = load_config(claude_home=claude_home) + self.assertEqual(skill_search_roots(cfg), [skills, plugin_skills]) + + def test_absent_roots_are_skipped(self): + with tempfile.TemporaryDirectory() as tmp: + cfg = load_config(claude_home=os.path.join(tmp, ".claude")) + self.assertEqual(skill_search_roots(cfg), []) + + def test_resolution_through_config_roots_prefers_the_user_skill(self): + with tempfile.TemporaryDirectory() as tmp: + claude_home = os.path.join(tmp, ".claude") + expected = _write_skill(os.path.join(claude_home, "skills"), "example-skill") + cfg = load_config(claude_home=claude_home) + res = resolve_skill("example-skill", skill_search_roots(cfg)) + self.assertEqual(res.status, FOUND) + self.assertEqual(res.path, os.path.realpath(expected)) + + def test_legacy_target_skill_path_behavior_is_untouched(self): + with tempfile.TemporaryDirectory() as tmp: + target = os.path.join(tmp, "repo", "SKILL.md") + cfg = load_config(claude_home=os.path.join(tmp, ".claude"), + target_skill_path=target) + self.assertEqual(cfg.managed_skill_path(), os.path.abspath(target)) + + +if __name__ == "__main__": + unittest.main()