(self, service_name, path, callback, errback, timeout = 30)
| 1671 | sendCreate(self.connected_trees[service_name]) |
| 1672 | |
| 1673 | def _listSnapshots_SMB2(self, service_name, path, callback, errback, timeout = 30): |
| 1674 | if not self.has_authenticated: |
| 1675 | raise NotReadyError('SMB connection not authenticated') |
| 1676 | |
| 1677 | expiry_time = time.time() + timeout |
| 1678 | path = path.replace('/', '\\') |
| 1679 | if path.startswith('\\'): |
| 1680 | path = path[1:] |
| 1681 | if path.endswith('\\'): |
| 1682 | path = path[:-1] |
| 1683 | messages_history = [ ] |
| 1684 | |
| 1685 | def sendCreate(tid): |
| 1686 | create_context_data = binascii.unhexlify(b""" |
| 1687 | 28 00 00 00 10 00 04 00 00 00 18 00 10 00 00 00 |
| 1688 | 44 48 6e 51 00 00 00 00 00 00 00 00 00 00 00 00 |
| 1689 | 00 00 00 00 00 00 00 00 00 00 00 00 10 00 04 00 |
| 1690 | 00 00 18 00 00 00 00 00 4d 78 41 63 00 00 00 00 |
| 1691 | """.replace(b' ', b'').replace(b'\n', b'')) |
| 1692 | m = SMB2Message(SMB2CreateRequest(path, |
| 1693 | file_attributes = 0, |
| 1694 | access_mask = FILE_READ_DATA | FILE_READ_ATTRIBUTES | SYNCHRONIZE, |
| 1695 | share_access = FILE_SHARE_READ | FILE_SHARE_WRITE, |
| 1696 | oplock = SMB2_OPLOCK_LEVEL_NONE, |
| 1697 | impersonation = SEC_IMPERSONATE, |
| 1698 | create_options = FILE_SYNCHRONOUS_IO_NONALERT, |
| 1699 | create_disp = FILE_OPEN, |
| 1700 | create_context_data = create_context_data)) |
| 1701 | m.tid = tid |
| 1702 | self._sendSMBMessage(m) |
| 1703 | self.pending_requests[m.mid] = _PendingRequest(m.mid, int(time.time()) + timeout, createCB, errback, tid = tid) |
| 1704 | self._pushToArray(messages_history, m) |
| 1705 | |
| 1706 | def createCB(create_message, **kwargs): |
| 1707 | self._pushToArray(messages_history, create_message) |
| 1708 | if create_message.status == 0: |
| 1709 | sendEnumSnapshots(kwargs['tid'], create_message.payload.fid) |
| 1710 | else: |
| 1711 | errback(OperationFailure('Failed to list snapshots %s on %s: Unable to open file/directory' % ( path, service_name ), messages_history)) |
| 1712 | |
| 1713 | def sendEnumSnapshots(tid, fid): |
| 1714 | m = SMB2Message(SMB2IoctlRequest(fid, |
| 1715 | ctlcode = 0x00144064, # FSCTL_SRV_ENUMERATE_SNAPSHOTS |
| 1716 | flags = 0x0001, |
| 1717 | in_data = b'')) |
| 1718 | m.tid = tid |
| 1719 | self._sendSMBMessage(m) |
| 1720 | self.pending_requests[m.mid] = _PendingRequest(m.mid, int(time.time()) + timeout, enumSnapshotsCB, errback, tid = tid, fid = fid) |
| 1721 | self._pushToArray(messages_history, m) |
| 1722 | |
| 1723 | def enumSnapshotsCB(enum_message, **kwargs): |
| 1724 | self._pushToArray(messages_history, enum_message) |
| 1725 | if enum_message.status == 0: |
| 1726 | results = [ ] |
| 1727 | snapshots_count = struct.unpack('<I', enum_message.payload.out_data[4:8])[0] |
| 1728 | for i in range(0, snapshots_count): |
| 1729 | s = DataFaultToleranceStrategy.data_bytes_decode(enum_message.payload.out_data[12+i*50:12+48+i*50]) |
| 1730 | results.append(datetime(*list(map(int, ( s[5:9], s[10:12], s[13:15], s[16:18], s[19:21], s[22:24] ))))) |
nothing calls this directly
no test coverage detected