Ping+多线程/协程的Python脚本

前言

前些日子开始使用bugku的awd平台进行练习，靶机ip还得自己扫，于是为了方便就写了个简单的ping+文件写入脚本，用于扫好ip放入特定txt便于后续脚本读取。

前几天想着有点慢，于是加了个多线程。

最近又正好学了异步，抱着大过年的来都来了还只是个孩子心态就尝试用异步来ping。

没想到一写，才发现之前写的Asyncio笔记有很多错误，果然还是得实践出真知。

贴个笔记链接：异步Asyncio学习笔记

脚本

ping3.ping的timeout参数类型说明是int，但我换成小数也没问题。

太小容易误判，太大又过慢，想要最佳化，还是得把握好timeout时间。

多线程

from ping3 import ping, verbose_ping
import threading
import traceback
import queue
import time

class Attack(threading.Thread):
    def __init__(self, queue):
        threading.Thread.__init__(self)
        self._queue = queue

    def run(self):
        while not self._queue.empty():
            try:
                target = self._queue.get()
                self.detect(target)
            except Exception:
                traceback.print_exc()

    def detect(self, target):
        result = ping(target, timeout=timeout)
        if result is not None and result != False:
            alive_target.append(target)


def create():
    target_exp = '192.168.123.{}'
    self_port = 999
    ips = []
    for i in range(1, 255):
        if i == self_port:
            continue
        ips.append(target_exp.format(i))
    return ips


def main(ips):
    clearfile(bugku_ips_path)
    if len(ips) < 5:
        threadsnum = len(ips)
    else:
        threadsnum = 20  # 线程默认值
    LocalQueue = queue.Queue()
    for ip in ips:
        LocalQueue.put(ip)
    threads = []
    threadsCount = int(threadsnum)
    for i in range(threadsCount):
        threads.append(Attack(LocalQueue))
    for j in threads:
        j.start()
    for j in threads:
        j.join()


if __name__ == '__main__':
    start_time = time.time()
    timeout = 1
    alive_target = []
    main(create())
    print(alive_target)
    print(time.time()-start_time)

协程

因为ping不支持异步，所以用线程池形式，以空间换时间。

这个其实和多线程没啥区别，无非一个用手动设置线程，一个自动增加线程。

不过按理来说还是这个要更好一点，无论是时间还是空间上来说，更为”自动化“，代码更少，而且Python因为有GIL锁，用协程更为适合。

from ping3 import ping, verbose_ping
import asyncio
import time

async def _ping(target):
    loop = asyncio.get_event_loop()
    future = loop.run_in_executor(None, detect, target, timeout)
    await future


def detect(target, timeout):
    result = ping(target, timeout=timeout)
    if result is not None and result != False:
        alive_target.append(target)


def create():
    ips = []
    target_exp = '192.168.123.{}'
    self_port = 999
    for i in range(1, 255):
        if i == self_port:
            continue
        ips.append(target_exp.format(i))
    return ips


if __name__ == '__main__':
    start_time = time.time()
    timeout = 1
    alive_target = []
    target_list = create()
    tasks = [_ping(i) for i in target_list]
    asyncio.run(asyncio.wait(tasks))
    print(alive_target)
    print(time.time()-start_time)

协程uvloop

姑且测试了一下，暂时没感觉出太大区别，可能测试样本太少了。

python模块uvloop暂不支持windows和python3.9

from ping3 import ping, verbose_ping
import asyncio
import time
import uvloop

asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())


async def _ping(target):
    loop = asyncio.get_event_loop()
    future = loop.run_in_executor(None, detect, target, timeout)
    await future


def detect(target, timeout):
    result = ping(target, timeout=timeout)
    if result is not None and result != False:
        alive_target.append(target)


def create():
    ips = []
    target_exp = '192.168.123.{}'
    self_port = 999
    for i in range(1, 255):
        if i == self_port:
            continue
        ips.append(target_exp.format(i))
    return ips


if __name__ == '__main__':
    start_time = time.time()
    timeout = 1
    alive_target = []
    target_list = create()
    tasks = [_ping(i) for i in target_list]
    asyncio.run(asyncio.wait(tasks))
    print(alive_target)
    print(time.time()-start_time)

结语

要想更快，还是得从协议入手，换syn扫描之类的。可以，但没必要。再怎么写肯定也没有nmap好用。