内网在线设备扫描

1
import subprocess
2
import socket
3
import threading
4
import ipaddress
5
import platform
6
from queue import Queue
7
from datetime import datetime
8

9
# 设置扫描的线程数，线程越多扫描越快，但过高可能导致网络拥堵或报错
10
THREAD_COUNT = 100
11
queue = Queue()
12
active_hosts = []
13
lock = threading.Lock()
14

15
def get_local_ip_info():
16
    """
17
    获取本机IP和大致的网段信息
18
    注意：这是一个简化的网段获取方式，假设是常见的 /24 子网
19
    """
20
    try:
21
        # 创建一个UDP套接字连接到外部以获取本机IP（不会实际发送数据）
22
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
23
        s.connect(("8.8.8.8", 80))
24
        local_ip = s.getsockname()[0]
25
        s.close()
26

27
        # 假设子网掩码是 255.255.255.0 (/24)
28
        # 将IP的最后一段替换为0，例如 192.168.1.5 -> 192.168.1.0/24
29
        network_prefix = '.'.join(local_ip.split('.')[:-1]) + '.0/24'
30
        return local_ip, network_prefix
31
    except Exception as e:
32
        print(f"获取本机IP失败: {e}")
33
        return None, None
34

35
def ping_host(ip):
36
    """
37
    对单个IP执行Ping操作
38
    Windows下参数是 -n, Linux/Mac下是 -c
39
    """
40
    param = '-n' if platform.system().lower() == 'windows' else '-c'
41
    # timeout参数在不同系统表现不同，Windows ping 默认等待时间较长
42
    # -w 500 表示等待500毫秒
43
    timeout_param = '-w' if platform.system().lower() == 'windows' else '-W'
44
    timeout_val = '500' if platform.system().lower() == 'windows' else '1'
45

46
    command = ['ping', param, '1', timeout_param, timeout_val, str(ip)]
47

48
    # 禁止弹出控制台窗口 (Windows特有优化)
49
    startupinfo = None
50
    if platform.system().lower() == 'windows':
51
        startupinfo = subprocess.STARTUPINFO()
52
        startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
53

54
    try:
55
        response = subprocess.call(
56
            command,
57
            stdout=subprocess.DEVNULL,
58
            stderr=subprocess.DEVNULL,
59
            startupinfo=startupinfo
60
        )
61
        return response == 0
62
    except Exception:
63
        return False
64

65
def get_hostname(ip):
66
    """尝试解析主机名"""
67
    try:
68
        hostname, _, _ = socket.gethostbyaddr(str(ip))
69
        return hostname
70
    except socket.herror:
71
        return "Unknown"
72

73
def worker():
74
    """线程工作函数"""
75
    while True:
76
        ip = queue.get()
77
        if ip is None:
78
            break
79

80
        if ping_host(ip):
81
            hostname = get_hostname(ip)
82
            with lock:
83
                print(f"[+] 发现设备: {ip} ({hostname})")
84
                active_hosts.append({'ip': str(ip), 'hostname': hostname})
85

86
        queue.task_done()
87

88
def scan_network():
89
    print("-" * 50)
90
    print("简单 Python 内网扫描器 (基于 Ping)")
91
    print("-" * 50)
92

93
    local_ip, network_str = get_local_ip_info()
94
    if not local_ip:
95
        return
96

97
    print(f"本机 IP: {local_ip}")
98
    print(f"扫描目标网段: {network_str}")
99
    print(f"开始扫描... (线程数: {THREAD_COUNT})")
100
    print("-" * 50)
101

102
    start_time = datetime.now()
103

104
    # 创建网络对象
105
    try:
106
        network = ipaddress.ip_network(network_str, strict=False)
107
    except ValueError:
108
        print("网段格式错误")
109
        return
110

111
    # 启动线程
112
    threads = []
113
    for _ in range(THREAD_COUNT):
114
        t = threading.Thread(target=worker)
115
        t.daemon = True # 设置为守护线程
116
        t.start()
117
        threads.append(t)
118

119
    # 将所有IP放入队列 (跳过网络地址和广播地址)
120
    for ip in network.hosts():
121
        queue.put(str(ip))
122

123
    # 等待队列清空
124
    queue.join()
125

126
    # 停止线程
127
    for _ in range(THREAD_COUNT):
128
        queue.put(None)
129
    for t in threads:
130
        t.join()
131

132
    end_time = datetime.now()
133
    duration = end_time - start_time
134

135
    print("-" * 50)
136
    print(f"扫描完成。耗时: {duration}")
137
    print(f"共发现 {len(active_hosts)} 个在线设备:")
138

139
    # 简单的排序并输出
140
    active_hosts.sort(key=lambda x: ipaddress.IPv4Address(x['ip']))
141
    for host in active_hosts:
142
        print(f"IP: {host['ip'].ljust(15)} | Hostname: {host['hostname']}")
143

144
if __name__ == "__main__":
145
    scan_network()
146
    input("\n按回车键退出...")