Update mypy and run it in strict mode

.github/workflows/run_mypy.yaml

@@ -8,11 +8,14 @@ jobs:
     runs-on: ubuntu-22.04
 
     steps:
-    - uses: actions/checkout@v1
+    - uses: actions/checkout@v7
     - name: Install mypy
-      run: sudo apt-get install -y mypy python3-mypy
+      run: |
+        sudo apt-get install -y python3
+        python3 -m venv .venv
+        .venv/bin/pip install mypy==2.4.0
     - name: Create default local config
       run: cp chb/util/ConfigLocal.template chb/util/ConfigLocal.py
     - name: Run mypy
       run: |
-        PYTHONPATH=$PWD mypy chb
+        PYTHONPATH=$PWD .venv/bin/mypy --strict chb

chb/api/BTerm.py

@@ -59,7 +59,7 @@
 from chb.util.IndexedTable import IndexedTableValue
 
 if TYPE_CHECKING:
-    from chb.api.InterfaceDictionary import FtsParameter
+    from chb.api.FtsParameter import FtsParameter
     from chb.api.InterfaceDictionary import InterfaceDictionary
 
 

chb/api/FormatStringSpec.py

@@ -124,7 +124,7 @@ def arg_type(self) -> str:
     def flags(self) -> str:
         return "".join(chr(i) for i in self.args)
 
-    def __str__(self):
+    def __str__(self) -> str:
         return (
             "%"
             + self.flags

chb/app/AppAccess.py

@@ -30,21 +30,17 @@
 
 from abc import ABC, abstractmethod
 from typing import (
-    Any,
     Callable,
     cast,
     Dict,
     List,
     Mapping,
     Optional,
     Sequence,
-    Generic,
     Tuple,
     Type,
     TYPE_CHECKING,
-    TypeVar,
-    Union,
-    overload)
+)
 
 from chb.api.CallTarget import CallTarget, IndirectTarget, CallbackTableTarget
 from chb.api.InterfaceDictionary import InterfaceDictionary
@@ -91,25 +87,23 @@
 if TYPE_CHECKING:
     from chb.app.Instruction import Instruction
 
-HeaderTy = TypeVar('HeaderTy', PEHeader, ELFHeader, Union[PEHeader, ELFHeader])
 
-
-class AppAccess(ABC, Generic[HeaderTy]):
+class AppAccess(ABC):
     def __init__(
             self,
             path: str,
             filename: str,
-            fileformat: Type[HeaderTy],
+            fileformat: Type[UF.FileHeader],
             deps: List[str] = []) -> None:
         """Initializes access to analysis results."""
         self._path = path
         self._filename = filename
         self._deps = deps  # list of summary zips registered as dependencies
-        self._header_ty: Type[HeaderTy] = fileformat  # currently supported: elf, pe
+        self._header_ty: Type[UF.FileHeader] = fileformat  # currently supported: elf, pe
 
         self._userdata: Optional[UserData] = None
 
-        self._header: Optional[HeaderTy] = None
+        self._header: Optional[UF.FileHeader] = None
 
         # functions
         self._appresultdata: Optional[AppResultData] = None
@@ -227,13 +221,30 @@ def bcfiles(self) -> BCFiles:
 
     # File format --------------------------------------------------------------
     @property
-    def header(self) -> HeaderTy:
+    def header(self) -> UF.FileHeader:
         if self._header is None:
             x = self._header_ty.get_xnode(self.path, self.filename)
             self._header = self._header_ty(
                 self.path, self.filename, x, self.dependencies)
         return self._header
 
+    # Format specific. Will raise exceptions if the wrong format is requested
+    @property
+    def elf_header(self) -> ELFHeader:
+        if not self.elf:
+            raise RuntimeError("Requested ELF header but binary is not ELF: %s" % self._header_ty)
+        header = self.header
+        assert isinstance(header, ELFHeader)
+        return header
+
+    @property
+    def pe_header(self) -> PEHeader:
+        if not self.pe:
+            raise RuntimeError("Requested PE header but binary is not PE: %s" % self._header_ty)
+        header = self.header
+        assert isinstance(header, PEHeader)
+        return header
+
     # Systeminfo ---------------------------------------------------------------
 
     @property

chb/app/Cfg.py

@@ -188,7 +188,7 @@ def inverse_with_phantom_exit_node(self) -> 'FlowGraph':
         augedges[phantomend] = terminators
         return FlowGraph(list(self.nodes) + [phantomend], augedges, phantomend)
 
-    def ipostdoms(rrg: 'FlowGraph') -> Dict[UserNodeID, UserNodeID]:
+    def ipostdoms(self, rrg: 'FlowGraph') -> Dict[UserNodeID, UserNodeID]:
         idoms = rrg.idoms.copy()
         # The start node of the reverse graph is a phantom node that doesn't
         # exist in the original graph.
@@ -211,7 +211,7 @@ def intersect(b1: UserNodeID, b2: UserNodeID) -> UserNodeID:
             def idom(n: UserNodeID) -> UserNodeID:
                 i = idoms[n]
                 assert i is not None # Should always hit start node.
-                return cast(UserNodeID, i)
+                return i
 
             finger1: UserNodeID = b1
             finger2: UserNodeID = b2
@@ -582,7 +582,7 @@ def node_within(x: str, merges: List[str], ctx: ControlFlowContext) -> List[AST.
                         for succ in succs:
                             if jumptable.has_target(succ) and succ not in succlabels:
                                 cvs = jumptable.get_target(succ)
-                                def rawlabel(c):
+                                def rawlabel(c: int) -> AST.ASTCaseLabel:
                                     r = astree.mk_case_label(astree.mk_integer_constant(c))
                                     astree.add_instruction_span(r.locationid, succ, "")
                                     return r
@@ -598,7 +598,7 @@ def switch_case_ctx(mb_nextsucc: Optional[str]) -> ControlFlowContext:
                             return ctx.in_switch(mergeaddr, mb_nextsucc)
                         return ctx
 
-                    def switch_case_stmts(succ, mb_nextsucc):
+                    def switch_case_stmts(succ: str, mb_nextsucc: Optional[str]) -> List[AST.ASTStmt]:
                         sctx = switch_case_ctx(mb_nextsucc)
                         casebody = do_branch(x, succ, sctx)
                         if casebody == [] and mb_nextsucc is None:
@@ -607,7 +607,12 @@ def switch_case_stmts(succ, mb_nextsucc):
                             # we insert a synthetic break statement to maintain
                             # syntactic compatibility.
                             casebody = [astree.mk_break_stmt()]
-                        return succlabels.get(succ, []) + casebody
+                        # typing note: mypy complains that we're mixing up ASTStmts (casebody) with
+                        # ASTStmtLabels (from succlabels). With the current implementation,
+                        # making ASTStmtLabels into ASTStmts would cause other issues.
+                        # This function is also not well understood so we're wary of mucking with
+                        # it too much. type: ignore is the best approach for the time being.
+                        return succlabels.get(succ, []) + casebody  # type: ignore[return-value]
 
                     bodystmts: List[AST.ASTStmt] = []
                     for idx in range(1, len(succs)):
@@ -627,8 +632,7 @@ def switch_case_stmts(succ, mb_nextsucc):
                       bodystmts.append(defaultcase)
 
                     switchbody = astree.mk_block(bodystmts)
-                    switchstmt = cast(AST.ASTSwitchStmt,
-                        astree.mk_switch_stmt(switchcondition, switchbody, mergeaddr))
+                    switchstmt = astree.mk_switch_stmt(switchcondition, switchbody, mergeaddr)
                     return (xstmts + [switchstmt])
                     # We don't use with_last_insn_span(x, switchstmt) because when we have
                     # a switch lifted from a conditional jump (for explicit cases) plus an

chb/arm/ARMAccess.py

@@ -29,24 +29,22 @@
 from typing import Callable, Dict, List, Mapping, Optional, Type
 
 from chb.app.Callgraph import Callgraph, mk_tgt_callgraph_node, mk_app_callgraph_node
-from chb.app.AppAccess import AppAccess, HeaderTy
+from chb.app.AppAccess import AppAccess
 
 from chb.arm.ARMDictionary import ARMDictionary
 from chb.arm.ARMFunction import ARMFunction
 from chb.arm.ARMInstruction import ARMInstruction
 
-from chb.elfformat.ELFHeader import ELFHeader
-
 import chb.util.fileutil as UF
 
 
-class ARMAccess(AppAccess[HeaderTy]):
+class ARMAccess(AppAccess):
 
     def __init__(
             self,
             path: str,
             filename: str,
-            fileformat: Type[HeaderTy],
+            fileformat: Type[UF.FileHeader],
             deps: List[str] = []) -> None:
         AppAccess.__init__(self, path, filename, fileformat, deps)
         self._armd: Optional[ARMDictionary] = None

chb/arm/ARMFunction.py

@@ -209,12 +209,12 @@ def f(baddr: str, block: ARMBlock) -> None:
 
     def iter_blocks(self, f: Callable[[str, ARMBlock], None]) -> None:
         for (ba, block) in self.blocks.items():
-            armblock = cast(ARMBlock, block)
+            armblock = block
             f(ba, armblock)
 
     def iter_instructions(self, f: Callable[[str, ARMInstruction], None]) -> None:
         for (ia, instr) in self.instructions.items():
-            arminstr = cast(ARMInstruction, instr)
+            arminstr = instr
             f(ia, arminstr)
 
     @property

chb/arm/ARMOperandKind.py

@@ -612,7 +612,11 @@ def scale_factor(self) -> Optional[int]:
                 return 1
             else:
                 if srt.is_shift_left:
-                    return 2 ** shiftamount
+                    # For some unknown reason, mypy (2.4.0) complains that we're returning
+                    # `Any` here instead of [int | None]. This only happens when we use the
+                    # `**` operator, so work around it by using `<<` Original code was:
+                    #   return 2 ** shiftamount
+                    return 1 << shiftamount
                 else:
                     return None
         else:

chb/arm/opcodes/ARMLoadMultipleIncrementAfter.py

@@ -116,7 +116,7 @@ def memrhss(self) -> List["XXpr"]:
     def are_memrhss_ok(self) -> bool:
         return all(self.is_xpr_ok(i) for i in self.memrhs_range)
 
-    def memrhs(self, index) -> "XXpr":
+    def memrhs(self, index: int) -> "XXpr":
         """returns the rhs expression at (1-based) position index"""
 
         if self.has_index(index):
@@ -135,7 +135,7 @@ def rmemrhss(self) -> List["XXpr"]:
     def are_rmemrhss_ok(self) -> bool:
         return all(self.is_xpr_ok(i) for i in self.rmemrhs_range)
 
-    def rmemrhs(self, index) -> "XXpr":
+    def rmemrhs(self, index: int) -> "XXpr":
         """returns the rewritten rhs expr at (1-based) position index"""
 
         if self.has_index(index):
@@ -154,7 +154,7 @@ def xaddrs(self) -> List["XXpr"]:
     def are_xaddrs_ok(self) -> bool:
         return all(self.is_xpr_ok(i) for i in self.xaddr_range)
 
-    def xaddr(self, index) -> "XXpr":
+    def xaddr(self, index: int) -> "XXpr":
         """returns the rhs addr at (1-based) position index"""
 
         if self.has_index(index):
@@ -173,7 +173,7 @@ def cmemrhss(self) -> List["XXpr"]:
     def are_cmemrhss_ok(self) -> bool:
         return all(self.is_cxpr_ok(i) for i in range(0, self.regcount))
 
-    def cmemrhs(self, index) -> "XXpr":
+    def cmemrhs(self, index: int) -> "XXpr":
         """returns the rhs addr at (1-based) position index"""
 
         if self.has_index(index):

chb/arm/opcodes/ARMLoadMultipleIncrementBefore.py

@@ -116,7 +116,7 @@ def memrhss(self) -> List["XXpr"]:
     def are_memrhss_ok(self) -> bool:
         return all(self.is_xpr_ok(i) for i in self.memrhs_range)
 
-    def memrhs(self, index) -> "XXpr":
+    def memrhs(self, index: int) -> "XXpr":
         """returns the rhs expression at (1-based) position index"""
 
         if self.has_index(index):
@@ -135,7 +135,7 @@ def rmemrhss(self) -> List["XXpr"]:
     def are_rmemrhss_ok(self) -> bool:
         return all(self.is_xpr_ok(i) for i in self.rmemrhs_range)
 
-    def rmemrhs(self, index) -> "XXpr":
+    def rmemrhs(self, index: int) -> "XXpr":
         """returns the rewritten rhs expr at (1-based) position index"""
 
         if self.has_index(index):
@@ -154,7 +154,7 @@ def xaddrs(self) -> List["XXpr"]:
     def are_xaddrs_ok(self) -> bool:
         return all(self.is_xpr_ok(i) for i in self.xaddr_range)
 
-    def xaddr(self, index) -> "XXpr":
+    def xaddr(self, index: int) -> "XXpr":
         """returns the rhs addr at (1-based) position index"""
 
         if self.has_index(index):
@@ -173,7 +173,7 @@ def cmemrhss(self) -> List["XXpr"]:
     def are_cmemrhss_ok(self) -> bool:
         return all(self.is_cxpr_ok(i) for i in range(0, self.regcount))
 
-    def cmemrhs(self, index) -> "XXpr":
+    def cmemrhs(self, index: int) -> "XXpr":
         """returns the rhs addr at (1-based) position index"""
 
         if self.has_index(index):

chb/arm/opcodes/ARMVectorMoveDS.py

@@ -154,14 +154,14 @@ def _unpack_imm32(self, x: XXpr) -> float:
         # https://stackoverflow.com/questions/33483846/how-to-convert-32-bit-binary-to-float
 
         ci = cast(XprConstant, x).intvalue
-        return struct.unpack('f', struct.pack('I', ci))[0]
+        return cast(float, struct.unpack('f', struct.pack('I', ci))[0])
 
     def _unpack_imm64(self, x: XXpr) -> float:
         # from StackOverflow:
         # https://stackoverflow.com/questions/8751653/how-to-convert-a-binary-string-into-a-float-value
         ci = cast(XprConstant, x).intvalue
         b8 = struct.pack('Q', ci)
-        return struct.unpack('d', b8)[0]
+        return cast(float, struct.unpack('d', b8)[0])
 
     def annotation(self, xdata: InstrXData) -> str:
         xd = ARMVectorMoveDSXData(xdata)

chb/arm/opcodes/ARMVectorMultiplySubtract.py

@@ -154,7 +154,7 @@ def _unpack_imm32(self, x: XXpr) -> float:
         # https://stackoverflow.com/questions/33483846/how-to-convert-32-bit-binary-to-float
 
         ci = cast(XprConstant, x).intvalue
-        return struct.unpack('f', struct.pack('I', ci))[0]
+        return cast(float, struct.unpack('f', struct.pack('I', ci))[0])
 
     def ast_prov(
             self,

chb/ast/ASTDeserializer.py

@@ -26,7 +26,7 @@
 # ------------------------------------------------------------------------------
 """Simplified version of AST to check serialization."""
 
-from typing import Any, cast, Dict, List, NewType, Optional, Tuple, Union
+from typing import Any, cast, Dict, List, Optional, Tuple
 
 from chb.ast.AbstractSyntaxTree import (
     AbstractSyntaxTree, nooffset, voidtype, ASTSpanRecord)
@@ -362,15 +362,15 @@ def find_instr_by_id(
         def find_instr_address_by_locationid(locationid: int) -> str:
             for node in fdata["spans"]:
                 if "locationid" in node and int(node["locationid"]) == locationid:
-                    return node["spans"][0]["base_va"]
+                    return cast(str, node["spans"][0]["base_va"])
             else:
                 raise Exception(
                     "No span found for locationid: " + str(locationid))
 
         def find_expr_address_by_exprid(exprid: int) -> str:
             for node in fdata["spans"]:
                 if "exprid" in node and int(node["exprid"]) == exprid:
-                    return node["spans"][0]["base_va"]
+                    return cast(str, node["spans"][0]["base_va"])
             else:
                 raise Exception(
                     "Deserializer: No span found for exprid: " + str(exprid))

chb/ast/ASTNode.py

@@ -2904,7 +2904,6 @@ def field_at_offset(self, offset: int) -> Tuple["ASTFieldInfo", int]:
                 prev = (i, fname)
         else:
             if prev is not None:
-                prev = cast(Tuple[int, str], prev)
                 return (self.fieldinfo(prev[1]), offset - prev[0])
             else:
                 raise Exception(