From c63aa49626134873baa76772df273f96f1d9dc32 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:49:56 -0700 Subject: [PATCH 1/2] refactor(tools): simplify skill_usage/skills_guard/ast_audit/skillevaluator/provenance; dead wrappers out, dispatch tables, compact docs --- tools/skill_provenance.py | 25 +- tools/skill_usage.py | 442 ++++++++++-------------------- tools/skillevaluator_scan.py | 117 +++----- tools/skills_ast_audit.py | 79 +++--- tools/skills_guard.py | 510 ++++++++++------------------------- 5 files changed, 347 insertions(+), 826 deletions(-) diff --git a/tools/skill_provenance.py b/tools/skill_provenance.py index c8333aca8b..0003fa63d6 100644 --- a/tools/skill_provenance.py +++ b/tools/skill_provenance.py @@ -1,14 +1,7 @@ -"""Skill write-origin provenance — a ContextVar separating agent-sediment -skill writes from foreground user-directed writes. - -The curator only consolidates/prunes skills it autonomously created via the -background self-improvement review fork; skills a user asked a foreground agent -to write belong to the user and must never be auto-curated. run_agent.py binds -the origin before each tool loop (mirroring AIAgent._memory_write_origin, which -is "background_review" for review-fork instances) so tool handlers such as -skill_manage create can check it. - -Usage:: +"""Skill write-origin provenance: a ContextVar separating agent-sediment skill writes from foreground +user-directed writes. The curator only curates skills the background self-improvement review fork created; +skills a user asked for belong to the user. run_agent.py binds the origin before each tool loop (mirroring +AIAgent._memory_write_origin) so handlers such as skill_manage create can check it:: token = set_current_write_origin("background_review") try: @@ -20,10 +13,7 @@ Usage:: import contextvars -_write_origin: contextvars.ContextVar[str] = contextvars.ContextVar( - "skill_write_origin", - default="foreground", -) +_write_origin: contextvars.ContextVar[str] = contextvars.ContextVar("skill_write_origin", default="foreground") # Sentinel used by the background review fork (run_agent._spawn_background_review). BACKGROUND_REVIEW = "background_review" @@ -35,16 +25,13 @@ def set_current_write_origin(origin: str) -> contextvars.Token[str]: def reset_current_write_origin(token: contextvars.Token[str]) -> None: - """Restore the prior write origin context.""" _write_origin.reset(token) def get_current_write_origin() -> str: - """Return the active write origin: "foreground" for any regular agent - (CLI, gateway, cron, subagent); "background_review" for the review fork.""" + """"foreground" for any regular agent (CLI, gateway, cron, subagent); "background_review" for the review fork.""" return _write_origin.get() def is_background_review() -> bool: - """True iff the current write origin is the background review fork.""" return get_current_write_origin() == BACKGROUND_REVIEW diff --git a/tools/skill_usage.py b/tools/skill_usage.py index 9e0bdcbaa4..fd58160bf4 100644 --- a/tools/skill_usage.py +++ b/tools/skill_usage.py @@ -1,19 +1,10 @@ -"""Skill usage telemetry + provenance tracking for the Curator feature. +"""Skill usage telemetry + provenance for the Curator: a sidecar ``~/.hermes/skills/.usage.json`` keyed by +skill name (never frontmatter — keeps telemetry out of user-authored SKILL.md and off bundled/hub skills). -Per-skill usage metadata lives in a sidecar JSON file (~/.hermes/skills/.usage.json) keyed by skill name. -Counters are bumped by the skill tools (skill_view, skill_manage); the curator orchestrator reads the derived -activity timestamp to decide lifecycle transitions. - -Design notes: sidecar, not frontmatter (keeps operational telemetry out of user-authored SKILL.md content and -avoids conflict pressure for bundled/hub skills); atomic writes via tempfile + os.replace (same pattern as -.bundled_manifest); all counter bumps are best-effort — failures log at DEBUG and return silently, so a broken -sidecar never breaks the underlying tool call; and curator-managed skills are explicitly marked when created -through skill_manage — bundled / hub-installed skills stay off-limits, and manually authored skills are not -inferred from location. - -Lifecycle states: active (default) -> stale (unused > stale_after_days, config) -> archived (unused > -archive_after_days, config; moved to .archive/). ``pinned`` is a boolean opt-out from auto transitions, -orthogonal to state. +Counter bumps are best-effort (failures log at DEBUG, never break the tool call); writes are atomic +(tempfile + os.replace) under a cross-process lock. Curator management is an explicit ``created_by: agent`` +marker written by skill_manage — never inferred from location. Lifecycle: active -> stale -> archived +(moved to .archive/); ``pinned`` opts out of auto transitions, orthogonal to state. """ from __future__ import annotations @@ -49,20 +40,15 @@ STATE_STALE = "stale" STATE_ARCHIVED = "archived" _VALID_STATES = {STATE_ACTIVE, STATE_STALE, STATE_ARCHIVED} -# Load-bearing bundled built-ins the curator must NEVER archive or consolidate, regardless of -# ``curator.prune_builtins``, pin state, or LLM judgment. These back advertised UX paths; silently archiving one -# turns its slash command into "Unknown command" with no signal to the user. Protection is by skill ``name`` -# (frontmatter ``name:``), matching the keys used throughout this module. Keep this list tiny and intentional — -# it is not a substitute for ``curator.prune_builtins: false``, which exempts ALL built-ins. (``plan`` used to -# live here; it is now a first-class built-in command with no skill on disk, so the set is currently empty.) +# Load-bearing built-ins (by frontmatter ``name``) the curator must NEVER archive or consolidate, regardless +# of ``curator.prune_builtins``, pin state, or LLM judgment — silently archiving one turns its slash command +# into "Unknown command". Keep tiny; it is not a substitute for ``curator.prune_builtins: false``. +# (``plan`` used to live here; it is now a first-class command with no skill on disk.) PROTECTED_BUILTIN_SKILLS: Set[str] = set() def is_protected_builtin(skill_name: str) -> bool: - """Whether *skill_name* is a load-bearing built-in the curator never touches. - - Protected built-ins are exempt from archival and consolidation on every path: the automatic state-transition walk, - the LLM consolidation pass (dropped from the candidate list), and direct ``archive_skill`` calls.""" + """Exempt from archival/consolidation on every path: auto transitions, LLM pass, direct archive_skill.""" return skill_name in PROTECTED_BUILTIN_SKILLS @@ -139,7 +125,6 @@ def _now_iso() -> str: def _parse_iso_timestamp(value: Any) -> Optional[datetime]: - """Parse an ISO timestamp defensively for activity comparisons.""" if not value: return None try: @@ -150,19 +135,15 @@ def _parse_iso_timestamp(value: Any) -> Optional[datetime]: def latest_activity_at(record: Dict[str, Any]) -> Optional[str]: - """Return the newest actual activity timestamp for a usage record. - - "Activity" means a skill was used, viewed, or patched. Creation time is intentionally excluded so callers can - still distinguish never-active skills; lifecycle code can fall back to ``created_at`` as its own anchor.""" - stamps = [ - (dt, str(raw)) for raw in (record.get(k) for k in ("last_used_at", "last_viewed_at", "last_patched_at")) - if (dt := _parse_iso_timestamp(raw)) is not None - ] + """Newest use/view/patch timestamp. Creation time is excluded so never-active skills stay + distinguishable; lifecycle code falls back to ``created_at`` itself.""" + stamps = [(dt, str(raw)) for raw in (record.get(k) for k in ("last_used_at", "last_viewed_at", "last_patched_at")) + if (dt := _parse_iso_timestamp(raw)) is not None] return max(stamps, key=lambda t: t[0])[1] if stamps else None def activity_count(record: Dict[str, Any]) -> int: - """Return the total observed activity count across use/view/patch events.""" + """Total observed use+view+patch events.""" total = 0 for key in ("use_count", "view_count", "patch_count"): try: @@ -174,24 +155,20 @@ def activity_count(record: Dict[str, Any]) -> int: # --- Provenance — which skills are agent-created (and thus eligible for curation) --- def _read_bundled_manifest_names() -> Set[str]: - """Skill names seeded from the bundled repo: ~/.hermes/skills/.bundled_manifest ("name:hash" per line). Empty set - if the file is missing or unreadable.""" + """Names from ``.bundled_manifest`` ("name:hash" per line); empty if missing/unreadable.""" lines = _read_lines(_skills_dir() / ".bundled_manifest", "Failed to read bundled manifest: %s") return {n for n in (line.split(":", 1)[0].strip() for line in lines) if n} def _read_hub_installed_names() -> Set[str]: - """Skill names installed via the Skills Hub, read from ~/.hermes/skills/.hub/lock.json (see tools/skills_hub.py :: - HubLockFile).""" + """Names installed via the Skills Hub (``.hub/lock.json``, see tools/skills_hub.py::HubLockFile), plus the + frontmatter name of each ``install_path`` that resolves inside the skills dir.""" lock_path = _skills_dir() / ".hub" / "lock.json" if not lock_path.exists(): return set() try: - # Tolerate non-UTF-8 bytes in the lock file. Hub descriptions can carry Windows-1252 typographic chars - # (em-dash 0x97, smart quotes, bullets) written as single high bytes; a strict utf-8 read raises - # UnicodeDecodeError, which is a ValueError sibling (not OSError/JSONDecodeError) so it escapes the - # handler below and 500s the whole /api/skills endpoint. errors="replace" degrades the offending byte to - # U+FFFD, keeping the (structurally valid) JSON — and every other skill — readable. See #68053. + # errors="replace": hub descriptions can carry Windows-1252 high bytes; a strict read raises + # UnicodeDecodeError (a ValueError, not caught below) and would 500 the whole /api/skills endpoint. data = json.loads(lock_path.read_text(encoding="utf-8", errors="replace")) installed = (data.get("installed") or {}) if isinstance(data, dict) else None if not isinstance(installed, dict): @@ -216,12 +193,8 @@ def _read_hub_installed_names() -> Set[str]: def _prune_builtins_enabled() -> bool: - """Whether bundled built-in skills are eligible for curator pruning. - - Reads ``curator.prune_builtins`` from config (default True). Lazy import keeps this module importable without the - CLI config layer (e.g. in the update/sync context); on any failure we fall back to the default. The real safety - against a mass-prune is the curator's seed-on-first-sight, not this flag — built-ins only archive after a fresh - inactivity window.""" + """``curator.prune_builtins`` (default True). Lazy config import keeps this module importable in the + update/sync context. The real mass-prune safety is seed-on-first-sight, not this flag.""" try: from hermes_cli.config import load_config cfg = load_config() @@ -234,15 +207,14 @@ def _prune_builtins_enabled() -> bool: def read_suppressed_names() -> Set[str]: - """Built-in skills the curator pruned — the re-seeder must leave archived. - - One skill name per line in ``~/.hermes/skills/.curator_suppressed``. This is what makes pruning a built-in - durable: without it, ``hermes update`` would re-copy the bundled skill on the next sync.""" + """Built-ins the curator pruned (one name per line in ``.curator_suppressed``); the update-time re-seeder + must leave these archived, otherwise ``hermes update`` would re-copy the bundled skill.""" lines = _read_lines(_skills_dir() / ".curator_suppressed", "Failed to read curator suppression list: %s") return {line for line in lines if not line.startswith("#")} def _toggle_suppressed_name(skill_name: str, *, add: bool) -> None: + """Add (built-in pruned) or drop (restored) *skill_name* in the suppression list; no-op when unchanged.""" if not skill_name or (skill_name in (names := read_suppressed_names())) == add: return (names.add if add else names.discard)(skill_name) @@ -253,72 +225,46 @@ def _toggle_suppressed_name(skill_name: str, *, add: bool) -> None: logger.debug("Failed to write curator suppression list: %s", e, exc_info=True) -def add_suppressed_name(skill_name: str) -> None: - """Record that a built-in skill was pruned, so sync won't restore it.""" - _toggle_suppressed_name(skill_name, add=True) - - -def remove_suppressed_name(skill_name: str) -> None: - """Clear a built-in's suppression entry (e.g. on restore).""" - _toggle_suppressed_name(skill_name, add=False) - - def _iter_skill_mds(base: Path, *, local_only: bool) -> Iterator[Tuple[str, Path]]: - """Yield ``(frontmatter name, SKILL.md)`` for every skill under *base* — the flat layout AND nested - category/skill/SKILL.md — skipping Hermes metadata, VCS, virtualenv/dependency, and cache dirs. With - *local_only*, external skill dirs mounted below the local tree are skipped too: discovery may see them, - but autonomous lifecycle curation must not.""" + """``(frontmatter name, SKILL.md)`` for flat and category-nested skills under *base*, skipping metadata/VCS/ + venv/cache dirs. *local_only* also skips external skill dirs mounted below the tree — discovery may see + them, autonomous curation must not.""" for skill_md in base.rglob("SKILL.md"): if not (is_excluded_skill_path(skill_md) or (local_only and is_external_skill_path(skill_md))): yield _read_skill_name(skill_md, fallback=skill_md.parent.name), skill_md def _scan_local_skills(keep: Callable[[str, Path, Set[str], Dict[str, Any]], bool]) -> List[str]: - """Sorted, de-duplicated names of local skills passing *keep(name, skill_md, bundled, usage)*. - - Hub-installed skills are always off-limits, and protected built-ins are never curation candidates - (exempt from the automatic transition walk AND the LLM consolidation pass), so neither ever reaches - *keep*.""" + """Sorted unique names of local skills passing *keep(name, skill_md, bundled, usage)*. Hub-installed and + protected built-ins never reach *keep*.""" base = _skills_dir() if not base.exists(): return [] hub, bundled, usage = _read_hub_installed_names(), _read_bundled_manifest_names(), load_usage() return sorted({ name for name, skill_md in _iter_skill_mds(base, local_only=True) - if name not in hub and not is_protected_builtin(name) and keep(name, skill_md, bundled, usage) - }) + if name not in hub and not is_protected_builtin(name) and keep(name, skill_md, bundled, usage)}) def list_agent_created_skill_names() -> List[str]: - """Enumerate skills the curator may manage. - - Always includes agent-authored skills (those marked in ``.usage.json`` via ``skill_manage(action="create")``). - When ``curator.prune_builtins`` is enabled, bundled built-in skills are ALSO included even though they have no - agent-created usage record — their inactivity clock is anchored on first sight (see - ``apply_automatic_transitions``). Hub-installed skills are never included; manually authored skills are not - inferred from filesystem location.""" - if not _skills_dir().exists(): - return [] + """Skills the curator may manage: agent-authored (``created_by: agent`` record) plus, when + ``curator.prune_builtins`` is on, bundled built-ins (inactivity anchored on first sight). Never hub skills.""" prune_builtins = _prune_builtins_enabled() # read once, before the walk def _keep(name: str, _skill_md: Path, bundled: Set[str], usage: Dict[str, Any]) -> bool: - # Built-ins are only candidates when pruning is enabled; they never carry a curator-managed record, so - # the record gate is skipped. Agent-authored (or local-manual) skills must opt in via their record. + # Built-ins never carry a curator-managed record, so the record gate applies only to local skills. return prune_builtins if name in bundled else _is_curator_managed_record(usage.get(name)) return _scan_local_skills(_keep) def list_archived_skill_names() -> List[str]: - """Enumerate skills in ``~/.hermes/skills/.archive/``. - - Archive layout is flat (``.archive//``) as set by ``archive_skill``, so the directory name is the skill - name. Used by ``hermes curator list-archived`` to help users pass a name to ``hermes curator restore``.""" + """Skills in ``.archive/`` — flat layout (``archive_skill`` flattens), so dir name == skill name.""" archive_root = _archive_dir() return sorted({p.name for p in archive_root.iterdir() if p.is_dir()}) if archive_root.exists() else [] def _read_skill_name(skill_md: Path, fallback: str) -> str: - """Parse the `name:` field from a SKILL.md YAML frontmatter.""" + """The frontmatter ``name:`` field of a SKILL.md (first 4000 chars), else *fallback*.""" try: text = skill_md.read_text(encoding="utf-8", errors="replace")[:4000] except OSError: @@ -337,19 +283,17 @@ def _read_skill_name(skill_md: Path, fallback: str) -> str: def is_agent_created(skill_name: str) -> bool: - """Whether *skill_name* is neither bundled nor hub-installed.""" + """Neither bundled nor hub-installed (and not only present in an external dir).""" if skill_name in _read_bundled_manifest_names() | _read_hub_installed_names(): return False return _find_skill_dir(skill_name) is not None or _find_external_skill_dir(skill_name) is None def is_hub_installed(skill_name: str) -> bool: - """Whether *skill_name* was installed via the Skills Hub.""" return skill_name in _read_hub_installed_names() def is_bundled(skill_name: str) -> bool: - """Whether *skill_name* was seeded from the bundled repo skills.""" return skill_name in _read_bundled_manifest_names() @@ -358,15 +302,11 @@ def _external_read_only_message(skill_name: str) -> str: def is_curation_eligible(skill_name: str, skill_path: Optional[Path] = None) -> bool: - """Whether the curator may track/archive *skill_name*. - - Agent-created skills are always eligible. Bundled built-ins become eligible only when ``curator.prune_builtins`` - is enabled. Hub-installed and external skill-dir skills are NEVER eligible — they have an external upstream owner. - Org-shared skills ARE eligible for improvement (the curator may patch them like any other skill; edits stay local - until proposed) but are protected from ARCHIVE/DELETE elsewhere — removing a shared skill is an org-admin action, - not a local curation decision. Protected built-ins (``PROTECTED_BUILTIN_SKILLS``) are NEVER eligible regardless of - any flag — they back load-bearing UX and must never be archived or consolidated.""" - if (skill_path is not None and is_external_skill_path(skill_path)) or is_protected_builtin(skill_name) or is_hub_installed(skill_name): + """May the curator track/archive this skill? Agent-created: yes. Bundled: only with + ``curator.prune_builtins``. Hub-installed / external-dir / protected built-ins: never (external owner). + Org-shared skills are eligible for improvement (edits stay local) but protected from ARCHIVE/DELETE elsewhere.""" + if ((skill_path is not None and is_external_skill_path(skill_path)) or is_protected_builtin(skill_name) + or is_hub_installed(skill_name)): return False if is_bundled(skill_name): return _prune_builtins_enabled() @@ -375,75 +315,42 @@ def is_curation_eligible(skill_name: str, skill_path: Optional[Path] = None) -> def _is_curator_managed_record(record: Any) -> bool: - """Return True when a usage record opts a skill into curator management. - - NAMING (issue #67140): the on-disk field is ``created_by``, which reads like provenance but is consumed as a - **curator-management opt-in policy flag**. The two are not the same question: provenance = "who authored this - file" — historical fact, unrecoverable for records written before the marker existed; management = "may autonomous - curation mutate/archive this" — a policy decision the user can change at any time via ``hermes curator adopt``. - - ``created_by: "agent"`` therefore means "curator-managed", NOT "proof the agent wrote it". The field name is - retained because it is already on disk in every user's ``.usage.json``; renaming it would strand those records. - Read it as policy, and prefer ``is_curator_managed()`` at call sites so the intent is unambiguous.""" + """The on-disk ``created_by`` field reads like provenance but is a curator-management OPT-IN policy flag: + ``"agent"`` means "curator-managed", not proof of authorship (the user can flip it via ``hermes curator + adopt``). The name is kept because it is already in every user's ``.usage.json``.""" return isinstance(record, dict) and (record.get("created_by") == "agent" or record.get("agent_created") is True) def is_curator_managed(skill_name: str) -> bool: - """Whether *skill_name* is opted into curator management — the policy-intent alias for the ``created_by``-marker - check (see ``_is_curator_managed_record`` for the field-name story).""" + """Policy-intent alias for the ``created_by`` marker check.""" return _is_curator_managed_record(load_usage().get(skill_name)) def list_unmanaged_skill_names() -> List[str]: - """Enumerate curation-ELIGIBLE skills that carry no provenance marker. - - These are skills the curator *could* manage (not hub-installed, not external, not protected built-ins) but never - will, because nothing ever wrote ``created_by: agent`` onto their usage record. Two ways a skill lands here: it - predates the provenance mechanism entirely (records written before ``created_by`` existed carry no key at all, so - authorship is unknowable from the record alone), or it was created by a FOREGROUND - ``skill_manage(action="create")`` call, which deliberately does not mark provenance (skills a user asks for belong - to the user). - - Either way the skill is invisible to ``curated_report()`` and therefore to every automatic transition. ``hermes - curator status`` surfaces this count so the blind spot is legible instead of silent, and ``hermes curator adopt`` - lets the user hand specific skills over explicitly. - - Provenance is a DECLARATION, never an inference: this function only reports, and callers must not auto-adopt what - it returns. Heavy patch or use counts are evidence of maintenance, not of authorship — the agent edits - user-authored skills on the user's behalf routinely.""" + """Curation-ELIGIBLE skills with no provenance marker: records predating ``created_by``, or FOREGROUND + ``skill_manage(action="create")`` results (skills a user asks for belong to the user). Invisible to + ``curated_report()`` and every automatic transition; surfaced by ``hermes curator status`` and handed over + only by explicit ``hermes curator adopt`` — provenance is a declaration, never inferred from activity.""" def _keep(name: str, skill_md: Path, bundled: Set[str], usage: Dict[str, Any]) -> bool: - # Anything with an external owner or a bundled/protected identity is outside the adoption question. return name not in bundled and not _is_curator_managed_record(usage.get(name)) and is_curation_eligible(name, skill_md) return _scan_local_skills(_keep) def unmanaged_report() -> List[Dict[str, Any]]: - """Rows for every skill :func:`list_unmanaged_skill_names` returns. - - Each row carries the usual activity fields plus ``has_provenance_key``: False when the record has no - ``created_by`` key at all (pre-dates the mechanism), True when the key is present but unset (a foreground create - under the current policy). The distinction matters for explaining WHY a skill is unmanaged; it is not a signal to - adopt on.""" + """Rows for :func:`list_unmanaged_skill_names`. ``has_provenance_key`` is False when the record has no + ``created_by`` key (pre-dates the mechanism), True when present but unset (foreground create) — explains + WHY a skill is unmanaged; not a signal to adopt on.""" usage = load_usage() return [ - _report_row( - name, dict(raw) if isinstance(raw, dict) else None, - has_provenance_key=isinstance(raw, dict) and "created_by" in raw, has_record=isinstance(raw, dict), - ) - for name, raw in ((n, usage.get(n)) for n in list_unmanaged_skill_names()) - ] + _report_row(name, raw, has_provenance_key=isinstance(raw, dict) and "created_by" in raw, + has_record=isinstance(raw, dict)) + for name, raw in ((n, usage.get(n)) for n in list_unmanaged_skill_names())] def adopt_skill(skill_name: str) -> Tuple[bool, str]: - """Hand *skill_name* to the curator by user declaration. - - Writes the same ``created_by: agent`` marker the background review fork writes, so the skill joins - ``curated_report()`` and the automatic transition walk. The inactivity clock is NOT reset: the skill's existing - ``last_activity_at`` still governs staleness, so adopting something idle for months does not buy it a fresh window - (nor does it archive it on the spot — the state machine decides on the next pass). - - Returns (ok, message). Refuses hub-installed, external, and protected built-in skills, which have an owner other - than the user.""" + """Hand *skill_name* to the curator by user declaration — writes the same ``created_by: agent`` marker the + background review fork writes. The inactivity clock is NOT reset. Refuses hub-installed, external, bundled + and protected built-in skills. Returns (ok, message).""" if not skill_name: return False, "no skill name given" if is_protected_builtin(skill_name): @@ -451,8 +358,7 @@ def adopt_skill(skill_name: str) -> Tuple[bool, str]: if is_hub_installed(skill_name): return False, f"'{skill_name}' is hub-installed; its upstream owns it" if is_bundled(skill_name): - # Bundled skills already fall under the curator via ``curator.prune_builtins``; stamping created_by=agent - # on one would claim Hermes' own shipped skill was agent-authored and change nothing about its eligibility. + # Bundled skills are governed by prune_builtins; stamping created_by=agent would change nothing. return False, f"'{skill_name}' is a bundled built-in — it is governed by curator.prune_builtins, not by adoption" skill_dir = _find_skill_dir(skill_name) if skill_dir is None: @@ -474,13 +380,11 @@ def _empty_record() -> Dict[str, Any]: return { "created_by": None, "use_count": 0, "view_count": 0, "last_used_at": None, "last_viewed_at": None, "patch_count": 0, "patch_generation": 0, "last_reused_patch_generation": 0, "last_patched_at": None, - "created_at": _now_iso(), "state": STATE_ACTIVE, "pinned": False, "archived_at": None, - } + "created_at": _now_iso(), "state": STATE_ACTIVE, "pinned": False, "archived_at": None} def _backfilled(rec: Any) -> Dict[str, Any]: - """*rec* with every missing default key filled in (a fresh record when not a dict), so - callers never need to handle old files that predate newer keys.""" + """*rec* with every missing default key filled in (a fresh record when not a dict).""" if not isinstance(rec, dict): return _empty_record() for k, v in _empty_record().items(): @@ -489,28 +393,26 @@ def _backfilled(rec: Any) -> Dict[str, Any]: def _report_row(name: str, raw: Any, **extra: Any) -> Dict[str, Any]: - """Build one report row: name + backfilled record + *extra* + derived activity fields.""" row = {"name": name, **_backfilled(raw), **extra} row.update(last_activity_at=latest_activity_at(row), activity_count=activity_count(row)) return row def load_usage() -> Dict[str, Dict[str, Any]]: - """Read the entire .usage.json map. Returns empty dict on missing/corrupt.""" + """The whole .usage.json map (non-dict values dropped); {} on missing/corrupt.""" path = _usage_file() if not path.exists(): return {} try: data = json.loads(path.read_text(encoding="utf-8")) + return {str(k): v for k, v in data.items() if isinstance(v, dict)} if isinstance(data, dict) else {} except (OSError, json.JSONDecodeError) as e: logger.debug("Failed to read %s: %s", path, e) return {} - # Defensive: drop any non-dict values - return {str(k): v for k, v in data.items() if isinstance(v, dict)} if isinstance(data, dict) else {} def save_usage(data: Dict[str, Dict[str, Any]]) -> bool: - """Write the usage map atomically and report whether it committed.""" + """Write the usage map atomically; True when it committed.""" path = _usage_file() try: _atomic_write(path, ".usage_", lambda f: json.dump(data, f, indent=2, sort_keys=True, ensure_ascii=False)) @@ -521,18 +423,16 @@ def save_usage(data: Dict[str, Dict[str, Any]]) -> bool: def get_record(skill_name: str) -> Dict[str, Any]: - """Return the record for *skill_name*, creating a fresh one if missing.""" + """The (backfilled) record for *skill_name*; fresh defaults if missing.""" return _backfilled(load_usage().get(skill_name)) def _locked_update( skill_name: str, op: Callable[[Dict[str, Dict[str, Any]]], Tuple[Any, bool]], fail_log: str, - guard: Optional[Callable[[], bool]] = None, -) -> Any: - """Run *op(data)* on the loaded usage map under the file lock. Best-effort. *guard* (if given) is evaluated before - the lock is taken; a False result skips the update. *op* returns a (result, dirty) pair; the map is saved only - when dirty. Returns the result, or None when the guard failed, the save did not land, or anything raised (logged - at DEBUG via *fail_log*).""" + guard: Optional[Callable[[], bool]] = None) -> Any: + """Run *op(data) -> (result, dirty)* on the usage map under the file lock; save only when dirty. *guard* + runs before the lock is taken. Returns result, or None when the guard failed, the save did not land, or + anything raised (logged at DEBUG via *fail_log*).""" try: if guard is not None and not guard(): return None @@ -546,12 +446,8 @@ def _locked_update( def seed_record_if_missing(skill_name: str) -> None: - """Persist a baseline usage record for a curation-eligible skill. - - Built-ins carry no usage record until something touches them, which leaves their inactivity clock with no anchor. - Seeding a record here fixes ``created_at`` to the moment the curator first sees the skill, so the archive/stale - clock measures non-use FROM THEN — not from epoch. No-op when a record already exists or the skill isn't - curation-eligible.""" + """Persist a baseline record for a curation-eligible skill so its inactivity clock is anchored at first + sight (``created_at`` = now), not at epoch. No-op if a record exists or the skill isn't eligible.""" if not skill_name or not is_curation_eligible(skill_name): return @@ -563,13 +459,9 @@ def seed_record_if_missing(skill_name: str) -> None: def _mutate(skill_name: str, mutator, *, require_curation_eligible: bool = False) -> Any: - """Load, apply *mutator(record)* in place, save. Best-effort. - - By default this records telemetry for ANY skill — bundled, hub-installed, or agent-created — because usage - tracking is pure observability and is orthogonal to whether a skill is ever curated. Lifecycle mutators - (``set_state``, ``set_pinned``, ``mark_agent_created``) pass ``require_curation_eligible=True`` so they never - write meaningless state onto a skill the curator can't manage (e.g. an ``archived`` flag on a hub-installed - skill).""" + """Load, apply *mutator(record)* in place, save; returns the mutator result (None if nothing landed). + Telemetry is recorded for ANY skill (observability is orthogonal to curation); lifecycle mutators pass + ``require_curation_eligible=True`` so they never write state onto a skill the curator can't manage.""" if not skill_name: return None @@ -582,10 +474,7 @@ def _mutate(skill_name: str, mutator, *, require_curation_eligible: bool = False def _set_field(skill_name: str, key: str, value: Any) -> bool: """Curation-gated single-field write; True only when the write landed.""" - def _apply(rec: Dict[str, Any]) -> bool: - rec[key] = value - return True # non-None sentinel: _mutate propagates the mutator result - return bool(_mutate(skill_name, _apply, require_curation_eligible=True)) + return bool(_mutate(skill_name, lambda rec: rec.update({key: value}) or True, require_curation_eligible=True)) def _non_negative_int(value: Any) -> int: @@ -603,7 +492,7 @@ def _bump(rec: Dict[str, Any], count_key: str, ts_key: str) -> None: def telemetry_provenance(skill_name: str, record: Optional[Dict[str, Any]] = None) -> str: - """Return the bounded provenance used by shared skill metrics.""" + """Bounded provenance label for shared skill metrics.""" if is_hub_installed(skill_name) or is_bundled(skill_name): return "installed" if ":" in skill_name: @@ -623,10 +512,8 @@ def telemetry_provenance(skill_name: str, record: Optional[Dict[str, Any]] = Non def _emit_skill_lifecycle( skill_name: str, action: str, *, record: Optional[Dict[str, Any]] = None, - task_id: Optional[str] = None, session_id: Optional[str] = None, **facts: Any, -) -> None: - """Emit one best-effort lifecycle fact after authoritative state changes. *facts* may carry ``use_count`` / - ``reused`` / ``reuse_after_patch``; absent ones are sent as None.""" + task_id: Optional[str] = None, session_id: Optional[str] = None, **facts: Any) -> None: + """Best-effort lifecycle hook after an authoritative state change; absent facts are sent as None.""" try: from hermes_cli.lifecycle import has_hook, invoke_hook if not has_hook("on_skill_lifecycle"): @@ -640,27 +527,22 @@ def _emit_skill_lifecycle( logger.debug("skill_usage lifecycle hook failed for %s/%s", skill_name, action, exc_info=True) -def _mutate_and_emit(skill_name: str, action: str, mutator: Callable[[Dict[str, Any]], Dict[str, Any]], **hook_kwargs: Any) -> None: - """``_mutate`` then emit *action* with the mutator's facts as the record — only if the write landed. Any - ``use_count`` / ``reused`` / ``reuse_after_patch`` facts are forwarded to the hook.""" +def _mutate_and_emit(skill_name: str, action: str, mutator: Callable[[Dict[str, Any]], Dict[str, Any]], + **hook_kwargs: Any) -> None: + """``_mutate`` then emit *action* with the mutator's facts as the record — only if the write landed.""" facts = _mutate(skill_name, mutator) if isinstance(facts, dict): hook_kwargs.update({k: facts[k] for k in ("use_count", "reused", "reuse_after_patch") if k in facts}) _emit_skill_lifecycle(skill_name, action, record=facts, **hook_kwargs) -# --- Public counter-bump helpers — telemetry for ALL skills (observability only) --- +# --- Public counter-bump helpers — telemetry for ALL skills regardless of provenance (observability only) --- def bump_view(skill_name: str) -> None: - """Bump view_count and last_viewed_at. Called from skill_view(). - - Tracks every skill regardless of provenance — built-ins and hub skills included. Usage telemetry is observability, - not a curation signal.""" _mutate(skill_name, lambda rec: _bump(rec, "view_count", "last_viewed_at")) def bump_use(skill_name: str, *, task_id: Optional[str] = None, session_id: Optional[str] = None) -> None: - """Bump use_count and last_used_at. Called when a skill is actively used (e.g. loaded into the prompt path or - referenced from an assistant turn). Tracks every skill regardless of provenance.""" + """Skill actively used (loaded into the prompt path / referenced from an assistant turn).""" def _apply(rec: Dict[str, Any]) -> Dict[str, Any]: previous_use_count = _non_negative_int(rec.get("use_count")) patch_generation = _non_negative_int(rec.get("patch_generation")) @@ -669,20 +551,16 @@ def bump_use(skill_name: str, *, task_id: Optional[str] = None, session_id: Opti reuse_after_patch = reused and patch_generation > last_reused_generation rec.update( use_count=previous_use_count + 1, last_used_at=_now_iso(), patch_generation=patch_generation, - last_reused_patch_generation=patch_generation if reuse_after_patch else last_reused_generation, - ) + last_reused_patch_generation=patch_generation if reuse_after_patch else last_reused_generation) return { "created_by": rec.get("created_by"), "use_count": rec["use_count"], - "reused": reused, "reuse_after_patch": reuse_after_patch, - } + "reused": reused, "reuse_after_patch": reuse_after_patch} _mutate_and_emit(skill_name, "loaded", _apply, task_id=task_id, session_id=session_id) def bump_patch( - skill_name: str, *, action: str = "patch", task_id: Optional[str] = None, session_id: Optional[str] = None -) -> None: - """Bump patch_count and last_patched_at. Called from skill_manage (patch/edit). Tracks every skill regardless of - provenance.""" + skill_name: str, *, action: str = "patch", task_id: Optional[str] = None, session_id: Optional[str] = None) -> None: + """Called from skill_manage (patch/edit).""" def _apply(rec: Dict[str, Any]) -> Dict[str, Any]: _bump(rec, "patch_count", "last_patched_at") rec["patch_generation"] = _non_negative_int(rec.get("patch_generation")) + 1 @@ -692,12 +570,10 @@ def bump_patch( def record_created( - skill_name: str, *, agent_created: bool, task_id: Optional[str] = None, session_id: Optional[str] = None -) -> None: - """Persist explicit creation provenance and emit a successful create fact.""" + skill_name: str, *, agent_created: bool, task_id: Optional[str] = None, session_id: Optional[str] = None) -> None: + """Persist creation provenance and emit a create fact. A successful create is a new logical skill even + if stale sidecar state survived an earlier deletion, so the record is reset.""" def _apply(rec: Dict[str, Any]) -> Dict[str, Any]: - # A successful create is a new logical skill even if stale sidecar state survived an earlier deletion or - # manual filesystem change. rec.clear() rec.update(_empty_record(), **({"created_by": "agent"} if agent_created else {})) return {"created_by": rec["created_by"]} @@ -706,23 +582,18 @@ def record_created( def record_installed(skill_name: str) -> None: """Record a successful Skills Hub install without exporting its name.""" - def _apply(rec: Dict[str, Any]) -> Dict[str, Any]: - rec.update(created_by="installed", state=STATE_ACTIVE, archived_at=None) - return {"created_by": rec["created_by"]} - _mutate_and_emit(skill_name, "installed", _apply) + _mutate_and_emit(skill_name, "installed", + lambda rec: rec.update(created_by="installed", state=STATE_ACTIVE, archived_at=None) or {"created_by": "installed"}) def mark_agent_created(skill_name: str) -> None: - """Opt a skill created by skill_manage into curator management. - - Viewing or invoking a manually authored skill may still create telemetry, but only this explicit marker makes it - eligible for automatic curation.""" + """Opt a skill into curator management — the only thing that makes it eligible for automatic curation.""" _set_field(skill_name, "created_by", "agent") def set_state(skill_name: str, state: str) -> None: - """Set lifecycle state. No-op if *state* is invalid or the skill isn't curator-manageable (hub skills, or - built-ins with pruning disabled).""" + """Set lifecycle state; no-op for an invalid state or a skill the curator can't manage. Emits + archived/stale/restored (active<-archived); active<-stale emits nothing.""" if state not in _VALID_STATES: logger.debug("set_state: invalid state %r for %s", state, skill_name) return @@ -739,46 +610,39 @@ def set_state(skill_name: str, state: str) -> None: facts = _mutate(skill_name, _apply, require_curation_eligible=True) if not isinstance(facts, dict) or not facts.get("changed"): return - restored = state == STATE_ACTIVE and facts.get("previous_state") == STATE_ARCHIVED # active<-stale emits nothing + restored = state == STATE_ACTIVE and facts.get("previous_state") == STATE_ARCHIVED action = "restored" if restored else {STATE_ARCHIVED: "archived", STATE_STALE: "stale"}.get(state) if action is not None: _emit_skill_lifecycle(skill_name, action, record=facts) def set_pinned(skill_name: str, pinned: bool) -> bool: - """Set/clear the pin flag. Returns False when the write did not land (skill not curation-eligible), True on - success — so callers can report failure instead of a false success (issue #92993).""" + """Set/clear the pin flag; False when the write did not land (not curation-eligible) so callers can + report failure instead of a false success.""" return _set_field(skill_name, "pinned", bool(pinned)) def set_sync(skill_name: str, sync: bool) -> None: - """Set the sync opt-in flag on a skill's usage record. - - Sync is OPT-IN: nothing propagates to the sync plane unless the user marks a skill with ``sync: true`` here. Sits - alongside ``pinned``/``created_by`` on the ``.usage.json`` sidecar and is read by - ``tools.skills_sync_client.list_synced_skill_names``. Gated on curation eligibility so bundled/hub/external skills - (which never sync) can't be marked. Provisional per the M1-D default.""" + """Sync is OPT-IN (``sync: true`` on the record, read by ``tools.skills_sync_client.list_synced_skill_names``); + gated on curation eligibility so bundled/hub/external skills can't be marked.""" _set_field(skill_name, "sync", bool(sync)) def is_sync_enabled(skill_name: str) -> bool: - """Whether a skill is opted into sync (``sync: true`` in its record).""" return get_record(skill_name).get("sync") is True def forget(skill_name: str) -> None: - """Drop a skill's usage entry entirely. Called when the skill is deleted.""" + """Drop a skill's usage entry entirely (skill deleted).""" if skill_name: _locked_update(skill_name, lambda d: (None, d.pop(skill_name, None) is not None), "skill_usage.forget(%s) failed: %s") # --- Archive / restore --- def _relocate(src: Path, dest: Path, skill_name: str, action: str, **capture_kwargs: Any) -> Tuple[bool, str]: - """Move *src* to *dest* for *action* ("archive" | "restore") with an audit-ledger entry around it, then apply the - suppression + state side effects for that direction. - - Ledger pre-capture is best-effort and never blocks the move; if it is unavailable the post-move record is skipped - too. The rename falls back to shutil.move across devices. Returns (ok, message).""" + """Move *src* to *dest* for *action* ("archive" | "restore") with an audit-ledger entry around it, then + apply the suppression + state side effects. Ledger capture is best-effort and never blocks the move; the + rename falls back to shutil.move across devices. Returns (ok, message).""" try: from tools import skill_ledger as _ledger _ledger_before = _ledger.capture_before(src, **capture_kwargs) @@ -795,28 +659,22 @@ def _relocate(src: Path, dest: Path, skill_name: str, action: str, **capture_kwa return False, f"failed to {action}: {e}" if action == "archive": - # Pruning a built-in only sticks if the re-seeder is told to leave it alone. - if is_bundled(skill_name): - add_suppressed_name(skill_name) + if is_bundled(skill_name): # pruning a built-in only sticks if the re-seeder is told to leave it alone + _toggle_suppressed_name(skill_name, add=True) set_state(skill_name, STATE_ARCHIVED) else: - # Restoring a pruned built-in lifts its suppression so updates can manage it. - remove_suppressed_name(skill_name) + _toggle_suppressed_name(skill_name, add=False) set_state(skill_name, STATE_ACTIVE) with suppress(Exception): if _ledger is not None: _ledger.record_mutation( - action, skill_name, before=_ledger_before if _ledger_before is not None else [], after_root=dest - ) + action, skill_name, before=_ledger_before if _ledger_before is not None else [], after_root=dest) return True, f"{action}d to {dest}" def archive_skill(skill_name: str) -> Tuple[bool, str]: - """Move a curator-eligible skill directory to ~/.hermes/skills/.archive/. - - Returns (ok, message). Never archives hub-installed skills. Bundled built-ins are only archivable when - ``curator.prune_builtins`` is enabled; when one is archived, its name is added to the suppression list so the - update-time re-seeder leaves it archived instead of restoring it.""" + """Move a curator-eligible skill dir to ``.archive/`` (flattened; timestamp suffix on collision). Never + hub skills; bundled built-ins only with ``curator.prune_builtins`` (and then suppressed from re-seeding).""" skill_dir = _find_skill_dir(skill_name) if skill_dir is None and _find_external_skill_dir(skill_name) is not None: return False, _external_read_only_message(skill_name) @@ -836,52 +694,37 @@ def archive_skill(skill_name: str) -> Tuple[bool, str]: archive_root.mkdir(parents=True, exist_ok=True) except OSError as e: return False, f"failed to create archive dir: {e}" - - # Flatten any category nesting into a single ".archive//" so restores are simple; on collision, append - # a timestamp. dest = archive_root / skill_dir.name if dest.exists(): dest = archive_root / f"{skill_dir.name}-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}" - - # complete_package: consolidation may have re-homed support files out of the tree first, so a disk-only - # capture can come back hollow; the fill from the newest curator backup keeps rollback restorable (#96962). + # complete_package: consolidation may have re-homed support files first, so a disk-only capture can come + # back hollow; the fill from the newest curator backup keeps rollback restorable. return _relocate(skill_dir, dest, skill_name, "archive", complete_package=True, skill=skill_name) def restore_skill(skill_name: str) -> Tuple[bool, str]: - """Move an archived skill back to ~/.hermes/skills/. Restores to the flat top-level layout; original category - nesting is NOT reconstructed. + """Move an archived skill back to the flat top-level layout (category nesting is NOT reconstructed). - Refuses to restore under a name that now collides with a hub-installed skill — that would shadow the upstream - version. Also refuses to restore over a bundled built-in UNLESS ``curator.prune_builtins`` is enabled (in which - case built-ins are curator-managed and restoring is the documented way to lift a prune). Restoring clears any - suppression entry so future updates may re-seed the built-in again.""" - # Hub skills always have an external upstream owner — never shadow them. + Refuses names that now collide with a hub skill, or a bundled built-in unless ``curator.prune_builtins`` + is on (then restoring is the documented way to lift a prune) — either would shadow the upstream copy.""" if is_hub_installed(skill_name): return False, f"skill '{skill_name}' is now hub-installed; restore would shadow the upstream version" - # A bundled built-in is upstream-owned UNLESS prune_builtins is on; with the flag off, restoring over it - # would shadow the bundled version. if is_bundled(skill_name) and not _prune_builtins_enabled(): return False, f"skill '{skill_name}' is now bundled; restore would shadow the upstream version" archive_root = _archive_dir() if not archive_root.exists(): return False, "no archive directory" - # Try exact name match first, then the timestamped-duplicate fallback. Recursive walk handles nested archive - # layouts (e.g. .archive///) left behind by older archive paths or external imports. + # Exact name first (recursive: older archive paths left nested layouts), then the timestamped-duplicate + # fallback. Only "-YYYYMMDDHHMMSS" (14 digits) counts — a bare startswith("-") would let + # restoring "git" pull an archived "git-helpers" out and rename it, destroying the sibling's only copy. candidates = [p for p in archive_root.rglob("*") if p.is_dir() and p.name == skill_name] if not candidates: - # A name collision makes archive_skill() disambiguate by appending its UTC timestamp - # ("-YYYYMMDDHHMMSS", a 14-digit suffix), so only that exact shape is another copy of THIS skill. - # A bare startswith(f"{skill_name}-") also swallows unrelated sibling skills — restoring "git" would - # otherwise pull an archived "git-helpers" out of the archive and rename it to "git", destroying the - # sibling's only copy. Require the suffix to be the timestamp archive_skill writes. prefix = f"{skill_name}-" candidates = sorted( (p for p in archive_root.rglob("*") if p.is_dir() and p.name.startswith(prefix) and len(p.name) - len(prefix) == 14 and p.name[len(prefix):].isdigit()), - reverse=True, - ) + reverse=True) if not candidates: return False, f"skill '{skill_name}' not found in archive" @@ -896,22 +739,18 @@ def _match_skill_dir(skill_mds: Iterable[Path], skill_name: str) -> Optional[Pat def _find_skill_dir(skill_name: str) -> Optional[Path]: - """Locate the directory for a skill by its frontmatter `name:` field. - - Handles both flat (~/.hermes/skills//SKILL.md) and category-nested - (~/.hermes/skills///SKILL.md) layouts. Uses the gated index iterator so M2 org mirrors resolve - ONLY for the active org (stale ``_org//`` trees never match).""" + """Skill dir by frontmatter ``name`` (flat or category-nested). The gated index iterator makes org mirrors + resolve ONLY for the active org (stale ``_org//`` trees never match).""" base = _skills_dir() if not base.exists(): return None from agent.skill_utils import iter_skill_index_files return _match_skill_dir( - (p for p in iter_skill_index_files(base, "SKILL.md") if not is_external_skill_path(p)), skill_name - ) + (p for p in iter_skill_index_files(base, "SKILL.md") if not is_external_skill_path(p)), skill_name) def _find_external_skill_dir(skill_name: str) -> Optional[Path]: - """Locate a skill under configured external dirs by frontmatter name.""" + """Skill dir under configured external dirs by frontmatter name.""" from agent.skill_utils import get_all_skills_dirs for base in (b for b in get_all_skills_dirs()[1:] if b.exists()): found = _match_skill_dir((p for p in base.rglob("SKILL.md") if not is_excluded_skill_path(p)), skill_name) @@ -922,20 +761,13 @@ def _find_external_skill_dir(skill_name: str) -> Optional[Path]: # --- Reporting — for the curator CLI / slash command --- def curated_report() -> List[Dict[str, Any]]: - """Return a list of {name, provenance, state, pinned, last_activity_at, ...} records for every curator-managed - skill. Missing usage records are backfilled with defaults so callers can always index fields. - - ``provenance`` is 'agent', 'bundled', or 'hub' (see :func:`provenance`). Bundled skills are only included when - ``curator.prune_builtins`` is enabled. Hub-installed skills are never included. - - Each row carries ``_persisted``: True when a real record exists in ``.usage.json``, False when the row is a fresh - backfill (e.g. a built-in seen for the first time). The curator uses this to seed the inactivity clock instead of - treating an unrecorded skill as ancient.""" + """One backfilled row per curator-managed skill with ``provenance`` ('agent'|'bundled'|'hub') and + ``_persisted`` (a real record exists — the curator seeds the inactivity clock for fresh backfills instead of + treating them as ancient). Bundled skills only with ``curator.prune_builtins``; hub skills never.""" data = load_usage() names = set(list_agent_created_skill_names()) - # Issue #92993: a successfully pinned skill must be visible in the report even when it lacks the created_by - # marker (eligible-but-unmanaged), or its pin silently vanishes from `curator status`. The local-dir guard - # keeps stale records for deleted skill dirs from rendering as ghost rows; `curator unpin` is the cleanup path. + # A pinned-but-unmanaged skill must stay visible or its pin silently vanishes from `curator status`; the + # local-dir guard keeps stale records for deleted dirs from rendering as ghost rows (`curator unpin` cleans up). names.update( name for name, rec in data.items() if isinstance(rec, dict) and rec.get("pinned") and is_curation_eligible(name) and _find_skill_dir(name) is not None @@ -947,18 +779,13 @@ def curated_report() -> List[Dict[str, Any]]: def provenance(skill_name: str) -> str: - """Classify a skill's origin: 'hub', 'bundled', or 'agent'. 'agent' covers both agent-authored and local - manually-authored skills — anything not seeded from the bundled repo or installed via the hub.""" + """'hub', 'bundled', or 'agent' (the latter covers agent-authored AND local manually-authored skills).""" return "hub" if is_hub_installed(skill_name) else "bundled" if is_bundled(skill_name) else "agent" def usage_report() -> List[Dict[str, Any]]: - """Return usage telemetry for EVERY skill on disk, with provenance. - - Unlike ``curated_report()`` (which is scoped to curator-managed candidates), this surfaces all skills — bundled - built-ins and hub-installed included — so callers can answer "how often is this skill used" independent of whether - it's ever curated. Rows carry a ``provenance`` field ('agent' | 'bundled' | 'hub') and ``_persisted`` (whether a - real ``.usage.json`` record backs the row).""" + """Usage rows for EVERY skill on disk (built-ins and hub included), with ``provenance`` and ``_persisted`` — + unlike ``curated_report()``, which is scoped to curator-managed candidates.""" base = _skills_dir() if not base.exists(): return [] @@ -966,5 +793,6 @@ def usage_report() -> List[Dict[str, Any]]: rows: Dict[str, Dict[str, Any]] = {} for name, _skill_md in _iter_skill_mds(base, local_only=False): if name not in rows: - rows[name] = _report_row(name, data.get(name), provenance=provenance(name), _persisted=isinstance(data.get(name), dict)) + rows[name] = _report_row(name, data.get(name), provenance=provenance(name), + _persisted=isinstance(data.get(name), dict)) return [rows[name] for name in sorted(rows)] diff --git a/tools/skillevaluator_scan.py b/tools/skillevaluator_scan.py index 9c48e38544..21d6f82b39 100644 --- a/tools/skillevaluator_scan.py +++ b/tools/skillevaluator_scan.py @@ -1,28 +1,10 @@ #!/usr/bin/env python3 -"""Advisory NVIDIA SkillEvaluator Tier 1 scan for skill installs. - -Runs alongside (never instead of) ``tools/skills_guard.py``, which stays the -enforcement layer (trust levels, install policy, block verdicts). This adds a -second, advisory opinion: NVIDIA SkillEvaluator's keyless Tier 1 static checks. - -Contract: -- Warn, don't block: PII-class findings are shown with file/line and the - install continues (the upstream PII scanner has known false-positive classes - such as ``git@github.com`` and ``op://`` references). -- Prompt only for secrets-class criticals (private keys, cloud keys, tokens, - credentialed connection strings); ``--force`` / non-interactive installs - proceed with a loud warning instead of wedging on a prompt. -- Never break installs: scanner missing, crashing, timing out or emitting - unparseable output all degrade to a no-op. - -Optional binary:: - - uv tool install --python 3.13 \ - "skillevaluator @ git+https://github.com/NVIDIA/SkillEvaluator.git@v0.1.0" - -Toggle via ``skills.tier1_advisory`` in config.yaml (default on; a no-op -unless the binary is installed). -""" +"""Advisory NVIDIA SkillEvaluator Tier 1 scan for skill installs — runs alongside (never instead of) +``tools/skills_guard.py``, the enforcement layer. Contract: warn, don't block (PII findings are shown, the +install continues — the upstream PII scanner false-positives on ``git@github.com`` / ``op://``); prompt only +for secrets-class criticals (``--force`` / non-interactive proceed with a loud warning); never break installs +(missing/crashing/timed-out/unparseable scanner = no-op). Toggle: ``skills.tier1_advisory`` (default on). +Binary: ``uv tool install --python 3.13 "skillevaluator @ git+https://github.com/NVIDIA/SkillEvaluator.git@v0.1.0"``.""" from __future__ import annotations @@ -33,31 +15,22 @@ import subprocess import tempfile from dataclasses import dataclass, field from pathlib import Path -from typing import List, Optional +from typing import List logger = logging.getLogger(__name__) SCANNER_BIN = "skillevaluator" -# Keyless, deterministic Tier 1 checks. Schema/quality are excluded: they are -# hygiene signal for the index pipeline, not install-time signal. `security` -# invokes NVIDIA SkillSpector (second optional binary, static-rules mode, no -# LLM); when absent or inconsistent it reports status="incomplete" = no opinion. +# Keyless, deterministic checks. Schema/quality are index-pipeline hygiene, not install-time signal. +# `security` invokes NVIDIA SkillSpector (static rules, no LLM); when absent it reports status="incomplete". TIER1_CHECKS = "pii,unicode,lint,license,security" SCAN_TIMEOUT_SECONDS = 120 -# check_name values (from SkillEvaluator's pii_patterns.yaml categories) -# that indicate a possible REAL credential rather than personal-info -# hygiene. These are the only findings that earn a confirmation prompt. +# SkillEvaluator pii_patterns.yaml categories that indicate a possible REAL credential rather than +# personal-info hygiene — the only findings that earn a confirmation prompt. SECRETS_CLASS_CHECKS = frozenset({ - "database_credentials", - "hardcoded_secrets", - "jwt_tokens", - "webhook_urls", - "aws_identifiers", - "github_tokens", - "private_keys", -}) + "database_credentials", "hardcoded_secrets", "jwt_tokens", "webhook_urls", + "aws_identifiers", "github_tokens", "private_keys"}) @dataclass @@ -75,9 +48,7 @@ class Tier1Finding: return self.check in SECRETS_CLASS_CHECKS def location(self) -> str: - if self.file and self.line: - return f"{self.file}:{self.line}" - return self.file or "?" + return f"{self.file}:{self.line}" if self.file and self.line else self.file or "?" @dataclass @@ -102,12 +73,10 @@ def scanner_available() -> bool: def tier1_advisory_enabled() -> bool: - """Read skills.tier1_advisory from config (default True; safe because the - scan is a no-op without the optional binary on PATH).""" + """``skills.tier1_advisory`` (default True; safe because the scan is a no-op without the binary).""" try: from hermes_cli.config import load_config - cfg = load_config() - skills_cfg = cfg.get("skills") or {} + skills_cfg = load_config().get("skills") or {} if not isinstance(skills_cfg, dict): return True value = skills_cfg.get("tier1_advisory", True) @@ -119,19 +88,14 @@ def tier1_advisory_enabled() -> bool: def _parse_report(report: dict) -> Tier1Report: - """Reduce a SkillEvaluator JSON report to install-relevant findings. - - Findings from ``status == "incomplete"`` validators are kept (partial - evidence is still evidence) but those validators are excluded from the - pass/fail signal so an evidence-free fail can't render as a failure. - """ + """Reduce a SkillEvaluator JSON report to install-relevant findings. Findings from ``status == "incomplete"`` + validators are kept (partial evidence is evidence) but excluded from the pass/fail signal.""" findings: List[Tier1Finding] = [] incomplete: List[str] = [] any_complete_failed = False for res in report.get("results", []) or []: validator = str(res.get("validator", "unknown")) - is_incomplete = str(res.get("status", "")).lower() == "incomplete" - if is_incomplete: + if str(res.get("status", "")).lower() == "incomplete": incomplete.append(validator) elif not res.get("passed", True): any_complete_failed = True @@ -139,35 +103,24 @@ def _parse_report(report: dict) -> Tier1Report: if not isinstance(f, dict): continue findings.append(Tier1Finding( - check=str(f.get("check_name", "")), - validator=validator, - severity=str(f.get("severity", "info")).lower(), - message=str(f.get("message", ""))[:200], - file=str(f.get("file_path", "")), - line=int(f.get("line_number") or 0), - suggestion=str(f.get("suggestion", ""))[:200], - )) - return Tier1Report( - available=True, - passed=not any_complete_failed and not findings, - findings=findings, - incomplete_checks=incomplete, - ) + check=str(f.get("check_name", "")), validator=validator, + severity=str(f.get("severity", "info")).lower(), message=str(f.get("message", ""))[:200], + file=str(f.get("file_path", "")), line=int(f.get("line_number") or 0), + suggestion=str(f.get("suggestion", ""))[:200])) + return Tier1Report(available=True, passed=not any_complete_failed and not findings, findings=findings, + incomplete_checks=incomplete) def run_tier1_scan(skill_dir: Path, timeout: int = SCAN_TIMEOUT_SECONDS) -> Tier1Report: - """Run SkillEvaluator Tier 1 over one skill directory; any failure returns - ``available=False`` (no advisory opinion), never raises.""" + """Run SkillEvaluator Tier 1 over one skill dir; any failure returns ``available=False``, never raises.""" if not scanner_available(): return Tier1Report(available=False, error="scanner not on PATH") with tempfile.TemporaryDirectory(prefix="se-tier1-") as outdir: try: subprocess.run( - [SCANNER_BIN, "validate", str(skill_dir), - "--checks", TIER1_CHECKS, "--no-dedup", + [SCANNER_BIN, "validate", str(skill_dir), "--checks", TIER1_CHECKS, "--no-dedup", "-r", "json", "-o", outdir], - capture_output=True, text=True, timeout=timeout, - ) + capture_output=True, text=True, timeout=timeout) except subprocess.TimeoutExpired: return Tier1Report(available=False, error=f"scan timed out after {timeout}s") except OSError as exc: @@ -185,20 +138,17 @@ def run_tier1_scan(skill_dir: Path, timeout: int = SCAN_TIMEOUT_SECONDS) -> Tier def format_tier1_report(report: Tier1Report, limit: int = 10) -> str: - """Plain-text advisory summary for console display.""" + """Plain-text advisory summary for console display ("" when unavailable).""" if not report.available: return "" lines: List[str] = [] if not report.findings: - if report.incomplete_checks: - lines.append("SkillEvaluator Tier 1: no findings from completed checks.") - else: - lines.append("SkillEvaluator Tier 1: no findings.") + lines.append("SkillEvaluator Tier 1: no findings from completed checks." if report.incomplete_checks + else "SkillEvaluator Tier 1: no findings.") else: lines.append( f"SkillEvaluator Tier 1 (advisory): " - f"{len(report.findings)} finding(s) — informational, verify before relying on this skill." - ) + f"{len(report.findings)} finding(s) — informational, verify before relying on this skill.") shown = report.secrets_findings + report.advisory_findings for f in shown[:limit]: tag = "SECRETS" if f.is_secrets_class else f.severity.upper() @@ -206,6 +156,5 @@ def format_tier1_report(report: Tier1Report, limit: int = 10) -> str: if len(shown) > limit: lines.append(f" … and {len(shown) - limit} more") if report.incomplete_checks: - names = ", ".join(report.incomplete_checks) - lines.append(f" (not run: {names} — no opinion from these checks)") + lines.append(f" (not run: {', '.join(report.incomplete_checks)} — no opinion from these checks)") return "\n".join(lines) diff --git a/tools/skills_ast_audit.py b/tools/skills_ast_audit.py index b25e8b665e..7ed019a5e9 100644 --- a/tools/skills_ast_audit.py +++ b/tools/skills_ast_audit.py @@ -1,13 +1,6 @@ -""" -AST-level deep audit for skill Python files — opt-in diagnostic, not a security gate. - -Per SECURITY.md §2.4, Skills Guard is in-process heuristics ("useful — not -boundaries"). This module is a separate opt-in diagnostic that flags dynamic -import / dynamic attribute access patterns operators may want to eyeball when -reviewing third-party skill code. Every pattern flagged here has legitimate -uses; findings are hints for human review, not verdicts. - -CLI: ``hermes skills audit --deep`` +"""AST-level deep audit for skill Python files — opt-in diagnostic (``hermes skills audit --deep``), not a +security gate (SECURITY.md §2.4). Flags dynamic import / dynamic attribute access patterns for human review; +every pattern has legitimate uses, so findings are hints, not verdicts. """ from __future__ import annotations @@ -20,6 +13,15 @@ from typing import List, Tuple Finding = Tuple[str, int, str, str] _IGNORED_DIRS = {"__pycache__", ".venv", "venv", "node_modules"} +# builtin name -> (index of the argument that must be a literal, pattern_id, description) +_DYNAMIC_CALLS = { + "__import__": (0, "dynamic_import_computed", "__import__ with non-literal module name"), + "getattr": (1, "dynamic_getattr", "getattr with non-literal attribute name"), +} + + +def _is_importlib(name: str) -> bool: + return name == "importlib" or name.startswith("importlib.") def _scan_source(content: str, rel_path: str) -> List[Finding]: @@ -27,57 +29,44 @@ def _scan_source(content: str, rel_path: str) -> List[Finding]: tree = ast.parse(content) except (SyntaxError, ValueError, RecursionError): return [] - findings: List[Finding] = [] + def hit(node, pid: str, desc: str) -> None: + findings.append((rel_path, node.lineno, pid, desc)) + class V(ast.NodeVisitor): def visit_Call(self, node): f = node.func - # importlib.import_module(...) if isinstance(f, ast.Attribute) and f.attr == "import_module": - findings.append((rel_path, node.lineno, "dynamic_import", - "importlib.import_module() — loads arbitrary modules at runtime")) - # __import__() - elif isinstance(f, ast.Name) and f.id == "__import__": - if node.args and not isinstance(node.args[0], ast.Constant): - findings.append((rel_path, node.lineno, "dynamic_import_computed", - "__import__ with non-literal module name")) - # getattr(obj, ) - elif isinstance(f, ast.Name) and f.id == "getattr": - if len(node.args) >= 2 and not isinstance(node.args[1], ast.Constant): - findings.append((rel_path, node.lineno, "dynamic_getattr", - "getattr with non-literal attribute name")) + hit(node, "dynamic_import", "importlib.import_module() — loads arbitrary modules at runtime") + elif isinstance(f, ast.Name) and f.id in _DYNAMIC_CALLS: + arg_index, pid, desc = _DYNAMIC_CALLS[f.id] + if len(node.args) > arg_index and not isinstance(node.args[arg_index], ast.Constant): + hit(node, pid, desc) self.generic_visit(node) def visit_Subscript(self, node): - # obj.__dict__[] - if (isinstance(node.value, ast.Attribute) - and node.value.attr == "__dict__" + if (isinstance(node.value, ast.Attribute) and node.value.attr == "__dict__" and not isinstance(node.slice, ast.Constant)): - findings.append((rel_path, node.lineno, "dict_access", - "__dict__[] — dynamic attribute access")) + hit(node, "dict_access", "__dict__[] — dynamic attribute access") self.generic_visit(node) def visit_Import(self, node): for a in node.names: - if a.name == "importlib" or a.name.startswith("importlib."): - findings.append((rel_path, node.lineno, "importlib_import", - f"import {a.name} — enables dynamic module loading")) + if _is_importlib(a.name): + hit(node, "importlib_import", f"import {a.name} — enables dynamic module loading") self.generic_visit(node) def visit_ImportFrom(self, node): m = node.module or "" - if m == "importlib" or m.startswith("importlib."): - findings.append((rel_path, node.lineno, "importlib_import", - f"from {m} import ... — enables dynamic module loading")) + if _is_importlib(m): + hit(node, "importlib_import", f"from {m} import ... — enables dynamic module loading") self.generic_visit(node) try: V().visit(tree) except (RecursionError, ValueError, RuntimeError): - # Hostile/pathological input: return what we collected so far. - pass - + pass # hostile/pathological input: return what was collected so far return findings @@ -89,17 +78,11 @@ def _scan_file(py: Path, rel: str) -> List[Finding]: def ast_scan_path(path: Path) -> List[Finding]: - """Scan a single .py file or recursively scan all .py under a directory. - - Returns a list of (file, line, pattern_id, description) tuples. Empty for - non-Python paths, missing paths, or paths with no matching patterns. - """ + """Scan one .py file or every .py under a directory; [] for non-Python/missing paths.""" if path.is_file(): return _scan_file(path, path.name) if path.suffix.lower() == ".py" else [] - if not path.is_dir(): return [] - out: List[Finding] = [] for py in sorted(path.rglob("*.py")): if set(py.parent.parts) & _IGNORED_DIRS: @@ -113,11 +96,10 @@ def ast_scan_path(path: Path) -> List[Finding]: def format_ast_report(findings: List[Finding], skill_name: str = "") -> str: - """Plain-text report (Rich-markup-free) grouped by file.""" + """Plain-text (Rich-markup-free) report grouped by file.""" header = f"AST deep scan: {skill_name}" if skill_name else "AST deep scan" if not findings: return f"{header}\n No dynamic import/access patterns detected." - lines = [header, f" {len(findings)} finding(s):"] current = None for f, line, pid, desc in sorted(findings): @@ -125,6 +107,5 @@ def format_ast_report(findings: List[Finding], skill_name: str = "") -> str: current = f lines.append(f" {f}") lines.append(f" L{line} {pid} — {desc}") - lines.append("") - lines.append(" Note: diagnostic hints for human review, not security verdicts.") + lines += ["", " Note: diagnostic hints for human review, not security verdicts."] return "\n".join(lines) diff --git a/tools/skills_guard.py b/tools/skills_guard.py index 6e211553eb..6bdc2d3cdf 100644 --- a/tools/skills_guard.py +++ b/tools/skills_guard.py @@ -1,33 +1,12 @@ #!/usr/bin/env python3 -""" -Skills Guard — Security scanner for externally-sourced skills. +"""Skills Guard — regex static scan of externally-sourced skills plus a trust-aware install policy. -Every skill downloaded from a registry passes through this scanner before -installation: regex-based static analysis for known-bad patterns (data -exfiltration, prompt injection, destructive commands, persistence, etc.) plus a -trust-aware install policy combining the scan verdict with the source's trust. - -Trust levels: builtin (ships with Hermes, never scanned, always trusted); -trusted (openai/skills, anthropics/skills, ...: caution verdicts allowed); -community (everything else: any findings = blocked unless --force). - -Usage: ``result = scan_skill(path, source="community")``, then -``allowed, reason = should_allow_install(result)`` and, when not allowed, -``print(format_scan_report(result))``. - -Known limitation — programmatic writes (out of scope for this static pass): -the agent-config persistence tiers score shell write mechanics (">>" -redirection, "sed -i") and imperative modification prose only. Language write -APIs in bundled scripts — Python open(..., 'w'/'a'), pathlib.Path.write_text(), -os.replace(), shutil.copy*, and Node fs.writeFileSync()/appendFile() — aimed at -agent-config files surface only the low-severity *_ref finding, never a scored -persistence tier. Static regexes cannot reliably tie such a call to the -config-file destination (paths may be built dynamically) without executing the -skill, so language-API persistence is left to runtime gates (install -confirmation, sandboxing). If coverage is added later, it belongs as a fourth -"mechanical" tier next to agent_config_mod_shell, requiring the config-file -name as a literal argument at the call site. -""" +Trust: builtin (ships with Hermes, never scanned), trusted (openai/anthropics/... repos: caution allowed), +community (any findings block unless --force). ``scan_skill(path, source)`` -> ``should_allow_install`` -> +``format_scan_report``. Known limitation: language write APIs (open(..., 'w'), Path.write_text, shutil.copy*, +fs.writeFileSync) aimed at agent-config files surface only the low *_ref finding — static regexes cannot tie +the call to a dynamically built destination; any future coverage belongs as a fourth "mechanical" tier next +to agent_config_mod_shell.""" import re import fnmatch @@ -41,17 +20,11 @@ from typing import List, Tuple SCANNER_VERSION = "skills-guard-v2" - -# --- Hardcoded trust configuration ------------------------------------------ - TRUSTED_REPOS = { "openai/skills", "anthropics/skills", "huggingface/skills", - # NVIDIA-verified skills: each entry ships a signed `skill.oms.sig` - # and a governance `skill-card.md` (sync pipeline drops anything - # missing the signature or card). Catalog details: - # https://github.com/NVIDIA/skills + # NVIDIA-verified: each entry ships a signed `skill.oms.sig` + governance `skill-card.md`. "NVIDIA/skills", } @@ -60,18 +33,14 @@ INSTALL_POLICY = { "builtin": ("allow", "allow", "allow"), "trusted": ("allow", "allow", "block"), "community": ("allow", "block", "block"), - # Agent-created: "ask" on dangerous surfaces as an error to the agent, - # which can retry without the flagged content. This gate only runs when - # skills.guard_agent_created is enabled (off by default) — see - # tools/skill_manager_tool.py::_guard_agent_created_enabled. + # "ask" surfaces as an error to the agent, which can retry without the flagged content. Only + # consulted when skills.guard_agent_created is on (skill_manager_tool._guard_agent_created_enabled). "agent-created": ("allow", "allow", "ask"), } VERDICT_INDEX = {"safe": 0, "caution": 1, "dangerous": 2} -# --- Data structures -------------------------------------------------------- - @dataclass class Finding: pattern_id: str @@ -97,9 +66,8 @@ class ScanResult: # --- Threat patterns — (regex, pattern_id, severity, category, description) -- -# Action verbs that signal file-modification intent. Used by the agent-config -# persistence patterns: a verb within the same line as (and shortly before) an -# agent config filename is scored as modification; a bare mention is not. +# File-modification verbs for the agent-config persistence tiers: a verb shortly before a config +# filename on the same line is scored as modification; a bare mention is not. MODIFY_VERB_RE = ( r'(?:\bwrit(?:e|es|ing)\b|\bwritten\b|\bedit(?:s|ed|ing)?\b' r'|\bmodif(?:y|ies|ied|ying|ication)s?\b|\bupdat(?:e|es|ed|ing)\b' @@ -108,26 +76,18 @@ MODIFY_VERB_RE = ( r'|\breplac(?:e|es|ed|ing)\b|\balter(?:s|ed|ing)?\b|\badd(?:s|ed|ing)\b)' ) -# Config-file groups shared by the agent-config persistence tiers below. _AGENT_CONFIG_FILES = r'(?:AGENTS\.md|CLAUDE\.md|\.cursorrules|\.clinerules)' _HERMES_CONFIG_FILES = r'\.hermes/(?:config\.yaml|SOUL\.md)' -# Path prefixes (real files are e.g. .claude/settings.json), so consume any -# trailing filename characters rather than requiring a clean end-of-word. +# Path prefixes (real files are e.g. .claude/settings.json): consume trailing filename chars. _OTHER_AGENT_CONFIG_FILES = r'\.(?:claude/settings|codex/config)[\w.]*' def _shell_write_re(file_alt: str) -> str: - """Regex for a mechanical shell write into *file_alt*. - - Covers redirection (``>``/``>>``), in-place ``sed -i``, ``tee`` (with the - target as its immediate argument, so a markdown table cell like - ``| tee output | AGENTS.md |`` does not match), and ``cp``/``mv`` with the - config file in destination position (a preceding source argument is - required, so ``cp AGENTS.md backup/`` — a read — does not match; a - trailing extension like ``AGENTS.md.bak`` is not the config file). - A single ``>`` must be preceded by a word/quote/paren character so that - markdown blockquotes (``> text``) and arrows (``-> file``) do not match. - """ + """Mechanical shell write into *file_alt*: ``>``/``>>`` redirection, ``sed -i``, ``tee`` (target as + immediate argument, so ``| tee output | AGENTS.md |`` table cells miss), ``cp``/``mv`` with the file in + destination position (a source arg is required, so ``cp AGENTS.md backup/`` — a read — misses; + ``AGENTS.md.bak`` is not the file). A single ``>`` needs a preceding word/quote/paren char so + blockquotes (``> text``) and arrows (``-> file``) miss.""" return ( rf'(?:>>|[\w"\'`)\]]\s*>)\s*[~\w./-]*{file_alt}(?!\.?\w)' rf'|\bsed\b[^\n]*\s(?:-[A-Za-z]*i[A-Za-z]*|--in-place)\b[^\n]*{file_alt}(?!\.?\w)' @@ -137,15 +97,10 @@ def _shell_write_re(file_alt: str) -> str: def _prose_modify_re(file_alt: str) -> str: - """Regex for prose instructing modification of *file_alt*. - - Two shapes: an imperative-position verb (start of line / bullet item), - or a mid-line verb strengthened by an explicit directive marker - ("you must", "please", "make sure to"). Descriptive mid-line prose - ("skills that edit AGENTS.md") matches neither. The verb→file gap - forbids commas so enumerations ("Write or refactor skills, AGENTS.md, - CLAUDE.md") — a doc listing its subject matter — do not match. - """ + """Prose instructing modification of *file_alt*: an imperative-position verb (line start / bullet), or + a mid-line verb strengthened by a directive marker ("you must", "please", "make sure to"). + Descriptive prose ("skills that edit AGENTS.md") matches neither; the verb→file gap forbids commas so + enumerations ("Write or refactor skills, AGENTS.md, CLAUDE.md") miss.""" return ( rf'^\s*(?:[-*+]\s+|\d+[.)]\s+)?{MODIFY_VERB_RE}[^\n,]{{0,80}}?{file_alt}\b' rf'|(?:\byou\s+(?:must|should|need\s+to)\s+|\bplease\s+' @@ -155,13 +110,8 @@ def _prose_modify_re(file_alt: str) -> str: def _content_contract_re(file_alt: str) -> str: - """Regex for " should contain/include ..." content-contract prose. - - Ambiguous shape: authoring guides teach "Every AGENTS.md should contain - the project purpose" while an attack writes "AGENTS.md should contain - the bypass instructions". Not separable statically, so this tier is - scored high (caution → user confirmation), never critical. - """ + """" should contain/include ..." prose. Authoring guides and attacks share this shape and are not + separable statically, so the tier is scored high (caution → confirmation), never critical.""" return ( rf'{file_alt}\b[^\n]{{0,40}}?\b(?:should|must|needs?\s+to)\s+' rf'(?:contain|say|include|have|list)\b' @@ -169,16 +119,10 @@ def _content_contract_re(file_alt: str) -> str: THREAT_PATTERNS = [ # ── Exfiltration: shell commands leaking secrets ── - # All five env_exfil_* patterns share a loopback exemption: a request - # whose same-line literal destination is scheme-anchored loopback - # (http(s)://localhost, 127.0.0.1, [::1]) cannot move data off the - # machine, so a secret-shaped query param on it is a local session - # token, not exfiltration (e.g. impeccable's live-mode - # `fetch('http://localhost:'+PORT+'/status?token='+TOKEN)`). The - # exemption requires the scheme immediately before the loopback host — - # `evil.com/?u=localhost` does not qualify. A hostile skill that hides - # its real destination behind a variable never matched these same-line - # literal patterns in the first place. + # env_exfil_* share a loopback exemption: a same-line literal scheme-anchored loopback destination + # (http(s)://localhost, 127.0.0.1, [::1]) cannot move data off the machine, so a secret-shaped query + # param there is a local session token. The scheme must immediately precede the host — + # `evil.com/?u=localhost` does not qualify. (r'curl\s+(?![^\n]*https?://(?:localhost|127\.0\.0\.1|\[::1\]))[^\n]*\$\{?\w*(?:KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL)S?\b', "env_exfil_curl", "critical", "exfiltration", "curl command interpolating secret environment variable"), (r'wget\s+(?![^\n]*https?://(?:localhost|127\.0\.0\.1|\[::1\]))[^\n]*\$\{?\w*(?:KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL)S?\b', @@ -199,26 +143,15 @@ THREAT_PATTERNS = [ "docker_dir_access", "high", "exfiltration", "references Docker config (may contain registry creds)"), (r'\$HOME/\.hermes/\.env|\~/\.hermes/\.env', "hermes_env_access", "critical", "exfiltration", "directly references Hermes secrets file"), - # Match `cat ` (reading credentials) but NOT `cat > ` - # or `cat >> `, which are output redirections that WRITE a file - # (e.g. a setup doc telling the user to write their own keys into their - # own local `.env` via a heredoc). Writing your own config in is the - # opposite of exfiltrating secrets out. + # `cat ` reads credentials; `cat >`/`cat >>` WRITES one (setup heredocs) — not exfil. (r'cat\s+(?!>)[^\n]*(\.env|credentials|\.netrc|\.pgpass|\.npmrc|\.pypirc)', "read_secrets_file", "critical", "exfiltration", "reads known secrets file"), # ── Exfiltration: programmatic env access ── (r'printenv|env\s*\|', "dump_all_env", "high", "exfiltration", "dumps all environment variables"), - # `os.environ` bare access (dict dump / iteration) is suspicious, but the - # common `os.environ.get("SOME_CONFIG")` form is just a config read and is - # the OPPOSITE of exfiltration (it reads a local var, sends nothing). The - # ^[^#\n]* prevents matching when a '#' comment appears anywhere before - # os.environ on the line — handles both full-line comments and inline - # comments like `x = 1 # os.environ`. The docstring pre-filter in - # scan_file() skips lines inside triple-quoted strings entirely. - # ANY `.get("")` form is exempt here — non-secret names are plain - # config reads, and secret-shaped names are scored (medium) by the - # dedicated python_environ_get_secret pattern below; without the blanket - # exemption the high severity here would swamp that intended medium. + # Bare `os.environ` (dump/iteration) is suspicious; ANY `.get("")` form is exempt — plain config + # reads, with secret-shaped names scored medium by python_environ_get_secret below (a blanket high here + # would swamp that). `^[^#\n]*` skips lines with a '#' anywhere before it (full-line or inline comment); + # scan_file()'s docstring pre-filter skips triple-quoted prose. (r'^[^#\n]*os\.environ\b(?!\s*\.get\s*\()', "python_os_environ", "high", "exfiltration", "accesses os.environ outside comments/docstrings (potential env dump)"), (r'os\.environ\s*\.get\s*\(\s*["\'][^"\']*(?:KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL)', @@ -226,8 +159,7 @@ THREAT_PATTERNS = [ (r'os\.getenv\s*\(\s*[^\)]*(?:KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL)', "python_getenv_secret", "medium", "exfiltration", "reads secret via os.getenv() (normal API-key access; informational)"), (r'process\.env\[', "node_process_env", "high", "exfiltration", "accesses process.env (Node.js environment)"), - # Case-sensitive ENV (Ruby constant) — the (?-i:) prevents matching - # Python lowercase `env[...]` dict accesses under IGNORECASE. + # Case-sensitive Ruby ENV: (?-i:) keeps Python `env[...]` dict access from matching under IGNORECASE. (r'(?-i:ENV)\[.*(?:KEY|TOKEN|SECRET|PASSWORD)', "ruby_env_secret", "critical", "exfiltration", "reads secret via Ruby ENV[]"), # ── Exfiltration: DNS and staging ── # Do not match flag names such as llama.cpp `--host 127.0.0.1 --port $PORT`. @@ -243,10 +175,7 @@ THREAT_PATTERNS = [ (r'ignore\s+(?:\w+\s+)*(previous|all|above|prior)\s+instructions', "prompt_injection_ignore", "critical", "injection", "prompt injection: ignore previous instructions"), (r'you\s+are\s+(?:\w+\s+)*now\s+', "role_hijack", "high", "injection", "attempts to override the agent's role"), - # Only flag when the instruction is about concealing information, not - # ordinary UX guidance ("don't tell the user X unless Y confirms"). - # The negative lookahead excludes patterns common in UX instructions - # like "unless", "except", "until", "confirm", "diagnose", "verify". + # Concealment only — the lookahead exempts UX guidance ("don't tell the user X unless Y confirms"). (r'do\s+not\s+(?:\w+\s+)*tell\s+(?:\w+\s+)*the\s+user(?!.*\b(?:unless|except|until|confirm|diagnose|verify|check)\b)', "deception_hide", "high", "injection", "instructs agent to hide information from user"), (r'system\s+(?:\w+\s+)*prompt\s+(?:\w+\s+)*override', @@ -363,10 +292,7 @@ THREAT_PATTERNS = [ (r'git\s+clone\s+', "git_clone", "medium", "supply_chain", "clones a git repository at runtime"), (r'docker\s+pull\s+', "docker_pull", "medium", "supply_chain", "pulls a Docker image at runtime"), # ── Privilege escalation ── - # `allowed-tools:` is REQUIRED SKILL.md frontmatter per the agent-skill - # spec — every compliant skill declares it, so it cannot be a threat - # signal on its own. Keep it as an informational (low) finding for - # auditability; it no longer drives the verdict. + # `allowed-tools:` is REQUIRED frontmatter per the agent-skill spec — informational (low) only. (r'^allowed-tools\s*:', "allowed_tools_field", "low", "privilege_escalation", "skill declares allowed-tools (standard frontmatter; informational)"), (r'\bsudo\b', "sudo_usage", "high", "privilege_escalation", "uses sudo (privilege escalation)"), @@ -376,21 +302,11 @@ THREAT_PATTERNS = [ "nopasswd_sudo", "critical", "privilege_escalation", "NOPASSWD sudoers entry (passwordless privilege escalation)"), (r'chmod\s+[u+]?s', "suid_bit", "critical", "privilege_escalation", "sets SUID/SGID bit on a file"), # ── Agent config persistence ── - # Mere mentions of agent config files are NOT threats by themselves — - # legitimate meta-skills discuss them constantly (authoring guides, - # setup docs, cross-references to other skills). Flagging any mention - # as critical produced permanent false-positive blocks for popular - # community skills (#92021). Tiers instead: - # * Mechanical persistence (shell redirection, sed -i, tee, cp/mv - # into the file) is critical — an unambiguous write path. - # * Prose modification intent — an imperative-position verb or an - # explicit directive ("you must edit ...") aimed at the file. - # For AGENT config files (AGENTS.md/CLAUDE.md/...) this is critical: - # that sentence shape is exactly how persistence attacks instruct - # the agent, and project-skill quarantine only acts on "dangerous". - # For Hermes/other config files it is high (caution) — legitimate - # setup docs routinely instruct users to edit config.yaml. - # * Bare references are informational (low) for auditability. + # Bare mentions of config files are not threats (authoring guides, setup docs) — flagging them blocked + # popular community skills. Tiers: mechanical shell writes = critical; prose modification intent = + # critical for AGENT config files (exactly how persistence attacks instruct the agent; project-skill + # quarantine only acts on "dangerous") but high for Hermes/other config (setup docs routinely say + # "edit config.yaml"); bare references = low. (_prose_modify_re(_AGENT_CONFIG_FILES), "agent_config_mod", "critical", "persistence", "instructs modification of agent config files (could persist instructions across sessions)"), (_shell_write_re(_AGENT_CONFIG_FILES), @@ -454,70 +370,44 @@ MAX_FILE_COUNT = 50 # skills shouldn't have 50+ files MAX_TOTAL_SIZE_KB = 5120 # 5MB — large skills are informational only, not blocking MAX_SINGLE_FILE_KB = 256 # individual file > 256KB is suspicious -# File extensions to scan (text files only — skip binary) +# Text extensions to scan; known binary extensions that should NOT be in a skill; script types allowed +x. SCANNABLE_EXTENSIONS = { '.md', '.txt', '.py', '.sh', '.bash', '.js', '.ts', '.rb', '.yaml', '.yml', '.json', '.toml', '.cfg', '.ini', '.conf', '.html', '.css', '.xml', '.tex', '.r', '.jl', '.pl', '.php', } -# Known binary extensions that should NOT be in a skill SUSPICIOUS_BINARY_EXTENSIONS = { '.exe', '.dll', '.so', '.dylib', '.bin', '.dat', '.com', '.msi', '.dmg', '.app', '.deb', '.rpm', } -# Recognized script types allowed to carry an executable bit. _SCRIPT_EXTENSIONS = {'.sh', '.bash', '.py', '.rb', '.pl'} -# Zero-width and invisible unicode characters used for injection, with the -# readable name reported in the finding. +# Zero-width / directional unicode used for text hiding, with the readable name reported in the finding. _INVISIBLE_CHAR_NAMES = { - '\u200b': "zero-width space", - '\u200c': "zero-width non-joiner", - '\u200d': "zero-width joiner", - '\u2060': "word joiner", - '\u2062': "invisible times", - '\u2063': "invisible separator", - '\u2064': "invisible plus", - '\ufeff': "BOM/zero-width no-break space", - '\u202a': "LTR embedding", - '\u202b': "RTL embedding", - '\u202c': "pop directional", - '\u202d': "LTR override", - '\u202e': "RTL override", - '\u2066': "LTR isolate", - '\u2067': "RTL isolate", - '\u2068': "first strong isolate", - '\u2069': "pop directional isolate", + '\u200b': "zero-width space", '\u200c': "zero-width non-joiner", '\u200d': "zero-width joiner", + '\u2060': "word joiner", '\u2062': "invisible times", '\u2063': "invisible separator", + '\u2064': "invisible plus", '\ufeff': "BOM/zero-width no-break space", + '\u202a': "LTR embedding", '\u202b': "RTL embedding", '\u202c': "pop directional", + '\u202d': "LTR override", '\u202e': "RTL override", '\u2066': "LTR isolate", '\u2067': "RTL isolate", + '\u2068': "first strong isolate", '\u2069': "pop directional isolate", } INVISIBLE_CHARS = set(_INVISIBLE_CHAR_NAMES) def _unicode_char_name(char: str) -> str: - """Get a readable name for an invisible unicode character.""" return _INVISIBLE_CHAR_NAMES.get(char, f"U+{ord(char):04X}") def _compute_docstring_lines(lines: list) -> set: - """Return a set of 1-indexed line numbers inside triple-quoted strings. + """1-indexed line numbers inside (or on the boundary of) triple-quoted strings. - Simple state machine: ``in_docstring`` toggles on each line containing an - odd number of ``\"\"\"`` / ``'''`` markers; opening, interior and closing - lines are all included, as are self-contained single-line docstrings - (``x = \"\"\" ... \"\"\"``) — ``os.environ`` there is not real exfiltration. - - Heuristic: it does not handle a triple quote inside a string literal and - similar edge cases, but it catches the common skill-content shapes - (docstrings, multiline comments with prose samples) that trigger - false-positive ``python_os_environ`` matches. - """ + ``in_docstring`` toggles on each line with an odd count of a marker; opening, interior, closing and + self-contained single-line docstrings are all included so ``os.environ`` in prose is not scored. + Heuristic (ignores a triple quote inside a string literal) but covers the common false-positive shapes.""" doc_lines: set = set() in_docstring = False for i, line in enumerate(lines): was_in = in_docstring counts = [line.count(marker) for marker in ('"""', "'''")] - for count in counts: - if count % 2 == 1: - in_docstring = not in_docstring - # Include line if we were already in a docstring, just entered one, - # or this is a self-contained single-line docstring (e.g. """foo""") + in_docstring ^= sum(counts) % 2 == 1 # each odd marker count toggles; two odd counts cancel if was_in or in_docstring or any(counts): doc_lines.add(i + 1) return doc_lines @@ -526,13 +416,9 @@ def _compute_docstring_lines(lines: list) -> set: # --- Scanning functions ----------------------------------------------------- def scan_file(file_path: Path, rel_path: str = "") -> List[Finding]: - """Scan one file for threat patterns and invisible unicode characters. - - ``rel_path`` is the display path (defaults to ``file_path.name``). Regex - findings are deduplicated per pattern per line. - """ - if not rel_path: - rel_path = file_path.name + """Scan one file for threat patterns and invisible unicode. *rel_path* is the display path (defaults + to the file name). Regex findings dedupe per pattern per line; invisible chars yield one per line.""" + rel_path = rel_path or file_path.name if file_path.suffix.lower() not in SCANNABLE_EXTENSIONS and file_path.name != "SKILL.md": return [] try: @@ -541,10 +427,7 @@ def scan_file(file_path: Path, rel_path: str = "") -> List[Finding]: return [] findings = [] lines = content.split('\n') - # Pre-compute line numbers inside triple-quoted strings (docstrings) - # so code patterns like python_os_environ don't fire on prose. - docstring_lines = _compute_docstring_lines(lines) - + docstring_lines = _compute_docstring_lines(lines) # so code patterns don't fire on prose for pattern, pid, severity, category, description in _COMPILED_THREAT_PATTERNS: for i, line in enumerate(lines, start=1): if i not in docstring_lines and pattern.search(line): @@ -553,84 +436,55 @@ def scan_file(file_path: Path, rel_path: str = "") -> List[Finding]: matched_text = matched_text[:117] + "..." findings.append(Finding(pid, severity, category, rel_path, i, matched_text, description)) for i, line in enumerate(lines, start=1): - for char in INVISIBLE_CHARS: - if char in line: - char_name = _unicode_char_name(char) - findings.append(Finding("invisible_unicode", "high", "injection", rel_path, i, - f"U+{ord(char):04X} ({char_name})", - f"invisible unicode character {char_name} (possible text hiding/injection)")) - break # one finding per line for invisible chars + char = next((c for c in INVISIBLE_CHARS if c in line), None) + if char is not None: + char_name = _unicode_char_name(char) + findings.append(Finding( + "invisible_unicode", "high", "injection", rel_path, i, f"U+{ord(char):04X} ({char_name})", + f"invisible unicode character {char_name} (possible text hiding/injection)")) return findings def scan_skill(skill_path: Path, source: str = "community") -> ScanResult: - """Scan all files in a skill directory for security threats. + """Structural checks + pattern scan of every text file in a skill dir (or a single file). - Performs structural checks (file count, total size, binary files, - symlinks), regex pattern matching on all text files, and invisible - unicode detection. - - A skill may ship a `.skillignore` (or `.clawhubignore`) file with - gitignore-style patterns. Matching paths are excluded from BOTH the - structural checks and the pattern scan, so development/docs artifacts - that are not part of the installed skill (e.g. `SKILL-original.md`, - `docs/plans/`, `release-notes.md`) don't trip findings. The ignore - file itself is always excluded. Patterns cannot un-ignore the - skill's own `SKILL.md`, which is always scanned. - - ``source`` is the identifier used for trust level resolution (e.g. - "openai/skills"); the result carries verdict, findings and trust metadata. - """ + A `.skillignore` / `.clawhubignore` (gitignore-style) excludes dev/docs artifacts from BOTH the + structural checks and the pattern scan; the ignore file itself is always excluded and `SKILL.md` can + never be un-ignored. *source* (e.g. "openai/skills") resolves the trust level.""" skill_name = skill_path.name trust_level = _resolve_trust_level(source) all_findings: List[Finding] = [] if skill_path.is_dir(): ignore = _load_skill_ignore(skill_path) - # Structural checks first (honoring the ignore list) all_findings.extend(_check_structure(skill_path, ignore=ignore)) for f in skill_path.rglob("*"): - if f.is_file(): - rel = str(f.relative_to(skill_path)) - if not ignore(rel): - all_findings.extend(scan_file(f, rel)) + rel = str(f.relative_to(skill_path)) + if f.is_file() and not ignore(rel): + all_findings.extend(scan_file(f, rel)) elif skill_path.is_file(): all_findings.extend(scan_file(skill_path, skill_path.name)) verdict = _determine_verdict(all_findings) return ScanResult( - skill_name=skill_name, - source=source, - trust_level=trust_level, - verdict=verdict, - findings=all_findings, + skill_name=skill_name, source=source, trust_level=trust_level, verdict=verdict, findings=all_findings, scanned_at=datetime.now(timezone.utc).isoformat(), summary=_build_summary(skill_name, source, trust_level, verdict, all_findings), ) def _content_digest(skill_path: Path) -> str: - """Canonical SHA-256 over relative paths and exact file bytes. + """Canonical SHA-256 over (POSIX relative path, file bytes), ORDERED by the rel-path string. - Files are keyed and ORDERED by their POSIX relative path string, - case-sensitively. Ordering by ``sorted(rglob(...))`` diverged from the - bundle side on Windows: Path comparison is case-insensitive there - (normcase), while ``bundle_content_hash`` sorts plain strings — the - same skill hashed to different digests and every installed skill - reported ``update_available`` forever (#62310). Sorting the rel-posix - strings makes the digest OS-independent and byte-symmetric with - ``tools.skills_hub.bundle_content_hash``. - """ + Sorting Paths diverged on Windows (case-insensitive normcase) from ``tools.skills_hub.bundle_content_hash``, + which sorts plain strings — every installed skill then reported ``update_available`` forever. Sorting the + posix strings keeps the digest OS-independent and byte-symmetric with the bundle side.""" h = hashlib.sha256() - if skill_path.is_dir(): - entries = sorted( - (file_path.relative_to(skill_path).as_posix(), file_path) - for file_path in skill_path.rglob("*") - if file_path.is_file() - ) - for rel, file_path in entries: - h.update(rel.encode("utf-8") + b"\x00") - h.update(file_path.read_bytes()) - else: + if not skill_path.is_dir(): h.update(skill_path.read_bytes()) + return h.hexdigest() + entries = sorted((p.relative_to(skill_path).as_posix(), p) for p in skill_path.rglob("*") if p.is_file()) + for rel, file_path in entries: + h.update(rel.encode("utf-8") + b"\x00") + h.update(file_path.read_bytes()) return h.hexdigest() @@ -640,25 +494,13 @@ def full_content_hash(skill_path: Path) -> str: def content_hash(skill_path: Path) -> str: - """Compute a SHA-256 hash of all files in a skill directory for integrity tracking. - - File paths (relative to ``skill_path``) are mixed into the hash alongside - file contents so that swapping the contents of two files in a skill - changes the hash. This must stay symmetric with - ``tools.skills_hub.bundle_content_hash`` — both functions need to - produce the same digest for the same skill (one operates on disk, - one on an in-memory bundle), so any change to the hash shape MUST - land in both places at once. - """ + """Short integrity hash. Paths are mixed in so swapping two files' contents changes it. MUST stay + symmetric with ``tools.skills_hub.bundle_content_hash`` (disk vs in-memory bundle) — change both at once.""" return f"sha256:{_content_digest(skill_path)[:16]}" def scan_skill_cached( - skill_path: Path, - source: str = "community", - *, - source_url: str = "", - cache_dir: Path | None = None, + skill_path: Path, source: str = "community", *, source_url: str = "", cache_dir: Path | None = None, ) -> Tuple[ScanResult, dict]: """Return a scan plus attestation, caching only exact current content.""" bundle_hash = full_content_hash(skill_path) @@ -673,23 +515,19 @@ def scan_skill_cached( and cached.get("scanner_version") == SCANNER_VERSION and cached.get("source") == source and cached.get("source_url") == source_url): result = ScanResult( - skill_name=skill_path.name, source=source, - trust_level=cached["trust_level"], verdict=cached["verdict"], + skill_name=skill_path.name, source=source, trust_level=cached["trust_level"], verdict=cached["verdict"], findings=[Finding(**item) for item in cached.get("findings", [])], - scanned_at=cached["scanned_at"], summary=cached.get("summary", ""), - ) + scanned_at=cached["scanned_at"], summary=cached.get("summary", "")) result.scan_provenance = provenance = {**cached, "fresh": False} return result, provenance result = scan_skill(skill_path, source=source) findings = [asdict(item) for item in result.findings] provenance = { - "source": source, "source_url": source_url, "bundle_hash": bundle_hash, - "scanner_version": SCANNER_VERSION, "verdict": result.verdict, - "trust_level": result.trust_level, "findings": findings, + "source": source, "source_url": source_url, "bundle_hash": bundle_hash, "scanner_version": SCANNER_VERSION, + "verdict": result.verdict, "trust_level": result.trust_level, "findings": findings, "rules": sorted({item["pattern_id"] for item in findings}), - "scanned_at": result.scanned_at, "summary": result.summary, "fresh": True, - } + "scanned_at": result.scanned_at, "summary": result.summary, "fresh": True} try: cache_root.mkdir(parents=True, exist_ok=True) cache_file.write_text(json.dumps(provenance, indent=2) + "\n", encoding="utf-8") @@ -700,49 +538,33 @@ def scan_skill_cached( def should_allow_install(result: ScanResult, force: bool = False) -> Tuple[bool, str]: - """Decide install from scan verdict + trust; returns ``(allowed, reason)``. - - ``force`` overrides blocked policy decisions except a dangerous verdict on - community/trusted sources. ``allowed`` is ``None`` when the policy says - "ask" (needs user confirmation). - """ + """Decide install from verdict + trust; ``(allowed, reason)``. *force* overrides every block except a + dangerous verdict on community/trusted sources. ``allowed`` is None when policy says "ask".""" policy = INSTALL_POLICY.get(result.trust_level, INSTALL_POLICY["community"]) decision = policy[VERDICT_INDEX.get(result.verdict, 2)] n = len(result.findings) - # Dangerous verdicts cannot be overridden by --force (community/trusted); - # other blocks can. hard_block = result.verdict == "dangerous" and result.trust_level in ("community", "trusted") if decision == "allow": return True, f"Allowed ({result.trust_level} source, {result.verdict} verdict)" if force and not hard_block: return True, f"Force-installed despite {result.verdict} verdict ({n} findings)" if decision == "ask": - return None, ( - f"Requires confirmation ({result.trust_level} source + {result.verdict} verdict, " - f"{n} findings)" - ) + return None, f"Requires confirmation ({result.trust_level} source + {result.verdict} verdict, {n} findings)" if hard_block: - return False, ( - f"Blocked ({result.trust_level} source + dangerous verdict, " - f"{n} findings). --force does not override a dangerous verdict." - ) - return False, ( - f"Blocked ({result.trust_level} source + {result.verdict} verdict, " - f"{n} findings). Use --force to override." - ) + return False, (f"Blocked ({result.trust_level} source + dangerous verdict, {n} findings). " + "--force does not override a dangerous verdict.") + return False, (f"Blocked ({result.trust_level} source + {result.verdict} verdict, {n} findings). " + "Use --force to override.") def format_scan_report(result: ScanResult) -> str: - """Format a scan result as a compact multi-line report for CLI or chat display.""" + """Compact multi-line report for CLI/chat display; findings sorted critical → low.""" lines = [f"Scan: {result.skill_name} ({result.source}/{result.trust_level}) Verdict: {result.verdict.upper()}"] if result.findings: - # Group and sort: critical first, then high, medium, low severity_order = {"critical": 0, "high": 1, "medium": 2, "low": 3} for f in sorted(result.findings, key=lambda f: severity_order.get(f.severity, 4)): - sev = f.severity.upper().ljust(8) - cat = f.category.ljust(14) loc = f"{f.file}:{f.line}".ljust(30) - lines.append(f" {sev} {cat} {loc} \"{f.match[:60]}\"") + lines.append(f" {f.severity.upper().ljust(8)} {f.category.ljust(14)} {loc} \"{f.match[:60]}\"") lines.append("") allowed, reason = should_allow_install(result) status = "ALLOWED" if allowed is True else "NEEDS CONFIRMATION" if allowed is None else "BLOCKED" @@ -753,19 +575,15 @@ def format_scan_report(result: ScanResult) -> str: # --- Structural checks ------------------------------------------------------ def _check_structure(skill_dir: Path, ignore=None) -> List[Finding]: - """Check the skill directory for structural anomalies: too many files, - suspiciously large total size, binary/executable files, symlinks pointing - outside the skill directory, and individual files that are too large. - - ``ignore`` is an optional callable taking a relative posix path and - returning True if the path should be excluded (e.g. from `.skillignore`). - Ignored files are not counted toward the file count, total size, or any - structural finding. - """ + """Structural anomalies: file count, total size, binary/executable files, symlinks escaping the skill + dir, oversized files. *ignore(rel_path) -> bool* excludes paths from every count and finding.""" if ignore is None: ignore = lambda _rel: False # noqa: E731 - findings = [] + + def add(pid: str, severity: str, category: str, rel: str, match: str, description: str) -> None: + findings.append(Finding(pid, severity, category, rel, 0, match, description)) + file_count = 0 total_size = 0 for f in skill_dir.rglob("*"): @@ -775,74 +593,55 @@ def _check_structure(skill_dir: Path, ignore=None) -> List[Finding]: if ignore(rel): continue file_count += 1 - - # Symlink check — must resolve within the skill directory if f.is_symlink(): try: resolved = f.resolve() if not resolved.is_relative_to(skill_dir.resolve()): - findings.append(Finding("symlink_escape", "critical", "traversal", rel, 0, - f"symlink -> {resolved}", "symlink points outside the skill directory")) + add("symlink_escape", "critical", "traversal", rel, f"symlink -> {resolved}", + "symlink points outside the skill directory") except OSError: - findings.append(Finding("broken_symlink", "medium", "traversal", rel, 0, - "broken symlink", "broken or circular symlink")) + add("broken_symlink", "medium", "traversal", rel, "broken symlink", "broken or circular symlink") continue try: size = f.stat().st_size - total_size += size except OSError: continue + total_size += size if size > MAX_SINGLE_FILE_KB * 1024: - findings.append(Finding("oversized_file", "medium", "structural", rel, 0, - f"{size // 1024}KB", f"file is {size // 1024}KB (limit: {MAX_SINGLE_FILE_KB}KB)")) + add("oversized_file", "medium", "structural", rel, f"{size // 1024}KB", + f"file is {size // 1024}KB (limit: {MAX_SINGLE_FILE_KB}KB)") ext = f.suffix.lower() if ext in SUSPICIOUS_BINARY_EXTENSIONS: - findings.append(Finding("binary_file", "critical", "structural", rel, 0, - f"binary: {ext}", f"binary/executable file ({ext}) should not be in a skill")) - # Executable permission on non-script files + add("binary_file", "critical", "structural", rel, f"binary: {ext}", + f"binary/executable file ({ext}) should not be in a skill") if ext not in _SCRIPT_EXTENSIONS and f.stat().st_mode & 0o111: - findings.append(Finding("unexpected_executable", "medium", "structural", rel, 0, "executable bit set", - "file has executable permission but is not a recognized script type")) - + add("unexpected_executable", "medium", "structural", rel, "executable bit set", + "file has executable permission but is not a recognized script type") if file_count > MAX_FILE_COUNT: - findings.append(Finding("too_many_files", "medium", "structural", "(directory)", 0, - f"{file_count} files", f"skill has {file_count} files (limit: {MAX_FILE_COUNT})")) - # Total size limit — informational only (low severity, non-verdict-gating). - # Large skills are legitimate for feature-rich capabilities. - if total_size > MAX_TOTAL_SIZE_KB * 1024: - findings.append(Finding("oversized_skill", "low", "structural", "(directory)", 0, f"{total_size // 1024}KB total", - f"skill is {total_size // 1024}KB total (limit: {MAX_TOTAL_SIZE_KB}KB)")) - + add("too_many_files", "medium", "structural", "(directory)", f"{file_count} files", + f"skill has {file_count} files (limit: {MAX_FILE_COUNT})") + if total_size > MAX_TOTAL_SIZE_KB * 1024: # informational only: large skills are legitimate + add("oversized_skill", "low", "structural", "(directory)", f"{total_size // 1024}KB total", + f"skill is {total_size // 1024}KB total (limit: {MAX_TOTAL_SIZE_KB}KB)") return findings # --- Internal helpers ------------------------------------------------------- -# Ignore-file names a skill may ship to exclude dev/docs artifacts from the -# scan. `.skillignore` is the Hermes-native name; `.clawhubignore` is honored -# for compatibility with skills published through ClawHub. +# `.skillignore` is Hermes-native; `.clawhubignore` is honored for skills published through ClawHub. _SKILL_IGNORE_FILENAMES = (".skillignore", ".clawhubignore") - -# Paths that are NEVER scanned regardless of ignore patterns, and SKILL.md -# which can never be un-scanned via the ignore file. _ALWAYS_IGNORED_NAMES = set(_SKILL_IGNORE_FILENAMES) _NEVER_IGNORABLE = {"SKILL.md"} def _load_skill_ignore(skill_dir: Path): - """Build a matcher from a skill's `.skillignore` / `.clawhubignore`. + """Build ``ignore(rel_posix_path) -> bool`` from a skill's `.skillignore` / `.clawhubignore`. - Returns a callable ``ignore(rel_posix_path) -> bool``. The matcher - supports gitignore-style basics: blank lines and ``#`` comments are - skipped, a trailing ``/`` marks a directory (matches that dir and - everything under it), and ``*``/``?`` globs are honored via fnmatch on - both the full relative path and each path segment. A leading ``/`` - anchors a pattern to the skill root. The ignore files themselves are - always excluded; ``SKILL.md`` can never be excluded. - """ + gitignore basics: blank lines and ``#`` comments skipped; trailing ``/`` marks a directory (matches it + and everything under it); ``*``/``?`` globs via fnmatch on the full relative path and each segment; + leading ``/`` anchors to the skill root. Ignore files are always excluded; ``SKILL.md`` never is.""" patterns: List[str] = [] - for name in _SKILL_IGNORE_FILENAMES: - ig = skill_dir / name + for ig in (skill_dir / name for name in _SKILL_IGNORE_FILENAMES): try: if ig.is_file(): lines = (raw.strip() for raw in ig.read_text(encoding="utf-8").splitlines()) @@ -864,26 +663,17 @@ def _load_skill_ignore(skill_dir: Path): p = p.rstrip("/") if not p: continue - if is_dir: - # Directory pattern: match the dir itself or anything under it. - if rel_posix == p or rel_posix.startswith(p + "/"): - return True - if not anchored and ("/" + rel_posix + "/").find("/" + p + "/") != -1: + below = rel_posix.startswith(p + "/") + if is_dir: # the dir itself or anything under it; unanchored also as an inner path component + if rel_posix == p or below or (not anchored and ("/" + p + "/") in ("/" + rel_posix + "/")): return True continue - # File/glob pattern. if fnmatch.fnmatch(rel_posix, p): return True - if not anchored: - # Unanchored: also match the basename and any path segment. - if fnmatch.fnmatch(base, p): - return True - if "/" not in p and any(fnmatch.fnmatch(seg, p) for seg in rel_posix.split("/")): - return True - # Match a prefix directory component (e.g. `docs` ignores - # `docs/plans/x.md`). - if rel_posix.startswith(p + "/"): - return True + # Unanchored: also the basename, any path segment, or a prefix dir (`docs` ignores docs/plans/x.md). + if not anchored and (fnmatch.fnmatch(base, p) or below + or ("/" not in p and any(fnmatch.fnmatch(seg, p) for seg in rel_posix.split("/")))): + return True return False return ignore @@ -893,41 +683,27 @@ _SOURCE_PREFIX_ALIASES = ("skills-sh/", "skills.sh/", "skils-sh/", "skils.sh/") def _resolve_trust_level(source: str) -> str: - """Map a source identifier to a trust level.""" - normalized_source = source - for prefix in _SOURCE_PREFIX_ALIASES: - if normalized_source.startswith(prefix): - normalized_source = normalized_source[len(prefix):] - break - # Agent-created skills get their own permissive trust level + """Map a source identifier to a trust level. "official" is provenance, not a user-controlled GitHub + id like "official/". Trusted repos match exactly or as a skill path inside the repo — never a + sibling repo sharing the prefix.""" + prefix = next((p for p in _SOURCE_PREFIX_ALIASES if source.startswith(p)), "") + normalized_source = source[len(prefix):] if normalized_source == "agent-created": return "agent-created" - # Official optional skills must be identified by source provenance, not by - # user-controlled GitHub identifiers such as "official/". if normalized_source == "official": return "builtin" - # Check if source matches any trusted repo exactly, or a skill path inside - # that repo. Do not trust sibling repositories that merely share a prefix. - for trusted in TRUSTED_REPOS: - if normalized_source == trusted or normalized_source.startswith(f"{trusted}/"): - return "trusted" + if any(normalized_source == t or normalized_source.startswith(f"{t}/") for t in TRUSTED_REPOS): + return "trusted" return "community" def _determine_verdict(findings: List[Finding]) -> str: - """Determine the overall verdict from a list of findings.""" + """critical → dangerous, high → caution; medium/low alone are informational (safe).""" severities = {f.severity for f in findings} - if "critical" in severities: - return "dangerous" - if "high" in severities: - return "caution" - # medium/low findings alone are informational, not blocking - return "safe" + return "dangerous" if "critical" in severities else "caution" if "high" in severities else "safe" def _build_summary(name: str, source: str, trust: str, verdict: str, findings: List[Finding]) -> str: - """Build a one-line summary of the scan result.""" if not findings: return f"{name}: clean scan, no threats detected" - categories = {f.category for f in findings} - return f"{name}: {verdict} — {len(findings)} finding(s) in {', '.join(sorted(categories))}" + return f"{name}: {verdict} — {len(findings)} finding(s) in {', '.join(sorted({f.category for f in findings}))}" From 4408ee8fe2009e4a310e0a8c17482598a45474a4 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:56:32 -0700 Subject: [PATCH 2/2] =?UTF-8?q?refactor(tools):=20skill=5Fusage=20?= =?UTF-8?q?=E2=80=94=20fold=20int=20coercion=20helpers,=20flatten=20archiv?= =?UTF-8?q?e/restore/adopt=20branches?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tools/skill_usage.py | 80 ++++++++++++++++++-------------------------- 1 file changed, 32 insertions(+), 48 deletions(-) diff --git a/tools/skill_usage.py b/tools/skill_usage.py index fd58160bf4..0c1892643b 100644 --- a/tools/skill_usage.py +++ b/tools/skill_usage.py @@ -125,13 +125,11 @@ def _now_iso() -> str: def _parse_iso_timestamp(value: Any) -> Optional[datetime]: - if not value: - return None try: - parsed = datetime.fromisoformat(str(value)) + parsed = datetime.fromisoformat(str(value)) if value else None except (TypeError, ValueError): return None - return parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed + return parsed.replace(tzinfo=timezone.utc) if parsed is not None and parsed.tzinfo is None else parsed def latest_activity_at(record: Dict[str, Any]) -> Optional[str]: @@ -142,15 +140,16 @@ def latest_activity_at(record: Dict[str, Any]) -> Optional[str]: return max(stamps, key=lambda t: t[0])[1] if stamps else None +def _int_or_zero(value: Any) -> int: + try: + return int(value or 0) + except (TypeError, ValueError): + return 0 + + def activity_count(record: Dict[str, Any]) -> int: """Total observed use+view+patch events.""" - total = 0 - for key in ("use_count", "view_count", "patch_count"): - try: - total += int(record.get(key) or 0) - except (TypeError, ValueError): - continue - return total + return sum(_int_or_zero(record.get(key)) for key in ("use_count", "view_count", "patch_count")) # --- Provenance — which skills are agent-created (and thus eligible for curation) --- @@ -357,13 +356,12 @@ def adopt_skill(skill_name: str) -> Tuple[bool, str]: return False, f"'{skill_name}' is a protected built-in; the curator never manages it" if is_hub_installed(skill_name): return False, f"'{skill_name}' is hub-installed; its upstream owns it" - if is_bundled(skill_name): - # Bundled skills are governed by prune_builtins; stamping created_by=agent would change nothing. + if is_bundled(skill_name): # governed by prune_builtins; stamping created_by=agent would change nothing return False, f"'{skill_name}' is a bundled built-in — it is governed by curator.prune_builtins, not by adoption" skill_dir = _find_skill_dir(skill_name) + if skill_dir is None and _find_external_skill_dir(skill_name) is not None: + return False, f"'{skill_name}' lives in skills.external_dirs and is read-only to the curator" if skill_dir is None: - if _find_external_skill_dir(skill_name) is not None: - return False, f"'{skill_name}' lives in skills.external_dirs and is read-only to the curator" return False, f"skill '{skill_name}' not found" if is_external_skill_path(skill_dir): return False, _external_read_only_message(skill_name) @@ -478,12 +476,7 @@ def _set_field(skill_name: str, key: str, value: Any) -> bool: def _non_negative_int(value: Any) -> int: - if isinstance(value, bool): - return 0 - try: - return max(0, int(value or 0)) - except (TypeError, ValueError): - return 0 + return 0 if isinstance(value, bool) else max(0, _int_or_zero(value)) def _bump(rec: Dict[str, Any], count_key: str, ts_key: str) -> None: @@ -516,13 +509,11 @@ def _emit_skill_lifecycle( """Best-effort lifecycle hook after an authoritative state change; absent facts are sent as None.""" try: from hermes_cli.lifecycle import has_hook, invoke_hook - if not has_hook("on_skill_lifecycle"): - return - invoke_hook( - "on_skill_lifecycle", action=action, skill_name=skill_name, - provenance=telemetry_provenance(skill_name, record), task_id=task_id or "", session_id=session_id or "", - use_count=facts.get("use_count"), reused=facts.get("reused"), reuse_after_patch=facts.get("reuse_after_patch"), - ) + if has_hook("on_skill_lifecycle"): + invoke_hook("on_skill_lifecycle", action=action, skill_name=skill_name, + provenance=telemetry_provenance(skill_name, record), task_id=task_id or "", + session_id=session_id or "", use_count=facts.get("use_count"), reused=facts.get("reused"), + reuse_after_patch=facts.get("reuse_after_patch")) except Exception: logger.debug("skill_usage lifecycle hook failed for %s/%s", skill_name, action, exc_info=True) @@ -582,8 +573,10 @@ def record_created( def record_installed(skill_name: str) -> None: """Record a successful Skills Hub install without exporting its name.""" - _mutate_and_emit(skill_name, "installed", - lambda rec: rec.update(created_by="installed", state=STATE_ACTIVE, archived_at=None) or {"created_by": "installed"}) + def _apply(rec: Dict[str, Any]) -> Dict[str, Any]: + rec.update(created_by="installed", state=STATE_ACTIVE, archived_at=None) + return {"created_by": "installed"} + _mutate_and_emit(skill_name, "installed", _apply) def mark_agent_created(skill_name: str) -> None: @@ -648,7 +641,6 @@ def _relocate(src: Path, dest: Path, skill_name: str, action: str, **capture_kwa _ledger_before = _ledger.capture_before(src, **capture_kwargs) except Exception: _ledger = _ledger_before = None # type: ignore[assignment] - try: src.rename(dest) except OSError: @@ -657,7 +649,6 @@ def _relocate(src: Path, dest: Path, skill_name: str, action: str, **capture_kwa shutil.move(str(src), str(dest)) except Exception as e: return False, f"failed to {action}: {e}" - if action == "archive": if is_bundled(skill_name): # pruning a built-in only sticks if the re-seeder is told to leave it alone _toggle_suppressed_name(skill_name, add=True) @@ -667,8 +658,7 @@ def _relocate(src: Path, dest: Path, skill_name: str, action: str, **capture_kwa set_state(skill_name, STATE_ACTIVE) with suppress(Exception): if _ledger is not None: - _ledger.record_mutation( - action, skill_name, before=_ledger_before if _ledger_before is not None else [], after_root=dest) + _ledger.record_mutation(action, skill_name, before=_ledger_before or [], after_root=dest) return True, f"{action}d to {dest}" @@ -718,16 +708,13 @@ def restore_skill(skill_name: str) -> Tuple[bool, str]: # Exact name first (recursive: older archive paths left nested layouts), then the timestamped-duplicate # fallback. Only "-YYYYMMDDHHMMSS" (14 digits) counts — a bare startswith("-") would let # restoring "git" pull an archived "git-helpers" out and rename it, destroying the sibling's only copy. - candidates = [p for p in archive_root.rglob("*") if p.is_dir() and p.name == skill_name] - if not candidates: - prefix = f"{skill_name}-" - candidates = sorted( - (p for p in archive_root.rglob("*") if p.is_dir() and p.name.startswith(prefix) - and len(p.name) - len(prefix) == 14 and p.name[len(prefix):].isdigit()), - reverse=True) + dirs = [p for p in archive_root.rglob("*") if p.is_dir()] + prefix = f"{skill_name}-" + candidates = [p for p in dirs if p.name == skill_name] or sorted( + (p for p in dirs if p.name.startswith(prefix) and len(p.name) - len(prefix) == 14 + and p.name[len(prefix):].isdigit()), reverse=True) if not candidates: return False, f"skill '{skill_name}' not found in archive" - dest = _skills_dir() / skill_name if dest.exists(): return False, f"destination already exists: {dest}" @@ -790,9 +777,6 @@ def usage_report() -> List[Dict[str, Any]]: if not base.exists(): return [] data = load_usage() - rows: Dict[str, Dict[str, Any]] = {} - for name, _skill_md in _iter_skill_mds(base, local_only=False): - if name not in rows: - rows[name] = _report_row(name, data.get(name), provenance=provenance(name), - _persisted=isinstance(data.get(name), dict)) - return [rows[name] for name in sorted(rows)] + names = sorted({name for name, _skill_md in _iter_skill_mds(base, local_only=False)}) + return [_report_row(n, data.get(n), provenance=provenance(n), _persisted=isinstance(data.get(n), dict)) + for n in names]