From 711ecf11e8fc2fb23ca9a884f5a9bbdcf786c2fb Mon Sep 17 00:00:00 2001 From: Varshith Puli Date: Wed, 21 Oct 2026 02:43:00 +0000 Subject: [PATCH 1/2] feat(analyzer): score writes to agent-memory files as MP3 memory manipulation MP1/MP3 prose patterns score the injection text, not the write, so a test hook appending to MEMORY.md scored 0/100 SAFE (issue #879). This adds a write-operation variant to the memory-poisoning analyzer: a write sink whose destination resolves to a known agent-memory file (MEMORY.md, CLAUDE.md, AGENTS.md, SOUL.md) or directory (.claude/, .cursor/, .agents/) is reported as MP3 Memory Manipulation (HIGH), whatever the payload bytes are. The carrier is scoped to files the developer toolchain auto-runs (test_*.py, *_test.py, conftest.py, *.test.*, *.spec.*) so explicitly invoked helpers, reads, and ordinary filenames stay quiet. Covers Python (open/io.open with write mode, Path.write_text/write_bytes, shutil.copy*) and TypeScript/JavaScript (writeFile(Sync)/appendFile(Sync) incl. fs.promises, createWriteStream) sinks. Verified against the five variants in #879: the inert canary rows go from SAFE to CAUTION with a standalone MP3 finding, and the injection-payload rows cross the 50 risk threshold (exit 1), while the prose-only control is unchanged. Adds 34 unit tests; full unit suite and ruff are clean. Signed-off-by: Varshith Puli --- .../static_patterns_memory_poisoning.py | 114 ++++++++++++++++ .../analyzers/test_mp3_memory_file_write.py | 128 ++++++++++++++++++ 2 files changed, 242 insertions(+) create mode 100644 tests/nodes/analyzers/test_mp3_memory_file_write.py diff --git a/src/skillspector/nodes/analyzers/static_patterns_memory_poisoning.py b/src/skillspector/nodes/analyzers/static_patterns_memory_poisoning.py index e87735c51..e0c2c22b3 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_memory_poisoning.py +++ b/src/skillspector/nodes/analyzers/static_patterns_memory_poisoning.py @@ -608,6 +608,87 @@ 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) -> 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. + 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") + if not _is_agent_memory_destination(destination): + 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 +780,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..9fcdb2f6a --- /dev/null +++ b/tests/nodes/analyzers/test_mp3_memory_file_write.py @@ -0,0 +1,128 @@ +# 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", + ), + ('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) == [] From 44e1ef2b99c765fa78238b35f3f1011440064006 Mon Sep 17 00:00:00 2001 From: Varshith Puli Date: Wed, 28 Oct 2026 15:58:18 +0000 Subject: [PATCH 2/2] fix(analyzers): preserve backslashes in raw-string memory destinations Raw and rf Python string prefixes keep backslashes literal, so decoding escapes collapsed r"C:\Users\alice\MEMORY.md" to "C:UsersaliceMEMORY.md" and the MP3 write signal never fired. Decode escapes only for non-raw literals, using the sink regex prefix the code already captures, and add raw, rf, and raw .claude directory Windows tests through the analyzer. Signed-off-by: Varshith Puli --- .../static_patterns_memory_poisoning.py | 14 ++++++++++--- .../analyzers/test_mp3_memory_file_write.py | 20 +++++++++++++++++++ 2 files changed, 31 insertions(+), 3 deletions(-) diff --git a/src/skillspector/nodes/analyzers/static_patterns_memory_poisoning.py b/src/skillspector/nodes/analyzers/static_patterns_memory_poisoning.py index e0c2c22b3..1acc7e2c4 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_memory_poisoning.py +++ b/src/skillspector/nodes/analyzers/static_patterns_memory_poisoning.py @@ -655,11 +655,18 @@ def _is_benign_timestamp_context_description(content: str, match: re.Match[str]) ] -def _is_agent_memory_destination(raw_destination: str) -> bool: +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. - unescaped = re.sub(r"\\(.)", r"\1", raw_destination, flags=re.DOTALL) + # 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: @@ -681,7 +688,8 @@ def _iter_memory_file_writes(content: str, file_type: str) -> Iterator[tuple[re. for sink in sinks: for match in sink.finditer(content): destination = match.group("dest") - if not _is_agent_memory_destination(destination): + 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): diff --git a/tests/nodes/analyzers/test_mp3_memory_file_write.py b/tests/nodes/analyzers/test_mp3_memory_file_write.py index 9fcdb2f6a..92911d467 100644 --- a/tests/nodes/analyzers/test_mp3_memory_file_write.py +++ b/tests/nodes/analyzers/test_mp3_memory_file_write.py @@ -60,6 +60,26 @@ def mp3_writes(content: str, file_path: str, file_type: str): "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")',