ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity

Merge branch 'simp/r3-34-C' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 23:12:39 -07:00
parent bab76177f0 4408ee8fe2
commit b33efc5386
5 changed files with 369 additions and 864 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/skill_provenance.py
+6 -19
View File
@@ -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
tools/skill_usage.py
+157 -345
View File
@@ -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,59 +125,49 @@ 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:
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]:
"""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 _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:
"""Return the total observed activity count across 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
"""Total observed use+view+patch events."""
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) ---
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 +192,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 +206,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 +224,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/<skill>/``) 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 +282,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 +301,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,89 +314,54 @@ 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):
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 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.
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)
@@ -474,13 +378,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 +391,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 +421,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 +444,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 +457,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,19 +472,11 @@ 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:
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:
@@ -603,7 +485,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,44 +505,35 @@ 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"):
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)
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 +542,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 +561,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"]}
@@ -708,21 +575,18 @@ 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"]}
return {"created_by": "installed"}
_mutate_and_emit(skill_name, "installed", _apply)
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,52 +603,44 @@ 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)
except Exception:
_ledger = _ledger_before = None # type: ignore[assignment]
try:
src.rename(dest)
except OSError:
@@ -793,30 +649,22 @@ 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":
# 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
)
_ledger.record_mutation(action, skill_name, before=_ledger_before or [], 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,55 +684,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/<skill>/" 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/<category>/<skill>/) left behind by older archive paths or external imports.
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
# ("<skill>-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,
)
# Exact name first (recursive: older archive paths left nested layouts), then the timestamped-duplicate
# fallback. Only "<skill>-YYYYMMDDHHMMSS" (14 digits) counts — a bare startswith("<skill>-") would let
# restoring "git" pull an archived "git-helpers" out and rename it, destroying the sibling's only copy.
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}"
@@ -896,22 +726,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>/SKILL.md) and category-nested
(~/.hermes/skills/<category>/<skill>/SKILL.md) layouts. Uses the gated index iterator so M2 org mirrors resolve
ONLY for the active org (stale ``_org/<other>/`` 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/<other>/`` 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 +748,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,24 +766,17 @@ 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 []
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]
tools/skillevaluator_scan.py
+33 -84
View File
@@ -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)
tools/skills_ast_audit.py
+30 -49
View File
@@ -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__(<computed>)
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, <computed>)
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__[<computed>]
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__[<computed>] — dynamic attribute access"))
hit(node, "dict_access", "__dict__[<computed>] — 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)
tools/skills_guard.py
+143 -367
View File
@@ -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 "<file> 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.
"""
""""<file> 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 <secrets-file>` (reading credentials) but NOT `cat > <file>`
# or `cat >> <file>`, 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 <secrets-file>` 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("<name>")` 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("<name>")` 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/<repo>". 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/<repo>".
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}))}"