Class TcpConnectionTable

mitmproxy/platform/windows.py:195–263

class TcpConnectionTable(collections.abc.Mapping):
    DEFAULT_TABLE_SIZE = 4096

    def __init__(self):
        self._tcp = MIB_TCPTABLE_OWNER_PID(self.DEFAULT_TABLE_SIZE)
        self._tcp_size = ctypes.wintypes.DWORD(self.DEFAULT_TABLE_SIZE)
        self._tcp6 = MIB_TCP6TABLE_OWNER_PID(self.DEFAULT_TABLE_SIZE)
        self._tcp6_size = ctypes.wintypes.DWORD(self.DEFAULT_TABLE_SIZE)
        self._map = {}

    def __getitem__(self, item):
        return self._map[item]

    def __iter__(self):
        return self._map.__iter__()

    def __len__(self):
        return self._map.__len__()

    def refresh(self):
        self._map = {}
        self._refresh_ipv4()
        self._refresh_ipv6()

    def _refresh_ipv4(self):
        ret = ctypes.windll.iphlpapi.GetExtendedTcpTable(  # type: ignore
            ctypes.byref(self._tcp),
            ctypes.byref(self._tcp_size),
            False,
            socket.AF_INET,
            TCP_TABLE_OWNER_PID_CONNECTIONS,
            0,
        )
        if ret == 0:
            for row in self._tcp.table[: self._tcp.dwNumEntries]:
                local_ip = socket.inet_ntop(socket.AF_INET, bytes(row.ucLocalAddr))
                local_port = socket.htons(row.dwLocalPort)
                self._map[(local_ip, local_port)] = row.dwOwningPid
        elif ret == ERROR_INSUFFICIENT_BUFFER:
            self._tcp = MIB_TCPTABLE_OWNER_PID(self._tcp_size.value)
            # no need to update size, that's already done.
            self._refresh_ipv4()
        else:
            raise RuntimeError(
                "[IPv4] Unknown GetExtendedTcpTable return code: %s" % ret
            )

    def _refresh_ipv6(self):
        ret = ctypes.windll.iphlpapi.GetExtendedTcpTable(  # type: ignore
            ctypes.byref(self._tcp6),
            ctypes.byref(self._tcp6_size),
            False,
            socket.AF_INET6,
            TCP_TABLE_OWNER_PID_CONNECTIONS,
            0,
        )
        if ret == 0:
            for row in self._tcp6.table[: self._tcp6.dwNumEntries]: