Polish up ldap3 tests

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

@@ -1,56 +0,0 @@
-import ldap3
-from flask import request, Flask
-
-app = Flask(__name__)
-
-
-@app.route("/tainted_var")
-def tainted_var():
-    unsanitized_dn = "dc=%s" % request.args['dc']
-    unsanitized_filter = "(&(objectClass=*)(uid=%s))" % request.args['username']
-
-    srv = ldap3.Server('localhost', port=1337)
-    conn = ldap3.Connection(srv, user=unsanitized_dn, auto_bind=True)
-    conn.search(unsanitized_dn, unsanitized_filter)
-    return conn.response
-
-
-@app.route("/var_tainted")
-def var_tainted():
-    unsanitized_dn = request.args['dc']
-    unsanitized_filter = request.args['username']
-
-    dn = "dc=%s" % unsanitized_dn
-    search_filter = "(&(objectClass=*)(uid=%s))" % unsanitized_filter
-
-    srv = ldap3.Server('localhost', port=1337)
-    conn = ldap3.Connection(srv, user=dn, auto_bind=True)
-    conn.search(dn, search_filter)
-    return conn.response
-
-
-@app.route("/direct")
-def direct():
-    srv = ldap3.Server('localhost', port=1337)
-    conn = ldap3.Connection(srv, user="dc=%s" %
-                            request.args['dc'], auto_bind=True)
-    conn.search("dc=%s" % unsanitized_dn,
-                "(&(objectClass=*)(uid=%s))" % request.args['username'])
-    return conn.response
-
-
-@app.route("/with_")
-def with_():
-    unsanitized_dn = request.args['dc']
-    unsanitized_filter = request.args['username']
-
-    dn = "dc=%s" % unsanitized_dn
-    search_filter = "(&(objectClass=*)(uid=%s))" % unsanitized_filter
-
-    srv = ldap3.Server('localhost', port=1337)
-    with ldap3.Connection(server, auto_bind=True) as conn:
-        conn.search(dn, search_filter)
-        return conn.response
-
-# if __name__ == "__main__":
-#     app.run(debug=True)

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

@@ -1,58 +0,0 @@
-import ldap3
-from ldap3.utils.conv import escape_filter_chars
-from flask import request, Flask
-
-app = Flask(__name__)
-
-
-@app.route("/tainted_var")
-def tainted_var():
-    sanitized_dn = "dc=%s" % request.args['dc']
-    sanitized_filter = "(&(objectClass=*)(uid=%s))" % escape_filter_chars(
-        request.args['username'])
-
-    srv = ldap3.Server('localhost', port=1337)
-    conn = ldap3.Connection(srv, user=sanitized_dn, auto_bind=True)
-    conn.search(sanitized_dn, sanitized_filter)
-    return conn.response
-
-
-@app.route("/var_tainted")
-def var_tainted():
-    sanitized_dn = request.args['dc']
-    sanitized_filter = request.args['username']
-
-    dn = "dc=%s" % sanitized_dn
-    search_filter = "(&(objectClass=*)(uid=%s))" % escape_filter_chars(sanitized_filter)
-
-    srv = ldap3.Server('localhost', port=1337)
-    conn = ldap3.Connection(srv, user=dn, auto_bind=True)
-    conn.search(dn, search_filter)
-    return conn.response
-
-
-@app.route("/direct")
-def direct():
-    srv = ldap3.Server('localhost', port=1337)
-    conn = ldap3.Connection(srv, user="dc=%s" %
-                            request.args['dc'], auto_bind=True)
-    conn.search("dc=%s" % request.args['dc'], "(&(objectClass=*)(uid=%s))" %
-                escape_filter_chars(request.args['username']))
-    return conn.response
-
-
-@ app.route("/with_")
-def with_():
-    sanitized_dn = request.args['dc']
-    sanitized_filter = escape_filter_chars(request.args['username'])
-
-    dn = "dc=%s" % sanitized_dn
-    search_filter = "(&(objectClass=*)(uid=%s))" % sanitized_filter
-
-    srv = ldap3.Server('localhost', port=1337)
-    with ldap3.Connection(server, auto_bind=True) as conn:
-        conn.search(dn, search_filter)
-        return conn.response
-
-# if __name__ == "__main__":
-#     app.run(debug=True)

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)