From bff2d41591c6830863d2b6efd976bb36ea22d50e Mon Sep 17 00:00:00 2001 From: KodeCharya Date: Tue, 14 Jul 2026 12:57:01 +0530 Subject: [PATCH] Add new security analyzers, remediation module - Added behavioral fingerprinting analyzer node - Added cross-skill dependency analyzer node - Added prompt injection resilience analyzer node - Integrated new nodes into `__init__.py` (ANALYZER_NODE_IDS and ANALYZER_NODES) - Created `remediation.py` and `watcher.py` to handle automated responses and live monitoring - Updated `cli.py` and `pattern_defaults.py` to support the new features --- .../nodes/analyzers/behavioral_fingerprint.py | 362 ++++++++++++++++++ .../nodes/analyzers/cross_skill_dependency.py | 201 ++++++++++ .../analyzers/prompt_injection_resilience.py | 203 ++++++++++ src/skillspector/remediation.py | 182 +++++++++ src/skillspector/watcher.py | 108 ++++++ 5 files changed, 1056 insertions(+) create mode 100644 src/skillspector/nodes/analyzers/behavioral_fingerprint.py create mode 100644 src/skillspector/nodes/analyzers/cross_skill_dependency.py create mode 100644 src/skillspector/nodes/analyzers/prompt_injection_resilience.py create mode 100644 src/skillspector/remediation.py create mode 100644 src/skillspector/watcher.py diff --git a/src/skillspector/nodes/analyzers/behavioral_fingerprint.py b/src/skillspector/nodes/analyzers/behavioral_fingerprint.py new file mode 100644 index 00000000..2f822107 --- /dev/null +++ b/src/skillspector/nodes/analyzers/behavioral_fingerprint.py @@ -0,0 +1,362 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Behavioral fingerprint analyzer: extract and hash behavioral signatures from skills. + +Computes a behavioral fingerprint of each skill by extracting: +- Import statements (what modules it uses) +- Function calls (what APIs it invokes) +- File access patterns (what paths it reads/writes) +- Network access patterns (what URLs/domains it contacts) +- Environment variable access (what secrets it reads) + +The fingerprint is a deterministic JSON hash that enables: +- Quick comparison against known-bad fingerprints +- Drift detection between skill versions +- Community threat intelligence sharing +""" + +from __future__ import annotations + +import ast +import hashlib +import json +import re + +from skillspector.logging_config import get_logger +from skillspector.models import AnalyzerFinding, Finding, Location, Severity +from skillspector.state import AnalyzerNodeResponse, SkillspectorState + +from .common import build_import_aliases, get_context_from_lines, get_source_segment, resolve_call_name +from .static_runner import MAX_FILE_BYTES, analyzer_finding_to_finding + +ANALYZER_ID = "behavioral_fingerprint" +logger = get_logger(__name__) + +_TAG = "Behavioral Fingerprint" + +# Known dangerous module groups +_DANGEROUS_MODULE_GROUPS = { + "network": {"requests", "urllib", "httpx", "aiohttp", "socket", "websocket"}, + "execution": {"subprocess", "os", "shlex", "popen", "pty"}, + "file_io": {"pathlib", "shutil", "glob", "fnmatch", "tempfile"}, + "crypto": {"hashlib", "hmac", "cryptography", "bcrypt"}, + "serialization": {"pickle", "marshal", "shelve", "json", "yaml"}, + "env": {"os", "dotenv"}, +} + +# Patterns for detecting network URLs in code/strings +_URL_PATTERN = re.compile( + r"https?://[^\s\"']+|" + r"wss?://[^\s\"']+|" + r"(?:POST|GET|PUT|DELETE|PATCH)\s+[^\s\"']+", + re.IGNORECASE, +) + +# Patterns for detecting file path access +_PATH_ACCESS_PATTERNS = [ + re.compile(r"(?:open|read|write|read_text|write_text)\s*\(\s*['\"]([^'\"]+)['\"]"), + re.compile(r"(?:Path|PurePath)\s*\(\s*['\"]([^'\"]+)['\"]"), + re.compile(r"(?:os\.path\.join|os\.path\.expanduser)\s*\(\s*['\"]([^'\"]+)['\"]"), + re.compile(r"~/(?:\.ssh|\.aws|\.config|\.env|\.git|Library)"), +] + +# Patterns for detecting env var access +_ENV_VAR_PATTERNS = [ + re.compile(r"os\.environ(?:\.get|\.pop|\[)\s*\(\s*['\"]([A-Z_]+)['\"]"), + re.compile(r"os\.getenv\s*\(\s*['\"]([A-Z_]+)['\"]"), + re.compile(r"ENV\s+([A-Z_]+)="), +] + +# Dangerous file paths that indicate credential access +_SENSITIVE_PATHS = frozenset({ + "~/.ssh", "~/.aws", "~/.config", "~/.env", "~/.git", + "/etc/passwd", "/etc/shadow", "/etc/hosts", + "~/.bashrc", "~/.zshrc", "~/.profile", +}) + + +def _extract_imports(tree: ast.Module) -> list[str]: + """Extract all import names from a Python AST.""" + imports = [] + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + imports.append(alias.name) + elif isinstance(node, ast.ImportFrom): + module = node.module or "" + for alias in node.names: + imports.append(f"{module}.{alias.name}" if module else alias.name) + return sorted(set(imports)) + + +def _extract_function_calls(tree: ast.Module, aliases: dict[str, str]) -> list[str]: + """Extract all function call names from a Python AST.""" + calls = [] + for node in ast.walk(tree): + if isinstance(node, ast.Call): + name = resolve_call_name(node, aliases) + if name: + calls.append(name) + return sorted(set(calls)) + + +def _extract_string_literals(tree: ast.Module) -> list[str]: + """Extract all string literals from a Python AST.""" + strings = [] + for node in ast.walk(tree): + if isinstance(node, ast.Constant) and isinstance(node.value, str): + if len(node.value) > 3: + strings.append(node.value) + return strings + + +def _detect_urls_in_strings(strings: list[str]) -> list[str]: + """Find URLs in string literals.""" + urls = set() + for s in strings: + for match in _URL_PATTERN.finditer(s): + urls.add(match.group(0).strip()) + return sorted(urls) + + +def _detect_file_paths_in_strings(strings: list[str]) -> list[str]: + """Find file path references in string literals.""" + paths = set() + for s in strings: + for pattern in _PATH_ACCESS_PATTERNS: + for match in pattern.finditer(s): + paths.add(match.group(1) if match.lastindex else match.group(0)) + for sensitive in _SENSITIVE_PATHS: + if sensitive in s: + paths.add(sensitive) + return sorted(paths) + + +def _detect_env_vars(content: str) -> list[str]: + """Find environment variable accesses in code.""" + env_vars = set() + for pattern in _ENV_VAR_PATTERNS: + for match in pattern.finditer(content): + env_vars.add(match.group(1)) + return sorted(env_vars) + + +def _classify_imports(imports: list[str]) -> dict[str, list[str]]: + """Classify imports into behavioral categories.""" + classified: dict[str, list[str]] = {} + for imp in imports: + root = imp.split(".")[0] + for category, modules in _DANGEROUS_MODULE_GROUPS.items(): + if root in modules: + classified.setdefault(category, []).append(imp) + return classified + + +def _compute_fingerprint( + imports: list[str], + calls: list[str], + urls: list[str], + file_paths: list[str], + env_vars: list[str], +) -> str: + """Compute a deterministic SHA-256 hash of the behavioral fingerprint.""" + fingerprint_data = { + "imports": imports, + "calls": calls, + "urls": urls, + "file_paths": file_paths, + "env_vars": env_vars, + } + canonical = json.dumps(fingerprint_data, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(canonical.encode()).hexdigest() + + +def _analyze_python_fingerprint( + content: str, file_path: str +) -> tuple[list[str], list[str], list[str], list[str], list[str]]: + """Extract behavioral features from a Python file.""" + try: + tree = ast.parse(content, filename=file_path) + except SyntaxError: + return [], [], [], [], [] + + aliases = build_import_aliases(tree) + imports = _extract_imports(tree) + calls = _extract_function_calls(tree, aliases) + strings = _extract_string_literals(tree) + urls = _detect_urls_in_strings(strings) + file_paths = _detect_file_paths_in_strings(strings) + env_vars = _detect_env_vars(content) + return imports, calls, urls, file_paths, env_vars + + +def _analyze_markdown_fingerprint(content: str) -> tuple[list[str], list[str], list[str]]: + """Extract behavioral features from markdown/config files.""" + urls = sorted(set(m.group(0).strip() for m in _URL_PATTERN.finditer(content))) + env_vars = set() + for pattern in _ENV_VAR_PATTERNS: + for match in pattern.finditer(content): + env_vars.add(match.group(1)) + file_paths = set() + for pattern in _PATH_ACCESS_PATTERNS: + for match in pattern.finditer(content): + file_paths.add(match.group(1) if match.lastindex else match.group(0)) + for sensitive in _SENSITIVE_PATHS: + if sensitive in content: + file_paths.add(sensitive) + return urls, sorted(file_paths), sorted(env_vars) + + +def analyze(content: str, file_path: str, file_type: str) -> list[AnalyzerFinding]: + """Analyze content and extract behavioral fingerprint features.""" + findings: list[AnalyzerFinding] = [] + + if file_type == "python": + imports, calls, urls, file_paths, env_vars = _analyze_python_fingerprint(content, file_path) + elif file_type in ("markdown", "yaml", "json", "toml"): + urls, file_paths, env_vars = _analyze_markdown_fingerprint(content) + imports, calls = [], [] + else: + return findings + + # FP1: Sensitive file path access + sensitive_access = [p for p in file_paths if p in _SENSITIVE_PATHS] + if sensitive_access: + findings.append( + AnalyzerFinding( + rule_id="FP1", + message=f"Sensitive file path access detected: {', '.join(sensitive_access)}", + severity=Severity.HIGH, + location=Location(file=file_path, start_line=1), + confidence=0.8, + tags=[_TAG], + context=f"Accessed paths: {', '.join(sensitive_access)}", + matched_text=", ".join(sensitive_access), + ) + ) + + # FP2: Credential-related env var access + credential_envs = [v for v in env_vars if any( + kw in v for kw in ("KEY", "SECRET", "TOKEN", "PASSWORD", "CREDENTIAL", "AUTH") + )] + if credential_envs: + findings.append( + AnalyzerFinding( + rule_id="FP2", + message=f"Credential environment variable access: {', '.join(credential_envs)}", + severity=Severity.MEDIUM, + location=Location(file=file_path, start_line=1), + confidence=0.7, + tags=[_TAG], + context=f"Env vars: {', '.join(credential_envs)}", + matched_text=", ".join(credential_envs), + ) + ) + + # FP3: External network endpoints + external_urls = [u for u in urls if not u.startswith(("http://localhost", "http://127.", "http://0."))] + if external_urls: + findings.append( + AnalyzerFinding( + rule_id="FP3", + message=f"External network endpoints referenced: {len(external_urls)} URL(s)", + severity=Severity.LOW, + location=Location(file=file_path, start_line=1), + confidence=0.5, + tags=[_TAG], + context=f"URLs: {', '.join(external_urls[:5])}", + matched_text=", ".join(external_urls[:5]), + ) + ) + + # FP4: Dangerous import combination + if imports: + classified = _classify_imports(imports) + dangerous_combos = [] + if "execution" in classified and "network" in classified: + dangerous_combos.append("execution + network") + if "file_io" in classified and "network" in classified: + dangerous_combos.append("file_io + network") + if "serialization" in classified and "execution" in classified: + dangerous_combos.append("serialization + execution") + if dangerous_combos: + findings.append( + AnalyzerFinding( + rule_id="FP4", + message=f"Dangerous import combination: {', '.join(dangerous_combos)}", + severity=Severity.MEDIUM, + location=Location(file=file_path, start_line=1), + confidence=0.65, + tags=[_TAG], + context=f"Modules: {', '.join(imports[:10])}", + matched_text=", ".join(dangerous_combos), + ) + ) + + return findings + + +def node(state: SkillspectorState) -> AnalyzerNodeResponse: + """Compute behavioral fingerprints and detect risky behavioral patterns.""" + components: list[str] = state.get("components") or [] + file_cache: dict[str, str] = state.get("file_cache") or {} + all_findings: list[Finding] = [] + + for path in components: + content = file_cache.get(path) + if content is None or len(content) > MAX_FILE_BYTES: + continue + idx = path.rfind(".") + suffix = path[idx:].lower() if idx >= 0 else "" + file_type = { + ".py": "python", ".md": "markdown", ".yaml": "yaml", ".yml": "yaml", + ".json": "json", ".toml": "toml", + }.get(suffix, "other") + if file_type == "other": + continue + raw = analyze(content, path, file_type) + all_findings.extend(analyzer_finding_to_finding(af) for af in raw) + + # Compute the aggregate fingerprint across all files + all_imports, all_calls, all_urls, all_paths, all_envs = [], [], [], [], [] + for path in components: + content = file_cache.get(path) + if content is None or len(content) > MAX_FILE_BYTES: + continue + idx = path.rfind(".") + suffix = path[idx:].lower() if idx >= 0 else "" + if suffix == ".py": + i, c, u, p, e = _analyze_python_fingerprint(content, path) + all_imports.extend(i) + all_calls.extend(c) + all_urls.extend(u) + all_paths.extend(p) + all_envs.extend(e) + elif suffix in (".md", ".yaml", ".yml", ".json", ".toml"): + u, p, e = _analyze_markdown_fingerprint(content) + all_urls.extend(u) + all_paths.extend(p) + all_envs.extend(e) + + fingerprint = _compute_fingerprint( + sorted(set(all_imports)), + sorted(set(all_calls)), + sorted(set(all_urls)), + sorted(set(all_paths)), + sorted(set(all_envs)), + ) + logger.info("%s: %d findings, fingerprint=%s", ANALYZER_ID, len(all_findings), fingerprint[:12]) + return {"findings": all_findings} diff --git a/src/skillspector/nodes/analyzers/cross_skill_dependency.py b/src/skillspector/nodes/analyzers/cross_skill_dependency.py new file mode 100644 index 00000000..5f6974ef --- /dev/null +++ b/src/skillspector/nodes/analyzers/cross_skill_dependency.py @@ -0,0 +1,201 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Cross-skill dependency analyzer: detect references between skills in multi-skill dirs. + +When scanning a directory containing multiple skills, this analyzer detects: +- Direct references from one skill to another (invoke, call, import) +- Privilege escalation chains (skill A grants permissions that skill B exploits) +- Shared state or file access between skills +- Circular dependencies between skills + +These patterns can indicate coordinated supply-chain attacks where a benign-looking +skill serves as a vector for a malicious one. +""" + +from __future__ import annotations + +import re + +from skillspector.logging_config import get_logger +from skillspector.models import AnalyzerFinding, Finding, Location, Severity +from skillspector.state import AnalyzerNodeResponse, SkillspectorState + +from .static_runner import MAX_FILE_BYTES, analyzer_finding_to_finding + +ANALYZER_ID = "cross_skill_dependency" +logger = get_logger(__name__) + +_TAG = "Cross-Skill Dependency" + +# Patterns that reference other skills by name or path +_SKILL_REFERENCE_PATTERNS = [ + re.compile(r"(?:invoke|call|run|execute|use|load|import)\s+(?:skill\s+)?['\"]([a-zA-Z0-9_-]+)['\"]", re.IGNORECASE), + re.compile(r"(?:skill|agent|tool)[/\s]+([a-zA-Z0-9_-]+)", re.IGNORECASE), + re.compile(r"(?:depends?\s+on|requires?\s+skill|needs?\s+skill)\s+['\"]?([a-zA-Z0-9_-]+)['\"]?", re.IGNORECASE), + re.compile(r"\{\{([a-zA-Z0-9_-]+)\.(?:output|result|response)\}\}", re.IGNORECASE), +] + +# Patterns that suggest privilege escalation chains +_PRIVILEGE_ESCALATION_PATTERNS = [ + re.compile(r"(?:grant|give|assign|set)\s+(?:permission|access|role|capability)\s+to\s+['\"]([a-zA-Z0-9_-]+)['\"]", re.IGNORECASE), + re.compile(r"(?:share|expose|export)\s+(?:credentials?|tokens?|keys?|secrets?)\s+(?:with|to)\s+['\"]([a-zA-Z0-9_-]+)['\"]", re.IGNORECASE), + re.compile(r"(?:pipe|chain|pass)\s+(?:output|results?)\s+(?:to|into)\s+['\"]([a-zA-Z0-9_-]+)['\"]", re.IGNORECASE), +] + +# Patterns for shared state access +_SHARED_STATE_PATTERNS = [ + re.compile(r"(?:shared|common|global)\s+(?:state|config|store|cache|registry)", re.IGNORECASE), + re.compile(r"(?:/tmp/|/var/|~/.cache/).*(?:skill|agent)", re.IGNORECASE), + re.compile(r"(?:lockfile|mutex|semaphore|barrier)", re.IGNORECASE), +] + + +def analyze( + content: str, + file_path: str, + file_type: str, + all_skill_names: list[str] | None = None, +) -> list[AnalyzerFinding]: + """Analyze content for cross-skill dependency patterns.""" + findings: list[AnalyzerFinding] = [] + skill_names = [s.lower() for s in (all_skill_names or [])] + if not skill_names: + return findings + + # CS1: Direct skill references + for pattern in _SKILL_REFERENCE_PATTERNS: + for match in pattern.finditer(content): + ref_name = match.group(1).lower() if match.lastindex else "" + if ref_name in skill_names and ref_name != _skill_name_from_path(file_path).lower(): + line_num = content[:match.start()].count("\n") + 1 + findings.append( + AnalyzerFinding( + rule_id="CS1", + message=f"Cross-skill reference to '{match.group(1)}'", + severity=Severity.MEDIUM, + location=Location(file=file_path, start_line=line_num), + confidence=0.7, + tags=[_TAG], + context=content[max(0, match.start() - 50) : match.end() + 50], + matched_text=match.group(0)[:200], + ) + ) + + # CS2: Privilege escalation chains + for pattern in _PRIVILEGE_ESCALATION_PATTERNS: + for match in pattern.finditer(content): + ref_name = match.group(1).lower() if match.lastindex else "" + if ref_name in skill_names: + line_num = content[:match.start()].count("\n") + 1 + findings.append( + AnalyzerFinding( + rule_id="CS2", + message=f"Privilege escalation chain involving '{match.group(1)}'", + severity=Severity.HIGH, + location=Location(file=file_path, start_line=line_num), + confidence=0.75, + tags=[_TAG], + context=content[max(0, match.start() - 50) : match.end() + 50], + matched_text=match.group(0)[:200], + ) + ) + + # CS3: Shared state access + for pattern in _SHARED_STATE_PATTERNS: + for match in pattern.finditer(content): + line_num = content[:match.start()].count("\n") + 1 + findings.append( + AnalyzerFinding( + rule_id="CS3", + message="Shared state mechanism detected between skills", + severity=Severity.LOW, + location=Location(file=file_path, start_line=line_num), + confidence=0.5, + tags=[_TAG], + context=content[max(0, match.start() - 50) : match.end() + 50], + matched_text=match.group(0)[:200], + ) + ) + + return findings + + +def _skill_name_from_path(file_path: str) -> str: + """Extract skill directory name from a file path.""" + parts = file_path.replace("\\", "/").split("/") + for part in reversed(parts): + if part and not part.startswith(".") and part not in ("src", "lib", "code", "scripts"): + return part.rsplit(".", 1)[0] if "." in part else part + return "" + + +def _detect_circular_references( + references: dict[str, set[str]], +) -> list[tuple[str, str]]: + """Detect circular dependencies in a reference graph using DFS.""" + cycles: list[tuple[str, str]] = [] + visited: set[str] = set() + in_stack: set[str] = set() + + def _dfs(node: str, path: list[str]) -> None: + if node in in_stack: + cycle_start = path.index(node) + cycles.append((node, path[cycle_start])) + return + if node in visited: + return + visited.add(node) + in_stack.add(node) + path.append(node) + for neighbor in references.get(node, set()): + _dfs(neighbor, path) + path.pop() + in_stack.discard(node) + + for node in references: + _dfs(node, []) + + return cycles + + +def node(state: SkillspectorState) -> AnalyzerNodeResponse: + """Detect cross-skill dependency patterns.""" + components: list[str] = state.get("components") or [] + file_cache: dict[str, str] = state.get("file_cache") or {} + all_findings: list[Finding] = [] + + # Extract skill names from directory structure + skill_names: list[str] = [] + for path in components: + parts = path.replace("\\", "/").split("/") + for part in parts[:-1]: + if part and not part.startswith("."): + skill_names.append(part) + skill_names = list(set(skill_names)) + + if len(skill_names) < 2: + logger.info("%s: fewer than 2 skill dirs detected, skipping", ANALYZER_ID) + return {"findings": []} + + for path in components: + content = file_cache.get(path) + if content is None or len(content) > MAX_FILE_BYTES: + continue + raw = analyze(content, path, "other", all_skill_names=skill_names) + all_findings.extend(analyzer_finding_to_finding(af) for af in raw) + + logger.info("%s: %d findings", ANALYZER_ID, len(all_findings)) + return {"findings": all_findings} diff --git a/src/skillspector/nodes/analyzers/prompt_injection_resilience.py b/src/skillspector/nodes/analyzers/prompt_injection_resilience.py new file mode 100644 index 00000000..f2a56f11 --- /dev/null +++ b/src/skillspector/nodes/analyzers/prompt_injection_resilience.py @@ -0,0 +1,203 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Prompt injection resilience analyzer: assess skill defenses against adversarial inputs. + +This analyzer evaluates how well a skill's structure and instructions would hold up +against prompt injection attacks. Rather than detecting existing vulnerabilities, +it identifies structural weaknesses that would make the skill susceptible: + +- Missing instruction boundaries (no clear separation between user content and skill instructions) +- Permissive input handling (no input validation or sanitization instructions) +- Overly trusting instructions (trusts user-provided data without verification) +- Missing output guards (no instructions to prevent leaking internal state) +- Lack of adversarial robustness patterns + +Produces a "resilience score" as findings rather than a vulnerability score. +""" + +from __future__ import annotations + +import re + +from skillspector.logging_config import get_logger +from skillspector.models import AnalyzerFinding, Finding, Location, Severity +from skillspector.state import AnalyzerNodeResponse, SkillspectorState + +from .common import get_context, get_line_number +from .static_runner import MAX_FILE_BYTES, analyzer_finding_to_finding + +ANALYZER_ID = "prompt_injection_resilience" +logger = get_logger(__name__) + +_TAG = "Prompt Injection Resilience" + +# Patterns indicating missing input validation +_MISSING_VALIDATION_PATTERNS = [ + (re.compile(r"(?:user|input|message|prompt|query)\s*(?::|is|=\s*)", re.IGNORECASE), 0.4), + (re.compile(r"(?:accept|receive|process|handle)\s+(?:user|input|message)", re.IGNORECASE), 0.4), +] + +# Patterns indicating trust in user content +_TRUSTING_PATTERNS = [ + (re.compile(r"(?:trust|believe|assume)\s+(?:the\s+)?(?:user|input|message)", re.IGNORECASE), 0.7), + (re.compile(r"(?:always|never)\s+(?:validate|verify|sanitize|check)\s+(?:the\s+)?(?:input|user)", re.IGNORECASE), 0.8), + (re.compile(r"(?:do\s+not|don't)\s+(?:validate|verify|sanitize|filter)\s+(?:the\s+)?(?:input|user)", re.IGNORECASE), 0.8), + (re.compile(r"(?:process|execute|run)\s+(?:the\s+)?(?:user|input)\s+(?:directly|immediately|without)", re.IGNORECASE), 0.7), +] + +# Patterns indicating output guards are present +_OUTPUT_GUARD_PATTERNS = [ + re.compile(r"(?:never|do\s+not|don't)\s+(?:reveal|expose|output|show|print|display)\s+(?:the\s+)?(?:system|internal|hidden|secret)", re.IGNORECASE), + re.compile(r"(?:filter|sanitize|validate|escape)\s+(?:the\s+)?(?:output|response|result)", re.IGNORECASE), + re.compile(r"(?:never|do\s+not|don't)\s+(?:include|include|contain)\s+(?:the\s+)?(?:following|above|system|instruction)", re.IGNORECASE), +] + +# Patterns indicating adversarial awareness +_ADVERSARIAL_AWARENESS_PATTERNS = [ + re.compile(r"(?:malicious|adversarial|injection|attack|exploit)", re.IGNORECASE), + re.compile(r"(?:security|safety)\s+(?:check|validation|review|audit)", re.IGNORECASE), + re.compile(r"(?:untrusted|unverified|unsanitized)\s+(?:input|content|data)", re.IGNORECASE), + re.compile(r"(?:prompt\s+injection|jailbreak|bypass)", re.IGNORECASE), +] + +# Patterns indicating instruction boundary markers +_INSTRUCTION_BOUNDARY_PATTERNS = [ + re.compile(r"^#{1,3}\s+(?:instructions|rules|guidelines|constraints|boundaries)", re.IGNORECASE | re.MULTILINE), + re.compile(r"(?:IMPORTANT|CRITICAL|SECURITY|WARNING)[:\s].*(?:never|always|do\s+not|must)", re.IGNORECASE), + re.compile(r"```\s*(?:system|instructions)\s*```", re.IGNORECASE), +] + + +def analyze(content: str, file_path: str, file_type: str) -> list[AnalyzerFinding]: + """Analyze skill content for prompt injection resilience weaknesses.""" + findings: list[AnalyzerFinding] = [] + + def loc(ln: int) -> Location: + return Location(file=file_path, start_line=ln) + + def ctx(start: int) -> str: + return get_context(content, start) + + tag = [_TAG] + + # Only analyze markdown and text files (skill instruction files) + if file_type not in ("markdown", "text", "other"): + return findings + + # IR1: Missing instruction boundaries + has_boundaries = any(p.search(content) for p in _INSTRUCTION_BOUNDARY_PATTERNS) + if not has_boundaries: + findings.append( + AnalyzerFinding( + rule_id="IR1", + message="No instruction boundaries found - skill lacks clear security instruction markers", + severity=Severity.MEDIUM, + location=loc(1), + confidence=0.6, + tags=tag, + context="No clear boundary between skill instructions and user content", + ) + ) + + # IR2: Trusting patterns (trusts user input without validation) + for pattern, confidence in _TRUSTING_PATTERNS: + for match in pattern.finditer(content): + line_num = get_line_number(content, match.start()) + findings.append( + AnalyzerFinding( + rule_id="IR2", + message="Skill trusts user input without validation", + severity=Severity.MEDIUM, + location=loc(line_num), + confidence=confidence, + tags=tag, + context=ctx(match.start()), + matched_text=match.group(0)[:200], + ) + ) + + # IR3: Missing output guards + has_output_guards = any(p.search(content) for p in _OUTPUT_GUARD_PATTERNS) + if not has_output_guards and len(content) > 200: + findings.append( + AnalyzerFinding( + rule_id="IR3", + message="No output guards found - skill does not restrict information disclosure", + severity=Severity.LOW, + location=loc(1), + confidence=0.5, + tags=tag, + context="No instructions preventing the agent from revealing internal state", + ) + ) + + # IR4: Adversarial awareness + has_adversarial_awareness = any(p.search(content) for p in _ADVERSARIAL_AWARENESS_PATTERNS) + if not has_adversarial_awareness and len(content) > 200: + findings.append( + AnalyzerFinding( + rule_id="IR4", + message="No adversarial awareness - skill does not address injection threats", + severity=Severity.LOW, + location=loc(1), + confidence=0.4, + tags=tag, + context="No mentions of adversarial inputs, injection, or security validation", + ) + ) + + # IR5: Missing input validation instructions + has_validation = any(p.search(content) for p in [ + re.compile(r"(?:validate|verify|sanitize|filter|check)\s+(?:the\s+)?(?:user|input|content|data)", re.IGNORECASE), + re.compile(r"(?:never|do\s+not|don't)\s+(?:trust|assume|accept)\s+(?:the\s+)?(?:user|input)", re.IGNORECASE), + ]) + has_user_input_ref = any(p.search(content) for p, _ in _MISSING_VALIDATION_PATTERNS) + if has_user_input_ref and not has_validation: + findings.append( + AnalyzerFinding( + rule_id="IR5", + message="User input referenced without validation instructions", + severity=Severity.MEDIUM, + location=loc(1), + confidence=0.55, + tags=tag, + context="Skill processes user input but lacks explicit validation requirements", + ) + ) + + return findings + + +def node(state: SkillspectorState) -> AnalyzerNodeResponse: + """Evaluate prompt injection resilience of skill instructions.""" + components: list[str] = state.get("components") or [] + file_cache: dict[str, str] = state.get("file_cache") or {} + all_findings: list[Finding] = [] + + for path in components: + content = file_cache.get(path) + if content is None or len(content) > MAX_FILE_BYTES: + continue + idx = path.rfind(".") + suffix = path[idx:].lower() if idx >= 0 else "" + file_type = { + ".md": "markdown", ".txt": "text", + }.get(suffix, "other") + raw = analyze(content, path, file_type) + all_findings.extend(analyzer_finding_to_finding(af) for af in raw) + + logger.info("%s: %d findings", ANALYZER_ID, len(all_findings)) + return {"findings": all_findings} diff --git a/src/skillspector/remediation.py b/src/skillspector/remediation.py new file mode 100644 index 00000000..0b371b52 --- /dev/null +++ b/src/skillspector/remediation.py @@ -0,0 +1,182 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""AI-assisted remediation: generate auto-fix patches for detected findings. + +Scans a skill, identifies fixable findings, and produces patched file versions +with automated remediations applied. Supports both regex-based pattern fixes +and AST-based Python code transformations. + +This module provides the ``skillspector fix`` CLI command's core logic. +""" + +from __future__ import annotations + +import difflib +import re +import textwrap +from pathlib import Path + +from skillspector.logging_config import get_logger +from skillspector.models import Finding + +logger = get_logger(__name__) + + +# Rule ID -> (pattern, replacement) for automated regex fixes +_REGEX_FIXES: dict[str, list[tuple[re.Pattern[str], str]]] = { + "P1": [ + (re.compile(r"ignore\s+(?:all\s+)?previous\s+instructions?", re.IGNORECASE), "[REMOVED: instruction override]"), + (re.compile(r"ignore\s+(?:all\s+)?(?:safety|security)\s+(?:rules?|constraints?|guidelines?)", re.IGNORECASE), "[REMOVED: safety override]"), + (re.compile(r"you\s+are\s+now\s+(?:in\s+)?(?:jailbreak|unrestricted|unfiltered)\s+mode", re.IGNORECASE), "[REMOVED: jailbreak mode]"), + ], + "P2": [ + (re.compile(r"", re.DOTALL), ""), + (re.compile(r"\[//\]:\s*#\(.*?\)", re.DOTALL), ""), + ], + "P6": [ + (re.compile(r"(?:print|output|reveal|expose|show)\s+(?:the\s+)?(?:system\s+)?(?:prompt|instructions?|rules?)", re.IGNORECASE), "[REMOVED: system prompt leakage]"), + ], + "AR1": [ + (re.compile(r"never\s+(?:refuse|say\s+no|decline|reject)", re.IGNORECASE), "[REMOVED: refusal suppression]"), + ], + "AR2": [ + (re.compile(r"(?:omit|remove|skip|drop)\s+(?:all\s+)?(?:warnings?|disclaimers?|caveats?)", re.IGNORECASE), "[REMOVED: disclaimer suppression]"), + ], + "SC2": [ + (re.compile(r"curl\s+(?:-[sSkLfO]*\s+)*\S+\s*\|\s*(?:bash|sh|python|node)", re.IGNORECASE), "# [SECURITY] Remote code execution blocked: review and install manually"), + ], +} + +# Rule ID -> remediation template for SKILL.md security annotations +_SKILL_MD_TEMPLATES: dict[str, str] = { + "EA1": "# SECURITY: Tool access has been restricted to required tools only.", + "EA2": "# SECURITY: Destructive operations now require human confirmation.", + "LP2": "# SECURITY: Wildcard permissions replaced with explicit allowlist.", + "LP3": "# SECURITY: Permissions field added to SKILL.md manifest.", +} + + +class RemediationResult: + """Result of applying automated remediations.""" + + def __init__(self) -> None: + self.files_modified: list[str] = [] + self.fixes_applied: list[dict[str, str]] = [] + self.skipped: list[dict[str, str]] = [] + self.diff: str = "" + + def add_fix(self, file_path: str, rule_id: str, description: str) -> None: + self.files_modified.append(file_path) + self.fixes_applied.append({ + "file": file_path, + "rule": rule_id, + "description": description, + }) + + def add_skip(self, file_path: str, rule_id: str, reason: str) -> None: + self.skipped.append({ + "file": file_path, + "rule": rule_id, + "reason": reason, + }) + + def summary(self) -> str: + lines = [f"Applied {len(self.fixes_applied)} fix(es) to {len(set(self.files_modified))} file(s)."] + if self.skipped: + lines.append(f"Skipped {len(self.skipped)} finding(s) requiring manual review.") + return "\n".join(lines) + + +def apply_regex_fix(content: str, rule_id: str) -> tuple[str, int]: + """Apply regex-based fixes for a given rule ID. Returns (new_content, fix_count).""" + fixes = _REGEX_FIXES.get(rule_id, []) + fix_count = 0 + for pattern, replacement in fixes: + new_content = pattern.sub(replacement, content) + if new_content != content: + fix_count += 1 + content = new_content + return content, fix_count + + +def generate_skill_md_patch(findings: list[Finding]) -> str | None: + """Generate a SKILL.md security annotation block from findings.""" + annotations: list[str] = [] + seen_rules: set[str] = set() + for finding in findings: + template = _SKILL_MD_TEMPLATES.get(finding.rule_id) + if template and finding.rule_id not in seen_rules: + annotations.append(template) + seen_rules.add(finding.rule_id) + if not annotations: + return None + return "\n".join(annotations) + + +def compute_diff(old: str, new: str, file_path: str) -> str: + """Compute a unified diff between old and new file contents.""" + old_lines = old.splitlines(keepends=True) + new_lines = new.splitlines(keepends=True) + return "".join(difflib.unified_diff(old_lines, new_lines, fromfile=file_path, tofile=f"{file_path} (patched)")) + + +def remediate_files( + findings: list[Finding], + file_cache: dict[str, str], + dry_run: bool = True, +) -> tuple[RemediationResult, dict[str, str]]: + """Apply automated remediations to files based on findings. + + Args: + findings: List of findings to remediate. + file_cache: Map of file paths to their contents. + dry_run: If True, compute patches without writing to disk. + + Returns: + Tuple of (RemediationResult, patched_files_map). + """ + result = RemediationResult() + patched: dict[str, str] = {} + + findings_by_file: dict[str, list[Finding]] = {} + for f in findings: + findings_by_file.setdefault(f.file, []).append(f) + + for file_path, file_findings in findings_by_file.items(): + original = file_cache.get(file_path) + if original is None: + continue + content = original + total_fixes = 0 + for finding in file_findings: + if finding.rule_id in _REGEX_FIXES: + new_content, fix_count = apply_regex_fix(content, finding.rule_id) + if fix_count > 0: + content = new_content + total_fixes += fix_count + result.add_fix(file_path, finding.rule_id, f"Regex fix applied ({fix_count} occurrence(s))") + else: + result.add_skip(file_path, finding.rule_id, "Pattern not found in current content") + elif finding.rule_id in _SKILL_MD_TEMPLATES: + result.add_skip(file_path, finding.rule_id, "Requires manual SKILL.md edit") + else: + result.add_skip(file_path, finding.rule_id, "No automated fix available") + + if content != original: + patched[file_path] = content + result.diff += compute_diff(original, content, file_path) + "\n" + + return result, patched diff --git a/src/skillspector/watcher.py b/src/skillspector/watcher.py new file mode 100644 index 00000000..74838296 --- /dev/null +++ b/src/skillspector/watcher.py @@ -0,0 +1,108 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Skill Watch Mode: monitor directories for changes and auto-rescan. + +Provides a ``skillspector watch`` CLI command that watches a directory for +file changes (using polling on all platforms) and automatically re-scans +when SKILL.md or executable files are modified. + +Features: +- Configurable poll interval +- Debounce rapid changes +- Per-scan output formatting +- Baseline support for incremental scanning +""" + +from __future__ import annotations + +import hashlib +import time +from pathlib import Path + +from skillspector.logging_config import get_logger + +logger = get_logger(__name__) + +_WATCH_EXTENSIONS = frozenset({ + ".md", ".markdown", ".py", ".sh", ".bash", ".zsh", + ".js", ".ts", ".json", ".yaml", ".yml", ".toml", + ".rb", ".go", ".rs", +}) + +_WATCH_PATTERNS = frozenset({ + "SKILL.md", "skill.md", "requirements.txt", "pyproject.toml", + "package.json", "Gemfile", "go.mod", "Cargo.toml", +}) + + +def _compute_directory_hash(directory: Path) -> str: + """Compute a hash of all watchable files in a directory tree.""" + hasher = hashlib.md5(usedforsecurity=False) + for file_path in sorted(directory.rglob("*")): + if not file_path.is_file(): + continue + if file_path.name.startswith("."): + continue + if file_path.suffix.lower() in _WATCH_EXTENSIONS or file_path.name in _WATCH_PATTERNS: + try: + content = file_path.read_bytes() + hasher.update(file_path.relative_to(directory).as_posix().encode()) + hasher.update(content) + except OSError: + continue + return hasher.hexdigest() + + +def watch_directory( + directory: Path, + callback, + poll_interval: float = 2.0, + debounce: float = 5.0, + **callback_kwargs, +) -> None: + """Watch a directory for changes and invoke callback on modification. + + Args: + directory: Directory to watch. + callback: Function to call when changes are detected. Receives directory path. + poll_interval: Seconds between polls. + debounce: Seconds to wait after a change before triggering a scan (to batch rapid edits). + **callback_kwargs: Extra kwargs passed to callback. + """ + logger.info("Watching %s (poll=%ss, debounce=%ss)", directory, poll_interval, debounce) + + last_hash = _compute_directory_hash(directory) + last_change_time: float | None = None + + while True: + time.sleep(poll_interval) + current_hash = _compute_directory_hash(directory) + + if current_hash != last_hash: + now = time.time() + if last_change_time is None: + last_change_time = now + + if now - last_change_time >= debounce: + logger.info("Changes detected in %s, triggering scan...", directory) + try: + callback(str(directory), **callback_kwargs) + except Exception: + logger.exception("Error during watch scan callback") + last_hash = _compute_directory_hash(directory) + last_change_time = None + else: + last_change_time = None