Python: Add models for synchronised iterators

python/ql/lib/change-notes/2026-09-30-python315-threading-iterators.md

@@ -0,0 +1,4 @@
+---
+category: minorAnalysis
+---
+* Added data-flow modeling for `threading.serialize_iterator` and `threading.concurrent_tee`, introduced in Python 3.15.

python/ql/lib/semmle/python/frameworks/Stdlib.model.yml

@@ -138,6 +138,17 @@ extensions:
       - ["tempfile", "Member[mkstemp]", "Argument[0,suffix:,1,prefix:,2,dir:]", "ReturnValue.TupleElement[0,1]", "taint"]
       # See https://docs.python.org/3/library/textwrap.html#textwrap.dedent
       - ["textwrap", "Member[dedent]", "Argument[0,text:]", "ReturnValue", "taint"]
+      # See https://docs.python.org/3/library/threading.html#threading.serialize_iterator
+      - ["threading", "Member[serialize_iterator]", "Argument[0,iterable:].ListElement", "ReturnValue.ListElement", "value"]
+      - ["threading", "Member[serialize_iterator]", "Argument[0,iterable:].SetElement", "ReturnValue.ListElement", "value"]
+      - ["threading", "Member[serialize_iterator]", "Argument[0,iterable:].AnyTupleElement", "ReturnValue.ListElement", "value"]
+      - ["threading", "Member[serialize_iterator]", "Argument[0,iterable:]", "ReturnValue", "taint"]
+      # See https://docs.python.org/3/library/threading.html#threading.concurrent_tee
+      # All returned iterators have the same contents, so the outer tuple is modeled as an index-insensitive sequence.
+      - ["threading", "Member[concurrent_tee]", "Argument[0,iterable:].ListElement", "ReturnValue.ListElement.ListElement", "value"]
+      - ["threading", "Member[concurrent_tee]", "Argument[0,iterable:].SetElement", "ReturnValue.ListElement.ListElement", "value"]
+      - ["threading", "Member[concurrent_tee]", "Argument[0,iterable:].AnyTupleElement", "ReturnValue.ListElement.ListElement", "value"]
+      - ["threading", "Member[concurrent_tee]", "Argument[0,iterable:]", "ReturnValue", "taint"]
       # See https://docs.python.org/3/library/traceback.html#traceback.StackSummary.from_list
       - ["traceback.StackSummary", "Member[from_list]", "Argument[0,a_list:]", "ReturnValue", "taint"]
       # See https://docs.python.org/3/library/typing.html#typing.cast

python/ql/test/library-tests/dataflow/threading-iterators/InlineTaintTest.expected

@@ -0,0 +1,3 @@
+argumentToEnsureNotTaintedNotMarkedAsSpurious
+untaintedArgumentToEnsureTaintedNotMarkedAsMissing
+testFailures

python/ql/test/library-tests/dataflow/threading-iterators/InlineTaintTest.ql

@@ -0,0 +1,2 @@
+import experimental.meta.InlineTaintTest
+import MakeInlineTaintTest<TestTaintTrackingConfig>

python/ql/test/library-tests/dataflow/threading-iterators/NormalDataflowTest.expected

@@ -0,0 +1,2 @@
+missingAnnotationOnSink
+testFailures

python/ql/test/library-tests/dataflow/threading-iterators/NormalDataflowTest.ql

@@ -0,0 +1,2 @@
+import python
+import utils.test.dataflow.NormalDataflowTest

python/ql/test/library-tests/dataflow/threading-iterators/options

@@ -0,0 +1 @@
+semmle-extractor-options: --lang=3 --max-import-depth=1

python/ql/test/library-tests/dataflow/threading-iterators/test.py

@@ -0,0 +1,110 @@
+import threading
+from threading import concurrent_tee, serialize_iterator
+
+
+def serialized_list():
+    iterator = threading.serialize_iterator([SOURCE])
+    SINK(next(iterator)) # $ flow="SOURCE, l:-1 -> next(..)"
+    SINK_F(iterator)
+
+
+def serialized_tuple():
+    iterator = serialize_iterator(iterable=(SOURCE,))
+    SINK(next(iterator)) # $ flow="SOURCE, l:-1 -> next(..)"
+
+
+def serialized_set():
+    wrap = serialize_iterator
+    iterator = wrap({SOURCE})
+    SINK(next(iterator)) # $ flow="SOURCE, l:-1 -> next(..)"
+
+
+def serialized_generator():
+    iterator = serialize_iterator(value for value in [SOURCE])
+    for value in iterator:
+        SINK(value) # $ flow="SOURCE, l:-2 -> value"
+
+
+def serialized_contents():
+    iterator = serialize_iterator([{"tainted": SOURCE, "clean": NONSOURCE}])
+    value = next(iterator)
+    SINK(value["tainted"]) # $ flow="SOURCE, l:-2 -> value['tainted']"
+    SINK_F(value["clean"])
+
+
+def serialized_callback():
+    iterator = next(map(serialize_iterator, [[SOURCE]]))
+    SINK(next(iterator)) # $ flow="SOURCE, l:-1 -> next(..)"
+
+
+def tee_list():
+    first, second = threading.concurrent_tee([SOURCE])
+    SINK(next(first)) # $ flow="SOURCE, l:-1 -> next(..)"
+    SINK(next(second)) # $ flow="SOURCE, l:-2 -> next(..)"
+    SINK_F(first)
+
+
+def tee_tuple():
+    copies = concurrent_tee(iterable=(SOURCE,), n=3)
+    SINK(next(copies[0])) # $ flow="SOURCE, l:-1 -> next(..)"
+    SINK(next(copies[2])) # $ flow="SOURCE, l:-2 -> next(..)"
+
+
+def tee_set():
+    split = concurrent_tee
+    first, second = split({SOURCE}, 2)
+    SINK(next(first)) # $ flow="SOURCE, l:-1 -> next(..)"
+    SINK(next(second)) # $ flow="SOURCE, l:-2 -> next(..)"
+
+
+def tee_many():
+    copies = concurrent_tee([SOURCE], n=10)
+    SINK(next(copies[9])) # $ flow="SOURCE, l:-1 -> next(..)"
+    SINK(next(copies[-1])) # $ flow="SOURCE, l:-2 -> next(..)"
+
+
+def tee_generator():
+    copies = concurrent_tee(value for value in [SOURCE])
+    for iterator in copies:
+        SINK(next(iterator)) # $ flow="SOURCE, l:-2 -> next(..)"
+
+
+def tee_contents():
+    first, second = concurrent_tee([{"tainted": SOURCE, "clean": NONSOURCE}])
+    value = next(second)
+    SINK(value["tainted"]) # $ flow="SOURCE, l:-2 -> value['tainted']"
+    SINK_F(value["clean"])
+
+
+def tee_callback():
+    copies = next(map(concurrent_tee, [[SOURCE]]))
+    SINK(next(copies[0])) # $ flow="SOURCE, l:-1 -> next(..)"
+
+
+# The decorator form is already handled without a dedicated model.
+@threading.synchronized_iterator
+def decorated_generator(value):
+    yield {"tainted": value, "clean": NONSOURCE}
+
+
+def synchronized_generator():
+    value = next(decorated_generator(SOURCE))
+    SINK(value["tainted"]) # $ flow="SOURCE, l:-1 -> value['tainted']"
+    SINK_F(value["clean"])
+
+
+def clean_inputs():
+    SINK_F(next(serialize_iterator([NONSOURCE])))
+    SINK_F(next(concurrent_tee([NONSOURCE])[0]))
+    SINK_F(next(concurrent_tee([NONSOURCE], n=SOURCE)[0]))
+
+
+def shadowed_functions():
+    def serialize_iterator(iterable):
+        return iter([NONSOURCE])
+
+    def concurrent_tee(iterable, n=2):
+        return iter([NONSOURCE]), iter([NONSOURCE])
+
+    SINK_F(next(serialize_iterator([SOURCE])))
+    SINK_F(next(concurrent_tee([SOURCE])[0]))

python/ql/test/library-tests/dataflow/threading-iterators/test_taint.py

@@ -0,0 +1,32 @@
+import threading
+
+
+def serialized_taint():
+    iterator = threading.serialize_iterator(TAINTED_LIST)
+    ensure_tainted(iterator) # $ tainted
+    ensure_tainted(next(iterator)) # $ tainted
+
+
+def tee_taint():
+    first, second = threading.concurrent_tee(TAINTED_LIST)
+    ensure_tainted(first) # $ tainted
+    ensure_tainted(second) # $ tainted
+    ensure_tainted(next(first)) # $ tainted
+    ensure_tainted(next(second)) # $ tainted
+
+
+def nested_contents():
+    iterator = threading.serialize_iterator([{"tainted": TAINTED_STRING, "clean": "safe"}])
+    value = next(iterator)
+    ensure_tainted(value["tainted"]) # $ tainted
+    ensure_not_tainted(value["clean"])
+
+    first, second = threading.concurrent_tee([{"tainted": TAINTED_STRING, "clean": "safe"}])
+    value = next(second)
+    ensure_tainted(value["tainted"]) # $ tainted
+    ensure_not_tainted(value["clean"])
+
+
+def count_is_not_data():
+    copies = threading.concurrent_tee(["clean"], n=taint(2))
+    ensure_not_tainted(next(copies[0]))