feat(skills): enforce capability/trust-tier checks at install and run time (#639)

Wires the previously-dead skills/security.py checks into two places. SkillImporter.import_skill() classifies trust tier before writing to disk, refuses unreviewed skills requesting dangerous capabilities unless confirmed (confirm_dangerous=True, or --yes-dangerous on skill install/sync), and persists tier/capabilities to the .source sidecar. SkillExecutor.run() gains opt-in capability enforcement: allowed_capabilities=None (the default, used by all existing call sites) means no policy; passing a set blocks skills whose required_capabilities are not covered before any step runs. Bulk sync reports refused skills instead of silently skipping them. Both enforcement points are covered by tests.

src/openjarvis/cli/skill_cmd.py

@@ -199,7 +199,16 @@ def _get_resolver(source: str, url: str = ""):
     default="",
     help="Repo URL (required when source is 'github').",
 )
-def install(query: str, with_scripts: bool, force: bool, url: str):
+@click.option(
+    "--yes-dangerous",
+    is_flag=True,
+    default=False,
+    help=(
+        "Confirm installing an unreviewed skill that requests dangerous "
+        "capabilities (shell/network-listen/filesystem-write)."
+    ),
+)
+def install(query: str, with_scripts: bool, force: bool, url: str, yes_dangerous: bool):
     """Install a skill from a source.
 
     Example: ``jarvis skill install hermes:apple-notes``
@@ -233,7 +242,12 @@ def install(query: str, with_scripts: bool, force: bool, url: str):
     from openjarvis.skills.tool_translator import ToolTranslator
 
     importer = SkillImporter(parser=SkillParser(), tool_translator=ToolTranslator())
-    result = importer.import_skill(matches[0], with_scripts=with_scripts, force=force)
+    result = importer.import_skill(
+        matches[0],
+        with_scripts=with_scripts,
+        force=force,
+        confirm_dangerous=yes_dangerous,
+    )
 
     if result.success:
         if result.skipped:
@@ -270,6 +284,15 @@ def install(query: str, with_scripts: bool, force: bool, url: str):
     help="Import scripts/ directories.",
 )
 @click.option("--force", is_flag=True, default=False, help="Re-import existing skills.")
+@click.option(
+    "--yes-dangerous",
+    is_flag=True,
+    default=False,
+    help=(
+        "Confirm installing unreviewed skills that request dangerous "
+        "capabilities (shell/network-listen/filesystem-write)."
+    ),
+)
 def sync(
     source: str,
     category: str,
@@ -277,6 +300,7 @@ def sync(
     search: str,
     with_scripts: bool,
     force: bool,
+    yes_dangerous: bool,
 ):
     """Bulk install + update from a source (or all configured sources)."""
     console = Console()
@@ -343,9 +367,20 @@ def sync(
 
         installed_count = 0
         for resolved in skills_to_import:
-            r = importer.import_skill(resolved, with_scripts=with_scripts, force=force)
+            r = importer.import_skill(
+                resolved,
+                with_scripts=with_scripts,
+                force=force,
+                confirm_dangerous=yes_dangerous,
+            )
             if r.success and not r.skipped:
                 installed_count += 1
+            elif not r.success and r.requires_confirmation:
+                console.print(
+                    f"  [yellow]Skipped {resolved.name}: requests dangerous "
+                    f"capabilities {r.dangerous_capabilities} "
+                    "(re-run with --yes-dangerous to install)[/yellow]"
+                )
         console.print(f"  Imported {installed_count}/{len(skills_to_import)} skills")
         total_installed += installed_count
 

src/openjarvis/skills/executor.py

@@ -5,10 +5,11 @@ from __future__ import annotations
 import json
 import re
 from dataclasses import dataclass, field
-from typing import Any, Callable, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional, Set
 
 from openjarvis.core.events import EventBus, EventType
 from openjarvis.core.types import ToolCall, ToolResult
+from openjarvis.skills.security import validate_capabilities
 from openjarvis.skills.types import SkillManifest
 from openjarvis.tools._stubs import ToolExecutor
 
@@ -37,10 +38,16 @@ class SkillExecutor:
         tool_executor: ToolExecutor,
         *,
         bus: Optional[EventBus] = None,
+        allowed_capabilities: Optional[Set[str]] = None,
     ) -> None:
         self._tool_executor = tool_executor
         self._bus = bus
         self._skill_resolver: Optional[SkillResolver] = None
+        # None means "no capability policy" — every skill runs, matching the
+        # behavior before capability enforcement existed. Pass a set (even an
+        # empty one) to enforce: skills whose required_capabilities are not a
+        # subset of it are blocked before any step runs.
+        self._allowed_capabilities: Optional[Set[str]] = allowed_capabilities
 
     def set_skill_resolver(self, resolver: SkillResolver) -> None:
         """Register a callback used to delegate ``skill_name`` steps."""
@@ -53,6 +60,38 @@ class SkillExecutor:
         initial_context: Optional[Dict[str, Any]] = None,
     ) -> SkillResult:
         """Execute all steps in a skill manifest."""
+        missing = (
+            validate_capabilities(manifest, self._allowed_capabilities)
+            if self._allowed_capabilities is not None
+            else []
+        )
+        if missing:
+            if self._bus:
+                self._bus.publish(
+                    EventType.SKILL_EXECUTE_START,
+                    {"skill": manifest.name, "steps": len(manifest.steps)},
+                )
+                self._bus.publish(
+                    EventType.SKILL_EXECUTE_END,
+                    {"skill": manifest.name, "success": False},
+                )
+            return SkillResult(
+                skill_name=manifest.name,
+                success=False,
+                step_results=[
+                    ToolResult(
+                        tool_name=manifest.name,
+                        content=(
+                            f"Blocked: skill '{manifest.name}' requires "
+                            f"capabilities {missing} that were not granted "
+                            "for this session."
+                        ),
+                        success=False,
+                    )
+                ],
+                context=dict(initial_context or {}),
+            )
+
         ctx: Dict[str, Any] = dict(initial_context or {})
         all_results: List[ToolResult] = []
 

src/openjarvis/skills/importer.py

@@ -25,6 +25,11 @@ import yaml
 
 from openjarvis.core.paths import get_config_dir
 from openjarvis.skills.parser import SkillParser
+from openjarvis.skills.security import (
+    TrustTier,
+    classify_trust_tier,
+    has_dangerous_capabilities,
+)
 from openjarvis.skills.sources.base import ResolvedSkill
 from openjarvis.skills.tool_translator import ToolTranslator
 
@@ -43,6 +48,9 @@ class ImportResult:
     untranslated_tools: List[str] = field(default_factory=list)
     scripts_imported: bool = False
     warnings: List[str] = field(default_factory=list)
+    trust_tier: TrustTier = TrustTier.UNREVIEWED
+    dangerous_capabilities: List[str] = field(default_factory=list)
+    requires_confirmation: bool = False
 
 
 class SkillImporter:
@@ -66,6 +74,7 @@ class SkillImporter:
         *,
         with_scripts: bool = False,
         force: bool = False,
+        confirm_dangerous: bool = False,
     ) -> ImportResult:
         """Install *resolved* into ``<target_root>/<source>/<name>/``.
 
@@ -95,12 +104,43 @@ class SkillImporter:
 
         try:
             frontmatter, body = self._read_skill_md(source_md)
-            self._parser.parse_frontmatter(frontmatter, markdown_content=body)
+            manifest = self._parser.parse_frontmatter(frontmatter, markdown_content=body)
         except Exception as exc:
             result.success = False
             result.warnings.append(f"Parse error: {exc}")
             return result
 
+        # 1a. Classify trust and check for dangerous capabilities *before*
+        # writing anything to disk. Everything the importer handles comes from
+        # an external source (github/hermes/openclaw), so the BUNDLED and
+        # WORKSPACE tiers never apply here, and no resolver verifies index
+        # membership yet — a signature alone still classifies as UNREVIEWED.
+        # Community skills get no special treatment just because they came
+        # from a named source.
+        result.trust_tier = classify_trust_tier(
+            has_signature=bool(manifest.signature),
+        )
+        result.dangerous_capabilities = has_dangerous_capabilities(manifest)
+
+        if result.dangerous_capabilities and result.trust_tier == TrustTier.UNREVIEWED:
+            result.requires_confirmation = True
+            if not confirm_dangerous:
+                result.success = False
+                result.warnings.append(
+                    "Refusing to install: this unreviewed skill requests "
+                    f"dangerous capabilities {result.dangerous_capabilities}. "
+                    "Re-run with confirm_dangerous=True (or `--yes-dangerous` "
+                    "on the CLI) only if you trust the source and have "
+                    "reviewed what it does."
+                )
+                return result
+            result.warnings.append(
+                "Installed with dangerous capabilities "
+                f"{result.dangerous_capabilities} — confirmed by caller. "
+                "This skill can run shell commands, open network listeners, "
+                "and/or write to the filesystem."
+            )
+
         # 2. Translate tool references
         translated_body, untranslated = self._translator.translate_markdown(body)
         result.untranslated_tools = untranslated
@@ -180,6 +220,7 @@ class SkillImporter:
         translated_str = ", ".join(f'"{t}"' for t in result.translated_tools)
         missing_str = ", ".join(f'"{t}"' for t in result.untranslated_tools)
         scripts_lower = "true" if result.scripts_imported else "false"
+        dangerous_str = ", ".join(f'"{c}"' for c in result.dangerous_capabilities)
 
         content = (
             f'source = "{resolved.source}:{resolved.name}"\n'
@@ -189,6 +230,8 @@ class SkillImporter:
             f"translated_tools = [{translated_str}]\n"
             f"missing_tools = [{missing_str}]\n"
             f"scripts_imported = {scripts_lower}\n"
+            f'trust_tier = "{result.trust_tier.value}"\n'
+            f"dangerous_capabilities = [{dangerous_str}]\n"
         )
         (target_dir / ".source").write_text(content, encoding="utf-8")
 

tests/skills/test_importer.py

@@ -202,3 +202,75 @@ class TestImportSkill:
 
         installed = target_root / "hermes" / "my-skill" / "SKILL.md"
         assert "Original" in installed.read_text()
+
+
+class TestDangerousCapabilityGate:
+    def _make_importer(self, tmp_path: Path) -> SkillImporter:
+        return SkillImporter(
+            parser=SkillParser(),
+            tool_translator=ToolTranslator(),
+            target_root=tmp_path / "skills",
+        )
+
+    def _make_resolved_with_caps(
+        self, tmp_path: Path, caps: list[str]
+    ) -> ResolvedSkill:
+        src_dir = tmp_path / "source" / "my-skill"
+        src_dir.mkdir(parents=True)
+        caps_yaml = "".join(f"  - {c}\n" for c in caps)
+        (src_dir / "SKILL.md").write_text(
+            "---\n"
+            "name: my-skill\n"
+            "description: A test skill\n"
+            f"required_capabilities:\n{caps_yaml}"
+            "---\n"
+            "Body"
+        )
+        return ResolvedSkill(
+            name="my-skill",
+            source="hermes",
+            path=src_dir,
+            category="testing",
+            description="A test skill",
+            commit="abc123",
+        )
+
+    def test_refuses_unreviewed_dangerous_skill(self, tmp_path: Path):
+        importer = self._make_importer(tmp_path)
+        resolved = self._make_resolved_with_caps(tmp_path, ["shell:execute"])
+        result = importer.import_skill(resolved)
+
+        assert not result.success
+        assert result.requires_confirmation
+        assert result.dangerous_capabilities == ["shell:execute"]
+        assert any("dangerous" in w.lower() for w in result.warnings)
+        # Nothing may be written to disk on refusal
+        assert not (tmp_path / "skills" / "hermes" / "my-skill").exists()
+
+    def test_confirm_dangerous_installs_and_records_tier(self, tmp_path: Path):
+        importer = self._make_importer(tmp_path)
+        resolved = self._make_resolved_with_caps(tmp_path, ["shell:execute"])
+        result = importer.import_skill(resolved, confirm_dangerous=True)
+
+        assert result.success
+        assert result.requires_confirmation
+        assert any("confirmed by caller" in w for w in result.warnings)
+        content = (
+            tmp_path / "skills" / "hermes" / "my-skill" / ".source"
+        ).read_text()
+        assert 'trust_tier = "unreviewed"' in content
+        assert 'dangerous_capabilities = ["shell:execute"]' in content
+
+    def test_benign_capabilities_need_no_confirmation(self, tmp_path: Path):
+        importer = self._make_importer(tmp_path)
+        resolved = self._make_resolved_with_caps(tmp_path, ["network:fetch"])
+        result = importer.import_skill(resolved)
+
+        assert result.success
+        assert not result.requires_confirmation
+        assert result.dangerous_capabilities == []
+        content = (
+            tmp_path / "skills" / "hermes" / "my-skill" / ".source"
+        ).read_text()
+        assert 'trust_tier = "unreviewed"' in content
+        assert "dangerous_capabilities = []" in content

tests/skills/test_skills.py

@@ -148,6 +148,58 @@ class TestSkillExecutor:
         assert EventType.SKILL_EXECUTE_END in event_types
 
 
+class TestSkillExecutorCapabilities:
+    def _manifest(self):
+        return SkillManifest(
+            name="capskill",
+            required_capabilities=["network:fetch"],
+            steps=[
+                SkillStep(
+                    tool_name="echo",
+                    arguments_template='{"text": "hello"}',
+                    output_key="result",
+                )
+            ],
+        )
+
+    def test_no_policy_runs_capability_skills(self):
+        """Default construction (no allowed_capabilities) must not enforce —
+        this is the pre-enforcement behavior every manager.py call site relies on."""
+        executor = SkillExecutor(ToolExecutor([EchoTool()]))
+        result = executor.run(self._manifest())
+        assert result.success
+        assert result.context.get("result") == "hello"
+
+    def test_policy_blocks_missing_capability(self):
+        executor = SkillExecutor(
+            ToolExecutor([EchoTool()]), allowed_capabilities=set()
+        )
+        result = executor.run(self._manifest())
+        assert not result.success
+        assert len(result.step_results) == 1
+        assert "Blocked" in result.step_results[0].content
+        assert "network:fetch" in result.step_results[0].content
+
+    def test_policy_allows_granted_capability(self):
+        executor = SkillExecutor(
+            ToolExecutor([EchoTool()]),
+            allowed_capabilities={"network:fetch"},
+        )
+        result = executor.run(self._manifest())
+        assert result.success
+        assert result.context.get("result") == "hello"
+
+    def test_blocked_run_publishes_events(self):
+        bus = EventBus(record_history=True)
+        executor = SkillExecutor(
+            ToolExecutor([EchoTool()]), bus=bus, allowed_capabilities=set()
+        )
+        executor.run(self._manifest())
+        event_types = {e.event_type for e in bus.history}
+        assert EventType.SKILL_EXECUTE_START in event_types
+        assert EventType.SKILL_EXECUTE_END in event_types
+
+
 class TestSkillStepExtended:
     def test_step_with_skill_name(self):
         step = SkillStep(skill_name="summarize", output_key="result")