Further cleanups before publication

blog.py

@@ -0,0 +1,16 @@
+from striga import Semantics
+from llvm import global_context
+
+with global_context().create_module("blog") as module:
+    sem = Semantics(module, verbose=True)
+    entry = 0x140001000
+    # sem.begin(entry)
+    # sem.lift_bytes(entry, b"\x48\x89\xC8") # mov rax, rcx
+    # sem.lift_bytes(entry, b"\x48\x8B\x44\x8B\x2A") # mov rax, [rbx+rcx*4+42]
+    # sem.lift_bytes(entry, b"\x48\x8b\x43\x2a")  # mov rax, [rbx+42]
+    # sem.lift_bytes(entry, b"\x48\x31\xD8") # xor rax, rbx
+    sem.lift_bytes(entry, b"\x75\x12")  # je imm
+    # sem.lift_bytes(entry, b"\xff\xe3") # jmp rbx
+    # sem.lift_bytes(entry, b"\xe8\x00\x00\x10\x00") # call imm
+    # sem.lift_bytes(entry, b"\xc3")  # ret
+    print(module)

src/striga/semantics.py

@@ -138,7 +138,7 @@ class Semantics:
             for name in FLAGS
         }
 
-        # Set per function lifting
+        # Set per function lifted
         self.insn_blocks: dict[int, BasicBlock] = {}
         self.function: Function
         self.reg_ptrs: dict[str, Value] = {}
@@ -215,22 +215,25 @@ class Semantics:
         assert block.function == self.function
         return block
 
-    def lift_bytes(self, address: int, code: bytes):
+    def lift_bytes(self, address: int, code: bytes) -> list[Successor]:
+        # Ensure we have a function to lift into
+        if not hasattr(self, "function"):
+            self.begin(address)
+
         insn = self.cs_disasm(address, code)
         if self.verbose:
             print(";", hex(insn.address), insn.mnemonic, insn.op_str)
 
-        # Get or create - the block may already exist as a branch target.
-        # If the block is already populated, this function has already been
-        # lifted in this module; do not append a second terminator.
+        # Skip lifting if the block is already populated
         block = self.get_or_create_block(address)
-        assert block.first_instruction
+        assert block.first_instruction, "unreachable"
         if block.first_instruction.opcode == Opcode.Ret:
             block.first_instruction.erase_from_parent()
         else:
             return []
 
         with block.create_builder() as ir:
+            # State used by semantic handlers
             self.ir = ir
             self.insn = insn
             # Intentional: RIP records the current instruction, not the next PC.
@@ -251,6 +254,7 @@ class Semantics:
                 ir.br(self.get_or_create_block(fallthrough))
                 successors = [Successor(address, self.const64(fallthrough))]
 
+            # Make sure the handler produced valid IR
             self.module.verify_or_raise()
             return successors
 
@@ -304,12 +308,6 @@ class Semantics:
             widened = self.ir.shl(widened, self.const64(bit_offset))
         self.ir.store(self.ir.or_(cleared, widened), full_ptr)
 
-    def mem_write(self, addr: Value, value: Value):
-        memory = self.function.get_param(0)
-        ptr = self.ir.gep(self.types.i8, memory, [addr])
-        store = self.ir.store(value, ptr)
-        store.inst_alignment = 1
-
     def mem_read(self, addr: Value, ty: Type) -> Value:
         memory = self.function.get_param(0)
         ptr = self.ir.gep(self.types.i8, memory, [addr])
@@ -317,6 +315,12 @@ class Semantics:
         load.inst_alignment = 1
         return load
 
+    def mem_write(self, addr: Value, value: Value):
+        memory = self.function.get_param(0)
+        ptr = self.ir.gep(self.types.i8, memory, [addr])
+        store = self.ir.store(value, ptr)
+        store.inst_alignment = 1
+
     def op_mem(self, op: X86Op) -> Value:
         assert op.type == CS_OP_MEM
 
@@ -381,30 +385,6 @@ class Semantics:
             # TODO: narrow the write?
             self.mem_write(addr, value)
 
-    def push(self, value: Value):
-        byte_width = value.type.int_width // 8
-        rsp = self.reg_read("rsp")
-        rsp_sub = self.ir.sub(rsp, self.const64(byte_width))
-        self.reg_write("rsp", rsp_sub)
-        self.mem_write(rsp_sub, value)
-
-    def pop(self, ty: Type) -> Value:
-        byte_width = ty.int_width // 8
-        rsp = self.reg_read("rsp")
-        value = self.mem_read(rsp, ty)
-        rsp_add = self.ir.add(rsp, self.const64(byte_width))
-        self.reg_write("rsp", rsp_add)
-        return value
-
-    def rflags_value(self) -> Value:
-        value = self.const64(1 << 1)  # Reserved bit 1 is always set.
-        for name, bit in FLAGS.items():
-            flag = self.ir.zext(self.flag_read(name), self.i64)
-            if bit:
-                flag = self.ir.shl(flag, self.const64(bit))
-            value = self.ir.or_(value, flag)
-        return value
-
     def _bool_to_flag(self, value: Value) -> Value:
         """Convert an LLVM i1 flag predicate to the i8 state representation."""
         if value.type == self.types.i8:
@@ -436,6 +416,30 @@ class Semantics:
     def flag_write_undef_if(self, cond: Value, name: str):
         self.flag_write_if(cond, name, self.flag_undef(name))
 
+    def push(self, value: Value):
+        byte_width = value.type.int_width // 8
+        rsp = self.reg_read("rsp")
+        rsp_sub = self.ir.sub(rsp, self.const64(byte_width))
+        self.reg_write("rsp", rsp_sub)
+        self.mem_write(rsp_sub, value)
+
+    def pop(self, ty: Type) -> Value:
+        byte_width = ty.int_width // 8
+        rsp = self.reg_read("rsp")
+        value = self.mem_read(rsp, ty)
+        rsp_add = self.ir.add(rsp, self.const64(byte_width))
+        self.reg_write("rsp", rsp_add)
+        return value
+
+    def rflags_value(self) -> Value:
+        value = self.const64(1 << 1)  # Reserved bit 1 is always set.
+        for name, bit in FLAGS.items():
+            flag = self.ir.zext(self.flag_read(name), self.i64)
+            if bit:
+                flag = self.ir.shl(flag, self.const64(bit))
+            value = self.ir.or_(value, flag)
+        return value
+
     def result_is_zero(self, result: Value) -> Value:
         return self.ir.icmp(IntPredicate.EQ, result, result.type.constant(0))
 

src/striga/x86/bitwise.py

@@ -42,6 +42,11 @@ def xor(sem: Semantics):
     logical_binop(sem, Opcode.Xor)
 
 
+@semantic
+def or_(sem: Semantics):
+    logical_binop(sem, Opcode.Or)
+
+
 @semantic
 def xorps(sem: Semantics):
     dst = sem.op_read(0)
@@ -49,11 +54,6 @@ def xorps(sem: Semantics):
     sem.op_write(0, sem.ir.xor(dst, src))
 
 
-@semantic
-def or_(sem: Semantics):
-    logical_binop(sem, Opcode.Or)
-
-
 @semantic
 def test(sem: Semantics):
     lhs = sem.op_read(0)

src/striga/x86/control.py

@@ -11,46 +11,45 @@ def bool_eq(sem: Semantics, lhs: Value, rhs: Value) -> Value:
 
 
 def cc_cond(sem: Semantics, cc: str) -> Value:
-    # TODO: can we do this lazily?
-    cf = sem.flag_read("cf")
-    zf = sem.flag_read("zf")
-    sf = sem.flag_read("sf")
-    of = sem.flag_read("of")
-    pf = sem.flag_read("pf")
+    cf = lambda: sem.flag_read("cf")
+    zf = lambda: sem.flag_read("zf")
+    sf = lambda: sem.flag_read("sf")
+    of = lambda: sem.flag_read("of")
+    pf = lambda: sem.flag_read("pf")
 
     match cc:
         case "a" | "nbe":
-            return sem.ir.and_(bool_not(sem, cf), bool_not(sem, zf))
+            return sem.ir.and_(bool_not(sem, cf()), bool_not(sem, zf()))
         case "ae" | "nb" | "nc":
-            return bool_not(sem, cf)
+            return bool_not(sem, cf())
         case "b" | "nae" | "c":
-            return cf
+            return cf()
         case "be" | "na":
-            return sem.ir.or_(cf, zf)
+            return sem.ir.or_(cf(), zf())
         case "e" | "z":
-            return zf
+            return zf()
         case "g" | "nle":
-            return sem.ir.and_(bool_not(sem, zf), bool_eq(sem, sf, of))
+            return sem.ir.and_(bool_not(sem, zf()), bool_eq(sem, sf(), of()))
         case "ge" | "nl":
-            return bool_eq(sem, sf, of)
+            return bool_eq(sem, sf(), of())
         case "l" | "nge":
-            return sem.ir.xor(sf, of)
+            return sem.ir.xor(sf(), of())
         case "le" | "ng":
-            return sem.ir.or_(zf, sem.ir.xor(sf, of))
+            return sem.ir.or_(zf(), sem.ir.xor(sf(), of()))
         case "ne" | "nz":
-            return bool_not(sem, zf)
+            return bool_not(sem, zf())
         case "no":
-            return bool_not(sem, of)
+            return bool_not(sem, of())
         case "np" | "po":
-            return bool_not(sem, pf)
+            return bool_not(sem, pf())
         case "ns":
-            return bool_not(sem, sf)
+            return bool_not(sem, sf())
         case "o":
-            return of
+            return of()
         case "p" | "pe":
-            return pf
+            return pf()
         case "s":
-            return sf
+            return sf()
     raise NotImplementedError(f"condition code {cc}")
 
 
@@ -340,16 +339,6 @@ def jrcxz(sem: Semantics):
     return jcxz(sem, "rcx")
 
 
-@semantic
-def call(sem: Semantics):
-    dst = sem.op_read(0)
-    fallthrough = sem.insn.address + sem.insn.size
-    sem.push(sem.const64(fallthrough))
-    sem.ir.call(sem.call_handler, [dst])
-    sem.ir.br(sem.get_or_create_block(fallthrough))
-    return [Successor(sem.insn.address, sem.const64(fallthrough))]
-
-
 @semantic
 def jmp(sem: Semantics):
     dst = sem.op_read(0)
@@ -361,6 +350,16 @@ def jmp(sem: Semantics):
     return [Successor(sem.insn.address, dst)]
 
 
+@semantic
+def call(sem: Semantics):
+    dst = sem.op_read(0)
+    fallthrough = sem.insn.address + sem.insn.size
+    sem.push(sem.const64(fallthrough))
+    sem.ir.call(sem.call_handler, [dst])
+    sem.ir.br(sem.get_or_create_block(fallthrough))
+    return [Successor(sem.insn.address, sem.const64(fallthrough))]
+
+
 @semantic
 def ret(sem: Semantics):
     dst = sem.pop(sem.i64)