Split shorten_code() formatting from dedup tracking (GHSA-cffv-grgg-g429)

AnalysisContext.shorten_code() mutated a shared reported_shortened_code
set as a side effect of formatting, even though most callers only wanted
the formatted string. It could then shadow findings from future analysis
runs.

For instance, UnsafeImportsML processed every import and would
leave MLAllowlist with an already-reported state for all of them.
Any import outside ML_ALLOWLIST would not be flagged by the second run.

Split shorten_code() into a pure formatter and an explicit mark_reported()
helper.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: thomas-chauchefoin-tob

fickling/analysis.py

@@ -78,19 +78,25 @@ def analyze(self, analysis: Analysis) -> list[AnalysisResult]:
     def results(self) -> AnalysisResults:
         return AnalysisResults(pickled=self.pickled, results=self.previous_results)
 
-    def shorten_code(self, ast_node) -> tuple[str, bool]:
+    def shorten_code(self, ast_node) -> str:
+        """Return a short, human-readable form of an AST node for use in
+        analysis messages. Pure formatter — does not touch dedup state.
+        """
         code = unparse(ast_node).strip()
         if len(code) > 32:
             cutoff = code.find("(")
-            if code[cutoff] == "(":
-                shortened_code = f"{code[: code.find('(')].strip()}(...)"
-            else:
-                shortened_code = code
-        else:
-            shortened_code = code
-        was_already_reported = shortened_code in self.reported_shortened_code
-        self.reported_shortened_code.add(shortened_code)
-        return shortened_code, was_already_reported
+            if cutoff >= 0:
+                return f"{code[:cutoff].strip()}(...)"
+        return code
+
+    def mark_reported(self, shortened: str) -> bool:
+        """Mark a shortened code fragment as reported. Returns True if
+        this was the first mark, False if a prior call already marked it.
+        """
+        if shortened in self.reported_shortened_code:
+            return False
+        self.reported_shortened_code.add(shortened)
+        return True
 
 
 class Analyzer(metaclass=AnalyzerMeta):
@@ -255,8 +261,8 @@ def analyze(self, context: AnalysisContext) -> Iterator[AnalysisResult]:
 class NonStandardImports(Analysis):
     def analyze(self, context: AnalysisContext) -> Iterator[AnalysisResult]:
         for node in context.pickled.non_standard_imports():
-            shortened, already_reported = context.shorten_code(node)
-            if not already_reported:
+            shortened = context.shorten_code(node)
+            if context.mark_reported(shortened):
                 yield AnalysisResult(
                     Severity.LIKELY_UNSAFE,
                     f"`{shortened}` imports a Python module that is not a part of "
@@ -324,7 +330,7 @@ class UnsafeImportsML(Analysis):
 
     def analyze(self, context: AnalysisContext) -> Iterator[AnalysisResult]:
         for node in context.pickled.properties.imports:
-            shortened, _ = context.shorten_code(node)
+            shortened = context.shorten_code(node)
             all_modules = [
                 node.module.rsplit(".", i)[0] for i in range(0, node.module.count(".") + 1)
             ]
@@ -377,7 +383,7 @@ class BadCalls(Analysis):
 
     def analyze(self, context: AnalysisContext) -> Iterator[AnalysisResult]:
         for node in context.pickled.properties.calls:
-            shortened, _already_reported = context.shorten_code(node)
+            shortened = context.shorten_code(node)
             if any(shortened.startswith(f"{c}(") for c in self.BAD_CALLS):
                 yield AnalysisResult(
                     Severity.OVERTLY_MALICIOUS,
@@ -397,7 +403,7 @@ def analyze(self, context: AnalysisContext) -> Iterator[AnalysisResult]:
                 # if the call is to a constructor of an object imported from the Python
                 # standard library, it's probably okay
                 continue
-            shortened, already_reported = context.shorten_code(node)
+            shortened = context.shorten_code(node)
             if (
                 shortened.startswith("eval(")
                 or shortened.startswith("exec(")
@@ -413,7 +419,7 @@ def analyze(self, context: AnalysisContext) -> Iterator[AnalysisResult]:
                     "OvertlyBadEval",
                     trigger=shortened,
                 )
-            elif not already_reported:
+            elif context.mark_reported(shortened):
                 yield AnalysisResult(
                     Severity.LIKELY_UNSAFE,
                     f"Call to `{shortened}` can execute arbitrary code and is inherently unsafe",
@@ -429,7 +435,7 @@ def analyze(self, context: AnalysisContext) -> Iterator[AnalysisResult]:
                 n.name in SAFE_BUILTINS for n in node.names
             ):
                 continue
-            shortened, _ = context.shorten_code(node)
+            shortened = context.shorten_code(node)
             yield AnalysisResult(
                 Severity.LIKELY_OVERTLY_MALICIOUS,
                 f"`{shortened}` is suspicious and indicative of an overtly malicious pickle file",
@@ -447,7 +453,7 @@ def analyze(self, context: AnalysisContext) -> Iterator[AnalysisResult]:
             # Malformed pickle or resource exhaustion - dedicated analyses will report this
             return
         for varname, asmt in unused.items():
-            shortened, _ = context.shorten_code(asmt.value)
+            shortened = context.shorten_code(asmt.value)
             yield AnalysisResult(
                 Severity.SUSPICIOUS,
                 f"Variable `{varname}` is assigned value `{shortened}` but unused afterward; "

fickling/ml.py

@@ -295,9 +295,7 @@ def __init__(self):
 
     def analyze(self, context: AnalysisContext) -> Iterator[AnalysisResult]:
         for node in context.pickled.properties.imports:
-            shortened, already_reported = context.shorten_code(node)
-            if already_reported:
-                continue
+            shortened = context.shorten_code(node)
 
             if isinstance(node, ast.ImportFrom):
                 # from module import x

test/test_bypasses.py

@@ -683,6 +683,28 @@ def test_missing_osx_support(self):
             "from _osx_support import _find_build_tool",
         )
 
+    # https://github.com/trailofbits/fickling/security/advisories/GHSA-cffv-grgg-g429
+    def test_ml_allowlist_not_shadowed_by_unsafe_imports_ml(self):
+        """MLAllowlist must flag imports outside ML_ALLOWLIST even when another
+        analysis (UnsafeImportsML) has already iterated the same import.
+        """
+        pickled = Pickled(
+            [
+                op.Proto.create(4),
+                op.ShortBinUnicode("ast"),
+                op.ShortBinUnicode("parse"),
+                op.StackGlobal(),
+                op.Stop(),
+            ]
+        )
+        res = check_safety(pickled)
+        self.assertGreater(res.severity, Severity.LIKELY_SAFE)
+        detailed = res.detailed_results().get("AnalysisResult", {})
+        self.assertIsNotNone(
+            detailed.get("MLAllowlist"),
+            "MLAllowlist did not produce a finding for `from ast import parse`",
+        )
+
 
 class TestUnsafeModuleCoverage(TestCase):
     """Verify every entry in UNSAFE_MODULES and UNSAFE_IMPORTS triggers detection."""