[WIP] Fix false-negatives and false-positives

example/november_2017_evaluation/false_negatives/source_nested_in_sink.py

@@ -0,0 +1,9 @@
+"""
+This is very similar to one of the false-negatives found in our November 16th/17th 2017 evaluation
+http://pyt.readthedocs.io/en/latest/past_evaluations.html#november-16-17th-2017
+"""
+
+
+@app.route('/login', methods=['GET', 'POST'])
+def login():
+    return redirect(request.args.get("next"))

pyt/expr_visitor.py

@@ -1,4 +1,5 @@
 import ast
+import itertools
 
 from .alias_helper import (
     handle_aliases_in_calls
@@ -9,6 +10,7 @@
 )
 from .expr_visitor_helper import (
     BUILTINS,
+    CALL_IDENTIFIER,
     CFG,
     return_connection_handler,
     SavedVariable
@@ -28,7 +30,7 @@
 )
 from .right_hand_side_visitor import RHSVisitor
 from .stmt_visitor import StmtVisitor
-from .stmt_visitor_helper import CALL_IDENTIFIER
+from .vars_visitor import VarsVisitor
 
 
 class ExprVisitor(StmtVisitor):
@@ -537,6 +539,108 @@ def process_function(self, call_node, definition):
 
         return self.nodes[-1]
 
+    def add_blackbox_or_builtin_call(self, node, blackbox):
+        """Processes a blackbox or builtin function when it is called.
+        Nothing gets assigned to ret_func_foo in the builtin/blackbox case.
+
+        Increments self.function_call_index each time it is called, we can refer to it as N in the comments.
+        Create e.g. ¤call_1 = ret_func_foo RestoreNode.
+
+        Create e.g. temp_N_def_arg1 = call_arg1_label_visitor.result for each argument.
+        Visit the arguments if they're calls. (save_def_args_in_temp)
+
+        I do not think I care about this one actually -- Create e.g. def_arg1 = temp_N_def_arg1 for each argument.
+        (create_local_scope_from_def_args)
+
+        Add RestoreNode to the end of the Nodes.
+
+        Args:
+            node(ast.Call) : The node that calls the definition.
+            blackbox(bool): Whether or not it is a builtin or blackbox call.
+        Returns:
+            call_node(BBorBInode): The call node.
+        """
+        self.function_call_index += 1
+        saved_function_call_index = self.function_call_index
+        self.undecided = False
+
+        call_label = LabelVisitor()
+        call_label.visit(node)
+
+        index = call_label.result.find('(')
+
+        # Create e.g. ¤call_1 = ret_func_foo
+        LHS = CALL_IDENTIFIER + 'call_' + str(saved_function_call_index)
+        RHS = 'ret_' + call_label.result[:index] + '('
+
+        call_node = BBorBInode(
+            label='',
+            left_hand_side=LHS,
+            right_hand_side_variables=[],
+            line_number=node.lineno,
+            path=self.filenames[-1],
+            func_name=call_label.result[:index]
+        )
+        visual_args = list()
+        rhs_vars = list()
+        last_return_value_of_nested_call = None
+
+        for arg in itertools.chain(node.args, node.keywords):
+            if isinstance(arg, ast.Call):
+                return_value_of_nested_call = self.visit(arg)
+
+                if last_return_value_of_nested_call:
+                    # connect inner to other_inner in e.g.
+                    # `scrypt.outer(scrypt.inner(image_name), scrypt.other_inner(image_name))`
+                    # I should probably loop to the inner most call of other_inner here.
+                    try:
+                        last_return_value_of_nested_call.connect(return_value_of_nested_call.first_node)
+                    except AttributeError:
+                        last_return_value_of_nested_call.connect(return_value_of_nested_call)
+                else:
+                    # I should only set this once per loop, inner in e.g.
+                    # `scrypt.outer(scrypt.inner(image_name), scrypt.other_inner(image_name))`
+                    # (inner_most_call is used when predecessor is a ControlFlowNode in connect_control_flow_node)
+                    call_node.inner_most_call = return_value_of_nested_call
+                last_return_value_of_nested_call = return_value_of_nested_call
+
+                visual_args.append(return_value_of_nested_call.left_hand_side)
+                rhs_vars.append(return_value_of_nested_call.left_hand_side)
+            else:
+                label = LabelVisitor()
+                label.visit(arg)
+                visual_args.append(label.result)
+
+                vv = VarsVisitor()
+                vv.visit(arg)
+                rhs_vars.extend(vv.result)
+        if last_return_value_of_nested_call:
+            # connect other_inner to outer in e.g.
+            # `scrypt.outer(scrypt.inner(image_name), scrypt.other_inner(image_name))`
+            last_return_value_of_nested_call.connect(call_node)
+
+        if len(visual_args) > 0:
+            for arg in visual_args:
+                RHS = RHS + arg + ", "
+            # Replace the last ", " with a )
+            RHS = RHS[:len(RHS) - 2] + ')'
+        else:
+            RHS = RHS + ')'
+        call_node.label = LHS + " = " + RHS
+
+        # .args is only used in get_sink_args
+        # We make a new list because right_hand_side_variables is extended in assignment_call_node
+        call_node.right_hand_side_variables = rhs_vars
+        call_node.args = list(rhs_vars)
+
+        if blackbox:
+            self.blackbox_assignments.add(call_node)
+
+        self.connect_if_allowed(self.nodes[-1], call_node)
+        self.nodes.append(call_node)
+
+        return call_node
+
     def visit_Call(self, node):
         _id = get_call_names_as_string(node.func)
         local_definitions = self.module_definitions_stack[-1]

pyt/expr_visitor_helper.py

@@ -3,13 +3,7 @@
 from .node_types import ConnectToExitNode
 
 
-SavedVariable = namedtuple(
-    'SavedVariable',
-    (
-        'LHS',
-        'RHS'
-    )
-)
+CALL_IDENTIFIER = '¤'
 BUILTINS = (
     'get',
     'Flask',
@@ -31,6 +25,13 @@
     'flash',
     'jsonify'
 )
+SavedVariable = namedtuple(
+    'SavedVariable',
+    (
+        'LHS',
+        'RHS'
+    )
+)
 
 
 class CFG():

pyt/stmt_visitor.py

@@ -1,5 +1,4 @@
 import ast
-import itertools
 import os.path
 
 from .alias_helper import (
@@ -36,7 +35,6 @@
 from .project_handler import get_directory_modules
 from .right_hand_side_visitor import RHSVisitor
 from .stmt_visitor_helper import (
-    CALL_IDENTIFIER,
     ConnectStatements,
     connect_nodes,
     extract_left_hand_side,
@@ -528,109 +526,6 @@ def visit_While(self, node):
 
         return self.loop_node_skeleton(test, node)
 
-    def add_blackbox_or_builtin_call(self, node, blackbox):
-        """Processes a blackbox or builtin function when it is called.
-        Nothing gets assigned to ret_func_foo in the builtin/blackbox case.
-
-        Increments self.function_call_index each time it is called, we can refer to it as N in the comments.
-        Create e.g. ¤call_1 = ret_func_foo RestoreNode.
-
-        Create e.g. temp_N_def_arg1 = call_arg1_label_visitor.result for each argument.
-        Visit the arguments if they're calls. (save_def_args_in_temp)
-
-        I do not think I care about this one actually -- Create e.g. def_arg1 = temp_N_def_arg1 for each argument.
-        (create_local_scope_from_def_args)
-
-        Add RestoreNode to the end of the Nodes.
-
-        Args:
-            node(ast.Call) : The node that calls the definition.
-            blackbox(bool): Whether or not it is a builtin or blackbox call.
-        Returns:
-            call_node(BBorBInode): The call node.
-        """
-        self.function_call_index += 1
-        saved_function_call_index = self.function_call_index
-        self.undecided = False
-
-        call_label = LabelVisitor()
-        call_label.visit(node)
-
-        index = call_label.result.find('(')
-
-        # Create e.g. ¤call_1 = ret_func_foo
-        LHS = CALL_IDENTIFIER + 'call_' + str(saved_function_call_index)
-        RHS = 'ret_' + call_label.result[:index] + '('
-
-        call_node = BBorBInode(
-            label='',
-            left_hand_side=LHS,
-            right_hand_side_variables=[],
-            line_number=node.lineno,
-            path=self.filenames[-1],
-            func_name=call_label.result[:index]
-        )
-        visual_args = list()
-        rhs_vars = list()
-        last_return_value_of_nested_call = None
-
-        for arg in itertools.chain(node.args, node.keywords):
-            if isinstance(arg, ast.Call):
-                return_value_of_nested_call = self.visit(arg)
-
-                if last_return_value_of_nested_call:
-                    # connect inner to other_inner in e.g.
-                    # `scrypt.outer(scrypt.inner(image_name), scrypt.other_inner(image_name))`
-                    # I should probably loop to the inner most call of other_inner here.
-                    try:
-                        last_return_value_of_nested_call.connect(return_value_of_nested_call.first_node)
-                    except AttributeError:
-                        last_return_value_of_nested_call.connect(return_value_of_nested_call)
-                else:
-                    # I should only set this once per loop, inner in e.g.
-                    # `scrypt.outer(scrypt.inner(image_name), scrypt.other_inner(image_name))`
-                    # (inner_most_call is used when predecessor is a ControlFlowNode in connect_control_flow_node)
-                    call_node.inner_most_call = return_value_of_nested_call
-                last_return_value_of_nested_call = return_value_of_nested_call
-
-                visual_args.append(return_value_of_nested_call.left_hand_side)
-                rhs_vars.append(return_value_of_nested_call.left_hand_side)
-            else:
-                label = LabelVisitor()
-                label.visit(arg)
-                visual_args.append(label.result)
-
-                vv = VarsVisitor()
-                vv.visit(arg)
-                rhs_vars.extend(vv.result)
-        if last_return_value_of_nested_call:
-            # connect other_inner to outer in e.g.
-            # `scrypt.outer(scrypt.inner(image_name), scrypt.other_inner(image_name))`
-            last_return_value_of_nested_call.connect(call_node)
-
-        if len(visual_args) > 0:
-            for arg in visual_args:
-                RHS = RHS + arg + ", "
-            # Replace the last ", " with a )
-            RHS = RHS[:len(RHS) - 2] + ')'
-        else:
-            RHS = RHS + ')'
-        call_node.label = LHS + " = " + RHS
-
-        call_node.right_hand_side_variables = rhs_vars
-        # Used in get_sink_args, not using right_hand_side_variables because it is extended in assignment_call_node
-        rhs_visitor = RHSVisitor()
-        rhs_visitor.visit(node)
-        call_node.args = rhs_visitor.result
-
-        if blackbox:
-            self.blackbox_assignments.add(call_node)
-
-        self.connect_if_allowed(self.nodes[-1], call_node)
-        self.nodes.append(call_node)
-
-        return call_node
-
     def visit_With(self, node):
         label_visitor = LabelVisitor()
         label_visitor.visit(node.items[0])

pyt/stmt_visitor_helper.py

@@ -2,6 +2,7 @@
 import random
 from collections import namedtuple
 
+from .expr_visitor_helper import CALL_IDENTIFIER
 from .node_types import (
     AssignmentCallNode,
     BBorBInode,
@@ -11,7 +12,6 @@
 )
 
 
-CALL_IDENTIFIER = '¤'
 ConnectStatements = namedtuple(
     'ConnectStatements',
     (

tests/vulnerabilities_test.py

@@ -12,17 +12,13 @@
     UImode,
     VulnerabilityFiles
 )
-from pyt.constraint_table import(
-    constraint_table,
-    initialize_constraint_table
-)
+from pyt.constraint_table import initialize_constraint_table
 from pyt.fixed_point import analyse
 from pyt.framework_adaptor import FrameworkAdaptor
-from pyt.framework_helper import(
+from pyt.framework_helper import (
     is_django_view_function,
     is_flask_route_function
 )
-from pyt.lattice import Lattice
 from pyt.node_types import Node
 from pyt.reaching_definitions_taint import ReachingDefinitionsTaintAnalysis
 
@@ -99,17 +95,15 @@ def test_find_triggers(self):
         l = vulnerabilities.find_triggers(XSS1.nodes, trigger_words)
         self.assert_length(l, expected_length=1)
 
-
     def test_find_sanitiser_nodes(self):
         cfg_node = Node(None, None, line_number=None, path=None)
-        sanitiser_tuple  = vulnerabilities.Sanitiser('escape', cfg_node)
+        sanitiser_tuple = vulnerabilities.Sanitiser('escape', cfg_node)
         sanitiser = 'escape'
 
         result = list(vulnerabilities.find_sanitiser_nodes(sanitiser, [sanitiser_tuple]))
         self.assert_length(result, expected_length=1)
         self.assertEqual(result[0], cfg_node)
 
-
     def test_build_sanitiser_node_dict(self):
         self.cfg_create_from_file('example/vulnerable_code/XSS_sanitised.py')
         cfg_list = [self.cfg]
@@ -118,7 +112,7 @@ def test_build_sanitiser_node_dict(self):
 
         cfg = cfg_list[1]
 
-        cfg_node = Node(None, None,  line_number=None, path=None)
+        cfg_node = Node(None, None, line_number=None, path=None)
         sinks_in_file = [vulnerabilities.TriggerNode('replace', ['escape'], cfg_node)]
 
         sanitiser_dict = vulnerabilities.build_sanitiser_node_dict(cfg, sinks_in_file)
@@ -146,7 +140,6 @@ def run_analysis(self, path):
             )
         )
 
-
     def test_find_vulnerabilities_assign_other_var(self):
         vulnerability_log = self.run_analysis('example/vulnerable_code/XSS_assign_to_other_var.py')
         self.assert_length(vulnerability_log.vulnerabilities, expected_length=1)
@@ -512,6 +505,24 @@ def test_XSS_variable_multiple_assign_result(self):
 
         self.assertTrue(self.string_compare_alpha(vulnerability_description, EXPECTED_VULNERABILITY_DESCRIPTION))
 
+    def test_source_nested_in_sink(self):
+        vulnerability_log = self.run_analysis('example/november_2017_evaluation/false_negatives/source_nested_in_sink.py')
+        self.assert_length(vulnerability_log.vulnerabilities, expected_length=1)
+        vulnerability_description = str(vulnerability_log.vulnerabilities[0])
+
+        EXPECTED_VULNERABILITY_DESCRIPTION = """
+            File: example/november_2017_evaluation/false_negatives/source_nested_in_sink.py
+             > User input at line 8, trigger word "request.args.get(":
+                 ¤call_2 = ret_request.args.get('next')
+            Reassigned in:
+                File: example/november_2017_evaluation/false_negatives/source_nested_in_sink.py
+                 > Line 8: ret_login = ¤call_1
+            File: example/november_2017_evaluation/false_negatives/source_nested_in_sink.py
+             > reaches line 8, trigger word "redirect(":
+                ¤call_1 = ret_redirect(¤call_2)
+        """
+        self.assertTrue(self.string_compare_alpha(vulnerability_description, EXPECTED_VULNERABILITY_DESCRIPTION))
+
 
 class EngineDjangoTest(BaseTestCase):
     def run_empty(self):