Python子网扫描器socket.connect()冻结问题及解决咨询

问题描述 #

我正在尝试用Python编写一个子网扫描器：代码第一部分接收用户输入并确定子网，第二部分遍历子网下全部255台主机。但扫描离线主机时，代码会出现冻结情况。

现有代码 #

main.py #

import os
import socket


def spread():
    ip_list = []
    lst = []
    new_ip = []
    ip = input("Enter any ip on your network: ")

    for ch in range(0, len(ip)):
        new_ip += ip[ch]
        if ip[ch] == ".":
            print(ip[ch])
            lst.append(ch)

    t = True
    while t == True:
        try:
            new_ip.pop(lst[2] + 1)
        except:
            t = False
            print(new_ip)
            break
        else:
            pass

    target = ""
    for char in new_ip:
        target += char
    print(target)

    #print(f"{ip} == {new_ip}")
    for i in range(1,255):
        print("socket initialized")
        from portscanner import scanner as scan
        for port in range(1,1025):
            try:

                scan(target + str(i))
                #HERE IS WHERE THE CODE FREEZES 
            except:
                continue
            else:
                pass

portscanner.py #

def scanner(ip):
    for port in range(2, 1025):
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        socket.setdefaulttimeout(0.002)

        # returns an error indicator
        try:
            result = s.connect_ex((ip, port))
            #More specifically, here is where the code freezes
        except:
            continue
        else:
            pass
        if result == 0:
            print("Port {} is open".format(port))
        else:
            print("Port {} is CLOSED on ".format(port) + ip)
            continue
            #print(port)
        s.close()

问题原因 #

代码冻结的核心原因是：对离线主机发起TCP连接请求时，单线程下逐个端口尝试连接的等待时间叠加，导致程序假死；同时代码存在循环内重复导入模块、超时设置不合理等结构问题，进一步加剧了性能瓶颈。

解决方案 #

1. 先检测主机存活，跳过无效扫描 #

在扫描端口前，通过ICMP ping判断主机是否在线，避免对离线主机做无意义的端口扫描：

import subprocess

def is_host_alive(ip):
    # 根据系统调整ping参数：Windows用-n，Linux/macOS用-c
    param = '-n' if os.name == 'nt' else '-c'
    # 发送1个包，超时500ms
    command = ['ping', param, '1', '-w', '500', ip]
    return subprocess.call(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) == 0

2. 优化超时设置与代码结构 #

• 将超时设置移到函数外部，避免重复配置
• 把模块导入放在文件开头，禁止循环内重复导入
• 确保socket连接后及时关闭，避免资源泄漏

修改后的portscanner.py：

import socket

# 设置合理的超时时间，0.002过短易导致误判
socket.setdefaulttimeout(0.5)

def scanner(ip):
    for port in range(2, 1025):
        s = None
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            result = s.connect_ex((ip, port))
            if result == 0:
                print(f"Port {port} is open on {ip}")
        except Exception:
            continue
        finally:
            if s:
                s.close()

3. 多线程并行扫描，避免程序冻结 #

单线程逐个扫描速度极慢，用threading实现多线程并行处理，提升扫描效率：

import threading

def scan_host(ip):
    if is_host_alive(ip):
        print(f"Host {ip} is alive, scanning ports...")
        scanner(ip)
    else:
        print(f"Host {ip} is offline")

4. 简化子网提取逻辑 #

原代码提取子网的逻辑过于繁琐，用字符串分割简化：

ip = input("Enter any ip on your network: ")
# 取IP前三段组成子网前缀
target = '.'.join(ip.split('.')[:3]) + '.'

最终优化后的main.py示例 #

import os
import subprocess
import threading
from portscanner import scanner

def is_host_alive(ip):
    param = '-n' if os.name == 'nt' else '-c'
    command = ['ping', param, '1', '-w', '500', ip]
    return subprocess.call(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) == 0

def scan_host(ip):
    if is_host_alive(ip):
        print(f"Host {ip} is alive, scanning ports...")
        scanner(ip)
    else:
        print(f"Host {ip} is offline")

def spread():
    ip = input("Enter any ip on your network: ")
    target = '.'.join(ip.split('.')[:3]) + '.'
    print(f"Scanning subnet: {target}0/24")

    threads = []
    for i in range(1,255):
        host_ip = target + str(i)
        thread = threading.Thread(target=scan_host, args=(host_ip,))
        threads.append(thread)
        thread.start()
    
    # 等待所有线程执行完成
    for thread in threads:
        thread.join()

if __name__ == "__main__":
    spread()

内容的提问来源于stack exchange，提问作者Pork Lord