在 python 中使用 GetExtendedTcpTable 时出错

Question

我在使用“GetExtendedTcpTable”时遇到了一些麻烦。当我尝试运行我的脚本时，我收到了这样的消息：

AssertionError: [Error 0] 操作成功完成

很少有脚本正常工作，我不明白这个消息，操作完成，怎么了？

这是代码，我尝试执行：

from ctypes import *
from ctypes.wintypes import *
from socket import  inet_aton,   inet_ntoa,  htons


AF_INET = 2
TCP_TABLE_BASIC_LISTENER = 0
TCP_TABLE_BASIC_CONNECTIONS = 1
TCP_TABLE_BASIC_ALL = 2
TCP_TABLE_OWNER_PID_LISTENER = 3
TCP_TABLE_OWNER_PID_CONNECTIONS = 4
TCP_TABLE_OWNER_PID_ALL = 5
TCP_TABLE_OWNER_MODULE_LISTENER = 6
TCP_TABLE_OWNER_MODULE_CONNECTIONS = 7
TCP_TABLE_OWNER_MODULE_ALL = 8

# for storing socket info python style.  
class socket_info:

    State = None
    LocalAddr = None
    LocalPort = None
    RemoteAddr = None
    RemotePort = None

    def __init__ (self, **kwargs):

        for key, word in kwargs.items():
            setattr(self, key, word)

def formatip (ip):
    ip = inet_aton (str(ip))
    return inet_ntoa (ip[::-1])

states = {
    1 : "TCP_STATE_CLOSED",
    2 : "TCP_STATE_LISTEN",
    3 : "TCP_STATE_SYN_SENT",
    4 : "TCP_STATE_SYN_RCVD",
    5 : "TCP_STATE_ESTAB",
    6 : "TCP_STATE_FIN_WAIT",
    7 : "TCP_STATE_FIN_WAIT2",
    8 : "TCP_STATE_CLOSE_WAIT",
    9 : "TCP_STATE_CLOSING",
    10 : "TCP_STATE_LAST_ACK",
    11 : "TCP_STATE_TIME_WAIT",
    12 : "TCP_STATE_DELETE_TCB",

    "TCP_STATE_CLOSED" : 1,
    "TCP_STATE_LISTEN" : 2,
    "TCP_STATE_SYN_SENT" : 3,
    "TCP_STATE_SYN_RCVD" : 4,
    "TCP_STATE_ESTAB" : 5,
    "TCP_STATE_FIN_WAIT" : 6,
    "TCP_STATE_FIN_WAIT2" : 7,
    "TCP_STATE_CLOSE_WAIT" : 8,
    "TCP_STATE_CLOSING" : 9,
    "TCP_STATE_LAST_ACK" :10,
    "TCP_STATE_TIME_WAIT" : 11,
    "TCP_STATE_DELETE_TCB" : 12 }

class MIB_TCPROW_OWNER_PID(Structure):
    _fields_ = [
        ("dwState", DWORD),
        ("dwLocalAddr", DWORD),
        ("dwLocalPort", DWORD),
        ("dwRemoteAddr", DWORD),
        ("dwRemotePort", DWORD),
        ("dwOwningPid", DWORD)
        ]


class MIB_TCPTABLE_OWNER_PID(Structure):
    _fields_ = [
        ("dwNumEntries", DWORD),
        ("MIB_TCPROW_OWNER_PID", MIB_TCPROW_OWNER_PID * 100)
        ]



def GetExtendedTcpTable (vip=AF_INET):
    table = MIB_TCPTABLE_OWNER_PID ()
    so = sizeof (table)
    size = DWORD (so)
    order = c_int(1)

    failure= windll.iphlpapi.GetExtendedTcpTable (
        byref (table),
        addressof (size),
        order,
        vip,
        TCP_TABLE_OWNER_PID_ALL,
        0    )

    assert not failure,  WinError (GetLastError ())

    pytables = []
    tables = table.MIB_TCPROW_OWNER_PID

    for index in range(table.dwNumEntries):
        table = tables [index]
        pytables.append (
            socket_info (
                State=states.get (table.dwState, "UNKNOWN_STATE_%s" %(str(table.dwState))),
                LocalAddr=formatip (table.dwLocalAddr),
                LocalPort=htons(table.dwLocalPort),
                RemoteAddr=formatip (table.dwRemoteAddr),
                RemotePort=htons(table.dwRemotePort),
                OwningPid = int (table.dwOwningPid)
            )
        )
    return pytables


def GetTcpTableForPid (pid):
    tables = GetExtendedTcpTable ()
    for table in tables:
        if table.OwningPid == pid: return table
    raise "Cannot find tcp table for pid %s" %pid

dict_process = {}
pid_set =set()
pid_list = []
tcp_info_list = []
tcp_info = GetExtendedTcpTable()
for item in tcp_info:
    LocalAddr = item.LocalAddr
    LocalPort = item.LocalPort
    RemoteAddr = item.RemoteAddr
    RemotePort = item.RemotePort
    OwningPid = item.OwningPid
    print('local Addr: '+ LocalAddr,'local port: '+ str(LocalPort),'remote Addr: ' + RemoteAddr, 'Remote Port: ' + str(RemotePort), OwningPid)

脚本会不时运行。它可以运行 5 分钟，然后由于这个愚蠢的错误而无法运行大约一个小时。如何绕过它？

我真的不知道，它是怎么回事。请帮帮我，我做错了什么？

我在 Win7 SP1 x64 上使用 python 3.2

非常感谢！

Answer 1

您不应该使用addressof(size)。这将返回一个 Python 整数，该整数将被转换为 32 位 C int。使用byref(size) 创建一个指针，如果您使用的是 64 位 Python，它将是一个 64 位值。

GetExtendedTcpTable 不调用SetLastError。它返回带有以下代码之一的DWORD：

NO_ERROR = 0
ERROR_INVALID_PARAMETER = 87
ERROR_INSUFFICIENT_BUFFER = 122

如果缓冲区太小，pdwSize 参数具有所需的大小。这里的一种选择是从长度为 0 的数组开始；然后resize 结构；最后将cast 数组调整为正确的大小：

class MIB_TCPTABLE_OWNER_PID(Structure):
    _fields_ = [
        ("dwNumEntries", DWORD),
        ("MIB_TCPROW_OWNER_PID", MIB_TCPROW_OWNER_PID * 0),
    ]

_GetExtendedTcpTable = windll.iphlpapi.GetExtendedTcpTable

def GetExtendedTcpTable(vip=AF_INET):
    table = MIB_TCPTABLE_OWNER_PID()
    size = DWORD() 
    order = 1

    failure = _GetExtendedTcpTable(
                  byref(table),
                  byref(size),
                  order,
                  vip,
                  TCP_TABLE_OWNER_PID_ALL,
                  0)

    if failure == ERROR_INSUFFICIENT_BUFFER:
        resize(table, size.value)
        memset(byref(table), 0, sizeof(table))
        failure = _GetExtendedTcpTable(
                      byref(table),
                      byref(size),
                      order,
                      vip,
                      TCP_TABLE_OWNER_PID_ALL,
                      0)

    if failure: 
        raise WinError(failure)

    ptr_type = POINTER(MIB_TCPROW_OWNER_PID * table.dwNumEntries)
    tables = cast(table.MIB_TCPROW_OWNER_PID, ptr_type)[0]

    pytables = []
    for table in tables:
        # rest unchanged

---

关于 Win32 的 LastError 值，一般来说你不应该依赖 Python 中的 GetLastError。您不知道您是否看到来自先前调用的旧错误代码，或者是否有干预调用修改了 LastError 值。如果您正在检查使用 LastError 的单个 API 调用，那么如果调用失败，则应该可以立即检查 GetLastError。但更一般地，您可能需要使用 use_last_error=True 加载 DLL：

iphlpapi = WinDLL('iphlpapi', use_last_error=True)

从此WinDLL 实例创建的函数指针将在调用返回后立即将LastError 保存到线程本地存储。调用 get_last_error 返回保存的错误代码。在调用函数之前，您可以事先调用set_last_error(0) 将0 换成LastError。