Python: Add jump steps for global variable nested field access

python/ql/lib/change-notes/2025-08-11-jump-step-global-nested.md

@@ -0,0 +1,4 @@
+---
+category: minorAnalysis
+---
+* Data flow tracking through global variables now supports nested field access patterns such as `global_var.obj.field`. This improves the precision of taint tracking analysis when data flows through complex global variable structures.

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

@@ -64,6 +64,15 @@ abstract class AttrRef extends Node {
 abstract class AttrWrite extends AttrRef {
   /** Gets the data flow node corresponding to the value that is written to the attribute. */
   abstract Node getValue();
+
+  /**
+   * Holds if this attribute write writes the attribute named `attrName` on object `object` with
+   * value `value`.
+   */
+  predicate writes(Node object, string attrName, Node value) {
+    this.accesses(object, attrName) and
+    this.getValue() = value
+  }
 }
 
 /**
@@ -225,7 +234,10 @@ private class ClassDefinitionAsAttrWrite extends AttrWrite, CfgNode {
  * - Dynamic attribute reads using `getattr`: `getattr(object, attr)`
  * - Qualified imports: `from module import attr as name`
  */
-abstract class AttrRead extends AttrRef, Node, LocalSourceNode { }
+abstract class AttrRead extends AttrRef, Node, LocalSourceNode {
+  /** Holds if this attribute read reads the attribute named `attrName` on the object `object`. */
+  predicate reads(Node object, string attrName) { this.accesses(object, attrName) }
+}
 
 /** A simple attribute read, e.g. `object.attr` */
 private class AttributeReadAsAttrRead extends AttrRead, CfgNode {

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

@@ -556,6 +556,75 @@ predicate runtimeJumpStep(Node nodeFrom, Node nodeTo) {
     nodeFrom.asCfgNode() = param.getDefault() and
     nodeTo.asCfgNode() = param.getDefiningNode()
   )
+  or
+  // Enhanced global variable field access tracking
+  globalVariableNestedFieldJumpStep(nodeFrom, nodeTo)
+}
+
+/** Helper predicate for `globalVariableNestedFieldJumpStep`. */
+pragma[nomagic]
+private predicate globalVariableAttrPathRead(
+  ModuleVariableNode globalVar, string accessPath, AttrRead r, string attrName
+) {
+  globalVariableAttrPathAtDepth(globalVar, accessPath, r.getObject(), _) and
+  attrName = r.getAttributeName()
+}
+
+/** Helper predicate for `globalVariableNestedFieldJumpStep`. */
+pragma[nomagic]
+private predicate globalVariableAttrPathWrite(
+  ModuleVariableNode globalVar, string accessPath, AttrWrite w, string attrName
+) {
+  globalVariableAttrPathAtDepth(globalVar, accessPath, w.getObject(), _) and
+  attrName = w.getAttributeName()
+}
+
+/**
+ * Holds if there is a jump step from `nodeFrom` to `nodeTo` through global variable field access.
+ * This supports tracking nested object field access through global variables like `app.obj.foo`.
+ */
+pragma[nomagic]
+private predicate globalVariableNestedFieldJumpStep(Node nodeFrom, Node nodeTo) {
+  exists(ModuleVariableNode globalVar, AttrWrite write, AttrRead read |
+    // Match writes and reads on the same global variable attribute path
+    exists(string accessPath, string attrName |
+      globalVariableAttrPathRead(globalVar, accessPath, read, attrName) and
+      globalVariableAttrPathWrite(globalVar, accessPath, write, attrName)
+    ) and
+    nodeFrom = write.getValue() and
+    nodeTo = read
+  )
+}
+
+/**
+ * Maximum depth for global variable nested attribute access.
+ * Depth 1 = globalVar.foo, depth 2 = globalVar.foo.bar, depth 3 = globalVar.foo.bar.baz, etc.
+ */
+private int getMaxGlobalVariableDepth() { result = 2 }
+
+/**
+ * Holds if `node` is an attribute access path starting from global variable `globalVar` at specific `depth`.
+ */
+private predicate globalVariableAttrPathAtDepth(
+  ModuleVariableNode globalVar, string accessPath, Node node, int depth
+) {
+  // Base case: Direct global variable access (depth 0)
+  depth = 0 and
+  // We use `globalVar` instead of `globalVar.getAWrite()` due to some weirdness with how
+  // attribute writes are handled in the global scope (see `GlobalAttributeAssignmentAsAttrWrite`).
+  node in [globalVar.getARead(), globalVar] and
+  accessPath = ""
+  or
+  exists(Node obj, string attrName, string parentAccessPath, int parentDepth |
+    node.(AttrRead).reads(obj, attrName)
+    or
+    any(AttrWrite aw).writes(obj, attrName, node)
+  |
+    globalVariableAttrPathAtDepth(globalVar, parentAccessPath, obj, parentDepth) and
+    accessPath = parentAccessPath + "." + attrName and
+    depth = parentDepth + 1 and
+    depth <= getMaxGlobalVariableDepth()
+  )
 }
 
 //--------

python/ql/test/library-tests/dataflow/fieldflow/test.py

@@ -301,12 +301,12 @@ def set_to_source():
 @expects(4) # $ unresolved_call=expects(..) unresolved_call=expects(..)(..)
 def test_global_flow_to_class_attribute():
     inst = WithTuple2()
-    SINK_F(WithTuple2.my_tuple[0])
+    SINK_F(WithTuple2.my_tuple[0])  # $ SPURIOUS: flow="SOURCE, l:-5 -> WithTuple2.my_tuple[0]"
     SINK_F(inst.my_tuple[0])
 
     set_to_source()
 
-    SINK(WithTuple2.my_tuple[0]) # $ MISSING: flow="SOURCE, l:-10 -> WithTuple2.my_tuple[0]"
+    SINK(WithTuple2.my_tuple[0]) # $ flow="SOURCE, l:-10 -> WithTuple2.my_tuple[0]"
     SINK(inst.my_tuple[0]) # $ MISSING: flow="SOURCE, l:-11 -> inst.my_tuple[0]"
 
 

python/ql/test/library-tests/dataflow/fieldflow/test_global.py

@@ -146,6 +146,7 @@ def fields_with_local_flow(x):
 class NestedObj(object):
     def __init__(self):
         self.obj = MyObj("OK")
+        self.obj.foo = "default"
 
     def getObj(self):
         return self.obj
@@ -163,17 +164,31 @@ def getObj(self):
 a2 = NestedObj()
 a2.getObj().foo = x2
 SINK(a2.obj.foo) # $ flow="SOURCE, l:-3 -> a2.obj.foo"
+
+# Global variable
+app = NestedObj()
+
+def init_global():
+    app.obj.foo = SOURCE
+
+def read_global(): 
+    return app.obj.foo
+
+def test_global_nested_attributes():
+    init_global()
+    result = read_global() 
+    SINK(result) # $ flow="SOURCE, l:-8 -> result"
 
 # ------------------------------------------------------------------------------
 # Global scope interaction
 # ------------------------------------------------------------------------------
 
 def func_defined_before():
-    SINK(global_obj.foo) # $ MISSING: flow="SOURCE, l:+3 -> global_obj.foo"
+    SINK(global_obj.foo) # $ flow="SOURCE, l:+3 -> global_obj.foo"
 
 global_obj = MyObj(NONSOURCE)
 global_obj.foo = SOURCE
 SINK(global_obj.foo) # $ flow="SOURCE, l:-1 -> global_obj.foo"
 
 def func_defined_after():
-    SINK(global_obj.foo) # $ MISSING: flow="SOURCE, l:-4 -> global_obj.foo"
+    SINK(global_obj.foo) # $ flow="SOURCE, l:-4 -> global_obj.foo"