(self, service_name, path, callback, errback, timeout = 30)
| 743 | sendCreate(self.connected_trees[service_name]) |
| 744 | |
| 745 | def _getAttributes_SMB2(self, service_name, path, callback, errback, timeout = 30): |
| 746 | if not self.has_authenticated: |
| 747 | raise NotReadyError('SMB connection not authenticated') |
| 748 | |
| 749 | expiry_time = time.time() + timeout |
| 750 | path = path.replace('/', '\\') |
| 751 | if path.startswith('\\'): |
| 752 | path = path[1:] |
| 753 | if path.endswith('\\'): |
| 754 | path = path[:-1] |
| 755 | messages_history = [ ] |
| 756 | |
| 757 | def sendCreate(tid): |
| 758 | create_context_data = binascii.unhexlify(""" |
| 759 | 28 00 00 00 10 00 04 00 00 00 18 00 10 00 00 00 |
| 760 | 44 48 6e 51 00 00 00 00 00 00 00 00 00 00 00 00 |
| 761 | 00 00 00 00 00 00 00 00 18 00 00 00 10 00 04 00 |
| 762 | 00 00 18 00 00 00 00 00 4d 78 41 63 00 00 00 00 |
| 763 | 00 00 00 00 10 00 04 00 00 00 18 00 00 00 00 00 |
| 764 | 51 46 69 64 00 00 00 00 |
| 765 | """.replace(' ', '').replace('\n', '')) |
| 766 | m = SMB2Message(SMB2CreateRequest(path, |
| 767 | file_attributes = 0, |
| 768 | access_mask = FILE_READ_DATA | FILE_READ_EA | FILE_READ_ATTRIBUTES | SYNCHRONIZE, |
| 769 | share_access = FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE, |
| 770 | oplock = SMB2_OPLOCK_LEVEL_NONE, |
| 771 | impersonation = SEC_IMPERSONATE, |
| 772 | create_options = 0, |
| 773 | create_disp = FILE_OPEN, |
| 774 | create_context_data = create_context_data)) |
| 775 | m.tid = tid |
| 776 | self._sendSMBMessage(m) |
| 777 | self.pending_requests[m.mid] = _PendingRequest(m.mid, expiry_time, createCB, errback, tid = tid) |
| 778 | self._pushToArray(messages_history, m) |
| 779 | |
| 780 | def createCB(create_message, **kwargs): |
| 781 | self._pushToArray(messages_history, create_message) |
| 782 | if create_message.status == 0: |
| 783 | p = create_message.payload |
| 784 | filename = self._extractLastPathComponent(path) |
| 785 | info = SharedFile(p.create_time, p.lastaccess_time, p.lastwrite_time, p.change_time, |
| 786 | p.file_size, p.allocation_size, p.file_attributes, |
| 787 | filename, filename) |
| 788 | closeFid(kwargs['tid'], p.fid, info = info) |
| 789 | else: |
| 790 | errback(OperationFailure('Failed to get attributes for %s on %s: Unable to open remote file object' % ( path, service_name ), messages_history)) |
| 791 | |
| 792 | def closeFid(tid, fid, info = None, error = None): |
| 793 | m = SMB2Message(SMB2CloseRequest(fid)) |
| 794 | m.tid = tid |
| 795 | self._sendSMBMessage(m) |
| 796 | self.pending_requests[m.mid] = _PendingRequest(m.mid, expiry_time, closeCB, errback, info = info, error = error) |
| 797 | self._pushToArray(messages_history, m) |
| 798 | |
| 799 | def closeCB(close_message, **kwargs): |
| 800 | if kwargs['info'] is not None: |
| 801 | callback(kwargs['info']) |
| 802 | elif kwargs['error'] is not None: |
nothing calls this directly
no test coverage detected