Merge pull request github#4710 from yoff/python-dataflow-variable-capture

Python: Dataflow, variable capture

Author: RasmusWL

python/ql/test/experimental/dataflow/coverage/argumentPassing.py

@@ -112,7 +112,7 @@ def with_multiple_kw_args(a, b, c):
     SINK3(c)
 
 
-@expects(9)
+@expects(12)
 def test_multiple_kw_args():
     with_multiple_kw_args(b=arg2, c=arg3, a=arg1)
     with_multiple_kw_args(arg1, *(arg2,), arg3)

python/ql/test/experimental/dataflow/coverage/classesCallGraph.expected

@@ -1,8 +1,5 @@
 | classes.py:45:16:45:35 | ControlFlowNode for Attribute() | classes.py:45:16:45:35 | ControlFlowNode for Attribute() |
 | classes.py:60:17:60:27 | [pre objCreate] ControlFlowNode for With_init() | classes.py:54:18:54:21 | ControlFlowNode for self |
-| classes.py:242:9:242:24 | ControlFlowNode for set() | classes.py:242:9:242:24 | ControlFlowNode for set() |
-| classes.py:247:9:247:30 | ControlFlowNode for frozenset() | classes.py:247:9:247:30 | ControlFlowNode for frozenset() |
-| classes.py:252:9:252:28 | ControlFlowNode for dict() | classes.py:252:9:252:28 | ControlFlowNode for dict() |
 | classes.py:565:5:565:16 | ControlFlowNode for with_getitem | classes.py:555:21:555:24 | ControlFlowNode for self |
 | classes.py:565:18:565:21 | ControlFlowNode for arg2 | classes.py:555:27:555:29 | ControlFlowNode for key |
 | classes.py:581:5:581:16 | ControlFlowNode for with_setitem | classes.py:570:21:570:24 | ControlFlowNode for self |

python/ql/test/experimental/dataflow/coverage/testlib.py renamed to python/ql/test/experimental/dataflow/testlib.py

@@ -50,6 +50,9 @@ def check_tests_valid(testFile):
 
 
 if __name__ == "__main__":
-    check_tests_valid("classes")
-    check_tests_valid("test")
-    check_tests_valid("argumentPassing")
+    check_tests_valid("coverage.classes")
+    check_tests_valid("coverage.test")
+    check_tests_valid("coverage.argumentPassing")
+    check_tests_valid("variable-capture.in")
+    check_tests_valid("variable-capture.nonlocal")
+    check_tests_valid("variable-capture.dict")

python/ql/test/experimental/dataflow/coverage/validTest.py renamed to python/ql/test/experimental/dataflow/validTest.py

@@ -0,0 +1,21 @@
+import python
+import semmle.python.dataflow.new.DataFlow
+import TestUtilities.InlineExpectationsTest
+import experimental.dataflow.testConfig
+
+class CaptureTest extends InlineExpectationsTest {
+  CaptureTest() { this = "CaptureTest" }
+
+  override string getARelevantTag() { result = "captured" }
+
+  override predicate hasActualResult(Location ___location, string element, string tag, string value) {
+    exists(DataFlow::Node source, DataFlow::Node sink |
+      exists(TestConfiguration cfg | cfg.hasFlow(source, sink))
+    |
+      ___location = sink.getLocation() and
+      tag = "captured" and
+      value = "" and
+      element = sink.toString()
+    )
+  }
+}

python/ql/test/experimental/dataflow/variable-capture/CaptureTest.expected

@@ -0,0 +1,93 @@
+# Here we test writing to a captured variable via a dictionary (see `out`).
+# We also test reading one captured variable and writing the value to another (see `through`).
+
+# All functions starting with "test_" should run and execute `print("OK")` exactly once.
+# This can be checked by running validTest.py.
+
+import sys
+import os
+
+sys.path.append(os.path.dirname(os.path.dirname((__file__))))
+from testlib import *
+
+# These are defined so that we can evaluate the test code.
+NONSOURCE = "not a source"
+SOURCE = "source"
+
+def is_source(x):
+    return x == "source" or x == b"source" or x == 42 or x == 42.0 or x == 42j
+
+
+def SINK(x):
+    if is_source(x):
+        print("OK")
+    else:
+        print("Unexpected flow", x)
+
+
+def SINK_F(x):
+    if is_source(x):
+        print("Unexpected flow", x)
+    else:
+        print("OK")
+
+
+def out():
+    sinkO1 = { "x": "" }
+    def captureOut1():
+        sinkO1["x"] = SOURCE
+    captureOut1()
+    SINK(sinkO1["x"]) #$ MISSING:captured
+
+    sinkO2 = { "x": "" }
+    def captureOut2():
+        def m():
+            sinkO2["x"] = SOURCE
+        m()
+    captureOut2()
+    SINK(sinkO2["x"]) #$ MISSING:captured
+
+    nonSink0 = { "x": "" }
+    def captureOut1NotCalled():
+        nonSink0["x"] = SOURCE
+    SINK_F(nonSink0["x"])
+
+    def captureOut2NotCalled():
+        def m():
+            nonSink0["x"] = SOURCE
+    captureOut2NotCalled()
+    SINK_F(nonSink0["x"])
+
+@expects(4)
+def test_out():
+    out()
+
+def through(tainted):
+    sinkO1 = { "x": "" }
+    def captureOut1():
+        sinkO1["x"] = tainted
+    captureOut1()
+    SINK(sinkO1["x"]) #$ MISSING:captured
+
+    sinkO2 = { "x": "" }
+    def captureOut2():
+        def m():
+            sinkO2["x"] = tainted
+        m()
+    captureOut2()
+    SINK(sinkO2["x"]) #$ MISSING:captured
+
+    nonSink0 = { "x": "" }
+    def captureOut1NotCalled():
+        nonSink0["x"] = tainted
+    SINK_F(nonSink0["x"])
+
+    def captureOut2NotCalled():
+        def m():
+            nonSink0["x"] = tainted
+    captureOut2NotCalled()
+    SINK_F(nonSink0["x"])
+
+@expects(4)
+def test_through():
+    through(SOURCE)

python/ql/test/experimental/dataflow/variable-capture/CaptureTest.ql

@@ -0,0 +1,96 @@
+# Here we test the case where a captured variable is being read.
+
+# All functions starting with "test_" should run and execute `print("OK")` exactly once.
+# This can be checked by running validTest.py.
+
+import sys
+import os
+
+sys.path.append(os.path.dirname(os.path.dirname((__file__))))
+from testlib import *
+
+# These are defined so that we can evaluate the test code.
+NONSOURCE = "not a source"
+SOURCE = "source"
+
+def is_source(x):
+    return x == "source" or x == b"source" or x == 42 or x == 42.0 or x == 42j
+
+
+def SINK(x):
+    if is_source(x):
+        print("OK")
+    else:
+        print("Unexpected flow", x)
+
+
+def SINK_F(x):
+    if is_source(x):
+        print("Unexpected flow", x)
+    else:
+        print("OK")
+
+# Capture the parameter of an outer function.
+def inParam(tainted):
+    def captureIn1():
+        sinkI1 = tainted
+        SINK(sinkI1) #$ MISSING:captured
+    captureIn1()
+
+    def captureIn2():
+        def m():
+            sinkI2 = tainted
+            SINK(sinkI2) #$ MISSING:captured
+        m()
+    captureIn2()
+
+    captureIn3 = lambda arg: SINK(tainted)
+    captureIn3("")
+
+    def captureIn1NotCalled():
+        nonSink0 = tainted
+        SINK_F(nonSink0)
+
+    def captureIn2NotCalled():
+        def m():
+            nonSink0 = tainted
+            SINK_F(nonSink0)
+    captureIn2NotCalled()
+
+@expects(3)
+def test_inParam():
+    inParam(SOURCE)
+
+# Capture the local variable of an outer function.
+def inLocal():
+    tainted = SOURCE
+
+    def captureIn1():
+        sinkI1 = tainted
+        SINK(sinkI1) #$ MISSING:captured
+    captureIn1()
+
+    def captureIn2():
+        def m():
+            sinkI2 = tainted
+            SINK(sinkI2) #$ MISSING:captured
+        m()
+    captureIn2()
+
+    captureIn3 = lambda arg: SINK(tainted)
+    captureIn3("")
+
+    def captureIn1NotCalled():
+        nonSink0 = tainted
+        SINK_F(nonSink0)
+
+    def captureIn2NotCalled():
+        def m():
+            nonSink0 = tainted
+            SINK_F(nonSink0)
+    captureIn2NotCalled()
+
+@expects(3)
+def test_inLocal():
+    inLocal()
+

python/ql/test/experimental/dataflow/variable-capture/dict.py

@@ -0,0 +1,101 @@
+# Here we test writing to a captured variable via the `nonlocal` keyword (see `out`).
+# We also test reading one captured variable and writing the value to another (see `through`).
+
+# All functions starting with "test_" should run and execute `print("OK")` exactly once.
+# This can be checked by running validTest.py.
+
+import sys
+import os
+
+sys.path.append(os.path.dirname(os.path.dirname((__file__))))
+from testlib import *
+
+# These are defined so that we can evaluate the test code.
+NONSOURCE = "not a source"
+SOURCE = "source"
+
+def is_source(x):
+    return x == "source" or x == b"source" or x == 42 or x == 42.0 or x == 42j
+
+
+def SINK(x):
+    if is_source(x):
+        print("OK")
+    else:
+        print("Unexpected flow", x)
+
+
+def SINK_F(x):
+    if is_source(x):
+        print("Unexpected flow", x)
+    else:
+        print("OK")
+
+
+def out():
+    sinkO1 = ""
+    def captureOut1():
+        nonlocal sinkO1
+        sinkO1 = SOURCE
+    captureOut1()
+    SINK(sinkO1) #$ MISSING:captured
+
+    sinkO2 = ""
+    def captureOut2():
+        def m():
+            nonlocal sinkO2
+            sinkO2 = SOURCE
+        m()
+    captureOut2()
+    SINK(sinkO2) #$ MISSING:captured
+
+    nonSink0 = ""
+    def captureOut1NotCalled():
+        nonlocal nonSink0
+        nonSink0 = SOURCE
+    SINK_F(nonSink0)
+
+    def captureOut2NotCalled():
+        def m():
+            nonlocal nonSink0
+            nonSink0 = SOURCE
+    captureOut2NotCalled()
+    SINK_F(nonSink0)
+
+@expects(4)
+def test_out():
+    out()
+
+def through(tainted):
+    sinkO1 = ""
+    def captureOut1():
+        nonlocal sinkO1
+        sinkO1 = tainted
+    captureOut1()
+    SINK(sinkO1) #$ MISSING:captured
+
+    sinkO2 = ""
+    def captureOut2():
+        def m():
+            nonlocal sinkO2
+            sinkO2 = tainted
+        m()
+    captureOut2()
+    SINK(sinkO2) #$ MISSING:captured
+
+    nonSink0 = ""
+    def captureOut1NotCalled():
+        nonlocal nonSink0
+        nonSink0 = tainted
+    SINK_F(nonSink0)
+
+    def captureOut2NotCalled():
+        def m():
+            nonlocal nonSink0
+            nonSink0 = tainted
+    captureOut2NotCalled()
+    SINK_F(nonSink0)
+
+@expects(4)
+def test_through():
+    through(SOURCE)