Fixed LDAP Auth (#2333)

mitmproxy/addons/proxyauth.py

@@ -47,12 +47,13 @@ class ProxyAuth:
         self.nonanonymous = False
         self.htpasswd = None
         self.singleuser = None
+        self.ldapconn = None
         self.ldapserver = None
         self.authenticated = weakref.WeakKeyDictionary()  # type: MutableMapping[connections.ClientConnection, Tuple[str, str]]
         """Contains all connections that are permanently authenticated after an HTTP CONNECT"""
 
     def enabled(self) -> bool:
-        return any([self.nonanonymous, self.htpasswd, self.singleuser, self.ldapserver])
+        return any([self.nonanonymous, self.htpasswd, self.singleuser, self.ldapconn, self.ldapserver])
 
     def is_proxy_auth(self) -> bool:
         """
@@ -101,19 +102,17 @@ class ProxyAuth:
         elif self.htpasswd:
             if self.htpasswd.check_password(username, password):
                 return username, password
-        elif self.ldapserver:
+        elif self.ldapconn:
             if not username or not password:
                 return None
-            dn = ctx.options.proxyauth.split(":")[2]
+            self.ldapconn.search(ctx.options.proxyauth.split(':')[4], '(cn=' + username + ')')
-            parts = dn.split("?")
+            if self.ldapconn.response:
-            conn = ldap3.Connection(
+                conn = ldap3.Connection(
-                self.ldapserver,
+                    self.ldapserver,
-                parts[0] + username + parts[1],
+                    self.ldapconn.response[0]['dn'],
-                password,
+                    password,
-                auto_bind=True)
+                    auto_bind=True)
-            if conn:
+                if conn:
-                conn.search(parts[1][1:], '(' + parts[0] + username + ')', attributes=['objectclass'])
-                if ctx.options.proxyauth.split(":")[3] in conn.entries[0]['objectclass']:
                     return username, password
         return None
 
@@ -146,19 +145,29 @@ class ProxyAuth:
                             "Could not open htpasswd file: %s" % p
                         )
                 elif ctx.options.proxyauth.startswith("ldap"):
-                    parts = ctx.options.proxyauth.split(":")
+                    parts = ctx.options.proxyauth.split(':')
-                    if len(parts) != 4:
+                    security = parts[0]
+                    ldap_server = parts[1]
+                    dn_baseauth = parts[2]
+                    password_baseauth = parts[3]
+                    if len(parts) != 5:
                         raise exceptions.OptionsError(
                             "Invalid ldap specification"
                         )
-                    if parts[0] == "ldaps":
+                    if security == "ldaps":
-                        server = ldap3.Server(parts[1], use_ssl=True)
+                        server = ldap3.Server(ldap_server, use_ssl=True)
-                    elif parts[0] == "ldap":
+                    elif security == "ldap":
-                        server = ldap3.Server(parts[1])
+                        server = ldap3.Server(ldap_server)
                     else:
                         raise exceptions.OptionsError(
                             "Invalid ldap specfication on the first part"
                         )
+                    conn = ldap3.Connection(
+                        server,
+                        dn_baseauth,
+                        password_baseauth,
+                        auto_bind=True)
+                    self.ldapconn = conn
                     self.ldapserver = server
                 else:
                     parts = ctx.options.proxyauth.split(':')

mitmproxy/options.py

@@ -206,10 +206,11 @@ class Options(optmanager.OptManager):
             authenticaiton but accept any credentials, start with "@" to specify
             a path to an Apache htpasswd file, be of the form
             "username:password", or be of the form
-            "ldap[s]:url_server_ldap:dn:group", the dn must include "?", which will be
+            "ldap[s]:url_server_ldap:dn_auth:password:dn_subtree",
-            the username prompted, and the group is the group the user must belong to
+            the dn_auth & password is the dn/pass used to authenticate
+            the dn subtree is the subtree that we will search to find the username
             an example would be
-            "ldap:ldap.forumsys.com:uid=?,dc=example,dc=com:person".
+            "ldap:localhost:cn=default,dc=example,dc=com:password:ou=application,dc=example,dc=com".
             """
         )
         self.add_option(

test/mitmproxy/addons/test_proxyauth.py

@@ -1,8 +1,8 @@
 import binascii
-import ldap3
 
 import pytest
 
+from unittest import mock
 from mitmproxy import exceptions
 from mitmproxy.addons import proxyauth
 from mitmproxy.test import taddons
@@ -42,19 +42,21 @@ def test_configure():
         ctx.configure(up, proxyauth=None)
         assert not up.nonanonymous
 
-        ctx.configure(up, proxyauth="ldap:fake_server:fake_dn:fake_group")
+        with mock.patch('ldap3.Server', return_value="ldap://fake_server:389 - cleartext"):
-        assert up.ldapserver
+            with mock.patch('ldap3.Connection', return_value="test"):
-
+                ctx.configure(up, proxyauth="ldap:localhost:cn=default,dc=cdhdt,dc=com:password:ou=application,dc=cdhdt,dc=com")
-        ctx.configure(up, proxyauth="ldap:fake_server:uid=?,dc=example,dc=com:person")
+                assert up.ldapserver
-        assert up.ldapserver
+                ctx.configure(up, proxyauth="ldaps:localhost:cn=default,dc=cdhdt,dc=com:password:ou=application,dc=cdhdt,dc=com")
-        ctx.configure(up, proxyauth="ldaps:fake_server.com:uid=?,dc=example,dc=com:person")
+                assert up.ldapserver
-        assert up.ldapserver
 
         with pytest.raises(exceptions.OptionsError):
+            ctx.configure(up, proxyauth="ldap:test:test:test")
+
+        with pytest.raises(IndexError):
             ctx.configure(up, proxyauth="ldap:fake_serveruid=?dc=example,dc=com:person")
 
         with pytest.raises(exceptions.OptionsError):
-            ctx.configure(up, proxyauth="ldapssssssss:fake_server.com:uid=?,dc=example,dc=com:person")
+            ctx.configure(up, proxyauth="ldapssssssss:fake_server:dn:password:tree")
 
         with pytest.raises(exceptions.OptionsError):
             ctx.configure(
@@ -124,26 +126,21 @@ def test_check(monkeypatch):
         )
         assert not up.check(f)
 
-        ctx.configure(
+        with mock.patch('ldap3.Server', return_value="ldap://fake_server:389 - cleartext"):
-            up,
+            with mock.patch('ldap3.Connection', search="test"):
-            proxyauth="ldap:fake-server:cn=?,ou=test,o=lab:test"
+                with mock.patch('ldap3.Connection.search', return_value="test"):
-        )
+                    ctx.configure(
-        conn = ldap3.Connection("fake-server", user="cn=user0,ou=test,o=lab", password="password", client_strategy=ldap3.MOCK_SYNC)
+                        up,
-        conn.bind()
+                        proxyauth="ldap:localhost:cn=default,dc=cdhdt,dc=com:password:ou=application,dc=cdhdt,dc=com"
-        conn.strategy.add_entry('cn=user0,ou=test,o=lab', {'userPassword': 'test0', 'sn': 'user0_sn', 'revision': 0, 'objectClass': 'test'})
+                    )
-
+                    f.request.headers["Proxy-Authorization"] = proxyauth.mkauth(
-        def conn_mp(ldap, user, password, **kwargs):
+                        "test", "test"
-            return conn
+                    )
-
+                    assert up.check(f)
-        monkeypatch.setattr(ldap3, "Connection", conn_mp)
+                    f.request.headers["Proxy-Authorization"] = proxyauth.mkauth(
-        f.request.headers["Proxy-Authorization"] = proxyauth.mkauth(
+                        "", ""
-            "user0", "test0"
+                    )
-        )
+                    assert not up.check(f)
-        assert up.check(f)
-        f.request.headers["Proxy-Authorization"] = proxyauth.mkauth(
-            "", ""
-        )
-        assert not up.check(f)
 
 
 def test_authenticate():