我试图维护用户在文本框中给出的子网中的 IP 地址列表,然后使用最多 50 个线程同时对它们进行 nmap,并在完成后立即显示每个 IP 的结果。代码抛出多个错误,特别是在子进程部分:-
# Scanning subnet of IP Addresses using NMAP
def Scan_IP(self):
num_threads = 50
global user_IP, i, scanout_q
scanin_q = queue.Queue()
scanout_q = queue.Queue()
text = str(self.textbox.text().strip())
# Fetching IP/ Subnet
if text.upper().isupper() is False:
try:
def scan(i, q):
while True:
# get an IP item from queue
IP = q.get()
# scan nw devices
arg = f'nmap, -Pn, --open, -T5, {str(IP)}'
p4 = subprocess.Popen(arg, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, encoding="utf-8", shell=True)
p5 = subprocess.Popen(
["awk", "/scan report/; /STATE/; /tcp/; /type:/; Running/; /details:/; /Service Info:/"],
stdin=p4.stdout, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, encoding="utf-8", shell = True)
output = p5.stdout.communicate()
# But do not wait till nmap finish, start displaying output immediately #
while True:
output = output.stderr.read()
if output == '':
pass
else:
sys.scanout_q.write(output)
sys.scanout_q.put(output)
sys.scanout_q.flush()
q.task_done()
# start thread pool
for i in range(num_threads):
worker = Thread(target=scan, args=(i, scanin_q))
worker.setDaemon(True)
worker.start()
# entering it into queue/ fill queue
IP_List = []
for IP in IPv4Network(text):
scanin_q.put(IP)
IP_List.append(IP)
# queue management: wait until worker threads are done to exit
scanin_q.join()
except Exception as e:
print(e)
else:
errorMsg = '\nInvalid IP Address/ Subnet; Try again!'
print(errorMsg)
错误:- 回溯(最近一次调用最后一次): 文件“C:\Users\user01\AppData\Local\Programs\Python\Python39\lib hreading.py”,第 917 行,运行 self._target(*self._args, **self._kwargs
文件“E:\PyCodes\NMAP.py”,第 218 行,扫描输出 = p5.stdout.communicate()
AttributeError:“_io.TextIOWrapper”对象没有属性“communicate”
最好运行
nmap
的单个实例并扫描多个目标。
可能有用的 nmap 选项(参见 man nmap
):
-iL <inputfilename>
目标列表
-oG -
以 grepable 格式输出到标准输出