(self, service_name, path_file_pattern, callback, errback, file_attributes = ATTR_NORMAL, timeout = 30)
| 1326 | sendCreate(tid) |
| 1327 | |
| 1328 | def _resetFileAttributes_SMB2(self, service_name, path_file_pattern, callback, errback, file_attributes = ATTR_NORMAL, timeout = 30): |
| 1329 | if not self.has_authenticated: |
| 1330 | raise NotReadyError('SMB connection not authenticated') |
| 1331 | |
| 1332 | expiry_time = time.time() + timeout |
| 1333 | path = path_file_pattern.replace('/', '\\') |
| 1334 | if path.startswith('\\'): |
| 1335 | path = path[1:] |
| 1336 | if path.endswith('\\'): |
| 1337 | path = path[:-1] |
| 1338 | messages_history = [ ] |
| 1339 | |
| 1340 | def sendCreate(tid): |
| 1341 | create_context_data = binascii.unhexlify(b""" |
| 1342 | 28 00 00 00 10 00 04 00 00 00 18 00 10 00 00 00 |
| 1343 | 44 48 6e 51 00 00 00 00 00 00 00 00 00 00 00 00 |
| 1344 | 00 00 00 00 00 00 00 00 18 00 00 00 10 00 04 00 |
| 1345 | 00 00 18 00 00 00 00 00 4d 78 41 63 00 00 00 00 |
| 1346 | 00 00 00 00 10 00 04 00 00 00 18 00 00 00 00 00 |
| 1347 | 51 46 69 64 00 00 00 00 |
| 1348 | """.replace(b' ', b'').replace(b'\n', b'')) |
| 1349 | |
| 1350 | m = SMB2Message(SMB2CreateRequest(path, |
| 1351 | file_attributes = 0, |
| 1352 | access_mask = FILE_WRITE_ATTRIBUTES, |
| 1353 | share_access = FILE_SHARE_READ | FILE_SHARE_WRITE, |
| 1354 | oplock = SMB2_OPLOCK_LEVEL_NONE, |
| 1355 | impersonation = SEC_IMPERSONATE, |
| 1356 | create_options = 0, |
| 1357 | create_disp = FILE_OPEN, |
| 1358 | create_context_data = create_context_data)) |
| 1359 | m.tid = tid |
| 1360 | self._sendSMBMessage(m) |
| 1361 | self.pending_requests[m.mid] = _PendingRequest(m.mid, int(time.time()) + timeout, createCB, errback, tid = tid) |
| 1362 | self._pushToArray(messages_history, m) |
| 1363 | |
| 1364 | def createCB(open_message, **kwargs): |
| 1365 | self._pushToArray(messages_history, open_message) |
| 1366 | if open_message.status == 0: |
| 1367 | sendReset(kwargs['tid'], open_message.payload.fid) |
| 1368 | else: |
| 1369 | errback(OperationFailure('Failed to reset attributes of %s on %s: Unable to open file' % ( path, service_name ), messages_history)) |
| 1370 | |
| 1371 | def sendReset(tid, fid): |
| 1372 | m = SMB2Message(SMB2SetInfoRequest(fid, |
| 1373 | additional_info = 0, |
| 1374 | info_type = SMB2_INFO_FILE, |
| 1375 | file_info_class = 4, # FileBasicInformation |
| 1376 | data = struct.pack('qqqqii', 0, 0, 0, 0, file_attributes, 0))) |
| 1377 | # [MS-SMB2]: 2.2.39, [MS-FSCC]: 2.4, [MS-FSCC]: 2.4.7, [MS-FSCC]: 2.6 |
| 1378 | m.tid = tid |
| 1379 | self._sendSMBMessage(m) |
| 1380 | self.pending_requests[m.mid] = _PendingRequest(m.mid, int(time.time()) + timeout, resetCB, errback, tid = tid, fid = fid) |
| 1381 | self._pushToArray(messages_history, m) |
| 1382 | |
| 1383 | def resetCB(reset_message, **kwargs): |
| 1384 | self._pushToArray(messages_history, reset_message) |
| 1385 | if reset_message.status == 0: |
nothing calls this directly
no test coverage detected