trailofbits
diff --git a/‎fickling/analysis.py‎
Lines changed: 141 additions & 3 deletions b/‎fickling/analysis.py‎
Lines changed: 141 additions & 3 deletions
diff --git a/‎fickling/cli.py‎
Lines changed: 14 additions & 5 deletions b/‎fickling/cli.py‎
Lines changed: 14 additions & 5 deletions
diff --git a/‎fickling/exception.py‎
Lines changed: 19 additions & 0 deletions b/‎fickling/exception.py‎
Lines changed: 19 additions & 0 deletions
@@ -7,13 +7,22 @@
 from collections.abc import Iterable, Iterator
 from enum import Enum
 
+from fickling.exception import ResourceExhaustionError
 from fickling.fickle import (
     BUILTIN_MODULE_NAMES,
     SAFE_BUILTINS,
+    BinGet,
+    BinPut,
+    Dup,
+    Get,
     InterpretationError,
     Interpreter,
+    LongBinGet,
+    LongBinPut,
+    Memoize,
     Pickled,
     Proto,
+    Put,
 )
 
 
@@ -35,7 +44,29 @@ def __init__(self, pickled: Pickled):
         self.results_by_analysis: dict[type[Analysis], list[AnalysisResult]] = defaultdict(list)
 
     def analyze(self, analysis: Analysis) -> list[AnalysisResult]:
-        results = list(analysis.analyze(self))
+        try:
+            results = list(analysis.analyze(self))
+        except ResourceExhaustionError as e:
+            # Resource limits exceeded - this is a DoS attack indicator
+            results = [
+                AnalysisResult(
+                    Severity.LIKELY_OVERTLY_MALICIOUS,
+                    f"Resource exhaustion detected during analysis: {e}; "
+                    f"this is indicative of an expansion attack (Billion Laughs style)",
+                    "ResourceExhaustion",
+                    trigger=f"{e.resource_type}: {e.actual}",
+                )
+            ]
+        except (ValueError, IndexError, RecursionError) as e:
+            # Malformed pickle caused an interpretation error
+            results = [
+                AnalysisResult(
+                    Severity.LIKELY_UNSAFE,
+                    f"The pickle file has malformed opcode sequences ({type(e).__name__}: {e}); "
+                    f"it is either corrupted or attempting to bypass the pickle security analysis",
+                    "InterpretationError",
+                )
+            ]
         if not results:
             self.results_by_analysis[type(analysis)].append(AnalysisResult(Severity.LIKELY_SAFE))
         else:
@@ -210,6 +241,17 @@ def analyze(self, context: AnalysisContext) -> Iterator[AnalysisResult]:
             )
 
 
+class ResourceExhaustionAnalysis(Analysis):
+    def analyze(self, context: AnalysisContext) -> Iterator[AnalysisResult]:
+        if context.pickled.has_resource_exhaustion:
+            yield AnalysisResult(
+                Severity.LIKELY_OVERTLY_MALICIOUS,
+                "Resource limits were exceeded during interpretation; "
+                "this is indicative of an expansion attack (Billion Laughs style)",
+                "ResourceExhaustion",
+            )
+
+
 class NonStandardImports(Analysis):
     def analyze(self, context: AnalysisContext) -> Iterator[AnalysisResult]:
         for node in context.pickled.non_standard_imports():
@@ -401,8 +443,8 @@ def analyze(self, context: AnalysisContext) -> Iterator[AnalysisResult]:
         interpreter = Interpreter(context.pickled)
         try:
             unused = interpreter.unused_assignments()
-        except InterpretationError:
-            # Malformed pickle - InterpretationErrorAnalysis will report this
+        except (InterpretationError, ResourceExhaustionError):
+            # Malformed pickle or resource exhaustion - dedicated analyses will report this
             return
         for varname, asmt in unused.items():
             shortened, _ = context.shorten_code(asmt.value)
@@ -415,6 +457,102 @@ def analyze(self, context: AnalysisContext) -> Iterator[AnalysisResult]:
             )
 
 
+class ExpansionAttackAnalysis(Analysis):
+    """Detects potential exponential expansion attacks (Billion Laughs style).
+
+    These attacks use:
+    - High GET/PUT ratio: Many GET operations retrieving memoized values
+    - Excessive DUP operations: Duplicating stack items repeatedly
+    """
+
+    # Thresholds for pattern detection
+    DEFAULT_GET_PUT_RATIO_THRESHOLD = 10  # GETs per PUT that is suspicious
+    DEFAULT_HIGH_GET_PUT_RATIO_THRESHOLD = 50  # Extremely high ratio
+    DEFAULT_DUP_COUNT_THRESHOLD = 100  # Number of DUPs that is suspicious
+
+    def __init__(
+        self,
+        *,
+        get_put_ratio_threshold: int = DEFAULT_GET_PUT_RATIO_THRESHOLD,
+        high_get_put_ratio_threshold: int = DEFAULT_HIGH_GET_PUT_RATIO_THRESHOLD,
+        dup_count_threshold: int = DEFAULT_DUP_COUNT_THRESHOLD,
+    ):
+        self._get_put_ratio_threshold = get_put_ratio_threshold
+        self._high_get_put_ratio_threshold = high_get_put_ratio_threshold
+        self._dup_count_threshold = dup_count_threshold
+
+    def analyze(self, context: AnalysisContext) -> Iterator[AnalysisResult]:
+        get_count = 0
+        put_count = 0
+        dup_count = 0
+
+        for opcode in context.pickled:
+            if isinstance(opcode, BinGet | LongBinGet | Get):
+                get_count += 1
+            elif isinstance(opcode, BinPut | LongBinPut | Put | Memoize):
+                put_count += 1
+            elif isinstance(opcode, Dup):
+                dup_count += 1
+
+        findings: list[AnalysisResult] = []
+
+        # Check for high GET/PUT ratio
+        if put_count > 0:
+            ratio = get_count / put_count
+            if ratio > self._high_get_put_ratio_threshold:
+                findings.append(
+                    AnalysisResult(
+                        Severity.LIKELY_UNSAFE,
+                        f"Extremely high GET/PUT ratio ({ratio:.1f}:1) detected; "
+                        f"this pattern is indicative of an exponential expansion attack "
+                        f"(Billion Laughs style) that could cause DoS",
+                        "ExpansionAttackAnalysis",
+                        trigger=f"GET/PUT ratio: {ratio:.1f}:1",
+                    )
+                )
+            elif ratio > self._get_put_ratio_threshold:
+                findings.append(
+                    AnalysisResult(
+                        Severity.SUSPICIOUS,
+                        f"High GET/PUT ratio ({ratio:.1f}:1) detected; "
+                        f"this may indicate an expansion attack pattern",
+                        "ExpansionAttackAnalysis",
+                        trigger=f"GET/PUT ratio: {ratio:.1f}:1",
+                    )
+                )
+        elif get_count > self._get_put_ratio_threshold:
+            # GETs with no PUTs is inherently malformed/malicious
+            findings.append(
+                AnalysisResult(
+                    Severity.LIKELY_UNSAFE,
+                    f"GET operations ({get_count}) with no PUT operations detected; "
+                    f"this is indicative of a malformed or malicious pickle",
+                    "ExpansionAttackAnalysis",
+                    trigger=f"GET count: {get_count}, PUT count: 0",
+                )
+            )
+
+        # Check for excessive DUP operations
+        if dup_count > self._dup_count_threshold:
+            findings.append(
+                AnalysisResult(
+                    Severity.SUSPICIOUS,
+                    f"Excessive DUP operations ({dup_count}) detected; "
+                    f"this may indicate a stack duplication attack",
+                    "ExpansionAttackAnalysis",
+                    trigger=f"DUP count: {dup_count}",
+                )
+            )
+
+        # Multiple indicators together are more severe
+        if len(findings) > 1:
+            for finding in findings:
+                if finding.severity < Severity.LIKELY_UNSAFE:
+                    finding.severity = Severity.LIKELY_UNSAFE
+
+        yield from findings
+
+
 class AnalysisResults:
     def __init__(self, pickled: Pickled, results: Iterable[AnalysisResult]):
         self.pickled: Pickled = pickled
 
@@ -7,6 +7,7 @@
 from . import __version__, fickle, tracing
 from .analysis import Severity, check_safety
 from .constants import EXIT_CLEAN, EXIT_ERROR, EXIT_UNSAFE
+from .exception import ResourceExhaustionError
 
 DEFAULT_JSON_OUTPUT_FILE = "safety_results.json"
 
@@ -183,11 +184,19 @@ def main(argv: list[str] | None = None) -> int:
                 interpreter = fickle.Interpreter(
                     pickled, first_variable_id=var_id, result_variable=f"result{i}"
                 )
-                if args.trace:
-                    trace = tracing.Trace(interpreter)
-                    print(unparse(trace.run()))
-                else:
-                    print(unparse(interpreter.to_ast()))
+                try:
+                    if args.trace:
+                        trace = tracing.Trace(interpreter)
+                        print(unparse(trace.run()))
+                    else:
+                        print(unparse(interpreter.to_ast()))
+                except ResourceExhaustionError as e:
+                    sys.stderr.write(
+                        f"Error: {e}\n"
+                        "This pickle file may contain an expansion attack. "
+                        "Use --check-safety to analyze it.\n"
+                    )
+                    return 1
                 var_id = interpreter.next_variable_id
     else:
         pickled = fickle.Pickled(
 
@@ -15,3 +15,22 @@ def __init__(self, msg):
 
     def __str__(self):
         return self.msg
+
+
+class ResourceExhaustionError(Exception):
+    """Raised when resource limits are exceeded during analysis."""
+
+    def __init__(self, resource_type: str, limit: int, actual: int):
+        self.resource_type = resource_type
+        self.limit = limit
+        self.actual = actual
+        super().__init__(
+            f"Resource limit exceeded: {resource_type} (limit={limit}, actual={actual})"
+        )
+
+
+class ExpansionAttackError(ResourceExhaustionError):
+    """Raised when exponential expansion attack (Billion Laughs style) is detected."""
+
+    def __init__(self, limit: int, actual: int):
+        super().__init__("get_ratio", limit, actual)