(self, packet: pydivert.Packet)
| 493 | self.api.shutdown() |
| 494 | |
| 495 | def redirect_request(self, packet: pydivert.Packet): |
| 496 | # print(" * Redirect client -> server to proxy") |
| 497 | # print(f"{packet.src_addr}:{packet.src_port} -> {packet.dst_addr}:{packet.dst_port}") |
| 498 | client = (packet.src_addr, packet.src_port) |
| 499 | |
| 500 | self.client_server_map[client] = (packet.dst_addr, packet.dst_port) |
| 501 | |
| 502 | # We do need to inject to an external IP here, 127.0.0.1 does not work. |
| 503 | if packet.address_family == socket.AF_INET: |
| 504 | assert self.ipv4_address |
| 505 | packet.dst_addr = self.ipv4_address |
| 506 | elif packet.address_family == socket.AF_INET6: |
| 507 | if not self.ipv6_address: |
| 508 | self.ipv6_address = get_local_ip6(packet.src_addr) |
| 509 | assert self.ipv6_address |
| 510 | packet.dst_addr = self.ipv6_address |
| 511 | else: |
| 512 | raise RuntimeError("Unknown address family") |
| 513 | packet.dst_port = self.proxy_port |
| 514 | packet.direction = pydivert.consts.Direction.INBOUND |
| 515 | |
| 516 | # We need a handle on the NETWORK layer. the local handle is not guaranteed to exist, |
| 517 | # so we use the response handle. |
| 518 | self.response.windivert.send(packet) |
| 519 | |
| 520 | def redirect_response(self, packet: pydivert.Packet): |
| 521 | """ |
no test coverage detected