Corrected black and isort

floss/api_hooks.py

@@ -6,8 +6,8 @@
 import envi
 import viv_utils.emulator_drivers
 
-import floss.logging_
 import floss.utils as fu
+import floss.logging_
 from floss.const import MOD_NAME
 
 logger = floss.logging_.getLogger(__name__)
@@ -139,6 +139,7 @@ def _fix_return(self, emu, return_address, return_addresses):
 
 class DemoHook:
     """A demo hook to demonstrate the API of the hook classes."""
+
     def __call__(
         self,
         emu: viv_utils.emulator_drivers.EmulatorDriver,
@@ -213,9 +214,7 @@ def _allocate_mem(self, emu, size):
         size = fu.round_(size, 16)
         size = fu.get_max_size(size, MAX_MEMORY_ALLOC_SIZE)
         logger.trace("mapping 0x%x bytes at 0x%x", size, va)
-        emu.addMemoryMap(
-            va, envi.memory.MM_RWX, "[heap allocation]", b"\x00" * (size + 4)
-        )
+        emu.addMemoryMap(va, envi.memory.MM_RWX, "[heap allocation]", b"\x00" * (size + 4))
         self._heap_addr += size
         return va
 
@@ -224,9 +223,7 @@ def __call__(self, emu, api, argv):
             size = argv[0]
         elif fu.contains_funcname(api, ("VirtualAlloc", "LocalAlloc", "GlobalAlloc")):
             size = argv[1]
-        elif fu.contains_funcname(
-            api, ("VirtualAllocEx", "HeapAlloc", "RtlAllocateHeap")
-        ):
+        elif fu.contains_funcname(api, ("VirtualAllocEx", "HeapAlloc", "RtlAllocateHeap")):
             size = argv[2]
         elif fu.contains_funcname(api, ("calloc", "calloc_base")):
             # size, count
@@ -245,27 +242,22 @@ class CppNewObjectHook(MemoryAllocationHook):
       - C++ new operator
     Thanks to @BenjaminSoelberg
     """
+
     ZNWJ = "Znwj"  # operator new(unsigned int)
     ZNAJ = "Znaj"  # operator new[](unsigned int)
     YAPAXI_Z_32 = "??2@YAPAXI@Z"  # void * __cdecl operator new(unsigned int)
-    YAPEAX_K_Z_64 = (
-        "??2@YAPEAX_K@Z"  # void * __ptr64 __cdecl operator new(unsigned __int64)
-    )
+    YAPEAX_K_Z_64 = "??2@YAPEAX_K@Z"  # void * __ptr64 __cdecl operator new(unsigned __int64)
     DEFAULT_SIZE = 0x1000
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
 
     def __call__(self, emu, api, argv):
-        if fu.contains_funcname(
-            api, (self.ZNWJ, self.ZNWJ, self.YAPAXI_Z_32, self.YAPEAX_K_Z_64)
-        ):
+        if fu.contains_funcname(api, (self.ZNWJ, self.ZNWJ, self.YAPAXI_Z_32, self.YAPEAX_K_Z_64)):
             if argv and len(argv) > 0:
                 size = argv[0]
             else:
-                size = (
-                    self.DEFAULT_SIZE
-                )  # will allocate a default block size if vivisect failed to extract argv
+                size = self.DEFAULT_SIZE  # will allocate a default block size if vivisect failed to extract argv
 
             va = self._allocate_mem(emu, size)
             fu.call_return(emu, api, argv, va)
@@ -274,17 +266,17 @@ def __call__(self, emu, api, argv):
 
 class MemoryFreeHook:
     """Hook calls to memory free functions: free memory and return success."""
+
     def __call__(self, emu, api, argv):
-        if fu.contains_funcname(
-            api, ("free", "free_base", "VirtualFree", "HeapFree", "RtlFreeHeap")
-        ):
+        if fu.contains_funcname(api, ("free", "free_base", "VirtualFree", "HeapFree", "RtlFreeHeap")):
             # If the function succeeds, the return value is nonzero.
             fu.call_return(emu, api, argv, 1)
             return True
 
 
 class MemcpyHook:
     """Hook calls to memory copy functions: copy memory from source to destination."""
+
     def __call__(self, emu, api, argv):
         if fu.contains_funcname(api, ("memcpy", "memmove")):
             dst, src, count = argv
@@ -323,6 +315,7 @@ def __call__(self, emu, api, argv):
 
 class StrncmpHook:
     """Hook calls to string compare functions: compare two strings."""
+
     def __call__(self, emu, api, argv):
         if fu.contains_funcname(api, ("strncmp",)):
             s1va, s2va, num = argv
@@ -349,6 +342,7 @@ def cmp(a, b):
 
 class MemchrHook:
     """Hook calls to memchr: search for a character in a memory block."""
+
     def __call__(self, emu, api, argv):
         if fu.contains_funcname(api, ("memchr",)):
             ptr, value, num = argv
@@ -365,6 +359,7 @@ def __call__(self, emu, api, argv):
 
 class MemsetHook:
     """Hook calls to memset: fill memory with a constant byte."""
+
     def __call__(self, emu, api, argv):
         if fu.contains_funcname(api, ("memset",)):
             ptr, value, num = argv
@@ -377,6 +372,7 @@ def __call__(self, emu, api, argv):
 
 class PrintfHook:
     """Hook calls to printf: write formatted data to stdout."""
+
     # TODO disabled for now as incomplete (need to implement string format) and could result in FP strings as is
     def __call__(self, emu, api, argv):
         # TODO vfprintf, vfwprintf, vfprintf_s, vfwprintf_s, vsnprintf, vsnwprintf, etc.
@@ -390,6 +386,7 @@ def __call__(self, emu, api, argv):
 
 class ExitExceptionHook:
     """Hook calls to exit and raise exception."""
+
     def __call__(self, emu, api, argv):
         if fu.contains_funcname(api, ("ExitProcess", "RaiseException")):
             raise viv_utils.emulator_drivers.StopEmulation()
@@ -401,6 +398,7 @@ def __call__(self, emu, api, argv):
 
 class SehPrologEpilogHook:
     """Hook calls to SEH prolog and epilog functions and return success."""
+
     def __call__(self, emu, api, argv):
         if fu.contains_funcname(
             api,
@@ -419,17 +417,17 @@ def __call__(self, emu, api, argv):
 
 class SecurityCheckCookieHook:
     """Hook calls to __security_check_cookie and return success."""
+
     def __call__(self, emu, api, argv):
-        if fu.contains_funcname(
-            api, ("__security_check_cookie", "@__security_check_cookie@4")
-        ):
+        if fu.contains_funcname(api, ("__security_check_cookie", "@__security_check_cookie@4")):
             # nop
             fu.call_return(emu, api, argv, 0)
             return True
 
 
 class GetLastErrorHook:
     """Hook calls to GetLastError and return success."""
+
     def __call__(self, emu, api, argv):
         if fu.contains_funcname(api, ("GetLastError",)):
             # always assuming success
@@ -440,6 +438,7 @@ def __call__(self, emu, api, argv):
 
 class GetCurrentProcessHook:
     """Hook calls to GetCurrentProcess and return a fake process handle."""
+
     def __call__(self, emu, api, argv):
         if fu.contains_funcname(api, ("GetCurrentProcess",)):
             fu.call_return(emu, api, argv, CURRENT_PROCESS_ID)
@@ -448,6 +447,7 @@ def __call__(self, emu, api, argv):
 
 class CriticalSectionHook:
     """Hook calls to InitializeCriticalSection and return a fake critical section handle."""
+
     def __call__(self, emu, api, argv):
         if fu.contains_funcname(api, ("InitializeCriticalSection",)):
             (hsection,) = argv

floss/decoding_manager.py

@@ -1,15 +1,15 @@
 # Copyright (C) 2017 Mandiant, Inc. All Rights Reserved.
 
-from dataclasses import dataclass
 from typing import List, Tuple
+from dataclasses import dataclass
 
-import envi.memory
 import viv_utils
+import envi.memory
 import viv_utils.emulator_drivers
 from envi import Emulator
 
-import floss.logging_
 import floss.utils
+import floss.logging_
 
 from . import api_hooks
 from .const import DS_MAX_ADDRESS_REVISITS_EMULATION
@@ -72,6 +72,7 @@ class Snapshot:
         sp: the stack counter
         pc: the instruction pointer
     """
+
     memory: Memory
     sp: int
     pc: int
@@ -94,6 +95,7 @@ def get_map_size(emu):
 
 class MapsTooLargeError(Exception):
     """Exception raised when the emulator has mapped too much memory."""
+
     pass
 
 
@@ -114,16 +116,18 @@ def make_snapshot(emu: Emulator) -> Snapshot:
 
 @dataclass
 class Delta:
-    """a pair of snapshots from before and after an operation. 
+    """a pair of snapshots from before and after an operation.
 
-    Facilitates diffing the state of an emulator. 
+    Facilitates diffing the state of an emulator.
     """
+
     pre: Snapshot
     post: Snapshot
 
 
 class DeltaCollectorHook(viv_utils.emulator_drivers.Hook):
     """hook that collects Deltas at each imported API call."""
+
     def __init__(self, pre_snap: Snapshot):
         super().__init__()
         self._pre_snap = pre_snap
@@ -224,9 +228,7 @@ def emulate_function(
         tos_val = floss.utils.get_stack_value(emu, 0)
         logger.debug("%s: top of stack (return address): 0x%x", e, tos_val)
     except envi.exc.DivideByZero as e:
-        logger.debug(
-            "vivisect encountered an emulation error. will continue processing. %s", e
-        )
+        logger.debug("vivisect encountered an emulation error. will continue processing. %s", e)
     except viv_utils.emulator_drivers.StopEmulation:
         pass
     except Exception:
@@ -242,9 +244,7 @@ def emulate_function(
     try:
         deltas.append(Delta(pre_snap, make_snapshot(emu)))
     except MapsTooLargeError:
-        logger.debug(
-            "failed to create final snapshot, emulator mapped too much memory, skipping"
-        )
+        logger.debug("failed to create final snapshot, emulator mapped too much memory, skipping")
         pass
 
     return deltas

floss/features/extract.py

@@ -1,20 +1,29 @@
 # Copyright (C) 2021 Mandiant, Inc. All Rights Reserved.
-from typing import Any, Callable, Iterator, Tuple
+from typing import Any, Tuple, Callable, Iterator
 
 import envi
 import networkx
-import viv_utils
 import vivisect
-from envi.archs.i386.opconst import (INS_CALL, INS_MOV, INS_ROL, INS_ROR,
-                                     INS_SHL, INS_SHR, INS_XOR)
+import viv_utils
 from networkx import strongly_connected_components
 from viv_utils import BasicBlock
+from envi.archs.i386.opconst import INS_MOV, INS_ROL, INS_ROR, INS_SHL, INS_SHR, INS_XOR, INS_CALL
 
 import floss.logging_
 from floss.const import TS_TIGHT_FUNCTION_MAX_BLOCKS
-from floss.features.features import (BlockCount, CallsTo, KindaTightLoop, Loop,
-                                     Mov, Nzxor, NzxorLoop, NzxorTightLoop,
-                                     Shift, TightFunction, TightLoop)
+from floss.features.features import (
+    Mov,
+    Loop,
+    Nzxor,
+    Shift,
+    CallsTo,
+    NzxorLoop,
+    TightLoop,
+    BlockCount,
+    TightFunction,
+    KindaTightLoop,
+    NzxorTightLoop,
+)
 
 # security cookie checks may perform non-zeroing XORs, these are expected within a certain
 # byte range within the first and returning basic blocks, this helps to reduce FP features
@@ -77,9 +86,7 @@ def is_security_cookie(f, bb, insn) -> bool:
         return True
 
     # ... or within last bytes (instructions) before a return
-    elif bb.instructions[-1].isReturn() and insn.va > (
-        bb.va + bb.size - SECURITY_COOKIE_BYTES_DELTA
-    ):
+    elif bb.instructions[-1].isReturn() and insn.va > (bb.va + bb.size - SECURITY_COOKIE_BYTES_DELTA):
         return True
 
     return False
@@ -149,9 +156,7 @@ def extract_function_calls_to(f):
     Returns:
         An iterator over CallsTo features, each representing a call made from the given function.
     """
-    yield CallsTo(
-        f.vw, [x[0] for x in f.vw.getXrefsTo(f.va, rtype=vivisect.const.REF_CODE)]
-    )
+    yield CallsTo(f.vw, [x[0] for x in f.vw.getXrefsTo(f.va, rtype=vivisect.const.REF_CODE)])
 
 
 def extract_function_kinda_tight_loop(f):
@@ -342,9 +347,7 @@ def abstract_nzxor_loop(features):
     Returns:
         An iterator over NzxorLoop features for each identified pattern.
     """
-    if any(isinstance(f, Nzxor) for f in features) and any(
-        isinstance(f, Loop) for f in features
-    ):
+    if any(isinstance(f, Nzxor) for f in features) and any(isinstance(f, Loop) for f in features):
         yield NzxorLoop()