4965 lines
193 KiB
Python
4965 lines
193 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Skills Hub — Source adapters and hub state management for the Hermes Skills Hub.
|
||
|
||
This is a library module (not an agent tool). It provides:
|
||
- GitHubAuth: Shared GitHub API authentication (PAT, gh CLI, GitHub App)
|
||
- SkillSource ABC: Interface for all skill registry adapters
|
||
- OptionalSkillSource: Official optional skills shipped with the repo (not activated by default)
|
||
- GitHubSource: Fetch skills from any GitHub repo via the Contents API
|
||
- HubLockFile: Track provenance of installed hub skills
|
||
- Hub state directory management (quarantine, audit log, taps, index cache)
|
||
|
||
Used by hermes_cli/skills_hub.py for CLI commands and the /skills slash command.
|
||
"""
|
||
|
||
import hashlib
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import time
|
||
from abc import ABC, abstractmethod
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path, PurePosixPath
|
||
from hermes_constants import get_hermes_home
|
||
from hermes_cli._subprocess_compat import windows_hide_flags
|
||
from agent.skill_utils import is_excluded_skill_path
|
||
from typing import Any, Dict, List, Optional, Tuple, Union
|
||
from urllib.parse import quote, unquote, urljoin, urlparse, urlsplit, urlunparse
|
||
|
||
import httpx
|
||
import yaml
|
||
|
||
from tools.skills_guard import (
|
||
ScanResult, content_hash, TRUSTED_REPOS,
|
||
)
|
||
from tools.url_safety import is_safe_url
|
||
from tools.website_policy import check_website_access
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Paths
|
||
# ---------------------------------------------------------------------------
|
||
# Resolved per-call (not frozen at import) so the profile override is honored;
|
||
# import-time constants leaked across profiles in single-process multi-profile
|
||
# runtimes. Legacy names (SKILLS_DIR, ...) are re-exposed via __getattr__ below
|
||
# so external `from tools.skills_hub import SKILLS_DIR` callers still work.
|
||
|
||
INDEX_CACHE_TTL = 3600 # 1 hour
|
||
|
||
|
||
# _override lets a test-injected real module attribute (patch.object/monkeypatch
|
||
# on SKILLS_DIR etc.) win over dynamic resolution; None means resolve live.
|
||
def _override(name: str):
|
||
return globals().get(name)
|
||
|
||
|
||
def _hermes_home() -> Path:
|
||
return get_hermes_home()
|
||
|
||
|
||
def _skills_dir() -> Path:
|
||
forced = _override("SKILLS_DIR")
|
||
return Path(forced) if forced is not None else _hermes_home() / "skills"
|
||
|
||
|
||
def _hub_dir() -> Path:
|
||
forced = _override("HUB_DIR")
|
||
return Path(forced) if forced is not None else _skills_dir() / ".hub"
|
||
|
||
|
||
def _lock_file() -> Path:
|
||
forced = _override("LOCK_FILE")
|
||
return Path(forced) if forced is not None else _hub_dir() / "lock.json"
|
||
|
||
|
||
def _quarantine_dir() -> Path:
|
||
forced = _override("QUARANTINE_DIR")
|
||
return Path(forced) if forced is not None else _hub_dir() / "quarantine"
|
||
|
||
|
||
def _audit_log() -> Path:
|
||
forced = _override("AUDIT_LOG")
|
||
return Path(forced) if forced is not None else _hub_dir() / "audit.log"
|
||
|
||
|
||
def _taps_file() -> Path:
|
||
forced = _override("TAPS_FILE")
|
||
return Path(forced) if forced is not None else _hub_dir() / "taps.json"
|
||
|
||
|
||
def _index_cache_dir() -> Path:
|
||
forced = _override("INDEX_CACHE_DIR")
|
||
return Path(forced) if forced is not None else _hub_dir() / "index-cache"
|
||
|
||
|
||
_DYNAMIC_PATH_RESOLVERS = {
|
||
"HERMES_HOME": _hermes_home,
|
||
"SKILLS_DIR": _skills_dir,
|
||
"HUB_DIR": _hub_dir,
|
||
"LOCK_FILE": _lock_file,
|
||
"QUARANTINE_DIR": _quarantine_dir,
|
||
"AUDIT_LOG": _audit_log,
|
||
"TAPS_FILE": _taps_file,
|
||
"INDEX_CACHE_DIR": _index_cache_dir,
|
||
}
|
||
|
||
|
||
def __getattr__(name: str):
|
||
"""Resolve legacy path constants dynamically (PEP 562) so they reflect the
|
||
active profile override; a test's patch.object-set real attribute shadows it."""
|
||
resolver = _DYNAMIC_PATH_RESOLVERS.get(name)
|
||
if resolver is not None:
|
||
return resolver()
|
||
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
|
||
|
||
_REDIRECT_STATUS_CODES = {301, 302, 303, 307, 308}
|
||
_MAX_SKILL_FETCH_REDIRECTS = 5
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Data models
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@dataclass
|
||
class SkillMeta:
|
||
"""Minimal metadata returned by search results."""
|
||
name: str
|
||
description: str
|
||
source: str # "official", "github", "clawhub", "lobehub"
|
||
identifier: str # source-specific ID (e.g. "openai/skills/skill-creator")
|
||
trust_level: str # "builtin" | "trusted" | "community"
|
||
repo: Optional[str] = None
|
||
path: Optional[str] = None
|
||
tags: List[str] = field(default_factory=list)
|
||
extra: Dict[str, Any] = field(default_factory=dict)
|
||
|
||
|
||
@dataclass
|
||
class SkillBundle:
|
||
"""A downloaded skill ready for quarantine/scanning/installation."""
|
||
name: str
|
||
files: Dict[str, Union[str, bytes]] # relative_path -> file content
|
||
source: str
|
||
identifier: str
|
||
trust_level: str
|
||
metadata: Dict[str, Any] = field(default_factory=dict)
|
||
|
||
|
||
_ALLOWED_SUPPORT_DIRS = frozenset({"references", "templates", "scripts", "assets", "examples"})
|
||
_LOCAL_LINK_RE = re.compile(
|
||
r"(?:\]\(|`|(?:^|[\s\"']))((?:references|templates|scripts|assets|examples)/[^\s)`\"'<>]+)",
|
||
re.MULTILINE,
|
||
)
|
||
_SUSPICIOUS_LOCAL_REF_RE = re.compile(
|
||
r"(?:references|templates|scripts|assets|examples)/(?:[^\s)`\"'<>]*/)?\.\.(?:/|$)"
|
||
)
|
||
_VALUELESS_QUERY_FLAG_RE = re.compile(
|
||
r"(?:[A-Za-z0-9_~-]|%[0-9A-Fa-f]{2})+\Z"
|
||
)
|
||
|
||
|
||
def _query_is_concrete(query: str) -> bool:
|
||
"""Whether ``query`` is unambiguously URL syntax rather than glob prose.
|
||
|
||
A non-empty ``key=value`` part is concrete URL syntax regardless of key
|
||
spelling, preserving the established behavior. Valueless flags are also
|
||
accepted when they are RFC 3986 unreserved-token shaped (including valid
|
||
percent escapes), rather than being restricted to a fixed allowlist.
|
||
|
||
Deliberately exclude ``.`` from valueless flags. A suffix such as
|
||
``?x.md`` or ``?.md`` is indistinguishable from a single-character glob
|
||
completing a filename in inline prose. Brackets and additional question
|
||
marks are excluded for the same reason. This syntactic ambiguity policy
|
||
preserves ordinary flags such as ``?view`` and ``?preview-mode`` while
|
||
rejecting glob-shaped references without guessing flag names.
|
||
"""
|
||
parts = query.split("&")
|
||
return all(
|
||
("=" in part and bool(part.split("=", 1)[0]))
|
||
or bool(_VALUELESS_QUERY_FLAG_RE.fullmatch(part))
|
||
for part in parts
|
||
)
|
||
|
||
# Same-directory links (``](./FILE.ext)`` / ``](FILE.ext)``) — siblings of
|
||
# SKILL.md that the document explicitly links. Skills legitimately ship
|
||
# supporting docs next to SKILL.md instead of under a support directory
|
||
# (e.g. mattpocock/skills' domain-modeling links ./CONTEXT-FORMAT.md);
|
||
# dropping them made the install "succeed" while the bundle came out with
|
||
# unresolved links (#96310). The trailing extension requirement keeps prose
|
||
# words out; the code-side checks keep this strictly to the skill's own
|
||
# directory (support-dir links stay on _LOCAL_LINK_RE).
|
||
_SAMEDIR_LINK_RE = re.compile(r"\]\(([^)\s\"'<>]+)")
|
||
_SAMEDIR_NAME_RE = re.compile(r"^(?:\./)?[A-Za-z0-9][A-Za-z0-9._-]*$")
|
||
|
||
|
||
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):
|
||
candidate = match.group(1).rstrip(".,;:")
|
||
if candidate.endswith("?"):
|
||
continue
|
||
parsed = urlsplit(candidate)
|
||
raw = unquote(parsed.path)
|
||
if any(char in raw for char in "*?[]"):
|
||
continue
|
||
if parsed.query and not _query_is_concrete(parsed.query):
|
||
continue
|
||
try:
|
||
safe = _validate_bundle_rel_path(raw)
|
||
except ValueError:
|
||
return None
|
||
if safe.split("/", 1)[0] in _ALLOWED_SUPPORT_DIRS:
|
||
# Prose placeholders — e.g. ``references/type-<name>.md`` (which
|
||
# the link regex truncates at ``<`` to the bare prefix
|
||
# ``references/type-``) — are agent instructions, not files.
|
||
# Glob shapes (*, ?, []) were already rejected on the raw
|
||
# candidate above; a truncated placeholder leaves a basename
|
||
# ending in a separator, which no real file uses. No extension
|
||
# requirement: extensionless support files
|
||
# (``references/LICENSE``) are legitimate.
|
||
base = safe.rsplit("/", 1)[-1]
|
||
if re.search(r"[*?<>]", safe) or not re.search(r"[A-Za-z0-9]$", base):
|
||
continue
|
||
paths.add(safe)
|
||
for match in _SAMEDIR_LINK_RE.finditer(normalized):
|
||
raw = match.group(1).rstrip(".,;:")
|
||
# Canonicalize first: drop query/fragment components (``?raw=1``,
|
||
# ``#section``) and percent-decode — the same normalization the
|
||
# support-dir branch applies via urlsplit+unquote — then strip a
|
||
# leading ``./``. The set below deduplicates case-SENSITIVE repeats;
|
||
# case-VARIANT collisions are rejected rather than merged (below).
|
||
name = unquote(urlsplit(raw).path)
|
||
name = name[2:] if name.startswith("./") else name
|
||
# External URLs, anchors, mailto and site-absolute targets are not
|
||
# same-directory file links — leave them to their own resolution.
|
||
if not name or "://" in raw or raw.startswith(("mailto:", "#", "/")):
|
||
continue
|
||
# A ``..`` prefix is a traversal attempt — same fail-closed contract
|
||
# as the support-dir branch above, before any shape-based skipping.
|
||
if name.startswith(".."):
|
||
return None
|
||
# Only unambiguous file links: an extension, no internal slash, and
|
||
# never SKILL.md itself (that IS the bundle root). The casefold
|
||
# check keeps a ``skill.md`` link from shipping as a bundle entry
|
||
# that collides with SKILL.md on case-insensitive filesystems
|
||
# (macOS/Windows) — skipped, not merged, so the bundle root is
|
||
# never overwritten (#96310 review).
|
||
if (
|
||
"/" in name
|
||
or name.casefold() == "skill.md"
|
||
or "." not in name.lstrip(".")
|
||
):
|
||
continue
|
||
if not _SAMEDIR_NAME_RE.match(name):
|
||
continue
|
||
try:
|
||
safe = _validate_bundle_rel_path(name)
|
||
except ValueError:
|
||
return None
|
||
paths.add(safe)
|
||
# Case-folded collision among the accepted same-dir names themselves
|
||
# (``A.md`` + ``a.md``) would also collide on install — drop the pair
|
||
# rather than guess which variant the author meant.
|
||
folded: dict[str, str] = {}
|
||
for p in sorted(paths):
|
||
key = p.casefold()
|
||
if key in folded:
|
||
paths.discard(folded[key])
|
||
paths.discard(p)
|
||
else:
|
||
folded[key] = p
|
||
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):
|
||
raise ValueError(f"Unsafe {field_name}: expected a string")
|
||
|
||
raw = path_value.strip()
|
||
if not raw:
|
||
raise ValueError(f"Unsafe {field_name}: empty path")
|
||
|
||
normalized = raw.replace("\\", "/")
|
||
path = PurePosixPath(normalized)
|
||
parts = [part for part in path.parts if part not in {"", "."}]
|
||
|
||
if normalized.startswith("/") or path.is_absolute():
|
||
raise ValueError(f"Unsafe {field_name}: {path_value}")
|
||
if not parts or any(part == ".." for part in parts):
|
||
raise ValueError(f"Unsafe {field_name}: {path_value}")
|
||
# Reject a colon in any component. On Windows a colon marks either a drive
|
||
# (``C:`` / ``C:foo``) or an NTFS Alternate Data Stream: a bundle member
|
||
# named ``file.py:payload`` writes hidden, scanner-invisible bytes into the
|
||
# visible ``file.py`` (rglob-based review never enumerates the stream).
|
||
# ``/`` is the only legal separator once normalized, so no portable bundle
|
||
# path needs a colon in a component.
|
||
if any(":" in part for part in parts):
|
||
raise ValueError(f"Unsafe {field_name}: {path_value}")
|
||
if not allow_nested and len(parts) != 1:
|
||
raise ValueError(f"Unsafe {field_name}: {path_value}")
|
||
|
||
return "/".join(parts)
|
||
|
||
|
||
def _validate_skill_name(name: str) -> str:
|
||
return _normalize_bundle_path(name, field_name="skill name", allow_nested=False)
|
||
|
||
|
||
def _validate_install_parent_path(category: str) -> str:
|
||
return _normalize_bundle_path(category, field_name="install parent path", allow_nested=True)
|
||
|
||
|
||
def _normalize_lock_install_path(install_path: str, skill_name: str) -> str:
|
||
"""Validate a skill install path before it touches the lock file or disk.
|
||
|
||
Lock-file ``install_path`` entries are the source-of-truth for where
|
||
``uninstall_skill`` will call ``shutil.rmtree``. A poisoned or buggy
|
||
entry — empty string, ``"."``, an absolute path, ``../..`` traversal,
|
||
or anything whose final component doesn't match the skill name — would
|
||
let ``rmtree`` wipe either the entire ``skills/`` tree or content
|
||
outside it.
|
||
|
||
Enforce that ``install_path`` ends with ``<skill_name>``. Nested
|
||
official optional skills may legitimately install below paths such as
|
||
``mlops/training/<skill_name>``; traversal, absolute paths, empty paths,
|
||
and mismatched final components are still rejected.
|
||
"""
|
||
safe_skill_name = _validate_skill_name(skill_name)
|
||
normalized = _normalize_bundle_path(
|
||
install_path,
|
||
field_name="install path",
|
||
allow_nested=True,
|
||
)
|
||
parts = normalized.split("/")
|
||
if not parts or parts[-1] != safe_skill_name:
|
||
raise ValueError(f"Unsafe install path: {install_path}")
|
||
return normalized
|
||
|
||
|
||
def _is_path_redirect(path: Path) -> bool:
|
||
"""True when ``path`` is a symlink or (on Windows) a directory junction.
|
||
|
||
Either form lets an attacker who can write into the ``skills/`` tree
|
||
redirect a subsequent ``rmtree`` to content outside it. ``is_junction``
|
||
only exists on Python 3.12+ Windows; gate with ``hasattr``.
|
||
"""
|
||
return path.is_symlink() or (hasattr(path, "is_junction") and path.is_junction())
|
||
|
||
|
||
def _resolve_lock_install_path(install_path: str, skill_name: str) -> Path:
|
||
"""Resolve a lock-file install path without allowing escapes from ``SKILLS_DIR``.
|
||
|
||
Two layers of defence on top of the existing ``is_relative_to`` check
|
||
that's been on main:
|
||
|
||
1. Walk the path component-by-component and refuse if any intermediate
|
||
component is a symlink/junction (a path resolution that follows a
|
||
symlink to outside skills/ would otherwise be hidden by Path.resolve).
|
||
2. After resolve(), reject not just escape-out but also ``resolved == SKILLS_DIR``
|
||
— an empty/``"."``/``""`` install_path resolves to the skills root itself,
|
||
and ``rmtree(SKILLS_DIR)`` would wipe every installed skill.
|
||
"""
|
||
normalized = _normalize_lock_install_path(install_path, skill_name)
|
||
skills_dir = _skills_dir()
|
||
skills_root = skills_dir.resolve()
|
||
|
||
target = skills_dir
|
||
for part in normalized.split("/"):
|
||
target = target / part
|
||
if _is_path_redirect(target):
|
||
raise ValueError(f"Unsafe install path: {install_path}")
|
||
|
||
target = target.resolve()
|
||
if target == skills_root or not target.is_relative_to(skills_root):
|
||
raise ValueError(f"Unsafe install path: {install_path}")
|
||
return target
|
||
|
||
|
||
def _ssrf_safe_http_get(url: str, *, timeout: int = 20) -> httpx.Response:
|
||
"""Fetch one URL with connect-time SSRF validation and no automatic redirects."""
|
||
from tools.url_safety import create_ssrf_safe_client
|
||
|
||
with create_ssrf_safe_client(timeout=timeout, follow_redirects=False) as client:
|
||
return client.get(url)
|
||
|
||
|
||
def _guarded_http_get(url: str, *, timeout: int = 20) -> Optional[httpx.Response]:
|
||
"""Fetch a URL with SSRF and redirect-target validation."""
|
||
from tools.url_safety import SSRFConnectionBlocked
|
||
|
||
current_url = url
|
||
|
||
for _ in range(_MAX_SKILL_FETCH_REDIRECTS + 1):
|
||
if not is_safe_url(current_url):
|
||
logger.warning("Blocked unsafe Skills Hub URL: %s", current_url)
|
||
return None
|
||
|
||
blocked = check_website_access(current_url)
|
||
if blocked:
|
||
logger.info(
|
||
"Blocked Skills Hub fetch for %s by rule %s",
|
||
blocked["host"],
|
||
blocked["rule"],
|
||
)
|
||
return None
|
||
|
||
try:
|
||
resp = _ssrf_safe_http_get(current_url, timeout=timeout)
|
||
except (SSRFConnectionBlocked, httpx.HTTPError) as exc:
|
||
logger.debug("Skills Hub fetch failed for %s: %s", current_url, exc)
|
||
return None
|
||
|
||
if resp.status_code in _REDIRECT_STATUS_CODES:
|
||
location = getattr(resp, "headers", {}).get("location")
|
||
if not location:
|
||
return None
|
||
current_url = urljoin(current_url, location)
|
||
continue
|
||
|
||
return resp
|
||
|
||
logger.warning("Skills Hub fetch exceeded redirect limit for %s", url)
|
||
return None
|
||
|
||
|
||
def _validate_bundle_rel_path(rel_path: str) -> str:
|
||
return _normalize_bundle_path(rel_path, field_name="bundle file path", allow_nested=True)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# GitHub Authentication
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class GitHubAuth:
|
||
"""
|
||
GitHub API authentication. Tries methods in priority order:
|
||
1. GITHUB_TOKEN / GH_TOKEN env var (PAT — the default)
|
||
2. `gh auth token` subprocess (if gh CLI is installed)
|
||
3. GitHub App JWT + installation token (if app credentials configured)
|
||
4. Unauthenticated (60 req/hr, public repos only)
|
||
"""
|
||
|
||
def __init__(self):
|
||
self._cached_token: Optional[str] = None
|
||
self._cached_method: Optional[str] = None
|
||
self._app_token_expiry: float = 0
|
||
|
||
def get_headers(self) -> Dict[str, str]:
|
||
"""Return authorization headers for GitHub API requests."""
|
||
token = self._resolve_token()
|
||
headers = {"Accept": "application/vnd.github.v3+json"}
|
||
if token:
|
||
headers["Authorization"] = f"token {token}"
|
||
return headers
|
||
|
||
def is_authenticated(self) -> bool:
|
||
return self._resolve_token() is not None
|
||
|
||
def auth_method(self) -> str:
|
||
"""Return which auth method is active: 'pat', 'gh-cli', 'github-app', or 'anonymous'."""
|
||
self._resolve_token()
|
||
return self._cached_method or "anonymous"
|
||
|
||
def _resolve_token(self) -> Optional[str]:
|
||
# Return cached token if still valid
|
||
if self._cached_token:
|
||
if self._cached_method != "github-app" or time.time() < self._app_token_expiry:
|
||
return self._cached_token
|
||
|
||
# 1. Environment variable (profile-scoped under a multiplexed gateway)
|
||
from agent.secret_scope import get_secret
|
||
token = get_secret("GITHUB_TOKEN") or get_secret("GH_TOKEN")
|
||
if token:
|
||
self._cached_token = token
|
||
self._cached_method = "pat"
|
||
return token
|
||
|
||
# 2. gh CLI
|
||
token = self._try_gh_cli()
|
||
if token:
|
||
self._cached_token = token
|
||
self._cached_method = "gh-cli"
|
||
return token
|
||
|
||
# 3. GitHub App
|
||
token = self._try_github_app()
|
||
if token:
|
||
self._cached_token = token
|
||
self._cached_method = "github-app"
|
||
self._app_token_expiry = time.time() + 3500 # ~58 min (tokens last 1 hour)
|
||
return token
|
||
|
||
self._cached_method = "anonymous"
|
||
return None
|
||
|
||
def _try_gh_cli(self) -> Optional[str]:
|
||
"""Try to get a token from the gh CLI."""
|
||
try:
|
||
result = subprocess.run(
|
||
["gh", "auth", "token"],
|
||
capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=5,
|
||
stdin=subprocess.DEVNULL,
|
||
creationflags=windows_hide_flags(),
|
||
)
|
||
if result.returncode == 0 and result.stdout.strip():
|
||
return result.stdout.strip()
|
||
except (FileNotFoundError, subprocess.TimeoutExpired) as e:
|
||
logger.debug("gh CLI token lookup failed: %s", e)
|
||
return None
|
||
|
||
def _try_github_app(self) -> Optional[str]:
|
||
"""Try GitHub App JWT authentication if credentials are configured."""
|
||
from agent.secret_scope import get_secret
|
||
app_id = get_secret("GITHUB_APP_ID")
|
||
key_path = get_secret("GITHUB_APP_PRIVATE_KEY_PATH")
|
||
installation_id = get_secret("GITHUB_APP_INSTALLATION_ID")
|
||
|
||
if not all([app_id, key_path, installation_id]):
|
||
return None
|
||
|
||
try:
|
||
import jwt # PyJWT
|
||
except ImportError:
|
||
logger.debug("PyJWT not installed, skipping GitHub App auth")
|
||
return None
|
||
|
||
try:
|
||
key_file = Path(key_path)
|
||
if not key_file.exists():
|
||
return None
|
||
private_key = key_file.read_text(encoding="utf-8")
|
||
|
||
now = int(time.time())
|
||
payload = {
|
||
"iat": now - 60,
|
||
"exp": now + (10 * 60),
|
||
"iss": app_id,
|
||
}
|
||
encoded_jwt = jwt.encode(payload, private_key, algorithm="RS256")
|
||
|
||
resp = httpx.post(
|
||
f"https://api.github.com/app/installations/{installation_id}/access_tokens",
|
||
headers={
|
||
"Authorization": f"Bearer {encoded_jwt}",
|
||
"Accept": "application/vnd.github.v3+json",
|
||
},
|
||
timeout=10,
|
||
)
|
||
if resp.status_code == 201:
|
||
return resp.json().get("token")
|
||
except Exception as e:
|
||
logger.debug("GitHub App auth failed: %s", e)
|
||
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Source adapter interface
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class SkillSource(ABC):
|
||
"""Abstract base for all skill registry adapters."""
|
||
|
||
@abstractmethod
|
||
def search(self, query: str, limit: int = 10) -> List[SkillMeta]:
|
||
"""Search for skills matching a query string."""
|
||
...
|
||
|
||
@abstractmethod
|
||
def fetch(self, identifier: str) -> Optional[SkillBundle]:
|
||
"""Download a skill bundle by identifier."""
|
||
...
|
||
|
||
@abstractmethod
|
||
def inspect(self, identifier: str) -> Optional[SkillMeta]:
|
||
"""Fetch metadata for a skill without downloading all files."""
|
||
...
|
||
|
||
@abstractmethod
|
||
def source_id(self) -> str:
|
||
"""Unique identifier for this source (e.g. 'github', 'clawhub')."""
|
||
...
|
||
|
||
def trust_level_for(self, identifier: str) -> str:
|
||
"""Determine trust level for a skill from this source."""
|
||
return "community"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# GitHub source adapter
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# Map a GitHub tap repo (owner/repo) to the human-facing provider label used
|
||
# in the docs-site catalog (website/scripts/extract-skills.py::GITHUB_TAP_LABELS).
|
||
# The runtime index collapses every GitHub tap into source="github"; stamping
|
||
# this provider label onto each skill's ``extra`` keeps the per-tap identity
|
||
# (NVIDIA / OpenAI / Anthropic / HuggingFace / gstack / ...) searchable and
|
||
# filterable at the CLI without disturbing the source="github" dedup / floor /
|
||
# index-skip logic that keys off the bare source id.
|
||
GITHUB_TAP_PROVIDERS = {
|
||
"openai/skills": "OpenAI",
|
||
"anthropics/skills": "Anthropic",
|
||
"huggingface/skills": "HuggingFace",
|
||
"nvidia/skills": "NVIDIA",
|
||
"voltagent/awesome-agent-skills": "VoltAgent",
|
||
"garrytan/gstack": "gstack",
|
||
"minimax-ai/cli": "MiniMax",
|
||
}
|
||
|
||
|
||
def github_provider_for(repo: str) -> Optional[str]:
|
||
"""Return the provider label for a GitHub tap repo, or None.
|
||
|
||
``repo`` is ``owner/repo``; matched case-insensitively so ``NVIDIA/skills``
|
||
and ``nvidia/skills`` both resolve to ``"NVIDIA"``.
|
||
"""
|
||
if not repo:
|
||
return None
|
||
return GITHUB_TAP_PROVIDERS.get(repo.strip().lower())
|
||
|
||
|
||
# Lowercased set of accepted ``--source`` provider filters. These are not real
|
||
# source ids — they narrow the merged results to GitHub-tap skills carrying the
|
||
# matching ``extra.provider`` label (see ``_filter_results_by_provider``).
|
||
_PROVIDER_FILTER_VALUES = frozenset(v.lower() for v in GITHUB_TAP_PROVIDERS.values())
|
||
|
||
|
||
def _filter_results_by_provider(
|
||
results: List["SkillMeta"], provider: str
|
||
) -> List["SkillMeta"]:
|
||
"""Keep only results whose ``extra.provider`` matches ``provider``.
|
||
|
||
An explicit provider filter (e.g. ``--source nvidia``) means "show me that
|
||
provider's skills" — so it narrows to exactly those, without injecting the
|
||
official catalog the unfiltered browse/search would lead with.
|
||
"""
|
||
want = provider.strip().lower()
|
||
return [
|
||
r for r in results
|
||
if str((r.extra or {}).get("provider", "")).lower() == want
|
||
]
|
||
|
||
|
||
class GitHubSource(SkillSource):
|
||
"""Fetch skills from GitHub repos via the Contents API."""
|
||
|
||
DEFAULT_TAPS = [
|
||
# NOTE: openai/skills moved its content into skills/.curated/ (and
|
||
# skills/.system/ for system-level skills). _list_skills_in_repo
|
||
# skips directories starting with "." or "_", so we point both
|
||
# entries at the inner paths directly.
|
||
{"repo": "openai/skills", "path": "skills/.curated/"},
|
||
{"repo": "openai/skills", "path": "skills/.system/"},
|
||
{"repo": "anthropics/skills", "path": "skills/"},
|
||
{"repo": "huggingface/skills", "path": "skills/"},
|
||
# NVIDIA/skills: NVIDIA-verified skills for CUDA-X, AIQ, cuOpt,
|
||
# cuPyNumeric, DeepStream, NeMo, NemoClaw, etc. Each skill ships
|
||
# alongside a signed `skill.oms.sig`, an OMS-signed `skill-card.md`
|
||
# (governance card), and an `evals/` directory — synced daily from
|
||
# the NVIDIA product repos. Treated as `trusted` (see
|
||
# `tools/skills_guard.py::TRUSTED_REPOS`). Sample layout:
|
||
# https://github.com/NVIDIA/skills/tree/main/skills
|
||
{"repo": "NVIDIA/skills", "path": "skills/"},
|
||
{"repo": "garrytan/gstack", "path": ""},
|
||
]
|
||
|
||
def __init__(self, auth: GitHubAuth, extra_taps: Optional[List[Dict]] = None):
|
||
self.auth = auth
|
||
self.taps = list(self.DEFAULT_TAPS)
|
||
if extra_taps:
|
||
self.taps.extend(extra_taps)
|
||
# Per-instance cache: repo -> (default_branch, tree_entries)
|
||
# Survives within a single search/install flow, avoiding redundant API calls.
|
||
self._tree_cache: Dict[str, Tuple[str, List[dict]]] = {}
|
||
self._tree_revisions: Dict[str, str] = {}
|
||
# Per-repo cache of the optional skills.sh.json grouping sidecar,
|
||
# mapping skill_name -> human-readable grouping title. ``None`` means
|
||
# "fetched, no sidecar"; a missing key means "not fetched yet".
|
||
self._skillsh_groupings: Dict[str, Optional[Dict[str, str]]] = {}
|
||
# Set when GitHub returns 403 with rate limit exhausted
|
||
self._rate_limited: bool = False
|
||
|
||
def source_id(self) -> str:
|
||
return "github"
|
||
|
||
@property
|
||
def is_rate_limited(self) -> bool:
|
||
"""Whether GitHub API rate limit was hit during operations."""
|
||
return self._rate_limited
|
||
|
||
def trust_level_for(self, identifier: str) -> str:
|
||
# identifier format: "owner/repo/path/to/skill"
|
||
parts = identifier.split("/", 2)
|
||
if len(parts) >= 2:
|
||
repo = f"{parts[0]}/{parts[1]}"
|
||
if repo in TRUSTED_REPOS:
|
||
return "trusted"
|
||
return "community"
|
||
|
||
def search(self, query: str, limit: int = 10) -> List[SkillMeta]:
|
||
"""Search all taps for skills matching the query."""
|
||
results: List[SkillMeta] = []
|
||
query_lower = query.lower()
|
||
|
||
for tap in self.taps:
|
||
try:
|
||
skills = self._list_skills_in_repo(tap["repo"], tap.get("path", ""))
|
||
for skill in skills:
|
||
searchable = f"{skill.name} {skill.description} {' '.join(skill.tags)}".lower()
|
||
if query_lower in searchable:
|
||
results.append(skill)
|
||
except Exception as e:
|
||
logger.debug("Failed to search %s: %s", tap['repo'], e)
|
||
continue
|
||
|
||
# Deduplicate by identifier, preferring higher trust levels.
|
||
# identifier is unique per skill; name is not (two configured taps can
|
||
# publish skills with the same name but different identifiers).
|
||
_trust_rank = {"builtin": 2, "trusted": 1, "community": 0}
|
||
seen = {}
|
||
for r in results:
|
||
if r.identifier not in seen:
|
||
seen[r.identifier] = r
|
||
elif _trust_rank.get(r.trust_level, 0) > _trust_rank.get(seen[r.identifier].trust_level, 0):
|
||
seen[r.identifier] = r
|
||
results = list(seen.values())
|
||
|
||
return results[:limit]
|
||
|
||
def fetch(self, identifier: str) -> Optional[SkillBundle]:
|
||
"""
|
||
Download a skill from GitHub.
|
||
identifier format: "owner/repo/path/to/skill-dir"
|
||
"""
|
||
parts = identifier.split("/", 2)
|
||
if len(parts) < 3:
|
||
return None
|
||
|
||
repo = f"{parts[0]}/{parts[1]}"
|
||
skill_path = parts[2]
|
||
|
||
# Resolve the tree FIRST so every byte fetch in this install —
|
||
# SKILL.md included — can be pinned to the same revision. Without the
|
||
# pin the /contents endpoint floats to the default-branch HEAD and
|
||
# the downloaded bytes can come from a NEWER revision than the tree
|
||
# the paths were validated against (TOCTOU between "the tree says
|
||
# this is a regular blob" and "what actually gets downloaded").
|
||
# Idempotent + cached, so callers that already primed the tree pay
|
||
# nothing extra.
|
||
tree = self._get_repo_tree(repo)
|
||
pinned_ref = self._tree_revisions.get(repo)
|
||
skill_md = self._fetch_file_content(
|
||
repo, f"{skill_path.rstrip('/')}/SKILL.md", ref=pinned_ref
|
||
)
|
||
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}
|
||
if tree is not None:
|
||
# Download the FULL skill directory, not just SKILL.md-linked
|
||
# paths. Link-driven fetching silently dropped every support file
|
||
# a skill keeps under a non-canonical dir name (`reference/`,
|
||
# `agents/`, root-level LICENSE/params files) or that only a
|
||
# reference file links — the exact gap the optional-skills live
|
||
# fetch already works around (see _fetch_live_optional_bundle).
|
||
# Everything still flows through quarantine + scan before
|
||
# install, and the scanner sees MORE this way, not less.
|
||
branch, entries = tree
|
||
prefix = f"{skill_path.rstrip('/')}/"
|
||
symlinked: set = set()
|
||
for item in entries:
|
||
item_path = item.get("path", "")
|
||
if not item_path.startswith(prefix):
|
||
continue
|
||
rel_path = item_path[len(prefix):]
|
||
if item.get("type") != "blob" or item.get("mode") == "120000":
|
||
symlinked.add(rel_path)
|
||
continue
|
||
if rel_path == "SKILL.md":
|
||
continue
|
||
base = rel_path.rsplit("/", 1)[-1]
|
||
if base.startswith(".") or base.endswith(".pyc") or "__pycache__" in rel_path.split("/"):
|
||
continue
|
||
try:
|
||
rel_path = _validate_bundle_rel_path(rel_path)
|
||
except ValueError:
|
||
logger.warning("Rejected unsafe file path in skill bundle: %s", item_path)
|
||
return None
|
||
content = self._fetch_file_bytes(repo, item_path, ref=pinned_ref)
|
||
if content is None:
|
||
logger.warning("Failed to fetch referenced skill support "
|
||
"file; continuing without it: %s", item_path)
|
||
continue
|
||
files[rel_path] = content
|
||
# A SKILL.md-linked support path that isn't in the tree is a
|
||
# dangling link — a repo-only dev tool, prose over-match, or a
|
||
# file the author forgot to push. Warn and install without it
|
||
# rather than aborting the whole install (#66760/#90081): the
|
||
# skill body still works, and the gap is visible in the log.
|
||
# A referenced path that IS in the tree but as a symlink (or any
|
||
# non-regular entry) stays a hard rejection — that shape is an
|
||
# escape attempt, not a forgotten file.
|
||
for rel_path in sorted(referenced):
|
||
if rel_path in symlinked:
|
||
logger.warning(
|
||
"Rejected non-regular referenced file in skill "
|
||
"bundle: %s%s", prefix, rel_path,
|
||
)
|
||
return None
|
||
if rel_path not in files:
|
||
logger.warning(
|
||
"Referenced skill support file is missing; "
|
||
"continuing without it: %s%s",
|
||
prefix, rel_path,
|
||
)
|
||
revision = self._tree_revisions.get(repo) or branch
|
||
else:
|
||
for rel_path in referenced:
|
||
content = self._fetch_file_bytes(repo, f"{skill_path.rstrip('/')}/{rel_path}")
|
||
if content is None:
|
||
logger.warning("Failed to fetch referenced skill support "
|
||
"file; continuing without it: %s", rel_path)
|
||
continue
|
||
files[rel_path] = content
|
||
revision = ""
|
||
|
||
skill_name = skill_path.rstrip("/").split("/")[-1]
|
||
trust = self.trust_level_for(identifier)
|
||
|
||
return SkillBundle(
|
||
name=skill_name,
|
||
files=files,
|
||
source="github",
|
||
identifier=identifier,
|
||
trust_level=trust,
|
||
metadata={
|
||
"source_url": (
|
||
f"https://github.com/{repo}/tree/{revision}/{skill_path}"
|
||
if revision else f"https://github.com/{repo}/{skill_path}"
|
||
),
|
||
"source_revision": revision,
|
||
},
|
||
)
|
||
|
||
def inspect(self, identifier: str) -> Optional[SkillMeta]:
|
||
"""Fetch just the SKILL.md metadata for preview."""
|
||
parts = identifier.split("/", 2)
|
||
if len(parts) < 3:
|
||
return None
|
||
|
||
repo = f"{parts[0]}/{parts[1]}"
|
||
skill_path = parts[2].rstrip("/")
|
||
skill_md_path = f"{skill_path}/SKILL.md"
|
||
|
||
content = self._fetch_file_content(repo, skill_md_path)
|
||
if not content:
|
||
return None
|
||
|
||
fm = self._parse_frontmatter_quick(content)
|
||
skill_name = fm.get("name", skill_path.split("/")[-1])
|
||
description = fm.get("description", "")
|
||
|
||
tags = []
|
||
metadata = fm.get("metadata", {})
|
||
if isinstance(metadata, dict):
|
||
hermes_meta = metadata.get("hermes", {})
|
||
if isinstance(hermes_meta, dict):
|
||
tags = hermes_meta.get("tags", [])
|
||
if not tags:
|
||
raw_tags = fm.get("tags", [])
|
||
tags = raw_tags if isinstance(raw_tags, list) else []
|
||
|
||
provider = github_provider_for(repo)
|
||
extra: Dict[str, Any] = {}
|
||
if provider:
|
||
extra["provider"] = provider
|
||
|
||
return SkillMeta(
|
||
name=skill_name,
|
||
description=str(description),
|
||
source="github",
|
||
identifier=identifier,
|
||
trust_level=self.trust_level_for(identifier),
|
||
repo=repo,
|
||
path=skill_path,
|
||
tags=[str(t) for t in tags],
|
||
extra=extra,
|
||
)
|
||
|
||
# -- Internal helpers --
|
||
|
||
def _list_skills_in_repo(self, repo: str, path: str) -> List[SkillMeta]:
|
||
"""List skill directories in a GitHub repo path, using cached index."""
|
||
cache_key = f"{repo}_{path}".replace("/", "_").replace(" ", "_")
|
||
cached = self._read_cache(cache_key)
|
||
if cached is not None:
|
||
return [SkillMeta(**s) for s in cached]
|
||
|
||
url = f"https://api.github.com/repos/{repo}/contents/{path.rstrip('/')}"
|
||
resp = self._github_get(url)
|
||
if resp is None or resp.status_code != 200:
|
||
return []
|
||
|
||
entries = resp.json()
|
||
if not isinstance(entries, list):
|
||
return []
|
||
|
||
skills: List[SkillMeta] = []
|
||
groupings = self._get_skillsh_groupings(repo)
|
||
for entry in entries:
|
||
if entry.get("type") != "dir":
|
||
continue
|
||
|
||
dir_name = entry["name"]
|
||
if dir_name.startswith((".", "_")):
|
||
continue
|
||
|
||
prefix = path.rstrip("/")
|
||
skill_identifier = f"{repo}/{prefix}/{dir_name}" if prefix else f"{repo}/{dir_name}"
|
||
meta = self.inspect(skill_identifier)
|
||
if meta:
|
||
if groupings:
|
||
category = groupings.get(meta.name) or groupings.get(dir_name)
|
||
if category:
|
||
meta.extra["category"] = category
|
||
skills.append(meta)
|
||
|
||
# Cache the results
|
||
self._write_cache(cache_key, [self._meta_to_dict(s) for s in skills])
|
||
return skills
|
||
|
||
# -- Repo tree cache (avoids redundant API calls) --
|
||
|
||
def _get_repo_tree(self, repo: str) -> Optional[Tuple[str, List[dict]]]:
|
||
"""Get cached or fresh repo tree.
|
||
|
||
Returns ``(default_branch, tree_entries)`` or ``None``.
|
||
A single install can call ``_download_directory_via_tree`` and
|
||
``_find_skill_in_repo_tree`` multiple times for the same repo — this
|
||
cache eliminates the redundant ``GET /repos/{repo}`` +
|
||
``GET /repos/{repo}/git/trees/{branch}`` round-trips (previously up to
|
||
6 duplicated pairs per install, consuming ~12 of the 60/hr
|
||
unauthenticated rate limit for nothing).
|
||
"""
|
||
if repo in self._tree_cache:
|
||
return self._tree_cache[repo]
|
||
|
||
headers = self.auth.get_headers()
|
||
|
||
# Resolve default branch
|
||
try:
|
||
resp = httpx.get(
|
||
f"https://api.github.com/repos/{repo}",
|
||
headers=headers, timeout=15, follow_redirects=True,
|
||
)
|
||
if resp.status_code != 200:
|
||
self._check_rate_limit_response(resp)
|
||
return None
|
||
default_branch = resp.json().get("default_branch", "main")
|
||
except (httpx.HTTPError, ValueError):
|
||
return None
|
||
|
||
# Fetch recursive tree
|
||
try:
|
||
resp = httpx.get(
|
||
f"https://api.github.com/repos/{repo}/git/trees/{default_branch}",
|
||
params={"recursive": "1"},
|
||
headers=headers, timeout=30, follow_redirects=True,
|
||
)
|
||
if resp.status_code != 200:
|
||
self._check_rate_limit_response(resp)
|
||
return None
|
||
tree_data = resp.json()
|
||
if tree_data.get("truncated"):
|
||
logger.debug("Git tree truncated for %s, cannot cache", repo)
|
||
return None
|
||
except (httpx.HTTPError, ValueError):
|
||
return None
|
||
|
||
entries = tree_data.get("tree", [])
|
||
revision = tree_data.get("sha")
|
||
if isinstance(revision, str) and revision:
|
||
self._tree_revisions[repo] = revision
|
||
self._tree_cache[repo] = (default_branch, entries)
|
||
return (default_branch, entries)
|
||
|
||
def _check_rate_limit_response(self, resp: "httpx.Response") -> None:
|
||
"""Flag the instance as rate-limited when GitHub returns 403 + exhausted quota."""
|
||
if resp.status_code in (403, 429):
|
||
remaining = resp.headers.get("X-RateLimit-Remaining", "")
|
||
if remaining == "0" or resp.status_code == 429:
|
||
self._rate_limited = True
|
||
logger.warning(
|
||
"GitHub API rate limit exhausted (unauthenticated: 60 req/hr). "
|
||
"Set GITHUB_TOKEN or install the gh CLI to raise the limit to 5,000/hr."
|
||
)
|
||
|
||
def _github_get(
|
||
self,
|
||
url: str,
|
||
*,
|
||
params: Optional[Dict] = None,
|
||
headers: Optional[Dict] = None,
|
||
timeout: float = 15.0,
|
||
max_retries: int = 3,
|
||
) -> Optional["httpx.Response"]:
|
||
"""GET against the GitHub API with retry/backoff on transient failures.
|
||
|
||
Returns the final ``httpx.Response`` (caller inspects status) or
|
||
``None`` when every attempt raised a transport error.
|
||
|
||
Retries on:
|
||
- 403/429 with ``X-RateLimit-Remaining: 0`` — waits until the
|
||
reset time (capped) when the header is present, else exponential
|
||
backoff. This is the all-GitHub-tap-collapse case: a single
|
||
shared rate limit zeroes github + well-known
|
||
at once during the index build.
|
||
- 5xx and connection/timeout errors — exponential backoff.
|
||
|
||
On terminal rate-limit exhaustion the instance is flagged via
|
||
``_check_rate_limit_response`` so the build can fail loud instead of
|
||
silently shipping an index with the GitHub sources dropped to zero.
|
||
"""
|
||
hdrs = headers if headers is not None else self.auth.get_headers()
|
||
backoff = 1.0
|
||
last_resp: Optional["httpx.Response"] = None
|
||
for attempt in range(max_retries):
|
||
try:
|
||
resp = httpx.get(
|
||
url, params=params, headers=hdrs,
|
||
timeout=timeout, follow_redirects=True,
|
||
)
|
||
except httpx.HTTPError as e:
|
||
logger.debug("GitHub GET %s failed (attempt %d/%d): %s",
|
||
url, attempt + 1, max_retries, e)
|
||
if attempt < max_retries - 1:
|
||
time.sleep(backoff)
|
||
backoff = min(backoff * 2, 30.0)
|
||
continue
|
||
return None
|
||
|
||
last_resp = resp
|
||
if resp.status_code == 200:
|
||
return resp
|
||
|
||
# Rate-limited: honor the reset header when present, else back off.
|
||
if resp.status_code in (403, 429):
|
||
remaining = resp.headers.get("X-RateLimit-Remaining", "")
|
||
is_rl = remaining == "0" or resp.status_code == 429
|
||
if is_rl and attempt < max_retries - 1:
|
||
wait = backoff
|
||
reset = resp.headers.get("X-RateLimit-Reset", "")
|
||
retry_after = resp.headers.get("Retry-After", "")
|
||
if retry_after.isdigit():
|
||
wait = min(float(retry_after), 60.0)
|
||
elif reset.isdigit():
|
||
delta = float(reset) - time.time()
|
||
if 0 < delta <= 60.0:
|
||
wait = delta
|
||
logger.debug(
|
||
"GitHub rate limited on %s, waiting %.1fs (attempt %d/%d)",
|
||
url, wait, attempt + 1, max_retries,
|
||
)
|
||
time.sleep(wait)
|
||
backoff = min(backoff * 2, 30.0)
|
||
continue
|
||
# Out of retries (or not a rate-limit 403) — flag and return.
|
||
self._check_rate_limit_response(resp)
|
||
return resp
|
||
|
||
# 5xx — retry; 4xx (other than rate limit) — return immediately.
|
||
if 500 <= resp.status_code < 600 and attempt < max_retries - 1:
|
||
time.sleep(backoff)
|
||
backoff = min(backoff * 2, 30.0)
|
||
continue
|
||
return resp
|
||
|
||
return last_resp
|
||
|
||
|
||
def _download_directory(self, repo: str, path: str) -> Dict[str, str]:
|
||
"""Recursively download all text files from a GitHub directory.
|
||
|
||
Uses the Git Trees API first (single call for the entire tree) to
|
||
avoid per-directory rate limiting that causes silent subdirectory
|
||
loss. Falls back to the recursive Contents API when the tree
|
||
endpoint is unavailable or the response is truncated.
|
||
"""
|
||
files = self._download_directory_via_tree(repo, path)
|
||
if files is not None:
|
||
return files
|
||
logger.debug("Tree API unavailable for %s/%s, falling back to Contents API", repo, path)
|
||
return self._download_directory_recursive(repo, path)
|
||
|
||
def _download_directory_via_tree(self, repo: str, path: str) -> Optional[Dict[str, str]]:
|
||
"""Download an entire directory using the Git Trees API (single request).
|
||
|
||
Returns:
|
||
dict of files if the path exists and has content,
|
||
empty dict ``{}`` if the tree is cached but the path doesn't exist
|
||
(prevents unnecessary Contents API fallback),
|
||
``None`` if the tree couldn't be fetched (triggers Contents API fallback).
|
||
"""
|
||
path = path.rstrip("/")
|
||
|
||
cached = self._get_repo_tree(repo)
|
||
if cached is None:
|
||
return None
|
||
_default_branch, tree_entries = cached
|
||
|
||
# Check if ANY entry lives under the target path
|
||
prefix = f"{path}/"
|
||
has_entries = any(
|
||
item.get("path", "").startswith(prefix) for item in tree_entries
|
||
)
|
||
if not has_entries:
|
||
# Path definitively doesn't exist in the repo — return empty
|
||
# instead of None to skip the Contents API fallback.
|
||
return {}
|
||
|
||
# Filter to blobs under our target path and fetch content
|
||
files: Dict[str, str] = {}
|
||
for item in tree_entries:
|
||
if item.get("type") != "blob":
|
||
continue
|
||
item_path = item.get("path", "")
|
||
if not item_path.startswith(prefix):
|
||
continue
|
||
rel_path = item_path[len(prefix):]
|
||
content = self._fetch_file_content(repo, item_path)
|
||
if content is not None:
|
||
files[rel_path] = content
|
||
else:
|
||
logger.debug("Skipped file (fetch failed): %s/%s", repo, item_path)
|
||
|
||
return files if files else None
|
||
|
||
def _download_directory_recursive(self, repo: str, path: str) -> Dict[str, str]:
|
||
"""Recursively download via Contents API (fallback)."""
|
||
url = f"https://api.github.com/repos/{repo}/contents/{path.rstrip('/')}"
|
||
# Route through _github_get so directory listing gets the same
|
||
# 429/403-rate-limit retry + backoff as file fetches (#3033).
|
||
resp = self._github_get(url)
|
||
if resp is None:
|
||
return {}
|
||
if resp.status_code != 200:
|
||
logger.debug("Contents API returned %d for %s/%s", resp.status_code, repo, path)
|
||
return {}
|
||
|
||
entries = resp.json()
|
||
if not isinstance(entries, list):
|
||
return {}
|
||
|
||
files: Dict[str, str] = {}
|
||
for entry in entries:
|
||
name = entry.get("name", "")
|
||
entry_type = entry.get("type", "")
|
||
|
||
if entry_type == "file":
|
||
content = self._fetch_file_content(repo, entry.get("path", ""))
|
||
if content is not None:
|
||
rel_path = name
|
||
files[rel_path] = content
|
||
elif entry_type == "dir":
|
||
sub_files = self._download_directory_recursive(repo, entry.get("path", ""))
|
||
if not sub_files:
|
||
logger.debug("Empty or failed subdirectory: %s/%s", repo, entry.get("path", ""))
|
||
for sub_name, sub_content in sub_files.items():
|
||
files[f"{name}/{sub_name}"] = sub_content
|
||
|
||
return files
|
||
|
||
def _find_skill_in_repo_tree(self, repo: str, skill_name: str) -> Optional[str]:
|
||
"""Use the GitHub Trees API to find a skill directory anywhere in the repo.
|
||
|
||
Returns the full identifier (``repo/path/to/skill``) or ``None``.
|
||
This is a single API call regardless of repo depth, so it efficiently
|
||
handles deeply nested directory structures like
|
||
``cli-tool/components/skills/development/<skill>/SKILL.md``.
|
||
"""
|
||
cached = self._get_repo_tree(repo)
|
||
if cached is None:
|
||
return None
|
||
_default_branch, tree_entries = cached
|
||
|
||
# Look for SKILL.md files inside directories named <skill_name>
|
||
skill_md_suffix = f"/{skill_name}/SKILL.md"
|
||
for entry in tree_entries:
|
||
if entry.get("type") != "blob":
|
||
continue
|
||
path = entry.get("path", "")
|
||
if path.endswith(skill_md_suffix) or path == f"{skill_name}/SKILL.md":
|
||
# Strip /SKILL.md to get the skill directory path
|
||
skill_dir = path[: -len("/SKILL.md")]
|
||
return f"{repo}/{skill_dir}"
|
||
|
||
return None
|
||
|
||
def _fetch_file_content(
|
||
self, repo: str, path: str, ref: Optional[str] = None
|
||
) -> Optional[str]:
|
||
"""Fetch a single text file from GitHub."""
|
||
content = self._fetch_file_bytes(repo, path, ref=ref)
|
||
if content is None:
|
||
return None
|
||
try:
|
||
return content.decode("utf-8")
|
||
except UnicodeDecodeError:
|
||
return None
|
||
|
||
def _fetch_file_bytes(
|
||
self, repo: str, path: str, ref: Optional[str] = None
|
||
) -> Optional[bytes]:
|
||
"""Fetch exact file bytes from GitHub without text decoding.
|
||
|
||
``ref`` pins the fetch to a specific commit/tree SHA. Without it the
|
||
contents endpoint floats to the default-branch HEAD, so the bytes can
|
||
come from a NEWER revision than the tree the paths were validated
|
||
against — a TOCTOU between "what the tree says is a regular blob"
|
||
and "what actually gets downloaded". Callers that resolved paths from
|
||
a tree pass that tree's SHA; ``None`` keeps the legacy unpinned
|
||
behavior for call sites with no revision in hand.
|
||
"""
|
||
encoded_path = quote(path, safe="/")
|
||
url = f"https://api.github.com/repos/{repo}/contents/{encoded_path}"
|
||
params = {"ref": ref} if ref else None
|
||
resp = self._github_get(
|
||
url,
|
||
params=params,
|
||
headers={**self.auth.get_headers(), "Accept": "application/vnd.github.v3.raw"},
|
||
)
|
||
if resp is not None and resp.status_code == 200:
|
||
return resp.content
|
||
return None
|
||
|
||
def _get_skillsh_groupings(self, repo: str) -> Optional[Dict[str, str]]:
|
||
"""Fetch and parse the repo-root ``skills.sh.json`` grouping sidecar.
|
||
|
||
``skills.sh.json`` is a published cross-ecosystem standard
|
||
(``$schema: https://skills.sh/schemas/skills.sh.schema.json``) that
|
||
lets a tap declare human-readable category groupings for its skills:
|
||
|
||
{"groupings": [{"title": "Inference AI", "skills": ["dynamo-..."]}]}
|
||
|
||
We flatten it into ``{skill_name: grouping_title}`` so the Skills Hub
|
||
UI can show a real category pill instead of a tag-derived guess. Any
|
||
tap that ships this file gets categorization for free — this is not
|
||
NVIDIA-specific.
|
||
|
||
Returns the map (possibly empty) on success, or ``None`` when the repo
|
||
has no sidecar / it couldn't be parsed. Cached per-repo on the instance.
|
||
"""
|
||
if repo in self._skillsh_groupings:
|
||
return self._skillsh_groupings[repo]
|
||
|
||
content = self._fetch_file_content(repo, "skills.sh.json")
|
||
groupings = self._parse_skillsh_groupings(content) if content else None
|
||
self._skillsh_groupings[repo] = groupings
|
||
return groupings
|
||
|
||
@staticmethod
|
||
def _parse_skillsh_groupings(content: str) -> Optional[Dict[str, str]]:
|
||
"""Flatten a ``skills.sh.json`` document into ``{skill_name: title}``.
|
||
|
||
Returns ``None`` when the content isn't a usable grouping document.
|
||
"""
|
||
try:
|
||
data = json.loads(content)
|
||
except (json.JSONDecodeError, TypeError):
|
||
return None
|
||
if not isinstance(data, dict):
|
||
return None
|
||
groupings = data.get("groupings")
|
||
if not isinstance(groupings, list):
|
||
return None
|
||
|
||
mapping: Dict[str, str] = {}
|
||
for group in groupings:
|
||
if not isinstance(group, dict):
|
||
continue
|
||
title = group.get("title")
|
||
members = group.get("skills")
|
||
if not isinstance(title, str) or not isinstance(members, list):
|
||
continue
|
||
for member in members:
|
||
if isinstance(member, str) and member:
|
||
# First grouping wins if a skill is listed twice.
|
||
mapping.setdefault(member, title)
|
||
return mapping
|
||
|
||
def _read_cache(self, key: str) -> Optional[list]:
|
||
"""Read cached index if not expired."""
|
||
cache_file = _index_cache_dir() / f"{key}.json"
|
||
if not cache_file.exists():
|
||
return None
|
||
try:
|
||
stat = cache_file.stat()
|
||
if time.time() - stat.st_mtime > INDEX_CACHE_TTL:
|
||
return None
|
||
return json.loads(cache_file.read_text(encoding="utf-8"))
|
||
except (OSError, json.JSONDecodeError):
|
||
return None
|
||
|
||
def _write_cache(self, key: str, data: list) -> None:
|
||
"""Write index data to cache."""
|
||
index_cache_dir = _index_cache_dir()
|
||
index_cache_dir.mkdir(parents=True, exist_ok=True)
|
||
cache_file = index_cache_dir / f"{key}.json"
|
||
try:
|
||
cache_file.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
|
||
except OSError as e:
|
||
logger.debug("Could not write cache: %s", e)
|
||
|
||
@staticmethod
|
||
def _meta_to_dict(meta: SkillMeta) -> dict:
|
||
return {
|
||
"name": meta.name,
|
||
"description": meta.description,
|
||
"source": meta.source,
|
||
"identifier": meta.identifier,
|
||
"trust_level": meta.trust_level,
|
||
"repo": meta.repo,
|
||
"path": meta.path,
|
||
"tags": meta.tags,
|
||
"extra": meta.extra,
|
||
}
|
||
|
||
@staticmethod
|
||
def _parse_frontmatter_quick(content: str) -> dict:
|
||
"""Parse YAML frontmatter from SKILL.md content."""
|
||
content = content.lstrip("\ufeff") # tolerate UTF-8 BOM (Windows editors)
|
||
if not content.startswith("---"):
|
||
return {}
|
||
match = re.search(r'\n---\s*\n', content[3:])
|
||
if not match:
|
||
return {}
|
||
yaml_text = content[3:match.start() + 3]
|
||
try:
|
||
parsed = yaml.safe_load(yaml_text)
|
||
return parsed if isinstance(parsed, dict) else {}
|
||
except yaml.YAMLError:
|
||
return {}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Well-known Agent Skills endpoint source adapter
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class WellKnownSkillSource(SkillSource):
|
||
"""Read skills from a domain exposing /.well-known/skills/index.json."""
|
||
|
||
BASE_PATH = "/.well-known/skills"
|
||
|
||
def source_id(self) -> str:
|
||
return "well-known"
|
||
|
||
def trust_level_for(self, identifier: str) -> str:
|
||
return "community"
|
||
|
||
def search(self, query: str, limit: int = 10) -> List[SkillMeta]:
|
||
index_url = self._query_to_index_url(query)
|
||
if not index_url:
|
||
return []
|
||
|
||
parsed = self._parse_index(index_url)
|
||
if not parsed:
|
||
return []
|
||
|
||
results: List[SkillMeta] = []
|
||
for entry in parsed["skills"][:limit]:
|
||
name = entry.get("name")
|
||
if not isinstance(name, str) or not name:
|
||
continue
|
||
description = entry.get("description", "")
|
||
files = entry.get("files", ["SKILL.md"])
|
||
results.append(SkillMeta(
|
||
name=name,
|
||
description=str(description),
|
||
source="well-known",
|
||
identifier=self._wrap_identifier(parsed["base_url"], name),
|
||
trust_level="community",
|
||
path=name,
|
||
extra={
|
||
"index_url": parsed["index_url"],
|
||
"base_url": parsed["base_url"],
|
||
"files": files if isinstance(files, list) else ["SKILL.md"],
|
||
},
|
||
))
|
||
return results
|
||
|
||
def inspect(self, identifier: str) -> Optional[SkillMeta]:
|
||
parsed = self._parse_identifier(identifier)
|
||
if not parsed:
|
||
return None
|
||
|
||
entry = self._index_entry(parsed["index_url"], parsed["skill_name"])
|
||
if not entry:
|
||
return None
|
||
|
||
skill_md = self._fetch_text(f"{parsed['skill_url']}/SKILL.md")
|
||
if skill_md is None:
|
||
return None
|
||
|
||
fm = GitHubSource._parse_frontmatter_quick(skill_md)
|
||
description = str(fm.get("description") or entry.get("description") or "")
|
||
name = str(fm.get("name") or parsed["skill_name"])
|
||
return SkillMeta(
|
||
name=name,
|
||
description=description,
|
||
source="well-known",
|
||
identifier=self._wrap_identifier(parsed["base_url"], parsed["skill_name"]),
|
||
trust_level="community",
|
||
path=parsed["skill_name"],
|
||
extra={
|
||
"index_url": parsed["index_url"],
|
||
"base_url": parsed["base_url"],
|
||
"files": entry.get("files", ["SKILL.md"]),
|
||
"endpoint": parsed["skill_url"],
|
||
},
|
||
)
|
||
|
||
def fetch(self, identifier: str) -> Optional[SkillBundle]:
|
||
parsed = self._parse_identifier(identifier)
|
||
if not parsed:
|
||
return None
|
||
|
||
try:
|
||
skill_name = _validate_skill_name(parsed["skill_name"])
|
||
except ValueError:
|
||
logger.warning("Well-known skill identifier contained unsafe skill name: %s", identifier)
|
||
return None
|
||
|
||
entry = self._index_entry(parsed["index_url"], parsed["skill_name"])
|
||
if not entry:
|
||
return None
|
||
|
||
files = entry.get("files", ["SKILL.md"])
|
||
if not isinstance(files, list) or not files:
|
||
files = ["SKILL.md"]
|
||
|
||
downloaded: Dict[str, str] = {}
|
||
for rel_path in files:
|
||
if not isinstance(rel_path, str) or not rel_path:
|
||
continue
|
||
try:
|
||
safe_rel_path = _validate_bundle_rel_path(rel_path)
|
||
except ValueError:
|
||
logger.warning(
|
||
"Well-known skill %s advertised unsafe file path: %r",
|
||
identifier,
|
||
rel_path,
|
||
)
|
||
return None
|
||
text = self._fetch_text(f"{parsed['skill_url']}/{safe_rel_path}")
|
||
if text is None:
|
||
return None
|
||
downloaded[safe_rel_path] = text
|
||
|
||
if "SKILL.md" not in downloaded:
|
||
return None
|
||
|
||
return SkillBundle(
|
||
name=skill_name,
|
||
files=downloaded,
|
||
source="well-known",
|
||
identifier=self._wrap_identifier(parsed["base_url"], skill_name),
|
||
trust_level="community",
|
||
metadata={
|
||
"index_url": parsed["index_url"],
|
||
"base_url": parsed["base_url"],
|
||
"endpoint": parsed["skill_url"],
|
||
"files": files,
|
||
},
|
||
)
|
||
|
||
def _query_to_index_url(self, query: str) -> Optional[str]:
|
||
query = query.strip()
|
||
if not query.startswith(("http://", "https://")):
|
||
return None
|
||
if query.endswith("/index.json"):
|
||
return query
|
||
if f"{self.BASE_PATH}/" in query:
|
||
base_url = query.split(f"{self.BASE_PATH}/", 1)[0] + self.BASE_PATH
|
||
return f"{base_url}/index.json"
|
||
return query.rstrip("/") + f"{self.BASE_PATH}/index.json"
|
||
|
||
def _parse_identifier(self, identifier: str) -> Optional[dict]:
|
||
raw = identifier[len("well-known:"):] if identifier.startswith("well-known:") else identifier
|
||
if not raw.startswith(("http://", "https://")):
|
||
return None
|
||
|
||
parsed_url = urlparse(raw)
|
||
clean_url = urlunparse(parsed_url._replace(fragment=""))
|
||
fragment = parsed_url.fragment
|
||
|
||
if clean_url.endswith("/index.json"):
|
||
if not fragment:
|
||
return None
|
||
base_url = clean_url[:-len("/index.json")]
|
||
skill_name = fragment
|
||
skill_url = f"{base_url}/{skill_name}"
|
||
return {
|
||
"index_url": clean_url,
|
||
"base_url": base_url,
|
||
"skill_name": skill_name,
|
||
"skill_url": skill_url,
|
||
}
|
||
|
||
if clean_url.endswith("/SKILL.md"):
|
||
skill_url = clean_url[:-len("/SKILL.md")]
|
||
else:
|
||
skill_url = clean_url.rstrip("/")
|
||
|
||
if f"{self.BASE_PATH}/" not in skill_url:
|
||
return None
|
||
|
||
base_url, skill_name = skill_url.rsplit("/", 1)
|
||
return {
|
||
"index_url": f"{base_url}/index.json",
|
||
"base_url": base_url,
|
||
"skill_name": skill_name,
|
||
"skill_url": skill_url,
|
||
}
|
||
|
||
def _parse_index(self, index_url: str) -> Optional[dict]:
|
||
cache_key = f"well_known_index_{hashlib.md5(index_url.encode()).hexdigest()}"
|
||
cached = _read_index_cache(cache_key)
|
||
if isinstance(cached, dict) and isinstance(cached.get("skills"), list):
|
||
return cached
|
||
|
||
resp = _guarded_http_get(index_url, timeout=20)
|
||
if resp is None or resp.status_code != 200:
|
||
return None
|
||
try:
|
||
data = resp.json()
|
||
except json.JSONDecodeError:
|
||
return None
|
||
|
||
skills = data.get("skills", []) if isinstance(data, dict) else []
|
||
if not isinstance(skills, list):
|
||
return None
|
||
|
||
parsed = {
|
||
"index_url": index_url,
|
||
"base_url": index_url[:-len("/index.json")],
|
||
"skills": skills,
|
||
}
|
||
_write_index_cache(cache_key, parsed)
|
||
return parsed
|
||
|
||
def _index_entry(self, index_url: str, skill_name: str) -> Optional[dict]:
|
||
parsed = self._parse_index(index_url)
|
||
if not parsed:
|
||
return None
|
||
for entry in parsed["skills"]:
|
||
if isinstance(entry, dict) and entry.get("name") == skill_name:
|
||
return entry
|
||
return None
|
||
|
||
@staticmethod
|
||
def _fetch_text(url: str) -> Optional[str]:
|
||
resp = _guarded_http_get(url, timeout=20)
|
||
if resp is not None and resp.status_code == 200:
|
||
return resp.text
|
||
return None
|
||
|
||
@staticmethod
|
||
def _wrap_identifier(base_url: str, skill_name: str) -> str:
|
||
return f"well-known:{base_url.rstrip('/')}/{skill_name}"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Direct URL source adapter
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class UrlSource(SkillSource):
|
||
"""Fetch SKILL.md plus explicitly referenced, allowlisted support files.
|
||
|
||
The identifier IS the URL (e.g. ``https://example.com/path/SKILL.md``).
|
||
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
|
||
``community`` and the same security scan runs as for every other source.
|
||
"""
|
||
|
||
def source_id(self) -> str:
|
||
return "url"
|
||
|
||
def trust_level_for(self, identifier: str) -> str:
|
||
return "community"
|
||
|
||
# Search is meaningless for a direct URL — skip (return empty).
|
||
def search(self, query: str, limit: int = 10) -> List[SkillMeta]:
|
||
return []
|
||
|
||
def _matches(self, identifier: str) -> bool:
|
||
"""Return True iff this source should handle ``identifier``.
|
||
|
||
We claim bare HTTP(S) URLs that end in ``.md`` (typically
|
||
``.../SKILL.md``). Wrapped identifiers (``github:``,
|
||
``well-known:``, etc.) and ``/.well-known/skills/`` URLs are
|
||
left for their respective adapters.
|
||
"""
|
||
if not isinstance(identifier, str):
|
||
return False
|
||
ident = identifier.strip()
|
||
if not ident.lower().startswith(("http://", "https://")):
|
||
return False
|
||
# Don't steal well-known URLs.
|
||
if "/.well-known/skills/" in ident or ident.rstrip("/").endswith("/index.json"):
|
||
return False
|
||
# Only claim URLs that look like a markdown file.
|
||
try:
|
||
path = urlparse(ident).path
|
||
except ValueError:
|
||
return False
|
||
return path.lower().endswith(".md")
|
||
|
||
def inspect(self, identifier: str) -> Optional[SkillMeta]:
|
||
if not self._matches(identifier):
|
||
return None
|
||
url = identifier.strip()
|
||
text = self._fetch_text(url)
|
||
if text is None:
|
||
return None
|
||
fm = GitHubSource._parse_frontmatter_quick(text)
|
||
name = self._resolve_skill_name(fm, url)
|
||
description = str(fm.get("description") or "")
|
||
tags: List[str] = []
|
||
metadata = fm.get("metadata", {})
|
||
if isinstance(metadata, dict):
|
||
hermes_meta = metadata.get("hermes", {})
|
||
if isinstance(hermes_meta, dict):
|
||
raw_tags = hermes_meta.get("tags", [])
|
||
if isinstance(raw_tags, list):
|
||
tags = [str(t) for t in raw_tags]
|
||
return SkillMeta(
|
||
name=name or "",
|
||
description=description,
|
||
source="url",
|
||
identifier=url,
|
||
trust_level="community",
|
||
path=name or "",
|
||
tags=tags,
|
||
extra={"url": url, "awaiting_name": name is None},
|
||
)
|
||
|
||
def fetch(self, identifier: str) -> Optional[SkillBundle]:
|
||
if not self._matches(identifier):
|
||
return None
|
||
url = identifier.strip()
|
||
text = self._fetch_text(url)
|
||
if text is None:
|
||
return None
|
||
|
||
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, quote(rel_path, safe="/"))
|
||
if urlparse(support_url).netloc != urlparse(url).netloc:
|
||
return None
|
||
content = self._fetch_bytes(support_url)
|
||
if content is None:
|
||
# A referenced support file that 404s (or is otherwise
|
||
# unreachable) shouldn't sink the whole install — skip it
|
||
# and let the bundle install without it.
|
||
logger.warning(
|
||
"URL skill %s: referenced support file %r could not be "
|
||
"fetched from %s; skipping it",
|
||
url, rel_path, support_url,
|
||
)
|
||
continue
|
||
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``)
|
||
# either prompts the user on a TTY or refuses with an actionable error
|
||
# on non-interactive surfaces. Keep the expensive HTTP fetch's result
|
||
# so the caller doesn't have to re-download after picking a name.
|
||
skill_name = ""
|
||
if name is not None:
|
||
try:
|
||
skill_name = _validate_skill_name(name)
|
||
except ValueError:
|
||
logger.warning("URL skill %s produced unsafe skill name: %r", url, name)
|
||
return None
|
||
|
||
return SkillBundle(
|
||
name=skill_name,
|
||
files=files,
|
||
source="url",
|
||
identifier=url,
|
||
trust_level="community",
|
||
metadata={"url": url, "source_url": url, "awaiting_name": not skill_name},
|
||
)
|
||
|
||
@staticmethod
|
||
def _fetch_text(url: str) -> Optional[str]:
|
||
resp = _guarded_http_get(url, timeout=20)
|
||
if resp is not None and resp.status_code == 200:
|
||
return resp.text
|
||
return None
|
||
|
||
@staticmethod
|
||
def _fetch_bytes(url: str) -> Optional[bytes]:
|
||
resp = _guarded_http_get(url, timeout=20)
|
||
if resp is not None and resp.status_code == 200:
|
||
return resp.content
|
||
return None
|
||
|
||
# Skill names must look like identifiers: lowercase letters/digits with
|
||
# optional hyphens/underscores. Blocks dangerous (``../evil``) AND useless
|
||
# (``SKILL``, ``README``, empty) candidates before they hit the disk.
|
||
_VALID_NAME_RE = re.compile(r"^[a-z][a-z0-9_-]*$")
|
||
|
||
@classmethod
|
||
def _is_valid_skill_name(cls, name: Optional[str]) -> bool:
|
||
if not isinstance(name, str):
|
||
return False
|
||
candidate = name.strip().lower()
|
||
if not candidate or candidate in {"skill", "readme", "index", "unnamed-skill"}:
|
||
return False
|
||
return bool(cls._VALID_NAME_RE.match(candidate))
|
||
|
||
@classmethod
|
||
def _resolve_skill_name(cls, fm: dict, url: str) -> Optional[str]:
|
||
"""Pick a skill name from frontmatter or URL.
|
||
|
||
Returns ``None`` when neither source produces a valid identifier;
|
||
callers (CLI ``do_install``) then prompt the user or refuse. Preferring
|
||
a clean failure over a useless auto-name like ``SKILL`` or ``unnamed-skill``.
|
||
"""
|
||
# 1. Frontmatter ``name:`` is authoritative when present and valid.
|
||
fm_name = fm.get("name") if isinstance(fm, dict) else None
|
||
if isinstance(fm_name, str) and cls._is_valid_skill_name(fm_name):
|
||
return fm_name.strip()
|
||
|
||
# 2. URL-slug heuristic: ``.../<name>/SKILL.md`` → ``<name>``;
|
||
# ``.../<name>.md`` → ``<name>``. Validate each candidate.
|
||
try:
|
||
path = urlparse(url).path
|
||
except ValueError:
|
||
return None
|
||
parts = [p for p in path.split("/") if p]
|
||
if parts and parts[-1].lower() == "skill.md" and len(parts) >= 2:
|
||
candidate = parts[-2]
|
||
if cls._is_valid_skill_name(candidate):
|
||
return candidate
|
||
if parts:
|
||
candidate = re.sub(r"\.md$", "", parts[-1], flags=re.IGNORECASE)
|
||
if cls._is_valid_skill_name(candidate):
|
||
return candidate
|
||
|
||
# Nothing usable — let the caller handle it.
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# skills.sh source adapter
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class SkillsShSource(SkillSource):
|
||
"""Discover skills via skills.sh and fetch content from the underlying GitHub repo."""
|
||
|
||
BASE_URL = "https://skills.sh"
|
||
SEARCH_URL = f"{BASE_URL}/api/search"
|
||
# Sitemap index — the real catalog source. The homepage scrape only
|
||
# exposes a curated featured strip (~200 entries); the sitemap covers
|
||
# the full ~20k+ catalog. https://www.skills.sh/sitemap.xml points at
|
||
# sitemap-skills-1.xml + sitemap-skills-2.xml, each up to 10k URLs.
|
||
SITEMAP_INDEX_URL = "https://www.skills.sh/sitemap.xml"
|
||
_SITEMAP_LOC_RE = re.compile(r"<loc>([^<]+)</loc>", re.IGNORECASE)
|
||
_SITEMAP_SKILL_RE = re.compile(
|
||
r"^https?://(?:www\.)?skills\.sh/(?P<owner>[^/]+)/(?P<repo>[^/]+)/(?P<skill>[^/]+)/?$",
|
||
re.IGNORECASE,
|
||
)
|
||
_SKILL_LINK_RE = re.compile(r'href=["\']/(?P<id>(?!agents/|_next/|api/)[^"\'/]+/[^"\'/]+/[^"\'/]+)["\']')
|
||
_INSTALL_CMD_RE = re.compile(
|
||
r'npx\s+skills\s+add\s+(?P<repo>https?://github\.com/[^\s<]+|[^\s<]+)'
|
||
r'(?:\s+--skill\s+(?P<skill>[^\s<]+))?',
|
||
re.IGNORECASE,
|
||
)
|
||
_PAGE_H1_RE = re.compile(r'<h1[^>]*>(?P<title>.*?)</h1>', re.IGNORECASE | re.DOTALL)
|
||
_PROSE_H1_RE = re.compile(
|
||
r'<div[^>]*class=["\'][^"\']*prose[^"\']*["\'][^>]*>.*?<h1[^>]*>(?P<title>.*?)</h1>',
|
||
re.IGNORECASE | re.DOTALL,
|
||
)
|
||
_PROSE_P_RE = re.compile(
|
||
r'<div[^>]*class=["\'][^"\']*prose[^"\']*["\'][^>]*>.*?<p[^>]*>(?P<body>.*?)</p>',
|
||
re.IGNORECASE | re.DOTALL,
|
||
)
|
||
_WEEKLY_INSTALLS_RE = re.compile(r'Weekly Installs.*?children\\":\\"(?P<count>[0-9.,Kk]+)\\"', re.DOTALL)
|
||
|
||
def __init__(self, auth: GitHubAuth):
|
||
self.auth = auth
|
||
self.github = GitHubSource(auth=auth)
|
||
|
||
def source_id(self) -> str:
|
||
return "skills-sh"
|
||
|
||
def trust_level_for(self, identifier: str) -> str:
|
||
return self.github.trust_level_for(self._normalize_identifier(identifier))
|
||
|
||
def search(self, query: str, limit: int = 10) -> List[SkillMeta]:
|
||
if not query.strip():
|
||
# Empty query = bulk catalog dump (what build_skills_index.py
|
||
# calls with). The homepage scrape only sees ~200 featured
|
||
# entries; the sitemap walks the full ~20k+ catalog.
|
||
return self._sitemap_catalog(limit)
|
||
|
||
cache_key = f"skills_sh_search_{hashlib.md5(f'{query}|{limit}'.encode()).hexdigest()}"
|
||
cached = _read_index_cache(cache_key)
|
||
if cached is not None:
|
||
return [SkillMeta(**item) for item in cached][:limit]
|
||
|
||
try:
|
||
resp = httpx.get(
|
||
self.SEARCH_URL,
|
||
params={"q": query, "limit": limit},
|
||
timeout=20,
|
||
)
|
||
if resp.status_code != 200:
|
||
return []
|
||
data = resp.json()
|
||
except (httpx.HTTPError, json.JSONDecodeError):
|
||
return []
|
||
|
||
items = data.get("skills", []) if isinstance(data, dict) else []
|
||
if not isinstance(items, list):
|
||
return []
|
||
|
||
results: List[SkillMeta] = []
|
||
for item in items[:limit]:
|
||
meta = self._meta_from_search_item(item)
|
||
if meta:
|
||
results.append(meta)
|
||
|
||
_write_index_cache(cache_key, [_skill_meta_to_dict(item) for item in results])
|
||
return results
|
||
|
||
def fetch(self, identifier: str) -> Optional[SkillBundle]:
|
||
canonical = self._normalize_identifier(identifier)
|
||
detail = self._fetch_detail_page(canonical)
|
||
for candidate in self._candidate_identifiers(canonical):
|
||
bundle = self.github.fetch(candidate)
|
||
if bundle:
|
||
bundle.source = "skills.sh"
|
||
bundle.identifier = self._wrap_identifier(canonical)
|
||
bundle.metadata.update(self._detail_to_metadata(canonical, detail))
|
||
return bundle
|
||
|
||
resolved = self._discover_identifier(canonical, detail=detail)
|
||
if resolved:
|
||
bundle = self.github.fetch(resolved)
|
||
if bundle:
|
||
bundle.source = "skills.sh"
|
||
bundle.identifier = self._wrap_identifier(canonical)
|
||
bundle.metadata.update(self._detail_to_metadata(canonical, detail))
|
||
return bundle
|
||
return None
|
||
|
||
def inspect(self, identifier: str) -> Optional[SkillMeta]:
|
||
canonical = self._normalize_identifier(identifier)
|
||
detail = self._fetch_detail_page(canonical)
|
||
meta = self._resolve_github_meta(canonical, detail=detail)
|
||
if meta:
|
||
return self._finalize_inspect_meta(meta, canonical, detail)
|
||
return None
|
||
|
||
def _sitemap_catalog(self, limit: int) -> List[SkillMeta]:
|
||
"""Walk the skills.sh sitemap to enumerate the full catalog.
|
||
|
||
Cached for the standard index TTL so we don't refetch ~2 MB of
|
||
sitemap XML per build. Falls back to ``_featured_skills`` if the
|
||
sitemap is unreachable or empty (network failure, hostname
|
||
change, etc.).
|
||
"""
|
||
cache_key = "skills_sh_sitemap_v1"
|
||
cached = _read_index_cache(cache_key)
|
||
if cached is not None:
|
||
metas = [SkillMeta(**item) for item in cached]
|
||
return metas[:limit] if limit > 0 else metas
|
||
|
||
# skills.sh serves the per-skill sitemaps brotli-compressed, and
|
||
# httpx's optional brotlicffi backend has a streaming-decode bug
|
||
# that fails on these specific payloads. Excluding "br" from
|
||
# Accept-Encoding makes the server fall back to gzip (or
|
||
# identity), which works on every httpx install.
|
||
sitemap_headers = {"Accept-Encoding": "gzip"}
|
||
|
||
# Step 1: fetch the sitemap index → list of skill-sitemap URLs.
|
||
skill_sitemap_urls: List[str] = []
|
||
try:
|
||
resp = httpx.get(
|
||
self.SITEMAP_INDEX_URL,
|
||
timeout=20,
|
||
follow_redirects=True,
|
||
headers=sitemap_headers,
|
||
)
|
||
if resp.status_code != 200:
|
||
return self._featured_skills(limit)
|
||
for match in self._SITEMAP_LOC_RE.finditer(resp.text):
|
||
loc = match.group(1).strip()
|
||
# Sitemap index entries that point at the per-skill maps.
|
||
if "sitemap-skills" in loc:
|
||
skill_sitemap_urls.append(loc)
|
||
except httpx.HTTPError:
|
||
return self._featured_skills(limit)
|
||
|
||
if not skill_sitemap_urls:
|
||
return self._featured_skills(limit)
|
||
|
||
# Step 2: fetch each skill sitemap and collect canonical "owner/repo/skill" IDs.
|
||
seen: set[str] = set()
|
||
results: List[SkillMeta] = []
|
||
for sitemap_url in skill_sitemap_urls:
|
||
try:
|
||
resp = httpx.get(
|
||
sitemap_url,
|
||
timeout=30,
|
||
follow_redirects=True,
|
||
headers=sitemap_headers,
|
||
)
|
||
if resp.status_code != 200:
|
||
continue
|
||
except httpx.HTTPError:
|
||
continue
|
||
for loc_match in self._SITEMAP_LOC_RE.finditer(resp.text):
|
||
url = loc_match.group(1).strip()
|
||
m = self._SITEMAP_SKILL_RE.match(url)
|
||
if not m:
|
||
continue
|
||
owner = m.group("owner")
|
||
repo_name = m.group("repo")
|
||
skill_name = m.group("skill")
|
||
canonical = f"{owner}/{repo_name}/{skill_name}"
|
||
if canonical in seen:
|
||
continue
|
||
seen.add(canonical)
|
||
repo = f"{owner}/{repo_name}"
|
||
results.append(SkillMeta(
|
||
name=skill_name,
|
||
description=f"Indexed by skills.sh from {repo}",
|
||
source="skills.sh",
|
||
identifier=self._wrap_identifier(canonical),
|
||
trust_level=self.github.trust_level_for(canonical),
|
||
repo=repo,
|
||
path=skill_name,
|
||
extra={
|
||
"detail_url": f"{self.BASE_URL}/{canonical}",
|
||
"repo_url": f"https://github.com/{repo}",
|
||
},
|
||
))
|
||
|
||
if not results:
|
||
return self._featured_skills(limit)
|
||
|
||
_write_index_cache(cache_key, [_skill_meta_to_dict(item) for item in results])
|
||
return results[:limit] if limit > 0 else results
|
||
|
||
def _featured_skills(self, limit: int) -> List[SkillMeta]:
|
||
cache_key = "skills_sh_featured"
|
||
cached = _read_index_cache(cache_key)
|
||
if cached is not None:
|
||
return [SkillMeta(**item) for item in cached][:limit]
|
||
|
||
try:
|
||
resp = httpx.get(self.BASE_URL, timeout=20)
|
||
if resp.status_code != 200:
|
||
return []
|
||
except httpx.HTTPError:
|
||
return []
|
||
|
||
seen: set[str] = set()
|
||
results: List[SkillMeta] = []
|
||
for match in self._SKILL_LINK_RE.finditer(resp.text):
|
||
canonical = match.group("id")
|
||
if canonical in seen:
|
||
continue
|
||
seen.add(canonical)
|
||
parts = canonical.split("/", 2)
|
||
if len(parts) < 3:
|
||
continue
|
||
repo = f"{parts[0]}/{parts[1]}"
|
||
skill_path = parts[2]
|
||
results.append(SkillMeta(
|
||
name=skill_path.split("/")[-1],
|
||
description=f"Featured on skills.sh from {repo}",
|
||
source="skills.sh",
|
||
identifier=self._wrap_identifier(canonical),
|
||
trust_level=self.github.trust_level_for(canonical),
|
||
repo=repo,
|
||
path=skill_path,
|
||
))
|
||
if len(results) >= limit:
|
||
break
|
||
|
||
_write_index_cache(cache_key, [_skill_meta_to_dict(item) for item in results])
|
||
return results
|
||
|
||
def _meta_from_search_item(self, item: dict) -> Optional[SkillMeta]:
|
||
if not isinstance(item, dict):
|
||
return None
|
||
|
||
canonical = item.get("id")
|
||
repo = item.get("source")
|
||
skill_path = item.get("skillId")
|
||
if not isinstance(canonical, str) or canonical.count("/") < 2:
|
||
if not (isinstance(repo, str) and isinstance(skill_path, str)):
|
||
return None
|
||
canonical = f"{repo}/{skill_path}"
|
||
|
||
parts = canonical.split("/", 2)
|
||
if len(parts) < 3:
|
||
return None
|
||
|
||
repo = f"{parts[0]}/{parts[1]}"
|
||
skill_path = parts[2]
|
||
installs = item.get("installs")
|
||
installs_label = f" · {int(installs):,} installs" if isinstance(installs, int) else ""
|
||
|
||
return SkillMeta(
|
||
name=str(item.get("name") or skill_path.split("/")[-1]),
|
||
description=f"Indexed by skills.sh from {repo}{installs_label}",
|
||
source="skills.sh",
|
||
identifier=self._wrap_identifier(canonical),
|
||
trust_level=self.github.trust_level_for(canonical),
|
||
repo=repo,
|
||
path=skill_path,
|
||
extra={
|
||
"installs": installs,
|
||
"detail_url": f"{self.BASE_URL}/{canonical}",
|
||
"repo_url": f"https://github.com/{repo}",
|
||
},
|
||
)
|
||
|
||
def _fetch_detail_page(self, identifier: str) -> Optional[dict]:
|
||
cache_key = f"skills_sh_detail_{hashlib.md5(identifier.encode()).hexdigest()}"
|
||
cached = _read_index_cache(cache_key)
|
||
if isinstance(cached, dict):
|
||
return cached
|
||
|
||
try:
|
||
resp = httpx.get(f"{self.BASE_URL}/{identifier}", timeout=20)
|
||
if resp.status_code != 200:
|
||
return None
|
||
except httpx.HTTPError:
|
||
return None
|
||
|
||
detail = self._parse_detail_page(identifier, resp.text)
|
||
if detail:
|
||
_write_index_cache(cache_key, detail)
|
||
return detail
|
||
|
||
def _parse_detail_page(self, identifier: str, html: str) -> Optional[dict]:
|
||
parts = identifier.split("/", 2)
|
||
if len(parts) < 3:
|
||
return None
|
||
|
||
default_repo = f"{parts[0]}/{parts[1]}"
|
||
skill_token = parts[2]
|
||
repo = default_repo
|
||
install_skill = skill_token
|
||
|
||
install_command = None
|
||
install_match = self._INSTALL_CMD_RE.search(html)
|
||
if install_match:
|
||
install_command = install_match.group(0).strip()
|
||
repo_value = (install_match.group("repo") or "").strip()
|
||
install_skill = (install_match.group("skill") or install_skill).strip()
|
||
repo = self._extract_repo_slug(repo_value) or repo
|
||
|
||
page_title = self._extract_first_match(self._PAGE_H1_RE, html)
|
||
body_title = self._extract_first_match(self._PROSE_H1_RE, html)
|
||
body_summary = self._extract_first_match(self._PROSE_P_RE, html)
|
||
weekly_installs = self._extract_weekly_installs(html)
|
||
security_audits = self._extract_security_audits(html, identifier)
|
||
|
||
return {
|
||
"repo": repo,
|
||
"install_skill": install_skill,
|
||
"page_title": page_title,
|
||
"body_title": body_title,
|
||
"body_summary": body_summary,
|
||
"weekly_installs": weekly_installs,
|
||
"install_command": install_command,
|
||
"repo_url": f"https://github.com/{repo}",
|
||
"detail_url": f"{self.BASE_URL}/{identifier}",
|
||
"security_audits": security_audits,
|
||
}
|
||
|
||
def _discover_identifier(self, identifier: str, detail: Optional[dict] = None) -> Optional[str]:
|
||
parts = identifier.split("/", 2)
|
||
if len(parts) < 3:
|
||
return None
|
||
|
||
default_repo = f"{parts[0]}/{parts[1]}"
|
||
repo = detail.get("repo", default_repo) if isinstance(detail, dict) else default_repo
|
||
skill_token=parts[2].split("/")[-1]
|
||
tokens=[skill_token]
|
||
if isinstance(detail, dict):
|
||
tokens.extend([
|
||
detail.get("install_skill", ""),
|
||
detail.get("page_title", ""),
|
||
detail.get("body_title", ""),
|
||
])
|
||
|
||
# Standard skill paths
|
||
base_paths = ["skills/", ".agents/skills/", ".claude/skills/"]
|
||
|
||
for base_path in base_paths:
|
||
try:
|
||
skills = self.github._list_skills_in_repo(repo, base_path)
|
||
except Exception:
|
||
continue
|
||
for meta in skills:
|
||
if self._matches_skill_tokens(meta, tokens):
|
||
return meta.identifier
|
||
|
||
# Prefer a single recursive tree lookup before brute-forcing every
|
||
# top-level directory. This avoids large request bursts on categorized
|
||
# repos like borghei/claude-skills.
|
||
tree_result = self.github._find_skill_in_repo_tree(repo, skill_token)
|
||
if tree_result:
|
||
return tree_result
|
||
|
||
# Fallback: scan repo root for directories that might contain skills
|
||
try:
|
||
root_url = f"https://api.github.com/repos/{repo}/contents/"
|
||
resp = httpx.get(root_url, headers=self.github.auth.get_headers(),
|
||
timeout=15, follow_redirects=True)
|
||
if resp.status_code == 200:
|
||
entries = resp.json()
|
||
if isinstance(entries, list):
|
||
for entry in entries:
|
||
if entry.get("type") != "dir":
|
||
continue
|
||
dir_name = entry["name"]
|
||
if dir_name.startswith((".", "_")):
|
||
continue
|
||
if dir_name in {"skills", ".agents", ".claude"}:
|
||
continue # already tried
|
||
# Try direct: repo/dir/skill_token
|
||
direct_id = f"{repo}/{dir_name}/{skill_token}"
|
||
meta = self.github.inspect(direct_id)
|
||
if meta:
|
||
return meta.identifier
|
||
# Try listing skills in this directory
|
||
try:
|
||
skills = self.github._list_skills_in_repo(repo, dir_name + "/")
|
||
except Exception:
|
||
continue
|
||
for meta in skills:
|
||
if self._matches_skill_tokens(meta, tokens):
|
||
return meta.identifier
|
||
except Exception:
|
||
pass
|
||
|
||
return None
|
||
|
||
def _resolve_github_meta(self, identifier: str, detail: Optional[dict] = None) -> Optional[SkillMeta]:
|
||
for candidate in self._candidate_identifiers(identifier):
|
||
meta = self.github.inspect(candidate)
|
||
if meta:
|
||
return meta
|
||
|
||
resolved = self._discover_identifier(identifier, detail=detail)
|
||
if resolved:
|
||
return self.github.inspect(resolved)
|
||
return None
|
||
|
||
def _finalize_inspect_meta(self, meta: SkillMeta, canonical: str, detail: Optional[dict]) -> SkillMeta:
|
||
meta.source = "skills.sh"
|
||
meta.identifier = self._wrap_identifier(canonical)
|
||
meta.trust_level = self.trust_level_for(canonical)
|
||
merged_extra = dict(meta.extra)
|
||
merged_extra.update(self._detail_to_metadata(canonical, detail))
|
||
meta.extra = merged_extra
|
||
|
||
if isinstance(detail, dict):
|
||
body_summary = detail.get("body_summary")
|
||
weekly_installs = detail.get("weekly_installs")
|
||
if body_summary:
|
||
meta.description = body_summary
|
||
elif meta.description and weekly_installs:
|
||
meta.description = f"{meta.description} · {weekly_installs} weekly installs on skills.sh"
|
||
return meta
|
||
|
||
@classmethod
|
||
def _matches_skill_tokens(cls, meta: SkillMeta, skill_tokens: List[str]) -> bool:
|
||
candidates = set()
|
||
candidates.update(cls._token_variants(meta.name))
|
||
candidates.update(cls._token_variants(meta.path))
|
||
candidates.update(cls._token_variants(meta.identifier.split("/", 2)[-1] if meta.identifier else None))
|
||
|
||
for token in skill_tokens:
|
||
variants = cls._token_variants(token)
|
||
if variants & candidates:
|
||
return True
|
||
return False
|
||
|
||
@staticmethod
|
||
def _token_variants(value: Optional[str]) -> set[str]:
|
||
if not value:
|
||
return set()
|
||
|
||
plain = SkillsShSource._strip_html(str(value)).strip().strip("/").lower()
|
||
if not plain:
|
||
return set()
|
||
|
||
base = plain.split("/")[-1]
|
||
sanitized = re.sub(r'[^a-z0-9/_-]+', '-', plain).strip('-')
|
||
sanitized_base = sanitized.split("/")[-1] if sanitized else ""
|
||
slash_tail = plain.split("/")[-1]
|
||
slash_tail_clean = slash_tail.lstrip('@')
|
||
slash_tail_clean = slash_tail_clean.split('/')[-1]
|
||
|
||
variants = {
|
||
plain,
|
||
plain.replace("_", "-"),
|
||
plain.replace("/", "-"),
|
||
base,
|
||
base.replace("_", "-"),
|
||
base.replace("/", "-"),
|
||
sanitized,
|
||
sanitized.replace("/", "-") if sanitized else "",
|
||
sanitized_base,
|
||
slash_tail_clean,
|
||
slash_tail_clean.replace("_", "-"),
|
||
}
|
||
return {v for v in variants if v}
|
||
|
||
@staticmethod
|
||
def _extract_repo_slug(repo_value: str) -> Optional[str]:
|
||
repo_value = repo_value.strip()
|
||
if repo_value.startswith("https://github.com/"):
|
||
repo_value = repo_value[len("https://github.com/"):]
|
||
repo_value = repo_value.strip("/")
|
||
parts = repo_value.split("/")
|
||
if len(parts) >= 2:
|
||
return f"{parts[0]}/{parts[1]}"
|
||
return None
|
||
|
||
@staticmethod
|
||
def _extract_first_match(pattern: re.Pattern, text: str) -> Optional[str]:
|
||
match = pattern.search(text)
|
||
if not match:
|
||
return None
|
||
value = next((group for group in match.groups() if group), None)
|
||
if value is None:
|
||
return None
|
||
return SkillsShSource._strip_html(value).strip() or None
|
||
|
||
def _detail_to_metadata(self, canonical: str, detail: Optional[dict]) -> Dict[str, Any]:
|
||
parts = canonical.split("/", 2)
|
||
repo = f"{parts[0]}/{parts[1]}" if len(parts) >= 2 else ""
|
||
metadata = {
|
||
"detail_url": f"{self.BASE_URL}/{canonical}",
|
||
}
|
||
if repo:
|
||
metadata["repo_url"] = f"https://github.com/{repo}"
|
||
if isinstance(detail, dict):
|
||
for key in ("weekly_installs", "install_command", "repo_url", "detail_url", "security_audits"):
|
||
value = detail.get(key)
|
||
if value:
|
||
metadata[key] = value
|
||
return metadata
|
||
|
||
@staticmethod
|
||
def _extract_weekly_installs(html: str) -> Optional[str]:
|
||
match = SkillsShSource._WEEKLY_INSTALLS_RE.search(html)
|
||
if not match:
|
||
return None
|
||
return match.group("count")
|
||
|
||
@staticmethod
|
||
def _extract_security_audits(html: str, identifier: str) -> Dict[str, str]:
|
||
audits: Dict[str, str] = {}
|
||
for audit in ("agent-trust-hub", "socket", "snyk"):
|
||
idx = html.find(f"/security/{audit}")
|
||
if idx == -1:
|
||
continue
|
||
window = html[idx:idx + 500]
|
||
match = re.search(r'(Pass|Warn|Fail)', window, re.IGNORECASE)
|
||
if match:
|
||
audits[audit] = match.group(1).title()
|
||
return audits
|
||
|
||
@staticmethod
|
||
def _strip_html(value: str) -> str:
|
||
return re.sub(r'<[^>]+>', '', value)
|
||
|
||
@staticmethod
|
||
def _normalize_identifier(identifier: str) -> str:
|
||
prefix_aliases = (
|
||
"skills-sh/",
|
||
"skills.sh/",
|
||
"skils-sh/",
|
||
"skils.sh/",
|
||
)
|
||
for prefix in prefix_aliases:
|
||
if identifier.startswith(prefix):
|
||
return identifier[len(prefix):]
|
||
return identifier
|
||
|
||
@staticmethod
|
||
def _candidate_identifiers(identifier: str) -> List[str]:
|
||
parts = identifier.split("/", 2)
|
||
if len(parts) < 3:
|
||
return [identifier]
|
||
|
||
repo = f"{parts[0]}/{parts[1]}"
|
||
skill_path = parts[2].lstrip("/")
|
||
candidates = [
|
||
f"{repo}/{skill_path}",
|
||
f"{repo}/skills/{skill_path}",
|
||
f"{repo}/.agents/skills/{skill_path}",
|
||
f"{repo}/.claude/skills/{skill_path}",
|
||
]
|
||
|
||
seen = set()
|
||
deduped: List[str] = []
|
||
for candidate in candidates:
|
||
if candidate not in seen:
|
||
seen.add(candidate)
|
||
deduped.append(candidate)
|
||
return deduped
|
||
|
||
@staticmethod
|
||
def _wrap_identifier(identifier: str) -> str:
|
||
return f"skills-sh/{identifier}"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# ClawHub source adapter
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class ClawHubSource(SkillSource):
|
||
"""
|
||
Fetch skills from ClawHub (clawhub.ai) via their HTTP API.
|
||
All skills are treated as community trust — ClawHavoc incident showed
|
||
their vetting is insufficient (341 malicious skills found Feb 2026).
|
||
"""
|
||
|
||
BASE_URL = "https://clawhub.ai/api/v1"
|
||
|
||
# Wall-clock budget for a full catalog walk. ClawHub has 50k+ skills and
|
||
# the walk is sequential (~250 requests, each under per-request
|
||
# timeout=30 so nothing errors), so an unbounded walk can block for
|
||
# minutes. Bound it so a slow/large catalog cannot hang the caller.
|
||
CATALOG_WALK_BUDGET_SECONDS = 12
|
||
|
||
def source_id(self) -> str:
|
||
return "clawhub"
|
||
|
||
def trust_level_for(self, identifier: str) -> str:
|
||
return "community"
|
||
|
||
@staticmethod
|
||
def _normalize_tags(tags: Any) -> List[str]:
|
||
if isinstance(tags, list):
|
||
return [str(t) for t in tags]
|
||
if isinstance(tags, dict):
|
||
return [str(k) for k in tags if str(k) != "latest"]
|
||
return []
|
||
|
||
@staticmethod
|
||
def _coerce_skill_payload(data: Any) -> Optional[Dict[str, Any]]:
|
||
if not isinstance(data, dict):
|
||
return None
|
||
nested = data.get("skill")
|
||
if isinstance(nested, dict):
|
||
merged = dict(nested)
|
||
latest_version = data.get("latestVersion")
|
||
if latest_version is not None and "latestVersion" not in merged:
|
||
merged["latestVersion"] = latest_version
|
||
# Carry over top-level fields that the listing API nests alongside
|
||
# the skill object — owner is needed for building valid detail URLs.
|
||
if "owner" in data and "owner" not in merged:
|
||
merged["owner"] = data["owner"]
|
||
return merged
|
||
return data
|
||
|
||
@staticmethod
|
||
def _query_terms(query: str) -> List[str]:
|
||
return [term for term in re.split(r"[^a-z0-9]+", query.lower()) if term]
|
||
|
||
@classmethod
|
||
def _search_score(cls, query: str, meta: SkillMeta) -> int:
|
||
query_norm = query.strip().lower()
|
||
if not query_norm:
|
||
return 1
|
||
|
||
identifier = (meta.identifier or "").lower()
|
||
name = (meta.name or "").lower()
|
||
description = (meta.description or "").lower()
|
||
normalized_identifier = " ".join(cls._query_terms(identifier))
|
||
normalized_name = " ".join(cls._query_terms(name))
|
||
query_terms = cls._query_terms(query_norm)
|
||
identifier_terms = cls._query_terms(identifier)
|
||
name_terms = cls._query_terms(name)
|
||
score = 0
|
||
|
||
if query_norm == identifier:
|
||
score += 140
|
||
if query_norm == name:
|
||
score += 130
|
||
if normalized_identifier == query_norm:
|
||
score += 125
|
||
if normalized_name == query_norm:
|
||
score += 120
|
||
if normalized_identifier.startswith(query_norm):
|
||
score += 95
|
||
if normalized_name.startswith(query_norm):
|
||
score += 90
|
||
if query_terms and identifier_terms[: len(query_terms)] == query_terms:
|
||
score += 70
|
||
if query_terms and name_terms[: len(query_terms)] == query_terms:
|
||
score += 65
|
||
if query_norm in identifier:
|
||
score += 40
|
||
if query_norm in name:
|
||
score += 35
|
||
if query_norm in description:
|
||
score += 10
|
||
|
||
for term in query_terms:
|
||
if term in identifier_terms:
|
||
score += 15
|
||
if term in name_terms:
|
||
score += 12
|
||
if term in description:
|
||
score += 3
|
||
|
||
return score
|
||
|
||
@staticmethod
|
||
def _dedupe_results(results: List[SkillMeta]) -> List[SkillMeta]:
|
||
seen: set[str] = set()
|
||
deduped: List[SkillMeta] = []
|
||
for result in results:
|
||
key = (result.identifier or result.name).lower()
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
deduped.append(result)
|
||
return deduped
|
||
|
||
def _exact_slug_meta(self, query: str) -> Optional[SkillMeta]:
|
||
query = query.strip()
|
||
parsed = self._parse_identifier(query)
|
||
query_terms = self._query_terms(query)
|
||
candidates: List[str] = []
|
||
|
||
if parsed:
|
||
candidates.append(parsed[0])
|
||
elif "/" not in query and self._SLUG_RE.fullmatch(query):
|
||
candidates.append(query)
|
||
|
||
if query_terms:
|
||
base_slug = "-".join(query_terms)
|
||
if len(query_terms) >= 2:
|
||
candidates.extend([
|
||
f"{base_slug}-agent",
|
||
f"{base_slug}-skill",
|
||
f"{base_slug}-tool",
|
||
f"{base_slug}-assistant",
|
||
f"{base_slug}-playbook",
|
||
base_slug,
|
||
])
|
||
else:
|
||
candidates.append(base_slug)
|
||
|
||
seen: set[str] = set()
|
||
for candidate in candidates:
|
||
if candidate in seen:
|
||
continue
|
||
seen.add(candidate)
|
||
meta = self.inspect(candidate)
|
||
if meta:
|
||
return meta
|
||
|
||
return None
|
||
|
||
def _finalize_search_results(self, query: str, results: List[SkillMeta], limit: int) -> List[SkillMeta]:
|
||
query_norm = query.strip()
|
||
if not query_norm:
|
||
return self._dedupe_results(results)[:limit]
|
||
|
||
filtered = [meta for meta in results if self._search_score(query_norm, meta) > 0]
|
||
filtered.sort(
|
||
key=lambda meta: (
|
||
-self._search_score(query_norm, meta),
|
||
meta.name.lower(),
|
||
meta.identifier.lower(),
|
||
)
|
||
)
|
||
filtered = self._dedupe_results(filtered)
|
||
|
||
exact = self._exact_slug_meta(query_norm)
|
||
if exact:
|
||
filtered = [meta for meta in filtered if self._search_score(query_norm, meta) >= 20]
|
||
filtered = self._dedupe_results([exact] + filtered)
|
||
|
||
if filtered:
|
||
return filtered[:limit]
|
||
|
||
if re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._/-]*", query_norm):
|
||
return []
|
||
|
||
return self._dedupe_results(results)[:limit]
|
||
|
||
def search(self, query: str, limit: int = 10) -> List[SkillMeta]:
|
||
query = query.strip()
|
||
|
||
if query:
|
||
query_terms = self._query_terms(query)
|
||
if len(query_terms) >= 2:
|
||
direct = self._exact_slug_meta(query)
|
||
if direct:
|
||
return [direct]
|
||
|
||
results = self._search_catalog(query, limit=limit)
|
||
if results:
|
||
return results
|
||
else:
|
||
# Empty query: route through the paginating catalog walker. When
|
||
# the full catalog is already disk-cached this returns it whole and
|
||
# the caller paginates client-side. On a cold cache, bound the walk
|
||
# to `limit` so a browse command renders its first page without
|
||
# walking the entire 50k+ catalog (max_items=0 → unbounded, used
|
||
# only by the offline index builder via search("", limit=0)).
|
||
catalog = self._load_catalog_index(max_items=limit if limit > 0 else 0)
|
||
if catalog:
|
||
return self._dedupe_results(catalog)[:limit] if limit > 0 else self._dedupe_results(catalog)
|
||
|
||
# Non-empty query catalog miss, or catalog walker failure: fall back to
|
||
# the lightweight listing API for a best-effort response.
|
||
cache_key = f"clawhub_search_listing_v1_{hashlib.md5(query.encode()).hexdigest()}_{limit}"
|
||
cached = _read_index_cache(cache_key)
|
||
if cached is not None:
|
||
return self._finalize_search_results(
|
||
query,
|
||
[SkillMeta(**s) for s in cached],
|
||
limit,
|
||
)
|
||
|
||
try:
|
||
resp = httpx.get(
|
||
f"{self.BASE_URL}/skills",
|
||
params={"search": query, "limit": limit},
|
||
timeout=15,
|
||
)
|
||
if resp.status_code != 200:
|
||
return []
|
||
data = resp.json()
|
||
except (httpx.HTTPError, json.JSONDecodeError):
|
||
return []
|
||
|
||
skills_data = data.get("items", data) if isinstance(data, dict) else data
|
||
if not isinstance(skills_data, list):
|
||
return []
|
||
|
||
results = []
|
||
for item in skills_data[:limit]:
|
||
slug = item.get("slug")
|
||
if not slug:
|
||
continue
|
||
display_name = item.get("displayName") or item.get("name") or slug
|
||
summary = item.get("summary") or item.get("description") or ""
|
||
tags = self._normalize_tags(item.get("tags", []))
|
||
extra: Dict[str, Any] = {}
|
||
owner = item.get("owner")
|
||
if isinstance(owner, dict):
|
||
handle = owner.get("handle")
|
||
if isinstance(handle, str) and handle:
|
||
extra["owner"] = handle
|
||
elif isinstance(owner, str) and owner:
|
||
extra["owner"] = owner
|
||
results.append(SkillMeta(
|
||
name=display_name,
|
||
description=summary,
|
||
source="clawhub",
|
||
identifier=slug,
|
||
trust_level="community",
|
||
tags=tags,
|
||
extra=extra,
|
||
))
|
||
|
||
final_results = self._finalize_search_results(query, results, limit)
|
||
_write_index_cache(cache_key, [_skill_meta_to_dict(s) for s in final_results])
|
||
return final_results
|
||
|
||
_SLUG_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]*$")
|
||
|
||
@classmethod
|
||
def _parse_identifier(cls, identifier: str) -> Optional[Tuple[str, Optional[str]]]:
|
||
"""Return ``(slug, expected_owner)`` for a ClawHub identifier.
|
||
|
||
Accepts a bare slug, ``clawhub/<slug>``, ``@owner/slug``, and the
|
||
clawhub.ai URL path ``owner/skills/slug``. GitHub-style
|
||
``owner/repo/skill`` and ``owner/repo/skills/skill`` identifiers
|
||
are not ClawHub's — claiming them by last path segment installs
|
||
a same-named skill from a different author.
|
||
"""
|
||
raw = (identifier or "").strip()
|
||
if not raw:
|
||
return None
|
||
had_at = raw.startswith("@")
|
||
ident = raw[1:] if had_at else raw
|
||
if ident.startswith("clawhub/"):
|
||
ident = ident[len("clawhub/"):]
|
||
parts = [part for part in ident.split("/") if part]
|
||
if len(parts) == 1:
|
||
slug = parts[0]
|
||
return (slug, None) if cls._SLUG_RE.fullmatch(slug) else None
|
||
if len(parts) == 2 and had_at:
|
||
owner, slug = parts
|
||
if cls._SLUG_RE.fullmatch(owner) and cls._SLUG_RE.fullmatch(slug):
|
||
return slug, owner
|
||
return None
|
||
if len(parts) == 3 and parts[1].lower() == "skills":
|
||
owner, _, slug = parts
|
||
if cls._SLUG_RE.fullmatch(owner) and cls._SLUG_RE.fullmatch(slug):
|
||
return slug, owner
|
||
return None
|
||
return None
|
||
|
||
@staticmethod
|
||
def _owner_from_payload(data: Optional[Dict[str, Any]]) -> Optional[str]:
|
||
if not isinstance(data, dict):
|
||
return None
|
||
owner = data.get("owner")
|
||
if isinstance(owner, dict):
|
||
handle = owner.get("handle")
|
||
if isinstance(handle, str) and handle.strip():
|
||
return handle.strip()
|
||
if isinstance(owner, str) and owner.strip():
|
||
return owner.strip()
|
||
return None
|
||
|
||
@classmethod
|
||
def _owner_matches(cls, expected_owner: Optional[str], data: Optional[Dict[str, Any]]) -> bool:
|
||
if not expected_owner:
|
||
return True
|
||
actual = cls._owner_from_payload(data)
|
||
if not actual:
|
||
return True
|
||
return actual.lower() == expected_owner.lower()
|
||
|
||
def fetch(self, identifier: str) -> Optional[SkillBundle]:
|
||
parsed = self._parse_identifier(identifier)
|
||
if parsed is None:
|
||
return None
|
||
slug, expected_owner = parsed
|
||
|
||
skill_data = self._coerce_skill_payload(self._get_json(f"{self.BASE_URL}/skills/{slug}"))
|
||
if not isinstance(skill_data, dict):
|
||
return None
|
||
if not self._owner_matches(expected_owner, skill_data):
|
||
return None
|
||
|
||
latest_version = self._resolve_latest_version(slug, skill_data)
|
||
if not latest_version:
|
||
logger.warning("ClawHub fetch failed for %s: could not resolve latest version", slug)
|
||
return None
|
||
|
||
# Primary method: download the skill as a ZIP bundle from /download
|
||
files = self._download_zip(slug, latest_version)
|
||
|
||
# Fallback: try the version metadata endpoint for inline/raw content
|
||
if "SKILL.md" not in files:
|
||
version_data = self._get_json(f"{self.BASE_URL}/skills/{slug}/versions/{latest_version}")
|
||
if isinstance(version_data, dict):
|
||
# Files may be nested under version_data["version"]["files"]
|
||
files = self._extract_files(version_data) or files
|
||
if "SKILL.md" not in files:
|
||
nested = version_data.get("version", {})
|
||
if isinstance(nested, dict):
|
||
files = self._extract_files(nested) or files
|
||
|
||
if "SKILL.md" not in files:
|
||
logger.warning(
|
||
"ClawHub fetch for %s resolved version %s but could not retrieve file content",
|
||
slug,
|
||
latest_version,
|
||
)
|
||
return None
|
||
|
||
return SkillBundle(
|
||
name=slug,
|
||
files=files,
|
||
source="clawhub",
|
||
identifier=slug,
|
||
trust_level="community",
|
||
)
|
||
|
||
def inspect(self, identifier: str) -> Optional[SkillMeta]:
|
||
parsed = self._parse_identifier(identifier)
|
||
if parsed is None:
|
||
return None
|
||
slug, expected_owner = parsed
|
||
data = self._coerce_skill_payload(self._get_json(f"{self.BASE_URL}/skills/{slug}"))
|
||
if not isinstance(data, dict):
|
||
return None
|
||
if not self._owner_matches(expected_owner, data):
|
||
return None
|
||
|
||
tags = self._normalize_tags(data.get("tags", []))
|
||
extra: Dict[str, Any] = {}
|
||
# The detail API returns owner info — capture it so callers can build
|
||
# valid ClawHub URLs (https://clawhub.ai/{owner}/skills/{slug}).
|
||
owner = self._owner_from_payload(data)
|
||
if owner:
|
||
extra["owner"] = owner
|
||
|
||
return SkillMeta(
|
||
name=data.get("displayName") or data.get("name") or data.get("slug") or slug,
|
||
description=data.get("summary") or data.get("description") or "",
|
||
source="clawhub",
|
||
identifier=data.get("slug") or slug,
|
||
trust_level="community",
|
||
tags=tags,
|
||
extra=extra,
|
||
)
|
||
|
||
def _search_catalog(self, query: str, limit: int = 10) -> List[SkillMeta]:
|
||
cache_key = f"clawhub_search_catalog_v1_{hashlib.md5(f'{query}|{limit}'.encode()).hexdigest()}"
|
||
cached = _read_index_cache(cache_key)
|
||
if cached is not None:
|
||
return [SkillMeta(**s) for s in cached][:limit]
|
||
|
||
catalog = self._load_catalog_index()
|
||
if not catalog:
|
||
return []
|
||
|
||
results = self._finalize_search_results(query, catalog, limit)
|
||
_write_index_cache(cache_key, [_skill_meta_to_dict(s) for s in results])
|
||
return results
|
||
|
||
def _load_catalog_index(self, max_items: int = 0) -> List[SkillMeta]:
|
||
"""Walk the ClawHub catalog via cursor pagination.
|
||
|
||
``max_items`` bounds the walk: once at least that many distinct skills
|
||
have been gathered the walk stops early. This is what browse's
|
||
cold-start fallback wants — it only renders one page, so walking the
|
||
entire 50k+ catalog just to slice off the first N is pure waste.
|
||
``max_items=0`` (the default, used by the offline index builder) means
|
||
walk to exhaustion.
|
||
|
||
Caching: only a *complete* catalog (cursor exhausted or page cap) is
|
||
written to the shared ``clawhub_catalog_v1`` cache. A walk truncated by
|
||
``max_items`` OR the wall-clock budget is partial, so caching it would
|
||
poison the full-catalog cache with an incomplete slice.
|
||
"""
|
||
cache_key = "clawhub_catalog_v1"
|
||
cached = _read_index_cache(cache_key)
|
||
if cached is not None:
|
||
return [SkillMeta(**s) for s in cached]
|
||
|
||
cursor: Optional[str] = None
|
||
results: List[SkillMeta] = []
|
||
seen: set[str] = set()
|
||
# ClawHub has 50k+ skills as of May 2026 (live E2E walked 49,698 with
|
||
# an active cursor still pending); 750 pages * 200/page = 150k ceiling
|
||
# leaves room for catalog growth. Walk-to-exhaustion typically
|
||
# terminates well before this on `nextCursor` going None — the cap is
|
||
# a safety rail against an infinite-cursor loop.
|
||
max_pages = 750
|
||
# Wall-clock budget is for interactive browse (max_items > 0) only.
|
||
# The offline index builder passes max_items=0 and must walk the full
|
||
# catalog — a 12s cap there ships ~3k skills and trips the deploy
|
||
# health floor (20k).
|
||
deadline = (
|
||
time.monotonic() + self.CATALOG_WALK_BUDGET_SECONDS
|
||
if max_items > 0
|
||
else None
|
||
)
|
||
hit_deadline = False
|
||
hit_max_items = False
|
||
|
||
for _ in range(max_pages):
|
||
if deadline is not None and time.monotonic() > deadline:
|
||
hit_deadline = True
|
||
break
|
||
params: Dict[str, Any] = {"limit": 200}
|
||
if cursor:
|
||
params["cursor"] = cursor
|
||
|
||
try:
|
||
resp = httpx.get(f"{self.BASE_URL}/skills", params=params, timeout=30)
|
||
if resp.status_code != 200:
|
||
break
|
||
data = resp.json()
|
||
except (httpx.HTTPError, json.JSONDecodeError):
|
||
break
|
||
|
||
items = data.get("items", []) if isinstance(data, dict) else []
|
||
if not isinstance(items, list) or not items:
|
||
break
|
||
|
||
for item in items:
|
||
slug = item.get("slug")
|
||
if not isinstance(slug, str) or not slug or slug in seen:
|
||
continue
|
||
seen.add(slug)
|
||
display_name = item.get("displayName") or item.get("name") or slug
|
||
summary = item.get("summary") or item.get("description") or ""
|
||
tags = self._normalize_tags(item.get("tags", []))
|
||
extra: Dict[str, Any] = {}
|
||
owner = self._owner_from_payload(item)
|
||
if owner:
|
||
extra["owner"] = owner
|
||
results.append(SkillMeta(
|
||
name=display_name,
|
||
description=summary,
|
||
source="clawhub",
|
||
identifier=slug,
|
||
trust_level="community",
|
||
tags=tags,
|
||
extra=extra,
|
||
))
|
||
|
||
cursor = data.get("nextCursor") if isinstance(data, dict) else None
|
||
if not isinstance(cursor, str) or not cursor:
|
||
break
|
||
|
||
# Browse's cold-start fallback only renders one page, so stop as
|
||
# soon as we have enough to satisfy the caller's bound. The index
|
||
# builder passes max_items=0 (unbounded) and walks to exhaustion.
|
||
if max_items > 0 and len(results) >= max_items:
|
||
hit_max_items = True
|
||
break
|
||
|
||
# Only cache a walk that reached a natural stop (cursor exhausted or
|
||
# page cap). A walk truncated by the wall-clock budget OR by max_items
|
||
# is partial, so writing it would poison the shared full-catalog cache
|
||
# with incomplete data.
|
||
if not hit_deadline and not hit_max_items:
|
||
_write_index_cache(cache_key, [_skill_meta_to_dict(s) for s in results])
|
||
return results
|
||
|
||
def _get_json(self, url: str, timeout: int = 20) -> Optional[Any]:
|
||
try:
|
||
resp = httpx.get(url, timeout=timeout)
|
||
if resp.status_code != 200:
|
||
return None
|
||
return resp.json()
|
||
except (httpx.HTTPError, json.JSONDecodeError):
|
||
return None
|
||
|
||
def _resolve_latest_version(self, slug: str, skill_data: Dict[str, Any]) -> Optional[str]:
|
||
latest = skill_data.get("latestVersion")
|
||
if isinstance(latest, dict):
|
||
version = latest.get("version")
|
||
if isinstance(version, str) and version:
|
||
return version
|
||
|
||
tags = skill_data.get("tags")
|
||
if isinstance(tags, dict):
|
||
latest_tag = tags.get("latest")
|
||
if isinstance(latest_tag, str) and latest_tag:
|
||
return latest_tag
|
||
|
||
versions_data = self._get_json(f"{self.BASE_URL}/skills/{slug}/versions")
|
||
if isinstance(versions_data, list) and versions_data:
|
||
first = versions_data[0]
|
||
if isinstance(first, dict):
|
||
version = first.get("version")
|
||
if isinstance(version, str) and version:
|
||
return version
|
||
return None
|
||
|
||
def _fetch_owner_handle(self, slug: str) -> Optional[str]:
|
||
"""Fetch the owner handle for a single ClawHub skill via the detail API.
|
||
|
||
Returns the owner handle string, or None if unavailable.
|
||
The detail endpoint at ``/api/v1/skills/{slug}`` returns an ``owner``
|
||
object with a ``handle`` field — the listing API does not include this.
|
||
|
||
Retry semantics (bounded):
|
||
- Up to 3 attempts total (initial + 2 retries).
|
||
- On HTTP 429: respects ``Retry-After`` header (seconds) when present,
|
||
otherwise exponential backoff (2s → 4s).
|
||
- On HTTP 5xx: exponential backoff (transient server errors).
|
||
- On HTTP 4xx (non-429): no retry — the resource doesn't exist.
|
||
"""
|
||
url = f"{self.BASE_URL}/skills/{slug}"
|
||
max_attempts = 3
|
||
backoff_base = 2.0 # seconds
|
||
|
||
for attempt in range(max_attempts):
|
||
try:
|
||
resp = httpx.get(url, timeout=20)
|
||
except (httpx.HTTPError, OSError):
|
||
# Network/transport error — treat as transient, retry with backoff.
|
||
if attempt < max_attempts - 1:
|
||
delay = backoff_base * (2 ** attempt)
|
||
logger.debug(
|
||
"_fetch_owner_handle(%s): transport error on attempt %d/%d, "
|
||
"retrying in %.1fs",
|
||
slug, attempt + 1, max_attempts, delay,
|
||
)
|
||
time.sleep(delay)
|
||
continue
|
||
return None
|
||
|
||
if resp.status_code == 200:
|
||
try:
|
||
raw = resp.json()
|
||
except (json.JSONDecodeError, ValueError):
|
||
return None
|
||
data = self._coerce_skill_payload(raw)
|
||
if not isinstance(data, dict):
|
||
return None
|
||
return self._owner_from_payload(data)
|
||
|
||
if resp.status_code == 429:
|
||
# Rate-limited — honour Retry-After if present, else backoff.
|
||
if attempt < max_attempts - 1:
|
||
retry_after_raw = resp.headers.get("Retry-After")
|
||
try:
|
||
delay = float(retry_after_raw) if retry_after_raw else backoff_base * (2 ** attempt)
|
||
except (TypeError, ValueError):
|
||
delay = backoff_base * (2 ** attempt)
|
||
logger.debug(
|
||
"_fetch_owner_handle(%s): HTTP 429 on attempt %d/%d, "
|
||
"retrying in %.1fs",
|
||
slug, attempt + 1, max_attempts, delay,
|
||
)
|
||
time.sleep(delay)
|
||
continue
|
||
return None
|
||
|
||
if 500 <= resp.status_code < 600:
|
||
# Transient server error — retry with backoff.
|
||
if attempt < max_attempts - 1:
|
||
delay = backoff_base * (2 ** attempt)
|
||
logger.debug(
|
||
"_fetch_owner_handle(%s): HTTP %d on attempt %d/%d, "
|
||
"retrying in %.1fs",
|
||
slug, resp.status_code, attempt + 1, max_attempts, delay,
|
||
)
|
||
time.sleep(delay)
|
||
continue
|
||
return None
|
||
|
||
# 4xx (non-429) — resource doesn't exist / bad request. No retry.
|
||
return None
|
||
|
||
return None
|
||
|
||
def enrich_owners(self, skills: List[SkillMeta], max_workers: int = 30) -> int:
|
||
"""Batch-fetch owner handles for ClawHub skills missing ``extra["owner"]``.
|
||
|
||
Mutates each SkillMeta in-place, setting ``extra["owner"]`` when the
|
||
detail API returns a handle. Returns the number of skills enriched.
|
||
|
||
This is intended for the offline index builder, which walks the full
|
||
50k+ catalog. The listing API does not include owner info, so we
|
||
fetch each skill's detail page concurrently. With ``max_workers=30``
|
||
the full catalog takes ~5–10 minutes — acceptable for a twice-daily
|
||
batch job.
|
||
|
||
Safety rails:
|
||
- Aborts early if 50 consecutive requests all fail (systemic outage).
|
||
- Respects HTTP 429 rate-limit responses with exponential backoff.
|
||
- Logs progress every 1000 skills so the batch job is observable.
|
||
"""
|
||
needs_enrichment = [
|
||
s for s in skills
|
||
if s.source == "clawhub" and not (s.extra or {}).get("owner")
|
||
]
|
||
if not needs_enrichment:
|
||
return 0
|
||
|
||
enriched = 0
|
||
consecutive_failures = 0
|
||
max_consecutive_failures = 50
|
||
processed = 0
|
||
import threading
|
||
lock = threading.Lock()
|
||
|
||
def _fetch(meta: SkillMeta) -> Optional[str]:
|
||
return self._fetch_owner_handle(meta.identifier)
|
||
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
with ThreadPoolExecutor(max_workers=max_workers) as pool:
|
||
futures = {pool.submit(_fetch, s): s for s in needs_enrichment}
|
||
for future in as_completed(futures):
|
||
meta = futures[future]
|
||
processed += 1
|
||
try:
|
||
handle = future.result()
|
||
if handle:
|
||
with lock:
|
||
if not meta.extra:
|
||
meta.extra = {}
|
||
meta.extra["owner"] = handle
|
||
enriched += 1
|
||
consecutive_failures = 0
|
||
else:
|
||
with lock:
|
||
consecutive_failures += 1
|
||
except Exception:
|
||
with lock:
|
||
consecutive_failures += 1
|
||
|
||
if processed % 1000 == 0:
|
||
logger.info(
|
||
"ClawHub owner enrichment: %d/%d processed, %d enriched",
|
||
processed, len(needs_enrichment), enriched,
|
||
)
|
||
|
||
with lock:
|
||
if consecutive_failures >= max_consecutive_failures:
|
||
logger.warning(
|
||
"ClawHub owner enrichment: %d consecutive failures — "
|
||
"aborting early (%d/%d processed, %d enriched). "
|
||
"The ClawHub API may be down or rate-limited.",
|
||
max_consecutive_failures, processed,
|
||
len(needs_enrichment), enriched,
|
||
)
|
||
# Cancel pending futures
|
||
for f in futures:
|
||
f.cancel()
|
||
break
|
||
|
||
return enriched
|
||
|
||
def _extract_files(self, version_data: Dict[str, Any]) -> Dict[str, str]:
|
||
files: Dict[str, str] = {}
|
||
file_list = version_data.get("files")
|
||
|
||
if isinstance(file_list, dict):
|
||
return {k: v for k, v in file_list.items() if isinstance(v, str)}
|
||
|
||
if not isinstance(file_list, list):
|
||
return files
|
||
|
||
for file_meta in file_list:
|
||
if not isinstance(file_meta, dict):
|
||
continue
|
||
|
||
fname = file_meta.get("path") or file_meta.get("name")
|
||
if not fname or not isinstance(fname, str):
|
||
continue
|
||
|
||
inline_content = file_meta.get("content")
|
||
if isinstance(inline_content, str):
|
||
files[fname] = inline_content
|
||
continue
|
||
|
||
raw_url = file_meta.get("rawUrl") or file_meta.get("downloadUrl") or file_meta.get("url")
|
||
if isinstance(raw_url, str) and raw_url.startswith("http"):
|
||
content = self._fetch_text(raw_url)
|
||
if content is not None:
|
||
files[fname] = content
|
||
|
||
return files
|
||
|
||
def _download_zip(self, slug: str, version: str) -> Dict[str, str]:
|
||
"""Download skill as a ZIP bundle from the /download endpoint and extract text files."""
|
||
import io
|
||
import zipfile
|
||
|
||
files: Dict[str, str] = {}
|
||
max_retries = 3
|
||
for attempt in range(max_retries):
|
||
try:
|
||
resp = httpx.get(
|
||
f"{self.BASE_URL}/download",
|
||
params={"slug": slug, "version": version},
|
||
timeout=30,
|
||
follow_redirects=True,
|
||
)
|
||
if resp.status_code == 429:
|
||
try:
|
||
retry_after = int(resp.headers.get("retry-after", "5"))
|
||
except (ValueError, TypeError):
|
||
retry_after = 5
|
||
retry_after = min(retry_after, 15) # Cap wait time
|
||
logger.debug(
|
||
"ClawHub download rate-limited for %s, retrying in %ds (attempt %d/%d)",
|
||
slug, retry_after, attempt + 1, max_retries,
|
||
)
|
||
time.sleep(retry_after)
|
||
continue
|
||
if resp.status_code != 200:
|
||
logger.debug("ClawHub ZIP download for %s v%s returned %s", slug, version, resp.status_code)
|
||
return files
|
||
|
||
with zipfile.ZipFile(io.BytesIO(resp.content)) as zf:
|
||
for info in zf.infolist():
|
||
if info.is_dir():
|
||
continue
|
||
try:
|
||
name = _validate_bundle_rel_path(info.filename)
|
||
except ValueError:
|
||
logger.debug("Skipping unsafe ZIP member path: %s", info.filename)
|
||
continue
|
||
# Only extract text-sized files (skip large binaries)
|
||
if info.file_size > 500_000:
|
||
logger.debug("Skipping large file in ZIP: %s (%d bytes)", name, info.file_size)
|
||
continue
|
||
try:
|
||
raw = zf.read(info.filename)
|
||
files[name] = raw.decode("utf-8")
|
||
except (UnicodeDecodeError, KeyError):
|
||
logger.debug("Skipping non-text file in ZIP: %s", name)
|
||
continue
|
||
|
||
return files
|
||
|
||
except zipfile.BadZipFile:
|
||
logger.warning("ClawHub returned invalid ZIP for %s v%s", slug, version)
|
||
return files
|
||
except httpx.HTTPError as exc:
|
||
logger.debug("ClawHub ZIP download failed for %s v%s: %s", slug, version, exc)
|
||
return files
|
||
|
||
logger.debug("ClawHub ZIP download exhausted retries for %s v%s", slug, version)
|
||
return files
|
||
|
||
def _fetch_text(self, url: str) -> Optional[str]:
|
||
resp = _guarded_http_get(url, timeout=20)
|
||
if resp is not None and resp.status_code == 200:
|
||
return resp.text
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# LobeHub source adapter
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class LobeHubSource(SkillSource):
|
||
"""
|
||
Fetch skills from LobeHub's agent marketplace (14,500+ agents).
|
||
LobeHub agents are system prompt templates — we convert them to SKILL.md on fetch.
|
||
Data lives in GitHub: lobehub/lobe-chat-agents.
|
||
"""
|
||
|
||
INDEX_URL = "https://chat-agents.lobehub.com/index.json"
|
||
|
||
def source_id(self) -> str:
|
||
return "lobehub"
|
||
|
||
def trust_level_for(self, identifier: str) -> str:
|
||
return "community"
|
||
|
||
def search(self, query: str, limit: int = 10) -> List[SkillMeta]:
|
||
index = self._fetch_index()
|
||
if not index:
|
||
return []
|
||
|
||
query_lower = query.lower()
|
||
results: List[SkillMeta] = []
|
||
|
||
agents = index.get("agents", index) if isinstance(index, dict) else index
|
||
if not isinstance(agents, list):
|
||
return []
|
||
|
||
for agent in agents:
|
||
meta = agent.get("meta", agent)
|
||
title = meta.get("title", agent.get("identifier", ""))
|
||
desc = meta.get("description", "")
|
||
tags = meta.get("tags", [])
|
||
|
||
searchable = f"{title} {desc} {' '.join(tags) if isinstance(tags, list) else ''}".lower()
|
||
if query_lower in searchable:
|
||
identifier = agent.get("identifier", title.lower().replace(" ", "-"))
|
||
results.append(SkillMeta(
|
||
name=identifier,
|
||
description=desc[:200],
|
||
source="lobehub",
|
||
identifier=f"lobehub/{identifier}",
|
||
trust_level="community",
|
||
tags=tags if isinstance(tags, list) else [],
|
||
))
|
||
|
||
if len(results) >= limit:
|
||
break
|
||
|
||
return results
|
||
|
||
def fetch(self, identifier: str) -> Optional[SkillBundle]:
|
||
# Strip "lobehub/" prefix if present
|
||
agent_id = identifier.split("/", 1)[-1] if identifier.startswith("lobehub/") else identifier
|
||
|
||
agent_data = self._fetch_agent(agent_id)
|
||
if not agent_data:
|
||
return None
|
||
|
||
skill_md = self._convert_to_skill_md(agent_data)
|
||
return SkillBundle(
|
||
name=agent_id,
|
||
files={"SKILL.md": skill_md},
|
||
source="lobehub",
|
||
identifier=f"lobehub/{agent_id}",
|
||
trust_level="community",
|
||
)
|
||
|
||
def inspect(self, identifier: str) -> Optional[SkillMeta]:
|
||
agent_id = identifier.split("/", 1)[-1] if identifier.startswith("lobehub/") else identifier
|
||
index = self._fetch_index()
|
||
if not index:
|
||
return None
|
||
|
||
agents = index.get("agents", index) if isinstance(index, dict) else index
|
||
if not isinstance(agents, list):
|
||
return None
|
||
|
||
for agent in agents:
|
||
if agent.get("identifier") == agent_id:
|
||
meta = agent.get("meta", agent)
|
||
return SkillMeta(
|
||
name=agent_id,
|
||
description=meta.get("description", ""),
|
||
source="lobehub",
|
||
identifier=f"lobehub/{agent_id}",
|
||
trust_level="community",
|
||
tags=meta.get("tags", []) if isinstance(meta.get("tags"), list) else [],
|
||
)
|
||
return None
|
||
|
||
def _fetch_index(self) -> Optional[Any]:
|
||
"""Fetch the LobeHub agent index (cached for 1 hour)."""
|
||
cache_key = "lobehub_index"
|
||
cached = _read_index_cache(cache_key)
|
||
if cached is not None:
|
||
return cached
|
||
|
||
try:
|
||
resp = httpx.get(self.INDEX_URL, timeout=30)
|
||
if resp.status_code != 200:
|
||
return None
|
||
data = resp.json()
|
||
except (httpx.HTTPError, json.JSONDecodeError):
|
||
return None
|
||
|
||
_write_index_cache(cache_key, data)
|
||
return data
|
||
|
||
def _fetch_agent(self, agent_id: str) -> Optional[dict]:
|
||
"""Fetch a single agent's JSON file."""
|
||
url = f"https://chat-agents.lobehub.com/{agent_id}.json"
|
||
try:
|
||
resp = httpx.get(url, timeout=15)
|
||
if resp.status_code == 200:
|
||
return resp.json()
|
||
except (httpx.HTTPError, json.JSONDecodeError) as e:
|
||
logger.debug("LobeHub agent fetch failed: %s", e)
|
||
return None
|
||
|
||
@staticmethod
|
||
def _convert_to_skill_md(agent_data: dict) -> str:
|
||
"""Convert a LobeHub agent JSON into SKILL.md format."""
|
||
meta = agent_data.get("meta", agent_data)
|
||
identifier = agent_data.get("identifier", "lobehub-agent")
|
||
title = meta.get("title", identifier)
|
||
description = meta.get("description", "")
|
||
tags = meta.get("tags", [])
|
||
system_role = agent_data.get("config", {}).get("systemRole", "")
|
||
|
||
tag_list = tags if isinstance(tags, list) else []
|
||
fm_lines = [
|
||
"---",
|
||
f"name: {identifier}",
|
||
f"description: {description[:500]}",
|
||
"metadata:",
|
||
" hermes:",
|
||
f" tags: [{', '.join(str(t) for t in tag_list)}]",
|
||
" lobehub:",
|
||
" source: lobehub",
|
||
"---",
|
||
]
|
||
|
||
body_lines = [
|
||
f"# {title}",
|
||
"",
|
||
description,
|
||
"",
|
||
"## Instructions",
|
||
"",
|
||
system_role if system_role else "(No system role defined)",
|
||
]
|
||
|
||
return "\n".join(fm_lines) + "\n\n" + "\n".join(body_lines) + "\n"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# browse.sh source adapter
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class BrowseShSource(SkillSource):
|
||
"""Discover and install site-specific browser automation skills from browse.sh.
|
||
|
||
browse.sh (https://browse.sh) is Browserbase's catalog of 200+ SKILL.md files
|
||
that describe how to automate specific websites (Airbnb, Amazon, arXiv, etc.).
|
||
The catalog lives at ``/api/skills`` and each skill's actual SKILL.md content
|
||
is fetched via ``/api/skills/{slug}`` which returns a ``skillMdUrl`` field
|
||
pointing at a CDN-hosted blob — the catalog's ``sourceUrl`` field is a GitHub
|
||
HTML URL whose underlying repository is not always public, so it cannot be
|
||
relied on for content fetch.
|
||
"""
|
||
|
||
CATALOG_URL = "https://browse.sh/api/skills"
|
||
SKILL_DETAIL_URL = "https://browse.sh/api/skills/{slug}"
|
||
_CACHE_KEY = "browse_sh_catalog"
|
||
|
||
def source_id(self) -> str:
|
||
return "browse-sh"
|
||
|
||
def trust_level_for(self, identifier: str) -> str:
|
||
return "community"
|
||
|
||
def _fetch_catalog(self) -> List[Dict]:
|
||
cached = _read_index_cache(self._CACHE_KEY)
|
||
if cached is not None:
|
||
return cached
|
||
try:
|
||
resp = httpx.get(self.CATALOG_URL, timeout=20)
|
||
if resp.status_code != 200:
|
||
return []
|
||
data = resp.json()
|
||
except (httpx.HTTPError, json.JSONDecodeError):
|
||
return []
|
||
skills = data.get("skills", []) if isinstance(data, dict) else []
|
||
if isinstance(skills, list):
|
||
_write_index_cache(self._CACHE_KEY, skills)
|
||
return skills if isinstance(skills, list) else []
|
||
|
||
def _item_to_meta(self, item: Dict) -> Optional[SkillMeta]:
|
||
slug = item.get("slug", "")
|
||
name = item.get("name", "")
|
||
title = item.get("title", name)
|
||
description = item.get("description", title)
|
||
if not slug or not name:
|
||
return None
|
||
if len(description) > 1024:
|
||
description = description[:1021] + "..."
|
||
return SkillMeta(
|
||
name=name,
|
||
description=description,
|
||
source="browse-sh",
|
||
identifier=f"browse-sh/{slug}",
|
||
trust_level="community",
|
||
tags=item.get("tags", []),
|
||
extra={
|
||
"slug": slug,
|
||
"hostname": item.get("hostname", ""),
|
||
"category": item.get("category", ""),
|
||
"source_url": item.get("sourceUrl", ""),
|
||
"recommended_method": item.get("recommendedMethod", ""),
|
||
"proxies": item.get("proxies", False),
|
||
"install_count": item.get("installCount", 0),
|
||
},
|
||
)
|
||
|
||
def search(self, query: str, limit: int = 10) -> List[SkillMeta]:
|
||
catalog = self._fetch_catalog()
|
||
query_lower = query.lower()
|
||
results = []
|
||
for item in catalog:
|
||
text = " ".join([
|
||
item.get("name", ""),
|
||
item.get("title", ""),
|
||
item.get("description", ""),
|
||
item.get("hostname", ""),
|
||
item.get("category", ""),
|
||
" ".join(item.get("tags", [])),
|
||
]).lower()
|
||
if not query_lower or query_lower in text:
|
||
meta = self._item_to_meta(item)
|
||
if meta:
|
||
results.append(meta)
|
||
if len(results) >= limit:
|
||
break
|
||
return results
|
||
|
||
def inspect(self, identifier: str) -> Optional[SkillMeta]:
|
||
slug = self._slug_from_identifier(identifier)
|
||
if not slug:
|
||
return None
|
||
catalog = self._fetch_catalog()
|
||
for item in catalog:
|
||
if item.get("slug") == slug:
|
||
return self._item_to_meta(item)
|
||
return None
|
||
|
||
def fetch(self, identifier: str) -> Optional[SkillBundle]:
|
||
slug = self._slug_from_identifier(identifier)
|
||
if not slug:
|
||
return None
|
||
catalog = self._fetch_catalog()
|
||
item = next((i for i in catalog if i.get("slug") == slug), None)
|
||
if not item:
|
||
return None
|
||
|
||
# Resolve the actual SKILL.md content URL via the per-skill detail
|
||
# endpoint, which returns a ``skillMdUrl`` (CDN blob). The catalog's
|
||
# ``sourceUrl`` is a GitHub HTML link whose underlying repo is not
|
||
# reliably public, so we don't use it for content.
|
||
md_url = self._resolve_skill_md_url(slug, item)
|
||
if not md_url:
|
||
return None
|
||
try:
|
||
resp = httpx.get(md_url, timeout=20, follow_redirects=True)
|
||
if resp.status_code != 200:
|
||
return None
|
||
content = resp.text
|
||
except httpx.HTTPError:
|
||
return None
|
||
|
||
meta = self._item_to_meta(item)
|
||
name = meta.name if meta else slug.split("/")[-1]
|
||
return SkillBundle(
|
||
name=name,
|
||
files={"SKILL.md": content},
|
||
source="browse-sh",
|
||
identifier=identifier,
|
||
trust_level="community",
|
||
metadata={
|
||
"slug": slug,
|
||
"hostname": item.get("hostname", ""),
|
||
"source_url": item.get("sourceUrl", ""),
|
||
"skill_md_url": md_url,
|
||
},
|
||
)
|
||
|
||
def _resolve_skill_md_url(self, slug: str, item: Dict) -> Optional[str]:
|
||
"""Resolve the SKILL.md content URL for a slug.
|
||
|
||
Primary path: hit ``/api/skills/{slug}`` and read ``skillMdUrl``.
|
||
Fallback: if the catalog item already has a ``raw.githubusercontent.com``
|
||
``sourceUrl`` (some entries may), use it directly.
|
||
"""
|
||
try:
|
||
detail = httpx.get(
|
||
self.SKILL_DETAIL_URL.format(slug=slug),
|
||
timeout=20,
|
||
follow_redirects=True,
|
||
)
|
||
if detail.status_code == 200:
|
||
data = detail.json()
|
||
if isinstance(data, dict):
|
||
md_url = data.get("skillMdUrl")
|
||
if isinstance(md_url, str) and md_url.startswith("http"):
|
||
return md_url
|
||
except (httpx.HTTPError, json.JSONDecodeError):
|
||
pass
|
||
|
||
source_url = item.get("sourceUrl", "") if isinstance(item, dict) else ""
|
||
from utils import base_url_host_matches
|
||
if source_url and base_url_host_matches(source_url, "raw.githubusercontent.com"):
|
||
return source_url
|
||
return None
|
||
|
||
def _slug_from_identifier(self, identifier: str) -> str:
|
||
"""Extract slug from identifier like 'browse-sh/airbnb.com/search-listings-abc'."""
|
||
if identifier.startswith("browse-sh/"):
|
||
return identifier[len("browse-sh/"):]
|
||
return identifier
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Official optional skills source adapter
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class OptionalSkillSource(SkillSource):
|
||
"""
|
||
Fetch skills from the optional-skills/ directory shipped with the repo.
|
||
|
||
These skills are official (maintained by Nous Research) but not activated
|
||
by default — they don't appear in the system prompt and aren't copied to
|
||
~/.hermes/skills/ during setup. They are discoverable via the Skills Hub
|
||
(search / install / inspect) and labelled "official" with "builtin" trust.
|
||
"""
|
||
|
||
OFFICIAL_REPO = "NousResearch/hermes-agent"
|
||
OPTIONAL_SKILLS_PREFIX = "optional-skills"
|
||
|
||
def __init__(self, auth: Optional[GitHubAuth] = None):
|
||
from hermes_constants import get_optional_skills_dir
|
||
|
||
self._optional_dir = get_optional_skills_dir(
|
||
Path(__file__).parent.parent / "optional-skills"
|
||
)
|
||
self._auth = auth
|
||
# Lazily created GitHubSource for the live-repo fallback — only
|
||
# instantiated when a skill is missing from the local checkout.
|
||
self._github: Optional[GitHubSource] = None
|
||
# rel_path ("category/skill") -> True, from the live repo tree.
|
||
# None = not fetched yet this process.
|
||
self._remote_dirs: Optional[Dict[str, bool]] = None
|
||
|
||
def source_id(self) -> str:
|
||
return "official"
|
||
|
||
def trust_level_for(self, identifier: str) -> str:
|
||
return "builtin"
|
||
|
||
# -- search -----------------------------------------------------------
|
||
|
||
def search(self, query: str, limit: int = 10) -> List[SkillMeta]:
|
||
results: List[SkillMeta] = []
|
||
query_lower = query.lower()
|
||
|
||
local_rels: set = set()
|
||
for meta in self._scan_all():
|
||
rel = meta.identifier.split("/", 1)[-1] if meta.identifier else ""
|
||
local_rels.add(rel)
|
||
searchable = f"{meta.name} {meta.description} {' '.join(meta.tags)}".lower()
|
||
if query_lower in searchable:
|
||
results.append(meta)
|
||
if len(results) >= limit:
|
||
break
|
||
|
||
# Also surface skills that landed on live main after this install was
|
||
# cut (missing from the local optional-skills/ checkout).
|
||
if len(results) < limit:
|
||
for rel_dir in sorted(self._list_remote_skill_dirs()):
|
||
if rel_dir in local_rels:
|
||
continue
|
||
name = rel_dir.rsplit("/", 1)[-1]
|
||
if query_lower and query_lower not in rel_dir.lower():
|
||
continue
|
||
results.append(SkillMeta(
|
||
name=name,
|
||
description="Official optional skill (from live repo; run install to fetch)",
|
||
source="official",
|
||
identifier=f"official/{rel_dir}",
|
||
trust_level="builtin",
|
||
repo=self.OFFICIAL_REPO,
|
||
path=f"{self.OPTIONAL_SKILLS_PREFIX}/{rel_dir}",
|
||
tags=[],
|
||
))
|
||
if len(results) >= limit:
|
||
break
|
||
|
||
return results
|
||
|
||
# -- fetch ------------------------------------------------------------
|
||
|
||
def fetch(self, identifier: str) -> Optional[SkillBundle]:
|
||
# identifier format: "official/category/skill" or "official/skill"
|
||
rel = identifier.split("/", 1)[-1] if identifier.startswith("official/") else identifier
|
||
skill_dir = self._optional_dir / rel
|
||
|
||
# Guard against path traversal (e.g. "official/../../etc")
|
||
try:
|
||
resolved = skill_dir.resolve()
|
||
optional_root = self._optional_dir.resolve()
|
||
if not resolved.is_relative_to(optional_root):
|
||
return None
|
||
except (OSError, ValueError):
|
||
return None
|
||
|
||
if not resolved.is_dir():
|
||
# Try searching by skill name only (last segment)
|
||
skill_name = rel.rsplit("/", 1)[-1]
|
||
skill_dir = self._find_skill_dir(skill_name)
|
||
if not skill_dir:
|
||
# Not in the local checkout — the skill may have landed on
|
||
# main after this install was cut. Fall back to the live repo.
|
||
return self._fetch_from_live_repo(rel)
|
||
else:
|
||
skill_dir = resolved
|
||
|
||
# Upstream-maintained entries: the local dir is a catalog stub whose
|
||
# frontmatter points at the real skill in an external repo the
|
||
# upstream project maintains (e.g. impeccable's Hermes-native bundle
|
||
# under pbakaus/impeccable:.hermes/skills/impeccable). Install pulls
|
||
# the live content from there instead of vendoring a fork here.
|
||
upstream = self._upstream_pointer(skill_dir)
|
||
if upstream is not None:
|
||
rel_id = skill_dir.resolve().relative_to(self._optional_dir.resolve()).as_posix()
|
||
return self._fetch_from_upstream(upstream, rel_id)
|
||
|
||
files: Dict[str, Union[str, bytes]] = {}
|
||
for f in skill_dir.rglob("*"):
|
||
if (
|
||
f.is_file()
|
||
and not f.name.startswith(".")
|
||
and "__pycache__" not in f.parts
|
||
and f.suffix != ".pyc"
|
||
):
|
||
rel_path = str(f.relative_to(skill_dir))
|
||
try:
|
||
files[rel_path] = f.read_bytes()
|
||
except OSError:
|
||
continue
|
||
|
||
if not files:
|
||
return None
|
||
|
||
# Determine category from directory structure
|
||
name = skill_dir.name
|
||
|
||
return SkillBundle(
|
||
name=name,
|
||
files=files,
|
||
source="official",
|
||
identifier=f"official/{skill_dir.resolve().relative_to(self._optional_dir.resolve()).as_posix()}",
|
||
trust_level="builtin",
|
||
)
|
||
|
||
# -- inspect ----------------------------------------------------------
|
||
|
||
def inspect(self, identifier: str) -> Optional[SkillMeta]:
|
||
rel = identifier.split("/", 1)[-1] if identifier.startswith("official/") else identifier
|
||
skill_name = rel.rsplit("/", 1)[-1]
|
||
|
||
for meta in self._scan_all():
|
||
if meta.name == skill_name:
|
||
return meta
|
||
|
||
# Not in the local checkout — check live main.
|
||
remote_dirs = self._list_remote_skill_dirs()
|
||
matches = [d for d in remote_dirs if d.rsplit("/", 1)[-1] == skill_name]
|
||
if len(matches) == 1:
|
||
rel_dir = matches[0]
|
||
return SkillMeta(
|
||
name=skill_name,
|
||
description="Official optional skill (from live repo; run install to fetch)",
|
||
source="official",
|
||
identifier=f"official/{rel_dir}",
|
||
trust_level="builtin",
|
||
repo=self.OFFICIAL_REPO,
|
||
path=f"{self.OPTIONAL_SKILLS_PREFIX}/{rel_dir}",
|
||
tags=[],
|
||
)
|
||
return None
|
||
|
||
# -- catalog ----------------------------------------------------------
|
||
|
||
def list_local(self) -> List[SkillMeta]:
|
||
"""Public catalog enumeration: every optional skill in the local
|
||
checkout, with frontmatter metadata. Backs the dashboard/desktop
|
||
"built-in optional skills" catalog surface."""
|
||
return self._scan_all()
|
||
|
||
# -- internal helpers -------------------------------------------------
|
||
|
||
def _get_github(self) -> "GitHubSource":
|
||
if self._github is None:
|
||
self._github = GitHubSource(auth=self._auth or GitHubAuth())
|
||
return self._github
|
||
|
||
def _fetch_from_live_repo(self, rel: str) -> Optional[SkillBundle]:
|
||
"""Fetch an optional skill straight from the live repo on GitHub.
|
||
|
||
Local installs lag `main` — a freshly merged optional skill isn't in
|
||
the user's `optional-skills/` checkout until they run
|
||
``hermes update``. Rather than telling them to update first, resolve
|
||
the skill against the live default branch.
|
||
|
||
``rel`` is the identifier without the ``official/`` prefix — either
|
||
``category/skill`` (used verbatim) or a bare skill name (located via
|
||
the repo tree).
|
||
"""
|
||
rel = rel.strip("/")
|
||
if not rel:
|
||
return None
|
||
# Reject traversal before it ever becomes a GitHub path.
|
||
parts = [p for p in rel.split("/") if p not in ("", ".")]
|
||
if not parts or any(p == ".." for p in parts):
|
||
return None
|
||
rel = "/".join(parts)
|
||
|
||
github = self._get_github()
|
||
remote_dirs = self._list_remote_skill_dirs()
|
||
|
||
if rel in remote_dirs:
|
||
repo_path = f"{self.OPTIONAL_SKILLS_PREFIX}/{rel}"
|
||
else:
|
||
# Bare name (or stale category) — locate by final path segment.
|
||
name = parts[-1]
|
||
matches = [d for d in remote_dirs if d.rsplit("/", 1)[-1] == name]
|
||
if len(matches) != 1:
|
||
return None
|
||
repo_path = f"{self.OPTIONAL_SKILLS_PREFIX}/{matches[0]}"
|
||
rel = matches[0]
|
||
|
||
# Download the FULL skill directory (byte-exact, including root-level
|
||
# install scripts, LICENSE, tests/). GitHubSource.fetch() would only
|
||
# pull SKILL.md + referenced support dirs, silently dropping files the
|
||
# local-checkout path preserves.
|
||
tree = github._get_repo_tree(self.OFFICIAL_REPO)
|
||
if tree is None:
|
||
return None
|
||
_branch, entries = tree
|
||
prefix = f"{repo_path}/"
|
||
files: Dict[str, Union[str, bytes]] = {}
|
||
for item in entries:
|
||
if item.get("type") != "blob" or item.get("mode") == "120000":
|
||
continue
|
||
item_path = item.get("path", "")
|
||
if not item_path.startswith(prefix):
|
||
continue
|
||
rel_file = item_path[len(prefix):]
|
||
base = rel_file.rsplit("/", 1)[-1]
|
||
if base.startswith(".") or base.endswith(".pyc") or "__pycache__" in rel_file.split("/"):
|
||
continue
|
||
content = github._fetch_file_bytes(self.OFFICIAL_REPO, item_path)
|
||
if content is None:
|
||
logger.warning("Live-repo optional skill fetch failed for %s", item_path)
|
||
return None
|
||
files[rel_file] = content
|
||
|
||
if "SKILL.md" not in files:
|
||
return None
|
||
|
||
# Live-fetched catalog stubs redirect the same way local ones do.
|
||
upstream = self._upstream_pointer_from_content(files["SKILL.md"])
|
||
if upstream is not None:
|
||
return self._fetch_from_upstream(upstream, rel)
|
||
|
||
logger.info("Optional skill '%s' fetched from live repo (not in local checkout)", rel)
|
||
return SkillBundle(
|
||
name=rel.rsplit("/", 1)[-1],
|
||
files=files,
|
||
source="official",
|
||
identifier=f"official/{rel}",
|
||
trust_level="builtin",
|
||
)
|
||
|
||
def _list_remote_skill_dirs(self) -> Dict[str, bool]:
|
||
"""Map of ``category/skill`` dirs under optional-skills/ on live main.
|
||
|
||
Derived from the repo tree (single API call, cached per-process by
|
||
GitHubSource, plus the shared on-disk index cache). Returns {} when
|
||
the network/API is unavailable — callers degrade to local-only.
|
||
"""
|
||
if self._remote_dirs is not None:
|
||
return self._remote_dirs
|
||
|
||
cache_key = "official_optional_dirs"
|
||
cached = _read_index_cache(cache_key)
|
||
if isinstance(cached, dict) and cached:
|
||
self._remote_dirs = cached
|
||
return cached
|
||
|
||
dirs: Dict[str, bool] = {}
|
||
tree = self._get_github()._get_repo_tree(self.OFFICIAL_REPO)
|
||
if tree is not None:
|
||
_branch, entries = tree
|
||
prefix = f"{self.OPTIONAL_SKILLS_PREFIX}/"
|
||
suffix = "/SKILL.md"
|
||
for item in entries:
|
||
path = item.get("path", "")
|
||
if (
|
||
item.get("type") == "blob"
|
||
and path.startswith(prefix)
|
||
and path.endswith(suffix)
|
||
):
|
||
rel_dir = path[len(prefix):-len(suffix)]
|
||
if rel_dir and not is_excluded_skill_path(
|
||
PurePosixPath(rel_dir + suffix)
|
||
):
|
||
dirs[rel_dir] = True
|
||
if dirs:
|
||
_write_index_cache(cache_key, dirs)
|
||
|
||
self._remote_dirs = dirs
|
||
return dirs
|
||
|
||
def _upstream_pointer(self, skill_dir: Path) -> Optional[Dict[str, str]]:
|
||
"""Return the upstream pointer for a catalog-stub skill dir, if any.
|
||
|
||
A stub declares ``metadata.hermes.upstream`` in its SKILL.md
|
||
frontmatter:
|
||
|
||
metadata:
|
||
hermes:
|
||
upstream:
|
||
repo: pbakaus/impeccable
|
||
path: .hermes/skills/impeccable
|
||
|
||
Returns ``{"repo": ..., "path": ...}`` or None for normal
|
||
(fully-vendored) optional skills.
|
||
"""
|
||
skill_md = skill_dir / "SKILL.md"
|
||
try:
|
||
content = skill_md.read_text(encoding="utf-8")
|
||
except (OSError, UnicodeDecodeError):
|
||
return None
|
||
return self._upstream_pointer_from_content(content)
|
||
|
||
def _upstream_pointer_from_content(self, content: Union[str, bytes]) -> Optional[Dict[str, str]]:
|
||
"""Parse ``metadata.hermes.upstream`` out of SKILL.md content."""
|
||
if isinstance(content, bytes):
|
||
try:
|
||
content = content.decode("utf-8")
|
||
except UnicodeDecodeError:
|
||
return None
|
||
fm = self._parse_frontmatter(content)
|
||
meta_block = fm.get("metadata")
|
||
if not isinstance(meta_block, dict):
|
||
return None
|
||
hermes_meta = meta_block.get("hermes")
|
||
if not isinstance(hermes_meta, dict):
|
||
return None
|
||
upstream = hermes_meta.get("upstream")
|
||
if not isinstance(upstream, dict):
|
||
return None
|
||
repo = str(upstream.get("repo", "")).strip().strip("/")
|
||
path = str(upstream.get("path", "")).strip().strip("/")
|
||
# repo must be exactly owner/name; path must be a clean relative path.
|
||
if not repo or repo.count("/") != 1 or not path:
|
||
return None
|
||
parts = [p for p in path.split("/") if p not in ("", ".")]
|
||
if not parts or any(p == ".." for p in parts):
|
||
return None
|
||
return {"repo": repo, "path": "/".join(parts)}
|
||
|
||
def _fetch_from_upstream(self, upstream: Dict[str, str], rel_id: str) -> Optional[SkillBundle]:
|
||
"""Fetch an upstream-maintained optional skill from its external repo.
|
||
|
||
Delegates to GitHubSource.fetch() (full-directory download through the
|
||
git tree, symlink/unsafe-path rejection, quarantine + scan downstream)
|
||
but re-labels the bundle as an official catalog entry so trust,
|
||
identifier, and update tracking stay in the optional-skills namespace.
|
||
"""
|
||
github = self._get_github()
|
||
bundle = github.fetch(f"{upstream['repo']}/{upstream['path']}")
|
||
if bundle is None:
|
||
logger.warning(
|
||
"Upstream fetch failed for optional skill %s (%s:%s)",
|
||
rel_id, upstream["repo"], upstream["path"],
|
||
)
|
||
return None
|
||
return SkillBundle(
|
||
name=bundle.name,
|
||
files=bundle.files,
|
||
source="official",
|
||
identifier=f"official/{rel_id}",
|
||
# Curated-catalog endorsement, but the content comes live from a
|
||
# third-party repo — "trusted", not "builtin", so a dangerous
|
||
# scan verdict still blocks install (INSTALL_POLICY).
|
||
trust_level="trusted",
|
||
metadata={
|
||
**bundle.metadata,
|
||
"upstream_repo": upstream["repo"],
|
||
"upstream_path": upstream["path"],
|
||
},
|
||
)
|
||
|
||
def _find_skill_dir(self, name: str) -> Optional[Path]:
|
||
"""Find a skill directory by name anywhere in optional-skills/."""
|
||
if not self._optional_dir.is_dir():
|
||
return None
|
||
for skill_md in self._optional_dir.rglob("SKILL.md"):
|
||
if is_excluded_skill_path(
|
||
skill_md.relative_to(self._optional_dir), root=self._optional_dir
|
||
):
|
||
continue
|
||
if skill_md.parent.name == name:
|
||
return skill_md.parent
|
||
return None
|
||
|
||
def _scan_all(self) -> List[SkillMeta]:
|
||
"""Enumerate all optional skills with metadata."""
|
||
if not self._optional_dir.is_dir():
|
||
return []
|
||
|
||
results: List[SkillMeta] = []
|
||
for skill_md in sorted(self._optional_dir.rglob("SKILL.md")):
|
||
if is_excluded_skill_path(
|
||
skill_md.relative_to(self._optional_dir), root=self._optional_dir
|
||
):
|
||
continue
|
||
parent = skill_md.parent
|
||
|
||
try:
|
||
content = skill_md.read_text(encoding="utf-8")
|
||
except (OSError, UnicodeDecodeError):
|
||
continue
|
||
|
||
fm = self._parse_frontmatter(content)
|
||
name = fm.get("name", parent.name)
|
||
desc = fm.get("description", "")
|
||
tags = []
|
||
meta_block = fm.get("metadata", {})
|
||
if isinstance(meta_block, dict):
|
||
hermes_meta = meta_block.get("hermes", {})
|
||
if isinstance(hermes_meta, dict):
|
||
tags = hermes_meta.get("tags", [])
|
||
|
||
rel_path = parent.relative_to(self._optional_dir).as_posix()
|
||
|
||
results.append(SkillMeta(
|
||
name=name,
|
||
description=desc[:200],
|
||
source="official",
|
||
identifier=f"official/{rel_path}",
|
||
trust_level="builtin",
|
||
repo=self.OFFICIAL_REPO,
|
||
# The centralized skills index consumes repo-root-relative paths.
|
||
path=f"optional-skills/{rel_path}",
|
||
tags=tags if isinstance(tags, list) else [],
|
||
))
|
||
|
||
return results
|
||
|
||
@staticmethod
|
||
def _parse_frontmatter(content: str) -> dict:
|
||
"""Parse YAML frontmatter from SKILL.md content."""
|
||
content = content.lstrip("\ufeff") # tolerate UTF-8 BOM (Windows editors)
|
||
if not content.startswith("---"):
|
||
return {}
|
||
match = re.search(r'\n---\s*\n', content[3:])
|
||
if not match:
|
||
return {}
|
||
yaml_text = content[3:match.start() + 3]
|
||
try:
|
||
parsed = yaml.safe_load(yaml_text)
|
||
return parsed if isinstance(parsed, dict) else {}
|
||
except yaml.YAMLError:
|
||
return {}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Shared cache helpers (used by multiple adapters)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _read_index_cache(key: str) -> Optional[Any]:
|
||
"""Read cached data if not expired."""
|
||
cache_file = _index_cache_dir() / f"{key}.json"
|
||
if not cache_file.exists():
|
||
return None
|
||
try:
|
||
stat = cache_file.stat()
|
||
if time.time() - stat.st_mtime > INDEX_CACHE_TTL:
|
||
return None
|
||
return json.loads(cache_file.read_text(encoding="utf-8"))
|
||
except (OSError, json.JSONDecodeError):
|
||
return None
|
||
|
||
|
||
def _write_index_cache(key: str, data: Any) -> None:
|
||
"""Write data to cache."""
|
||
index_cache_dir = _index_cache_dir()
|
||
index_cache_dir.mkdir(parents=True, exist_ok=True)
|
||
# Ensure .ignore exists so ripgrep (and tools respecting .ignore) skip
|
||
# this directory. Cache files contain unvetted community content that
|
||
# could include adversarial text (prompt injection via catalog entries).
|
||
ignore_file = _hub_dir() / ".ignore"
|
||
if not ignore_file.exists():
|
||
try:
|
||
ignore_file.write_text("# Exclude hub internals from search tools\n*\n", encoding="utf-8")
|
||
except OSError:
|
||
pass
|
||
cache_file = index_cache_dir / f"{key}.json"
|
||
try:
|
||
cache_file.write_text(json.dumps(data, ensure_ascii=False, default=str), encoding="utf-8")
|
||
except OSError as e:
|
||
logger.debug("Could not write cache: %s", e)
|
||
|
||
|
||
def _skill_meta_to_dict(meta: SkillMeta) -> dict:
|
||
"""Convert a SkillMeta to a dict for caching."""
|
||
return {
|
||
"name": meta.name,
|
||
"description": meta.description,
|
||
"source": meta.source,
|
||
"identifier": meta.identifier,
|
||
"trust_level": meta.trust_level,
|
||
"repo": meta.repo,
|
||
"path": meta.path,
|
||
"tags": meta.tags,
|
||
"extra": meta.extra,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Lock file management
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class HubLockFile:
|
||
"""Manages skills/.hub/lock.json — tracks provenance of installed hub skills."""
|
||
|
||
def __init__(self, path: Optional[Path] = None):
|
||
self.path = path if path is not None else _lock_file()
|
||
|
||
def load(self) -> dict:
|
||
if not self.path.exists():
|
||
return {"version": 1, "installed": {}}
|
||
try:
|
||
return json.loads(self.path.read_text(encoding="utf-8"))
|
||
except (json.JSONDecodeError, OSError):
|
||
return {"version": 1, "installed": {}}
|
||
|
||
def save(self, data: dict) -> None:
|
||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||
self.path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
|
||
|
||
def record_install(
|
||
self,
|
||
name: str,
|
||
source: str,
|
||
identifier: str,
|
||
trust_level: str,
|
||
scan_verdict: str,
|
||
skill_hash: str,
|
||
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
|
||
# for the uninstall_skill rmtree-escape; reject malformed input at
|
||
# write time so the file never carries the bad state.
|
||
safe_name = _validate_skill_name(name)
|
||
safe_install_path = _normalize_lock_install_path(install_path, safe_name)
|
||
data = self.load()
|
||
data["installed"][safe_name] = {
|
||
"source": source,
|
||
"identifier": identifier,
|
||
"trust_level": trust_level,
|
||
"scan_verdict": scan_verdict,
|
||
"content_hash": skill_hash,
|
||
"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(),
|
||
}
|
||
self.save(data)
|
||
|
||
def record_uninstall(self, name: str) -> None:
|
||
data = self.load()
|
||
data["installed"].pop(name, None)
|
||
self.save(data)
|
||
|
||
def get_installed(self, name: str) -> Optional[dict]:
|
||
data = self.load()
|
||
return data["installed"].get(name)
|
||
|
||
def list_installed(self) -> List[dict]:
|
||
data = self.load()
|
||
result = []
|
||
for name, entry in data["installed"].items():
|
||
result.append({"name": name, **entry})
|
||
return result
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Taps management
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TapsManager:
|
||
"""Manages the taps.json file — custom GitHub repo sources."""
|
||
|
||
def __init__(self, path: Optional[Path] = None):
|
||
self.path = path if path is not None else _taps_file()
|
||
|
||
def load(self) -> List[dict]:
|
||
if not self.path.exists():
|
||
return []
|
||
try:
|
||
data = json.loads(self.path.read_text(encoding="utf-8"))
|
||
return data.get("taps", [])
|
||
except (json.JSONDecodeError, OSError):
|
||
return []
|
||
|
||
def save(self, taps: List[dict]) -> None:
|
||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||
self.path.write_text(json.dumps({"taps": taps}, indent=2) + "\n", encoding="utf-8")
|
||
|
||
def add(self, repo: str, path: str = "skills/") -> bool:
|
||
"""Add a tap. Returns False if already exists."""
|
||
taps = self.load()
|
||
if any(t["repo"] == repo for t in taps):
|
||
return False
|
||
taps.append({"repo": repo, "path": path})
|
||
self.save(taps)
|
||
return True
|
||
|
||
def remove(self, repo: str) -> bool:
|
||
"""Remove a tap by repo name. Returns False if not found."""
|
||
taps = self.load()
|
||
new_taps = [t for t in taps if t["repo"] != repo]
|
||
if len(new_taps) == len(taps):
|
||
return False
|
||
self.save(new_taps)
|
||
return True
|
||
|
||
def list_taps(self) -> List[dict]:
|
||
return self.load()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Audit log
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def append_audit_log(action: str, skill_name: str, source: str,
|
||
trust_level: str, verdict: str, extra: str = "") -> None:
|
||
"""Append a line to the audit log."""
|
||
audit_log = _audit_log()
|
||
audit_log.parent.mkdir(parents=True, exist_ok=True)
|
||
timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||
parts = [timestamp, action, skill_name, f"{source}:{trust_level}", verdict]
|
||
if extra:
|
||
parts.append(extra)
|
||
line = " ".join(parts) + "\n"
|
||
try:
|
||
with open(audit_log, "a", encoding="utf-8") as f:
|
||
f.write(line)
|
||
except OSError as e:
|
||
logger.debug("Could not write audit log: %s", e)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Hub operations (high-level)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def ensure_hub_dirs() -> None:
|
||
"""Create the .hub directory structure if it doesn't exist."""
|
||
hub_dir = _hub_dir()
|
||
lock_file = _lock_file()
|
||
audit_log = _audit_log()
|
||
taps_file = _taps_file()
|
||
hub_dir.mkdir(parents=True, exist_ok=True)
|
||
_quarantine_dir().mkdir(exist_ok=True)
|
||
_index_cache_dir().mkdir(exist_ok=True)
|
||
if not lock_file.exists():
|
||
lock_file.write_text('{"version": 1, "installed": {}}\n', encoding="utf-8")
|
||
if not audit_log.exists():
|
||
audit_log.touch()
|
||
if not taps_file.exists():
|
||
taps_file.write_text('{"taps": []}\n', encoding="utf-8")
|
||
|
||
|
||
def quarantine_bundle(bundle: SkillBundle) -> Path:
|
||
"""Write a skill bundle to the quarantine directory for scanning."""
|
||
ensure_hub_dirs()
|
||
skill_name = _validate_skill_name(bundle.name)
|
||
validated_files: List[Tuple[str, Union[str, bytes]]] = []
|
||
for rel_path, file_content in bundle.files.items():
|
||
safe_rel_path = _validate_bundle_rel_path(rel_path)
|
||
validated_files.append((safe_rel_path, file_content))
|
||
|
||
dest = _quarantine_dir() / skill_name
|
||
if dest.exists():
|
||
shutil.rmtree(dest)
|
||
dest.mkdir(parents=True)
|
||
|
||
for rel_path, file_content in validated_files:
|
||
file_dest = dest.joinpath(*rel_path.split("/"))
|
||
file_dest.parent.mkdir(parents=True, exist_ok=True)
|
||
if isinstance(file_content, bytes):
|
||
file_dest.write_bytes(file_content)
|
||
else:
|
||
file_dest.write_text(file_content, encoding="utf-8")
|
||
|
||
return dest
|
||
|
||
|
||
def _category_skill_dirs(directory: Path) -> List[str]:
|
||
"""Names of direct children of *directory* that contain skills.
|
||
|
||
A child counts when it is a non-hidden directory holding at least one
|
||
active ``SKILL.md`` anywhere below it (recursive, so nested category
|
||
layouts like ``mlops/training/<skill>`` are detected). Vendored,
|
||
cache, and progressive-disclosure support paths are pruned via
|
||
:func:`is_excluded_skill_path` so a lone ``node_modules`` or
|
||
``references/pkg/SKILL.md`` hit does not misclassify the directory as
|
||
a category. Shared by the install-time category guard here and
|
||
``hermes_cli.skills_hub._existing_categories``.
|
||
"""
|
||
skill_dirs: List[str] = []
|
||
for entry in directory.iterdir():
|
||
if not entry.is_dir() or entry.name.startswith("."):
|
||
continue
|
||
for skill_md in entry.rglob("SKILL.md"):
|
||
if is_excluded_skill_path(
|
||
skill_md.relative_to(directory), root=directory
|
||
):
|
||
continue
|
||
skill_dirs.append(entry.name)
|
||
break
|
||
return skill_dirs
|
||
|
||
|
||
def install_from_quarantine(
|
||
quarantine_path: Path,
|
||
skill_name: str,
|
||
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)
|
||
safe_category = _validate_install_parent_path(category) if category else ""
|
||
quarantine_resolved = quarantine_path.resolve()
|
||
quarantine_root = _quarantine_dir().resolve()
|
||
if not quarantine_resolved.is_relative_to(quarantine_root):
|
||
raise ValueError(f"Unsafe quarantine path: {quarantine_path}")
|
||
|
||
if safe_category:
|
||
install_rel_path = f"{safe_category}/{safe_skill_name}"
|
||
else:
|
||
install_rel_path = safe_skill_name
|
||
|
||
# Resolve via the same lock-path validator the uninstaller uses. Catches
|
||
# symlink-in-skills-tree redirects at install time so the lock entry's
|
||
# path can never refer to a redirected target.
|
||
install_dir = _resolve_lock_install_path(install_rel_path, safe_skill_name)
|
||
|
||
# Refuse to nest a skill inside an existing skill directory. Installing
|
||
# with ``--category <name-of-an-existing-skill>`` would create a hybrid
|
||
# skill-plus-category directory; a later update or uninstall of the outer
|
||
# skill would then rmtree the inner one — the sibling case of the
|
||
# category-bucket wipe reported in issue #75983.
|
||
skills_root = _skills_dir().resolve()
|
||
ancestor = install_dir.parent
|
||
while ancestor != skills_root and ancestor.is_relative_to(skills_root):
|
||
if (ancestor / "SKILL.md").is_file():
|
||
raise ValueError(
|
||
f"Refusing to install into '{ancestor.name}': it is an "
|
||
f"existing skill directory, not a category. Choose a "
|
||
f"different category."
|
||
)
|
||
ancestor = ancestor.parent
|
||
|
||
if install_dir.exists():
|
||
if not install_dir.is_dir():
|
||
# A stray regular file at the install path. rmtree() on a file
|
||
# raises NotADirectoryError (an uncaught traceback at the CLI);
|
||
# refuse with the same actionable ValueError contract instead.
|
||
raise ValueError(
|
||
f"Refusing to install: '{install_dir.name}' already exists "
|
||
f"and is not a directory. Remove it or choose a different "
|
||
f"skill name."
|
||
)
|
||
# Guard against silent data loss when the install target collides with
|
||
# an existing category bucket (a directory that holds other skills).
|
||
# This was reported as GitHub issue #75983: installing a skill with
|
||
# --name matching an existing category directory caused rmtree to wipe
|
||
# all sibling skills. A directory that directly contains SKILL.md is
|
||
# an existing skill installation and stays overwritable (hub-installed
|
||
# skills are additionally guarded by the lock-file check in
|
||
# do_install()). But a directory that contains *other* skill
|
||
# directories is a category bucket and must NOT be silently deleted.
|
||
if not (install_dir / "SKILL.md").exists():
|
||
skill_dirs_in = _category_skill_dirs(install_dir)
|
||
if skill_dirs_in:
|
||
raise ValueError(
|
||
f"Refusing to overwrite category directory '{install_dir}' "
|
||
f"which contains {len(skill_dirs_in)} skill(s): "
|
||
f"{', '.join(sorted(skill_dirs_in))}. "
|
||
f"Use a different --name or install into a subcategory."
|
||
)
|
||
shutil.rmtree(install_dir)
|
||
|
||
# Warn (but don't block) if SKILL.md is very large
|
||
skill_md = quarantine_path / "SKILL.md"
|
||
if skill_md.exists():
|
||
try:
|
||
skill_size = skill_md.stat().st_size
|
||
if skill_size > 100_000:
|
||
logger.warning(
|
||
"Skill '%s' has a large SKILL.md (%s chars). "
|
||
"Large skills consume significant context when loaded. "
|
||
"Consider asking the author to split it into smaller files.",
|
||
safe_skill_name,
|
||
f"{skill_size:,}",
|
||
)
|
||
except OSError:
|
||
pass
|
||
|
||
# Reject symlinks inside the quarantined skill before moving it.
|
||
# A malicious skill bundle could include a symlink pointing outside the
|
||
# skills tree; its target contents would then be copied into skills/ and
|
||
# leaked to the agent on the next skill_view call.
|
||
for entry in quarantine_path.rglob("*"):
|
||
if not _is_path_redirect(entry):
|
||
continue
|
||
try:
|
||
rel = entry.relative_to(quarantine_resolved)
|
||
except ValueError:
|
||
rel = entry
|
||
raise ValueError(
|
||
f"Installed skill contains symlinks, which is not allowed: {rel}"
|
||
)
|
||
|
||
install_dir.parent.mkdir(parents=True, exist_ok=True)
|
||
shutil.move(str(quarantine_path), str(install_dir))
|
||
|
||
# Record in lock file
|
||
lock = HubLockFile()
|
||
lock.record_install(
|
||
name=safe_skill_name,
|
||
source=bundle.source,
|
||
identifier=bundle.identifier,
|
||
trust_level=bundle.trust_level,
|
||
scan_verdict=scan_result.verdict,
|
||
skill_hash=content_hash(install_dir),
|
||
install_path=install_dir.resolve().relative_to(_skills_dir().resolve()).as_posix(),
|
||
files=list(bundle.files.keys()),
|
||
metadata=bundle.metadata,
|
||
scan_provenance=scan_provenance or getattr(scan_result, "scan_provenance", None),
|
||
)
|
||
|
||
append_audit_log(
|
||
"INSTALL", safe_skill_name, bundle.source,
|
||
bundle.trust_level, scan_result.verdict,
|
||
content_hash(install_dir),
|
||
)
|
||
|
||
try:
|
||
from tools.skill_usage import record_installed
|
||
|
||
record_installed(safe_skill_name)
|
||
except Exception:
|
||
logger.debug(
|
||
"Unable to record skill install lifecycle for %s",
|
||
safe_skill_name,
|
||
exc_info=True,
|
||
)
|
||
|
||
return install_dir
|
||
|
||
|
||
def uninstall_skill(skill_name: str) -> Tuple[bool, str]:
|
||
"""Remove a hub-installed skill. Refuses to remove builtins."""
|
||
lock = HubLockFile()
|
||
entry = lock.get_installed(skill_name)
|
||
if not entry:
|
||
return False, f"'{skill_name}' is not a hub-installed skill (may be a builtin)"
|
||
|
||
# Validate the lock entry's install_path against the skill name. This is
|
||
# the destructive boundary — anything that falls through to the rmtree
|
||
# below MUST be inside SKILLS_DIR and MUST NOT be SKILLS_DIR itself
|
||
# (an empty/"."/"/" install_path would otherwise wipe the entire tree).
|
||
# _resolve_lock_install_path enforces a relative path ending in
|
||
# <skill_name>, rejects absolute/traversal paths, and walks the path
|
||
# component-by-component refusing symlink/junction redirects.
|
||
try:
|
||
install_path = _resolve_lock_install_path(
|
||
entry.get("install_path", ""), skill_name
|
||
)
|
||
except ValueError as exc:
|
||
return False, f"Refusing to uninstall '{skill_name}': {exc}"
|
||
|
||
if install_path.exists():
|
||
shutil.rmtree(install_path)
|
||
|
||
lock.record_uninstall(skill_name)
|
||
append_audit_log("UNINSTALL", skill_name, entry["source"], entry["trust_level"], "n/a", "user_request")
|
||
|
||
return True, f"Uninstalled '{skill_name}' from {entry['install_path']}"
|
||
|
||
|
||
def bundle_content_hash(bundle: SkillBundle) -> str:
|
||
"""Compute a deterministic hash for an in-memory skill bundle.
|
||
|
||
MUST stay symmetric with ``tools.skills_guard.content_hash`` (which
|
||
hashes the same skill from disk). That function keys files by
|
||
``relative_to(...).as_posix()`` — forward slashes on every OS. Bundle
|
||
keys built on Windows carry backslashes (``str(f.relative_to(dir))``),
|
||
which changed both the hashed bytes AND the sort order, so every
|
||
installed skill reported ``update_available`` forever on Windows
|
||
(#62310). Normalize to POSIX separators before sorting/hashing.
|
||
"""
|
||
h = hashlib.sha256()
|
||
normalized = {
|
||
rel_path.replace("\\", "/"): content
|
||
for rel_path, content in bundle.files.items()
|
||
}
|
||
for rel_path in sorted(normalized):
|
||
# Include the path so swapping file contents between two paths
|
||
# changes the hash (avoids filename-swap evading update detection).
|
||
h.update(rel_path.encode("utf-8"))
|
||
h.update(b"\x00")
|
||
content = normalized[rel_path]
|
||
if isinstance(content, bytes):
|
||
h.update(content)
|
||
else:
|
||
h.update(content.encode("utf-8"))
|
||
return f"sha256:{h.hexdigest()[:16]}"
|
||
|
||
|
||
def _source_matches(source: SkillSource, source_name: str) -> bool:
|
||
aliases = {
|
||
"skills.sh": "skills-sh",
|
||
}
|
||
normalized = aliases.get(source_name, source_name)
|
||
return source.source_id() == normalized
|
||
|
||
|
||
def check_for_skill_updates(
|
||
name: Optional[str] = None,
|
||
*,
|
||
lock: Optional[HubLockFile] = None,
|
||
sources: Optional[List[SkillSource]] = None,
|
||
auth: Optional[GitHubAuth] = None,
|
||
) -> List[dict]:
|
||
"""Check installed hub skills for upstream changes."""
|
||
lock = lock or HubLockFile()
|
||
installed = lock.list_installed()
|
||
if name:
|
||
installed = [entry for entry in installed if entry.get("name") == name]
|
||
|
||
if sources is None:
|
||
sources = create_source_router(auth=auth)
|
||
|
||
results: List[dict] = []
|
||
for entry in installed:
|
||
identifier = entry.get("identifier", "")
|
||
source_name = entry.get("source", "")
|
||
candidate_sources = [src for src in sources if _source_matches(src, source_name)]
|
||
if not candidate_sources:
|
||
# No adapter for the recorded source (e.g. a tap was removed, or the
|
||
# source was renamed upstream). Previously this fell back to *all*
|
||
# sources, which meant a same-named skill in a DIFFERENT registry
|
||
# could satisfy the fetch and be reported as an update for this
|
||
# entry -- silently reassigning provenance. Skill names are not
|
||
# namespaced across registries, so that fallback is unsafe by
|
||
# construction. Report unavailable instead and let the user decide.
|
||
results.append({
|
||
"name": entry.get("name", ""),
|
||
"identifier": identifier,
|
||
"source": source_name,
|
||
"status": "unavailable",
|
||
})
|
||
continue
|
||
|
||
bundle = None
|
||
for src in candidate_sources:
|
||
try:
|
||
bundle = src.fetch(identifier)
|
||
except Exception:
|
||
bundle = None
|
||
if bundle:
|
||
break
|
||
|
||
if not bundle:
|
||
results.append({
|
||
"name": entry.get("name", ""),
|
||
"identifier": identifier,
|
||
"source": source_name,
|
||
"status": "unavailable",
|
||
})
|
||
continue
|
||
|
||
current_hash = entry.get("content_hash", "")
|
||
latest_hash = bundle_content_hash(bundle)
|
||
status = "up_to_date" if current_hash == latest_hash else "update_available"
|
||
results.append({
|
||
"name": entry.get("name", ""),
|
||
"identifier": identifier,
|
||
"source": source_name,
|
||
"status": status,
|
||
"current_hash": current_hash,
|
||
"latest_hash": latest_hash,
|
||
"bundle": bundle,
|
||
})
|
||
|
||
return results
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Hermes centralized index source
|
||
# ---------------------------------------------------------------------------
|
||
|
||
HERMES_INDEX_URL = "https://hermes-agent.nousresearch.com/docs/api/skills-index.json"
|
||
HERMES_INDEX_TTL = 6 * 3600 # 6 hours
|
||
|
||
|
||
def _hermes_index_cache_file() -> Path:
|
||
return _index_cache_dir() / "hermes-index.json"
|
||
|
||
|
||
def _load_hermes_index() -> Optional[dict]:
|
||
"""Fetch the centralized skills index, with local cache.
|
||
|
||
The index is a JSON file hosted on the docs site, rebuilt daily by CI.
|
||
We cache it locally for HERMES_INDEX_TTL seconds to avoid repeated
|
||
downloads within a session.
|
||
"""
|
||
# Check local cache
|
||
hermes_index_cache_file = _hermes_index_cache_file()
|
||
if hermes_index_cache_file.exists():
|
||
try:
|
||
age = time.time() - hermes_index_cache_file.stat().st_mtime
|
||
if age < HERMES_INDEX_TTL:
|
||
return json.loads(hermes_index_cache_file.read_text(encoding="utf-8"))
|
||
except (OSError, json.JSONDecodeError):
|
||
pass
|
||
|
||
# Fetch from docs site.
|
||
#
|
||
# We deliberately DON'T let httpx negotiate Brotli here. The index is a
|
||
# large body (tens of MB); httpx's streaming Brotli decoder, backed by
|
||
# brotlicffi 1.2.0.1 (pinned for Discord attachment decoding), trips over
|
||
# its own output_buffer_limit on payloads this size and raises
|
||
# DecodingError("brotli: decoder process called with data when
|
||
# 'can_accept_more_data()' is False"). That surfaces as an empty Skills
|
||
# Hub (blank Browse-hub landing, index contributes 0 search hits) because
|
||
# the error is caught below and we silently fall back to a (often absent)
|
||
# stale cache. Requesting gzip/deflate sidesteps the broken decoder while
|
||
# still compressing the transfer. The identity retry is belt-and-braces
|
||
# for any future proxy that ignores the header and returns Brotli anyway.
|
||
data = None
|
||
for accept_encoding in ("gzip, deflate", "identity"):
|
||
try:
|
||
resp = httpx.get(
|
||
HERMES_INDEX_URL,
|
||
timeout=15,
|
||
follow_redirects=True,
|
||
headers={"Accept-Encoding": accept_encoding},
|
||
)
|
||
if resp.status_code != 200:
|
||
logger.debug("Hermes index fetch returned %d", resp.status_code)
|
||
return _load_stale_index_cache()
|
||
data = resp.json()
|
||
break
|
||
except httpx.DecodingError as e:
|
||
# Content-Encoding decode failed — retry once uncompressed before
|
||
# giving up on the network path entirely.
|
||
logger.debug(
|
||
"Hermes index decode failed (Accept-Encoding=%s): %s",
|
||
accept_encoding,
|
||
e,
|
||
)
|
||
continue
|
||
except (httpx.HTTPError, json.JSONDecodeError) as e:
|
||
logger.debug("Hermes index fetch failed: %s", e)
|
||
return _load_stale_index_cache()
|
||
|
||
if data is None:
|
||
return _load_stale_index_cache()
|
||
|
||
# Validate structure
|
||
if not isinstance(data, dict) or "skills" not in data:
|
||
return _load_stale_index_cache()
|
||
|
||
# Cache locally
|
||
try:
|
||
hermes_index_cache_file.parent.mkdir(parents=True, exist_ok=True)
|
||
hermes_index_cache_file.write_text(json.dumps(data), encoding="utf-8")
|
||
except OSError:
|
||
pass
|
||
|
||
return data
|
||
|
||
|
||
def _load_stale_index_cache() -> Optional[dict]:
|
||
"""Fall back to stale cache when the network fetch fails."""
|
||
hermes_index_cache_file = _hermes_index_cache_file()
|
||
if hermes_index_cache_file.exists():
|
||
try:
|
||
return json.loads(hermes_index_cache_file.read_text(encoding="utf-8"))
|
||
except (OSError, json.JSONDecodeError):
|
||
pass
|
||
return None
|
||
|
||
|
||
class HermesIndexSource(SkillSource):
|
||
"""Skill source backed by the centralized Hermes Skills Index.
|
||
|
||
The index is a JSON catalog published to the docs site and rebuilt
|
||
daily by CI. It contains metadata + resolved GitHub paths for every
|
||
skill, eliminating the need for users to hit the GitHub API for
|
||
search or path discovery.
|
||
|
||
When the index is unavailable, all methods return empty / None so
|
||
downstream sources take over transparently.
|
||
"""
|
||
|
||
def __init__(self, auth: GitHubAuth):
|
||
self._index: Optional[dict] = None
|
||
self._loaded = False
|
||
self.auth = auth
|
||
# Lazily create GitHubSource for fetch — only used when actually
|
||
# downloading files, which requires real GitHub API calls.
|
||
self._github: Optional[GitHubSource] = None
|
||
|
||
def _ensure_loaded(self) -> dict:
|
||
if not self._loaded:
|
||
self._index = _load_hermes_index()
|
||
self._loaded = True
|
||
return self._index or {}
|
||
|
||
def _get_github(self) -> GitHubSource:
|
||
if self._github is None:
|
||
self._github = GitHubSource(auth=self.auth)
|
||
return self._github
|
||
|
||
def source_id(self) -> str:
|
||
return "hermes-index"
|
||
|
||
@property
|
||
def is_available(self) -> bool:
|
||
"""Whether the index is loaded and has skills."""
|
||
index = self._ensure_loaded()
|
||
return bool(index.get("skills"))
|
||
|
||
def trust_level_for(self, identifier: str) -> str:
|
||
index = self._ensure_loaded()
|
||
for skill in index.get("skills", []):
|
||
if skill.get("identifier") == identifier:
|
||
return skill.get("trust_level", "community")
|
||
return "community"
|
||
|
||
def search(self, query: str, limit: int = 10) -> List[SkillMeta]:
|
||
"""Search the cached index. Zero API calls.
|
||
|
||
Matches against name, description, tags, identifier, and the per-tap
|
||
``extra.provider`` label (so a query like ``nvidia`` surfaces the
|
||
``NVIDIA/skills/...`` entries even though their ``source`` is the bare
|
||
``github``). Results are scored and ranked (exact name > name prefix >
|
||
whole-word > substring) rather than returned in raw index order and
|
||
truncated at the first ``limit`` hits — that earlier break-at-limit
|
||
behaviour returned an arbitrary file-order slice and buried the most
|
||
relevant skills.
|
||
"""
|
||
index = self._ensure_loaded()
|
||
skills = index.get("skills", [])
|
||
if not skills:
|
||
return []
|
||
|
||
if not query.strip():
|
||
# No query — return featured/popular (index order)
|
||
return [self._to_meta(s) for s in skills[:limit]]
|
||
|
||
query_lower = query.lower()
|
||
scored: List[Tuple[int, int, dict]] = []
|
||
for i, s in enumerate(skills):
|
||
name = str(s.get("name", "")).lower()
|
||
provider = str((s.get("extra") or {}).get("provider", "")).lower()
|
||
haystack = " ".join([
|
||
name,
|
||
str(s.get("description", "")).lower(),
|
||
" ".join(str(t).lower() for t in s.get("tags", [])),
|
||
str(s.get("identifier", "")).lower(),
|
||
provider,
|
||
])
|
||
if query_lower not in haystack:
|
||
continue
|
||
# Lower score sorts first.
|
||
if name == query_lower:
|
||
score = 0
|
||
elif name.startswith(query_lower):
|
||
score = 1
|
||
elif provider == query_lower:
|
||
score = 2
|
||
elif query_lower in name.split() or query_lower in provider.split():
|
||
score = 3
|
||
elif query_lower in name:
|
||
score = 4
|
||
else:
|
||
score = 5
|
||
# i (original index order) is the stable tiebreaker.
|
||
scored.append((score, i, s))
|
||
|
||
scored.sort(key=lambda x: (x[0], x[1]))
|
||
return [self._to_meta(s) for _, _, s in scored[:limit]]
|
||
|
||
def fetch(self, identifier: str) -> Optional[SkillBundle]:
|
||
"""Fetch a skill using the resolved path from the index.
|
||
|
||
If the index has a ``resolved_github_id`` for this skill, we skip
|
||
the entire candidate/discovery chain and go directly to GitHub
|
||
with the exact path. This reduces install from ~31 API calls to
|
||
just the file content downloads (~5-22 depending on skill size).
|
||
"""
|
||
index = self._ensure_loaded()
|
||
entry = self._find_entry(identifier, index)
|
||
if not entry:
|
||
return None
|
||
|
||
# Use resolved path if available
|
||
resolved = entry.get("resolved_github_id")
|
||
if resolved:
|
||
bundle = self._get_github().fetch(resolved)
|
||
if bundle:
|
||
bundle.source = entry.get("source", "hermes-index")
|
||
bundle.identifier = identifier
|
||
return bundle
|
||
|
||
# Fall back to identifier-based fetch via repo/path
|
||
repo = entry.get("repo", "")
|
||
path = entry.get("path", "")
|
||
if repo and path:
|
||
github_id = f"{repo}/{path}"
|
||
bundle = self._get_github().fetch(github_id)
|
||
if bundle:
|
||
bundle.source = entry.get("source", "hermes-index")
|
||
bundle.identifier = identifier
|
||
return bundle
|
||
|
||
return None
|
||
|
||
def inspect(self, identifier: str) -> Optional[SkillMeta]:
|
||
"""Return metadata from the index. Zero API calls."""
|
||
index = self._ensure_loaded()
|
||
entry = self._find_entry(identifier, index)
|
||
if entry:
|
||
return self._to_meta(entry)
|
||
return None
|
||
|
||
def _find_entry(self, identifier: str, index: dict) -> Optional[dict]:
|
||
"""Look up a skill in the index by identifier or name."""
|
||
skills = index.get("skills", [])
|
||
|
||
# Exact identifier match
|
||
for s in skills:
|
||
if s.get("identifier") == identifier:
|
||
return s
|
||
|
||
# Try without source prefix (e.g. "skills-sh/" stripped)
|
||
normalized = identifier
|
||
for prefix in ("skills-sh/", "skills.sh/", "official/", "github/", "clawhub/"):
|
||
if identifier.startswith(prefix):
|
||
normalized = identifier[len(prefix):]
|
||
break
|
||
|
||
# Match on normalized identifier or name
|
||
for s in skills:
|
||
sid = s.get("identifier", "")
|
||
# Strip prefix from stored identifier too
|
||
stored_normalized = sid
|
||
for prefix in ("skills-sh/", "skills.sh/", "official/", "github/", "clawhub/"):
|
||
if sid.startswith(prefix):
|
||
stored_normalized = sid[len(prefix):]
|
||
break
|
||
if stored_normalized == normalized:
|
||
return s
|
||
|
||
return None
|
||
|
||
@staticmethod
|
||
def _to_meta(entry: dict) -> SkillMeta:
|
||
return SkillMeta(
|
||
name=entry.get("name", ""),
|
||
description=entry.get("description", ""),
|
||
source=entry.get("source", "hermes-index"),
|
||
identifier=entry.get("identifier", ""),
|
||
trust_level=entry.get("trust_level", "community"),
|
||
repo=entry.get("repo"),
|
||
path=entry.get("path"),
|
||
tags=entry.get("tags", []),
|
||
extra=entry.get("extra", {}),
|
||
)
|
||
|
||
|
||
def create_source_router(auth: Optional[GitHubAuth] = None) -> List[SkillSource]:
|
||
"""
|
||
Create all configured source adapters.
|
||
Returns a list of active sources for search/fetch operations.
|
||
"""
|
||
if auth is None:
|
||
auth = GitHubAuth()
|
||
|
||
taps_mgr = TapsManager()
|
||
extra_taps = taps_mgr.list_taps()
|
||
|
||
sources: List[SkillSource] = [
|
||
OptionalSkillSource(auth=auth), # Official optional skills (highest priority)
|
||
HermesIndexSource(auth=auth), # Centralized index (search + resolved install paths)
|
||
SkillsShSource(auth=auth),
|
||
WellKnownSkillSource(),
|
||
UrlSource(), # Direct HTTP(S) URL to a SKILL.md file
|
||
GitHubSource(auth=auth, extra_taps=extra_taps),
|
||
ClawHubSource(),
|
||
LobeHubSource(),
|
||
BrowseShSource(), # browse.sh: 169+ site-specific browser automation skills
|
||
]
|
||
|
||
return sources
|
||
|
||
|
||
def _search_one_source(
|
||
src: SkillSource, query: str, limit: int
|
||
) -> Tuple[str, List[SkillMeta]]:
|
||
"""Search a single source. Runs in a thread for parallelism."""
|
||
try:
|
||
return src.source_id(), src.search(query, limit=limit)
|
||
except Exception as e:
|
||
logger.debug("Search failed for %s: %s", src.source_id(), e)
|
||
return src.source_id(), []
|
||
|
||
|
||
def parallel_search_sources(
|
||
sources: List[SkillSource],
|
||
query: str = "",
|
||
per_source_limits: Optional[Dict[str, int]] = None,
|
||
source_filter: str = "all",
|
||
overall_timeout: float = 30,
|
||
on_source_done: Optional[Any] = None,
|
||
) -> Tuple[List[SkillMeta], Dict[str, int], List[str]]:
|
||
"""Search all sources in parallel with per-source timeout.
|
||
|
||
Returns ``(all_results, source_counts, timed_out_ids)``.
|
||
|
||
*on_source_done* is an optional callback ``(source_id, count) -> None``
|
||
invoked as each source completes — useful for progress indicators.
|
||
"""
|
||
from concurrent.futures import as_completed
|
||
|
||
per_source_limits = per_source_limits or {}
|
||
|
||
# A provider filter (e.g. "nvidia", "openai") targets GitHub-tap skills
|
||
# that the runtime index stores under source="github" with an
|
||
# ``extra.provider`` label. It is NOT a real source id, so source-level
|
||
# selection must treat it like "all" (the index / github source carries
|
||
# the data); the per-provider narrowing happens downstream on the merged
|
||
# results (see ``_filter_results_by_provider``).
|
||
_provider_filter = source_filter.strip().lower() in _PROVIDER_FILTER_VALUES
|
||
_effective_filter = "all" if _provider_filter else source_filter
|
||
|
||
active: List[SkillSource] = []
|
||
# When the centralized index is available and the user hasn't filtered
|
||
# to a specific source, skip external API sources (github, skills-sh,
|
||
# clawhub, etc.) — the index already has their data. This avoids
|
||
# ~70 GitHub API calls per search for unauthenticated users.
|
||
_index_available = False
|
||
_api_source_ids = frozenset({"github", "skills-sh", "clawhub",
|
||
"lobehub", "well-known"})
|
||
if _effective_filter == "all":
|
||
for src in sources:
|
||
if (src.source_id() == "hermes-index"
|
||
and getattr(src, "is_available", False)):
|
||
_index_available = True
|
||
break
|
||
|
||
for src in sources:
|
||
sid = src.source_id()
|
||
if _effective_filter != "all" and sid != _effective_filter and sid != "official":
|
||
continue
|
||
# Skip external API sources when the index covers them
|
||
if _index_available and sid in _api_source_ids:
|
||
continue
|
||
active.append(src)
|
||
|
||
all_results: List[SkillMeta] = []
|
||
source_counts: Dict[str, int] = {}
|
||
timed_out_ids: List[str] = []
|
||
|
||
if not active:
|
||
return all_results, source_counts, timed_out_ids
|
||
|
||
# NOTE: a `with ThreadPoolExecutor(...) as pool` block calls
|
||
# ``shutdown(wait=True)`` on exit, which blocks until every submitted
|
||
# worker finishes — so a single slow source (e.g. ClawHub) keeps the
|
||
# caller blocked for minutes and renders ``overall_timeout`` a no-op.
|
||
# Manage the executor manually and shut it down with ``wait=False`` so
|
||
# the timeout is actually honoured. Daemon workers (tools.daemon_pool):
|
||
# an abandoned slow source must not block interpreter exit either —
|
||
# stdlib workers are joined unconditionally by the atexit hook.
|
||
from tools.daemon_pool import DaemonThreadPoolExecutor
|
||
pool = DaemonThreadPoolExecutor(max_workers=min(len(active), 8))
|
||
futures = {}
|
||
for src in active:
|
||
lim = per_source_limits.get(src.source_id(), 50)
|
||
fut = pool.submit(_search_one_source, src, query, lim)
|
||
futures[fut] = src.source_id()
|
||
|
||
try:
|
||
try:
|
||
for fut in as_completed(futures, timeout=overall_timeout):
|
||
try:
|
||
sid, results = fut.result(timeout=0)
|
||
source_counts[sid] = len(results)
|
||
all_results.extend(results)
|
||
if on_source_done:
|
||
on_source_done(sid, len(results))
|
||
except Exception:
|
||
pass
|
||
except TimeoutError:
|
||
timed_out_ids = [
|
||
futures[f] for f in futures if not f.done()
|
||
]
|
||
if timed_out_ids:
|
||
logger.debug(
|
||
"Skills browse timed out waiting for: %s",
|
||
", ".join(timed_out_ids),
|
||
)
|
||
finally:
|
||
# wait=False so a slow source cannot block the caller's return;
|
||
# cancel_futures drops not-yet-started work.
|
||
pool.shutdown(wait=False, cancel_futures=True)
|
||
|
||
return all_results, source_counts, timed_out_ids
|
||
|
||
|
||
def unified_search(query: str, sources: List[SkillSource],
|
||
source_filter: str = "all", limit: int = 10) -> List[SkillMeta]:
|
||
"""Search all sources (in parallel) and merge results."""
|
||
all_results, _, _ = parallel_search_sources(
|
||
sources,
|
||
query=query,
|
||
source_filter=source_filter,
|
||
overall_timeout=30,
|
||
)
|
||
|
||
# A provider filter (nvidia/openai/...) is applied here, on the merged set,
|
||
# because it targets the per-tap ``extra.provider`` label rather than a real
|
||
# source id (the runtime index stores every GitHub tap as source="github").
|
||
if source_filter.strip().lower() in _PROVIDER_FILTER_VALUES:
|
||
all_results = _filter_results_by_provider(all_results, source_filter)
|
||
|
||
# Deduplicate by identifier, preferring higher trust levels.
|
||
# identifier is always unique per skill (e.g. "browse-sh/airbnb.com/search-listings-ddgioa").
|
||
# Using name would incorrectly collapse browse-sh skills from different sites that share
|
||
# the same task name (e.g. "search-listings" from Airbnb and Booking.com).
|
||
_TRUST_RANK = {"builtin": 2, "trusted": 1, "community": 0}
|
||
seen: Dict[str, SkillMeta] = {}
|
||
for r in all_results:
|
||
if r.identifier not in seen:
|
||
seen[r.identifier] = r
|
||
elif _TRUST_RANK.get(r.trust_level, 0) > _TRUST_RANK.get(seen[r.identifier].trust_level, 0):
|
||
seen[r.identifier] = r
|
||
deduped = list(seen.values())
|
||
|
||
# Stable-sort by trust rank before truncating: the limit cut must not
|
||
# drop a builtin/official catalog entry because a high-volume community
|
||
# source (skills.sh mirrors every repo) happened to finish first and
|
||
# flood the merged list. Insertion order is preserved within each rank.
|
||
deduped.sort(key=lambda r: -_TRUST_RANK.get(r.trust_level, 0))
|
||
|
||
return deduped[:limit]
|