Format files to pass tests

Author: harelon

setup.py

@@ -24,9 +24,9 @@
                     "Operating System :: OS Independent",
                   ],
       entry_points={
-        'console_scripts': [
-          'karta_analyze_src = karta.karta_analyze_src:main'
-        ]
+            'console_scripts': [
+                  'karta_analyze_src = karta.karta_analyze_src:main'
+            ]
       },
       zip_safe=False
-)
+      )

src/karta/config/utils.py

@@ -10,7 +10,7 @@
 DISASSEMBLER_PATH = None
 CONFIG_DIR = os.path.dirname(os.path.realpath(__file__))
 DEFAULT_DISASSEMBLER = os.path.join(CONFIG_DIR, "default_disassembler_path")
-SCRIPT_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)),"..","analyze_src_file.py")
+SCRIPT_PATH = os.path.join(os.path.dirname(os.path.realpath(__file__)), "..", "analyze_src_file.py")
 
 LIBRARY_NAME            = "Karta"
 STATE_FILE_SUFFIX       = "_file_state.json"
@@ -459,18 +459,42 @@ def isMatching():
     return matching_mode
 
 def addDisassembler(name, path):
+    """Add an installation of a dissasembler.
+
+    Args:
+        name (str): name of the installation file
+        path (str): directory of the disassembler installtion
+    """
     with open(os.path.join(CONFIG_DIR, name), "w") as f:
         f.write(path)
 
 def disassemblerInstallationExists(name):
+    """Check whether there is an existing installation with the filename.
+
+    Args:
+        name (str): filename of the installtion to check
+
+    Return Value:
+        return true if such installtion exists
+    """
     return os.path.exists(os.path.join(CONFIG_DIR, name))
 
 def getDisassembler(name):
+    """Get directory of disassembler from configuration by it's configuration file name.
+
+    Args:
+        name (str): name of the disassembler to search for
+    """
     if disassemblerInstallationExists(name):
         with open(os.path.join(CONFIG_DIR, name), "r") as f:
             return f.read()
 
 def setDefaultDisassembler(name):
+    """Set the default disassembler in the configuration file.
+
+    Args:
+        name (str): name of the file that contains the default disassembler to use.
+    """
     with open(os.path.join(CONFIG_DIR, DEFAULT_DISASSEMBLER), "w") as f:
         if os.path.isfile(os.path.join(CONFIG_DIR, name)):
             f.write(name)
@@ -488,7 +512,7 @@ def getDisasPath(prompter):
         actual_disas_path = None
         with open(DEFAULT_DISASSEMBLER, 'r') as f:
             actual_disas_path = f.read()
-        full_disas_name_path = os.path.join(CONFIG_DIR ,actual_disas_path)
+        full_disas_name_path = os.path.join(CONFIG_DIR, actual_disas_path)
         if os.path.isfile(full_disas_name_path):
             with open(full_disas_name_path, 'r') as f:
                 DISASSEMBLER_PATH = f.read()

src/karta/disassembler/IDA/ida_cmd_api.py

@@ -33,7 +33,7 @@ def name():
 
     # Overridden base function
     async def createDatabase(self, binary_file, is_windows):
-        """Create a database file for the given binary file, compiled to windows or linux as specified.
+        """Create a database file asynchonously for the given binary file , compiled to windows or linux as specified.
 
         Args:
             binary_file (path): path to the input binary (*.o / *.obj) file
@@ -49,36 +49,57 @@ async def createDatabase(self, binary_file, is_windows):
 
         database_file = binary_file + self.suffix
         # execute the program
-        process = await asyncio.create_subprocess_exec(self._path, "-A" , "-B", f"-T{type}" ,f"-o{database_file}", binary_file)
+        process = await asyncio.create_subprocess_exec(self._path, "-A", "-B", f"-T{type}", f"-o{database_file}", binary_file)
         await process.wait()
         # return back the (should be) created database file path
         return database_file
 
     # Overridden base function
     async def executeScript(self, database, script):
-        """Execute the given script over the given database file that was created earlier.
+        """Execute the given script asynchonously over the given database file that was created earlier.
 
         Args:
             database (path): path to a database file created by the same program
             script (path): python script to be executed once the database is loaded
         """
         process = await asyncio.create_subprocess_exec(self._path, "-A", f"-S{script}", database)
         await process.wait()
-    
+
     def isSupported(self, feature_name):
+        """Check if feature is enabled.
+
+        Args:
+            feature_name (str): name of the feature to check if enabled
+
+        Return Value:
+            returns whether the current class has a member name "feature_name"
+        """
         return hasattr(self, feature_name)
 
     async def createAndExecute(self, binary_file, is_windows, script):
+        """Execute the given script over the given database asynchonously without closing it first.
+
+        Args:
+            binary_file (str): filename of the binary to analyze
+            is_windows (bool): whether the file is a windows compiled file or linux compiled file
+            script (str): filename of the script to use on the binary in ida
+        """
         type = "elf" if not is_windows else "coff"
 
         if not hasattr(self, "is64"):
             self.decideArchitecureChoices(binary_file, is_windows)
 
         # execute the program
-        process = await asyncio.create_subprocess_exec(self._path, "-A", "-c" , f"-S{script}", f"-T{type}", binary_file)
+        process = await asyncio.create_subprocess_exec(self._path, "-A", "-c", f"-S{script}", f"-T{type}", binary_file)
         await process.wait()
 
     def decideArchitecureChoices(self, binary_file, is_windows):
+        """Automate deciding whether to send the files to ida64 or ida.
+
+        Args:
+            binary_file (str): filename of the file to be a test case for analysis
+            is_windows (bool): whether the test file is a linux or windows binary
+        """
         # machine type header of pe
         # specified in that order, amd64, arm64, ia64, loongarch64, riscv64
         ARCH64PE = [b"\x64\x86", b"\x64\xaa", b"\x00\x02", b"\x64\x62", b"\x64\x50"]
@@ -106,6 +127,7 @@ def decideArchitecureChoices(self, binary_file, is_windows):
                     self.is64 = True
         self.suffix = ".i64" if self.is64 else ".idb"
         self._path += "64" if self.is64 else ""
-        
+
+
 # Don't forget to register at the factory
 registerDisassemblerCMD(IdaCMD.identify, IdaCMD)

src/karta/disassembler/disas_api.py

@@ -3,7 +3,6 @@
 ##################################################################################################################
 
 import os
-import asyncio
 from collections    import defaultdict
 
 class DisasAPI(object):
@@ -553,7 +552,7 @@ def name():
         raise NotImplementedError("Subclasses should implement this!")
 
     async def createDatabase(self, binary_file, is_windows):
-        """Create a database file for the given binary file, compiled to windows or linux as specified.
+        """Create a database file asynchonously for the given binary file, compiled to windows or linux as specified.
 
         Args:
             binary_file (path): path to the input binary (*.o / *.obj) file
@@ -565,15 +564,20 @@ async def createDatabase(self, binary_file, is_windows):
         raise NotImplementedError("Subclasses should implement this!")
 
     async def executeScript(self, database, script):
-        """Execute the given script over the given database file that was created earlier.
+        """Execute the given script asynchonously over the given database file that was created earlier.
 
         Args:
             database (path): path to a database file created by the same program
             script (path): python script to be executed once the database is loaded
         """
         raise NotImplementedError("Subclasses should implement this!")
-    
+
     def isSupported(self, feature_name):
+        """Allow script to query whether a feature is enabled in it's disas_api.
+
+        Return Value:
+            False, no featue is supported on this Abstract class
+        """
         return False
 
 class DisasVerifier(object):

src/karta/installers/common_installer.py

@@ -4,10 +4,8 @@
 from ..config.utils import addDisassembler, getDisassembler, setDefaultDisassembler
 
 
-def common_locations():
-    """
-        Gets a list of common places where a disassembler may be installed
-    """
+def commonLocations():
+    """Get a list of common places where a disassembler may be installed on any os."""
     location_list = list()
     # based on the operating system installation location may vary
     system_lower = platform.system().lower()
@@ -30,18 +28,21 @@ def common_locations():
     return location_list
 
 
-def detect_installation(pattern, installation_file):
-    """
-        If already installed get the installation path
-        Detect installation of a disassembler by a regex provided by its specific installer
-        Save it into a file so you may use it as a default
-        If its not in any of the common locations ask the user to enter it
+def detectInstallation(pattern, installation_file):
+    """Detect a disassembler's installation by regex, if not found get it with input.
+
+    Args:
+        pattern (re.Pattern): pattern to detect an installation
+        installation_file (str): name of the file to save the disassembler name in
+
+    Return Value:
+        Directory of the disassembler file
     """
     disassembler = getDisassembler(installation_file)
     if disassembler is not None:
         return disassembler
 
-    for location in common_locations():
+    for location in commonLocations():
         for directory in next(os.walk(location))[1]:
             if pattern.match(directory):
                 install_directory = os.path.join(location, directory)
@@ -59,5 +60,6 @@ def detect_installation(pattern, installation_file):
         exit(0)
 
 
-def set_default_disassembler(disassembler_name):
+def commonSetDefaultDisassembler(disassembler_name):
+    """Call config utils setDefaultDisassembler function."""
     setDefaultDisassembler(disassembler_name)

src/karta/installers/ida_installer.py

@@ -2,24 +2,27 @@
 import re
 from shutil import copyfile
 
-from .common_installer import detect_installation, set_default_disassembler
+from .common_installer import detectInstallation, commonSetDefaultDisassembler
 
 
 disassembler_name = "ida_path"
 
-def detect_ida(save_file):
-    """
-    find where ida is installed by this regex, should work for both windows and linux
+def detectIda(save_file):
+    """Find where ida is installed by this regex, should work for both windows and linux.
+
+    Args:
+        save_file (str): name of the file which will save the path to the disassembler directory
+
+    Return Value:
+        Installation folder for the ida pro disassembler
     """
-    pattern = re.compile("ida( pro )?-?\d\.\d", re.IGNORECASE)
-    return detect_installation(pattern, save_file)
+    pattern = re.compile("ida( pro )?-?\\d\\.\\d", re.IGNORECASE)
+    return detectInstallation(pattern, save_file)
 
 def main():
-    """
-    detect ida copy library files into its plugins directory and add the plugin to be run on start
-    """
-    path = detect_ida(disassembler_name)
-    set_default_disassembler(disassembler_name)
+    """Detect ida copy plugin file into its plugins directory."""
+    path = detectIda(disassembler_name)
+    commonSetDefaultDisassembler(disassembler_name)
     src_dir = os.path.join(os.path.dirname(os.path.realpath(__file__)), '..')
     ida_plugin = os.path.join(src_dir, "plugins", "ida_plugin.py")
     plugin_dst = os.path.join(path, "plugins", "ida_karta.py")
@@ -29,6 +32,5 @@ def main():
     )
 
 
-
 if __name__ == "__main__":
     main()

src/karta/karta_analyze_src.py

@@ -48,7 +48,7 @@ def locateFiles(bin_dir, file_list, suffix):
                 yield path.abspath(path.join(root, file)), file
 
 async def analyzeFile(full_file_path, is_windows):
-    """Analyze a single file using analyzer script.
+    """Analyze a single file asynchonously using analyzer script.
 
     Args:
         full_file_path (str): full path to the specific (*.obj / *.o) file
@@ -61,6 +61,16 @@ async def analyzeFile(full_file_path, is_windows):
         await disas_cmd.executeScript(database_path, SCRIPT_PATH)
 
 async def processFile(full_file_path, is_windows, compiled_file, progress_bar, prompter, semaphore):
+    """Analyze a file asynchonously using a disassembler and parse the file stats.
+
+    Args:
+        full_file_path (str): full path to the specific (*.obj / *.o) file
+        is_windows (bool): True iff a windows compilation (*.obj or *.o)
+        compiled_file (str): file name of the compiled_file
+        progress_bar (progressBar): progress bar to update once file processing is finished
+        prompter (prompter): notify the user when an error occured
+        semaphore (semaphore): release it when the disassembler finished processing our file
+    """
     # run the analsys on the files in an asynchrnous way
     prompter.debug(f"{full_file_path} - {compiled_file}")
     if progress_bar is None:
@@ -261,7 +271,14 @@ async def analyzeLibrary(config_name, bin_dirs, compiled_ars, concurrency, promp
     prompter.info(f"Anchor to function ratio is: {len(anchors_list)}/{len(src_functions_list)}")
     prompter.removeIndent()
 
-def verify_archives_and_objects(using_archives, couples, prompter):
+def verifyArchivesAndObjects(using_archives, couples, prompter):
+    """Verify that the archive and object files karta processes are valid.
+
+    Args:
+        using_archives (bool): is karta analyzing object files only or library files too
+        couples (list): couples of dir and lib files or only bin dirs according to the using_archives parameter
+        prompter (prompter): prompter to notify the user if any error occures
+    """
     bin_dirs = []
     archive_paths = []
     error_occured = False
@@ -275,7 +292,7 @@ def verify_archives_and_objects(using_archives, couples, prompter):
                 error_occured = True
             else:
                 bin_dirs.append(couples[i])
-            
+
             if not path.exists(couples[i + 1]):
                 prompter.error(f"Error the path {couples[i + 1]} does not exist!")
                 error_occured = True
@@ -332,8 +349,8 @@ def main(args=None):
     if using_archives:
         if len(couples) % 2 != 0:
             parser.error("Odd length in list of dir,archive couples, should be: [(directory, archive name), ...]")
-    
-    paths = verify_archives_and_objects(using_archives, couples, prompter)
+
+    paths = verifyArchivesAndObjects(using_archives, couples, prompter)
 
     if paths is None:
         return
@@ -354,7 +371,6 @@ def main(args=None):
     if is_windows:
         setWindowsMode()
 
-
     # analyze the open source library
     asyncio.run(analyzeLibrary(constructConfigPath(library_name, library_version), bin_dirs, archive_paths, concurrency, prompter))