python3检测一批主机ip的tcp端口

待检测的服务器ip地址;每行一个ip地址
`cat hostip.txt`
```
100.50.235.116
100.50.118.26
```

要检测哪些tcp端口，修改脚本中
`portlist = [9090, 10000, 20000, 30000]`

运行后生成的日志文件;[+]表示通，[-]表示不通
`cat port_scan_host.log`
```
2021 - 07 - 09 17 : 22 : 27 IP = 100.50.235.116 [+]网络tcp端口 9090 open
2021 - 07 - 09 17 : 22 : 29 IP = 100.50.118.26 [-]网络tcp端口 9090 close
```

## V 0.1版本，扫描只针对IP用多并发
port_scan_net_v0.1.py
```python
#import subprocess
from socket import *
import logging
import datetime
import time
import threading
from queue import Queue
import re
import sys
# 实现日志导出

#批量扫描端口，根据批量ping修改为tcp端口扫描
#hostip.txt中每行写一个ip地址，运行时根据输入ip或net来扫描单个地址或ip对应的C类网段。
#
def set_logging_format():
  logging.basicConfig(level=logging.INFO,
            format='%(message)s',
            filename='port_scan_host.log',
            filemode='w'
            )
  console = logging.StreamHandler()
  console.setLevel(logging.INFO)
  formatter = logging.Formatter('%(message)s')
  console.setFormatter(formatter)
  logging.getLogger('').addHandler(console)
# 将需要ping 连接的IP加入队列
def insert_ip_queue(ip_list_path):
  IP_QUEUE = Queue()
  with open (ip_list_path,'r') as f:
    for ip in f.readlines():
      ip = ip.strip('\n')
      # 判断一个字符串是否包含合法IP地址
      ip = re.search(r'((25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d)\.){3}(25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d)', ip)
      if ip:
        ip = ip.group()  # 提取出合法ip地址
        IP_QUEUE.put(ip)
    f.close()
  return IP_QUEUE
def IP_list (ip_list_path):
  ip_list = Queue()
  with open (ip_list_path,'r') as f:
    for ip in f.readlines():
      ip = ip.strip('\n')
      # 判断一个字符串是否包含合法IP地址
      ip = re.search(r'((25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d)\.){3}(25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d)', ip)
      if ip:
        ip = ip.group()  # 提取出合法ip地址
        list_segment = ip.split('.')
        ip_index = 1
        # 将需要 ping 的 ip 加入队列
        for i in range(0, 254):
          list_segment[-1] = str(ip_index + i)
          addr = ('.').join(list_segment)
          print(addr)
          ip_list.put(addr)
    f.close()
  return ip_list
# print (IP_list())
#print (IP_list())
#定义 ping 函数
def scan_IP (IP_QUEUE):
  while not IP_QUEUE.empty():
    ip = IP_QUEUE.get().strip('\n')
    #print (ip)
    ##res = subprocess.call('ping -w 1000 -n 1 %s' % ip , stdout=subprocess.PIPE,shell=True)
    portlist = [9090, 10000, 20000, 30000]
    for port in portlist:
      try:
        s = socket(AF_INET, SOCK_STREAM)
        s.connect((ip, port))
        print('[+] %d open' % port)
        res = ('[+]网络tcp端口 %d open' % port)
        today = datetime.datetime.now().strftime("%Y - %m - %d %H : %M : %S")
        logging.info("%s IP = %s %s" % (today, ip, res))
        s.close()
      except:
        print('[-] %d close' % port)
        res = ('[-]网络tcp端口 %d close' % port)
        today = datetime.datetime.now().strftime("%Y - %m - %d %H : %M : %S")
        logging.info("%s IP = %s %s" % (today, ip, res))

def main ():
  set_logging_format()
  ip_list_path = './hostip.txt'
  # IP_QUEUE = insert_ip_queue(ip_list_path)
  # IP_LIST = IP_list(ip_list_path)
  threads = []
  THREAD_NUM = 200
  user_iput = input('please input modren:(ip|net) ')
  if user_iput == 'net':
    IP_LIST = IP_list(ip_list_path)
    IP_L = IP_LIST
  if user_iput == 'ip':
    IP_QUEUE = insert_ip_queue(ip_list_path)
    IP_L = IP_QUEUE
  for i in range (THREAD_NUM):
    t = threading.Thread(target = scan_IP, args = (IP_L,))
    threads.append(t)
  for i in range (THREAD_NUM):
    threads[i].start()
  for i in range (THREAD_NUM):
    threads[i].join()
if __name__ == '__main__':
  main()

```

设置超时时间、设置端口为所有
```
# 需要导入模块: import socket [as 别名]
# 或者: from socket import setdefaulttimeout [as 别名]
# socket.setdefaulttimeout(1)

from socket import *
setdefaulttimeout(1)  #设置默认超时1秒  0.08为80毫秒

portlist = list(range(1,65536))
```

## V 0.2版本，扫描IP和端口都用多并发
```python
#import subprocess
from socket import *
import logging
import datetime
import time
import threading
from queue import Queue
import re
import sys
import traceback

setdefaulttimeout(0.08)
# 实现日志导出
#批量扫描端口，根据批量ping修改为tcp端口扫描
#hostip.txt中每行写一个ip地址，运行时根据输入ip或net来扫描单个地址或ip对应的C类网段。
#
def set_logging_format():
  logging.basicConfig(level=logging.INFO,
            format='%(message)s',
            filename='port_scan_host.log',
            filemode='w'
            )
  console = logging.StreamHandler()
  console.setLevel(logging.INFO)
  formatter = logging.Formatter('%(message)s')
  console.setFormatter(formatter)
  logging.getLogger('').addHandler(console)
# 将需要ping 连接的IP加入队列

def IP2ipp(dip):
  IP2ipp_list = []
  portlist = list(range(1,65536))#[3306]
  for dport in portlist:
    try:
      dipport = (':').join([dip,str(dport)])

print(dipport)
      IP2ipp_list.append(dipport)
      #return dipport
    except Exception as e: 
      print(traceback.format_exc())
      # print("addipport-err")
  return IP2ipp_list
      
def insert_ip_queue(ip_list_path):
  IP_QUEUE = Queue()
  with open (ip_list_path,'r') as f:
    for ip in f.readlines():
      ip = ip.strip('\n')
      # 判断一个字符串是否包含合法IP地址
      ip = re.search(r'((25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d)\.){3}(25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d)', ip)
      if ip:
        ip = ip.group()  # 提取出合法ip地址
        ip = IP2ipp(str(ip))
        for i in ip:
          IP_QUEUE.put(i)
    f.close()
  return IP_QUEUE
def IP_list (ip_list_path):
  ip_list = Queue()
  with open (ip_list_path,'r') as f:
    for ip in f.readlines():
      ip = ip.strip('\n')
      # 判断一个字符串是否包含合法IP地址
      ip = re.search(r'((25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d)\.){3}(25[0-5]|2[0-4]\d|1\d{2}|[1-9]?\d)', ip)
      if ip:
        ip = ip.group()  # 提取出合法ip地址
        list_segment = ip.split('.')
        ip_index = 1
        # 将需要 ping 的 ip 加入队列
        for i in range(0, 254):
          list_segment[-1] = str(ip_index + i)
          addr = ('.').join(list_segment)
          print(addr)
          ip = IP2ipp(addr)
          for i in ip:
            ip_list.put(i)
    f.close()
  return ip_list
# print (IP_list())
#print (IP_list())
#定义 ping 函数
def scan_IP (IP_QUEUE):
  while not IP_QUEUE.empty():
    # print(type(IP_QUEUE.get()),IP_QUEUE.get())
    ipp = IP_QUEUE.get().strip('\n')
    #print (ip)
    ##res = subprocess.call('ping -w 1000 -n 1 %s' % ip , stdout=subprocess.PIPE,shell=True)
    ip_port = ipp.split(':')
    #time.sleep(10)
    ip = ip_port[0]
    port = eval(ip_port[1])
    # portlist = list(range(1,65536))#[3306]
    #for port in portlist:
    try:
      s = socket(AF_INET, SOCK_STREAM)
      s.connect((ip, port))
      print('[+] %d open' % port)
      res = ('[+]网络tcp端口 %d open' % port)
      today = datetime.datetime.now().strftime("%Y - %m - %d %H : %M : %S")
      logging.info("%s IP = %s %s" % (today, ip, res))
      s.close()
    except:
      print('[-] %d close' % port)
      res = ('[-]网络tcp端口 %d close' % port)
      today = datetime.datetime.now().strftime("%Y - %m - %d %H : %M : %S")
      logging.info("%s IP = %s %s" % (today, ip, res))
def main ():
  set_logging_format()
  ip_list_path = './hostip.txt'
  # IP_QUEUE = insert_ip_queue(ip_list_path)
  # IP_LIST = IP_list(ip_list_path)
  threads = []
  THREAD_NUM = 2000
  user_iput = input('please input modren:(ip|net) ')
  if user_iput == 'net':
    IP_LIST = IP_list(ip_list_path)
    IP_L = IP_LIST
  if user_iput == 'ip':
    IP_QUEUE = insert_ip_queue(ip_list_path)
    IP_L = IP_QUEUE

for i in range (THREAD_NUM):
    t = threading.Thread(target = scan_IP, args = (IP_L,))
    threads.append(t)
  for i in range (THREAD_NUM):
    threads[i].start()
  for i in range (THREAD_NUM):
    threads[i].join()
if __name__ == '__main__':
  main()

```