(self, family, proto, src_ip, src_port, dst_ip, dst_port)
| 76 | _pf_context['started_by_sshuttle'] -= 1 |
| 77 | |
| 78 | def query_nat(self, family, proto, src_ip, src_port, dst_ip, dst_port): |
| 79 | [proto, family, src_port, dst_port] = [ |
| 80 | int(v) for v in [proto, family, src_port, dst_port]] |
| 81 | |
| 82 | packed_src_ip = socket.inet_pton(family, src_ip) |
| 83 | packed_dst_ip = socket.inet_pton(family, dst_ip) |
| 84 | |
| 85 | assert len(packed_src_ip) == len(packed_dst_ip) |
| 86 | length = len(packed_src_ip) |
| 87 | |
| 88 | pnl = self.pfioc_natlook() |
| 89 | pnl.proto = proto |
| 90 | pnl.direction = self.PF_OUT |
| 91 | pnl.af = family |
| 92 | memmove(addressof(pnl.saddr), packed_src_ip, length) |
| 93 | memmove(addressof(pnl.daddr), packed_dst_ip, length) |
| 94 | self._add_natlook_ports(pnl, src_port, dst_port) |
| 95 | |
| 96 | ioctl(pf_get_dev(), self.DIOCNATLOOK, |
| 97 | (c_char * sizeof(pnl)).from_address(addressof(pnl))) |
| 98 | |
| 99 | ip = socket.inet_ntop( |
| 100 | pnl.af, (c_char * length).from_address(addressof(pnl.rdaddr)).raw) |
| 101 | port = socket.ntohs(self._get_natlook_port(pnl.rdxport)) |
| 102 | return (ip, port) |
| 103 | |
| 104 | @staticmethod |
| 105 | def _add_natlook_ports(pnl, src_port, dst_port): |
no test coverage detected