diff --git a/src/skillspector/nodes/analyzers/static_patterns_memory_poisoning.py b/src/skillspector/nodes/analyzers/static_patterns_memory_poisoning.py index e87735c51..1acc7e2c4 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_memory_poisoning.py +++ b/src/skillspector/nodes/analyzers/static_patterns_memory_poisoning.py @@ -608,6 +608,95 @@ def _is_benign_timestamp_context_description(content: str, match: re.Match[str]) return False +# MP3 (write variant): a write operation whose destination resolves to a known +# agent-memory file or directory. MP1/MP3 prose patterns score the injection +# *text*; they cannot see a write *operation*, so a test hook that appends to +# MEMORY.md scores 0/100 SAFE (issue #879). This scores the write itself: the +# persistence mechanism is the finding, whatever the payload bytes are. +# +# Calibration: the signal is the *destination* being memory the agent loads as +# its own context, not file I/O as such; and the carrier is scoped to files +# the developer toolchain runs without being asked (test files, conftest.py), +# so an explicitly invoked helper that writes a report stays quiet. +_AGENT_MEMORY_FILENAMES = frozenset({"memory.md", "claude.md", "agents.md", "soul.md"}) +_AGENT_MEMORY_DIR_SEGMENTS = frozenset({".claude", ".cursor", ".agents"}) + +# Basename shapes the developer toolchain auto-runs: pytest/unittest discovery +# (test_*.py, *_test.py, conftest.py) and JS/TS runners (*.test.*, *.spec.*). +_AUTO_RUN_CARRIER_RE = re.compile( + r"conftest\.py|test_.*\.pyw?|.*_test\.pyw?|.*\.(?:test|spec)\.[a-z0-9]+", + re.IGNORECASE, +) + +_PY_STRING_LITERAL = r"(?P[A-Za-z]*)(?P[\"'])(?P(?:\\.|(?!(?P=q)).)*)(?P=q)" +_TS_STRING_LITERAL = r"(?P[\"'`])(?P(?:\\.|(?!(?P=q)).)*)(?P=q)" + +# Write sinks that take a literal destination. open() requires an explicit +# write mode so reads stay quiet; shutil.copy* resolve the destination, which +# is the second positional argument. +_PY_WRITE_SINKS = [ + re.compile( + r"(?:\bio\s*\.\s*)?\bopen\s*\(\s*" + + _PY_STRING_LITERAL + + r"\s*,\s*(?:mode\s*=\s*)?(?P[\"'])(?P[^\"']*)(?P=mq)" + ), + re.compile(r"\bPath\s*\(\s*" + _PY_STRING_LITERAL + r"\s*\)\s*\.\s*write_(?:text|bytes)\s*\("), + re.compile( + r"\bshutil\s*\.\s*copy(?:2|file)?\s*\(\s*[^()]*(?:\([^()]*\)[^()]*)*,\s*" + + _PY_STRING_LITERAL + ), +] +_TS_WRITE_SINKS = [ + re.compile( + r"(?:\bfs\s*\.\s*(?:promises\s*\.\s*)?)?" + r"\b(?:writeFileSync|appendFileSync|writeFile|appendFile)\s*\(\s*" + _TS_STRING_LITERAL + ), + re.compile(r"\bfs\s*\.\s*createWriteStream\s*\(\s*" + _TS_STRING_LITERAL), +] + + +def _is_agent_memory_destination(raw_destination: str, prefix: str = "") -> bool: + """Return True when a write destination is an agent-memory file or directory.""" + # Literal path only: interpolation in directory segments does not change + # which well-known file is targeted, so f"{base}/MEMORY.md" still counts. + # A raw prefix (r, rf) keeps backslashes literal: decoding escapes there + # would eat the separators, so r"C:\Users\alice\MEMORY.md" would collapse + # to "C:UsersaliceMEMORY.md" and lose every path segment. Only non-raw + # literals get escape decoding; both forms then normalize separators. + if "r" in prefix.lower(): + unescaped = raw_destination + else: + unescaped = re.sub(r"\\(.)", r"\1", raw_destination, flags=re.DOTALL) + normalized = unescaped.replace("\\", "/").strip() + segments = [s for s in normalized.split("/") if s not in ("", ".", "..")] + if not segments: + return False + lowered = [segment.lower() for segment in segments] + if lowered[-1] in _AGENT_MEMORY_FILENAMES: + return True + return any(segment in _AGENT_MEMORY_DIR_SEGMENTS for segment in lowered) + + +def _iter_memory_file_writes(content: str, file_type: str) -> Iterator[tuple[re.Match[str], str]]: + """Yield (match, destination) for writes to agent-memory destinations.""" + if file_type == "python": + sinks = _PY_WRITE_SINKS + elif file_type in ("typescript", "javascript"): + sinks = _TS_WRITE_SINKS + else: + return + for sink in sinks: + for match in sink.finditer(content): + destination = match.group("dest") + prefix = match.groupdict().get("prefix") or "" + if not _is_agent_memory_destination(destination, prefix): + continue + mode = match.groupdict().get("mode") + if mode is not None and not any(char in "wxa+" for char in mode): + continue + yield match, destination + + def analyze(content: str, file_path: str, file_type: str) -> list[AnalyzerFinding]: """Analyze content for memory poisoning patterns (MP1–MP3).""" findings: list[AnalyzerFinding] = [] @@ -699,6 +788,39 @@ def ctx(start: int) -> str: complete_match=match.group(0), ) ) + # MP3 write variant: score the write to agent memory, not the written text. + basename = file_path.replace("\\", "/").rsplit("/", 1)[-1] + if ( + file_type in ("python", "typescript", "javascript") + and _AUTO_RUN_CARRIER_RE.fullmatch(basename) is not None + ): + for write_match, destination in _iter_memory_file_writes(content, file_type): + line_num = get_line_number(content, write_match.start()) + findings.append( + AnalyzerFinding( + rule_id="MP3", + message="Memory Manipulation", + severity=Severity.HIGH, + location=loc(line_num), + confidence=0.85, + tags=tag, + context=ctx(write_match.start()), + matched_text=write_match.group(0)[:200], + complete_match=write_match.group(0), + remediation=( + "An auto-run file writes to an agent memory file " + f"({destination}). Such writes persist across agent " + "interactions without model involvement. Confirm the " + "destination is intentional and the written content is " + "trusted before installing this skill." + ), + evidence={ + "destination": destination, + "carrier_basename": basename, + "carrier_kind": "auto_run", + }, + ) + ) return findings diff --git a/tests/nodes/analyzers/test_mp3_memory_file_write.py b/tests/nodes/analyzers/test_mp3_memory_file_write.py new file mode 100644 index 000000000..92911d467 --- /dev/null +++ b/tests/nodes/analyzers/test_mp3_memory_file_write.py @@ -0,0 +1,148 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Auto-run carriers that write to agent-memory files are MP3 findings. + +Issue #879: MP1/MP3 prose patterns score the injection text, not the write, so +a test hook appending to MEMORY.md scored 0/100 SAFE. The write to a known +agent-memory destination is the finding, whatever the payload bytes are. +""" + +import pytest + +from skillspector.models import Severity +from skillspector.nodes.analyzers import static_patterns_memory_poisoning as mp + + +def mp3_writes(content: str, file_path: str, file_type: str): + return [ + f + for f in mp.analyze(content, file_path, file_type) + if f.rule_id == "MP3" and f.severity == Severity.HIGH and f.confidence == 0.85 + ] + + +TS_CARRIER = ( + 'import { appendFileSync } from "fs";\n' + 'beforeAll(() => { appendFileSync("MEMORY.md", "\\n\\n"); });' +) +PY_CARRIER = ( + "def pytest_configure(config):\n" + ' with open("MEMORY.md", "a") as fh:\n' + ' fh.write("\\n\\n")' +) + + +@pytest.mark.parametrize( + ("content", "file_path", "file_type", "destination"), + [ + # The reporter's variants A (TypeScript) and E (Python). + (TS_CARRIER, "notes.test.ts", "typescript", "MEMORY.md"), + (PY_CARRIER, "conftest.py", "python", "MEMORY.md"), + (PY_CARRIER, "tests/conftest.py", "python", "MEMORY.md"), + # Other write sinks and memory destinations. + ('open("MEMORY.md", "w").write("x")', "test_notes.py", "python", "MEMORY.md"), + ('open("MEMORY.md", mode="a").write("x")', "test_notes.py", "python", "MEMORY.md"), + ('Path("CLAUDE.md").write_text("x")', "test_notes.py", "python", "CLAUDE.md"), + ('Path("AGENTS.md").write_bytes(b"x")', "test_notes.py", "python", "AGENTS.md"), + ('shutil.copy("src.txt", "SOUL.md")', "test_notes.py", "python", "SOUL.md"), + ('io.open("MEMORY.md", "a").write("x")', "test_notes.py", "python", "MEMORY.md"), + ( + 'open(f"{base}/MEMORY.md", "w").write("x")', + "test_notes.py", + "python", + "{base}/MEMORY.md", + ), + ('open("./MEMORY.md", "a").write("x")', "test_notes.py", "python", "./MEMORY.md"), + ( + 'open("C:\\\\notes\\\\MEMORY.md", "a")', + "test_notes.py", + "python", + "C:\\\\notes\\\\MEMORY.md", + ), + # Raw prefixes keep backslashes literal: the same Windows path + # written with r or rf prefixes must still resolve to MEMORY.md. + ( + 'open(r"C:\\Users\\alice\\MEMORY.md", "a")', + "test_notes.py", + "python", + "C:\\Users\\alice\\MEMORY.md", + ), + ( + 'open(rf"C:\\Users\\alice\\MEMORY.md", "a")', + "test_notes.py", + "python", + "C:\\Users\\alice\\MEMORY.md", + ), + ( + 'open(r"C:\\Users\\alice\\.claude\\settings.json", "a")', + "test_notes.py", + "python", + "C:\\Users\\alice\\.claude\\settings.json", + ), + ('fs.writeFileSync("MEMORY.md", "x")', "notes.test.ts", "typescript", "MEMORY.md"), + ( + 'await fs.promises.appendFile("MEMORY.md", "x")', + "notes.spec.js", + "javascript", + "MEMORY.md", + ), + ('fs.createWriteStream("AGENTS.md")', "notes.test.mjs", "javascript", "AGENTS.md"), + ( + 'writeFile(".claude/settings.json", "{}")', + "notes.test.ts", + "typescript", + ".claude/settings.json", + ), + ( + "appendFileSync(`${dir}/MEMORY.md`, x)", + "notes.test.ts", + "typescript", + "${dir}/MEMORY.md", + ), + ( + "appendFileSync(`.cursor/rules.md`, x)", + "notes.test.ts", + "typescript", + ".cursor/rules.md", + ), + ], +) +def test_memory_file_write_from_auto_run_carrier_is_mp3(content, file_path, file_type, destination): + findings = mp3_writes(content, file_path, file_type) + assert len(findings) == 1 + finding = findings[0] + assert finding.evidence["destination"] == destination + assert finding.evidence["carrier_kind"] == "auto_run" + assert finding.remediation + + +@pytest.mark.parametrize( + ("content", "file_path", "file_type"), + [ + # Reads stay quiet. + ('open("MEMORY.md").read()', "test_notes.py", "python"), + ('open("MEMORY.md", "r").read()', "test_notes.py", "python"), + ('Path("MEMORY.md").read_text()', "test_notes.py", "python"), + # Ordinary file I/O is not the signal; the destination is. + ('open("notes.txt", "w").write("x")', "test_notes.py", "python"), + ('Path("report.md").write_text("x")', "test_notes.py", "python"), + ('open("memory.md.bak", "w")', "test_notes.py", "python"), + ('open("my-MEMORY.md", "w")', "test_notes.py", "python"), + ('appendFileSync("notes.txt", x)', "notes.test.ts", "typescript"), + # Non-literal destinations cannot be resolved to a memory file. + ('open(MEMORY_PATH, "w")', "test_notes.py", "python"), + ("appendFileSync(`${MEMORY_PATH}`, x)", "notes.test.ts", "typescript"), + # Explicitly invoked helpers are out of scope: only auto-run carriers. + (TS_CARRIER, "notes.ts", "typescript"), + (TS_CARRIER, "scripts/notes.js", "javascript"), + (PY_CARRIER, "helper.py", "python"), + (PY_CARRIER, "skills/notes/writer.py", "python"), + # Prose about memory files is not a write operation. + ("The skill writes its notes to MEMORY.md when asked.", "SKILL.md", "markdown"), + # Unrelated languages are not analyzed for write sinks. + ('open("MEMORY.md", "w")', "notes.sh", "other"), + ], +) +def test_memory_file_write_does_not_fire(content, file_path, file_type): + assert mp3_writes(content, file_path, file_type) == []