Compare commits

...

1 Commits

Author SHA1 Message Date
teknium1
ea39cf04d0 fix(skills): install referenced bundle files with scan provenance 2026-07-09 18:56:03 -07:00
4 changed files with 388 additions and 10 deletions

View File

@@ -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)

View File

@@ -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"}

View File

@@ -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.

View File

@@ -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(