Active Directory: fetch root domain of forest from LDAP (#105)

This change allows to use short domain name in login field.
As part of this PR ssl certificate validation in check_3pid_auth is fixed also.

Signed-off-by: Yuri Konotopov <[email protected]>

Author: nE0sIghT

README.rst

@@ -116,8 +116,8 @@ Let's say you have several domains in the ``example.com`` forest:
         bind_dn: "cn=hacker,ou=svcaccts,dc=example,dc=com"
         bind_password: "ch33kym0nk3y"
 
-With this configuration the user can log in with either ``main.example.com\someuser``,
-``someuser/main.example.com`` or ``someuser``.
+With this configuration the user can log in with either ``main\someuser``,
+``main.example.com\someuser``, ``someuser/main.example.com`` or ``someuser``.
 
 Users of other domains in the ``example.com`` forest can log in with ``domain\login``
 or ``login/domain``.

ldap_auth_provider.py

@@ -13,6 +13,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from typing import Optional
+
 from twisted.internet import threads
 
 
@@ -61,6 +63,7 @@ class LDAPMode(object):
 
 class LdapAuthProvider(object):
     __version__ = "0.1"
+    _ldap_tls = ldap3.Tls(validate=ssl.CERT_REQUIRED)
 
     def __init__(self, config, account_handler):
         self.account_handler = account_handler
@@ -84,6 +87,9 @@ def __init__(self, config, account_handler):
         self.ldap_active_directory = config.active_directory
         if self.ldap_active_directory:
             self.ldap_default_domain = config.default_domain
+            # Either: the Active Directory root domain (type str); empty string in case
+            # of error; or None if there was no attempt to fetch root domain yet
+            self.ldap_root_domain = None  # type: Optional[str]
 
     def get_supported_login_types(self):
         return {'m.login.password': ('password',)}
@@ -117,17 +123,14 @@ async def check_auth(self, username, login_type, login_dict):
 
         if self.ldap_active_directory:
             try:
-                (login, domain, localpart) = self._map_login_to_upn(username)
+                (login, domain, localpart) = await self._map_login_to_upn(username)
                 uid_value = login + "@" + domain
                 default_display_name = login
             except ActiveDirectoryUPNException:
                 return False
 
         try:
-            tls = ldap3.Tls(validate=ssl.CERT_REQUIRED)
-            server = ldap3.ServerPool(
-                [ldap3.Server(uri, get_info=None, tls=tls) for uri in self.ldap_uris],
-            )
+            server = self._get_server()
             logger.debug(
                 "Attempting LDAP connection with %s",
                 self.ldap_uris
@@ -254,9 +257,7 @@ async def check_3pid_auth(self, medium, address, password):
 
         # Talk to LDAP and check if this email/password combo is correct
         try:
-            server = ldap3.ServerPool(
-                [ldap3.Server(uri, get_info=None) for uri in self.ldap_uris],
-            )
+            server = self._get_server()
             logger.debug(
                 "Attempting LDAP connection with %s",
                 self.ldap_uris
@@ -404,6 +405,78 @@ class _LdapConfig(object):
 
         return ldap_config
 
+    def _get_server(self, get_info: Optional[str] = None) -> ldap3.ServerPool:
+        """Constructs ServerPool from configured LDAP URIs
+
+        Args:
+            get_info (str, optional): specifies if the server schema and server
+            specific info must be read. Defaults to None.
+
+        Returns:
+            Servers grouped in a ServerPool
+        """
+        return ldap3.ServerPool(
+            [
+                ldap3.Server(
+                    uri,
+                    get_info=get_info,
+                    tls=self._ldap_tls
+                )
+                for uri in self.ldap_uris
+            ],
+        )
+
+    async def _fetch_root_domain(self) -> str:
+        """Fetches root domain from LDAP and saves it to ``self.ldap_root_domain``
+
+        Returns:
+            The root domain of Active Directory forest
+        """
+        if self.ldap_root_domain is not None:
+            return self.ldap_root_domain
+
+        self.ldap_root_domain = ""
+
+        if self.ldap_mode != LDAPMode.SEARCH:
+            logger.info("Fetching root domain is supported in search mode only")
+            return self.ldap_root_domain
+
+        server = self._get_server(get_info=ldap3.DSA)
+        result, conn = await self._ldap_simple_bind(
+            server=server,
+            bind_dn=self.ldap_bind_dn,
+            password=self.ldap_bind_password,
+        )
+
+        if not result:
+            logger.warning("Unable to get root domain due to failed LDAP bind")
+            return self.ldap_root_domain
+
+        if (
+            conn.server.info.other
+            and conn.server.info.other.get("rootDomainNamingContext")
+        ):
+            # conn.server.info.other["rootDomainNamingContext"][0]
+            # is of the form DC=example,DC=org
+            self.ldap_root_domain = ".".join(
+                [
+                    dc.split("=")[1] for dc
+                    in conn.server.info.other["rootDomainNamingContext"][0].split(",")
+                    if "=" in dc
+                ]
+            )
+            logger.info('Obtained root domain "%s"', self.ldap_root_domain)
+
+        if not self.ldap_root_domain:
+            logger.warning(
+                "No valid `rootDomainNamingContext` attribute was found in the RootDSE. "
+                "Logging in using short domain name will be unavailable."
+            )
+
+        await threads.deferToThread(conn.unbind)
+
+        return self.ldap_root_domain
+
     async def _ldap_simple_bind(self, server, bind_dn, password):
         """ Attempt a simple bind with the credentials
             given by the user against the LDAP server.
@@ -556,7 +629,7 @@ async def _ldap_authenticated_search(self, server, password, filters):
             logger.warning("Error during LDAP authentication: %s", e)
             raise
 
-    def _map_login_to_upn(self, username):
+    async def _map_login_to_upn(self, username):
         """Maps user provided login to Active Directory UPN and
         local part of Matrix ID.
 
@@ -577,6 +650,9 @@ def _map_login_to_upn(self, username):
 
         if '\\' in username:
             (domain, login) = username.lower().rsplit('\\', 1)
+            ldap_root_domain = await self._fetch_root_domain()
+            if ldap_root_domain and not domain.endswith(ldap_root_domain):
+                domain += "." + ldap_root_domain
         elif "/" in username:
             (login, domain) = username.lower().rsplit("/", 1)
         else:

tests/__init__.py

@@ -1,5 +1,5 @@
 from asyncio.futures import Future
-from typing import Any, Awaitable
+from typing import Any, Awaitable, Type
 
 from twisted.internet.endpoints import serverFromString
 from twisted.internet.protocol import ServerFactory
@@ -117,10 +117,9 @@ async def _create_db():
 
 
 class _LDAPServerFactory(ServerFactory):
-    protocol = LDAPServer
-
-    def __init__(self, root):
+    def __init__(self, root, ldap_server_type: Type[LDAPServer] = LDAPServer):
         self.root = root
+        self.protocol = ldap_server_type
 
     def buildProtocol(self, addr):
         proto = self.protocol()
@@ -158,11 +157,11 @@ def close(self):
 )
 
 
-async def create_ldap_server():
+async def create_ldap_server(ldap_server_type: Type[LDAPServer] = LDAPServer):
     "Returns a context manager that represents the LDAP server."
 
     db = await _create_db()
-    factory = _LDAPServerFactory(db)
+    factory = _LDAPServerFactory(db, ldap_server_type)
     factory.debug = True
 
     # We just pick an arbitrary port to listen on.

tests/test_ad.py

@@ -12,6 +12,11 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+from ldaptor import interfaces
+from ldaptor.protocols import pureldap
+from ldaptor.protocols.ldap import ldaperrors
+from ldaptor.protocols.ldap.ldapserver import LDAPServer
+
 from twisted.internet.defer import ensureDeferred
 from twisted.trial import unittest
 from twisted.internet import defer
@@ -29,10 +34,30 @@
 logging.basicConfig()
 
 
+class _ActiveDirectoryLDAPServer(LDAPServer):
+    """Extends LDAPServer to return AD-specific attributes
+
+    Includes `rootDomainNamingContext` in bind responses.
+    """
+    def getRootDSE(self, request, reply):
+        root = interfaces.IConnectedLDAPEntry(self.factory)
+        reply(pureldap.LDAPSearchResultEntry(
+            objectName='',
+            attributes=[('supportedLDAPVersion', ['3']),
+                        ('namingContexts', [root.dn.getText()]),
+                        ('supportedExtension', [
+                            pureldap.LDAPPasswordModifyRequest.oid, ]),
+                        ('rootDomainNamingContext', ['DC=example,DC=org']), ], ))
+        return pureldap.LDAPSearchResultDone(
+            resultCode=ldaperrors.Success.resultCode)
+
+
 class AbstractLdapActiveDirectoryTestCase():
     @defer.inlineCallbacks
     def setUp(self):
-        self.ldap_server = yield ensureDeferred(create_ldap_server())
+        self.ldap_server = yield ensureDeferred(
+            create_ldap_server(_ActiveDirectoryLDAPServer)
+        )
         account_handler = Mock(spec_set=["check_user_exists", "get_qualified_user_id"])
         account_handler.check_user_exists.return_value = make_awaitable(True)
         account_handler.get_qualified_user_id = get_qualified_user_id
@@ -74,20 +99,41 @@ def test_correct_pwd(self):
         ))
         self.assertEqual(result, "@mainuser/main.example.org:test")
 
+        result = yield ensureDeferred(self.auth_provider.check_auth(
+            "main\\mainuser",
+            'm.login.password',
+            {"password": "abracadabra"}
+        ))
+        self.assertEqual(result, "@mainuser/main.example.org:test")
+
         result = yield ensureDeferred(self.auth_provider.check_auth(
             "subsidiary.example.org\\nonmainuser",
             'm.login.password',
             {"password": "simsalabim"}
         ))
         self.assertEqual(result, "@nonmainuser/subsidiary.example.org:test")
 
+        result = yield ensureDeferred(self.auth_provider.check_auth(
+            "subsidiary\\nonmainuser",
+            'm.login.password',
+            {"password": "simsalabim"}
+        ))
+        self.assertEqual(result, "@nonmainuser/subsidiary.example.org:test")
+
         result = yield ensureDeferred(self.auth_provider.check_auth(
             "subsidiary.example.org\\mainuser",
             'm.login.password',
             {"password": "changeit"}
         ))
         self.assertEqual(result, "@mainuser/subsidiary.example.org:test")
 
+        result = yield ensureDeferred(self.auth_provider.check_auth(
+            "subsidiary\\mainuser",
+            'm.login.password',
+            {"password": "changeit"}
+        ))
+        self.assertEqual(result, "@mainuser/subsidiary.example.org:test")
+
     @defer.inlineCallbacks
     def test_single_email(self):
         result = yield ensureDeferred(self.auth_provider.check_3pid_auth(