ISSUE_5068 (#5161)

* changes for custom port number

* indent correction

* test coverage

* coverage correction

* simplify LDAP auth

* make mypy hapy

Co-authored-by: Maximilian Hils <git@maximilianhils.com>

CHANGELOG.md

@@ -30,6 +30,7 @@
 * Allow addon hooks to be async (@nneonneo, #4207)
 * Reintroduce `Flow.live`, which signals if a flow belongs to a currently active connection. (@mhils, #4207)
 * Speculative fix for some rare HTTP/2 connection stalls (#5158, @EndUser509)
+* Add ability to specify custom ports with LDAP authentication (#5068, @demonoidvk)
 * Console Improvements on Windows (@mhils)
 
 ## 28 September 2021: mitmproxy 7.0.4

mitmproxy/addons/proxyauth.py

@@ -34,7 +34,7 @@ class ProxyAuth:
             "username:pass",
             "any" to accept any user/pass combination,
             "@path" to use an Apache htpasswd file,
-            or "ldap[s]:url_server_ldap:dn_auth:password:dn_subtree" for LDAP authentication.
+            or "ldap[s]:url_server_ldap[:port]:dn_auth:password:dn_subtree" for LDAP authentication.
             """
         )
 
@@ -207,16 +207,15 @@ class Ldap(Validator):
     dn_subtree: str
 
     def __init__(self, proxyauth: str):
-        try:
-            security, url, ldap_user, ldap_pass, self.dn_subtree = proxyauth.split(":")
-        except ValueError:
-            raise exceptions.OptionsError("Invalid ldap specification")
-        if security == "ldaps":
-            server = ldap3.Server(url, use_ssl=True)
-        elif security == "ldap":
-            server = ldap3.Server(url)
-        else:
-            raise exceptions.OptionsError("Invalid ldap specification on the first part")
+        (
+            use_ssl,
+            url,
+            port,
+            ldap_user,
+            ldap_pass,
+            self.dn_subtree,
+        ) = self.parse_spec(proxyauth)
+        server = ldap3.Server(url, port=port, use_ssl=use_ssl)
         conn = ldap3.Connection(
             server,
             ldap_user,
@@ -226,6 +225,34 @@ class Ldap(Validator):
         self.conn = conn
         self.server = server
 
+    @staticmethod
+    def parse_spec(spec: str) -> Tuple[bool, str, Optional[int], str, str, str]:
+        try:
+            if spec.count(":") > 4:
+                (
+                    security,
+                    url,
+                    port_str,
+                    ldap_user,
+                    ldap_pass,
+                    dn_subtree,
+                ) = spec.split(":")
+                port = int(port_str)
+            else:
+                security, url, ldap_user, ldap_pass, dn_subtree = spec.split(":")
+                port = None
+
+            if security == "ldaps":
+                use_ssl = True
+            elif security == "ldap":
+                use_ssl = False
+            else:
+                raise ValueError
+
+            return use_ssl, url, port, ldap_user, ldap_pass, dn_subtree
+        except ValueError:
+            raise exceptions.OptionsError(f"Invalid LDAP specification: {spec}")
+
     def __call__(self, username: str, password: str) -> bool:
         if not username or not password:
             return False

test/mitmproxy/addons/test_proxyauth.py

@@ -147,13 +147,19 @@ class TestProxyAuth:
             )
             assert isinstance(pa.validator, proxyauth.Ldap)
 
-            with pytest.raises(exceptions.OptionsError, match="Invalid ldap specification"):
+            ctx.configure(
+                pa,
+                proxyauth="ldap:localhost:1234:cn=default,dc=cdhdt,dc=com:password:ou=application,dc=cdhdt,dc=com"
+            )
+            assert isinstance(pa.validator, proxyauth.Ldap)
+
+            with pytest.raises(exceptions.OptionsError, match="Invalid LDAP specification"):
                 ctx.configure(pa, proxyauth="ldap:test:test:test")
 
-            with pytest.raises(exceptions.OptionsError, match="Invalid ldap specification"):
+            with pytest.raises(exceptions.OptionsError, match="Invalid LDAP specification"):
                 ctx.configure(pa, proxyauth="ldap:fake_serveruid=?dc=example,dc=com:person")
 
-            with pytest.raises(exceptions.OptionsError, match="Invalid ldap specification"):
+            with pytest.raises(exceptions.OptionsError, match="Invalid LDAP specification"):
                 ctx.configure(pa, proxyauth="ldapssssssss:fake_server:dn:password:tree")
 
             with pytest.raises(exceptions.OptionsError, match="Could not open htpasswd file"):
@@ -200,11 +206,15 @@ class TestProxyAuth:
             assert f2.metadata["proxyauth"] == ('test', 'test')
 
 
-def test_ldap(monkeypatch):
+@pytest.mark.parametrize("spec", [
+    "ldaps:localhost:cn=default,dc=cdhdt,dc=com:password:ou=application,dc=cdhdt,dc=com",
+    "ldap:localhost:1234:cn=default,dc=cdhdt,dc=com:password:ou=application,dc=cdhdt,dc=com"
+])
+def test_ldap(monkeypatch, spec):
     monkeypatch.setattr(ldap3, "Server", mock.MagicMock())
     monkeypatch.setattr(ldap3, "Connection", mock.MagicMock())
 
-    validator = proxyauth.Ldap("ldaps:localhost:cn=default,dc=cdhdt,dc=com:password:ou=application,dc=cdhdt,dc=com")
+    validator = proxyauth.Ldap(spec)
     assert not validator("", "")
     assert validator("foo", "bar")
     validator.conn.response = False