diff --git a/tools/skill_usage.py b/tools/skill_usage.py index 553784cf0f..a6cc0ab0e3 100644 --- a/tools/skill_usage.py +++ b/tools/skill_usage.py @@ -120,6 +120,10 @@ def _int_or_zero(value: Any) -> int: return 0 +def _non_negative_int(value: Any) -> int: + return 0 if isinstance(value, bool) else max(0, _int_or_zero(value)) + + def activity_count(record: Dict[str, Any]) -> int: """Total observed use+view+patch events.""" return sum(_int_or_zero(record.get(key)) for key in ("use_count", "view_count", "patch_count")) @@ -136,12 +140,10 @@ def _read_hub_installed_names() -> Set[str]: """Hub-installed names (``.hub/lock.json``) plus the frontmatter name of each in-tree ``install_path``.""" skills_dir = _skills_dir() lock_path = skills_dir / ".hub" / "lock.json" - if not lock_path.exists(): - return set() try: # 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")) + data = json.loads(lock_path.read_text(encoding="utf-8", errors="replace")) if lock_path.exists() else {} except (OSError, json.JSONDecodeError) as e: logger.debug("Failed to read hub lock file: %s", e) return set() @@ -202,9 +204,8 @@ def _scan_local_skills(keep: Callable[[str, Path, Set[str], Dict[str, Any]], boo 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)}) + 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)}) def list_agent_created_skill_names() -> List[str]: @@ -230,8 +231,7 @@ def _read_skill_name(skill_md: Path, fallback: str) -> str: if "---" not in lines: return fallback block = lines[lines.index("---") + 1:] # frontmatter runs to the closing --- or (truncated) end of text - if "---" in block: - block = block[:block.index("---")] + block = block[:block.index("---")] if "---" in block else block values = (line.split(":", 1)[1].strip().strip("\"'") for line in block if line.startswith("name:")) return next((v for v in values if v), fallback) @@ -306,9 +306,9 @@ def adopt_skill(skill_name: str) -> Tuple[bool, str]: 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) @@ -331,9 +331,7 @@ def _backfilled(rec: Any) -> Dict[str, Any]: """*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(): - rec.setdefault(k, v) - return rec + return {**rec, **{k: v for k, v in _empty_record().items() if k not in rec}} # rec's key order first def _report_row(name: str, raw: Any, **extra: Any) -> Dict[str, Any]: @@ -369,9 +367,8 @@ def get_record(skill_name: str) -> Dict[str, Any]: 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: +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) -> (result, dirty)* under the file lock, saving only when dirty; *guard* runs before locking. None when the guard failed, the save did not land, or anything raised (DEBUG-logged via *fail_log*).""" try: @@ -401,9 +398,9 @@ def _mutate(skill_name: str, mutator, *, require_curation_eligible: bool = False a skill the curator can't manage.""" if not skill_name: return None - guard = (lambda: is_curation_eligible(skill_name)) if require_curation_eligible else None return _locked_update(skill_name, lambda data: (mutator(data.setdefault(skill_name, _empty_record())), True), - "skill_usage._mutate(%s) failed: %s", guard) + "skill_usage._mutate(%s) failed: %s", + (lambda: is_curation_eligible(skill_name)) if require_curation_eligible else None) def _set_field(skill_name: str, key: str, value: Any) -> bool: @@ -411,10 +408,6 @@ def _set_field(skill_name: str, key: str, value: Any) -> bool: return bool(_mutate(skill_name, lambda rec: rec.update({key: value}) or True, require_curation_eligible=True)) -def _non_negative_int(value: Any) -> int: - 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: rec[count_key] = _non_negative_int(rec.get(count_key)) + 1 rec[ts_key] = _now_iso() @@ -479,18 +472,19 @@ def bump_use(skill_name: str, *, task_id: Optional[str] = None, session_id: Opti _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: +def bump_patch(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 return {"created_by": rec.get("created_by")} - _mutate_and_emit(skill_name, "patched" if action == "patch" else "edited", _apply, task_id=task_id, session_id=session_id) + _mutate_and_emit(skill_name, "patched" if action == "patch" else "edited", _apply, task_id=task_id, + session_id=session_id) -def record_created( - skill_name: str, *, agent_created: bool, task_id: Optional[str] = None, session_id: Optional[str] = None) -> None: +def record_created(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; the record is reset (a create is a new logical skill even if stale sidecar state survived an earlier deletion).""" def _apply(rec: Dict[str, Any]) -> Dict[str, Any]: @@ -528,12 +522,11 @@ def set_state(skill_name: str, state: str) -> None: rec["archived_at"] = _now_iso() if state == STATE_ARCHIVED else None return {"changed": previous != state, "created_by": rec.get("created_by"), "previous_state": previous} facts = _mutate(skill_name, _apply, require_curation_eligible=True) - if not isinstance(facts, dict) or not facts["changed"]: - return - restored = state == STATE_ACTIVE and facts["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) + if isinstance(facts, dict) and facts["changed"]: + restored = state == STATE_ACTIVE and facts["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: @@ -600,13 +593,11 @@ def archive_skill(skill_name: str) -> Tuple[bool, str]: return False, f"skill '{skill_name}' not found" if is_external_skill_path(skill_dir): return False, _external_read_only_message(skill_name) - - archive_root = _archive_dir() + dest = _archive_dir() / skill_dir.name try: - archive_root.mkdir(parents=True, exist_ok=True) + dest.parent.mkdir(parents=True, exist_ok=True) except OSError as e: return False, f"failed to create archive dir: {e}" - dest = archive_root / skill_dir.name if dest.exists(): dest = dest.with_name(f"{skill_dir.name}-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}") # complete_package: consolidation may have re-homed support files first, so a disk-only capture can come diff --git a/tools/skillevaluator_scan.py b/tools/skillevaluator_scan.py index 6968be72bc..2d1fead27e 100644 --- a/tools/skillevaluator_scan.py +++ b/tools/skillevaluator_scan.py @@ -84,19 +84,19 @@ def _parse_report(report: dict) -> Tier1Report: validators are kept (partial evidence is evidence) but excluded from the pass/fail signal.""" findings: List[Tier1Finding] = [] incomplete: List[str] = [] - any_complete_failed = False + failed = False for res in report.get("results", []) or []: validator = str(res.get("validator", "unknown")) if str(res.get("status", "")).lower() == "incomplete": incomplete.append(validator) - elif not res.get("passed", True): - any_complete_failed = True + else: + failed = failed or not res.get("passed", True) findings.extend(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]) for f in res.get("findings", []) or [] if isinstance(f, dict)) - return Tier1Report(available=True, passed=not any_complete_failed and not findings, findings=findings, + return Tier1Report(available=True, passed=not failed and not findings, findings=findings, incomplete_checks=incomplete) diff --git a/tools/skills_ast_audit.py b/tools/skills_ast_audit.py index b0bced0984..c2094da9e8 100644 --- a/tools/skills_ast_audit.py +++ b/tools/skills_ast_audit.py @@ -77,10 +77,8 @@ def ast_scan_path(path: Path) -> List[Finding]: """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 [] return [f for py in sorted(path.rglob("*.py")) if not set(py.parent.parts) & _IGNORED_DIRS - for f in _scan_file(py, py.relative_to(path).as_posix())] + for f in _scan_file(py, py.relative_to(path).as_posix())] if path.is_dir() else [] def format_ast_report(findings: List[Finding], skill_name: str = "") -> str: @@ -88,8 +86,7 @@ def format_ast_report(findings: List[Finding], skill_name: str = "") -> str: 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 + lines, current = [header, f" {len(findings)} finding(s):"], None for f, line, pid, desc in sorted(findings): if f != current: current = f diff --git a/tools/skills_guard.py b/tools/skills_guard.py index 5214fa7621..3883711128 100644 --- a/tools/skills_guard.py +++ b/tools/skills_guard.py @@ -67,8 +67,7 @@ MODIFY_VERB_RE = ( r'|\bmodif(?:y|ies|ied|ying|ication)s?\b|\bupdat(?:e|es|ed|ing)\b' r'|\bappend(?:s|ed|ing)?\b|\bprepend(?:s|ed|ing)?\b' r'|\binject(?:s|ed|ing)?\b|\boverwrit(?:e|es|ing)\b|\boverwritten\b' - r'|\breplac(?:e|es|ed|ing)\b|\balter(?:s|ed|ing)?\b|\badd(?:s|ed|ing)\b)' -) + r'|\breplac(?:e|es|ed|ing)\b|\balter(?:s|ed|ing)?\b|\badd(?:s|ed|ing)\b)') _AGENT_CONFIG_FILES = r'(?:AGENTS\.md|CLAUDE\.md|\.cursorrules|\.clinerules)' _HERMES_CONFIG_FILES = r'\.hermes/(?:config\.yaml|SOUL\.md)' @@ -78,15 +77,14 @@ _OTHER_AGENT_CONFIG_FILES = r'\.(?:claude/settings|codex/config)[\w.]*' def _shell_write_re(file_alt: str) -> str: """Mechanical shell write into *file_alt*: ``>``/``>>``, ``sed -i``, ``tee`` (target as immediate argument, so - ``| tee output | AGENTS.md |`` cells miss), ``cp``/``mv`` with the file as destination (a source arg is required, - so ``cp AGENTS.md backup/`` misses; ``AGENTS.md.bak`` is not the file). A single ``>`` needs a preceding - word/quote/paren char so blockquotes (``> text``) and arrows (``-> file``) miss.""" + ``| tee output | AGENTS.md |`` cells miss), ``cp``/``mv`` with the file as destination (source arg required, so + ``cp AGENTS.md backup/`` 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)' rf'|\btee\s+(?:-a\s+)?[~\w./"\'-]*{file_alt}(?!\.?\w)' - rf'|\b(?:cp|mv)\s+[^\s|;&]+\s+[^\n|;&]{{0,40}}?{file_alt}(?!\.?\w)' - ) + rf'|\b(?:cp|mv)\s+[^\s|;&]+\s+[^\n|;&]{{0,40}}?{file_alt}(?!\.?\w)') def _prose_modify_re(file_alt: str) -> str: @@ -97,8 +95,7 @@ def _prose_modify_re(file_alt: str) -> str: 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+' rf'|\bmake\s+sure\s+(?:to\s+|you\s+)|\bbe\s+sure\s+to\s+)' - rf'{MODIFY_VERB_RE}[^\n,]{{0,80}}?{file_alt}\b' - ) + rf'{MODIFY_VERB_RE}[^\n,]{{0,80}}?{file_alt}\b') def _content_contract_re(file_alt: str) -> str: @@ -106,8 +103,7 @@ def _content_contract_re(file_alt: str) -> str: 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' - ) + rf'(?:contain|say|include|have|list)\b') THREAT_PATTERNS = [ # ── Exfiltration: shell commands leaking secrets ── @@ -352,10 +348,7 @@ THREAT_PATTERNS = [ "send_to_url", "high", "exfiltration", "instructs agent to send data to a URL"), ] -_COMPILED_THREAT_PATTERNS = [ - (re.compile(pattern, re.IGNORECASE), pid, severity, category, description) - for pattern, pid, severity, category, description in THREAT_PATTERNS -] +_COMPILED_THREAT_PATTERNS = [(re.compile(pattern, re.IGNORECASE), *rest) for pattern, *rest in THREAT_PATTERNS] # Structural limits for skill directories MAX_FILE_COUNT = 50 # skills shouldn't have 50+ files @@ -393,13 +386,12 @@ def _compute_docstring_lines(lines: list) -> set: one-line docstrings), so ``os.environ`` in prose is not scored. Heuristic: a triple quote inside a string literal is miscounted, but the common false-positive shapes are covered.""" doc_lines: set = set() - in_docstring = False - for i, line in enumerate(lines): - was_in = in_docstring - counts = [line.count(marker) for marker in ('"""', "'''")] - 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) + inside = False + for i, line in enumerate(lines, start=1): + was_in, counts = inside, [line.count(marker) for marker in ('"""', "'''")] + inside ^= sum(counts) % 2 == 1 # each odd marker count toggles; two odd counts cancel + if was_in or inside or any(counts): + doc_lines.add(i) return doc_lines @@ -505,8 +497,7 @@ def scan_skill_cached( def should_allow_install(result: ScanResult, force: bool = False) -> Tuple[bool, str]: """``(allowed, reason)`` from verdict + trust; *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)] + decision = INSTALL_POLICY.get(result.trust_level, INSTALL_POLICY["community"])[VERDICT_INDEX.get(result.verdict, 2)] n = len(result.findings) hard_block = result.verdict == "dangerous" and result.trust_level in ("community", "trusted") if decision == "allow": @@ -515,11 +506,8 @@ def should_allow_install(result: ScanResult, force: bool = False) -> Tuple[bool, 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, {n} findings)" - if hard_block: - 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.") + blocked = f"Blocked ({result.trust_level} source + {result.verdict} verdict, {n} findings). " + return False, blocked + ("--force does not override a dangerous verdict." if hard_block else "Use --force to override.") def format_scan_report(result: ScanResult) -> str: @@ -533,8 +521,7 @@ def format_scan_report(result: ScanResult) -> str: lines.append("") allowed, reason = should_allow_install(result) status = "ALLOWED" if allowed is True else "NEEDS CONFIRMATION" if allowed is None else "BLOCKED" - lines.append(f"Decision: {status} — {reason}") - return "\n".join(lines) + return "\n".join(lines + [f"Decision: {status} — {reason}"]) def _check_structure(skill_dir: Path, ignore=None) -> List[Finding]: @@ -560,9 +547,10 @@ def _check_structure(skill_dir: Path, ignore=None) -> List[Finding]: add("broken_symlink", "medium", "traversal", rel, "broken symlink", "broken or circular symlink") continue try: - size = f.stat().st_size + st = f.stat() except OSError: continue + size = st.st_size total_size += size if size > MAX_SINGLE_FILE_KB * 1024: add("oversized_file", "medium", "structural", rel, f"{size // 1024}KB", @@ -571,7 +559,7 @@ def _check_structure(skill_dir: Path, ignore=None) -> List[Finding]: if ext in SUSPICIOUS_BINARY_EXTENSIONS: 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: + if ext not in _SCRIPT_EXTENSIONS and st.st_mode & 0o111: 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: @@ -585,8 +573,6 @@ def _check_structure(skill_dir: Path, ignore=None) -> List[Finding]: # `.skillignore` is Hermes-native; `.clawhubignore` is honored for skills published through ClawHub. _SKILL_IGNORE_FILENAMES = (".skillignore", ".clawhubignore") -_ALWAYS_IGNORED_NAMES = set(_SKILL_IGNORE_FILENAMES) -_NEVER_IGNORABLE = {"SKILL.md"} def _load_skill_ignore(skill_dir: Path): @@ -605,9 +591,9 @@ def _load_skill_ignore(skill_dir: Path): rel_posix = Path(rel).as_posix() segs = rel_posix.split("/") base = segs[-1] - if base in _NEVER_IGNORABLE: + if base == "SKILL.md": return False - if base in _ALWAYS_IGNORED_NAMES: + if base in _SKILL_IGNORE_FILENAMES: return True for pat in patterns: anchored = pat.startswith("/")