Method decode

pager_lib/smb/smb_structs.py:339–392

(self, message)

    PAYLOAD_STRUCT_SIZE = struct.calcsize(PAYLOAD_STRUCT_FORMAT)

    def decode(self, message):
        assert message.command == SMB_COM_NEGOTIATE

        if not message.isReply:
            raise ProtocolError('Not a SMB_COM_NEGOTIATE reply', message.raw_data, message)

        self.security_mode, self.max_mpx_count, self.max_number_vcs, self.max_buffer_size, \
        self.max_raw_size, self.session_key, self.capabilities, self.system_time, self.server_time_zone, \
        self.challenge_length = ( 0, ) * 10

        data_len = len(message.parameters_data)
        if data_len < 2:
            raise ProtocolError('Not enough data to decode SMB_COM_NEGOTIATE dialect_index field', message.raw_data, message)

        self.dialect_index = struct.unpack('<H', message.parameters_data[:2])[0]
        if self.dialect_index == NT_LAN_MANAGER_DIALECT:
            if data_len != (0x11 * 2):
                raise ProtocolError('NT LAN Manager dialect selected in SMB_COM_NEGOTIATE but parameters bytes count (%d) does not meet specs' % data_len,
                                    message.raw_data, message)
            else:
                _, self.security_mode, self.max_mpx_count, self.max_number_vcs, self.max_buffer_size, \
                self.max_raw_size, self.session_key, self.capabilities, self.system_time, self.server_time_zone, \
                self.challenge_length = struct.unpack(self.PAYLOAD_STRUCT_FORMAT, message.parameters_data[:self.PAYLOAD_STRUCT_SIZE])
        elif self.dialect_index == 0xFFFF:
            raise ProtocolError('Server does not support any of the pysmb dialects. Please email pysmb to add in support for your OS',
                                message.raw_data, message)
        else:
            raise ProtocolError('Unknown dialect index (0x%04X)' % self.dialect_index, message.raw_data, message)

        data_len = len(message.data)
        if not message.hasExtendedSecurity:
            self.challenge, self.domain = '', ''
            if self.challenge_length > 0:
                if data_len >= self.challenge_length:
                    self.challenge = message.data[:self.challenge_length]

                    s = b''
                    offset = self.challenge_length
                    while offset < data_len:
                        _s = message.data[offset:offset+2]
                        if _s == b'\0\0':
                            self.domain = DataFaultToleranceStrategy.data_bytes_decode(s)#.decode('UTF-16LE')
                            break
                        else:
                            s += _s
                            offset += 2
                else:
                    raise ProtocolError('Not enough data to decode SMB_COM_NEGOTIATE (without security extensions) Challenge field', message.raw_data, message)
        else:
            if data_len < 16:
                raise ProtocolError('Not enough data to decode SMB_COM_NEGOTIATE (with security extensions) ServerGUID field', message.raw_data, message)

            self.server_guid = message.data[:16]
            self.security_blob = message.data[16:]

    @property
    def supportsExtendedSecurity(self):