fix(core): address CodeRabbit runtime-safety findings

Harden the CodeRabbit-reported runtime-safety and API-correctness issues across the disassembly facade, Intel/CIL backends, label providers, loaders, report DTOs, and shared utilities.

Grouped fixes:

- guard optional timeout callbacks and malformed parser/provider inputs

- correct byte/string comparisons, alignment ordering, and provider call contracts

- make report DTO construction, statistics addition, and mutable defaults safer

- replace fragile/private APIs, direct prints, deprecated logger calls, and assert-only validation

- convert malformed Rust demangler inputs to demangler-specific errors

- preserve explicit backend selection during DEX autodetect

- move regression coverage into subsystem test suites instead of a CodeRabbit-specific test file

Validation:

- python -m pytest tests/testDalvikDisassembler.py tests/testEscaper.py tests/testRustSymbolProvider.py tests/testFileFormatParsers.py tests/testIntelDisassembler.py tests/testIndirectCallAnalyzer.py -q

- make test

- make lint

- python -m ruff check src/smda tests/testDalvikDisassembler.py

- git diff --check

Author: r0ny123

src/smda/Disassembler.py

@@ -26,6 +26,7 @@ def __init__(self, config=None, backend=None):
             config = SmdaConfig()
         self.config = config
         self.disassembler = None
+        self._explicit_backend = bool(backend)
         if backend == "intel":
             self.disassembler = IntelDisassembler(self.config)
         elif backend == "cil":
@@ -136,8 +137,8 @@ def disassembleFile(self, file_path, pdb_path=""):
                 self.disassembler.addPdbFile(binary_info, pdb_path)
             smda_report = self._disassemble(binary_info, timeout=self.config.TIMEOUT)
             if self.config.WITH_STRINGS:
-                is_go_binary = GoSymbolProvider(None).getPcLntabOffset(binary_info.binary)
-                string_mode = "go" if is_go_binary else None
+                go_pclntab_offset = GoSymbolProvider(None).getPcLntabOffset(binary_info.binary)
+                string_mode = "go" if go_pclntab_offset is not None else None
                 self._addStringsToReport(smda_report, binary_info.binary, mode=string_mode)
             if self.config.STORE_BUFFER:
                 smda_report.buffer = binary_info.binary
@@ -157,8 +158,8 @@ def disassembleUnmappedBuffer(self, file_content):
             self.initDisassembler(binary_info.architecture)
             smda_report = self._disassemble(binary_info, timeout=self.config.TIMEOUT)
             if self.config.WITH_STRINGS:
-                is_go_binary = GoSymbolProvider(None).getPcLntabOffset(binary_info.binary)
-                string_mode = "go" if is_go_binary else None
+                go_pclntab_offset = GoSymbolProvider(None).getPcLntabOffset(binary_info.binary)
+                string_mode = "go" if go_pclntab_offset is not None else None
                 self._addStringsToReport(smda_report, file_content, mode=string_mode)
             if self.config.STORE_BUFFER:
                 smda_report.buffer = file_content
@@ -186,7 +187,7 @@ def disassembleBuffer(
         # Auto-detect DEX when the caller did not explicitly override architecture.
         # disassembleUnmappedBuffer / disassembleFile already use FileLoader for detection;
         # this path bypasses it, so we check the magic bytes manually here.
-        if architecture == "intel" and DexFileLoader.isCompatible(file_content):
+        if not self._explicit_backend and architecture == "intel" and DexFileLoader.isCompatible(file_content):
             architecture = "dalvik"
             if bitness is None:
                 bitness = DexFileLoader.getBitness(file_content)
@@ -205,8 +206,8 @@ def disassembleBuffer(
         try:
             smda_report = self._disassemble(binary_info, timeout=self.config.TIMEOUT)
             if self.config.WITH_STRINGS:
-                is_go_binary = GoSymbolProvider(None).getPcLntabOffset(binary_info.binary)
-                string_mode = "go" if is_go_binary else None
+                go_pclntab_offset = GoSymbolProvider(None).getPcLntabOffset(binary_info.binary)
+                string_mode = "go" if go_pclntab_offset is not None else None
                 self._addStringsToReport(smda_report, file_content, mode=string_mode)
             if self.config.STORE_BUFFER:
                 smda_report.buffer = file_content

src/smda/DisassemblyStatistics.py

@@ -8,6 +8,17 @@ class DisassemblyStatistics:
     num_function_calls = None
     num_failed_functions = None
     num_failed_instructions = None
+    _fields = (
+        "num_functions",
+        "num_recursive_functions",
+        "num_leaf_functions",
+        "num_basic_blocks",
+        "num_instructions",
+        "num_api_calls",
+        "num_function_calls",
+        "num_failed_functions",
+        "num_failed_instructions",
+    )
 
     def __init__(self, disassembly_result=None):
         if disassembly_result is not None:
@@ -74,13 +85,14 @@ def toDict(self):
     def __add__(self, other):
         if not isinstance(other, DisassemblyStatistics):
             raise ValueError("Needs another DisassemblyStatistics to perform addition of values")
-        self.num_functions += other.num_functions
-        self.num_recursive_functions += other.num_recursive_functions
-        self.num_leaf_functions += other.num_leaf_functions
-        self.num_basic_blocks += other.num_basic_blocks
-        self.num_instructions += other.num_instructions
-        self.num_api_calls += other.num_api_calls
-        self.num_function_calls += other.num_function_calls
-        self.num_failed_functions += other.num_failed_functions
-        self.num_failed_instructions += other.num_failed_instructions
+        result = DisassemblyStatistics()
+        for field in self._fields:
+            setattr(result, field, (getattr(self, field) or 0) + (getattr(other, field) or 0))
+        return result
+
+    def __iadd__(self, other):
+        if not isinstance(other, DisassemblyStatistics):
+            raise ValueError("Needs another DisassemblyStatistics to perform addition of values")
+        for field in self._fields:
+            setattr(self, field, (getattr(self, field) or 0) + (getattr(other, field) or 0))
         return self

src/smda/cil/CilDisassembler.py

@@ -156,21 +156,6 @@ def analyzeFunction(self, pe, start_addr, method_body):
             i_size = len(i_bytes)
             i_mnemonic = str(insn.opcode)
             i_op_str = format_operand(pe, insn.operand)
-            # debug output for all instructions
-            if False:
-                from smda.cil.CilInstructionEscaper import CilInstructionEscaper
-                from smda.common.SmdaInstruction import SmdaInstruction
-
-                smda_ins = SmdaInstruction([i_address, i_bytes.hex(), i_mnemonic, i_op_str])
-                escaped = CilInstructionEscaper.escapeBinary(smda_ins)
-                print(
-                    f"{escaped:<20}"
-                    + f"{insn.offset:04X}"
-                    + "    "
-                    + f"{' '.join(f'{b:02x}' for b in insn.get_bytes()): <20}"
-                    + f"{str(insn.opcode): <15}"
-                    + format_operand(pe, insn.operand)
-                )
             # https://en.wikipedia.org/wiki/List_of_CIL_instructions
             if i_mnemonic in ["ret"]:
                 state.setNextInstructionReachable(False)
@@ -211,12 +196,13 @@ def analyzeFunction(self, pe, start_addr, method_body):
                 target = int(i_op_str, 16)
                 state.addCodeRef(i_address, target, by_jump=True)
             if i_mnemonic in ["jmp"]:
-                raise Exception(
-                    "Found unhandled CIL jmp instruction, report back its structure and have Daniel fix it."
-                )
                 target = int(i_op_str, 16)
                 state.addCodeRef(i_address, target, by_jump=True)
                 state.setNextInstructionReachable(False)
+                LOGGER.error(
+                    "Found unhandled CIL jmp instruction at 0x%x, report back its structure and have Daniel fix it.",
+                    i_address,
+                )
             if i_mnemonic in ["ldstr"]:
                 # we possibly want to extract and collect these and put them in the stringref part of SmdaFunction
                 self.disassembly.addStringRef(start_addr, i_address, i_op_str[1:-1])
@@ -276,6 +262,6 @@ def analyzeBuffer(self, binary_info, cbAnalysisTimeout=None):
             self.analyzeFunction(pe, method_body.offset, method_body)
         # package up and finish
         self.disassembly.analysis_end_ts = datetime.datetime.now(datetime.timezone.utc)
-        if cbAnalysisTimeout():
+        if cbAnalysisTimeout and cbAnalysisTimeout():
             self.disassembly.analysis_timeout = True
         return self.disassembly

src/smda/cil/CilInstructionEscaper.py

@@ -303,7 +303,6 @@ def escapeMnemonic(mnemonic):
                 mnemonic,
             )
             return "U"
-        return mnemonic
 
     @staticmethod
     def escapeField(op_field, escape_registers=True, escape_pointers=True, escape_constants=True):
@@ -451,6 +450,7 @@ def escapeBinaryJumpCall(ins, escape_intraprocedural_jumps=False):
                 "brtrue",
                 "brtrue.s",
                 "brzero",
+                "brzero.s",
                 "switch",
             ]:
                 escaped_sequence = CilInstructionEscaper.escapeToOpcodeOnly(ins)

src/smda/cil/FunctionAnalysisState.py

@@ -133,15 +133,14 @@ def getBlocks(self):
                     and i != len(self.instructions) - 1
                     and any(r != self.instructions[i + 1][0] for r in self.code_refs_from[current[0]])
                 ):
-                    break
                     # if we can reach a colliding address from here, the block is broken and should end.
                     reachable_collisions = self.code_refs_from[current[0]].intersection(self.colliding_addresses)
                     next_addr = current[0] + current[1]
                     is_next_addr = next_addr in reachable_collisions
                     if reachable_collisions and is_next_addr:
                         # we should remove the from/to code references for this collision as there should be no non CFG instruction references between instructions of different functions
                         self.removeCodeRef(current[0], next_addr)
-                        break
+                    break
                 if (
                     i != len(self.instructions) - 1
                     and self.instructions[i + 1][0] in self.code_refs_to

src/smda/common/BinaryInfo.py

@@ -17,7 +17,7 @@ class BinaryInfo:
     raw_data = b""
     binary_size = 0
     bitness = None
-    code_areas = []
+    code_areas = None
     component = ""
     family = ""
     file_path = ""
@@ -36,6 +36,7 @@ def __init__(self, binary):
         self.binary = binary
         self.raw_data = binary
         self.binary_size = len(binary)
+        self.code_areas = []
         self._lief_binary = None
         self.abi = ""
 

src/smda/common/SmdaBasicBlock.py

@@ -16,14 +16,18 @@ class SmdaBasicBlock:
     def __init__(self, instructions, smda_function=None):
         assert isinstance(instructions, list)
         self.smda_function = smda_function
+        self.instructions = instructions
+        self.offset = instructions[0].offset if instructions else None
+        self.length = len(instructions)
+        self.picblockhash = None
+        self.opcblockhash = None
         if instructions:
-            self.instructions = instructions
-            self.offset = instructions[0].offset
-            self.length = len(instructions)
             self.picblockhash = self.getPicBlockHash()
             self.opcblockhash = self.getOpcBlockHash()
 
     def getInstructions(self) -> Iterator["SmdaInstruction"]:
+        if self.instructions is None:
+            return
         yield from self.instructions
 
     def getPicBlockHash(self):
@@ -90,12 +94,11 @@ def getSuccessors(self):
 
     @classmethod
     def fromDict(cls, block_dict, smda_function=None) -> "SmdaBasicBlock":
-        smda_block = cls(None)
-        smda_block.smda_function = smda_function
-        smda_block.instructions = [SmdaInstruction.fromDict(d, smda_function=smda_function) for d in block_dict]
-        return smda_block
+        return cls([SmdaInstruction.fromDict(d, smda_function=smda_function) for d in block_dict], smda_function)
 
     def toDict(self) -> dict:
+        if self.instructions is None:
+            return []
         return [smda_ins.toDict() for smda_ins in self.instructions]
 
     def __int__(self):

src/smda/common/SmdaFunction.py

@@ -36,11 +36,13 @@ class SmdaFunction:
     function_name = ""
     pic_hash = None
     opc_hash = None
+    nesting_depth = 0
     strongly_connected_components = None
     tfidf = None
 
     def __init__(self, disassembly=None, function_offset=None, config=None, smda_report=None):
         self.smda_report = smda_report
+        self.nesting_depth = 0
         if disassembly is not None and function_offset is not None:
             self._escaper = IntelInstructionEscaper if disassembly.binary_info.architecture in ["intel"] else None
             self.offset = function_offset
@@ -72,14 +74,11 @@ def __init__(self, disassembly=None, function_offset=None, config=None, smda_rep
             # DEX strings are part of the parsed file structure, so they're always
             # populated for Dalvik regardless of WITH_STRINGS — no extra extraction
             # cost. For other architectures, honor WITH_STRINGS as usual.
-            if (
-                config
-                and config.WITH_STRINGS
-                or (
-                    disassembly.binary_info.architecture == "dalvik"
-                    and disassembly.getStringRefsForFunction(function_offset)
-                )
-            ):
+            is_dalvik_with_strings = (
+                disassembly.binary_info.architecture == "dalvik"
+                and disassembly.getStringRefsForFunction(function_offset)
+            )
+            if (config and config.WITH_STRINGS) or is_dalvik_with_strings:
                 self.stringrefs = (
                     self._normalizeDalvikStringRefs(disassembly.getStringRefsForFunction(function_offset))
                     if disassembly.binary_info.architecture == "dalvik"
@@ -145,7 +144,7 @@ def getPicHashAsLong(self):
         return self.pic_hash
 
     def getPicHashAsHex(self):
-        return struct.pack("l", self.pic_hash).hex()
+        return struct.pack("Q", self.pic_hash).hex()
 
     def getInstructions(self):
         for block in self.getBlocks():

src/smda/common/SmdaInstruction.py

@@ -75,7 +75,7 @@ def getDetailed(self):
             # WAIT instruction before the wait version and the NOP instruction
             # before the no-wait version.
             if len(with_details) > 1:
-                LOGGER.warn(
+                LOGGER.warning(
                     f"Sequence {self.bytes} disassembles to {len(with_details)} instructions but expected one - taking the last instruction only!"
                 )
                 self.detailed = with_details[-1]

src/smda/common/TailcallAnalyzer.py

@@ -1,10 +1,13 @@
 import bisect
 import json
+import logging
 from collections import defaultdict
 from operator import itemgetter
 
 from smda.common.ExceptionHandling import reraise_non_operational_exception
 
+LOGGER = logging.getLogger(__name__)
+
 
 class TailcallAnalyzer:
     def __init__(self):
@@ -87,15 +90,15 @@ def __printIntervals(self, intervals):
         # return
         if len(intervals) < 25:
             for one, two in intervals:
-                print(f"  0x{one:x} -> 0x{two:x}")
+                LOGGER.debug("  0x%x -> 0x%x", one, two)
         else:
-            print("Function has too many intervals to display")
+            LOGGER.debug("Function has too many intervals to display")
 
     def resolveTailcalls(self, disassembler, verbose=False):
         newly_created_functions = set()
         for tailcall in self.getTailcalls():
             if verbose:
-                print(f"Processing tailcall:\n{json.dumps(tailcall, indent=2, sort_keys=True)}")
+                LOGGER.debug("Processing tailcall:\n%s", json.dumps(tailcall, indent=2, sort_keys=True))
             # remove the information from the function-analysis state of the disassembly
             function = self.__getFunctionByStartAddr(tailcall["destination_function"])
             if not function or function.is_tailcall_function:
@@ -105,7 +108,7 @@ def resolveTailcalls(self, disassembler, verbose=False):
             self.__functions.remove(function)
             if function:
                 if verbose:
-                    print("Old function:")
+                    LOGGER.debug("Old function:")
                     self.__printIntervals(self.__getFunctionIntervals(function))
                 function.revertAnalysis()
 
@@ -123,7 +126,7 @@ def resolveTailcalls(self, disassembler, verbose=False):
                     reraise_non_operational_exception(exc)
                     # print ("0x{:x} -> 0x{:x}".format(tailcall["destination_function"], tailcall["destination_addr"]))
             elif verbose:
-                print(
+                LOGGER.debug(
                     "**** 0x{:x} IS NOW PART OF 0x{:x}".format(
                         tailcall["destination_function"], tailcall["destination_addr"]
                     )
@@ -132,10 +135,10 @@ def resolveTailcalls(self, disassembler, verbose=False):
             if verbose:
                 function = self.__getFunctionByStartAddr(tailcall["destination_function"])
                 new_function = self.__getFunctionByStartAddr(tailcall["destination_addr"])
-                print("New function:")
+                LOGGER.debug("New function:")
                 if new_function:
                     self.__printIntervals(self.__getFunctionIntervals(new_function))
-                print("Re-disassembled old function:")
+                LOGGER.debug("Re-disassembled old function:")
                 if function:
                     self.__printIntervals(self.__getFunctionIntervals(function))
         return sorted(newly_created_functions)