Merge pull request cannatag#1087 from ThePirateWhoSmellsOfSunflowers/…

…add_ntlm_channel_binding

Add TLS channel binding during NTLM authentication

docs/manual/source/bind.rst

@@ -272,6 +272,11 @@ When binding via NTLM, it is also possible to authenticate with an LM:NTLM hash
 
     c = Connection(s, user="AUTHTEST\\Administrator", password="E52CAC67419A9A224A3B108F3FA6CB6D:8846F7EAEE8FB117AD06BDD830B7586C", authentication=NTLM)
 
+You can enable LDAP Channel Binding over TLS with the argument ``channel_binding=TLS_CHANNEL_BINDING``::
+    c = Connection(s, user="AUTHTEST\\Administrator", password="E52CAC67419A9A224A3B108F3FA6CB6D:8846F7EAEE8FB117AD06BDD830B7586C"
+                    channel_binding=TLS_CHANNEL_BINDING, authentication=NTLM)
+
+This option is only relevant for NTLM authentication done over TLS.
 It also supports confidentiality when performing LDAP Queries using the following:
 
     c = Connection(s, user="AUTHTEST\\Administrator", password="E52CAC67419A9A224A3B108F3FA6CB6D:8846F7EAEE8FB117AD06BDD830B7586C", authentication=NTLM, session_security=ENCRYPT)

docs/manual/source/connection.rst

@@ -104,6 +104,7 @@ Connection parameters are:
 
 * auto_encode: automatically tries to convert from local encoding to UTF8 for well known syntaxes and types, default to True
 
+* channel_binding: enable Channel Binding. Currently only TLS_CHANNEL_BINDING is implemented and this parameter is only available when using NTLM authentication done over TLS.
 
 .. note::
    The *auto_range* feature is very useful when searching Active Directory servers. When an Active Directory search returns more than 1000 entries this feature is automatically used by the server.

ldap3/__init__.py

@@ -37,6 +37,8 @@
 KERBEROS = GSSAPI = 'GSSAPI'
 PLAIN = 'PLAIN'
 
+# LDAPS CHANNEL BINDING
+TLS_CHANNEL_BINDING = 'TLS_CHANNEL_BINDING'
 # SESSION SECURITY
 SIGN = 'sign'
 ENCRYPT = 'ENCRYPT'

ldap3/core/connection.py

@@ -32,8 +32,7 @@
     SUBTREE, ASYNC, SYNC, NO_ATTRIBUTES, ALL_ATTRIBUTES, ALL_OPERATIONAL_ATTRIBUTES, MODIFY_INCREMENT, LDIF, ASYNC_STREAM, \
     RESTARTABLE, ROUND_ROBIN, REUSABLE, AUTO_BIND_DEFAULT, AUTO_BIND_NONE, AUTO_BIND_TLS_BEFORE_BIND, SAFE_SYNC, SAFE_RESTARTABLE, \
     AUTO_BIND_TLS_AFTER_BIND, AUTO_BIND_NO_TLS, STRING_TYPES, SEQUENCE_TYPES, MOCK_SYNC, MOCK_ASYNC, NTLM, EXTERNAL,\
-    DIGEST_MD5, GSSAPI, PLAIN, DSA, SCHEMA, ALL, ENCRYPT, SIGN
-
+    DIGEST_MD5, GSSAPI, PLAIN, DSA, SCHEMA, ALL, TLS_CHANNEL_BINDING
 from .results import RESULT_SUCCESS, RESULT_COMPARE_TRUE, RESULT_COMPARE_FALSE
 from ..extend import ExtendedOperationsRoot
 from .pooling import ServerPool
@@ -158,6 +157,8 @@ class Connection(object):
     :type check_names: bool
     :param collect_usage: collect usage metrics in the usage attribute
     :type collect_usage: bool
+    :param channel_binding: Enable Channel Binding
+    :type channel_binding: str
     :param read_only: disable operations that modify data in the LDAP server
     :type read_only: bool
     :param lazy: open and bind the connection only when an actual operation is performed
@@ -200,6 +201,7 @@ def __init__(self,
                  sasl_credentials=None,
                  check_names=True,
                  collect_usage=False,
+                 channel_binding=None,
                  read_only=False,
                  lazy=False,
                  raise_exceptions=False,
@@ -336,6 +338,13 @@ def __init__(self,
                 self.server_pool = None
                 self.server = server
 
+            if channel_binding == TLS_CHANNEL_BINDING and not (self.authentication == NTLM and self.server.ssl):
+                self.last_error = '"channel_binding" option only available for NTLM authentication over LDAPS'
+                if log_enabled(ERROR):
+                    log(ERROR, '%s for <%s>', self.last_error, self)
+                raise LDAPInvalidValueError(self.last_error)
+            self.channel_binding = channel_binding
+
             # if self.authentication == SIMPLE and self.user and self.check_names:
             #     self.user = safe_dn(self.user)
             #     if log_enabled(EXTENDED):
@@ -1390,6 +1399,33 @@ def do_ntlm_bind(self,
                     if self.session_security == ENCRYPT:
                         self.ntlm_client.confidentiality = True
 
+                    if self.channel_binding == TLS_CHANNEL_BINDING:
+                        # To perform channel binding during NTLM authentication, we need to add a new AV_PAIR (MS-NLMP 2.2.2.1)
+                        # within the AUTHENTICATE_MESSAGE (MS-NLMP 2.2.1.3). This new AV_PAIR has AvId 0x000A (MsvAvChannelBindings).
+                        # The Value field contains an MD5 hash of a gss_channel_bindings_struct.
+                        # The logic here is heavly inspired by "msldap", "minikerberos" and "asysocks" projects by @skelsec.
+                        from hashlib import sha256, md5
+                        ntlm_client.tls_channel_binding = True
+                        peer_certificate_sha256 = sha256(self.server.tls.peer_certificate).digest()
+
+                        # https://datatracker.ietf.org/doc/html/rfc2744#section-3.11
+                        channel_binding_struct = bytes()
+                        initiator_address = b'\x00'*8
+                        acceptor_address = b'\x00'*8
+
+                        # https://datatracker.ietf.org/doc/html/rfc5929#section-4
+                        application_data_raw = b'tls-server-end-point:' + peer_certificate_sha256
+                        len_application_data = len(application_data_raw).to_bytes(4, byteorder='little', signed = False)
+                        application_data = len_application_data
+                        application_data += application_data_raw
+                        channel_binding_struct += initiator_address
+                        channel_binding_struct += acceptor_address
+                        channel_binding_struct += application_data
+
+                        # https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-nlmp/83f5e789-660d-4781-8491-5f8c6641f75e
+                        # "The Value field contains an MD5 hash of a gss_channel_bindings_struct"
+                        ntlm_client.client_av_channel_bindings = md5(channel_binding_struct).digest()
+
                     # as per https://msdn.microsoft.com/en-us/library/cc223501.aspx
                     # send a sicilyPackageDiscovery request (in the bindRequest)
                     request = bind_operation(self.version, 'SICILY_PACKAGE_DISCOVERY', self.ntlm_client)

ldap3/core/tls.py

@@ -79,7 +79,8 @@ def __init__(self,
                  ca_certs_data=None,
                  local_private_key_password=None,
                  ciphers=None,
-                 sni=None):
+                 sni=None,
+                 peer_certificate=None):
         if ssl_options is None:
             ssl_options = []
         self.ssl_options = ssl_options
@@ -239,6 +240,7 @@ def wrap_socket(self, connection, do_handshake=False):
         if do_handshake and (self.validate == ssl.CERT_REQUIRED or self.validate == ssl.CERT_OPTIONAL):
             check_hostname(wrapped_socket, connection.server.host, self.valid_names)
 
+        self.peer_certificate = wrapped_socket.getpeercert(binary_form=True)
         connection.socket = wrapped_socket
         return
 

ldap3/utils/ntlm.py

@@ -222,6 +222,8 @@ def __init__(self, domain, user_name, password):
         self.current_encoding = None
         self.client_challenge = None
         self.server_target_info_raw = None
+        self.client_av_channel_bindings = None
+        self.tls_channel_binding = None
 
     def get_client_flag(self, flag):
         if not self.client_config_flags:
@@ -457,7 +459,7 @@ def pack_av_info(avs):
         # avs is a list of tuples, each tuple is made of av_type and av_value
         info = b''
         for av_type, av_value in avs:
-            if av_type(0) == AV_END_OF_LIST:
+            if av_type == AV_END_OF_LIST:
                 continue
             info += pack('<H', av_type)
             info += pack('<H', len(av_value))
@@ -486,6 +488,13 @@ def compute_nt_response(self):
         temp += self.pack_windows_timestamp()  # time - 8 bytes
         temp += self.client_challenge  # random client challenge - 8 bytes
         temp += pack('<I', 0)  # Z(4)
+
+        if self.tls_channel_binding:
+            server_av_pairs_unpack = self.unpack_av_info(self.server_target_info_raw)
+            server_av_pairs_unpack.append((AV_CHANNEL_BINDINGS,self.client_av_channel_bindings))
+            server_av_pairs_pack = self.pack_av_info(server_av_pairs_unpack)
+            self.server_target_info_raw = server_av_pairs_pack
+
         temp += self.server_target_info_raw
         temp += pack('<I', 0)  # Z(4)
         response_key_nt = self.ntowf_v2()