diff --git a/src/skillspector/input_handler.py b/src/skillspector/input_handler.py index c42eaf1ce..dd60b1bd3 100644 --- a/src/skillspector/input_handler.py +++ b/src/skillspector/input_handler.py @@ -709,6 +709,57 @@ def _validate_zip_member_type(info: zipfile.ZipInfo) -> None: raise ValueError("Zip directory entry contains file data") +def selected_source_identity_for_input(input_path: str) -> str | None: + """Return a host/operator-selected skill identity from the original input. + + Temporary git/zip materialization uses ephemeral scan-root basenames such as + ``repo`` or ``extracted``. The repository, archive, or selected local path + name remains a trusted corroborating identity for AS3 self-reference + suppression without trusting contributor-controlled manifest data alone. + """ + text = input_path.strip() + if not text: + return None + + if text.startswith("git@"): + match = re.match(r"^git@[^:]+:(.+)$", text) + if match is None: + return None + repo_path = match.group(1).removesuffix(".git").rstrip("/") + name = repo_path.rsplit("/", 1)[-1] + return name.strip() or None + + if text.startswith(("https://", "http://")): + parsed = urlparse(text) + path = (parsed.path or "").removesuffix(".git").rstrip("/") + parts = [part for part in path.split("/") if part] + if not parts: + return None + # github.com/owner/repo[/...], raw.githubusercontent.com/owner/repo/... + name = parts[1] if len(parts) >= 2 else parts[0] + return name.strip() or None + + local = Path(text) + if local.suffix.lower() == ".zip": + stem = local.stem.strip() + return stem or None + if local.suffix.lower() == ".md": + parent_name = local.parent.name.strip() + if parent_name and parent_name not in {".", ".."}: + return parent_name + stem = local.stem.strip() + if stem and stem.casefold() != "skill": + return stem + return None + + name = local.name.strip() + if not name or name in {".", ".."}: + return None + if name in {"repo", "extracted"} or name.startswith("skillspector_"): + return None + return name + + class InputHandler: """ Handles input resolution for different source types. diff --git a/src/skillspector/nodes/analyzers/static_patterns_agent_snooping.py b/src/skillspector/nodes/analyzers/static_patterns_agent_snooping.py index 6e980c03d..362946854 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_agent_snooping.py +++ b/src/skillspector/nodes/analyzers/static_patterns_agent_snooping.py @@ -33,6 +33,7 @@ from collections.abc import Mapping from contextvars import ContextVar +from skillspector.input_handler import selected_source_identity_for_input from skillspector.logging_config import get_logger from skillspector.models import AnalyzerFinding, Location, Severity from skillspector.state import AnalyzerNodeResponse, SkillspectorState @@ -50,6 +51,8 @@ _CURRENT_SKILL_IDENTIFIERS: ContextVar[frozenset[str]] = ContextVar( "agent_snooping_current_skill_identifiers", default=frozenset() ) +# Ephemeral basenames created by InputHandler for git/zip/file materialization. +_EPHEMERAL_SCAN_ROOT_BASENAMES = frozenset({"repo", "extracted"}) # AS1: Agent Config Directory Access # Matches code/instructions that read from well-known agent config directories. @@ -208,8 +211,32 @@ def _normalize_skill_identifier(value: object) -> str | None: return normalized or None +def _is_ephemeral_scan_root_basename(identifier: str) -> bool: + """Return whether a scan-root basename is an InputHandler materialization stub.""" + return identifier in _EPHEMERAL_SCAN_ROOT_BASENAMES or identifier.startswith("skillspector_") + + +def _selected_source_identifier(state: SkillspectorState) -> str | None: + """Return the trusted repository/archive/selected-source identity from state.""" + selected = _normalize_skill_identifier(state.get("selected_source_identity")) + if selected is not None: + return selected + input_path = state.get("input_path") + if isinstance(input_path, str) and input_path.strip(): + return _normalize_skill_identifier(selected_source_identity_for_input(input_path.strip())) + return None + + def _current_skill_identifiers(state: SkillspectorState) -> frozenset[str]: - """Derive a trusted, internally consistent current-skill identity.""" + """Derive trusted current-skill identities for AS3 self-reference suppression. + + Host-derived scan-root basenames and selected repository/archive identities + are authoritative. Contributor-controlled ``manifest.name`` may only + corroborate those trusted identities; it never introduces a suppression + identity on its own. Ephemeral temp-clone basenames such as ``repo`` are + ignored so a matching selected-source identity can still suppress the real + skill self-path without opening a peer-skill false negative. + """ skill_path: object = state.get("skill_path") path_text: str | bytes | None = None @@ -225,19 +252,26 @@ def _current_skill_identifiers(state: SkillspectorState) -> frozenset[str]: normalized_path = path_text.replace("\\", "/").rstrip("/") path_identifier = _normalize_skill_identifier(normalized_path.rsplit("/", 1)[-1]) + source_identifier = _selected_source_identifier(state) + manifest = state.get("manifest") manifest_identifier: str | None = None if isinstance(manifest, Mapping): manifest_identifier = _normalize_skill_identifier(manifest.get("name")) - # The path is the only host-derived identity available here. A manifest - # name is contributor-controlled, so it may corroborate the path but must - # never introduce a second identity or override a disagreement. - if path_identifier is None: - return frozenset() - if manifest_identifier is not None and manifest_identifier != path_identifier: - return frozenset() - return frozenset({path_identifier}) + identifiers: set[str] = set() + if path_identifier is not None and not _is_ephemeral_scan_root_basename(path_identifier): + identifiers.add(path_identifier) + if source_identifier is not None: + identifiers.add(source_identifier) + + # Manifest data is contributor-controlled. Keep it only when it already + # matches a trusted host/operator identity (no-op add) so mismatched names + # such as ``name: victim`` cannot suppress peer ``skills/victim/SKILL.md``. + if manifest_identifier is not None and manifest_identifier in identifiers: + identifiers.add(manifest_identifier) + + return frozenset(identifiers) def _is_current_skill_path_reference( diff --git a/src/skillspector/nodes/resolve_input.py b/src/skillspector/nodes/resolve_input.py index 5ff22ac43..cc8a68fd5 100644 --- a/src/skillspector/nodes/resolve_input.py +++ b/src/skillspector/nodes/resolve_input.py @@ -27,6 +27,7 @@ from skillspector.input_handler import ( InputHandler, TransitiveIngestTruncatedError, + selected_source_identity_for_input, validate_local_input_path, ) from skillspector.logging_config import get_logger @@ -60,6 +61,7 @@ def resolve_input(state: SkillspectorState) -> dict[str, object]: resolved, source_type = handler.resolve(input_path.strip()) update: dict[str, object] = { "skill_path": str(resolved), + "selected_source_identity": selected_source_identity_for_input(input_path.strip()), "workflow_resource_budget": workflow_budget, } temp_dir = handler.temp_dir_for_cleanup() @@ -87,6 +89,7 @@ def resolve_input(state: SkillspectorState) -> dict[str, object]: resolved = validate_local_input_path(Path(skill_path)) return { "skill_path": str(resolved), + "selected_source_identity": selected_source_identity_for_input(skill_path.strip()), "temp_dir_for_cleanup": None, "workflow_resource_budget": workflow_budget, } @@ -94,12 +97,14 @@ def resolve_input(state: SkillspectorState) -> dict[str, object]: logger.warning("Could not resolve skill_path: %s", e) return { "skill_path": None, + "selected_source_identity": None, "temp_dir_for_cleanup": None, "workflow_resource_budget": workflow_budget, } return { "skill_path": None, + "selected_source_identity": None, "temp_dir_for_cleanup": None, "workflow_resource_budget": workflow_budget, } diff --git a/src/skillspector/state.py b/src/skillspector/state.py index 1a887e76f..2e1da2bf0 100644 --- a/src/skillspector/state.py +++ b/src/skillspector/state.py @@ -236,6 +236,10 @@ class SkillspectorState(TypedDict, total=False): # Input: resolve_input node consumes input_path or skill_path, sets skill_path input_path: str | None skill_path: str | None + # Host/operator-selected repository, archive, or path identity used to + # corroborate AS3 current-skill suppression when temp clones use ephemeral + # scan-root basenames such as ``repo`` / ``extracted``. + selected_source_identity: str | None # Set by resolve_input when a temp dir was created (git/url/zip/file); caller should clean up temp_dir_for_cleanup: str | None zip_bytes: bytes | None diff --git a/tests/nodes/analyzers/test_static_patterns.py b/tests/nodes/analyzers/test_static_patterns.py index 89c3a1ec1..d134e4337 100644 --- a/tests/nodes/analyzers/test_static_patterns.py +++ b/tests/nodes/analyzers/test_static_patterns.py @@ -895,10 +895,11 @@ def test_as3_scan_root_identity_marks_self_reference_when_manifest_is_absent(sel assert readme_event["emitted_finding_ids"] == [] assert not any(f.rule_id == "AS3" for f in result["findings"]) - def test_as3_inconsistent_manifest_identity_fails_closed(self): - """A contributor-controlled manifest cannot override the scan-root identity.""" + def test_as3_selected_source_identity_suppresses_temp_clone_self_reference(self): + """Temp-clone basename ``repo`` still suppresses when source identity corroborates.""" state = { - "skill_path": "/tmp/checkout-root", + "skill_path": "/tmp/skillspector_abc123/repo", + "selected_source_identity": "example-skill", "manifest": {"name": "example-skill"}, "components": ["README.md"], "file_cache": {"README.md": "Root skill: skills/example-skill/SKILL.md"}, @@ -906,11 +907,53 @@ def test_as3_inconsistent_manifest_identity_fails_closed(self): result = agent_snooping_module.node(state) + readme_event = next( + event for event in result["inspection_ledger"] if event["path"] == "README.md" + ) + assert readme_event["outcome"] == "completed" + assert readme_event["emitted_finding_ids"] == [] + assert not any(f.rule_id == "AS3" for f in result["findings"]) + + def test_as3_path_basename_suppresses_without_trusting_mismatched_manifest(self): + """Scan-root basename suppresses self-paths; uncorroborated manifest names do not.""" + state = { + "skill_path": "/tmp/checkout-root/example-skill", + "manifest": {"name": "published-name"}, + "components": ["README.md"], + "file_cache": { + "README.md": ( + "Root skill: skills/example-skill/SKILL.md\n" + "Also: skills/published-name/SKILL.md\n" + "Peer: skills/other-skill/SKILL.md" + ) + }, + } + + result = agent_snooping_module.node(state) + as3_findings = [finding for finding in result["findings"] if finding.rule_id == "AS3"] assert [finding.matched_text for finding in as3_findings] == [ - "skills/example-skill/SKILL.md" + "skills/published-name/SKILL.md", + "skills/other-skill/SKILL.md", ] + def test_as3_adversarial_mismatched_manifest_does_not_suppress_peer_path(self): + """Malicious manifest name unequal to trusted source identity cannot hide AS3.""" + state = { + "skill_path": "/tmp/skillspector_abc123/repo", + "selected_source_identity": "evil-skill", + "manifest": {"name": "victim"}, + "components": ["README.md"], + "file_cache": { + "README.md": ("Self: skills/evil-skill/SKILL.md\nPeer: skills/victim/SKILL.md") + }, + } + + result = agent_snooping_module.node(state) + + as3_findings = [finding for finding in result["findings"] if finding.rule_id == "AS3"] + assert [finding.matched_text for finding in as3_findings] == ["skills/victim/SKILL.md"] + def test_as3_long_current_skill_path_is_not_snooping(self): """Self-reference comparison uses the full path before evidence truncation.""" skill_name = f"example-{'a' * 190}" diff --git a/tests/nodes/test_resolve_input.py b/tests/nodes/test_resolve_input.py index 7adc6b360..41eca35ad 100644 --- a/tests/nodes/test_resolve_input.py +++ b/tests/nodes/test_resolve_input.py @@ -30,6 +30,7 @@ def test_resolve_input_with_input_path_directory(tmp_path: Path) -> None: update = resolve_input(state) assert update["skill_path"] == str(tmp_path.resolve()) assert update.get("temp_dir_for_cleanup") is None + assert update.get("selected_source_identity") == tmp_path.name def test_resolve_input_with_skill_path_only(tmp_path: Path) -> None: @@ -39,6 +40,7 @@ def test_resolve_input_with_skill_path_only(tmp_path: Path) -> None: update = resolve_input(state) assert update["skill_path"] == str(tmp_path.resolve()) assert update.get("temp_dir_for_cleanup") is None + assert update.get("selected_source_identity") == tmp_path.name def test_resolve_input_rejects_skill_path_with_symlinked_parent(tmp_path: Path) -> None: @@ -131,3 +133,19 @@ def cleanup(self) -> None: } assert cleaned == [True] assert "private/source" not in str(raised.value) + + +def test_selected_source_identity_from_git_and_archive_inputs() -> None: + """Repository and archive names become trusted selected-source identities.""" + from skillspector.input_handler import selected_source_identity_for_input + + assert ( + selected_source_identity_for_input("https://github.com/acme/example-skill.git") + == "example-skill" + ) + assert ( + selected_source_identity_for_input("git@github.com:acme/example-skill.git") + == "example-skill" + ) + assert selected_source_identity_for_input("/tmp/packs/example-skill.zip") == "example-skill" + assert selected_source_identity_for_input("/tmp/skillspector_abc/repo") is None