(data_bytes)
| 671 | closeFid(kwargs['tid'], kwargs['fid'], error = query_message.status) |
| 672 | |
| 673 | def decodeQueryStruct(data_bytes): |
| 674 | # FileIdBothDirectoryInformation structure. See [MS-SMB]: 2.2.8.1.3 and [MS-FSCC]: 2.4.17 |
| 675 | info_format = '<IIQQQQQQIIIBB24sHQ' |
| 676 | info_size = struct.calcsize(info_format) |
| 677 | |
| 678 | data_length = len(data_bytes) |
| 679 | offset = 0 |
| 680 | while offset < data_length: |
| 681 | if offset + info_size > data_length: |
| 682 | return data_bytes[offset:] |
| 683 | |
| 684 | next_offset, _, \ |
| 685 | create_time, last_access_time, last_write_time, last_attr_change_time, \ |
| 686 | file_size, alloc_size, file_attributes, filename_length, ea_size, \ |
| 687 | short_name_length, _, short_name, _, file_id = struct.unpack(info_format, data_bytes[offset:offset+info_size]) |
| 688 | |
| 689 | offset2 = offset + info_size |
| 690 | if offset2 + filename_length > data_length: |
| 691 | return data_bytes[offset:] |
| 692 | |
| 693 | filename = DataFaultToleranceStrategy.data_bytes_decode(data_bytes[offset2:offset2+filename_length]) |
| 694 | short_name = DataFaultToleranceStrategy.data_bytes_decode(short_name[:short_name_length]) |
| 695 | |
| 696 | accept_result = False |
| 697 | if (file_attributes & 0xff) in ( 0x00, ATTR_NORMAL ): # Only the first 8-bits are compared. We ignore other bits like temp, compressed, encryption, sparse, indexed, etc |
| 698 | accept_result = (search == SMB_FILE_ATTRIBUTE_NORMAL) or (search & SMB_FILE_ATTRIBUTE_INCL_NORMAL) |
| 699 | else: |
| 700 | accept_result = (file_attributes & search) > 0 |
| 701 | if accept_result: |
| 702 | results.append(SharedFile(convertFILETIMEtoEpoch(create_time), convertFILETIMEtoEpoch(last_access_time), |
| 703 | convertFILETIMEtoEpoch(last_write_time), convertFILETIMEtoEpoch(last_attr_change_time), |
| 704 | file_size, alloc_size, file_attributes, short_name, filename, file_id)) |
| 705 | |
| 706 | if next_offset: |
| 707 | offset += next_offset |
| 708 | else: |
| 709 | break |
| 710 | return b'' |
| 711 | |
| 712 | def closeFid(tid, fid, results = None, error = None): |
| 713 | m = SMB2Message(SMB2CloseRequest(fid)) |
nothing calls this directly
no test coverage detected