diff --git a/skillopt_sleep/skill_resolver.py b/skillopt_sleep/skill_resolver.py new file mode 100644 index 00000000..318664da --- /dev/null +++ b/skillopt_sleep/skill_resolver.py @@ -0,0 +1,204 @@ +"""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 +import re +from dataclasses import dataclass +from typing import List, Sequence, Tuple + +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: Tuple[str, ...] = () + 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 _listdir(path: str) -> List[str]: + """Sorted directory entries, or [] when the directory is absent/unreadable.""" + try: + return sorted(os.listdir(path)) + except OSError: + return [] + + +def _version_sort_key(name: str) -> tuple: + """Order version directory names newest-last, numerically where possible. + + ``1.10.0`` must sort above ``1.9.0``, so the leading numeric segments + compare as ints rather than as strings. Anything after the first + non-numeric segment is a prerelease suffix (``2.0.0-beta``), and a bare + release outranks any prerelease sharing its numeric prefix — comparing the + segment lists alone would do the opposite, because a shorter list that is a + prefix of a longer one sorts lower. The name is the final tie-break so the + order is always total and deterministic. + """ + numeric: List[int] = [] + suffix: List[str] = [] + for part in re.split(r"[._\-+]", name): + if part.isdigit() and not suffix: + numeric.append(int(part)) + else: + suffix.append(part) + return (numeric, 0 if suffix else 1, suffix, name) + + +def _plugin_skills_root(plugin_dir: str) -> str: + """The single skills root for one installed plugin, or "" if it has none. + + Claude marketplace installs are versioned — + ``//skills`` — and several versions of the same plugin can + be present at once. Returning each of them as a peer root would make an + ordinary upgraded plugin resolve AMBIGUOUS, so exactly one root is chosen: + the newest version, falling back to the legacy unversioned + ``/skills`` layout when no version directory carries skills. + """ + versioned = [] + for entry in _listdir(plugin_dir): + candidate = os.path.join(plugin_dir, entry, "skills") + if os.path.isdir(candidate): + versioned.append((_version_sort_key(entry), candidate)) + if versioned: + versioned.sort() + return versioned[-1][1] + + legacy = os.path.join(plugin_dir, "skills") + return legacy if os.path.isdir(legacy) else "" + + +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 the plugin cache, in either the versioned + marketplace layout ``plugins/cache////skills`` + or the legacy ``plugins/cache///skills``. At most one + root per installed plugin is returned, in that fixed precedence order. + """ + configured = str(getattr(cfg, "claude_home", "") or "").strip() + if not configured: + # Guard before abspath: os.path.abspath("") is the CWD, which would + # silently search a tree well outside the documented ~/.claude root. + return [] + claude_home = os.path.abspath(os.path.expanduser(configured)) + roots = [os.path.join(claude_home, "skills")] + + cache = os.path.join(claude_home, "plugins", "cache") + for marketplace in _listdir(cache): + plugins_dir = os.path.join(cache, marketplace) + if not os.path.isdir(plugins_dir): + continue + for plugin in _listdir(plugins_dir): + root = _plugin_skills_root(os.path.join(plugins_dir, plugin)) + if root: + roots.append(root) + 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 "" + try: + contained = os.path.commonpath([real_root, skill_file]) == real_root + except ValueError: + # Different drives / mixed path flavours (Windows): by definition the + # file is not inside this root, so refuse it rather than crash. + return "" + if not contained: + 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=tuple(matches), + reason="several skill roots define this skill", + ) + return SkillResolution( + name=normalized, status=FOUND, path=matches[0], candidates=tuple(matches) + ) diff --git a/tests/test_sleep_skill_resolver.py b/tests/test_sleep_skill_resolver.py new file mode 100644 index 00000000..00912349 --- /dev/null +++ b/tests/test_sleep_skill_resolver.py @@ -0,0 +1,340 @@ +"""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 + + +def _symlink(test, source, link_name): + """Create a symlink, or skip the test where the platform refuses one. + + Windows needs admin or Developer Mode for symlinks, and some CI sandboxes + disallow them outright. The behaviour under test is a security property + that only exists where symlinks do, so skipping is correct rather than + failing on an unrelated platform limitation. + """ + try: + os.symlink(source, link_name) + except (OSError, NotImplementedError, AttributeError) as exc: + test.skipTest(f"symlinks unavailable on this platform: {exc}") + + +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") + _symlink(self, 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") + _symlink(self, 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") + _symlink(self, 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") + with open(path, encoding="utf-8") as f: + before = (os.stat(path).st_size, f.read()) + resolve_skill("example-skill", [tmp]) + with open(path, encoding="utf-8") as f: + after = (os.stat(path).st_size, f.read()) + self.assertEqual(after, before) + + def test_no_roots_is_missing(self): + self.assertEqual(resolve_skill("example-skill", []).status, MISSING) + + def test_candidates_are_an_immutable_tuple(self): + with tempfile.TemporaryDirectory() as tmp: + _write_skill(tmp, "example-skill") + res = resolve_skill("example-skill", [tmp]) + self.assertIsInstance(res.candidates, tuple) + with self.assertRaises(AttributeError): + res.candidates.append("/injected") # type: ignore[attr-defined] + + def test_skill_file_on_another_drive_is_refused_not_crashed(self): + # os.path.commonpath raises ValueError for paths that share no root + # (mixed drives on Windows). Resolution must treat that as "outside". + with tempfile.TemporaryDirectory() as tmp: + _write_skill(tmp, "example-skill") + real_commonpath = os.path.commonpath + + def exploding_commonpath(paths): + raise ValueError("paths don't have the same drive") + + os.path.commonpath = exploding_commonpath + try: + self.assertEqual(resolve_skill("example-skill", [tmp]).status, MISSING) + finally: + os.path.commonpath = real_commonpath + + +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_blank_claude_home_never_falls_back_to_the_cwd(self): + # os.path.abspath("") is the CWD; a blank override must not turn the + # working directory into a skill root. + class _Cfg: + def __init__(self, claude_home): + self.claude_home = claude_home + + for blank in ["", " ", None]: + self.assertEqual(skill_search_roots(_Cfg(blank)), [], repr(blank)) + + def test_unreadable_plugin_cache_does_not_break_discovery(self): + with tempfile.TemporaryDirectory() as tmp: + claude_home = os.path.join(tmp, ".claude") + skills = os.path.join(claude_home, "skills") + os.makedirs(skills) + cache = os.path.join(claude_home, "plugins", "cache") + os.makedirs(cache) + try: + os.chmod(cache, 0o000) + except OSError as exc: # pragma: no cover - platform dependent + self.skipTest(f"cannot drop permissions on this platform: {exc}") + if os.access(cache, os.R_OK): + # Windows and some filesystems ignore mode bits, and root + # bypasses them, so the unreadable precondition never holds. + os.chmod(cache, 0o700) + self.skipTest("directory is still readable after chmod 000") + try: + cfg = load_config(claude_home=claude_home) + self.assertEqual(skill_search_roots(cfg), [skills]) + finally: + try: + os.chmod(cache, 0o700) + except OSError: # pragma: no cover - best-effort restore + pass + + 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_versioned_marketplace_layout_is_discovered(self): + # Sanitized mirror of a real marketplace install. Observed layout on a + # working machine: every plugin skills dir sits at + # ////skills — the unversioned + # layout did not occur at all, so discovery must handle this one. + with tempfile.TemporaryDirectory() as tmp: + claude_home = os.path.join(tmp, ".claude") + cache = os.path.join(claude_home, "plugins", "cache") + expected = _write_skill( + os.path.join(cache, "claude-plugins-official", "superpowers", "5.0.7", "skills"), + "brainstorming", + ) + cfg = load_config(claude_home=claude_home) + self.assertIn( + os.path.join(cache, "claude-plugins-official", "superpowers", "5.0.7", "skills"), + skill_search_roots(cfg), + ) + res = resolve_skill("brainstorming", skill_search_roots(cfg)) + self.assertEqual(res.status, FOUND) + self.assertEqual(res.path, os.path.realpath(expected)) + + def test_multiple_installed_versions_resolve_to_the_newest_not_ambiguous(self): + # A plugin upgraded in place keeps older version dirs alongside the new + # one (observed: three versions of the same plugin). Treating each as a + # peer root would make an ordinary upgrade resolve AMBIGUOUS. + with tempfile.TemporaryDirectory() as tmp: + claude_home = os.path.join(tmp, ".claude") + plugin = os.path.join(claude_home, "plugins", "cache", + "claude-plugins-official", "chrome-devtools-mcp") + for version in ["1.1.1", "1.5.0", "1.6.0"]: + _write_skill(os.path.join(plugin, version, "skills"), "chrome-devtools") + newest = os.path.join(plugin, "1.6.0", "skills") + + cfg = load_config(claude_home=claude_home) + roots = skill_search_roots(cfg) + self.assertEqual([r for r in roots if r.startswith(plugin)], [newest]) + + res = resolve_skill("chrome-devtools", roots) + self.assertEqual(res.status, FOUND) + self.assertEqual(res.path, + os.path.realpath(os.path.join(newest, "chrome-devtools", "SKILL.md"))) + + def test_version_ordering_is_numeric_not_lexicographic(self): + with tempfile.TemporaryDirectory() as tmp: + claude_home = os.path.join(tmp, ".claude") + plugin = os.path.join(claude_home, "plugins", "cache", "market", "plugin") + for version in ["1.9.0", "1.10.0"]: + _write_skill(os.path.join(plugin, version, "skills"), "example-skill") + cfg = load_config(claude_home=claude_home) + roots = [r for r in skill_search_roots(cfg) if r.startswith(plugin)] + # "1.10.0" < "1.9.0" as strings; it must still win as a version. + self.assertEqual(roots, [os.path.join(plugin, "1.10.0", "skills")]) + + def test_stable_release_beats_an_installed_prerelease(self): + # Segment lists alone would rank 2.0.0-beta above 2.0.0, because a + # shorter list that prefixes a longer one sorts lower. A prerelease + # must never be preferred over the stable release it precedes. + with tempfile.TemporaryDirectory() as tmp: + claude_home = os.path.join(tmp, ".claude") + plugin = os.path.join(claude_home, "plugins", "cache", "market", "plugin") + for version in ["2.0.0", "2.0.0-beta"]: + _write_skill(os.path.join(plugin, version, "skills"), "example-skill") + cfg = load_config(claude_home=claude_home) + roots = [r for r in skill_search_roots(cfg) if r.startswith(plugin)] + self.assertEqual(roots, [os.path.join(plugin, "2.0.0", "skills")]) + + def test_version_key_orders_release_forms_sensibly(self): + from skillopt_sleep.skill_resolver import _version_sort_key as key + self.assertGreater(key("2.0.0"), key("2.0.0-beta")) + self.assertGreater(key("1.10.0"), key("1.9.0")) + self.assertGreater(key("2.0.0-beta"), key("2.0.0-alpha")) + self.assertGreater(key("1.0.0"), key("1.0")) + self.assertGreater(key("1.0.0"), key("1.0.0rc1")) + self.assertGreater(key("1.0.0"), key("main")) + + def test_legacy_unversioned_layout_still_works(self): + with tempfile.TemporaryDirectory() as tmp: + claude_home = os.path.join(tmp, ".claude") + plugin = os.path.join(claude_home, "plugins", "cache", "market", "plugin") + expected = _write_skill(os.path.join(plugin, "skills"), "example-skill") + cfg = load_config(claude_home=claude_home) + self.assertIn(os.path.join(plugin, "skills"), skill_search_roots(cfg)) + res = resolve_skill("example-skill", skill_search_roots(cfg)) + self.assertEqual(res.status, FOUND) + self.assertEqual(res.path, os.path.realpath(expected)) + + def test_two_marketplaces_each_contribute_a_root(self): + with tempfile.TemporaryDirectory() as tmp: + claude_home = os.path.join(tmp, ".claude") + cache = os.path.join(claude_home, "plugins", "cache") + official = os.path.join(cache, "claude-plugins-official", "sentry", "1.0.0", "skills") + cognee = os.path.join(cache, "cognee", "cognee-memory", "1.0.0", "skills") + _write_skill(official, "sentry-skill") + _write_skill(cognee, "cognee-remember") + cfg = load_config(claude_home=claude_home) + roots = skill_search_roots(cfg) + self.assertIn(official, roots) + self.assertIn(cognee, roots) + + 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()