Merge pull request diffblue#534 from diffblue/smowton-on-behalf-of-marek/simple-taint-flow-check

SEC-620, SEC-650: Added simple early check for impossibility of taint flow to sink

Author: marek-trtik

driver/presentation.py

@@ -182,55 +182,56 @@ def build_HTML_interface_to_slicing_tasks(root_dir,sub_dir,ofile):
     if not os.path.exists(full_sub_dir):
          os.makedirs(full_sub_dir)
 
-    build_HTML_interface_to_slicer_call_graph(
-        os.path.abspath(os.path.join(full_sub_dir,"call_graph.html"))
+    instrumented_goto_program_fname = os.path.join(full_sub_dir, "instrumented_goto_program.json")
+    if os.path.isfile(instrumented_goto_program_fname):
+        build_HTML_interface_to_slicer_call_graph(
+            os.path.abspath(os.path.join(full_sub_dir,"call_graph.html"))
         )
-    build_HTML_interface_to_slicer_inverted_call_graph(
-        os.path.abspath(os.path.join(full_sub_dir,"inverted_call_graph.html"))
+        build_HTML_interface_to_slicer_inverted_call_graph(
+            os.path.abspath(os.path.join(full_sub_dir,"inverted_call_graph.html"))
         )
-    build_HTML_interface_to_slicer_tokens_propagation_graph(
-        os.path.abspath(os.path.join(full_sub_dir,"tokens_propagation_graph.html"))
+        build_HTML_interface_to_slicer_tokens_propagation_graph(
+            os.path.abspath(os.path.join(full_sub_dir,"tokens_propagation_graph.html"))
         )
 
-    instrumentation_props,instrumentation_props_fname = load_instrumentation_props_of_slicer(full_sub_dir)
-    fname = os.path.splitext(instrumentation_props_fname)[0] + ".html"
-    build_HTML_interface_to_slicer_instrumentation_props(instrumentation_props, fname)
+        instrumentation_props,instrumentation_props_fname = load_instrumentation_props_of_slicer(full_sub_dir)
+        fname = os.path.splitext(instrumentation_props_fname)[0] + ".html"
+        build_HTML_interface_to_slicer_instrumentation_props(instrumentation_props, fname)
 
-    ofile.write("<table>\n"
-                "<caption>Supporting and intermediate data structures.</caption>\n"
-                "  <tr>\n"
-                "    <th>Property</th>\n"
-                "    <th>Value</th>\n"
-                "  </tr>\n")
-    ofile.write("  <tr>\n")
-    ofile.write("    <td>Call graph</td>\n")
-    ofile.write("    <td align=\"center\"><a href=\"./" + sub_dir + "/call_graph.html\">here</a></td>\n")
-    ofile.write("  </tr>\n")
-    ofile.write("  <tr>\n")
-    ofile.write("    <td>Inverted call graph</td>\n")
-    ofile.write("    <td align=\"center\"><a href=\"./" + sub_dir + "/inverted_call_graph.html\">here</a></td>\n")
-    ofile.write("  </tr>\n")
-    ofile.write("  <tr>\n")
-    ofile.write("    <td>Tokens propagation graph</td>\n")
-    ofile.write("    <td align=\"center\"><a href=\"./" + sub_dir + "/tokens_propagation_graph.html\">here</a></td>\n")
-    ofile.write("  </tr>\n")
-    ofile.write("  <tr>\n")
-    ofile.write("    <td>Map from rules to their application sites (in JSON format)</td>\n")
-    ofile.write("    <td align=\"center\"><a href=\"./" + sub_dir +
-                "/map_from_functions_to_rule_application_sites.json\">here</a></td>\n")
-    ofile.write("  </tr>\n")
-    ofile.write("    <td>Instrumentation properties</td>\n")
-    ofile.write("    <td align=\"center\"><a href=\"./" +
-                os.path.join(sub_dir,os.path.splitext(os.path.basename(instrumentation_props_fname))[0] + ".html") +
-                "\">here</a></td>\n")
-    ofile.write("  </tr>\n")
-    ofile.write("</table>\n")
+        ofile.write("<table>\n"
+                    "<caption>Supporting and intermediate data structures.</caption>\n"
+                    "  <tr>\n"
+                    "    <th>Property</th>\n"
+                    "    <th>Value</th>\n"
+                    "  </tr>\n")
+        ofile.write("  <tr>\n")
+        ofile.write("    <td>Call graph</td>\n")
+        ofile.write("    <td align=\"center\"><a href=\"./" + sub_dir + "/call_graph.html\">here</a></td>\n")
+        ofile.write("  </tr>\n")
+        ofile.write("  <tr>\n")
+        ofile.write("    <td>Inverted call graph</td>\n")
+        ofile.write("    <td align=\"center\"><a href=\"./" + sub_dir + "/inverted_call_graph.html\">here</a></td>\n")
+        ofile.write("  </tr>\n")
+        ofile.write("  <tr>\n")
+        ofile.write("    <td>Tokens propagation graph</td>\n")
+        ofile.write("    <td align=\"center\"><a href=\"./" + sub_dir + "/tokens_propagation_graph.html\">here</a></td>\n")
+        ofile.write("  </tr>\n")
+        ofile.write("  <tr>\n")
+        ofile.write("    <td>Map from rules to their application sites (in JSON format)</td>\n")
+        ofile.write("    <td align=\"center\"><a href=\"./" + sub_dir +
+                    "/map_from_functions_to_rule_application_sites.json\">here</a></td>\n")
+        ofile.write("  </tr>\n")
+        ofile.write("    <td>Instrumentation properties</td>\n")
+        ofile.write("    <td align=\"center\"><a href=\"./" +
+                    os.path.join(sub_dir,os.path.splitext(os.path.basename(instrumentation_props_fname))[0] + ".html") +
+                    "\">here</a></td>\n")
+        ofile.write("  </tr>\n")
+        ofile.write("</table>\n")
 
-    ofile.write("<p></p>\n")
+        ofile.write("<p></p>\n")
 
-    with open(os.path.join(full_sub_dir,"instrumented_goto_program.json")) as json_file:
-        instrumented_goto_program = json.load(json_file)
-    if instrumented_goto_program is not None:
+        with open(instrumented_goto_program_fname) as json_file:
+            instrumented_goto_program = json.load(json_file)
         ofile.write("<table>\n"
                     "<caption>Instrumented GOTO program</caption>\n"
                     "  <tr>\n"
@@ -250,7 +251,9 @@ def build_HTML_interface_to_slicing_tasks(root_dir,sub_dir,ofile):
         ofile.write("  </tr>\n")
         ofile.write("</table>\n")
         return True
-    return False
+    else:
+        ofile.write("<p>The program is safe. No tainted data may reach any sink.</p>\n")
+        return False
 
 def build_HTML_interface_to_the_slicer(root_dir,sub_dir,ofile):
     full_sub_dir=os.path.join(root_dir, sub_dir)
@@ -686,12 +689,9 @@ def build_HTML_interface_to_results_and_statistics(
         #######################################################################################################
 
 
-        ofile.write("<h3>Phase 2: Application of program slicing</h3>\n")
-        ofile.write("<p>It is performed by the tool 'goto-instrument'.</p>\n")
-
-        if not has_instrumented_program:
-            ofile.write("<p>The instrumented GOTO binary was not generated from the previous stage.</p>\n")
-        else:
+        if has_instrumented_program:
+            ofile.write("<h3>Phase 2: Application of program slicing</h3>\n")
+            ofile.write("<p>It is performed by the tool 'goto-instrument'.</p>\n")
             build_HTML_interface_to_the_slicer(
                 cmdline.results_dir,
                 "program_slicing",
@@ -704,12 +704,9 @@ def build_HTML_interface_to_results_and_statistics(
         #######################################################################################################
 
 
-        ofile.write("<h3>Phase 3: Search for error-traces in the sliced program</h3>\n")
-        ofile.write("<p>It is currently performed only by the tool 'JBMC'.</p>\n")
-
-        if not has_instrumented_program:
-            ofile.write("<p>There are no error traces, because there is no GOTO binary available from previous stage(s).</p>\n")
-        else:
+        if has_instrumented_program:
+            ofile.write("<h3>Phase 3: Search for error-traces in the sliced program</h3>\n")
+            ofile.write("<p>It is currently performed only by the tool 'JBMC'.</p>\n")
             build_HTML_interface_to_error_traces(
                 cmdline.results_dir,
                 "search_for_error_traces",

driver/run.py

@@ -294,32 +294,29 @@ def run_scan(cmdline):
 
     instrumented_program_json_path = os.path.join(cmdline.results_dir, "program_slicing", "instrumented_goto_program.json")
 
-    if not os.path.exists(instrumented_program_json_path):
-        print("Instrumented program .json don't exist at " + instrumented_program_json_path + ".")
-        return
-
-    with open(instrumented_program_json_path, "r") as f:
-        instrumented_program = json.load(f)
-    # Instrumented GOTO binaries are specified relative to the results directory
-    # (the security-analyser working directory). Rewrite them with absolute paths:
-    instrumented_program["goto_binary_file"] = os.path.abspath(os.path.join(cmdline.results_dir, instrumented_program["goto_binary_file"]))
-
-    print("Starting program slicing.")
-    prof["program_slicing"] = analyser.run_program_slicing(
-        instrumented_program,
-        os.path.abspath(os.path.join(cmdline.results_dir,"program_slicing")),
-        cmdline.timeout,
-        cmdline.verbosity,
-        cmdline.dump_html_instrumented_goto
-        )
-
-    print("Starting the search for error traces.")
-    prof["search_for_error_traces"] = analyser.run_search_for_error_traces(
-        os.path.abspath(os.path.join(cmdline.results_dir,"program_slicing","sliced_goto_program.json")),
-        os.path.abspath(os.path.join(cmdline.results_dir,"search_for_error_traces")),
-        cmdline.timeout,
-        cmdline.verbosity
-        )
+    if os.path.exists(instrumented_program_json_path):
+        with open(instrumented_program_json_path, "r") as f:
+            instrumented_program = json.load(f)
+        # Instrumented GOTO binaries are specified relative to the results directory
+        # (the security-analyser working directory). Rewrite them with absolute paths:
+        instrumented_program["goto_binary_file"] = os.path.abspath(os.path.join(cmdline.results_dir, instrumented_program["goto_binary_file"]))
+
+        print("Starting program slicing.")
+        prof["program_slicing"] = analyser.run_program_slicing(
+            instrumented_program,
+            os.path.abspath(os.path.join(cmdline.results_dir,"program_slicing")),
+            cmdline.timeout,
+            cmdline.verbosity,
+            cmdline.dump_html_instrumented_goto
+            )
+
+        print("Starting the search for error traces.")
+        prof["search_for_error_traces"] = analyser.run_search_for_error_traces(
+            os.path.abspath(os.path.join(cmdline.results_dir,"program_slicing","sliced_goto_program.json")),
+            os.path.abspath(os.path.join(cmdline.results_dir,"search_for_error_traces")),
+            cmdline.timeout,
+            cmdline.verbosity
+            )
 
     print("Building performance plots.")
     prof_plots = {}

regression/end_to_end/driver.py

@@ -11,8 +11,9 @@
 
 class ErrorTraces:
 
-    def __init__(self, traces, cmdline):
+    def __init__(self, traces, pretty_cmdline, cmdline):
         self.traces = traces
+        self.pretty_cmdline = pretty_cmdline
         self.cmdline = cmdline
 
     def count_traces(self):
@@ -64,12 +65,21 @@ def trace_goes_through(
                         pass
         return False
 
+    def get_cmdline(self):
+        return self.cmdline
+
+    def get_results_dir(self):
+        return self.cmdline[self.cmdline.index("-R") + 1]
+
+    def get_temp_dir(self):
+        return self.cmdline[self.cmdline.index("-T") + 1]
+
     def __enter__(self):
         return self
 
     def __exit__(self, exc_type, exc_value, traceback):
         if exc_value is not None:
-            print("Failure may relate to command: ", self.cmdline, file=sys.stderr)
+            print("Failure may relate to command: ", self.pretty_cmdline, file=sys.stderr)
 
 
 def pretty_print_commandline(cmdline):
@@ -95,12 +105,17 @@ def run_security_driver_script(
     cmdline = ["python3", pipeline_driver_path]
     cmdline.extend(extra_commandline)
 
-    executable_runner = ExecutableRunner(cmdline)
-    (stdout, stderr, ret) = executable_runner.run()
-    if ret != 0:
-        raise Exception(
-            "Failed running \"%s\":\nstdout:\n\n%s\nstderr\n\n%s" % \
-            (pretty_print_commandline(cmdline), stdout, stderr))
+    analyzer_home = utils.get_security_analyzer_home()
+    if analyzer_home is None:
+        raise Exception("Set SECURITY_SCANNER_HOME to a path containing the 'security-analyzer' binary")
+
+    with utils.working_dir(analyzer_home):
+        executable_runner = ExecutableRunner(cmdline)
+        (stdout, stderr, ret) = executable_runner.run()
+        if ret != 0:
+            raise Exception(
+                "Failed running \"%s\":\nstdout:\n\n%s\nstderr\n\n%s" % \
+                (pretty_print_commandline(cmdline), stdout, stderr))
 
 
 def run_security_analyser_pipeline(
@@ -109,14 +124,12 @@ def run_security_analyser_pipeline(
     base_path,
     entry_point,
     load_strategy=LoadStrategy.conventional_lazy_loading,
-    extra_args=None):
+    extra_args=None,
+    keep_results=False):
 
     if extra_args is None:
         extra_args = []
 
-    analyzer_home = utils.get_security_analyzer_home()
-    if analyzer_home is None:
-        raise Exception("Set SECURITY_SCANNER_HOME to a path containing the 'security-analyzer' binary")
     absolute_binary_path = \
         os.path.join(base_path, relative_binary_path)
     absolute_rules_path = \
@@ -147,18 +160,20 @@ def run_security_analyser_pipeline(
         # optimisation matches:
         cmdline.append("--verify-csvsa-sparse-domains")
 
-    with utils.working_dir(analyzer_home), \
-         utils.temp_dir_deleter(results_dir), \
-         utils.temp_dir_deleter(temporary_dir), \
-         utils.temp_dir_deleter(common_dir):
+    with utils.temp_dir_deleter(results_dir, keep_results), \
+         utils.temp_dir_deleter(temporary_dir, keep_results), \
+         utils.temp_dir_deleter(common_dir, keep_results):
 
         run_security_driver_script(cmdline)
 
         trace_list = \
             os.path.join(
                 results_dir, entry_point, "search_for_error_traces", "error_traces.json")
-        with open(trace_list, "r") as f:
-            traces = json.load(f)
+        if os.path.isfile(trace_list):
+            with open(trace_list, "r") as f:
+                traces = json.load(f)
+        else:
+            traces = []
 
         # Replace all trace JSON file references with the trace itself:
         for i in range(len(traces)):
@@ -176,4 +191,4 @@ def run_security_analyser_pipeline(
             print("Test \"%s\" kept results (%s) and temporary directory (%s)" %
                   (pretty_cmdline, results_dir, temporary_dir))
 
-        return ErrorTraces(traces, pretty_cmdline)
+        return ErrorTraces(traces, pretty_cmdline, cmdline)

regression/end_to_end/early_check/build.xml

@@ -0,0 +1,17 @@
+<project name="Main" basedir="." default="compile">
+
+  <property name="root.dir"      value="./"/>
+  <property name="src.dir"       value="${root.dir}/src"/>
+  <property name="classes.dir"   value="${root.dir}/build"/>
+
+  <target name="compile">
+    <antcall target="clean" />
+    <mkdir dir="${classes.dir}"/>
+    <javac srcdir="${src.dir}" destdir="${classes.dir}" includeantruntime="false" debug="on" />
+  </target>
+
+  <target name="clean">
+    <delete dir="${classes.dir}"/>
+  </target>
+
+</project>

regression/end_to_end/early_check/rules.json

@@ -0,0 +1,32 @@
+{
+  "rules":
+    [
+      {
+        "comment": "Obtaining a tainted string.",
+        "class": "Main",
+        "method": "source:()Ljava/lang/String;",
+        "result": {
+          "location": "return_value",
+          "taint": "Tainted string"
+        }
+      },
+      {
+        "comment": "Returning a sanitized string.",
+        "class": "Main",
+        "method": "sanitize:(Ljava/lang/String;)Ljava/lang/String;",
+        "sanitizes": {
+          "location": "returns",
+          "taint": "Clean string"
+        }
+      },
+      {
+        "comment": "Writing a potentially tainted data into the sink",
+        "class": "Main",
+        "method": "sink:(Ljava/lang/String;)V",
+        "sinkTarget": {
+          "location": "arg0",
+          "vulnerability": "Tainted string"
+        }
+      }
+    ]
+}

regression/end_to_end/early_check/src/Main.java

@@ -0,0 +1,42 @@
+public class Main {
+
+  public static String source() {
+    return "Tainted string";
+  }
+
+  public static void sink(String s) {
+  }
+
+  public static String sanitize(String s) {
+    return "sanitised string";
+  }
+
+  // Test cases
+
+  public static void catch_impossible_taint_flow_01() {
+    String s = "no taint";
+    sink(s);
+  }
+
+  public static void catch_impossible_taint_flow_02() {
+    String s = source();
+    s = sanitize(s);
+    sink(s);
+  }
+
+  public static void miss_impossible_taint_flow_01() {
+    String s = source();
+    int i = 0;
+    if (i == 1)
+        sink(s);
+  }
+
+  public static void miss_impossible_taint_flow_02() {
+    String s = source();
+    int i = 20;
+    if (i == 20)
+        s = sanitize(s);
+    sink(s);
+  }
+}
+