(
self, flow: dns.DNSFlow, msg: dns.DNSMessage
)
| 92 | yield commands.SendData(self.context.server, packed) |
| 93 | |
| 94 | def handle_response( |
| 95 | self, flow: dns.DNSFlow, msg: dns.DNSMessage |
| 96 | ) -> layer.CommandGenerator[None]: |
| 97 | flow.response = msg |
| 98 | yield DnsResponseHook(flow) |
| 99 | if flow.response: |
| 100 | packed = pack_message(flow.response, flow.client_conn.transport_protocol) |
| 101 | yield commands.SendData(self.context.client, packed) |
| 102 | |
| 103 | def handle_error(self, flow: dns.DNSFlow, err: str) -> layer.CommandGenerator[None]: |
| 104 | flow.error = mflow.Error(err) |
no test coverage detected