Python: Add LDAP Injection query

python/ql/src/experimental/Security/CWE-090/LDAPInjection.qhelp

@@ -0,0 +1,28 @@
+<!DOCTYPE qhelp PUBLIC "-//Semmle//qhelp//EN" "qhelp.dtd">
+
+<qhelp>
+  <overview>
+    <p>If an LDAP query is built by a not sanitized user-provided value, a user is likely to be able to run malicious LDAP queries.</p>
+  </overview>
+
+  <recommendation>
+    <p>In case user input must compose an LDAP query, it should be escaped in order to avoid a malicious user supplying special characters that change the actual purpose of the query. To do so, functions that ldap frameworks provide such as <code>escape_filter_chars</code> should be applied to that user input.
+  <recommendation>
+
+
+  <references>
+    <li>
+      OWASP
+      <a href="https://owasp.org/www-community/attacks/LDAP_Injection">LDAP Injection</a>
+    </li>
+    <li>
+      SonarSource
+      <a href="https://rules.sonarsource.com/python/RSPEC-2078">RSPEC-2078</a>
+    </li>
+    <li>
+      Python
+      <a href="https://www.python-ldap.org/en/python-ldap-3.3.0/reference/ldap.html">LDAP Documentation</a>
+    </li>
+  </references>
+
+</qhelp>

python/ql/src/experimental/Security/CWE-090/LDAPInjection.ql

@@ -0,0 +1,24 @@
+/**
+ * @name Python LDAP Injection
+ * @description Python LDAP Injection through search filter
+ * @kind path-problem
+ * @problem.severity error
+ * @id python/ldap-injection
+ * @tags experimental	
+ *       security	
+ *       external/cwe/cwe-090
+ */
+
+// Determine precision above
+import python
+import experimental.semmle.python.security.injection.LDAPInjection
+import DataFlow::PathGraph
+
+from
+  LDAPInjectionFlowConfig config, DataFlow::PathNode source, DataFlow::PathNode sink,
+  LDAPInjectionSink castedSink
+where
+  config.hasFlowPath(source, sink) and
+  castedSink.getLDAPNode() = sink.getNode()
+select sink.getNode(), source, sink, "$@ LDAP query executes $@ as a $@.", castedSink, "This",
+  source.getNode(), "a user-provided value", castedSink.getLDAPNode(), castedSink.getLDAPPart()

python/ql/src/experimental/Security/CWE-090/unit_tests/ldap3_bad.py

@@ -0,0 +1,38 @@
+from flask import request, Flask
+import ldap3
+
+app = Flask(__name__)
+
+
+@app.route("/normal")
+def normal():
+    unsafe_dn = "dc=%s" % request.args['dc']
+    unsafe_filter = "(user=%s)" % request.args['username']
+
+    srv = ldap3.Server('ldap://127.0.0.1', port=1337)
+    conn = ldap3.Connection(srv, user=unsafe_dn, auto_bind=True)
+    conn.search(unsafe_dn, unsafe_filter, attributes=[
+                "testAttr1", "testAttr2"])
+
+
+@app.route("/normal_noAttrs")
+def normal_noAttrs():
+    unsafe_dn = "dc=%s" % request.args['dc']
+    unsafe_filter = "(user=%s)" % request.args['username']
+
+    srv = ldap3.Server('ldap://127.0.0.1', port=1337)
+    conn = ldap3.Connection(srv, user=unsafe_dn, auto_bind=True)
+    conn.search(unsafe_dn, unsafe_filter)
+
+
+@app.route("/direct")
+def direct():
+    unsafe_dn = "dc=%s" % request.args['dc']
+    unsafe_filter = "(user=%s)" % request.args['username']
+
+    srv = ldap3.Server('ldap://127.0.0.1', port=1337)
+    conn = ldap3.Connection(srv, user=unsafe_dn, auto_bind=True).search(unsafe_dn, unsafe_filter, attributes=[
+        "testAttr1", "testAttr2"])
+
+# if __name__ == "__main__":
+#     app.run(debug=True)

python/ql/src/experimental/Security/CWE-090/unit_tests/ldap3_good.py

@@ -0,0 +1,49 @@
+from flask import request, Flask
+import ldap3
+from ldap3.utils.dn import escape_rdn
+from ldap3.utils.conv import escape_filter_chars
+
+app = Flask(__name__)
+
+
+@app.route("/normal")
+def normal():
+    unsafe_dn = "dc=%s" % request.args['dc']
+    unsafe_filter = "(user=%s)" % request.args['username']
+
+    safe_dn = escape_rdn(unsafe_dn)
+    safe_filter = escape_filter_chars(unsafe_filter)
+
+    srv = ldap3.Server('ldap://127.0.0.1', port=1337)
+    conn = ldap3.Connection(srv, user=unsafe_dn, auto_bind=True)
+    conn.search(safe_dn, safe_filter, attributes=[
+                "testAttr1", "testAttr2"])
+
+
+@app.route("/normal_noAttrs")
+def normal_noAttrs():
+    unsafe_dn = "dc=%s" % request.args['dc']
+    unsafe_filter = "(user=%s)" % request.args['username']
+
+    safe_dn = escape_rdn(unsafe_dn)
+    safe_filter = escape_filter_chars(unsafe_filter)
+
+    srv = ldap3.Server('ldap://127.0.0.1', port=1337)
+    conn = ldap3.Connection(srv, user=unsafe_dn, auto_bind=True)
+    conn.search(safe_dn, safe_filter)
+
+
+@app.route("/direct")
+def direct():
+    unsafe_dn = "dc=%s" % request.args['dc']
+    unsafe_filter = "(user=%s)" % request.args['username']
+
+    safe_dn = escape_rdn(unsafe_dn)
+    safe_filter = escape_filter_chars(unsafe_filter)
+
+    srv = ldap3.Server('ldap://127.0.0.1', port=1337)
+    conn = ldap3.Connection(srv, user=unsafe_dn, auto_bind=True).search(safe_dn, safe_filter, attributes=[
+        "testAttr1", "testAttr2"])
+
+# if __name__ == "__main__":
+#     app.run(debug=True)

python/ql/src/experimental/Security/CWE-090/unit_tests/ldap_bad.py

@@ -0,0 +1,56 @@
+from flask import request, Flask
+import ldap
+
+app = Flask(__name__)
+
+
+@app.route("/normal")
+def normal():
+    unsafe_dn = "dc=%s" % request.args['dc']
+    unsafe_filter = "(user=%s)" % request.args['username']
+
+    ldap_connection = ldap.initialize("ldap://127.0.0.1:1337")
+    user = ldap_connection.search_s(
+        unsafe_dn, ldap.SCOPE_SUBTREE, unsafe_filter, ["testAttr1", "testAttr2"])
+
+
+@app.route("/normal_noAttrs")
+def normal_noAttrs():
+    unsafe_dn = "dc=%s" % request.args['dc']
+    unsafe_filter = "(user=%s)" % request.args['username']
+
+    ldap_connection = ldap.initialize("ldap://127.0.0.1:1337")
+    user = ldap_connection.search_s(
+        unsafe_dn, ldap.SCOPE_SUBTREE, unsafe_filter)
+
+
+@app.route("/direct")
+def direct():
+    unsafe_dn = "dc=%s" % request.args['dc']
+    unsafe_filter = "(user=%s)" % request.args['username']
+
+    user = ldap.initialize("ldap://127.0.0.1:1337").search_s(
+        unsafe_dn, ldap.SCOPE_SUBTREE, unsafe_filter, ["testAttr1", "testAttr2"])
+
+
+@app.route("/normal_argbyname")
+def normal_argbyname():
+    unsafe_dn = "dc=%s" % request.args['dc']
+    unsafe_filter = "(user=%s)" % request.args['username']
+
+    ldap_connection = ldap.initialize("ldap://127.0.0.1:1337")
+    user = ldap_connection.search_s(
+        unsafe_dn, ldap.SCOPE_SUBTREE, attrlist=["testAttr1", "testAttr2"], filterstr=unsafe_filter)
+
+
+@app.route("/direct_argbyname")
+def direct_argbyname():
+    unsafe_dn = "dc=%s" % request.args['dc']
+    unsafe_filter = "(user=%s)" % request.args['username']
+
+    user = ldap.initialize("ldap://127.0.0.1:1337").search_s(
+        unsafe_dn, ldap.SCOPE_SUBTREE, attrlist=["testAttr1", "testAttr2"], filterstr=unsafe_filter)
+
+
+# if __name__ == "__main__":
+#     app.run(debug=True)

python/ql/src/experimental/Security/CWE-090/unit_tests/ldap_good.py

@@ -0,0 +1,73 @@
+from flask import request, Flask
+import ldap
+import ldap.filter
+import ldap.dn
+
+app = Flask(__name__)
+
+
+@app.route("/normal")
+def normal():
+    unsafe_dn = "dc=%s" % request.args['dc']
+    unsafe_filter = "(user=%s)" % request.args['username']
+
+    safe_dn = ldap.dn.escape_dn_chars(unsafe_dn)
+    safe_filter = ldap.filter.escape_filter_chars(unsafe_filter)
+
+    ldap_connection = ldap.initialize("ldap://127.0.0.1:1337")
+    user = ldap_connection.search_s(
+        safe_dn, ldap.SCOPE_SUBTREE, safe_filter, ["testAttr1", "testAttr2"])
+
+
+@app.route("/normal_noAttrs")
+def normal_noAttrs():
+    unsafe_dn = "dc=%s" % request.args['dc']
+    unsafe_filter = "(user=%s)" % request.args['username']
+
+    safe_dn = ldap.dn.escape_dn_chars(unsafe_dn)
+    safe_filter = ldap.filter.escape_filter_chars(unsafe_filter)
+
+    ldap_connection = ldap.initialize("ldap://127.0.0.1:1337")
+    user = ldap_connection.search_s(
+        safe_dn, ldap.SCOPE_SUBTREE, safe_filter)
+
+
+@app.route("/direct")
+def direct():
+    unsafe_dn = "dc=%s" % request.args['dc']
+    unsafe_filter = "(user=%s)" % request.args['username']
+
+    safe_dn = ldap.dn.escape_dn_chars(unsafe_dn)
+    safe_filter = ldap.filter.escape_filter_chars(unsafe_filter)
+
+    user = ldap.initialize("ldap://127.0.0.1:1337").search_s(
+        safe_dn, ldap.SCOPE_SUBTREE, safe_filter, ["testAttr1", "testAttr2"])
+
+
+@app.route("/normal_argbyname")
+def normal_argbyname():
+    unsafe_dn = "dc=%s" % request.args['dc']
+    unsafe_filter = "(user=%s)" % request.args['username']
+
+    safe_dn = ldap.dn.escape_dn_chars(unsafe_dn)
+    safe_filter = ldap.filter.escape_filter_chars(unsafe_filter)
+
+    ldap_connection = ldap.initialize("ldap://127.0.0.1:1337")
+    user = ldap_connection.search_s(
+        safe_dn, ldap.SCOPE_SUBTREE, attrlist=["testAttr1", "testAttr2"], filterstr=safe_filter)
+
+
+@app.route("/direct_argbyname")
+def direct_argbyname():
+    unsafe_dn = "dc=%s" % request.args['dc']
+    unsafe_filter = "(user=%s)" % request.args['username']
+
+    safe_dn = ldap.dn.escape_dn_chars(unsafe_dn)
+    safe_filter = ldap.filter.escape_filter_chars(unsafe_filter)
+
+    user = ldap.initialize("ldap://127.0.0.1:1337").search_s(
+        safe_dn, ldap.SCOPE_SUBTREE, attrlist=["testAttr1", "testAttr2"], filterstr=safe_filter)
+
+
+# if __name__ == "__main__":
+#     app.run(debug=True)

python/ql/src/experimental/semmle/python/Concepts.qll

@@ -13,3 +13,36 @@ private import semmle.python.dataflow.new.DataFlow
 private import semmle.python.dataflow.new.RemoteFlowSources
 private import semmle.python.dataflow.new.TaintTracking
 private import experimental.semmle.python.Frameworks
+private import semmle.python.ApiGraphs
+
+module LDAPQuery {
+  abstract class Range extends DataFlow::Node {
+    abstract DataFlow::Node getLDAPNode();
+
+    abstract string getLDAPPart();
+  }
+}
+
+class LDAPQuery extends DataFlow::Node {
+  LDAPQuery::Range range;
+
+  LDAPQuery() { this = range }
+
+  DataFlow::Node getLDAPNode() { result = range.getLDAPNode() }
+
+  string getLDAPPart() { result = range.getLDAPPart() }
+}
+
+module LDAPEscape {
+  abstract class Range extends DataFlow::Node {
+    abstract DataFlow::Node getEscapeNode();
+  }
+}
+
+class LDAPEscape extends DataFlow::Node {
+  LDAPEscape::Range range;
+
+  LDAPEscape() { this = range }
+
+  DataFlow::Node getEscapeNode() { result = range.getEscapeNode() }
+}