Source code for ldap3.utils.ntlm

"""
"""

# Created on 2015.04.02
#
# Author: Giovanni Cannata
#
# Copyright 2015 - 2018 Giovanni Cannata
#
# This file is part of ldap3.
#
# ldap3 is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# ldap3 is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with ldap3 in the COPYING and COPYING.LESSER files.
# If not, see <http://www.gnu.org/licenses/>.

# NTLMv2 authentication as per [MS-NLMP] (https://msdn.microsoft.com/en-us/library/cc236621.aspx)

from struct import pack, unpack
from platform import system, version
from socket import gethostname
from time import time
import hmac
import hashlib
import binascii
from os import urandom

try:
    from locale import getpreferredencoding
    oem_encoding = getpreferredencoding()
except Exception:
    oem_encoding = 'utf-8'

from ..protocol.formatters.formatters import format_ad_timestamp

NTLM_SIGNATURE = b'NTLMSSP\x00'
NTLM_MESSAGE_TYPE_NTLM_NEGOTIATE = 1
NTLM_MESSAGE_TYPE_NTLM_CHALLENGE = 2
NTLM_MESSAGE_TYPE_NTLM_AUTHENTICATE = 3

FLAG_NEGOTIATE_56 = 31  # W
FLAG_NEGOTIATE_KEY_EXCH = 30  # V
FLAG_NEGOTIATE_128 = 29  # U
FLAG_NEGOTIATE_VERSION = 25  # T
FLAG_NEGOTIATE_TARGET_INFO = 23  # S
FLAG_REQUEST_NOT_NT_SESSION_KEY = 22  # R
FLAG_NEGOTIATE_IDENTIFY = 20  # Q
FLAG_NEGOTIATE_EXTENDED_SESSIONSECURITY = 19  # P
FLAG_TARGET_TYPE_SERVER = 17  # O
FLAG_TARGET_TYPE_DOMAIN = 16  # N
FLAG_NEGOTIATE_ALWAYS_SIGN = 15  # M
FLAG_NEGOTIATE_OEM_WORKSTATION_SUPPLIED = 13  # L
FLAG_NEGOTIATE_OEM_DOMAIN_SUPPLIED = 12  # K
FLAG_NEGOTIATE_ANONYMOUS = 11  # J
FLAG_NEGOTIATE_NTLM = 9  # H
FLAG_NEGOTIATE_LM_KEY = 7  # G
FLAG_NEGOTIATE_DATAGRAM = 6  # F
FLAG_NEGOTIATE_SEAL = 5  # E
FLAG_NEGOTIATE_SIGN = 4  # D
FLAG_REQUEST_TARGET = 2  # C
FLAG_NEGOTIATE_OEM = 1  # B
FLAG_NEGOTIATE_UNICODE = 0  # A

FLAG_TYPES = [FLAG_NEGOTIATE_56,
              FLAG_NEGOTIATE_KEY_EXCH,
              FLAG_NEGOTIATE_128,
              FLAG_NEGOTIATE_VERSION,
              FLAG_NEGOTIATE_TARGET_INFO,
              FLAG_REQUEST_NOT_NT_SESSION_KEY,
              FLAG_NEGOTIATE_IDENTIFY,
              FLAG_NEGOTIATE_EXTENDED_SESSIONSECURITY,
              FLAG_TARGET_TYPE_SERVER,
              FLAG_TARGET_TYPE_DOMAIN,
              FLAG_NEGOTIATE_ALWAYS_SIGN,
              FLAG_NEGOTIATE_OEM_WORKSTATION_SUPPLIED,
              FLAG_NEGOTIATE_OEM_DOMAIN_SUPPLIED,
              FLAG_NEGOTIATE_ANONYMOUS,
              FLAG_NEGOTIATE_NTLM,
              FLAG_NEGOTIATE_LM_KEY,
              FLAG_NEGOTIATE_DATAGRAM,
              FLAG_NEGOTIATE_SEAL,
              FLAG_NEGOTIATE_SIGN,
              FLAG_REQUEST_TARGET,
              FLAG_NEGOTIATE_OEM,
              FLAG_NEGOTIATE_UNICODE]

AV_END_OF_LIST = 0
AV_NETBIOS_COMPUTER_NAME = 1
AV_NETBIOS_DOMAIN_NAME = 2
AV_DNS_COMPUTER_NAME = 3
AV_DNS_DOMAIN_NAME = 4
AV_DNS_TREE_NAME = 5
AV_FLAGS = 6
AV_TIMESTAMP = 7
AV_SINGLE_HOST_DATA = 8
AV_TARGET_NAME = 9
AV_CHANNEL_BINDINGS = 10

AV_TYPES = [AV_END_OF_LIST,
            AV_NETBIOS_COMPUTER_NAME,
            AV_NETBIOS_DOMAIN_NAME,
            AV_DNS_COMPUTER_NAME,
            AV_DNS_DOMAIN_NAME,
            AV_DNS_TREE_NAME,
            AV_FLAGS,
            AV_TIMESTAMP,
            AV_SINGLE_HOST_DATA,
            AV_TARGET_NAME,
            AV_CHANNEL_BINDINGS]

AV_FLAG_CONSTRAINED = 0
AV_FLAG_INTEGRITY = 1
AV_FLAG_TARGET_SPN_UNTRUSTED = 2

AV_FLAG_TYPES = [AV_FLAG_CONSTRAINED,
                 AV_FLAG_INTEGRITY,
                 AV_FLAG_TARGET_SPN_UNTRUSTED]


[docs]def pack_windows_version(debug=False): if debug: if system().lower() == 'windows': try: major_release, minor_release, build = version().split('.') major_release = int(major_release) minor_release = int(minor_release) build = int(build) except Exception: major_release = 5 minor_release = 1 build = 2600 else: major_release = 5 minor_release = 1 build = 2600 else: major_release = 0 minor_release = 0 build = 0 return pack('<B', major_release) + \ pack('<B', minor_release) + \ pack('<H', build) + \ pack('<B', 0) + \ pack('<B', 0) + \ pack('<B', 0) + \ pack('<B', 15)
[docs]def unpack_windows_version(version_message): if len(version_message) != 8: raise ValueError('version field must be 8 bytes long') if str is bytes: # Python 2 return (unpack('<B', version_message[0])[0], unpack('<B', version_message[1])[0], unpack('<H', version_message[2:4])[0], unpack('<B', version_message[7])[0]) else: # Python 3 return (int(version_message[0]), int(version_message[1]), int(unpack('<H', version_message[2:4])[0]), int(version_message[7]))
[docs]class NtlmClient(object): def __init__(self, domain, user_name, password): self.client_config_flags = 0 self.exported_session_key = None self.negotiated_flags = None self.user_name = user_name self.user_domain = domain self.no_lm_response_ntlm_v1 = None self.client_blocked = False self.client_block_exceptions = [] self.client_require_128_bit_encryption = None self.max_life_time = None self.client_signing_key = None self.client_sealing_key = None self.sequence_number = None self.server_sealing_key = None self.server_signing_key = None self.integrity = False self.replay_detect = False self.sequence_detect = False self.confidentiality = False self.datagram = False self.identity = False self.client_supplied_target_name = None self.client_channel_binding_unhashed = None self.unverified_target_name = None self._password = password self.server_challenge = None self.server_target_name = None self.server_target_info = None self.server_version = None self.server_av_netbios_computer_name = None self.server_av_netbios_domain_name = None self.server_av_dns_computer_name = None self.server_av_dns_domain_name = None self.server_av_dns_forest_name = None self.server_av_target_name = None self.server_av_flags = None self.server_av_timestamp = None self.server_av_single_host_data = None self.server_av_channel_bindings = None self.server_av_flag_constrained = None self.server_av_flag_integrity = None self.server_av_flag_target_spn_untrusted = None self.current_encoding = None self.client_challenge = None self.server_target_info_raw = None
[docs] def get_client_flag(self, flag): if not self.client_config_flags: return False if flag in FLAG_TYPES: return True if self.client_config_flags & (1 << flag) else False raise ValueError('invalid flag')
[docs] def get_negotiated_flag(self, flag): if not self.negotiated_flags: return False if flag not in FLAG_TYPES: raise ValueError('invalid flag') return True if self.negotiated_flags & (1 << flag) else False
[docs] def get_server_av_flag(self, flag): if not self.server_av_flags: return False if flag not in AV_FLAG_TYPES: raise ValueError('invalid AV flag') return True if self.server_av_flags & (1 << flag) else False
[docs] def set_client_flag(self, flags): if type(flags) == int: flags = [flags] for flag in flags: if flag in FLAG_TYPES: self.client_config_flags |= (1 << flag) else: raise ValueError('invalid flag')
[docs] def reset_client_flags(self): self.client_config_flags = 0
[docs] def unset_client_flag(self, flags): if type(flags) == int: flags = [flags] for flag in flags: if flag in FLAG_TYPES: self.client_config_flags &= ~(1 << flag) else: raise ValueError('invalid flag')
[docs] def create_negotiate_message(self): """ Microsoft MS-NLMP 2.2.1.1 """ self.reset_client_flags() self.set_client_flag([FLAG_REQUEST_TARGET, FLAG_NEGOTIATE_56, FLAG_NEGOTIATE_128, FLAG_NEGOTIATE_NTLM, FLAG_NEGOTIATE_ALWAYS_SIGN, FLAG_NEGOTIATE_OEM, FLAG_NEGOTIATE_UNICODE, FLAG_NEGOTIATE_EXTENDED_SESSIONSECURITY]) message = NTLM_SIGNATURE # 8 bytes message += pack('<I', NTLM_MESSAGE_TYPE_NTLM_NEGOTIATE) # 4 bytes message += pack('<I', self.client_config_flags) # 4 bytes message += self.pack_field('', 40) # domain name field # 8 bytes if self.get_client_flag(FLAG_NEGOTIATE_VERSION): # version 8 bytes - used for debug in ntlm message += pack_windows_version(True) else: message += pack_windows_version(False) return message
[docs] def parse_challenge_message(self, message): """ Microsoft MS-NLMP 2.2.1.2 """ if len(message) < 56: # minimum size of challenge message return False if message[0:8] != NTLM_SIGNATURE: # NTLM signature - 8 bytes return False if int(unpack('<I', message[8:12])[0]) != NTLM_MESSAGE_TYPE_NTLM_CHALLENGE: # type of message - 4 bytes return False target_name_len, _, target_name_offset = self.unpack_field(message[12:20]) # targetNameFields - 8 bytes self.negotiated_flags = unpack('<I', message[20:24])[0] # negotiated flags - 4 bytes self.current_encoding = 'utf-16-le' if self.get_negotiated_flag( FLAG_NEGOTIATE_UNICODE) else oem_encoding # set encoding self.server_challenge = message[24:32] # server challenge - 8 bytes target_info_len, _, target_info_offset = self.unpack_field(message[40:48]) # targetInfoFields - 8 bytes self.server_version = unpack_windows_version(message[48:56]) if self.get_negotiated_flag(FLAG_REQUEST_TARGET) and target_name_len: self.server_target_name = message[target_name_offset: target_name_offset + target_name_len].decode( self.current_encoding) if self.get_negotiated_flag(FLAG_NEGOTIATE_TARGET_INFO) and target_info_len: self.server_target_info_raw = message[target_info_offset: target_info_offset + target_info_len] self.server_target_info = self.unpack_av_info(self.server_target_info_raw) for attribute, value in self.server_target_info: if attribute == AV_NETBIOS_COMPUTER_NAME: self.server_av_netbios_computer_name = value.decode('utf-16-le') # always unicode elif attribute == AV_NETBIOS_DOMAIN_NAME: self.server_av_netbios_domain_name = value.decode('utf-16-le') # always unicode elif attribute == AV_DNS_COMPUTER_NAME: self.server_av_dns_computer_name = value.decode('utf-16-le') # always unicode elif attribute == AV_DNS_DOMAIN_NAME: self.server_av_dns_domain_name = value.decode('utf-16-le') # always unicode elif attribute == AV_DNS_TREE_NAME: self.server_av_dns_forest_name = value.decode('utf-16-le') # always unicode elif attribute == AV_FLAGS: if self.get_server_av_flag(AV_FLAG_CONSTRAINED): self.server_av_flag_constrained = True if self.get_server_av_flag(AV_FLAG_INTEGRITY): self.server_av_flag_integrity = True if self.get_server_av_flag(AV_FLAG_TARGET_SPN_UNTRUSTED): self.server_av_flag_target_spn_untrusted = True elif attribute == AV_TIMESTAMP: self.server_av_timestamp = format_ad_timestamp(unpack('<Q', value)[0]) elif attribute == AV_SINGLE_HOST_DATA: self.server_av_single_host_data = value elif attribute == AV_TARGET_NAME: self.server_av_target_name = value.decode('utf-16-le') # always unicode elif attribute == AV_CHANNEL_BINDINGS: self.server_av_channel_bindings = value else: raise ValueError('unknown AV type')
[docs] def create_authenticate_message(self): """ Microsoft MS-NLMP 2.2.1.3 """ # 3.1.5.2 if not self.client_config_flags and not self.negotiated_flags: return False # 3.1.5.2 if self.get_client_flag(FLAG_NEGOTIATE_128) and not self.get_negotiated_flag(FLAG_NEGOTIATE_128): return False # 3.1.5.2 if (not self.server_av_netbios_computer_name or not self.server_av_netbios_domain_name) and self.server_av_flag_integrity: return False message = NTLM_SIGNATURE # 8 bytes message += pack('<I', NTLM_MESSAGE_TYPE_NTLM_AUTHENTICATE) # 4 bytes pos = 88 # payload starts at 88 # 3.1.5.2 if self.server_target_info: lm_challenge_response = b'' else: # computed LmChallengeResponse - todo lm_challenge_response = b'' message += self.pack_field(lm_challenge_response, pos) # LmChallengeResponseField field # 8 bytes pos += len(lm_challenge_response) nt_challenge_response = self.compute_nt_response() message += self.pack_field(nt_challenge_response, pos) # NtChallengeResponseField field # 8 bytes pos += len(nt_challenge_response) domain_name = self.user_domain.encode(self.current_encoding) message += self.pack_field(domain_name, pos) # DomainNameField field # 8 bytes pos += len(domain_name) user_name = self.user_name.encode(self.current_encoding) message += self.pack_field(user_name, pos) # UserNameField field # 8 bytes pos += len(user_name) if self.get_negotiated_flag(FLAG_NEGOTIATE_OEM_WORKSTATION_SUPPLIED) or self.get_negotiated_flag( FLAG_NEGOTIATE_VERSION): workstation = gethostname().encode(self.current_encoding) else: workstation = b'' message += self.pack_field(workstation, pos) # empty WorkstationField field # 8 bytes pos += len(workstation) encrypted_random_session_key = b'' message += self.pack_field(encrypted_random_session_key, pos) # EncryptedRandomSessionKeyField field # 8 bytes pos += len(encrypted_random_session_key) message += pack('<I', self.negotiated_flags) # negotiated flags - 4 bytes if self.get_negotiated_flag(FLAG_NEGOTIATE_VERSION): message += pack_windows_version(True) # windows version - 8 bytes else: message += pack_windows_version() # empty windows version - 8 bytes message += pack('<Q', 0) # mic message += pack('<Q', 0) # mic - total of 16 bytes # payload starts at 88 message += lm_challenge_response message += nt_challenge_response message += domain_name message += user_name message += workstation message += encrypted_random_session_key return message
[docs] @staticmethod def pack_field(value, offset): return pack('<HHI', len(value), len(value), offset)
[docs] @staticmethod def unpack_field(field_message): if len(field_message) != 8: raise ValueError('ntlm field must be 8 bytes long') return unpack('<H', field_message[0:2])[0], \ unpack('<H', field_message[2:4])[0], \ unpack('<I', field_message[4:8])[0]
[docs] @staticmethod def unpack_av_info(info): if info: avs = list() done = False pos = 0 while not done: av_type = unpack('<H', info[pos: pos + 2])[0] if av_type not in AV_TYPES: raise ValueError('unknown AV type') av_len = unpack('<H', info[pos + 2: pos + 4])[0] av_value = info[pos + 4: pos + 4 + av_len] pos += av_len + 4 if av_type == AV_END_OF_LIST: done = True else: avs.append((av_type, av_value)) else: return list() return avs
[docs] @staticmethod 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: continue info += pack('<H', av_type) info += pack('<H', len(av_value)) info += av_value # add AV_END_OF_LIST info += pack('<H', AV_END_OF_LIST) info += pack('<H', 0) return info
[docs] @staticmethod def pack_windows_timestamp(): return pack('<Q', (int(time()) + 11644473600) * 10000000)
[docs] def compute_nt_response(self): if not self.user_name and not self._password: # anonymous authentication return b'' self.client_challenge = urandom(8) temp = b'' temp += pack('<B', 1) # ResponseVersion - 1 byte temp += pack('<B', 1) # HiResponseVersion - 1 byte temp += pack('<H', 0) # Z(2) temp += pack('<I', 0) # Z(4) - total Z(6) temp += self.pack_windows_timestamp() # time - 8 bytes temp += self.client_challenge # random client challenge - 8 bytes temp += pack('<I', 0) # Z(4) temp += self.server_target_info_raw temp += pack('<I', 0) # Z(4) response_key_nt = self.ntowf_v2() nt_proof_str = hmac.new(response_key_nt, self.server_challenge + temp).digest() nt_challenge_response = nt_proof_str + temp return nt_challenge_response
[docs] def ntowf_v2(self): passparts = self._password.split(':') if len(passparts) == 2 and len(passparts[0]) == 32 and len(passparts[1]) == 32: # The specified password is an LM:NTLM hash password_digest = binascii.unhexlify(passparts[1]) else: password_digest = hashlib.new('MD4', self._password.encode('utf-16-le')).digest() return hmac.new(password_digest, (self.user_name.upper() + self.user_domain).encode('utf-16-le')).digest()