Add configurable loop_analysis_max_depth and max_xrefs limits across config, CLI, and batch extractor

Author: marcosd4h

deep_extract/config.py

@@ -43,6 +43,8 @@ class AnalysisConfig:
         force_reanalyze: Force re-analysis even if already completed
         thunk_depth: Maximum depth for thunk resolution
         min_conf: Minimum confidence for function call validation
+        loop_analysis_max_depth: Maximum recursion depth for loop analysis graph traversal
+        max_xrefs: Maximum xrefs serialized per JSON xref field
         
         # Computed file metadata
         file_hashes: MD5/SHA256 hashes of the input file
@@ -76,6 +78,8 @@ class AnalysisConfig:
     force_reanalyze: bool = False
     thunk_depth: Optional[int] = None
     min_conf: Optional[int] = None
+    loop_analysis_max_depth: Optional[int] = None
+    max_xrefs: Optional[int] = None
 
     # Advanced analysis options
     use_interprocedural_analysis: bool = True  # Enable deeper indirect call resolution
@@ -134,6 +138,16 @@ def __post_init__(self):
                 raise ValueError(f"min_conf must be a number, got: {type(self.min_conf)}")
             if not 10 <= self.min_conf <= 100:
                 raise ValueError(f"min_conf must be between 10 and 100, got: {self.min_conf}")
+
+        if self.loop_analysis_max_depth is not None:
+            if not isinstance(self.loop_analysis_max_depth, int) or self.loop_analysis_max_depth < 1:
+                raise ValueError(
+                    f"loop_analysis_max_depth must be a positive integer, got: {self.loop_analysis_max_depth}"
+                )
+
+        if self.max_xrefs is not None:
+            if not isinstance(self.max_xrefs, int) or self.max_xrefs < 1:
+                raise ValueError(f"max_xrefs must be a positive integer, got: {self.max_xrefs}")
 
         # Validate boolean flags
         bool_fields = [
@@ -237,6 +251,24 @@ def from_ida_args(cls, args: dict, input_file: str) -> 'AnalysisConfig':
                     raise ValueError(f"min_conf must be 10-100, got: {min_conf}")
             except (TypeError, ValueError) as e:
                 raise ValueError(f"Invalid min_conf: {e}")
+
+        loop_analysis_max_depth = args.get('loop_analysis_max_depth')
+        if loop_analysis_max_depth is not None:
+            try:
+                loop_analysis_max_depth = int(loop_analysis_max_depth)
+                if loop_analysis_max_depth < 1:
+                    raise ValueError(f"loop_analysis_max_depth must be positive, got: {loop_analysis_max_depth}")
+            except (TypeError, ValueError) as e:
+                raise ValueError(f"Invalid loop_analysis_max_depth: {e}")
+
+        max_xrefs = args.get('max_xrefs')
+        if max_xrefs is not None:
+            try:
+                max_xrefs = int(max_xrefs)
+                if max_xrefs < 1:
+                    raise ValueError(f"max_xrefs must be positive, got: {max_xrefs}")
+            except (TypeError, ValueError) as e:
+                raise ValueError(f"Invalid max_xrefs: {e}")
 
         return cls(
             sqlite_db_path=sqlite_db_path,
@@ -258,6 +290,8 @@ def from_ida_args(cls, args: dict, input_file: str) -> 'AnalysisConfig':
             use_interprocedural_analysis=bool(args.get('use_interprocedural_analysis', True)),
             thunk_depth=thunk_depth,
             min_conf=min_conf,
+            loop_analysis_max_depth=loop_analysis_max_depth,
+            max_xrefs=max_xrefs,
         )
 
     def to_dict(self) -> dict:
@@ -286,6 +320,8 @@ def to_dict(self) -> dict:
             'use_interprocedural_analysis': self.use_interprocedural_analysis,
             'thunk_depth': self.thunk_depth,
             'min_conf': self.min_conf,
+            'loop_analysis_max_depth': self.loop_analysis_max_depth,
+            'max_xrefs': self.max_xrefs,
         }
 
     def to_analysis_flags_json(self) -> str:
@@ -318,5 +354,7 @@ def get_analysis_args_dict(self) -> dict:
             'force_reanalyze': self.force_reanalyze,
             'generate_cpp': self.generate_cpp,
             'generate_asm': self.generate_asm,
+            'loop_analysis_max_depth': self.loop_analysis_max_depth,
+            'max_xrefs': self.max_xrefs,
         }
 

deep_extract/constants.py

@@ -39,8 +39,13 @@
 # Jump Table Analysis Thresholds
 MIN_JUMP_TABLE_CONFIDENCE = 30.0       # Minimum confidence percentage to accept jump table detection
 
+# Xref Serialization Configuration
+DEFAULT_MAX_XREFS = 500000       # Default maximum xrefs serialized per JSON xref field
+MIN_MAX_XREFS = 1000             # Minimum accepted EXTRACTOR_MAX_XREFS override
+
 # Loop Analysis Configuration
 DEFAULT_LOOP_BLOCK_LIMIT = 5000  # Default maximum basic blocks before skipping loop analysis
+DEFAULT_LOOP_ANALYSIS_MAX_DEPTH = 5000  # Default recursion depth guard for loop graph traversal
 LOOP_ANALYSIS_WARNING_THRESHOLD = 500  # Warn when function exceeds this block count
 
 # Batch Processing Configuration

deep_extract/json_safety.py

@@ -12,6 +12,7 @@
 from datetime import datetime
 from decimal import Decimal
 
+from . import constants
 from .logging_utils import debug_print
 
 
@@ -537,27 +538,32 @@ def apply_field_limits(xref_list: List[Any], max_xrefs: int = 10000) -> List[Any
 
 
 # Convenience functions for common field types
-def _get_max_xrefs() -> int:
+def _get_max_xrefs(override: Optional[int] = None) -> int:
     """Get the max xref limit from environment with sane fallback."""
-    raw = os.environ.get("EXTRACTOR_MAX_XREFS", "")
-    try:
-        value = int(raw)
-    except (TypeError, ValueError):
-        value = 50000
-    # Clamp to a reasonable range to avoid zero/negative or runaway values.
-    if value < 1000:
-        value = 1000
-    if value > 500000:
-        value = 500000
+    if override is not None:
+        value = int(override)
+    else:
+        raw = os.environ.get("EXTRACTOR_MAX_XREFS", "")
+        try:
+            value = int(raw)
+        except (TypeError, ValueError):
+            value = constants.DEFAULT_MAX_XREFS
+    # Clamp only the lower bound; callers can opt into larger values when needed.
+    if value < constants.MIN_MAX_XREFS:
+        value = constants.MIN_MAX_XREFS
     return value
 
 
-def safe_serialize_xrefs(xrefs: List[Dict[str, Any]], field_name: str = "xrefs") -> str:
+def safe_serialize_xrefs(
+    xrefs: List[Dict[str, Any]],
+    field_name: str = "xrefs",
+    max_xrefs: Optional[int] = None,
+) -> str:
     """Safely serialize cross-reference list."""
-    max_xrefs = _get_max_xrefs()
-    if len(xrefs) > max_xrefs:
-        debug_print(f"WARNING - Xref list exceeds limit ({len(xrefs)} > {max_xrefs}); truncation may occur")
-    return to_json_safe(xrefs, max_list_items=max_xrefs, field_name=field_name)
+    effective_max_xrefs = _get_max_xrefs(max_xrefs)
+    if len(xrefs) > effective_max_xrefs:
+        debug_print(f"WARNING - Xref list exceeds limit ({len(xrefs)} > {effective_max_xrefs}); truncation may occur")
+    return to_json_safe(xrefs, max_list_items=effective_max_xrefs, field_name=field_name)
 
 
 def safe_serialize_strings(strings: List[str], field_name: str = "strings") -> str:

deep_extract/loop_analysis.py

@@ -7,6 +7,8 @@
 """
 
 from typing import Dict, List, Tuple, Any, Optional
+import os
+import sys
 import ida_funcs
 import ida_bytes
 import ida_ua
@@ -24,6 +26,7 @@
 def extract_loop_analysis(function_ea: int, 
                          quantity_limit: bool = True, 
                          chunk_quantity: Optional[int] = None,
+                         max_depth: Optional[int] = None,
                          config: Optional[Any] = None) -> Dict[str, Any]:
     """
     Analyzes the control flow graph of a function to identify all loops.
@@ -35,6 +38,8 @@ def extract_loop_analysis(function_ea: int,
     :param quantity_limit: A flag to limit analysis on huge functions.
     :param chunk_quantity: The block count threshold to trigger the limit.
                           If None, uses DEFAULT_LOOP_BLOCK_LIMIT from constants.
+    :param max_depth: Recursion depth guard for graph traversals. If None, uses
+                      config, EXTRACTOR_LOOP_ANALYSIS_MAX_DEPTH, or the default.
     :param config: Optional AnalysisConfig object to get limit from config.
     :return: A dictionary containing the loops found and a count of them.
     """
@@ -49,8 +54,23 @@ def extract_loop_analysis(function_ea: int,
         else:
             # Fall back to constant
             chunk_quantity = constants.DEFAULT_LOOP_BLOCK_LIMIT
+
+    if max_depth is None:
+        if config and getattr(config, 'loop_analysis_max_depth', None) is not None:
+            max_depth = config.loop_analysis_max_depth
+        else:
+            raw_max_depth = os.environ.get("EXTRACTOR_LOOP_ANALYSIS_MAX_DEPTH", "")
+            try:
+                max_depth = int(raw_max_depth) if raw_max_depth else constants.DEFAULT_LOOP_ANALYSIS_MAX_DEPTH
+            except (TypeError, ValueError):
+                max_depth = constants.DEFAULT_LOOP_ANALYSIS_MAX_DEPTH
+    max_depth = max(1, int(max_depth))
 
     try:
+        previous_recursion_limit = sys.getrecursionlimit()
+        if previous_recursion_limit <= max_depth:
+            sys.setrecursionlimit(max_depth + 100)
+
         func = ida_funcs.get_func(function_ea)
         if not func:
             return {"error": "Function not found.", "loops": [], "loop_count": 0}
@@ -121,7 +141,12 @@ def extract_loop_analysis(function_ea: int,
         on_stack = [False] * n
         sccs = []
 
-        def strongconnect(v):
+        def strongconnect(v, depth=0):
+            if depth > max_depth:
+                raise RecursionError(
+                    f"loop analysis max depth exceeded ({max_depth}); "
+                    "set EXTRACTOR_LOOP_ANALYSIS_MAX_DEPTH or --loop-analysis-max-depth to increase"
+                )
             indices[v] = index_counter[0]
             lowlinks[v] = index_counter[0]
             index_counter[0] += 1
@@ -130,7 +155,7 @@ def strongconnect(v):
 
             for w in successors[v]:
                 if indices[w] == -1:
-                    strongconnect(w)
+                    strongconnect(w, depth + 1)
                     lowlinks[v] = min(lowlinks[v], lowlinks[w])
                 elif on_stack[w]:
                     lowlinks[v] = min(lowlinks[v], indices[w])
@@ -223,15 +248,20 @@ def _count_back_edges(loop_nodes: set) -> int:
             stack_set = set()
             count = 0
 
-            def dfs(node_id):
+            def dfs(node_id, depth=0):
                 nonlocal count
+                if depth > max_depth:
+                    raise RecursionError(
+                        f"loop analysis max depth exceeded ({max_depth}); "
+                        "set EXTRACTOR_LOOP_ANALYSIS_MAX_DEPTH or --loop-analysis-max-depth to increase"
+                    )
                 visited.add(node_id)
                 stack_set.add(node_id)
                 for succ_id in successors[node_id]:
                     if succ_id not in loop_nodes:
                         continue
                     if succ_id not in visited:
-                        dfs(succ_id)
+                        dfs(succ_id, depth + 1)
                     elif succ_id in stack_set:
                         count += 1
                 stack_set.remove(node_id)
@@ -332,4 +362,7 @@ def dfs(node_id):
     except Exception as e:
         debug_print(f"ERROR - [Loop Analysis] for 0x{function_ea:X}: {str(e)}")
         return {"error": str(e), "loops": [], "loop_count": 0}
+    finally:
+        if 'previous_recursion_limit' in locals() and sys.getrecursionlimit() != previous_recursion_limit:
+            sys.setrecursionlimit(previous_recursion_limit)
 

deep_extract/pe_context_extractor.py

@@ -775,6 +775,8 @@ def parse_arguments() -> Optional[dict]:
     parser.add_argument("--asm-output-dir", type=str, default=None, help="Directory for ASM output files (defaults to same as cpp-output-dir or extracted_raw_code/ next to db)")
     parser.add_argument("--thunk-depth", type=int, default=None)
     parser.add_argument("--min-call-conf", type=float, default=None)
+    parser.add_argument("--loop-analysis-max-depth", type=int, default=None)
+    parser.add_argument("--max-xrefs", type=int, default=None)
 
     # Process argv from IDA
     # idc.ARGV contains all arguments including script name
@@ -1272,7 +1274,10 @@ def _process_single_function(ea: int, string_map: Dict[int, Any], has_decompiler
         if options.get('analyze_loops') and extractor_core.HAS_LOOP_DETECTOR:
             loop_start = time.perf_counter()
             try:
-                loop_analysis = extractor_core.extract_loop_analysis(ea)
+                loop_analysis = extractor_core.extract_loop_analysis(
+                    ea,
+                    max_depth=options.get('loop_analysis_max_depth'),
+                )
             except Exception as e:
                 loop_analysis = None
                 analysis_errors.append({"stage": "loop_analysis", "error": str(e)})
@@ -1334,10 +1339,18 @@ def _process_single_function(ea: int, string_map: Dict[int, Any], has_decompiler
             'function_name': extractor_core.get_raw_function_name(ea),
             'assembly_code': '\n'.join(assembly_lines),
             'decompiled_code': decompiled_code,
-            'inbound_xrefs': json_safety.safe_serialize_xrefs(xref_data["inbound_xrefs"], "inbound_xrefs"),
-            'outbound_xrefs': json_safety.safe_serialize_xrefs(xref_data["outbound_xrefs"], "outbound_xrefs"),
-            'simple_inbound_xrefs': json_safety.safe_serialize_xrefs(simple_inbound, "simple_inbound_xrefs"),
-            'simple_outbound_xrefs': json_safety.safe_serialize_xrefs(simple_outbound, "simple_outbound_xrefs"),
+            'inbound_xrefs': json_safety.safe_serialize_xrefs(
+                xref_data["inbound_xrefs"], "inbound_xrefs", max_xrefs=options.get('max_xrefs')
+            ),
+            'outbound_xrefs': json_safety.safe_serialize_xrefs(
+                xref_data["outbound_xrefs"], "outbound_xrefs", max_xrefs=options.get('max_xrefs')
+            ),
+            'simple_inbound_xrefs': json_safety.safe_serialize_xrefs(
+                simple_inbound, "simple_inbound_xrefs", max_xrefs=options.get('max_xrefs')
+            ),
+            'simple_outbound_xrefs': json_safety.safe_serialize_xrefs(
+                simple_outbound, "simple_outbound_xrefs", max_xrefs=options.get('max_xrefs')
+            ),
             'vtable_contexts': json_safety.safe_serialize_vtable_contexts(xref_data["vtable_contexts"], "vtable_contexts"),
             'global_var_accesses': json_safety.to_json_safe(global_accesses, max_list_items=1000, field_name="global_var_accesses"),
             'loop_analysis': json_safety.safe_serialize_loop_analysis(loop_analysis, "loop_analysis"),
@@ -1368,6 +1381,8 @@ def extract_all_functions(sqlite_db_path, hashes, imports_json, exports_json, en
         'extract_stack_frame': kwargs.get('extract_stack_frame', True),
         'extract_globals': kwargs.get('extract_globals', True),
         'analyze_loops': kwargs.get('analyze_loops', True),
+        'loop_analysis_max_depth': kwargs.get('loop_analysis_max_depth'),
+        'max_xrefs': kwargs.get('max_xrefs'),
     }
 
     profile_enabled = get_log_level() in ("TRACE", "DEBUG")

headless_batch_extractor.ps1

@@ -34,6 +34,12 @@
 .PARAMETER TimeoutHours
     Maximum number of hours each IDA process may run before being terminated (default: 6).
 
+.PARAMETER LoopAnalysisMaxDepth
+    Maximum recursion depth for loop analysis graph traversal. Overrides the extractor default.
+
+.PARAMETER MaxXrefs
+    Maximum xrefs serialized per JSON xref field. Overrides the extractor default.
+
 .PARAMETER Help
     Display detailed help information with usage examples.
 
@@ -120,6 +126,12 @@ param(
     [ValidateRange(1, [int]::MaxValue)]
     [int]$TimeoutHours = 6,
 
+    [Parameter(HelpMessage = "Maximum recursion depth for loop analysis graph traversal (0 = extractor default)")]
+    [int]$LoopAnalysisMaxDepth = 0,
+
+    [Parameter(HelpMessage = "Maximum xrefs serialized per JSON xref field (0 = extractor default)")]
+    [int]$MaxXrefs = 0,
+
     # Display help
     [Parameter(ParameterSetName = 'Help', HelpMessage = "Display detailed help information")]
     [Alias('h', '?')]
@@ -194,6 +206,8 @@ function Show-Help {
     Write-Host "  -IdaPath <path>         Path to IDA Pro executable (auto-detected if not specified)"
     Write-Host "  -MaxConcurrentProcesses Number of parallel IDA processes (default: 4)"
     Write-Host "  -TimeoutHours <hours>   Per-process timeout before termination (default: 6)"
+    Write-Host "  -LoopAnalysisMaxDepth   Loop analysis recursion depth guard (extractor default if omitted)"
+    Write-Host "  -MaxXrefs <count>       Maximum xrefs serialized per JSON xref field (extractor default if omitted)"
     Write-Host "  -Help                   Display this help message"
     Write-Host ""
     Write-Host "ANALYSIS FLAGS (disable specific features):" -ForegroundColor Yellow
@@ -248,6 +262,12 @@ function Show-Help {
     Write-Host "  # Allow very large modules to run longer before timeout" -ForegroundColor Gray
     Write-Host "  .\headless_batch_extractor.ps1 -ExtractDirRecursive 'C:\Binaries' -StorageDir 'C:\Analysis' -TimeoutHours 50"
     Write-Host ""
+    Write-Host "  # Increase loop analysis depth for complex CFGs" -ForegroundColor Gray
+    Write-Host "  .\headless_batch_extractor.ps1 -ExtractDirRecursive 'C:\Binaries' -StorageDir 'C:\Analysis' -LoopAnalysisMaxDepth 10000"
+    Write-Host ""
+    Write-Host "  # Increase xref serialization limit for dense modules" -ForegroundColor Gray
+    Write-Host "  .\headless_batch_extractor.ps1 -ExtractDirRecursive 'C:\Binaries' -StorageDir 'C:\Analysis' -MaxXrefs 1000000"
+    Write-Host ""
     Write-Host "  # Skip string extraction to reduce analysis time" -ForegroundColor Gray
     Write-Host "  .\headless_batch_extractor.ps1 -ExtractDir 'C:\Binaries' -StorageDir 'C:\Analysis' -NoExtractStrings"
     Write-Host ""
@@ -1262,6 +1282,14 @@ function Start-IDAProcesses {
             if (-not $analysisFlags.extract_pe_metadata) { $idaScriptArgs.Add("--no-pe-metadata") }
             if (-not $analysisFlags.extract_advanced_pe) { $idaScriptArgs.Add("--no-advanced-pe") }
             if (-not $analysisFlags.extract_runtime_info) { $idaScriptArgs.Add("--no-runtime-info") }
+            if ($analysisFlags.loop_analysis_max_depth -and $analysisFlags.loop_analysis_max_depth -gt 0) {
+                $idaScriptArgs.Add("--loop-analysis-max-depth")
+                $idaScriptArgs.Add([string]$analysisFlags.loop_analysis_max_depth)
+            }
+            if ($analysisFlags.max_xrefs -and $analysisFlags.max_xrefs -gt 0) {
+                $idaScriptArgs.Add("--max-xrefs")
+                $idaScriptArgs.Add([string]$analysisFlags.max_xrefs)
+            }
             if ($analysisFlags.force_reanalyze) { $idaScriptArgs.Add("--force-reanalyze") }
             if ($analysisFlags.generate_cpp) { 
                 $idaScriptArgs.Add("--generate-cpp")
@@ -2095,6 +2123,8 @@ $analysisFlags = @{
     extract_pe_metadata    = -not $NoPeMetadata.IsPresent
     extract_advanced_pe    = -not $NoAdvancedPe.IsPresent
     extract_runtime_info   = -not $NoRuntimeInfo.IsPresent
+    loop_analysis_max_depth = $LoopAnalysisMaxDepth
+    max_xrefs              = $MaxXrefs
     force_reanalyze        = $ForceReanalyze.IsPresent
     generate_cpp           = -not $NoGenerateCpp.IsPresent
     generate_asm           = $GenerateAsm.IsPresent

ida-plugin.json

@@ -3,7 +3,7 @@
   "plugin": {
     "name": "DeepExtract",
     "entryPoint": "main.py",
-    "version": "0.9.12",
+    "version": "0.9.13",
     "idaVersions": [
       "9.0",
       "9.1",