1

I'm using the ctypes module to call GetTcpTable2. I've been slowly converting the example here in C++ to Python; but am getting crash during a field access.

if __name__ == "__main__":
    ptcp_table = POINTER(MIB_TCPTABLE2)()
    ptcp_table = cast(create_string_buffer(sizeof(MIB_TCPTABLE2)),
                      POINTER(MIB_TCPTABLE2))
    ip_addr = in_addr()
    size = c_ulong(sizeof(MIB_TCPTABLE2))
    retval = GetTcpTable2(ptcp_table, byref(size), TRUE)
    if retval == ERROR_INSUFFICIENT_BUFFER:
        ptcp_table = cast(create_string_buffer(size.value),
                          POINTER(MIB_TCPTABLE2))
        if not ptcp_table:
            #throw error
            pass

    retval = GetTcpTable2(ptcp_table, byref(size), TRUE)
    if retval == NO_ERROR:
        print("Entries %d" % ptcp_table[0].dwNumEntries)
        for i in range(0, ptcp_table[0].dwNumEntries):
            print(ptcp_table[0].table[i])
            #ip_addr.S_un.S_addr = ptcp_table[0].table[i].dwLocalAddr
            #ip_addr_string = inet_nota(ip_addr)
            #print(ip_addr_string)
            #print(string_at(ip_addr_string))

It crashes when trying to access dwLocalAddr apart of table[i].

ptcp_table[0].table[i].dwLocalAddr

However it doesn't crash when just printing ptcp_table[0].table[i]. I've tried printing and accessing other fields; but Python just crashes.

Here are my struct definitions:

class MIB_TCPROW2(Structure):
    _fields_ = [
        ("dwState", c_ulong),
        ("dwLocalAddr", c_ulong),
        ("dwLocalPort", c_ulong),
        ("dwRemoteAddr", c_ulong),
        ("dwRemotePort", c_ulong),
        ("dwOwningPid", c_ulong),
        ("dwOffloadState", c_int)
    ]


class MIB_TCPTABLE2(Structure):
    _fields_ = [
        ("dwNumEntries", c_ulong),
        ("table", POINTER(MIB_TCPROW2))
    ]

Definition of GetTcpTable2:

GetTcpTable2 = windll.iphlpapi.GetTcpTable2
GetTcpTable2.argtypes = [POINTER(MIB_TCPTABLE2), POINTER(c_ulong), c_char]
GetTcpTable2.restype = c_ulong

I have a small hunch that in the definition of the MIB_TCPTABLE2 struct; the documentation says that table is an array of MIB_TCPROW2 size ANY_SIZE; and further inspection is that ANY_SIZE is 1 from checking the iphlpapi.h file. And I know that the size of POINTER(MIB_TCPROW2) doesn't equal the size of MIB_TCPROW2.

jacob
  • 4,656
  • 1
  • 23
  • 32
  • 1
    `MIB_TCPTABLE2` is a variable-sized structure; `dwNumEntries` tells you how big the `table` array actually is. How you represent that to Python I have no idea though. – Jonathan Potter Jul 24 '16 at 21:48
  • @JonathanPotter Thank you for pointing this out. I've done a bit more searching and looked about creating a class factory method for `MIB_TCPTABLE2` to generate variable lengthed versions of it. – jacob Jul 24 '16 at 22:01
  • @J.J.Hakala Have to wait 2 days before I can do that. (3 hours from now) – jacob Jul 26 '16 at 18:27

2 Answers2

1

I looked into other ctypes questions revolving around variable length fields inside of a struct, and came to an answer which suggested to use a factory method to generate the class definitions.

def MIB_TCPTABLE2_FACTORY(size):
    class MIB_TCPTABLE2(Structure):
        _fields_ = [
            ("dwNumEntries", c_ulong),
            ("table", MIB_TCPROW2 * size)
        ]
    return MIB_TCPTABLE2

I can use this knowing the size returned from GetTcpTable2 to create a new type. And then all I have to do is change the argtypes of GetTcpTable2 to accept a void *.

GetTcpTable2.argtypes = [c_void_p, POINTER(c_ulong), c_char]
Community
  • 1
  • 1
jacob
  • 4,656
  • 1
  • 23
  • 32
0

This is how I solved it. I first got the required size by passing in the following arguments:

ret = windll.iphlpapi.GetTcpTable2(None, byref(tcp_table_size), True)

Notice None is the first argument, which is equivalent to NULL when passing it into ctypes windows function. Then I defined the MIB_TCPTABLE2 class and inside it, I passed in the size returned by the first call to GetTcpTable2:

class MIB_TCPTABLE2(Structure):
    _fields_ = [
        ("dwNumEntries", c_ulong),
        ("table", MIB_TCPROW2 * tcp_table_size.value),
    ]

Next, I created an instance of the structure and called GetTcpTable2 again passing in the newly created structure:

tcp_table = MIB_TCPTABL2()

ret = windll.iphlpapi.GetTcpTable2(byref(tcp_table), byref(tcp_table_size), True)

Here's the sample code:

from ctypes import *
import socket
import struct

NO_ERROR = 0
ERROR_INSUFFICIENT_BUFFER = 122

TcpConnectionOffloadStateInHost = 0
TcpConnectionOffloadStateOffloading = 1
TcpConnectionOffloadStateOffloaded = 2
TcpConnectionOffloadStateUploading = 3
TcpConnectionOffloadStateMax = 4

class MIB_TCPROW2(Structure):
    _fields_ = [
        ("dwState", c_ulong),
        ("dwLocalAddr", c_ulong),
        ("dwLocalPort", c_ulong),
        ("dwRemoteAddr", c_ulong),
        ("dwRemotePort", c_ulong),
        ("dwOwningPid", c_ulong),
        ("dwOffloadState", c_ulong),
    ]
    
def main():
    windll.iphlpapi.GetTcpTable2.argtypes = [c_void_p, POINTER(c_ulong), c_bool]
    tcp_table_size = c_ulong()
        
    ret = windll.iphlpapi.GetTcpTable2(None, byref(tcp_table_size), True)
    if ret == ERROR_INSUFFICIENT_BUFFER:
        class MIB_TCPTABLE2(Structure):
            _fields_ = [
                ("dwNumEntries", c_ulong),
                ("table", MIB_TCPROW2 * tcp_table_size.value),
            ]
        
        tcp_table = MIB_TCPTABLE2()
        
        ret = windll.iphlpapi.GetTcpTable2(byref(tcp_table), byref(tcp_table_size), True)
        if ret != NO_ERROR:
            print("ERROR: GetTcpTable2() failed, error = " + str(ret))
        else:
            for i in range(tcp_table.dwNumEntries):
                dest_ip = socket.inet_ntoa(struct.pack('<L', tcp_table.table[i].dwRemoteAddr))
                print("PID: " + str(tcp_table.table[i].dwOwningPid) + ", DEST IP: " + dest_ip)
    
    
if __name__ == "__main__":
    main()
jaemszy
  • 31
  • 2