Python 多服务器批量端口扫描工具|内网资产一键巡检

import socket                  from concurrent.futures import ThreadPoolExecutor, as_completed                  import datetime                  # 内置全网常见高危端口&服务映射                  COMMON_PORTS = {                  21: "FTP",                  22: "SSH",                  23: "Telnet",                  25: "SMTP",                  53: "DNS",                  80: "HTTP",                  110: "POP3",                  135: "RPC",                  139: "NetBIOS",                  143: "IMAP",                  443: "HTTPS",                  445: "SMB",                  993: "IMAPS",                  995: "POP3S",                  1433: "MSSQL",                  1521: "Oracle",                  3306: "MySQL",                  3389: "RDP",                  5432: "PostgreSQL",                  6379: "Redis",                  27017: "MongoDB",                  8080: "HTTP-Alt",                  8443: "HTTPS-Alt",                  }                  def check_port(ip, port, timeout=1):                  """                  单端口检测函数                  :param ip: 目标IP                  :param port: 目标端口                  :param timeout: 连接超时时间                  :return: 端口开放返回True，否则False                  """                  try:                  with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:                  sock.settimeout(timeout)                  connect_code = sock.connect_ex((ip, port))                  return connect_code == 0                  except Exception:                  return False                  def scan_single_target(ip, ports=None, timeout=1, max_workers=50):                  """                  单台服务器全端口扫描                  :param ip: 目标IP地址                  :param ports: 自定义扫描端口，默认扫描内置高危端口                  :param timeout: 超时时间                  :param max_workers: 单IP扫描并发线程数                  :return: 排序后的开放端口列表                  """                  if ports is None:                  ports = list(COMMON_PORTS.keys())                  open_ports = []                  with ThreadPoolExecutor(max_workers=max_workers) as pool:                  task_map = {pool.submit(check_port, ip, port, timeout): port for port in ports}                  for future in as_completed(task_map):                  port = task_map[future]                  if future.result():                  open_ports.append(port)                  return sorted(open_ports)                  def batch_scan(ip_list, timeout=1, thread_num=80):                  """                  多服务器批量扫描主函数                  :param ip_list: 待扫描IP列表                  :param timeout: 端口连接超时                  :param thread_num: 单服务器扫描并发数                  """                  total_start = datetime.datetime.now()                  print(f"===== 批量端口扫描启动，共 {len(ip_list)} 台服务器 =====\n")                  scan_result = {}                  # 逐台扫描服务器                  for target_ip in ip_list:                  ip_start = datetime.datetime.now()                  print(f"正在扫描服务器: {target_ip}")                  open_port_list = scan_single_target(target_ip, timeout=timeout, max_workers=thread_num)                  scan_result[target_ip] = open_port_list                  cost = (datetime.datetime.now() - ip_start).total_seconds()                  print(f"{target_ip} 扫描完成，耗时 {cost:.2f} 秒\n")                  # 输出全局扫描报告                  print("===== 全部服务器扫描完成｜汇总报告 =====")                  total_cost = (datetime.datetime.now() - total_start).total_seconds()                  print(f"本次批量扫描总耗时：{total_cost:.2f} 秒\n")                  # 格式化展示每台服务器结果                  for ip, ports in scan_result.items():                  print(f"【目标服务器】{ip}")                  if ports:                  print(f"发现 {len(ports)} 个开放高危端口：")                  for port in ports:                  service = COMMON_PORTS.get(port, "未知服务")                  print(f"端口 {port}：{service}")                  else:                  print("未检测到常见高危端口开放")                  print("-" * 50)                  if __name__ == "__main__":                  # ================== 配置区域（自行修改）==================                  # 填写需要批量扫描的所有服务器IP                  SCAN_IP_LIST = [                  "192.168.52.15",                  "192.168.52.16",                  "192.168.52.17",                  "127.0.0.1"           ]                  SCAN_TIMEOUT = 1# 单端口超时时间                  SCAN_THREAD = 80# 单服务器并发线程数                  # ======================================================                  # 执行批量扫描                  batch_scan(SCAN_IP_LIST, timeout=SCAN_TIMEOUT, thread_num=SCAN_THREAD)                  

python port_scan.py