(self)
| 217 | self._refresh_ipv6() |
| 218 | |
| 219 | def _refresh_ipv4(self): |
| 220 | ret = ctypes.windll.iphlpapi.GetExtendedTcpTable( # type: ignore |
| 221 | ctypes.byref(self._tcp), |
| 222 | ctypes.byref(self._tcp_size), |
| 223 | False, |
| 224 | socket.AF_INET, |
| 225 | TCP_TABLE_OWNER_PID_CONNECTIONS, |
| 226 | 0, |
| 227 | ) |
| 228 | if ret == 0: |
| 229 | for row in self._tcp.table[: self._tcp.dwNumEntries]: |
| 230 | local_ip = socket.inet_ntop(socket.AF_INET, bytes(row.ucLocalAddr)) |
| 231 | local_port = socket.htons(row.dwLocalPort) |
| 232 | self._map[(local_ip, local_port)] = row.dwOwningPid |
| 233 | elif ret == ERROR_INSUFFICIENT_BUFFER: |
| 234 | self._tcp = MIB_TCPTABLE_OWNER_PID(self._tcp_size.value) |
| 235 | # no need to update size, that's already done. |
| 236 | self._refresh_ipv4() |
| 237 | else: |
| 238 | raise RuntimeError( |
| 239 | "[IPv4] Unknown GetExtendedTcpTable return code: %s" % ret |
| 240 | ) |
| 241 | |
| 242 | def _refresh_ipv6(self): |
| 243 | ret = ctypes.windll.iphlpapi.GetExtendedTcpTable( # type: ignore |
no test coverage detected