1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
| import ipaddress
import logging
import time
from multiprocessing import Queue, Process
from scapy.layers.l2 import Ether, ARP
from scapy.sendrecv import srp
logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
def scapy_arp_requests(host, queue=None, ifname="eth0"):
"""
构造ARP包进行扫描
:param host: 需要扫描的主机
:param queue: 存储数据的队列
:param ifname: 网卡名称
:return: 无队列时返回 IP 和 Mac 地址,否则返回 None
"""
# 构造ARP包
result_raw = srp(Ether(dst="FF:FF:FF:FF:FF:FF")
/ ARP(op=1, hwdst="00:00:00:00:00:00", pdst=host),
timeout=1, iface=ifname, verbose=False)
try:
# 取出成功响应的ARP包数据
result_list = result_raw[0].res
if queue == None:
return result_list[0][1].getlayer(ARP).fields['hwsrc']
else:
# 将数据加入队列
queue.put((host, result_list[0][1].getlayer(ARP).fields['hwsrc']))
except:
return
def scan(network, func):
"""
扫描主机
:param network: 扫描的网段
:param func: 扫描调用的函数
:return:
"""
queue = Queue()
net = ipaddress.ip_network(network)
for ip in net:
ip = str(ip)
arp = Process(target=func, args=(ip, queue)) # 创建进程
arp.start() # 开始进程
time.sleep(3)
successful_mac_list = []
while not queue.empty():
ip, mac = queue.get()
successful_mac_list.append((ip, mac))
return successful_mac_list
if __name__ == '__main__':
network = input("Please enter the network segment to be scanned:")
start = time.time()
print("Start scanning ...")
active_ip = scan(network, scapy_arp_requests)
print("Scan complete!")
print("The hosts successfully scanned are:")
for i, (ip, mac) in enumerate(active_ip):
print("{}: IP:{} -- Mac:{}".format(i, ip, mac))
print("\nA total of {} addresses were successful!\n".format(len(active_ip)))
end = time.time()
print("This scan takes a total of {} seconds.".format(end - start))
|