diff --git a/hermes_cli/skills_hub.py b/hermes_cli/skills_hub.py index ff8f18256115..66fb903ab3e0 100644 --- a/hermes_cli/skills_hub.py +++ b/hermes_cli/skills_hub.py @@ -516,7 +516,7 @@ def do_install(identifier: str, category: str = "", force: bool = False, GitHubAuth, create_source_router, ensure_hub_dirs, quarantine_bundle, install_from_quarantine, HubLockFile, ) - from tools.skills_guard import scan_skill, should_allow_install, format_scan_report + from tools.skills_guard import scan_skill_cached, should_allow_install, format_scan_report c = console or _console ensure_hub_dirs() @@ -648,8 +648,24 @@ def do_install(identifier: str, category: str = "", force: bool = False, or getattr(meta, "identifier", "") or identifier ) - result = scan_skill(q_path, source=scan_source) + from tools.skills_hub import HUB_DIR, source_url_for_bundle + result, scan_provenance = scan_skill_cached( + q_path, + source=scan_source, + source_url=source_url_for_bundle(bundle), + cache_dir=HUB_DIR / "scan-cache", + ) c.print(format_scan_report(result)) + freshness = "fresh" if scan_provenance["fresh"] else "cached" + c.print( + f"[dim]Scan provenance: {freshness}; scanner " + f"{scan_provenance['scanner_version']}; hash {scan_provenance['bundle_hash']}[/]" + ) + rules = ", ".join(scan_provenance["rules"]) or "none" + c.print( + f"[dim]Source: {scan_provenance['source_url']}; scanned " + f"{scan_provenance['scanned_at']}; rules: {rules}[/]" + ) # Check install policy allowed, reason = should_allow_install(result, force=force) diff --git a/tests/tools/test_skill_bundle_provenance.py b/tests/tools/test_skill_bundle_provenance.py new file mode 100644 index 000000000000..25f696c775bf --- /dev/null +++ b/tests/tools/test_skill_bundle_provenance.py @@ -0,0 +1,199 @@ +"""Multi-file third-party skill bundles and scanner provenance (#60598).""" + +import json +import subprocess +import threading +from functools import partial +from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer +from io import StringIO +from pathlib import Path + +import pytest +from rich.console import Console + +from tools.skills_guard import SCANNER_VERSION, scan_skill_cached +from tools.skills_hub import GitHubAuth, GitHubSource, HubLockFile, SkillBundle, UrlSource + + +SKILL_MD = """--- +name: demo-bundle +description: A multi-file test skill. +--- +# Demo +Read [the guide](references/guide.md), use `templates/report.md`, and run +`scripts/run.py`. The repository also contains assets/logo.txt. +""" + + +class _QuietHandler(SimpleHTTPRequestHandler): + def log_message(self, *_args): + pass + + +@pytest.fixture +def served_repo(tmp_path): + repo = tmp_path / "upstream" + repo.mkdir() + (repo / "SKILL.md").write_text(SKILL_MD) + for rel, text in { + "references/guide.md": "safe guide\n", + "templates/report.md": "report\n", + "scripts/run.py": "print('ok')\n", + "assets/logo.txt": "logo\n", + "examples/not-installed.md": "must not be copied\n", + "README.md": "must not be copied\n", + }.items(): + path = repo / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text) + subprocess.run(["git", "init", "-q"], cwd=repo, check=True) + subprocess.run(["git", "add", "."], cwd=repo, check=True) + subprocess.run( + ["git", "-c", "user.name=Test", "-c", "user.email=test@example.com", "commit", "-qm", "fixture"], + cwd=repo, + check=True, + ) + + server = ThreadingHTTPServer( + ("127.0.0.1", 0), partial(_QuietHandler, directory=str(repo)) + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield repo, f"http://127.0.0.1:{server.server_port}/SKILL.md" + finally: + server.shutdown() + thread.join() + + +def test_url_source_fetches_only_referenced_allowed_support_directories(served_repo, monkeypatch): + _repo, url = served_repo + monkeypatch.setattr("tools.skills_hub.is_safe_url", lambda _url: True) + monkeypatch.setattr("tools.skills_hub.check_website_access", lambda _url: None) + + bundle = UrlSource().fetch(url) + + assert bundle is not None + assert set(bundle.files) == { + "SKILL.md", + "references/guide.md", + "templates/report.md", + "scripts/run.py", + "assets/logo.txt", + } + assert bundle.metadata["source_url"] == url + + +def test_url_source_rejects_traversal_reference(monkeypatch): + source = UrlSource() + skill = "---\nname: bad\ndescription: bad\n---\n[bad](references/../../secret.txt)\n" + monkeypatch.setattr(source, "_fetch_text", lambda _url: skill) + + assert source.fetch("https://example.com/bad/SKILL.md") is None + + +def test_github_source_rejects_symlink_in_referenced_directory(monkeypatch): + source = GitHubSource(GitHubAuth()) + monkeypatch.setattr(source, "_fetch_file_content", lambda _repo, path: SKILL_MD if path.endswith("SKILL.md") else "x") + source._tree_cache["owner/repo"] = ( + "main", + [ + {"path": "skill/SKILL.md", "type": "blob", "mode": "100644"}, + {"path": "skill/references/guide.md", "type": "blob", "mode": "120000"}, + ], + ) + + assert source.fetch("owner/repo/skill") is None + + +def test_scan_cache_records_full_provenance_and_hash_change_forces_rescan(tmp_path): + skill = tmp_path / "skill" + skill.mkdir() + (skill / "SKILL.md").write_text("---\nname: skill\ndescription: test\n---\n# safe\n") + cache = tmp_path / "scan-cache" + + first, first_provenance = scan_skill_cached( + skill, source="owner/repo/skill", source_url="https://github.com/owner/repo", cache_dir=cache + ) + second, second_provenance = scan_skill_cached( + skill, source="owner/repo/skill", source_url="https://github.com/owner/repo", cache_dir=cache + ) + (skill / "SKILL.md").write_text("---\nname: skill\ndescription: changed\n---\n# safe\n") + third, third_provenance = scan_skill_cached( + skill, source="owner/repo/skill", source_url="https://github.com/owner/repo", cache_dir=cache + ) + + assert first.verdict == second.verdict == third.verdict == "safe" + assert first_provenance["fresh"] is True + assert second_provenance["fresh"] is False + assert third_provenance["fresh"] is True + assert first_provenance["bundle_hash"].startswith("sha256:") + assert len(first_provenance["bundle_hash"].split(":", 1)[1]) == 64 + assert third_provenance["bundle_hash"] != first_provenance["bundle_hash"] + assert first_provenance["scanner_version"] == SCANNER_VERSION + assert first_provenance["source_url"] == "https://github.com/owner/repo" + assert isinstance(first_provenance["findings"], list) + assert isinstance(first_provenance["rules"], list) + assert first_provenance["scanned_at"] + + +def test_lock_file_persists_scan_provenance(tmp_path): + lock = HubLockFile(tmp_path / "lock.json") + provenance = { + "source_url": "https://example.com/SKILL.md", + "bundle_hash": "sha256:" + "a" * 64, + "scanner_version": SCANNER_VERSION, + "findings": [], + "rules": [], + "scanned_at": "2026-07-09T00:00:00+00:00", + "fresh": True, + } + lock.record_install( + name="demo", source="url", identifier="https://example.com/SKILL.md", + trust_level="community", scan_verdict="safe", skill_hash="sha256:legacy", + install_path="demo", files=["SKILL.md"], scan_provenance=provenance, + ) + + assert lock.get_installed("demo")["scan_provenance"] == provenance + + +def test_real_temp_repo_and_home_install_e2e(served_repo, monkeypatch, tmp_path): + from hermes_cli.skills_hub import do_install + import tools.skills_hub as hub + + _repo, url = served_repo + home = tmp_path / "home" + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setattr("tools.skills_hub.is_safe_url", lambda _url: True) + monkeypatch.setattr("tools.skills_hub.check_website_access", lambda _url: None) + monkeypatch.setattr(hub, "create_source_router", lambda auth=None: [UrlSource()]) + + sink = StringIO() + do_install(url, console=Console(file=sink, force_terminal=False), skip_confirm=True) + + installed = home / "skills" / "demo-bundle" + assert (installed / "references" / "guide.md").read_text() == "safe guide\n" + assert (installed / "templates" / "report.md").is_file() + assert (installed / "scripts" / "run.py").is_file() + assert (installed / "assets" / "logo.txt").is_file() + assert not (installed / "examples").exists() + entry = json.loads((home / "skills" / ".hub" / "lock.json").read_text())["installed"]["demo-bundle"] + assert entry["scan_provenance"]["source_url"] == url + assert entry["scan_provenance"]["fresh"] is True + assert "Scan provenance: fresh" in sink.getvalue() + + +def test_bundled_optional_source_still_includes_support_files(tmp_path, monkeypatch): + from tools.skills_hub import OptionalSkillSource + + root = tmp_path / "optional-skills" + skill = root / "category" / "official-demo" + (skill / "references").mkdir(parents=True) + (skill / "SKILL.md").write_text("---\nname: official-demo\ndescription: demo\n---\n") + (skill / "references" / "all.md").write_text("all") + source = OptionalSkillSource() + source._optional_dir = root + + bundle = source.fetch("official/category/official-demo") + assert bundle is not None + assert set(bundle.files) == {"SKILL.md", "references/all.md"} diff --git a/tools/skills_guard.py b/tools/skills_guard.py index a1a8606d6d47..98acf6a9ff45 100644 --- a/tools/skills_guard.py +++ b/tools/skills_guard.py @@ -25,12 +25,16 @@ Usage: import re import fnmatch import hashlib +import json from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import List, Tuple +SCANNER_VERSION = "skills-guard-v1" + + # --------------------------------------------------------------------------- @@ -87,6 +91,7 @@ class ScanResult: findings: List[Finding] = field(default_factory=list) scanned_at: str = "" summary: str = "" + scan_provenance: dict = field(default_factory=dict) # --------------------------------------------------------------------------- @@ -683,6 +688,76 @@ def scan_skill(skill_path: Path, source: str = "community") -> ScanResult: ) +def _full_content_hash(skill_path: Path) -> str: + """Complete SHA-256 over paths and bytes for scan cache identity.""" + h = hashlib.sha256() + if skill_path.is_dir(): + for file_path in sorted(skill_path.rglob("*")): + rel = file_path.relative_to(skill_path).as_posix() + if file_path.is_symlink(): + h.update(rel.encode() + b"\x00SYMLINK\x00") + elif file_path.is_file(): + h.update(rel.encode() + b"\x00") + h.update(file_path.read_bytes()) + else: + h.update(skill_path.read_bytes()) + return f"sha256:{h.hexdigest()}" + + +def _finding_dict(finding: Finding) -> dict: + return {key: getattr(finding, key) for key in ( + "pattern_id", "severity", "category", "file", "line", "match", "description" + )} + + +def scan_skill_cached( + skill_path: Path, + source: str = "community", + *, + source_url: str = "", + cache_dir: Path | None = None, +) -> Tuple[ScanResult, dict]: + """Return a scan plus attestation, caching only exact current content.""" + bundle_hash = _full_content_hash(skill_path) + cache_root = cache_dir or skill_path.parent / ".scan-cache" + cache_file = cache_root / f"{bundle_hash.split(':', 1)[1]}.json" + try: + cached = json.loads(cache_file.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + cached = None + if (isinstance(cached, dict) + and cached.get("bundle_hash") == bundle_hash + and cached.get("scanner_version") == SCANNER_VERSION + and cached.get("source") == source): + result = ScanResult( + skill_name=skill_path.name, source=source, + trust_level=cached["trust_level"], verdict=cached["verdict"], + findings=[Finding(**item) for item in cached.get("findings", [])], + scanned_at=cached["scanned_at"], summary=cached.get("summary", ""), + ) + provenance = dict(cached) + provenance["fresh"] = False + result.scan_provenance = provenance + return result, provenance + + result = scan_skill(skill_path, source=source) + findings = [_finding_dict(item) for item in result.findings] + provenance = { + "source": source, "source_url": source_url, "bundle_hash": bundle_hash, + "scanner_version": SCANNER_VERSION, "verdict": result.verdict, + "trust_level": result.trust_level, "findings": findings, + "rules": sorted({item["pattern_id"] for item in findings}), + "scanned_at": result.scanned_at, "summary": result.summary, "fresh": True, + } + try: + cache_root.mkdir(parents=True, exist_ok=True) + cache_file.write_text(json.dumps(provenance, indent=2) + "\n", encoding="utf-8") + except OSError: + pass + result.scan_provenance = provenance + return result, provenance + + def should_allow_install(result: ScanResult, force: bool = False) -> Tuple[bool, str]: """ Determine whether a skill should be installed based on scan result and trust. diff --git a/tools/skills_hub.py b/tools/skills_hub.py index 9883b720ee02..c3c2ec2faf56 100644 --- a/tools/skills_hub.py +++ b/tools/skills_hub.py @@ -152,6 +152,46 @@ class SkillBundle: metadata: Dict[str, Any] = field(default_factory=dict) +_ALLOWED_SUPPORT_DIRS = frozenset({"references", "templates", "scripts", "assets"}) +_LOCAL_LINK_RE = re.compile( + r"(?:\]\(|`|(?:^|[\s\"']))((?:references|templates|scripts|assets)/[^\s)`\"'<>]+)", + re.MULTILINE, +) +_SUSPICIOUS_LOCAL_REF_RE = re.compile( + r"(?:references|templates|scripts|assets)/(?:[^\s)`\"'<>]*/)?\.\.(?:/|$)" +) + + +def _referenced_support_paths(skill_md: str) -> Optional[set[str]]: + """Extract safe referenced paths; return None on a traversal attempt.""" + normalized = skill_md.replace("\\", "/") + if _SUSPICIOUS_LOCAL_REF_RE.search(normalized): + return None + paths: set[str] = set() + for match in _LOCAL_LINK_RE.finditer(normalized): + raw = match.group(1).rstrip(".,;:") + try: + safe = _validate_bundle_rel_path(raw) + except ValueError: + return None + if safe.split("/", 1)[0] in _ALLOWED_SUPPORT_DIRS: + paths.add(safe) + return paths + + +def source_url_for_bundle(bundle: SkillBundle) -> str: + """Best available human-facing immutable-source provenance URL.""" + explicit = bundle.metadata.get("source_url") or bundle.metadata.get("url") + if explicit: + return str(explicit) + if bundle.source == "github": + parts = bundle.identifier.split("/", 2) + if len(parts) >= 2: + suffix = f"/tree/main/{parts[2]}" if len(parts) == 3 else "" + return f"https://github.com/{parts[0]}/{parts[1]}{suffix}" + return bundle.identifier + + def _normalize_bundle_path(path_value: str, *, field_name: str, allow_nested: bool) -> str: """Normalize and validate bundle-controlled paths before touching disk.""" if not isinstance(path_value, str): @@ -601,9 +641,39 @@ class GitHubSource(SkillSource): repo = f"{parts[0]}/{parts[1]}" skill_path = parts[2] - files = self._download_directory(repo, skill_path) - if not files or "SKILL.md" not in files: + skill_md = self._fetch_file_content(repo, f"{skill_path.rstrip('/')}/SKILL.md") + if skill_md is None: return None + referenced = _referenced_support_paths(skill_md) + if referenced is None: + return None + + files: Dict[str, Union[str, bytes]] = {"SKILL.md": skill_md} + tree = self._get_repo_tree(repo) + if tree is not None: + _branch, entries = tree + prefix = f"{skill_path.rstrip('/')}/" + wanted_roots = {path.split("/", 1)[0] for path in referenced} + for item in entries: + item_path = item.get("path", "") + if not item_path.startswith(prefix): + continue + rel_path = item_path[len(prefix):] + if rel_path.split("/", 1)[0] not in wanted_roots: + continue + if item.get("type") != "blob" or item.get("mode") == "120000": + logger.warning("Rejected non-regular file in skill bundle: %s", item_path) + return None + content = self._fetch_file_content(repo, item_path) + if content is None: + return None + files[_validate_bundle_rel_path(rel_path)] = content + else: + for rel_path in referenced: + content = self._fetch_file_content(repo, f"{skill_path.rstrip('/')}/{rel_path}") + if content is None: + return None + files[rel_path] = content skill_name = skill_path.rstrip("/").split("/")[-1] trust = self.trust_level_for(identifier) @@ -614,6 +684,7 @@ class GitHubSource(SkillSource): source="github", identifier=identifier, trust_level=trust, + metadata={"source_url": f"https://github.com/{repo}/tree/main/{skill_path}"}, ) def inspect(self, identifier: str) -> Optional[SkillMeta]: @@ -1318,12 +1389,12 @@ class WellKnownSkillSource(SkillSource): # --------------------------------------------------------------------------- class UrlSource(SkillSource): - """Fetch a single-file SKILL.md skill directly from an HTTP(S) URL. + """Fetch SKILL.md plus explicitly referenced, allowlisted support files. The identifier IS the URL (e.g. ``https://example.com/path/SKILL.md``). - Only single-file skills are supported — multi-file skills with - ``references/`` or ``scripts/`` subfolders need a manifest we can't - discover from a bare URL. + Bare URLs cannot safely enumerate a repository, so only exact references + below references/templates/scripts/assets are fetched. Other repository + files are never copied. The skill name is read from the ``name:`` field in the SKILL.md YAML frontmatter (with a URL-slug fallback). Trust level is always @@ -1402,6 +1473,19 @@ class UrlSource(SkillSource): fm = GitHubSource._parse_frontmatter_quick(text) name = self._resolve_skill_name(fm, url) + referenced = _referenced_support_paths(text) + if referenced is None: + return None + files: Dict[str, Union[str, bytes]] = {"SKILL.md": text} + base_url = url.rsplit("/", 1)[0] + "/" + for rel_path in sorted(referenced): + support_url = urljoin(base_url, rel_path) + if urlparse(support_url).netloc != urlparse(url).netloc: + return None + content = self._fetch_text(support_url) + if content is None: + return None + files[rel_path] = content # When auto-resolution fails, return a bundle with an empty name and # ``awaiting_name=True`` in metadata. The install flow (``do_install``) @@ -1418,11 +1502,11 @@ class UrlSource(SkillSource): return SkillBundle( name=skill_name, - files={"SKILL.md": text}, + files=files, source="url", identifier=url, trust_level="community", - metadata={"url": url, "awaiting_name": not skill_name}, + metadata={"url": url, "source_url": url, "awaiting_name": not skill_name}, ) @staticmethod @@ -3304,6 +3388,7 @@ class HubLockFile: install_path: str, files: List[str], metadata: Optional[Dict[str, Any]] = None, + scan_provenance: Optional[Dict[str, Any]] = None, ) -> None: # Validate both the skill name and the install path SHAPE before # writing into lock.json. A poisoned lock entry is the precondition @@ -3321,6 +3406,7 @@ class HubLockFile: "install_path": safe_install_path, "files": files, "metadata": metadata or {}, + "scan_provenance": scan_provenance or {}, "installed_at": datetime.now(timezone.utc).isoformat(), "updated_at": datetime.now(timezone.utc).isoformat(), } @@ -3461,6 +3547,7 @@ def install_from_quarantine( category: str, bundle: SkillBundle, scan_result: ScanResult, + scan_provenance: Optional[Dict[str, Any]] = None, ) -> Path: """Move a scanned skill from quarantine into the skills directory.""" safe_skill_name = _validate_skill_name(skill_name) @@ -3529,6 +3616,7 @@ def install_from_quarantine( install_path=str(install_dir.relative_to(_skills_dir())), files=list(bundle.files.keys()), metadata=bundle.metadata, + scan_provenance=scan_provenance or getattr(scan_result, "scan_provenance", None), ) append_audit_log(