(self, service_name, old_path, new_path, callback, errback, timeout = 30)
| 1574 | sendCreate(self.connected_trees[service_name]) |
| 1575 | |
| 1576 | def _rename_SMB2(self, service_name, old_path, new_path, callback, errback, timeout = 30): |
| 1577 | if not self.has_authenticated: |
| 1578 | raise NotReadyError('SMB connection not authenticated') |
| 1579 | |
| 1580 | expiry_time = time.time() + timeout |
| 1581 | messages_history = [ ] |
| 1582 | |
| 1583 | new_path = new_path.replace('/', '\\') |
| 1584 | if new_path.startswith('\\'): |
| 1585 | new_path = new_path[1:] |
| 1586 | if new_path.endswith('\\'): |
| 1587 | new_path = new_path[:-1] |
| 1588 | |
| 1589 | old_path = old_path.replace('/', '\\') |
| 1590 | if old_path.startswith('\\'): |
| 1591 | old_path = old_path[1:] |
| 1592 | if old_path.endswith('\\'): |
| 1593 | old_path = old_path[:-1] |
| 1594 | |
| 1595 | def sendCreate(tid): |
| 1596 | create_context_data = binascii.unhexlify(b""" |
| 1597 | 28 00 00 00 10 00 04 00 00 00 18 00 10 00 00 00 |
| 1598 | 44 48 6e 51 00 00 00 00 00 00 00 00 00 00 00 00 |
| 1599 | 00 00 00 00 00 00 00 00 18 00 00 00 10 00 04 00 |
| 1600 | 00 00 18 00 00 00 00 00 4d 78 41 63 00 00 00 00 |
| 1601 | 00 00 00 00 10 00 04 00 00 00 18 00 00 00 00 00 |
| 1602 | 51 46 69 64 00 00 00 00 |
| 1603 | """.replace(b' ', b'').replace(b'\n', b'')) |
| 1604 | m = SMB2Message(SMB2CreateRequest(old_path, |
| 1605 | file_attributes = 0, |
| 1606 | access_mask = DELETE | FILE_READ_ATTRIBUTES | SYNCHRONIZE, |
| 1607 | share_access = FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE, |
| 1608 | oplock = SMB2_OPLOCK_LEVEL_NONE, |
| 1609 | impersonation = SEC_IMPERSONATE, |
| 1610 | create_options = FILE_SYNCHRONOUS_IO_NONALERT, |
| 1611 | create_disp = FILE_OPEN, |
| 1612 | create_context_data = create_context_data)) |
| 1613 | m.tid = tid |
| 1614 | self._sendSMBMessage(m) |
| 1615 | self.pending_requests[m.mid] = _PendingRequest(m.mid, int(time.time()) + timeout, createCB, errback, tid = tid) |
| 1616 | self._pushToArray(messages_history, m) |
| 1617 | |
| 1618 | def createCB(create_message, **kwargs): |
| 1619 | self._pushToArray(messages_history, create_message) |
| 1620 | if create_message.status == 0: |
| 1621 | sendRename(kwargs['tid'], create_message.payload.fid) |
| 1622 | else: |
| 1623 | errback(OperationFailure('Failed to rename %s on %s: Unable to open file/directory' % ( old_path, service_name ), messages_history)) |
| 1624 | |
| 1625 | def sendRename(tid, fid): |
| 1626 | data = b'\x00'*16 + struct.pack('<I', len(new_path)*2) + new_path.encode('UTF-16LE') |
| 1627 | m = SMB2Message(SMB2SetInfoRequest(fid, |
| 1628 | additional_info = 0, |
| 1629 | info_type = SMB2_INFO_FILE, |
| 1630 | file_info_class = 0x0a, # SMB2_FILE_RENAME_INFO |
| 1631 | data = data)) |
| 1632 | m.tid = tid |
| 1633 | self._sendSMBMessage(m) |
nothing calls this directly
no test coverage detected