Add mypy type checking to disassembler modules; fix mypy errors

disassemblers/ofrak_cached_disassembly/CHANGELOG.md

@@ -3,6 +3,14 @@ All notable changes to `ofrak-cached-disassembly` will be documented in this fil
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) and adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## 1.0.0 - 2025-07-25
+## [Unreleased 0.2.0](https://github.com/redballoonsecurity/ofrak/tree/master)
+
+### Added
+- Add mypy type checking enforcement to `make test` ([#702](https://github.com/redballoonsecurity/ofrak/pull/702))
+
+### Fixed
+- Fix type annotations for mypy compliance (Optional types, config defaults) ([#702](https://github.com/redballoonsecurity/ofrak/pull/702))
+
+## 0.1.0 - 2025-07-25
 ### Added
 Initial release. Hello world!

disassemblers/ofrak_cached_disassembly/Makefile

@@ -1,11 +1,18 @@
 PYTHON=python3
 
+.PHONY: install
 install:
 	$(PYTHON) -m pip install .
 
+.PHONY: develop
 develop:
 	$(PYTHON) -m pip install -e .
 
-test:
+.PHONY: inspect
+inspect:
+	$(PYTHON) -m mypy
+
+.PHONY: test
+test: inspect
 	$(PYTHON) -m pytest --cov=ofrak_cached_disassembly --cov-report=term-missing tests
 	fun-coverage --cov-fail-under=100

disassemblers/ofrak_cached_disassembly/mypy.ini

@@ -0,0 +1,2 @@
+[mypy]
+files = src

disassemblers/ofrak_cached_disassembly/setup.py

@@ -21,7 +21,7 @@ def run(self):
 
 setuptools.setup(
     name="ofrak_cached_disassembly",
-    version="0.1.0",
+    version="0.2.0rc1",
     author="Red Balloon Security",
     author_email="ofrak@redballoonsecurity.com",
     description="OFRAK Disassembler Components for Cached Results",

disassemblers/ofrak_cached_disassembly/src/ofrak_cached_disassembly/components/cached_disassembly_unpacker.py

@@ -16,7 +16,6 @@
 from ofrak.core.decompilation import (
     DecompilationAnalysis,
     DecompilationAnalyzer,
-    DecompilationAnalysis,
     ResourceView,
 )
 
@@ -37,7 +36,7 @@
 
 @dataclass
 class CachedAnalysis(ResourceView):
-    base_address: int = None
+    base_address: Optional[int] = None
 
 
 @dataclass
@@ -79,7 +78,7 @@ async def analyze(self, resource: Resource, config: CachedAnalysisAnalyzerConfig
             resource.has_tag(Program) or resource.has_tag(Ihex)
         ) and not resource.has_attributes(ProgramAttributes):
             raise AttributeError(
-                f"The resource with ID {resource.get_id()} is not an analyzable program format and does not have ProgramAttributes set."
+                f"The resource with ID {resource.get_id().hex()} is not an analyzable program format and does not have ProgramAttributes set."
             )
         self.analysis_store.store_analysis(resource.get_id(), config.filename)
         if not config.force:
@@ -125,7 +124,7 @@ def __init__(
         super().__init__(resource_factory, data_service, resource_service, component_locator)
         self.analysis_store = analysis_store
 
-    async def unpack(self, resource: Resource, config: None):
+    async def unpack(self, resource: Resource, config: None = None):
         analysis = self.analysis_store.get_analysis(resource.get_id())
         for key, mem_region in analysis.items():
             if key.startswith("seg"):
@@ -235,7 +234,7 @@ def __init__(
         super().__init__(resource_factory, data_service, resource_service, component_locator)
         self.analysis_store = analysis_store
 
-    async def unpack(self, resource: Resource, config: None):
+    async def unpack(self, resource: Resource, config: None = None):
         program_r = await resource.get_only_ancestor(ResourceFilter.with_tags(CachedAnalysis))
         analysis = self.analysis_store.get_analysis(program_r.get_id())
         if analysis["metadata"]["backend"] == "ghidra":
@@ -269,7 +268,7 @@ def __init__(
         super().__init__(resource_factory, data_service, resource_service, component_locator)
         self.analysis_store = analysis_store
 
-    async def unpack(self, resource: Resource, config: None):
+    async def unpack(self, resource: Resource, config: None = None):
         program_r = await resource.get_only_ancestor(ResourceFilter.with_tags(CachedAnalysis))
         analysis = self.analysis_store.get_analysis(program_r.get_id())
         program_attributes = self.analysis_store.get_program_attributes(program_r.get_id())
@@ -323,7 +322,7 @@ def __init__(
         super().__init__(resource_factory, data_service, resource_service, component_locator)
         self.analysis_store = analysis_store
 
-    async def unpack(self, resource: Resource, config: None):
+    async def unpack(self, resource: Resource, config: None = None):
         program_r = await resource.get_only_ancestor(ResourceFilter.with_tags(CachedAnalysis))
         analysis = self.analysis_store.get_analysis(program_r.get_id())
 
@@ -365,7 +364,7 @@ def __init__(
         super().__init__(resource_factory, data_service, resource_service)
         self.analysis_store = analysis_store
 
-    async def analyze(self, resource: Resource, config: None) -> DecompilationAnalysis:
+    async def analyze(self, resource: Resource, config: None = None) -> DecompilationAnalysis:
         # Run / fetch ghidra analyzer
         program_r = await resource.get_only_ancestor(ResourceFilter.with_tags(CachedAnalysis))
         analysis = self.analysis_store.get_analysis(program_r.get_id())

disassemblers/ofrak_ghidra/CHANGELOG.md

@@ -7,6 +7,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 
 ### Added
 - Add OFRAK requirements, requirement to test mapping, test specifications ([#656](https://github.com/redballoonsecurity/ofrak/pull/656))
+- Add mypy type checking enforcement to `make test` ([#702](https://github.com/redballoonsecurity/ofrak/pull/702))
 
 ### Changed
 - Minor update to OFRAK Community License, add OFRAK Pro License ([#478](https://github.com/redballoonsecurity/ofrak/pull/478))
@@ -20,6 +21,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - Handle escape (\) character in the JSON going from Java (Ghidra) to Python ([#604](https://github.com/redballoonsecurity/ofrak/pull/604))
 - Fix Ghidra and pyghidra CodeRegion unpacker to take into account the base address that ghidra sets for PIE executables.([#627](https://github.com/redballoonsecurity/ofrak/pull/627))
 - Pin java version ([#683](https://github.com/redballoonsecurity/ofrak/pull/683))
+- Fix `_do_ghidra_analyze_and_serve` crash when Ghidra exits unexpectedly by capturing stderr ([#702](https://github.com/redballoonsecurity/ofrak/pull/702))
+- Fix `GhidraDecompilationAnalyzer` silently swallowing non-JSONDecodeError exceptions and incorrect error string handling ([#702](https://github.com/redballoonsecurity/ofrak/pull/702))
+- Fix `_arch_info_to_processor_id` to raise an explicit error for unsupported ISAs instead of proceeding with `None` family ([#702](https://github.com/redballoonsecurity/ofrak/pull/702))
 
 ## 0.1.1 - 2024-02-15
 ### Added

disassemblers/ofrak_ghidra/Makefile

@@ -1,11 +1,18 @@
 PYTHON=python3
 
+.PHONY: install
 install:
 	$(PYTHON) -m pip install .
 
+.PHONY: develop
 develop:
 	$(PYTHON) -m pip install -e . --config-settings editable_mode=compat
 
-test:
+.PHONY: inspect
+inspect:
+	$(PYTHON) -m mypy
+
+.PHONY: test
+test: inspect
 	$(PYTHON) -m pytest --cov=ofrak_ghidra --cov-report=term-missing tests
 	fun-coverage --cov-fail-under=100

disassemblers/ofrak_ghidra/mypy.ini

@@ -0,0 +1,5 @@
+[mypy]
+files = src
+
+[mypy-yaml.*]
+ignore_missing_imports = True

disassemblers/ofrak_ghidra/setup.py

@@ -21,7 +21,7 @@ def run(self):
 
 setuptools.setup(
     name="ofrak_ghidra",
-    version="0.2.0rc4",
+    version="0.2.0rc5",
     author="Red Balloon Security",
     author_email="ofrak@redballoonsecurity.com",
     description="OFRAK Ghidra Components",

disassemblers/ofrak_ghidra/src/ofrak_ghidra/components/blocks/bb_unpacker.py

@@ -2,7 +2,8 @@
 import os
 import re
 from collections import defaultdict
-from typing import Tuple, Dict, Union, List, Iterable
+from typing import Tuple, List, Iterable
+from typing_extensions import NotRequired, TypedDict
 
 from ofrak.core.architecture import ProgramAttributes
 from ofrak_type.architecture import InstructionSet, InstructionSetMode
@@ -18,10 +19,24 @@
 )
 from ofrak_ghidra.constants import CORE_OFRAK_GHIDRA_SCRIPTS
 from ofrak_ghidra.ghidra_model import OfrakGhidraMixin, OfrakGhidraScript
-from ofrak_io.batch_manager import make_batch_manager
+from ofrak_io.batch_manager import make_batch_manager, BatchManagerInterface
+
+
+class InstructionInfo(TypedDict):
+    """Instruction data returned by GetInstructions.java - all fields always present."""
+
+    instr_offset: int
+    instr_size: int
+    mnem: str
+    operands: str
+    results: str
+    regs_read: str
+    regs_written: str
+    instr_mode: NotRequired[str]
+
 
 _GetInstructionsRequest = Tuple[Resource, int, int]
-_GetInstructionsResult = List[Dict[str, Union[str, int]]]
+_GetInstructionsResult = List[InstructionInfo]
 
 
 class GhidraBasicBlockUnpacker(
@@ -35,6 +50,8 @@ class GhidraBasicBlockUnpacker(
         os.path.join(CORE_OFRAK_GHIDRA_SCRIPTS, "GetInstructions.java"),
     )
 
+    batch_manager: BatchManagerInterface[_GetInstructionsRequest, _GetInstructionsResult]
+
     def __init__(
         self,
         resource_factory: ResourceFactory,
@@ -58,13 +75,13 @@ async def unpack(self, resource: Resource, config=None):
         program_attrs = await resource.analyze(ProgramAttributes)
 
         children_created = []
-        for instruction in instructions:
-            vaddr = instruction["instr_offset"]
-            size = instruction["instr_size"]
+        for instr_info in instructions:
+            vaddr = instr_info["instr_offset"]
+            size = instr_info["instr_size"]
             mnem, operands = _asm_fixups(
-                instruction["mnem"].lower(), instruction["operands"].lower(), program_attrs
+                instr_info["mnem"].lower(), instr_info["operands"].lower(), program_attrs
             )
-            results = instruction["results"].split(",")
+            results = instr_info["results"].split(",")
             regs_read = list()
             regs_written = list()
             # TODO A way to standardize register representations
@@ -74,14 +91,14 @@ async def unpack(self, resource: Resource, config=None):
                 regs_written.append("rsp")
                 regs_read.append("rsp")
 
-            for reg in instruction["regs_read"].lower().split(","):
+            for reg in instr_info["regs_read"].lower().split(","):
                 if reg not in regs_read and reg != "":
                     regs_read.append(reg)
-            for reg in instruction["regs_written"].lower().split(","):
+            for reg in instr_info["regs_written"].lower().split(","):
                 if reg not in regs_written and reg != "":
                     regs_written.append(reg)
 
-            mode_string = instruction.get("instr_mode", "NONE")
+            mode_string = instr_info.get("instr_mode", "NONE")
             mode = InstructionSetMode[mode_string]
             assert mode == bb_view.mode, (
                 f"The instruction mode {mode.name} returned by Ghidra does not match the basic "
@@ -111,16 +128,16 @@ async def _handle_get_instructions_batch(
             requests_by_resource[ghidra_project.resource.get_id()].append(req)
             ghidra_project_resources_by_id[ghidra_project.resource.get_id()] = resource
 
-        all_results = []
+        all_results: List[Tuple[_GetInstructionsRequest, _GetInstructionsResult]] = []
 
-        for resource_id, requests in requests_by_resource.items():
+        for resource_id, reqs in requests_by_resource.items():
             resource = ghidra_project_resources_by_id[resource_id]
-            bb_starts = ",".join(hex(bb_start) for _, bb_start, _ in requests)
-            bb_ends = ",".join(hex(bb_end) for _, _, bb_end in requests)
+            bb_starts = ",".join(hex(bb_start) for _, bb_start, _ in reqs)
+            bb_ends = ",".join(hex(bb_end) for _, _, bb_end in reqs)
 
             results = await self.get_instructions_script.call_script(resource, bb_starts, bb_ends)
 
-            all_results.extend(zip(requests, results))
+            all_results.extend(zip(reqs, results))
 
         return all_results