Merge remote-tracking branch 'origin/main' into jorgectf/python/headerInjection

python/ql/test/experimental/dataflow/method-calls/test.expected

@@ -0,0 +1,8 @@
+conjunctive_lookup
+| test.py:6:1:6:6 | ControlFlowNode for meth() | meth() | obj1 | bar |
+| test.py:6:1:6:6 | ControlFlowNode for meth() | meth() | obj1 | foo |
+| test.py:6:1:6:6 | ControlFlowNode for meth() | meth() | obj2 | bar |
+| test.py:6:1:6:6 | ControlFlowNode for meth() | meth() | obj2 | foo |
+calls_lookup
+| test.py:6:1:6:6 | ControlFlowNode for meth() | meth() | obj1 | foo |
+| test.py:6:1:6:6 | ControlFlowNode for meth() | meth() | obj2 | bar |

python/ql/test/experimental/dataflow/method-calls/test.py

@@ -0,0 +1,6 @@
+if cond:
+    meth = obj1.foo
+else:
+    meth = obj2.bar
+
+meth()

python/ql/test/experimental/dataflow/method-calls/test.ql

@@ -0,0 +1,18 @@
+import python
+import semmle.python.dataflow.new.DataFlow
+import experimental.dataflow.TestUtil.PrintNode
+
+query predicate conjunctive_lookup(
+  DataFlow::MethodCallNode methCall, string call, string object, string methodName
+) {
+  call = prettyNode(methCall) and
+  object = prettyNode(methCall.getObject()) and
+  methodName = methCall.getMethodName()
+}
+
+query predicate calls_lookup(
+  DataFlow::MethodCallNode methCall, string call, string object, string methodName
+) {
+  call = prettyNode(methCall) and
+  exists(DataFlow::Node o | methCall.calls(o, methodName) and object = prettyNode(o))
+}

python/ql/test/experimental/dataflow/sensitive-data/test.py

@@ -78,3 +78,37 @@ request.args.getlist("password")[0] # $ MISSING: SensitiveDataSource=password
 
 from not_found import password2 as foo # $ SensitiveDataSource=password
 print(foo) # $ SensitiveUse=password
+
+# ------------------------------------------------------------------------------
+# cross-talk between different calls
+# ------------------------------------------------------------------------------
+
+# Case 1: providing name as argument
+
+_configuration = {"sleep_timer": 5, "mysql_password": "1234"}
+
+def get_config(key):
+    # Treating this as a SensitiveDataSource is questionable, since that will result in
+    # _all_ calls to `get_config` being treated as giving sensitive data
+    return _configuration[key]
+
+foo = get_config("mysql_password")
+print(foo) # $ MISSING: SensitiveUse=password
+
+bar = get_config("sleep_timer")
+print(bar)
+
+# Case 2: Providing function as argument
+
+def call_wrapper(func):
+    print("Will call", func)
+    # Treating this as a SensitiveDataSource is questionable, since that will result in
+    # _all_ calls to `call_wrapper` being treated as giving sensitive data
+    return func() # $ SensitiveDataSource=password
+
+foo = call_wrapper(get_password)
+print(foo) # $ SensitiveUse=password
+
+harmless = lambda: "bar"
+bar = call_wrapper(harmless)
+print(bar) # $ SPURIOUS: SensitiveUse=password

python/ql/test/experimental/dataflow/tainttracking/defaultAdditionalTaintStep/test_string.py

@@ -104,7 +104,7 @@ def non_syntactic():
     _str = str
     ensure_tainted(
         meth(), # $ MISSING: tainted
-        _str(ts), # $ MISSING: tainted
+        _str(ts), # $ tainted
     )
 
 

python/ql/test/experimental/dataflow/typetracking/moduleattr.ql

@@ -2,7 +2,7 @@ import python
 import semmle.python.dataflow.new.DataFlow
 import semmle.python.dataflow.new.TypeTracker
 
-private DataFlow::LocalSourceNode module_tracker(TypeTracker t) {
+private DataFlow::TypeTrackingNode module_tracker(TypeTracker t) {
   t.start() and
   result = DataFlow::importNode("module")
   or
@@ -13,7 +13,7 @@ query DataFlow::Node module_tracker() {
   module_tracker(DataFlow::TypeTracker::end()).flowsTo(result)
 }
 
-private DataFlow::LocalSourceNode module_attr_tracker(TypeTracker t) {
+private DataFlow::TypeTrackingNode module_attr_tracker(TypeTracker t) {
   t.startInAttr("attr") and
   result = module_tracker()
   or

python/ql/test/experimental/dataflow/typetracking/tracked.ql

@@ -6,7 +6,7 @@ import TestUtilities.InlineExpectationsTest
 // -----------------------------------------------------------------------------
 // tracked
 // -----------------------------------------------------------------------------
-private DataFlow::LocalSourceNode tracked(TypeTracker t) {
+private DataFlow::TypeTrackingNode tracked(TypeTracker t) {
   t.start() and
   result.asCfgNode() = any(NameNode n | n.getId() = "tracked")
   or
@@ -34,14 +34,14 @@ class TrackedTest extends InlineExpectationsTest {
 // -----------------------------------------------------------------------------
 // int + str
 // -----------------------------------------------------------------------------
-private DataFlow::LocalSourceNode int_type(TypeTracker t) {
+private DataFlow::TypeTrackingNode int_type(TypeTracker t) {
   t.start() and
   result.asCfgNode() = any(CallNode c | c.getFunction().(NameNode).getId() = "int")
   or
   exists(TypeTracker t2 | result = int_type(t2).track(t2, t))
 }
 
-private DataFlow::LocalSourceNode string_type(TypeTracker t) {
+private DataFlow::TypeTrackingNode string_type(TypeTracker t) {
   t.start() and
   result.asCfgNode() = any(CallNode c | c.getFunction().(NameNode).getId() = "str")
   or
@@ -83,7 +83,7 @@ class TrackedStringTest extends InlineExpectationsTest {
 // -----------------------------------------------------------------------------
 // tracked_self
 // -----------------------------------------------------------------------------
-private DataFlow::LocalSourceNode tracked_self(TypeTracker t) {
+private DataFlow::TypeTrackingNode tracked_self(TypeTracker t) {
   t.start() and
   exists(Function f |
     f.isMethod() and
@@ -117,7 +117,7 @@ class TrackedSelfTest extends InlineExpectationsTest {
 // -----------------------------------------------------------------------------
 // This modeling follows the same pattern that we currently use in our real library modeling.
 /** Gets a reference to `foo` (fictive module). */
-private DataFlow::LocalSourceNode foo(DataFlow::TypeTracker t) {
+private DataFlow::TypeTrackingNode foo(DataFlow::TypeTracker t) {
   t.start() and
   result = DataFlow::importNode("foo")
   or
@@ -128,7 +128,7 @@ private DataFlow::LocalSourceNode foo(DataFlow::TypeTracker t) {
 DataFlow::Node foo() { foo(DataFlow::TypeTracker::end()).flowsTo(result) }
 
 /** Gets a reference to `foo.bar` (fictive module). */
-private DataFlow::LocalSourceNode foo_bar(DataFlow::TypeTracker t) {
+private DataFlow::TypeTrackingNode foo_bar(DataFlow::TypeTracker t) {
   t.start() and
   result = DataFlow::importNode("foo.bar")
   or
@@ -142,7 +142,7 @@ private DataFlow::LocalSourceNode foo_bar(DataFlow::TypeTracker t) {
 DataFlow::Node foo_bar() { foo_bar(DataFlow::TypeTracker::end()).flowsTo(result) }
 
 /** Gets a reference to `foo.bar.baz` (fictive attribute on `foo.bar` module). */
-private DataFlow::LocalSourceNode foo_bar_baz(DataFlow::TypeTracker t) {
+private DataFlow::TypeTrackingNode foo_bar_baz(DataFlow::TypeTracker t) {
   t.start() and
   result = DataFlow::importNode("foo.bar.baz")
   or

python/ql/test/experimental/library-tests/frameworks/sqlalchemy/ConceptsTest.ql

@@ -0,0 +1,3 @@
+import python
+import experimental.meta.ConceptsTest
+import experimental.semmle.python.frameworks.SqlAlchemy

python/ql/test/experimental/library-tests/frameworks/sqlalchemy/InlineTaintTest.expected

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

python/ql/test/experimental/library-tests/frameworks/sqlalchemy/InlineTaintTest.ql

@@ -0,0 +1,2 @@
+import experimental.meta.InlineTaintTest
+import experimental.semmle.python.frameworks.SqlAlchemy

python/ql/test/experimental/library-tests/frameworks/sqlalchemy/SqlExecution.py

@@ -0,0 +1,57 @@
+import sqlalchemy
+from sqlalchemy import Column, Integer, String, ForeignKey, create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.pool import StaticPool
+from sqlalchemy.orm import relationship, backref, sessionmaker, joinedload
+from sqlalchemy.sql import text
+
+engine = create_engine(
+    'sqlite:///:memory:',
+    echo=True,
+    connect_args={"check_same_thread": False},
+    poolclass=StaticPool
+)
+
+Base = declarative_base()
+
+class User(Base):
+    __tablename__ = 'users'
+
+    id = Column(Integer, primary_key=True)
+    name = Column(String)
+
+Base.metadata.create_all(engine)
+
+Session = sessionmaker(bind=engine)
+session = Session()
+
+ed_user = User(name='ed')
+ed_user2 = User(name='george')
+
+session.add(ed_user)
+session.add(ed_user2)
+
+session.commit()
+
+# Injection without requiring the text() taint-step
+session.query(User).filter_by(name="some sql")  # $ MISSING: getSql="some sql"
+session.scalar("some sql")  # $ getSql="some sql"
+engine.scalar("some sql")  # $ getSql="some sql"
+session.execute("some sql")  # $ getSql="some sql"
+
+with engine.connect() as connection:
+    connection.execute("some sql")  # $ getSql="some sql"
+
+with engine.begin() as connection:
+    connection.execute("some sql")  # $ getSql="some sql"
+
+# Injection requiring the text() taint-step
+t = text("some sql")
+session.query(User).filter(t)  # $ getSql=t
+session.query(User).group_by(User.id).having(t)  # $ getSql=User.id MISSING: getSql=t
+session.query(User).group_by(t).first()  # $ getSql=t
+session.query(User).order_by(t).first()  # $ getSql=t
+
+query = select(User).where(User.name == t)  # $ MISSING: getSql=t
+with engine.connect() as conn:
+    conn.execute(query) # $ getSql=query

python/ql/test/experimental/library-tests/frameworks/sqlalchemy/taint_test.py

@@ -0,0 +1,12 @@
+import sqlalchemy
+
+def test_taint():
+    ts = TAINTED_STRING
+
+    ensure_tainted(
+        ts, # $ tainted
+        sqlalchemy.text(ts), # $ tainted
+        sqlalchemy.sql.text(ts),# $ tainted
+        sqlalchemy.sql.expression.text(ts),# $ tainted
+        sqlalchemy.sql.expression.TextClause(ts),# $ tainted
+    )

python/ql/test/experimental/meta/ConceptsTest.qll

@@ -93,6 +93,23 @@ class EncodingTest extends InlineExpectationsTest {
   }
 }
 
+class LoggingTest extends InlineExpectationsTest {
+  LoggingTest() { this = "LoggingTest" }
+
+  override string getARelevantTag() { result in ["loggingInput"] }
+
+  override predicate hasActualResult(Location location, string element, string tag, string value) {
+    exists(location.getFile().getRelativePath()) and
+    exists(Logging logging, DataFlow::Node data |
+      location = data.getLocation() and
+      element = data.toString() and
+      value = prettyNodeForInlineTest(data) and
+      data = logging.getAnInput() and
+      tag = "loggingInput"
+    )
+  }
+}
+
 class CodeExecutionTest extends InlineExpectationsTest {
   CodeExecutionTest() { this = "CodeExecutionTest" }
 
@@ -129,6 +146,38 @@ class SqlExecutionTest extends InlineExpectationsTest {
   }
 }
 
+class EscapingTest extends InlineExpectationsTest {
+  EscapingTest() { this = "EscapingTest" }
+
+  override string getARelevantTag() { result in ["escapeInput", "escapeOutput", "escapeKind"] }
+
+  override predicate hasActualResult(Location location, string element, string tag, string value) {
+    exists(location.getFile().getRelativePath()) and
+    exists(Escaping esc |
+      exists(DataFlow::Node data |
+        location = data.getLocation() and
+        element = data.toString() and
+        value = prettyNodeForInlineTest(data) and
+        (
+          data = esc.getAnInput() and
+          tag = "escapeInput"
+          or
+          data = esc.getOutput() and
+          tag = "escapeOutput"
+        )
+      )
+      or
+      exists(string format |
+        location = esc.getLocation() and
+        element = format and
+        value = format and
+        format = esc.getKind() and
+        tag = "escapeKind"
+      )
+    )
+  }
+}
+
 class HttpServerRouteSetupTest extends InlineExpectationsTest {
   HttpServerRouteSetupTest() { this = "HttpServerRouteSetupTest" }
 
@@ -252,6 +301,38 @@ class HttpServerHttpRedirectResponseTest extends InlineExpectationsTest {
   }
 }
 
+class HttpServerCookieWriteTest extends InlineExpectationsTest {
+  HttpServerCookieWriteTest() { this = "HttpServerCookieWriteTest" }
+
+  override string getARelevantTag() {
+    result in ["CookieWrite", "CookieRawHeader", "CookieName", "CookieValue"]
+  }
+
+  override predicate hasActualResult(Location location, string element, string tag, string value) {
+    exists(location.getFile().getRelativePath()) and
+    exists(HTTP::Server::CookieWrite cookieWrite |
+      location = cookieWrite.getLocation() and
+      (
+        element = cookieWrite.toString() and
+        value = "" and
+        tag = "CookieWrite"
+        or
+        element = cookieWrite.toString() and
+        value = prettyNodeForInlineTest(cookieWrite.getHeaderArg()) and
+        tag = "CookieRawHeader"
+        or
+        element = cookieWrite.toString() and
+        value = prettyNodeForInlineTest(cookieWrite.getNameArg()) and
+        tag = "CookieName"
+        or
+        element = cookieWrite.toString() and
+        value = prettyNodeForInlineTest(cookieWrite.getValueArg()) and
+        tag = "CookieValue"
+      )
+    )
+  }
+}
+
 class FileSystemAccessTest extends InlineExpectationsTest {
   FileSystemAccessTest() { this = "FileSystemAccessTest" }
 
@@ -269,6 +350,23 @@ class FileSystemAccessTest extends InlineExpectationsTest {
   }
 }
 
+class FileSystemWriteAccessTest extends InlineExpectationsTest {
+  FileSystemWriteAccessTest() { this = "FileSystemWriteAccessTest" }
+
+  override string getARelevantTag() { result = "fileWriteData" }
+
+  override predicate hasActualResult(Location location, string element, string tag, string value) {
+    exists(location.getFile().getRelativePath()) and
+    exists(FileSystemWriteAccess write, DataFlow::Node data |
+      data = write.getADataNode() and
+      location = data.getLocation() and
+      element = data.toString() and
+      value = prettyNodeForInlineTest(data) and
+      tag = "fileWriteData"
+    )
+  }
+}
+
 class PathNormalizationTest extends InlineExpectationsTest {
   PathNormalizationTest() { this = "PathNormalizationTest" }
 

python/ql/test/experimental/query-tests/Security/CWE-287/ImproperLdapAuth.expected

@@ -0,0 +1,10 @@
+| auth_bad_2.py:19:5:19:42 | ControlFlowNode for Attribute() | The following LDAP bind operation is executed without authentication |
+| auth_bad_2.py:33:5:33:44 | ControlFlowNode for Attribute() | The following LDAP bind operation is executed without authentication |
+| auth_bad_2.py:47:5:47:43 | ControlFlowNode for Attribute() | The following LDAP bind operation is executed without authentication |
+| auth_bad_2.py:60:5:60:52 | ControlFlowNode for Attribute() | The following LDAP bind operation is executed without authentication |
+| auth_bad_2.py:73:5:73:39 | ControlFlowNode for Attribute() | The following LDAP bind operation is executed without authentication |
+| auth_bad_2.py:87:5:87:48 | ControlFlowNode for Attribute() | The following LDAP bind operation is executed without authentication |
+| auth_bad_3.py:19:12:19:43 | ControlFlowNode for Connection() | The following LDAP bind operation is executed without authentication |
+| auth_bad_3.py:33:12:33:57 | ControlFlowNode for Connection() | The following LDAP bind operation is executed without authentication |
+| auth_bad_3.py:46:12:46:55 | ControlFlowNode for Connection() | The following LDAP bind operation is executed without authentication |
+| auth_bad_3.py:60:12:60:42 | ControlFlowNode for Connection() | The following LDAP bind operation is executed without authentication |

python/ql/test/experimental/query-tests/Security/CWE-287/ImproperLdapAuth.qlref

@@ -0,0 +1 @@
+experimental/Security/CWE-287/ImproperLdapAuth.ql

python/ql/test/experimental/query-tests/Security/CWE-287/auth_bad_2.py

@@ -0,0 +1,92 @@
+from flask import request, Flask
+import ldap
+import ldap.filter
+import ldap.dn
+
+app = Flask(__name__)
+
+
+@app.route("/simple_bind_example")
+def simple_bind_example():
+    """
+    The bind's password argument is not set
+    """
+
+    dn = "dc={}".format(ldap.dn.escape_dn_chars(request.args['dc']))
+    search_filter = "(user={})".format(ldap.filter.escape_filter_chars(request.args['search']))
+
+    ldap_connection = ldap.initialize("ldap://127.0.0.1:1337")
+    ldap_connection.simple_bind('cn=root')
+    user = ldap_connection.search_s(dn, ldap.SCOPE_SUBTREE, search_filter)
+
+
+@app.route("/simple_bind_s_example")
+def simple_bind_s_example():
+    """
+    The bind's password argument is not set
+    """
+
+    dn = "dc={}".format(ldap.dn.escape_dn_chars(request.args['dc']))
+    search_filter = "(user={})".format(ldap.filter.escape_filter_chars(request.args['search']))
+
+    ldap_connection = ldap.initialize("ldap://127.0.0.1:1337")
+    ldap_connection.simple_bind_s('cn=root')
+    user = ldap_connection.search_s(dn, ldap.SCOPE_SUBTREE, search_filter)
+
+
+@app.route("/bind_s_example")
+def bind_s_example():
+    """
+    The bind's password argument is set to None
+    """
+
+    dn = "dc={}".format(ldap.dn.escape_dn_chars(request.args['dc']))
+    search_filter = "(user={})".format(ldap.filter.escape_filter_chars(request.args['search']))
+
+    ldap_connection = ldap.initialize("ldap://127.0.0.1:1337")
+    ldap_connection.bind_s('cn=root', None)
+    user = ldap_connection.search_s(dn, ldap.SCOPE_SUBTREE, search_filter)
+
+@app.route("/bind_s_example")
+def bind_s_example_kwargs():
+    """
+    The bind's password argument is set to None
+    """
+
+    dn = "dc={}".format(ldap.dn.escape_dn_chars(request.args['dc']))
+    search_filter = "(user={})".format(ldap.filter.escape_filter_chars(request.args['search']))
+
+    ldap_connection = ldap.initialize("ldap://127.0.0.1:1337")
+    ldap_connection.bind_s(who='cn=root', cred=None)
+    user = ldap_connection.search_s(dn, ldap.SCOPE_SUBTREE, search_filter)
+
+@app.route("/bind_example")
+def bind_example():
+    """
+    The bind's password argument is an empty string
+    """
+
+    dn = "dc={}".format(ldap.dn.escape_dn_chars(request.args['dc']))
+    search_filter = "(user={})".format(ldap.filter.escape_filter_chars(request.args['search']))
+
+    ldap_connection = ldap.initialize("ldap://127.0.0.1:1337")
+    ldap_connection.bind('cn=root', "")
+    user = ldap_connection.search_s(dn, ldap.SCOPE_SUBTREE, search_filter)
+
+
+@app.route("/bind_example")
+def bind_example():
+    """
+    The bind's password argument is an empty string
+    """
+
+    dn = "dc={}".format(ldap.dn.escape_dn_chars(request.args['dc']))
+    search_filter = "(user={})".format(ldap.filter.escape_filter_chars(request.args['search']))
+
+    ldap_connection = ldap.initialize("ldap://127.0.0.1:1337")
+    ldap_connection.bind(who='cn=root', cred="")
+    user = ldap_connection.search_s(dn, ldap.SCOPE_SUBTREE, search_filter)
+
+
+# if __name__ == "__main__":
+#     app.run(debug=True)

python/ql/test/experimental/query-tests/Security/CWE-287/auth_bad_3.py

@@ -0,0 +1,65 @@
+from ldap3 import Server, Connection, ALL
+from flask import request, Flask
+from ldap3.utils.dn import escape_rdn
+from ldap3.utils.conv import escape_filter_chars
+
+app = Flask(__name__)
+
+
+@app.route("/passwordNone")
+def passwordNone():
+    """
+    The bind's password argument is set to None
+    """
+
+    dn = "dc={}".format(escape_rdn(request.args['dc']))
+    search_filter = "(user={})".format(escape_filter_chars(request.args['search']))
+
+    srv = Server('servername', get_info=ALL)
+    conn = Connection(srv, 'user_dn', None)
+    status, result, response, _ = conn.search(dn, search_filter)
+
+
+@app.route("/passwordNone")
+def passwordNoneKwargs():
+    """
+    The bind's password argument is set to None
+    """
+
+    dn = "dc={}".format(escape_rdn(request.args['dc']))
+    search_filter = "(user={})".format(escape_filter_chars(request.args['search']))
+
+    srv = Server('servername', get_info=ALL)
+    conn = Connection(srv, user='user_dn', password=None)
+    status, result, response, _ = conn.search(dn, search_filter)
+
+@app.route("/passwordEmpty")
+def passwordEmpty():
+    """
+    The bind's password argument is an empty string
+    """
+
+    dn = "dc={}".format(escape_rdn(request.args['dc']))
+    search_filter = "(user={})".format(escape_filter_chars(request.args['search']))
+
+    srv = Server('servername', get_info=ALL)
+    conn = Connection(srv, user='user_dn', password="")
+    status, result, response, _ = conn.search(dn, search_filter)
+
+
+@app.route("/notPassword")
+def notPassword():
+    """
+    The bind's password argument is not set
+    """
+
+    dn = "dc={}".format(escape_rdn(request.args['dc']))
+    search_filter = "(user={})".format(escape_filter_chars(request.args['search']))
+
+    srv = Server('servername', get_info=ALL)
+    conn = Connection(srv, user='user_dn')
+    status, result, response, _ = conn.search(dn, search_filter)
+
+
+# if __name__ == "__main__":
+#     app.run(debug=True)

python/ql/test/experimental/query-tests/Security/CWE-287/auth_good_2.py

@@ -0,0 +1,65 @@
+from flask import request, Flask
+import ldap
+import ldap.filter
+import ldap.dn
+
+app = Flask(__name__)
+
+
+@app.route("/simple_bind_example")
+def simple_bind_example():
+    """
+    The bind's password argument is a non-empty string
+    """
+
+    dn = "dc={}".format(ldap.dn.escape_dn_chars(request.args['dc']))
+    search_filter = "(user={})".format(ldap.filter.escape_filter_chars(request.args['search']))
+
+    ldap_connection = ldap.initialize("ldap://127.0.0.1:1337")
+    ldap_connection.simple_bind('cn=root', "SecurePa$$!")
+    user = ldap_connection.search_s(dn, ldap.SCOPE_SUBTREE, search_filter)
+
+
+@app.route("/simple_bind_s_example")
+def simple_bind_s_example():
+    """
+    The bind's password argument is a non-empty string
+    """
+
+    dn = "dc={}".format(ldap.dn.escape_dn_chars(request.args['dc']))
+    search_filter = "(user={})".format(ldap.filter.escape_filter_chars(request.args['search']))
+
+    ldap_connection = ldap.initialize("ldap://127.0.0.1:1337")
+    ldap_connection.simple_bind_s('cn=root', "SecurePa$$!")
+    user = ldap_connection.search_s(dn, ldap.SCOPE_SUBTREE, search_filter)
+
+
+@app.route("/bind_s_example")
+def bind_s_example():
+    """
+    The bind's password argument is a non-empty string
+    """
+
+    dn = "dc={}".format(ldap.dn.escape_dn_chars(request.args['dc']))
+    search_filter = "(user={})".format(ldap.filter.escape_filter_chars(request.args['search']))
+
+    ldap_connection = ldap.initialize("ldap://127.0.0.1:1337")
+    ldap_connection.bind_s('cn=root', "SecurePa$$!")
+    user = ldap_connection.search_s(dn, ldap.SCOPE_SUBTREE, search_filter)
+
+
+@app.route("/bind_example")
+def bind_example():
+    """
+    The bind's password argument is a non-empty string
+    """
+
+    dn = "dc={}".format(ldap.dn.escape_dn_chars(request.args['dc']))
+    search_filter = "(user={})".format(ldap.filter.escape_filter_chars(request.args['search']))
+
+    ldap_connection = ldap.initialize("ldap://127.0.0.1:1337")
+    ldap_connection.bind('cn=root', "SecurePa$$!")
+    user = ldap_connection.search_s(dn, ldap.SCOPE_SUBTREE, search_filter)
+
+# if __name__ == "__main__":
+#     app.run(debug=True)

python/ql/test/experimental/query-tests/Security/CWE-287/auth_good_3.py

@@ -0,0 +1,24 @@
+from ldap3 import Server, Connection, ALL
+from flask import request, Flask
+from ldap3.utils.dn import escape_rdn
+from ldap3.utils.conv import escape_filter_chars
+
+app = Flask(__name__)
+
+
+@app.route("/passwordFromEnv")
+def passwordFromEnv():
+    """
+    The bind's password argument is a non-empty string
+    """
+
+    dn = "dc={}".format(escape_rdn(request.args['dc']))
+    search_filter = "(user={})".format(escape_filter_chars(request.args['search']))
+
+    srv = Server('servername', get_info=ALL)
+    conn = Connection(srv, user='user_dn',
+                      password="SecurePa$$!")
+    status, result, response, _ = conn.search(dn, search_filter)
+
+# if __name__ == "__main__":
+#     app.run(debug=True)