Python: Move dataflow TestUtil to importable location

python/ql/test/TestUtilities/dataflow/DataflowQueryTest.qll

@@ -0,0 +1,117 @@
+import python
+import semmle.python.dataflow.new.DataFlow
+import semmle.python.dataflow.new.TaintTracking
+import TestUtilities.InlineExpectationsTest
+private import semmle.python.dataflow.new.internal.PrintNode
+
+signature module QueryTestSig {
+  predicate isSink(DataFlow::Node sink);
+
+  predicate flowTo(DataFlow::Node sink);
+}
+
+module MakeQueryTest<QueryTestSig Impl> {
+  module DataFlowQueryTest implements TestSig {
+    string getARelevantTag() { result = "result" }
+
+    predicate hasActualResult(Location location, string element, string tag, string value) {
+      exists(DataFlow::Node sink | Impl::flowTo(sink) |
+        location = sink.getLocation() and
+        tag = "result" and
+        value = "BAD" and
+        element = sink.toString()
+      )
+    }
+
+    // We allow annotating any sink with `result=OK` to signal
+    // safe sinks.
+    // Sometimes a line contains both an alert and a safe sink.
+    // In this situation, the annotation form `OK(safe sink)`
+    // can be useful.
+    predicate hasOptionalResult(Location location, string element, string tag, string value) {
+      exists(DataFlow::Node sink | Impl::isSink(sink) |
+        location = sink.getLocation() and
+        tag = "result" and
+        value in ["OK", "OK(" + prettyNode(sink) + ")"] and
+        element = sink.toString()
+      )
+    }
+  }
+
+  import MakeTest<DataFlowQueryTest>
+
+  query predicate missingAnnotationOnSink(Location location, string error, string element) {
+    error = "ERROR, you should add `# $ MISSING: result=BAD` or `result=OK` annotation" and
+    exists(DataFlow::Node sink |
+      exists(sink.getLocation().getFile().getRelativePath()) and
+      Impl::isSink(sink) and
+      location = sink.getLocation() and
+      element = prettyExpr(sink.asExpr()) and
+      not Impl::flowTo(sink) and
+      not exists(FalseNegativeTestExpectation missingResult |
+        missingResult.getTag() = "result" and
+        missingResult.getValue() = "BAD" and
+        missingResult.getLocation().getFile() = location.getFile() and
+        missingResult.getLocation().getStartLine() = location.getStartLine()
+      ) and
+      not exists(GoodTestExpectation okResult |
+        okResult.getTag() = "result" and
+        okResult.getValue() in ["OK", "OK(" + prettyNode(sink) + ")"] and
+        okResult.getLocation().getFile() = location.getFile() and
+        okResult.getLocation().getStartLine() = location.getStartLine()
+      )
+    )
+  }
+}
+
+module FromDataFlowConfig<DataFlow::ConfigSig C> {
+  module Impl implements QueryTestSig {
+    predicate isSink(DataFlow::Node sink) { C::isSink(sink) }
+
+    predicate flowTo(DataFlow::Node sink) { DataFlow::Global<C>::flowTo(sink) }
+  }
+
+  import MakeQueryTest<Impl>
+}
+
+module FromDataFlowStateConfig<DataFlow::StateConfigSig C> {
+  module Impl implements QueryTestSig {
+    predicate isSink(DataFlow::Node sink) { C::isSink(sink) or C::isSink(sink, _) }
+
+    predicate flowTo(DataFlow::Node sink) { DataFlow::GlobalWithState<C>::flowTo(sink) }
+  }
+
+  import MakeQueryTest<Impl>
+}
+
+module FromTaintTrackingConfig<DataFlow::ConfigSig C> {
+  module Impl implements QueryTestSig {
+    predicate isSink(DataFlow::Node sink) { C::isSink(sink) }
+
+    predicate flowTo(DataFlow::Node sink) { TaintTracking::Global<C>::flowTo(sink) }
+  }
+
+  import MakeQueryTest<Impl>
+}
+
+module FromTaintTrackingStateConfig<DataFlow::StateConfigSig C> {
+  module Impl implements QueryTestSig {
+    predicate isSink(DataFlow::Node sink) { C::isSink(sink) or C::isSink(sink, _) }
+
+    predicate flowTo(DataFlow::Node sink) { TaintTracking::GlobalWithState<C>::flowTo(sink) }
+  }
+
+  import MakeQueryTest<Impl>
+}
+
+deprecated signature class LegacyConfiguration extends DataFlow::Configuration;
+
+deprecated module FromLegacyConfiguration<LegacyConfiguration C> {
+  module Impl implements QueryTestSig {
+    predicate isSink(DataFlow::Node sink) { any(C c).isSink(sink) or any(C c).isSink(sink, _) }
+
+    predicate flowTo(DataFlow::Node sink) { any(C c).hasFlowTo(sink) }
+  }
+
+  import MakeQueryTest<Impl>
+}

python/ql/test/TestUtilities/dataflow/FlowTest.qll

@@ -0,0 +1,39 @@
+import python
+import semmle.python.dataflow.new.DataFlow
+import TestUtilities.InlineExpectationsTest
+private import semmle.python.dataflow.new.internal.PrintNode
+
+signature module FlowTestSig {
+  string flowTag();
+
+  predicate relevantFlow(DataFlow::Node fromNode, DataFlow::Node toNode);
+}
+
+module MakeTestSig<FlowTestSig Impl> implements TestSig {
+  string getARelevantTag() { result = Impl::flowTag() }
+
+  predicate hasActualResult(Location location, string element, string tag, string value) {
+    exists(DataFlow::Node fromNode, DataFlow::Node toNode | Impl::relevantFlow(fromNode, toNode) |
+      location = toNode.getLocation() and
+      tag = Impl::flowTag() and
+      value =
+        "\"" + prettyNode(fromNode).replaceAll("\"", "'") + lineStr(fromNode, toNode) + " -> " +
+          prettyNode(toNode).replaceAll("\"", "'") + "\"" and
+      element = toNode.toString()
+    )
+  }
+
+  pragma[inline]
+  private string lineStr(DataFlow::Node fromNode, DataFlow::Node toNode) {
+    exists(int delta |
+      delta = fromNode.getLocation().getStartLine() - toNode.getLocation().getStartLine()
+    |
+      if delta = 0
+      then result = ""
+      else
+        if delta > 0
+        then result = ", l:+" + delta.toString()
+        else result = ", l:" + delta.toString()
+    )
+  }
+}

python/ql/test/TestUtilities/dataflow/LocalFlowStepTest.qll

@@ -0,0 +1,13 @@
+import python
+import semmle.python.dataflow.new.DataFlow
+import FlowTest
+
+module LocalFlowStepTest implements FlowTestSig {
+  string flowTag() { result = "step" }
+
+  predicate relevantFlow(DataFlow::Node fromNode, DataFlow::Node toNode) {
+    DataFlow::localFlowStep(fromNode, toNode)
+  }
+}
+
+import MakeTest<MakeTestSig<LocalFlowStepTest>>

python/ql/test/TestUtilities/dataflow/MaximalFlowTest.qll

@@ -0,0 +1,42 @@
+import python
+import semmle.python.dataflow.new.DataFlow
+private import semmle.python.dataflow.new.internal.DataFlowPrivate
+import FlowTest
+
+module MaximalFlowTest implements FlowTestSig {
+  string flowTag() { result = "flow" }
+
+  predicate relevantFlow(DataFlow::Node source, DataFlow::Node sink) {
+    source != sink and
+    MaximalFlows::flow(source, sink)
+  }
+}
+
+import MakeTest<MakeTestSig<MaximalFlowTest>>
+
+/**
+ * A configuration to find all "maximal" flows.
+ * To be used on small programs.
+ */
+module MaximalFlowsConfig implements DataFlow::ConfigSig {
+  predicate isSource(DataFlow::Node node) {
+    exists(node.getLocation().getFile().getRelativePath()) and
+    not node.asCfgNode() instanceof CallNode and
+    not node.asCfgNode().getNode() instanceof Return and
+    not node instanceof DataFlow::ParameterNode and
+    not node instanceof DataFlow::PostUpdateNode and
+    // not node.asExpr() instanceof FunctionExpr and
+    // not node.asExpr() instanceof ClassExpr and
+    not DataFlow::localFlowStep(_, node)
+  }
+
+  predicate isSink(DataFlow::Node node) {
+    exists(node.getLocation().getFile().getRelativePath()) and
+    not any(CallNode c).getArg(_) = node.asCfgNode() and
+    not node instanceof DataFlow::ArgumentNode and
+    not node.asCfgNode().(NameNode).getId().matches("SINK%") and
+    not DataFlow::localFlowStep(node, _)
+  }
+}
+
+module MaximalFlows = DataFlow::Global<MaximalFlowsConfig>;

python/ql/test/TestUtilities/dataflow/NormalDataflowTest.qll

@@ -0,0 +1,34 @@
+import python
+import TestUtilities.dataflow.FlowTest
+import TestUtilities.dataflow.testConfig
+private import semmle.python.dataflow.new.internal.PrintNode
+
+module DataFlowTest implements FlowTestSig {
+  string flowTag() { result = "flow" }
+
+  predicate relevantFlow(DataFlow::Node source, DataFlow::Node sink) {
+    TestFlow::flow(source, sink)
+  }
+}
+
+import MakeTest<MakeTestSig<DataFlowTest>>
+
+query predicate missingAnnotationOnSink(Location location, string error, string element) {
+  error = "ERROR, you should add `# $ MISSING: flow` annotation" and
+  exists(DataFlow::Node sink |
+    TestConfig::isSink(sink) and
+    // note: we only care about `SINK` and not `SINK_F`, so we have to reconstruct manually.
+    exists(DataFlow::CallCfgNode call |
+      call.getFunction().asCfgNode().(NameNode).getId() = "SINK" and
+      (sink = call.getArg(_) or sink = call.getArgByName(_))
+    ) and
+    location = sink.getLocation() and
+    element = prettyExpr(sink.asExpr()) and
+    not TestFlow::flowTo(sink) and
+    not exists(FalseNegativeTestExpectation missingResult |
+      missingResult.getTag() = "flow" and
+      missingResult.getLocation().getFile() = location.getFile() and
+      missingResult.getLocation().getStartLine() = location.getStartLine()
+    )
+  )
+}

python/ql/test/TestUtilities/dataflow/NormalTaintTrackingTest.qll

@@ -0,0 +1,33 @@
+import python
+import TestUtilities.dataflow.FlowTest
+import TestUtilities.dataflow.testTaintConfig
+private import semmle.python.dataflow.new.internal.PrintNode
+
+module DataFlowTest implements FlowTestSig {
+  string flowTag() { result = "flow" }
+
+  predicate relevantFlow(DataFlow::Node source, DataFlow::Node sink) {
+    TestFlow::flow(source, sink)
+  }
+}
+
+import MakeTest<MakeTestSig<DataFlowTest>>
+
+query predicate missingAnnotationOnSink(Location location, string error, string element) {
+  error = "ERROR, you should add `# $ MISSING: flow` annotation" and
+  exists(DataFlow::Node sink |
+    exists(DataFlow::CallCfgNode call |
+      // note: we only care about `SINK` and not `SINK_F`, so we have to reconstruct manually.
+      call.getFunction().asCfgNode().(NameNode).getId() = "SINK" and
+      (sink = call.getArg(_) or sink = call.getArgByName(_))
+    ) and
+    location = sink.getLocation() and
+    element = prettyExpr(sink.asExpr()) and
+    not TestFlow::flowTo(sink) and
+    not exists(FalseNegativeTestExpectation missingResult |
+      missingResult.getTag() = "flow" and
+      missingResult.getLocation().getFile() = location.getFile() and
+      missingResult.getLocation().getStartLine() = location.getStartLine()
+    )
+  )
+}

python/ql/test/TestUtilities/dataflow/RoutingTest.qll

@@ -0,0 +1,58 @@
+import python
+import semmle.python.dataflow.new.DataFlow
+import TestUtilities.InlineExpectationsTest
+private import semmle.python.dataflow.new.internal.PrintNode
+private import semmle.python.dataflow.new.internal.DataFlowPrivate as DataFlowPrivate
+
+/**
+ * A routing test is designed to test that values are routed to the
+ * correct arguments of the correct functions. It is assumed that
+ * the functions tested sink their arguments sequentially, that is
+ * `SINK1(arg1)`, etc.
+ */
+signature module RoutingTestSig {
+  class Argument;
+
+  string flowTag(Argument arg);
+
+  predicate relevantFlow(DataFlow::Node fromNode, DataFlow::Node toNode, Argument arg);
+}
+
+module MakeTestSig<RoutingTestSig Impl> implements TestSig {
+  string getARelevantTag() { result in ["func", Impl::flowTag(_)] }
+
+  predicate hasActualResult(Location location, string element, string tag, string value) {
+    exists(DataFlow::Node fromNode, DataFlow::Node toNode, Impl::Argument arg |
+      Impl::relevantFlow(fromNode, toNode, arg)
+    |
+      location = fromNode.getLocation() and
+      element = fromNode.toString() and
+      (
+        tag = Impl::flowTag(arg) and
+        if "\"" + tag + "\"" = fromValue(fromNode) then value = "" else value = fromValue(fromNode)
+        or
+        // only have result for `func` tag if the function where `arg<n>` is used, is
+        // different from the function name of the call where `arg<n>` was specified as
+        // an argument
+        tag = "func" and
+        value = toFunc(toNode) and
+        not value = fromFunc(fromNode)
+      )
+    )
+  }
+}
+
+pragma[inline]
+private string fromValue(DataFlow::Node fromNode) {
+  result = "\"" + prettyNode(fromNode).replaceAll("\"", "'") + "\""
+}
+
+pragma[inline]
+private string fromFunc(DataFlow::ArgumentNode fromNode) {
+  result = fromNode.getCall().getNode().(CallNode).getFunction().getNode().(Name).getId()
+}
+
+pragma[inline]
+private string toFunc(DataFlow::Node toNode) {
+  result = toNode.getEnclosingCallable().getQualifiedName()
+}

python/ql/test/TestUtilities/dataflow/UnresolvedCalls.qll

@@ -0,0 +1,37 @@
+import python
+private import semmle.python.dataflow.new.internal.PrintNode
+private import semmle.python.dataflow.new.internal.DataFlowPrivate as DataFlowPrivate
+private import semmle.python.ApiGraphs
+import TestUtilities.InlineExpectationsTest
+
+signature module UnresolvedCallExpectationsSig {
+  predicate unresolvedCall(CallNode call);
+}
+
+module DefaultUnresolvedCallExpectations implements UnresolvedCallExpectationsSig {
+  predicate unresolvedCall(CallNode call) {
+    not exists(DataFlowPrivate::DataFlowCall dfc |
+      exists(dfc.getCallable()) and dfc.getNode() = call
+    ) and
+    not DataFlowPrivate::resolveClassCall(call, _) and
+    not call = API::builtin(_).getACall().asCfgNode()
+  }
+}
+
+module MakeUnresolvedCallExpectations<UnresolvedCallExpectationsSig Impl> {
+  private module UnresolvedCallExpectations implements TestSig {
+    string getARelevantTag() { result = "unresolved_call" }
+
+    predicate hasActualResult(Location location, string element, string tag, string value) {
+      exists(location.getFile().getRelativePath()) and
+      exists(CallNode call | Impl::unresolvedCall(call) |
+        location = call.getLocation() and
+        tag = "unresolved_call" and
+        value = prettyExpr(call.getNode()) and
+        element = call.toString()
+      )
+    }
+  }
+
+  import MakeTest<UnresolvedCallExpectations>
+}

python/ql/test/TestUtilities/dataflow/callGraphConfig.qll

@@ -0,0 +1,22 @@
+private import python
+import semmle.python.dataflow.new.DataFlow
+private import semmle.python.dataflow.new.internal.DataFlowPrivate as DataFlowPrivate
+
+/**
+ * A configuration to find the call graph edges.
+ */
+module CallGraphConfig implements DataFlow::ConfigSig {
+  predicate isSource(DataFlow::Node node) {
+    node instanceof DataFlowPrivate::ReturnNode
+    or
+    node instanceof DataFlow::ArgumentNode
+  }
+
+  predicate isSink(DataFlow::Node node) {
+    node instanceof DataFlowPrivate::OutNode
+    or
+    node instanceof DataFlow::ParameterNode
+  }
+}
+
+module CallGraphFlow = DataFlow::Global<CallGraphConfig>;

python/ql/test/TestUtilities/dataflow/testConfig.qll

@@ -0,0 +1,49 @@
+/**
+ * Configuration to test selected data flow
+ * Sources in the source code are denoted by the special name `SOURCE`,
+ * and sinks are denoted by arguments to the special function `SINK`.
+ * For example, given the test code
+ * ```python
+ *  def test():
+ *      s = SOURCE
+ *      SINK(s)
+ * ```
+ * `SOURCE` will be a source and the second occurrence of `s` will be a sink.
+ *
+ * In order to test literals, alternative sources are defined for each type:
+ *
+ *  for | use
+ * ----------
+ * string | `"source"`
+ * integer | `42`
+ * float | `42.0`
+ * complex | `42j` (not supported yet)
+ */
+
+private import python
+import semmle.python.dataflow.new.DataFlow
+
+module TestConfig implements DataFlow::ConfigSig {
+  predicate isSource(DataFlow::Node node) {
+    node.(DataFlow::CfgNode).getNode().(NameNode).getId() = "SOURCE"
+    or
+    node.(DataFlow::CfgNode).getNode().getNode().(StringLiteral).getS() = "source"
+    or
+    node.(DataFlow::CfgNode).getNode().getNode().(IntegerLiteral).getN() = "42"
+    or
+    node.(DataFlow::CfgNode).getNode().getNode().(FloatLiteral).getN() = "42.0"
+    // No support for complex numbers
+  }
+
+  predicate isSink(DataFlow::Node node) {
+    exists(DataFlow::CallCfgNode call |
+      call.getFunction().asCfgNode().(NameNode).getId() in ["SINK", "SINK_F"] and
+      (node = call.getArg(_) or node = call.getArgByName(_)) and
+      not node = call.getArgByName("not_present_at_runtime")
+    )
+  }
+
+  predicate isBarrierIn(DataFlow::Node node) { isSource(node) }
+}
+
+module TestFlow = DataFlow::Global<TestConfig>;

python/ql/test/TestUtilities/dataflow/testTaintConfig.qll

@@ -0,0 +1,49 @@
+/**
+ * Configuration to test selected data flow
+ * Sources in the source code are denoted by the special name `SOURCE`,
+ * and sinks are denoted by arguments to the special function `SINK`.
+ * For example, given the test code
+ * ```python
+ *  def test():
+ *      s = SOURCE
+ *      SINK(s)
+ * ```
+ * `SOURCE` will be a source and the second occurrence of `s` will be a sink.
+ *
+ * In order to test literals, alternative sources are defined for each type:
+ *
+ *  for | use
+ * ----------
+ * string | `"source"`
+ * integer | `42`
+ * float | `42.0`
+ * complex | `42j` (not supported yet)
+ */
+
+private import python
+import semmle.python.dataflow.new.DataFlow
+import semmle.python.dataflow.new.TaintTracking
+
+module TestConfig implements DataFlow::ConfigSig {
+  predicate isSource(DataFlow::Node node) {
+    node.(DataFlow::CfgNode).getNode().(NameNode).getId() = "SOURCE"
+    or
+    node.(DataFlow::CfgNode).getNode().getNode().(StringLiteral).getS() = "source"
+    or
+    node.(DataFlow::CfgNode).getNode().getNode().(IntegerLiteral).getN() = "42"
+    or
+    node.(DataFlow::CfgNode).getNode().getNode().(FloatLiteral).getN() = "42.0"
+    // No support for complex numbers
+  }
+
+  predicate isSink(DataFlow::Node node) {
+    exists(CallNode call |
+      call.getFunction().(NameNode).getId() in ["SINK", "SINK_F"] and
+      node.(DataFlow::CfgNode).getNode() = call.getAnArg()
+    )
+  }
+
+  predicate isBarrierIn(DataFlow::Node node) { isSource(node) }
+}
+
+module TestFlow = TaintTracking::Global<TestConfig>;