(self, pkt)
| 1557 | |
| 1558 | @ATMT.action(receive_read_request) |
| 1559 | def send_read_response(self, pkt): |
| 1560 | self.update_smbheader(pkt) |
| 1561 | resp = SMB2_Read_Response() |
| 1562 | fid = self.get_file_id(pkt) |
| 1563 | if self.current_tree() == "IPC$": |
| 1564 | # Read output from DCE/RPC server |
| 1565 | r = self.rpc_server.get_response() |
| 1566 | resp.Data = bytes(r) |
| 1567 | else: |
| 1568 | # Read file and send content |
| 1569 | pth, _ = self.current_handles[fid] |
| 1570 | length = pkt[SMB2_Read_Request].Length |
| 1571 | off = pkt[SMB2_Read_Request].Offset |
| 1572 | self.vprint("Reading %s bytes at %s" % (length, off)) |
| 1573 | with open(pth, "rb") as fd: |
| 1574 | fd.seek(off) |
| 1575 | resp.Data = fd.read(length) |
| 1576 | self.send(self.smb_header.copy() / resp) |
| 1577 | |
| 1578 | @ATMT.receive_condition(SERVING) |
| 1579 | def receive_close_request(self, pkt): |
nothing calls this directly
no test coverage detected