(self, pkt)
| 1526 | |
| 1527 | @ATMT.action(receive_write_request) |
| 1528 | def send_write_response(self, pkt): |
| 1529 | self.update_smbheader(pkt) |
| 1530 | resp = SMB2_Write_Response(Count=len(pkt.Data)) |
| 1531 | fid = self.get_file_id(pkt) |
| 1532 | if self.current_tree() == "IPC$": |
| 1533 | if fid in self.PIPES_TABLE.values(): |
| 1534 | # A pipe |
| 1535 | self.rpc_server.recv(pkt.Data) |
| 1536 | else: |
| 1537 | if self.readonly: |
| 1538 | # Read only ! |
| 1539 | resp = SMB2_Error_Response() |
| 1540 | resp.Command = "SMB2_WRITE" |
| 1541 | resp.Status = "ERROR_FILE_READ_ONLY" |
| 1542 | else: |
| 1543 | # Write file |
| 1544 | pth, _ = self.current_handles[fid] |
| 1545 | length = pkt[SMB2_Write_Request].DataLen |
| 1546 | off = pkt[SMB2_Write_Request].Offset |
| 1547 | self.vprint("Writing %s bytes at %s" % (length, off)) |
| 1548 | with open(pth, "r+b") as fd: |
| 1549 | fd.seek(off) |
| 1550 | resp.Count = fd.write(pkt[SMB2_Write_Request].Data) |
| 1551 | self.send(self.smb_header.copy() / resp) |
| 1552 | |
| 1553 | @ATMT.receive_condition(SERVING) |
| 1554 | def receive_read_request(self, pkt): |
nothing calls this directly
no test coverage detected