Update of Python driver scripts: It provides driving of the whole analysis pipeline up to recognition of error traces by CBMC, including presentation of results in HTML format (not complete yet though - presentation of results of last 2 phases is not clear/implemented yet).

Author: marek-trtik

security-scanner/analyser.py

@@ -1,27 +1,32 @@
 import os
 import time
 import json
+import shutil
 
 
 def __get_my_dir(): return os.path.dirname(os.path.realpath(__file__))
 
 
-def get_possible_analyser_binary_pathnames():
+def get_possible_directories_of_analysers():
+    return ["./", "../src/driver", "../cbmc/src/cbmc", "../cbmc/src/symex"]
+
+
+def get_possible_analyser_binary_pathnames(analyser_name):
     result = []
-    for rdir in ["./", "../src/driver"]:
-        fname = os.path.abspath(os.path.join(__get_my_dir(), rdir, "security-analyzer"))
+    for rdir in get_possible_directories_of_analysers():
+        fname = os.path.abspath(os.path.join(__get_my_dir(), rdir, analyser_name))
         result.append(fname)
     return result
 
 
-def get_analyser_binary_pathname():
-    for fname in get_possible_analyser_binary_pathnames():
+def get_analyser_binary_pathname(analyser_name):
+    for fname in get_possible_analyser_binary_pathnames(analyser_name):
         if os.path.exists(fname) and os.path.isfile(fname):
             return fname
     return None
 
 
-def run_analyser(
+def run_secutiry_analyser(
         root_jar,
         transition_rules_file_pathname,
         timeout,
@@ -86,12 +91,12 @@ def run_analyser(
     old_cwd = os.getcwd()
     os.chdir(results_dir)
     command = (
-        get_analyser_binary_pathname() + " "
+        get_analyser_binary_pathname("security-analyzer") + " "
         + "'./" + os.path.relpath(root_jar,results_dir) + "' "
         "--security-scanner '" + os.path.relpath(root_config_json_fname,results_dir) + "' "
         )
     prof["calling_goto_analyser"] = { "duration": time.time() }
-    print("Invoking 'security_analyser' ...")
+    print("Invoking 'security-analyser' ...")
     if verbosity >= 9:
         print("CWD: " + results_dir)
         print("CMD: " + command)
@@ -101,3 +106,99 @@ def run_analyser(
     prof["duration"] = time.time() - prof["duration"]
 
     return prof
+
+
+def run_program_slicing(
+        json_cfg_fname,
+        results_dir,
+        timeout,
+        verbosity
+        ):
+    prof = { "duration": time.time() }
+
+    # TODO: once the slicer in "goto-instrument" tool is available, then provide proper implementation here!
+    #       Until then we assume the slicing removes nothing from the instrumented program.
+    shutil.copyfile(json_cfg_fname, os.path.join(os.path.dirname(json_cfg_fname),"sliced_goto_programs.json"))
+
+    prof["duration"] = time.time() - prof["duration"]
+    return prof
+
+
+def run_search_for_error_traces(
+        json_cfg_fname,
+        results_dir,
+        timeout,
+        verbosity
+        ):
+    prof = { "duration": time.time() }
+    if not os.path.exists(results_dir):
+        os.makedirs(results_dir)
+
+    trace_id = 0
+    traces_list = []
+
+    with open(json_cfg_fname) as data_file:
+        cfgs = json.load(data_file)
+    for cfg in cfgs:
+        old_cwd = os.getcwd()
+        os.chdir(results_dir)
+        command = (
+            get_analyser_binary_pathname("cbmc") + " " +
+            cfg["goto_binary_file"] + " " +
+            " --trace --lazy-methods --json-ui --unwind 10 " +
+            "--verbosity " + str(1) + " "
+            "> \"cbmc_results.json\""
+            )
+        prof["calling_cbmc"] = { "duration": time.time() }
+        print("Invoking 'cbmc' ...")
+        if verbosity >= 9:
+            print("CWD: " + results_dir)
+            print("CMD: " + command)
+        os.system(command)
+        prof["calling_cbmc"]["duration"] = time.time() - prof["calling_cbmc"]["duration"]
+        os.chdir(old_cwd)
+
+        with open(os.path.join(results_dir,"cbmc_results.json")) as data_file:
+            cbmc_results = json.load(data_file)
+        for outer_element in cbmc_results:
+            if "result" in outer_element:
+                for inner_element in outer_element["result"]:
+                    if "trace" in inner_element and len(inner_element["trace"]) > 0:
+                        trace=inner_element["trace"]
+                        last_rec=trace[len(trace)-1]
+                        if "sourceLocation" in last_rec:
+                            loc=last_rec["sourceLocation"]
+                            for sink in cfg["sinks"]:
+                                if loc["file"] == sink["file"] and \
+                                   loc["function"] == sink["function"] and \
+                                   loc["line"] == str(sink["line"]):
+                                    trace_fname = os.path.join(results_dir, "error_trace_" + str(trace_id) + ".json")
+                                    with open(trace_fname,"w") as trace_file:
+                                        trace_file.write(json.dumps(trace,sort_keys=False,indent=4))
+                                    trace_id += 1
+
+                                    found = False
+                                    for rec in traces_list:
+                                        if rec["file"] == sink["file"] and \
+                                           rec["function"] == sink["function"] and \
+                                           rec["line"] == sink["line"] and \
+                                           rec["goto_binary_file"] == cfg["goto_binary_file"]:
+                                            rec["error_traces"].append(trace_fname)
+                                    if found == False:
+                                        rec=sink
+                                        rec["goto_binary_file"]=cfg["goto_binary_file"]
+                                        rec["error_traces"]=[trace_fname]
+                                        traces_list.append(rec)
+                                    break
+
+    with open(os.path.join(results_dir, "error_traces.json"), "w") as traces_file:
+        traces_file.write(json.dumps(traces_list, sort_keys=True, indent=4))
+
+    prof["duration"] = time.time() - prof["duration"]
+    return prof
+
+
+
+
+if __name__ == "__main__":
+    pass