Merge branch 'ossf:main' into fix-api-for-analysis

oss_fuzz_integration/oss-fuzz-patches.diff

@@ -29,7 +29,7 @@ index 64d11095b..ec8ad457c 100644
  # /ccache/bin will contain the compiler wrappers, and /ccache/cache will
  # contain the actual cache, which can be saved.
 diff --git a/infra/base-images/base-builder/compile b/infra/base-images/base-builder/compile
-index 1c10d9e23..25caa5897 100755
+index 1c10d9e23..5e10ccc14 100755
 --- a/infra/base-images/base-builder/compile
 +++ b/infra/base-images/base-builder/compile
 @@ -209,9 +209,9 @@ if [ "$SANITIZER" = "introspector" ] || [ "$RUST_SANITIZER" = "introspector" ];
@@ -143,20 +143,47 @@ index 1c10d9e23..25caa5897 100755
 
    # Move coverage report.
    if [ -d "$OUT/textcov_reports" ]
-@@ -356,6 +373,7 @@ if [ "$SANITIZER" = "introspector" ] || [ "$RUST_SANITIZER" = "introspector" ];
+@@ -356,17 +373,18 @@ if [ "$SANITIZER" = "introspector" ] || [ "$RUST_SANITIZER" = "introspector" ];
    cd $SRC/inspector
 
    # Make fuzz-introspector HTML report.
 +  export PROJECT_NAME="123"
    REPORT_ARGS="--name=$PROJECT_NAME"
    # Only pass coverage_url when COVERAGE_URL is set (in cloud builds)
    if [[ ! -z "${COVERAGE_URL+x}" ]]; then
-@@ -386,16 +404,19 @@ if [ "$SANITIZER" = "introspector" ] || [ "$RUST_SANITIZER" = "introspector" ];
+-    REPORT_ARGS="$REPORT_ARGS --coverage_url=${COVERAGE_URL}"
++    REPORT_ARGS="$REPORT_ARGS --coverage-url=${COVERAGE_URL}"
+   fi
+
+   # Do different things depending on languages
+   if [ "$FUZZING_LANGUAGE" = "python" ]; then
+     echo "GOING python route"
+     set -x
+-    REPORT_ARGS="$REPORT_ARGS --target_dir=$SRC/inspector"
++    REPORT_ARGS="$REPORT_ARGS --target-dir=$SRC/inspector"
+     REPORT_ARGS="$REPORT_ARGS --language=python"
+     python3 /fuzz-introspector/src/main.py report $REPORT_ARGS
+     rsync -avu --delete "$SRC/inspector/" "$OUT/inspector"
+@@ -374,28 +392,31 @@ if [ "$SANITIZER" = "introspector" ] || [ "$RUST_SANITIZER" = "introspector" ];
+     echo "GOING jvm route"
+     set -x
+     find $OUT/ -name "jacoco.xml" -exec cp {} $SRC/inspector/ \;
+-    REPORT_ARGS="$REPORT_ARGS --target_dir=$SRC/inspector"
++    REPORT_ARGS="$REPORT_ARGS --target-dir=$SRC/inspector"
+     REPORT_ARGS="$REPORT_ARGS --language=jvm"
+     python3 /fuzz-introspector/src/main.py report $REPORT_ARGS
+     rsync -avu --delete "$SRC/inspector/" "$OUT/inspector"
+   elif [ "$FUZZING_LANGUAGE" = "rust" ]; then
+     echo "GOING rust route"
+-    REPORT_ARGS="$REPORT_ARGS --target_dir=$SRC/inspector"
++    REPORT_ARGS="$REPORT_ARGS --target-dir=$SRC/inspector"
+     REPORT_ARGS="$REPORT_ARGS --language=rust"
+     python3 /fuzz-introspector/src/main.py report $REPORT_ARGS
      rsync -avu --delete "$SRC/inspector/" "$OUT/inspector"
    else
      # C/C++
 +    mkdir -p $SRC/inspector
-+    fuzz-introspector full --target_dir=$SRC --language=${FUZZING_LANGUAGE} --out-dir=$SRC/inspector ${REPORT_ARGS}
++    fuzz-introspector full --target-dir=$SRC --language=${FUZZING_LANGUAGE} --out-dir=$SRC/inspector ${REPORT_ARGS}
 +    #fuzz-introspector full --target_dir $SRC --language=${FUZZING_LANGUAGE}
 
      # Correlate fuzzer binaries to fuzz-introspector's raw data

src/fuzz_introspector/analyses/annotated_cfg.py

@@ -17,14 +17,14 @@
 import logging
 import json
 
-from typing import (Any, Dict, List)
+from typing import (Any, Dict, List, Optional)
 
 from fuzz_introspector import analysis
 from fuzz_introspector import utils
 from fuzz_introspector import cfg_load
 from fuzz_introspector import json_report
 from fuzz_introspector import html_helpers
-from fuzz_introspector.datatypes import project_profile, fuzzer_profile
+from fuzz_introspector.datatypes import project_profile, fuzzer_profile, function_profile
 
 logger = logging.getLogger(name=__name__)
 
@@ -35,7 +35,7 @@ class FuzzAnnotatedCFG(analysis.AnalysisInterface):
     def __init__(self) -> None:
         logger.info("Creating annotated CFG")
         self.json_string_result = ""
-        self.json_results: Dict[str, Any] = dict()
+        self.json_results: Dict[str, Any] = {}
         self.dump_files = False
 
     @classmethod
@@ -62,13 +62,13 @@ def analysis_func(self,
         logger.info("Creating annotated CFGs")
 
         for profile in profiles:
-            logger.info("Analysing: %s" % (profile.identifier))
+            logger.info('Analysing: %s', profile.identifier)
             destinations = []
             src_file = None
             is_first = True
             # We must haev a high number here initially, to trigger the first
             # catch.
-            callsite_stack = dict()
+            callsite_stack = {}
             for callsite in cfg_load.extract_all_callsites(
                     profile.fuzzer_callsite_calltree):
 
@@ -103,11 +103,16 @@ def analysis_func(self,
                                      parent_callsite.dst_function_name))
 
                 # To be a top level target a callsite should:
-                # 1.0) Not be in the fuzzer source file and one of the following:
+                # 1.0) Not be in the fuzzer source file and one of the
+                #    following:
                 #    1a) have parent callsite be in the fuzzer, i.e. transition
                 #        from files.
-                #    1b) Have calldepth 1 (i.e. directly from LLVMFuzzerTestOneInput),
+                #    1b) Have calldepth 1 (i.e. directly from
+                #        LLVMFuzzerTestOneInput),
                 #        since we know parent then is in fuzzer source file.
+                if dst_fd is None:
+                    continue
+
                 cond1 = dst_fd is not None and dst_fd.function_source_file != src_file
                 cond2 = (par_fd is not None and par_fd.function_source_file
                          == src_file) or callsite.depth == 1
@@ -146,7 +151,10 @@ def analysis_func(self,
             self.get_name(), self.get_json_string_result(), out_dir)
         return ""
 
-    def get_profile_sourcefile(self, profile, func_name):
+    def get_profile_sourcefile(
+            self, profile,
+            func_name) -> Optional[function_profile.FunctionProfile]:
+        """Gets the source file of a function within a project profile"""
         dst_options = [
             func_name,
             utils.demangle_cpp_func(func_name),
@@ -165,7 +173,11 @@ def get_profile_sourcefile(self, profile, func_name):
                 pass
         return None
 
-    def get_profile_sourcefile_merged(self, merged_profile, func_name):
+    def get_profile_sourcefile_merged(
+            self, merged_profile,
+            func_name) -> Optional[function_profile.FunctionProfile]:
+        """Gets the source file of a function withing a merged project
+        profile"""
         dst_options = [
             func_name,
             utils.demangle_cpp_func(func_name),

src/fuzz_introspector/analyses/calltree_analysis.py

@@ -51,9 +51,11 @@ def get_name(cls):
         return cls.name
 
     def get_json_string_result(self):
+        """Helper for getting json string"""
         return self.json_string_result
 
     def set_json_string_result(self, json_string):
+        """Helper for setting json string"""
         self.json_string_result = json_string
 
     def analysis_func(self,
@@ -100,14 +102,12 @@ def create_calltree(self, profile: fuzzer_profile.FuzzerProfile,
         nodes = cfg_load.extract_all_callsites(
             profile.fuzzer_callsite_calltree)
 
-        for i in range(len(nodes)):
+        for i, node in enumerate(nodes):
             # All divs created in this loop must also be closed in this loop.
-            node = nodes[i]
-
-            if (profile.target_lang == "jvm"):
+            if profile.target_lang == "jvm":
                 demangled_name = utils.demangle_jvm_func(
                     node.dst_function_source_file, node.dst_function_name)
-            elif (profile.target_lang == "rust"):
+            elif profile.target_lang == "rust":
                 demangled_name = utils.demangle_rust_func(
                     node.dst_function_name)
             else:
@@ -120,7 +120,8 @@ def create_calltree(self, profile: fuzzer_profile.FuzzerProfile,
             link = node.cov_link
             ct_idx_str = self.create_str_node_ctx_idx(str(node.cov_ct_idx))
 
-            # Only display [function] link if we have, otherwhise show no [function] text.
+            # Only display [function] link if we have, otherwhise show no
+            # [function] text.
             if node.dst_function_source_file.replace(" ", "") != "":
                 func_href = f"""<a href="{link}">[function]</a>"""
             else:
@@ -154,8 +155,8 @@ def create_calltree(self, profile: fuzzer_profile.FuzzerProfile,
             if i < len(nodes) - 1:
                 next_node = nodes[i + 1]
 
-                # If depth is increasing then we should open a new div for folding
-                # the calltree.
+                # If depth is increasing then we should open a new div for
+                # folding the calltree.
                 if next_node.depth > node.depth:
                     calltree_html_section_string += f"""<div
         class="calltree-line-wrapper open level-{int(node.depth)}"
@@ -183,9 +184,9 @@ def create_calltree(self, profile: fuzzer_profile.FuzzerProfile,
         # visualisation happens in javascript rather than here.
         calltree_html_section_string += "<div id=\"side-overview-wrapper\"></div>"
 
-        logger.info("calltree_html_section_string: <divs>: %d -- </divs>: %d" %
-                    (calltree_html_section_string.count("<div"),
-                     calltree_html_section_string.count("</div>")))
+        logger.info('calltree_html_section_string: <divs>: %d -- </divs>: %d',
+                    calltree_html_section_string.count("<div"),
+                    calltree_html_section_string.count("</div>"))
 
         calltree_html_string += calltree_html_section_string + "</div>"  # calltree-wrapper
         logger.info("Calltree created")
@@ -220,20 +221,22 @@ def collect_calltree_nodes(
                 "the blocker table won't have correct links to calltree.")
 
         blocker_node_map: Dict[analysis.FuzzBranchBlocker,
-                               cfg_load.CalltreeCallsite] = dict()
+                               cfg_load.CalltreeCallsite] = {}
         for blocker in branch_blockers:
             func_name = blocker.function_name
             branch_linenumber = int(blocker.branch_line_number)
             for idx, node in enumerate(all_callsites):
                 if func_name == node.dst_function_name:
                     depth = node.depth + 1
                     found_node = node
-                    # Try to adjust the blocker node in the callees of current func
+                    # Try to adjust the blocker node in the callees of the
+                    # current func.
                     for i in range(idx + 1, nodes_num):
                         new_node = all_callsites[i]
                         if depth > new_node.depth:
                             break  # Reached the caller of the node
-                        if depth == new_node.depth and branch_linenumber >= new_node.src_linenumber:
+                        if (depth == new_node.depth and
+                                branch_linenumber >= new_node.src_linenumber):
                             found_node = new_node
                     blocker_node_map[blocker] = found_node
                     break
@@ -246,7 +249,8 @@ def html_create_dedicated_calltree_file(
         """
         Write a wrapped HTML file with the tags needed from fuzz-introspector
         We use this only for wrapping calltrees at the moment, however, down
-        the line it makes sense to have an easy wrapper for other HTML pages too.
+        the line it makes sense to have an easy wrapper for other HTML pages
+        too.
         """
         complete_html_string = ""
         blocker_infos = {}
@@ -319,15 +323,16 @@ def get_fuzz_blockers(
             max_blockers_to_extract: int = 999
     ) -> List[cfg_load.CalltreeCallsite]:
         """Gets a list of fuzz blockers"""
-        blocker_list: List[cfg_load.CalltreeCallsite] = list()
+        blocker_list: List[cfg_load.CalltreeCallsite] = []
 
         # Extract all callsites in calltree and exit early if none
         all_callsites = cfg_load.extract_all_callsites(
             profile.fuzzer_callsite_calltree)
         if len(all_callsites) == 0:
             return blocker_list
 
-        # Filter nodes that has forward reds. Extract maximum max_blockers_to_extract nodes.
+        # Filter nodes that has forward reds. Extract maximum
+        # max_blockers_to_extract nodes.
         nodes_sorted_by_red_ahead = sorted(all_callsites,
                                            key=lambda x: x.cov_forward_reds,
                                            reverse=True)
@@ -377,7 +382,7 @@ def create_fuzz_blocker_table(
             sort_order="desc")
         for node in fuzz_blockers:
             link_prefix = "0" * (5 - len(str(node.cov_ct_idx)))
-            node_id = "%s%s" % (link_prefix, node.cov_ct_idx)
+            node_id = f'{link_prefix}{node.cov_ct_idx}'
             if file_link is not None:
                 cs_link = (
                     "<span class=\"text-link\">"

src/fuzz_introspector/analyses/engine_input.py

@@ -31,6 +31,7 @@
 
 
 class EngineInput(analysis.AnalysisInterface):
+    """Generates content that can be used by fuzz engines."""
     name: str = "FuzzEngineInputAnalysis"
 
     def __init__(self) -> None:
@@ -50,12 +51,12 @@ def set_json_string_result(self, json_string):
     def analysis_func(self,
                       table_of_contents: html_helpers.HtmlTableOfContents,
                       tables: List[str],
-                      project_profile: project_profile.MergedProjectProfile,
+                      proj_profile: project_profile.MergedProjectProfile,
                       profiles: List[fuzzer_profile.FuzzerProfile],
                       basefolder: str, coverage_url: str,
                       conclusions: List[html_helpers.HTMLConclusion],
                       out_dir) -> str:
-        logger.info(f" - Running analysis {self.get_name()}")
+        logger.info('- Running analysis %s', self.get_name())
 
         if not self.display_html:
             # Overwrite the table of contents variable, to avoid displaying
@@ -72,29 +73,27 @@ def analysis_func(self,
                        "to a fuzz engine when running a given fuzz target. The current " \
                        "focus is on providing input that is usable by libFuzzer.</p>"
 
-        for profile_idx in range(len(profiles)):
-            logger.info(
-                f"Generating input for {profiles[profile_idx].identifier}")
+        for prof in profiles:
+            logger.info('Generating input for %s', prof.identifier)
             html_string += html_helpers.html_add_header_with_link(
-                profiles[profile_idx].fuzzer_source_file,
-                html_helpers.HTML_HEADING.H2, table_of_contents)
+                prof.fuzzer_source_file, html_helpers.HTML_HEADING.H2,
+                table_of_contents)
 
             # Create dictionary section
-            html_string += self.get_dictionary_section(profiles[profile_idx],
-                                                       table_of_contents,
+            html_string += self.get_dictionary_section(prof, table_of_contents,
                                                        out_dir)
 
             html_string += "<br>"
 
             # Create focus function section
             html_string += self.get_fuzzer_focus_function_section(
-                profiles[profile_idx],
+                prof,
                 table_of_contents,
             )
         html_string += "</div>"  # .collapsible
         html_string += "</div>"  # report-box
 
-        logger.info(f" - Completed analysis {self.get_name()}")
+        logger.info('- Completed analysis %s', self.get_name())
         if not self.display_html:
             html_string = ""
 
@@ -167,8 +166,8 @@ def get_fuzzer_focus_function_section(
                     focus_functions.append(utils.demangle_rust_func(ffname))
                 else:
                     focus_functions.append(utils.demangle_cpp_func(ffname))
-                logger.info(
-                    f"Found focus function: {fuzz_blocker.src_function_name}")
+                logger.info('Found focus function: %s',
+                            fuzz_blocker.src_function_name)
 
         if len(focus_functions) == 0:
             return ""
@@ -186,18 +185,19 @@ def get_fuzzer_focus_function_section(
 
     def add_to_json_file(self, json_file_path: str, fuzzer_name: str, key: str,
                          val: List[str]) -> None:
+        """Add key to json dictionary in `json_file_path`."""
         # Create file if it does not exist
         if not os.path.isfile(json_file_path):
-            json_data = dict()
+            json_data = {}
         else:
             json_fd = open(json_file_path)
             json_data = json.load(json_fd)
             json_fd.close()
         if 'fuzzers' not in json_data:
-            json_data['fuzzers'] = dict()
+            json_data['fuzzers'] = {}
 
         if fuzzer_name not in json_data['fuzzers']:
-            json_data['fuzzers'][fuzzer_name] = dict()
+            json_data['fuzzers'][fuzzer_name] = {}
 
         json_data['fuzzers'][fuzzer_name][key] = val
 

src/fuzz_introspector/analyses/filepath_analyser.py

@@ -58,7 +58,7 @@ def analysis_func(self,
                       basefolder: str, coverage_url: str,
                       conclusions: List[html_helpers.HTMLConclusion],
                       out_dir) -> str:
-        logger.info(f" - Running analysis {self.get_name()}")
+        logger.info('- Running analysis %s', self.get_name())
 
         all_proj_files = self.all_files_targeted(proj_profile)
         all_proj_dirs = set()