| 0. 1. 2. 3. 4. 5. 6. 7. 8. 9. 10. 11. 12. 13. 14. 15. 16. 17. 18. 19. 20. 21. 22. 23. 24. 25. 26. 27. 28. 29. 30. 31. 32. 33. 34. 35. 36. 37. 38. 39. 40. 41. 42. 43. 44. 45. 46. 47. 48. 49. 50. 51. 52. 53. 54. 55. 56. 57. 58. 59. 60. 61. 62. 63. 64. 65. 66. 67. 68. 69. 70. 71. 72. 73. 74. 75. 76. 77. 78. 79. 80. 81. 82. 83. 84. 85. 86. 87. 88. 89. 90. 91. 92. 93. 94. 95. 96. 97. 98. 99. 100. 101. 102. 103. 104. 105. 106. 107. 108. 109. 110. 111. 112. 113. 114. 115. 116. 117. 118. 119. 120. 121. 122. 123. 124. 125. 126. 127. 128. 129. 130. 131. 132. 133. 134. 135. 136. 137. 138. 139. 140. 141. 142. 143. 144. 145. 146. 147. 148. | # openPorts.py import ctypes import socket import struct def getOpenPorts(): """ This function will return a list of ports (TCP/UDP) that the current machine is listening on. It's basically a replacement for parsing netstat output but also serves as a good example for using the IP Helper API: http://msdn.microsoft.com/library/default.asp?url=/library/en- us/iphlp/iphlp/ip_helper_start_page.asp. I also used the following post as a guide myself (in case it's useful to anyone): http://aspn.activestate.com/ASPN/Mail/Message/ctypes-users/1966295 """ portList = [] DWORD = ctypes.c_ulong NO_ERROR = 0 NULL = "" bOrder = 0 # define some MIB constants used to identify the state of a TCP port MIB_TCP_STATE_CLOSED = 1 MIB_TCP_STATE_LISTEN = 2 MIB_TCP_STATE_SYN_SENT = 3 MIB_TCP_STATE_SYN_RCVD = 4 MIB_TCP_STATE_ESTAB = 5 MIB_TCP_STATE_FIN_WAIT1 = 6 MIB_TCP_STATE_FIN_WAIT2 = 7 MIB_TCP_STATE_CLOSE_WAIT = 8 MIB_TCP_STATE_CLOSING = 9 MIB_TCP_STATE_LAST_ACK = 10 MIB_TCP_STATE_TIME_WAIT = 11 MIB_TCP_STATE_DELETE_TCB = 12 ANY_SIZE = 1 # defing our MIB row structures class MIB_TCPROW(ctypes.Structure): _fields_ = [('dwState', DWORD), ('dwLocalAddr', DWORD), ('dwLocalPort', DWORD), ('dwRemoteAddr', DWORD), ('dwRemotePort', DWORD)] class MIB_UDPROW(ctypes.Structure): _fields_ = [('dwLocalAddr', DWORD), ('dwLocalPort', DWORD)] dwSize = DWORD(0) # call once to get dwSize ctypes.windll.iphlpapi.GetTcpTable(NULL, ctypes.byref(dwSize), bOrder) # ANY_SIZE is used out of convention (to be like MS docs); even setting this # to dwSize will likely be much larger than actually necessary but much # more efficient that just declaring ANY_SIZE = 65500. # (in C we would use malloc to allocate memory for the *table pointer and # then have ANY_SIZE set to 1 in the structure definition) ANY_SIZE = dwSize.value class MIB_TCPTABLE(ctypes.Structure): _fields_ = [('dwNumEntries', DWORD), ('table', MIB_TCPROW * ANY_SIZE)] tcpTable = MIB_TCPTABLE() tcpTable.dwNumEntries = 0 # define as 0 for our loops sake # now make the call to GetTcpTable to get the data if (ctypes.windll.iphlpapi.GetTcpTable(ctypes.byref(tcpTable), ctypes.byref(dwSize), bOrder) == NO_ERROR): maxNum = tcpTable.dwNumEntries placeHolder = 0 # loop through every connection while placeHolder < maxNum: item = tcpTable.table[placeHolder] placeHolder += 1 # format the data we need (there is more data if it is useful - # see structure definition) lPort = item.dwLocalPort lPort = socket.ntohs(lPort) lAddr = item.dwLocalAddr lAddr = socket.inet_ntoa(struct.pack('L', lAddr)) portState = item.dwState # only record TCP ports where we're listening on our external # (or all) connections if str(lAddr) != "127.0.0.1" and portState == MIB_TCP_STATE_LISTEN: portList.append(str(lPort) + "/TCP") else: print "Error occurred when trying to get TCP Table" dwSize = DWORD(0) # call once to get dwSize ctypes.windll.iphlpapi.GetUdpTable(NULL, ctypes.byref(dwSize), bOrder) ANY_SIZE = dwSize.value # again, used out of convention # (see notes in TCP section) class MIB_UDPTABLE(ctypes.Structure): _fields_ = [('dwNumEntries', DWORD), ('table', MIB_UDPROW * ANY_SIZE)] udpTable = MIB_UDPTABLE() udpTable.dwNumEntries = 0 # define as 0 for our loops sake # now make the call to GetUdpTable to get the data if (ctypes.windll.iphlpapi.GetUdpTable(ctypes.byref(udpTable), ctypes.byref(dwSize), bOrder) == NO_ERROR): maxNum = udpTable.dwNumEntries placeHolder = 0 while placeHolder < maxNum: item = udpTable.table[placeHolder] placeHolder += 1 lPort = item.dwLocalPort lPort = socket.ntohs(lPort) lAddr = item.dwLocalAddr lAddr = socket.inet_ntoa(struct.pack('L', lAddr)) # only record UDP ports where we're listening on our external # (or all) connections if str(lAddr) != "127.0.0.1": portList.append(str(lPort) + "/UDP") else: print "Error occurred when trying to get UDP Table" portList.sort() return portList ListofOpenPorts = getOpenPorts() |