syz-introspector: more refactoring and cleaning up

tools/syz-introspector/src/fuzz_introspector_utils.py

@@ -80,7 +80,7 @@ def cleanup_files(workdir: str = ""):
         os.remove('targetCalltree.txt')
 
 
-def get_all_c_files_mentioned_in_light(workdir, all_source) -> List[str]:
+def get_c_files_mentioned_in_light(workdir) -> List[str]:
     """Gets C source files mention in light FI report."""
     with open(os.path.join(workdir, 'report.yaml'), 'r',
               encoding='utf-8') as f:
@@ -92,7 +92,7 @@ def get_all_c_files_mentioned_in_light(workdir, all_source) -> List[str]:
     return all_files
 
 
-def get_all_header_files_in_light(workdir, all_sources) -> List[str]:
+def get_all_header_files_in_light(workdir) -> List[str]:
     """Gets list of header files in light introspector report and
     finds them in the target kernel folder."""
     all_header_files = []
@@ -108,7 +108,7 @@ def get_all_header_files_in_light(workdir, all_sources) -> List[str]:
     return all_header_files
 
 
-def extract_calltree_light(target_function, kernel_dir, workdir, target_dir):
+def extract_calltree_light(target_function, workdir, target_dir):
     """Light introspector run"""
 
     # logging.info('Analysing: %s' % (workdir))

tools/syz-introspector/src/main.py

@@ -102,11 +102,10 @@ def parse_args() -> argparse.Namespace:
     return args
 
 
-def extract_source_loc_analysis(workdir: str, all_sources: List[str],
-                                report) -> None:
+def extract_source_loc_analysis(workdir: str, report) -> None:
     """Extracts the lines of code in each C source code file."""
-    all_c_files = fuzz_introspector_utils.get_all_c_files_mentioned_in_light(
-        workdir, all_sources)
+    all_c_files = fuzz_introspector_utils.get_c_files_mentioned_in_light(
+        workdir)
     logger.info('[+] Source files:')
     source_files = []
     total_loc = 0
@@ -161,7 +160,7 @@ def get_possible_devnodes(ioctl_handlers):
     return all_devnodes
 
 
-def analyse_ioctl_handler(ioctl_handler, workdir, kernel_folder, target_path):
+def analyse_ioctl_handler(ioctl_handler, workdir, target_path):
     """Extracts the calltree from a given ioctl handler and creates a Fuzz
     Introspector HTML report for it as well. The data will be written in a
     folder within the working directory.
@@ -180,8 +179,7 @@ def analyse_ioctl_handler(ioctl_handler, workdir, kernel_folder, target_path):
     # Extract the calltree. Do this by running an introspector run
     # to generate the calltree as well as analysis files.
     calltree = fuzz_introspector_utils.extract_calltree_light(
-        ioctl_handler['func']['functionName'], kernel_folder, fi_data_dir,
-        target_path)
+        ioctl_handler['func']['functionName'], fi_data_dir, target_path)
 
     if calltree:
         ioctl_handler['calltree'] = calltree
@@ -224,13 +222,12 @@ def analyse_ioctl_handler(ioctl_handler, workdir, kernel_folder, target_path):
         ioctl_handler['calltree'] = ''
 
 
-def extract_ioctls_in_driver(kernel_folder, report, workdir, all_sources,
-                             strict_ioctls):
+def extract_ioctls_in_driver(kernel_folder, report, workdir, strict_ioctls):
     """Extracts ioctls defined/used in driver"""
-    ioctls_defined = textual_source_analysis.extract_raw_ioctls_text_from_header_files(
+    ioctls_defined = textual_source_analysis.extract_ioctls_from_files(
         report['header_files'], kernel_folder)
-    c_files_in_driver = fuzz_introspector_utils.get_all_c_files_mentioned_in_light(
-        workdir, all_sources)
+    c_files_in_driver = fuzz_introspector_utils.get_c_files_mentioned_in_light(
+        workdir)
     if strict_ioctls:
         refined_ioctls = []
         for ioctl in ioctls_defined:
@@ -269,28 +266,27 @@ def main() -> None:
         os.environ['FI_KERNEL_COV'] = args.coverage_report
 
     # Extract source file structure.
-    all_sources = identify_kernel_source_files(kernel_folder)
+    identify_kernel_source_files(kernel_folder)
 
     # Run base introspector. In this run there are no entrypoints analysed.
     run_light_fi(target_path, workdir)
-    extract_source_loc_analysis(workdir, all_sources, report)
+    extract_source_loc_analysis(workdir, report)
 
     # Find all header files.
     logger.info('[+] Finding header files')
-    report['header_files'] = syz_core.extract_header_files_referenced(
-        workdir, all_sources)
-    logger.info('Found a total of %d header files',
-                len(report['header_files']))
+    report['header_files'] = syz_core.extract_header_files_referenced(workdir)
+    logger.debug('Found a total of %d header files',
+                 len(report['header_files']))
     for header_file in report['header_files']:
-        logger.info('- %s', header_file)
+        logger.debug('- %s', header_file)
 
     new_headers = []
     logger.info('Refining header files')
     for header_file in report['header_files']:
-        logger.info('r: %s', header_file)
+        logger.debug('r: %s', header_file)
         vt = textual_source_analysis.find_file(header_file)
         if vt:
-            logger.info('--- %s', vt)
+            logger.debug('--- %s', vt)
             new_headers.append(vt)
     logger.info('Refined to %d', len(new_headers))
 
@@ -301,15 +297,14 @@ def main() -> None:
     # Extract ioctls.
     logger.info('[+] Extracting raw ioctls')
     report['ioctls'] = extract_ioctls_in_driver(kernel_folder, report, workdir,
-                                                all_sources,
                                                 args.strict_ioctls)
+    logger.info('[+] Found the following ioctls')
     for ioctl in report['ioctls']:
         logger.info('%s ::: %s', ioctl.raw_definition, ioctl.name)
 
     logger.info('[+] Scanning for ioctl handler using text analysis')
     ioctl_handlers = syz_core.get_ioctl_handlers(report['ioctls'],
-                                                 kernel_folder, report,
-                                                 workdir)
+                                                 kernel_folder, workdir)
 
     # Get possible set of devnodes.
     all_devnodes = get_possible_devnodes(ioctl_handlers)
@@ -318,8 +313,7 @@ def main() -> None:
 
     # extract calltrees
     for ioctl_handler in report['ioctl_handlers']:
-        analyse_ioctl_handler(ioctl_handler, workdir, kernel_folder,
-                              target_path)
+        analyse_ioctl_handler(ioctl_handler, workdir, target_path)
 
     logger.info('[+] Showing complexity of ioctl handlers')
     syz_core.interpret_complexity_of_ioctl_handlers(report['ioctl_handlers'])

tools/syz-introspector/src/syz_core.py

@@ -53,27 +53,28 @@ def find_ioctl_first_case_uses(ioctl_handler, kernel_folder):
     for idx, line in enumerate(content.split('\n')):
         already_seen = set()
         for ioctl in ioctl_handler['ioctls']:
-            if ioctl.name in line and ioctl.name not in already_seen and 'case' in line:
+            if (ioctl.name in line and ioctl.name not in already_seen
+                    and 'case' in line):
                 logger.debug("%s :: %d", line.replace("\n", ""), idx)
                 already_seen.add(ioctl.name)
                 pair_starts.append((ioctl.name, idx + 1))
     return pair_starts
 
 
-def extract_header_files_referenced(workdir, all_sources) -> Set[str]:
+def extract_header_files_referenced(workdir) -> Set[str]:
     """extract the source of all header files from FI output."""
 
     raw_header_file_references = fuzz_introspector_utils.get_all_header_files_in_light(
-        workdir, all_sources)
+        workdir)
 
     all_files = set()
     for raw_file_reference in raw_header_file_references:
-        logger.info('Header file -: %s', raw_file_reference)
+        logger.debug('Header file -: %s', raw_file_reference)
         path2 = raw_file_reference.replace('file_location:',
                                            '').strip().split(':')[0].replace(
                                                "'", '')
         normalized = os.path.normpath(path2)
-        logger.info('Adding %s', normalized)
+        logger.debug('Adding %s', normalized)
         all_files.add(normalized)
 
     logger.debug('Files found')
@@ -99,13 +100,13 @@ def extract_header_files_referenced(workdir, all_sources) -> Set[str]:
                     '>', '').replace('<', '').replace('\"',
                                                       '').replace(' ', '')
 
-                logger.info('Including: %s', header_included)
+                logger.debug('Including: %s', header_included)
                 new_files.add(header_included)
     all_files = all_files.union(new_files)
 
     found_files = []
     for header_file in all_files:
-        logger.info('Finding F1')
+        logger.debug('Finding F1')
         valid_target = textual_source_analysis.find_file(header_file)
         if valid_target:
             found_files.append(valid_target)
@@ -115,8 +116,7 @@ def extract_header_files_referenced(workdir, all_sources) -> Set[str]:
     return found_files
 
 
-def extract_syzkaller_types_to_analyze(ioctls, syzkaller_description,
-                                       typedict) -> Dict[str, str]:
+def extract_syzkaller_types_to_analyze(ioctls, typedict) -> Dict[str, str]:
     """goes through ioctls_per_fp and syzkaller_description, and sets something
     in syzkaller_description and returns a set of types to analyse."""
 
@@ -181,8 +181,8 @@ def extract_syzkaller_types_to_analyze(ioctls, syzkaller_description,
                         target = '%s  array[%s, %s]' % (field_name, arr_type,
                                                         array_count)
                     else:
-                        target = '%s     %s' % (syz_name, syz_type)
-                    description_str += ' %s\n' % (target)
+                        target = f'{syz_name}     {syz_type}'
+                    description_str += f' {target}\n'
 
                 description_str += '}'
                 description_types[dtype] = description_str
@@ -201,7 +201,7 @@ def get_next_syzkaller_workdir():
 
 
 def write_syzkaller_description(ioctls, syzkaller_description, workdir,
-                                all_devnodes, header_file, target_path):
+                                all_devnodes, target_path):
     """Writes a description-X.txt representing syzkaller description."""
 
     # Ensure there are actually ioctls to generate
@@ -257,7 +257,7 @@ def write_syzkaller_description(ioctls, syzkaller_description, workdir,
         f.write('\n' * 2)
 
         # Describe the types
-        for st, text in syzkaller_description.items():  #['types']:
+        for _, text in syzkaller_description.items():  #['types']:
             f.write(text)
             f.write('\n' * 2)
     return next_syz_descr
@@ -274,13 +274,12 @@ def get_function_containing_line_idx(line_idx, sorted_functions):
     return None
 
 
-def check_source_files_for_ioctl(kernel_folder, src_file, ioctls,
-                                 all_files_with_func):
+def check_source_files_for_ioctl(src_file, ioctls, all_files_with_func):
     """For a given set of of IOCTLs and a source file, finds the functions
     in the source file from the `all_files_with_func` that uses the IOCTLs."""
     all_ioctl_func_handlers = list()
 
-    logger.info('Finding F2')
+    logger.info('Finding source files with ioctls usage.')
     target_file = textual_source_analysis.find_file(src_file)
     if not target_file:
         return []
@@ -323,14 +322,10 @@ def check_source_files_for_ioctl(kernel_folder, src_file, ioctls,
 
     if len(functions_with_ioctls_in_them) > 0:
         logger.debug('Functions with ioctls in them')
-        for interesting_func in functions_with_ioctls_in_them:
-            all_ioctl_func_handlers.append(
-                functions_with_ioctls_in_them[interesting_func])
-            logger.debug(
-                "- %s",
-                functions_with_ioctls_in_them[interesting_func]['func'])
-            for ioctl in functions_with_ioctls_in_them[interesting_func][
-                    'ioctls']:
+        for func_obj in functions_with_ioctls_in_them.values():
+            all_ioctl_func_handlers.append(func_obj)
+            logger.debug("- %s", func_obj['func'])
+            for ioctl in func_obj['ioctls']:
                 logger.debug('  - %s', (ioctl.name))
 
     return all_ioctl_func_handlers
@@ -375,7 +370,7 @@ def find_all_unlocked_ioctls(source_files_to_functions_mapping):
     return unlocked_ioctl_functions
 
 
-def get_ioctl_handlers(ioctls, kernel_folder, report, fi_data_dir):
+def get_ioctl_handlers(ioctls, kernel_folder, fi_data_dir):
     """Finds the places in the source code where IOCTL commands are used."""
 
     logger.info('Getting ioctl handlers')
@@ -408,8 +403,7 @@ def get_ioctl_handlers(ioctls, kernel_folder, report, fi_data_dir):
 
         for src_file in source_files_to_functions_mapping:
             tmp_ioctl_handlers = check_source_files_for_ioctl(
-                kernel_folder, src_file, ioctls,
-                source_files_to_functions_mapping)
+                src_file, ioctls, source_files_to_functions_mapping)
             ioctl_handlers += tmp_ioctl_handlers
 
     for unlocked_ioctl_handler in unlocked_ioctl_handlers:
@@ -667,9 +661,8 @@ def extract_types_of_syzkaller_description(ioctls, fi_data_dir):
         type_dict['structs'] += struct_list
         type_dict['typedefs'] += typedefs
 
-    syzkaller_description = {'types': dict()}
     all_types_to_decipher = extract_syzkaller_types_to_analyze(
-        ioctls, syzkaller_description, type_dict)
+        ioctls, type_dict)
 
     logger.info('All types extracted from struct to include in description:')
     logger.info(json.dumps(list(all_types_to_decipher), indent=2))
@@ -695,11 +688,8 @@ def create_and_dump_syzkaller_description(ioctls_per_fp, workdir: str,
 
     syzkaller_description_types = {}
     all_ioctls = []
-    for header_file, ioctls in header_files_to_ioctls.items():
-        logger.info('Header file:')
-        logger.info(header_file)
+    for _, ioctls in header_files_to_ioctls.items():
         all_ioctls.extend(ioctls)
-
         types_to_dump = extract_types_of_syzkaller_description(
             ioctls, fi_data_dir)
         for type_to_dump, type_text in types_to_dump.items():
@@ -710,7 +700,7 @@ def create_and_dump_syzkaller_description(ioctls_per_fp, workdir: str,
     logger.info('[+] Creating syzkaller description for ')
 
     syzkaller_description_path = write_syzkaller_description(
-        all_ioctls, syzkaller_description_types, workdir, all_devnodes, '',
+        all_ioctls, syzkaller_description_types, workdir, all_devnodes,
         target_path)
     if syzkaller_description_path:
         logger.info('[+] - auto-generated description: %s',

tools/syz-introspector/src/syzkaller_util.py

@@ -16,27 +16,26 @@
 
 def convert_raw_type_to_syzkaller_type(raw_type) -> str:
     """Converts type seen llvm ir/debug data to syzkaller type"""
-    if raw_type == '__s32':
-        return 'int32'
-    if raw_type == '__u32':
-        return 'int32'
-    if raw_type == 'unsigned int':
-        return 'int32'
-    if raw_type == '__u64':
-        return 'int64'
-    if raw_type == 'int32_t' or raw_type == 'int':
-        return 'int32'
-    if raw_type == 'uint32_t':
-        return 'int32'
-    if raw_type == '__u64 *':
-        return 'int64'
-    if raw_type == 'char *':
-        return 'int8'
-    if raw_type == 'char':
-        return 'int8'
-    if raw_type == '__u16':
-        return 'int16'
-    return raw_type
+    type_mapping = {
+        '__u8': 'int8',
+        '__s8': 'int8',
+        'char': 'int8',
+        'char *': 'int8',
+        '__s16': 'int16',
+        '__u16': 'int16',
+        'short': 'int16',
+        '__s32': 'int32',
+        '__u32': 'int32',
+        'int': 'int32',
+        'int32_t': 'int32',
+        'uint32_t': 'int32',
+        'unsigned int': 'int32',
+        '__s64': 'int64',
+        '__u64': 'int64',
+        '__u64 *': 'int64',
+    }
+
+    return type_mapping.get(raw_type, raw_type)
 
 
 def get_type_ptr_of_syzkaller(ioctl) -> str: