Merge pull request #17663 from aschackmull/dataflow/speculative-flow

Dataflow: Add support for speculative taint flow.

python/ql/lib/semmle/python/dataflow/new/internal/DataFlowImpl1.qll

@@ -261,13 +261,17 @@ deprecated private module Config implements FullStateConfigSig {
     model = ""
   }
 
-  predicate isAdditionalFlowStep(Node node1, FlowState state1, Node node2, FlowState state2) {
+  predicate isAdditionalFlowStep(
+    Node node1, FlowState state1, Node node2, FlowState state2, string model
+  ) {
     getConfig(state1).isAdditionalFlowStep(node1, getState(state1), node2, getState(state2)) and
-    getConfig(state2) = getConfig(state1)
+    getConfig(state2) = getConfig(state1) and
+    model = ""
     or
     not singleConfiguration() and
     getConfig(state1).isAdditionalFlowStep(node1, node2) and
-    state2 = state1
+    state2 = state1 and
+    model = ""
   }
 
   predicate allowImplicitRead(Node node, ContentSet c) {

python/ql/lib/semmle/python/dataflow/new/internal/DataFlowImpl2.qll

@@ -261,13 +261,17 @@ deprecated private module Config implements FullStateConfigSig {
     model = ""
   }
 
-  predicate isAdditionalFlowStep(Node node1, FlowState state1, Node node2, FlowState state2) {
+  predicate isAdditionalFlowStep(
+    Node node1, FlowState state1, Node node2, FlowState state2, string model
+  ) {
     getConfig(state1).isAdditionalFlowStep(node1, getState(state1), node2, getState(state2)) and
-    getConfig(state2) = getConfig(state1)
+    getConfig(state2) = getConfig(state1) and
+    model = ""
     or
     not singleConfiguration() and
     getConfig(state1).isAdditionalFlowStep(node1, node2) and
-    state2 = state1
+    state2 = state1 and
+    model = ""
   }
 
   predicate allowImplicitRead(Node node, ContentSet c) {

python/ql/lib/semmle/python/dataflow/new/internal/DataFlowImpl3.qll

@@ -261,13 +261,17 @@ deprecated private module Config implements FullStateConfigSig {
     model = ""
   }
 
-  predicate isAdditionalFlowStep(Node node1, FlowState state1, Node node2, FlowState state2) {
+  predicate isAdditionalFlowStep(
+    Node node1, FlowState state1, Node node2, FlowState state2, string model
+  ) {
     getConfig(state1).isAdditionalFlowStep(node1, getState(state1), node2, getState(state2)) and
-    getConfig(state2) = getConfig(state1)
+    getConfig(state2) = getConfig(state1) and
+    model = ""
     or
     not singleConfiguration() and
     getConfig(state1).isAdditionalFlowStep(node1, node2) and
-    state2 = state1
+    state2 = state1 and
+    model = ""
   }
 
   predicate allowImplicitRead(Node node, ContentSet c) {

python/ql/lib/semmle/python/dataflow/new/internal/DataFlowImpl4.qll

@@ -261,13 +261,17 @@ deprecated private module Config implements FullStateConfigSig {
     model = ""
   }
 
-  predicate isAdditionalFlowStep(Node node1, FlowState state1, Node node2, FlowState state2) {
+  predicate isAdditionalFlowStep(
+    Node node1, FlowState state1, Node node2, FlowState state2, string model
+  ) {
     getConfig(state1).isAdditionalFlowStep(node1, getState(state1), node2, getState(state2)) and
-    getConfig(state2) = getConfig(state1)
+    getConfig(state2) = getConfig(state1) and
+    model = ""
     or
     not singleConfiguration() and
     getConfig(state1).isAdditionalFlowStep(node1, node2) and
-    state2 = state1
+    state2 = state1 and
+    model = ""
   }
 
   predicate allowImplicitRead(Node node, ContentSet c) {

python/ql/lib/semmle/python/dataflow/new/internal/TaintTrackingPrivate.qll

@@ -219,3 +219,36 @@ predicate asyncWithStep(DataFlow::Node nodeFrom, DataFlow::Node nodeTo) {
     contextManager.strictlyDominates(var)
   )
 }
+
+import SpeculativeTaintFlow
+
+private module SpeculativeTaintFlow {
+  private import semmle.python.dataflow.new.internal.DataFlowDispatch as DataFlowDispatch
+  private import semmle.python.dataflow.new.internal.DataFlowPublic as DataFlowPublic
+
+  /**
+   * Holds if the additional step from `src` to `sink` should be considered in
+   * speculative taint flow exploration.
+   */
+  predicate speculativeTaintStep(DataFlow::Node src, DataFlow::Node sink) {
+    exists(DataFlowDispatch::DataFlowCall call, DataFlowDispatch::ArgumentPosition argpos |
+      // TODO: exclude neutrals and anything that has QL modeling.
+      not exists(DataFlowDispatch::DataFlowCall call0 |
+        // Workaround for the fact that python currently associates several
+        // DataFlowCalls with a single call.
+        src.(DataFlowPublic::ArgumentNode).argumentOf(call0, _) and
+        exists(DataFlowDispatch::viableCallable(call0))
+      ) and
+      call instanceof DataFlowDispatch::PotentialLibraryCall and
+      src.(DataFlowPublic::ArgumentNode).argumentOf(call, argpos)
+    |
+      not argpos.isSelf() and
+      sink.(DataFlowPublic::PostUpdateNode)
+          .getPreUpdateNode()
+          .(DataFlowPublic::ArgumentNode)
+          .argumentOf(call, any(DataFlowDispatch::ArgumentPosition qualpos | qualpos.isSelf()))
+      or
+      sink.(DataFlowDispatch::OutNode).getCall(_) = call
+    )
+  }
+}