构建你比较满意的分布式扫描平台（上）

0x00、前言

在云安全内部安全能力建设中，对云资产的端口扫描是一个必须要做的事情，因为开放一个端口对外提供一个服务都是扩大了您在云上攻击面。对于这种危险需要尽早的通知云上用户。那么如何对几万甚至几十万云主机做有效的端口扫描和精确的服务识别？这需要一套分布式的扫描系统来支撑。

0x01、实践的认知

在此之前，做了一些的小实践

方向3的代码如下：

# coding=utf-8 #!/usr/env/bin python //存储到redis def store(result): r=redis.Redis(host='127.0.0.1',port=6379,decode_responses=True,password=xxxx) with open(result,'r') as f: for line in f: if line.startswith('{ '}: try: temp = json.loads(line[:-2]) tmp1=temp["ports"][0] r.append(temp["ip"],str(tmp1["port"])+",") except: continue return r //masscan扫描模块 def Scan(): try: global g_queue while not g_queue.empty(): item = g_queue.get() result = "result"+item+".json" p = subprocess.Popen("/root/masscan/bin/masscan "+item+" -p T:21-23,25,80,81,88,110,143,443,1080,1433,1521,1158,3306-3308,3389,3690,5432,5900,6379,7001,8000,8080,9000,9418,27017-27019,50060,111,11211,2049 -oJ "+result, shell=True) p.wait() if p.returncode==0: print ('ok') if os.path.getsize(result) != 0: print item store(result) if g_queue.qsize() == 0: print (u'公有云高危端口扫描结束') return "ok" except Exception,e: print e return e if __name__ == '__main__': // ip地址压入队列 csvfile2 = file('xxx_public_ip.csv', 'r') reader = csv.reader(csvfile2) for x in reader: ips = IP(x[0]) for y in ips: g_queue.put(y.strNormal(0))

Nmap扫描模块

def NmapScan(): try: global g_queue while not g_queue.empty(): item = g_queue.get() filename = item.split(' ')[1]+"_"+item.split(' ')[0] result = "result"+filename.strip()+".xml" print result p = subprocess.Popen("/usr/bin/nmap -oX "+result+" -sV -p"+item, shell=True) p.wait() if p.returncode==0: nmap_report = NmapParser.parse_fromfile(result) for scanned_hosts in nmap_report.hosts: print scanned_hosts.address for serv in scanned_hosts.services: if serv.state == "open": m = serv.service_dict.get('extrainfo', '') print m if m.find('\'') != -1: pass else: writer.writerow([scanned_hosts.address,str(serv.port),serv.service,serv.service_dict.get('product', ''),serv.service_dict.get('version', ''),serv.service_dict.get('extrainfo', '')]) print "size = ", g_queue.qsize() if g_queue.qsize() == 0: print (u'公网服务指纹扫描结束') return "ok" except Exception,e: print e return e

//从队列中读取扫描目标

pool = redis.ConnectionPool(host='127.0.0.1',port=6379,decode_responses=True,password=xxxx,db=0) r = redis.Redis(connection_pool=pool) pipe = r.pipeline() pipe_size = 100000 len = 0 key_list = [] print r.pipeline() keys = r.keys() for key in keys: key_list.append(key) pipe.get(key) if len < pipe_size: len += 1 else: for (k, v) in zip(key_list, pipe.execute()): print k, v len = 0 key_list = [] for (k, v) in zip(key_list, pipe.execute()): for sport in v.split(','): if sport!='': item = sport+" "+k print item g_queue.put(item)

还是出现阻塞的情况。

不得不要从架构上考虑一下了。

0x02 分布式扫描架构探讨

思维是循序渐进的，否则你在认知的路上永远是抄袭别人，但是不知道为什么这么做。其实核心需要解决的是阻塞的事情，还有通过分布式扩展计算和带宽的能力以满足大规模扫描的安全能力。

思考1：阻塞问题那通过异步IO处理

1、web控制层

·使用Nginx搭建，解析VUE生成的html文件。做界面展示

·使用uWSGI搭建web中间件服务器，解析Django开发的python程序，做OpenAPI层。

·数据库使用postgresql。存储扫描配置扫描任务等信息

2、Web数据展示层

·底层扫描数据直接通过logstash进入Elastic Search中，所有的报表都都通过API访问ES，这样方便做变化对比。

3、任务调度层

· 消息队列使用自建的Redis做。

·异步IO使用tornado做。

4、Worker执行者

·控制参数通过redis获取

·扫描插件使用python编写，那么对于端口扫描其实就两个一个masscan和nmapscan插件。

·扫描结果通过logstash传入到Elastic Search

5、数据层

打算用kafka实现，这个要看会不会有突然大量数据涌入到ES中。

思考2：有没有成熟的组件完成上述事情呢？

最近在朋友圈传的百度陆奇《如何成为一名优秀的工程师》提到了上述观点，然后我豁然开朗，异步IO处理，在Django生态中有celery呀。它其实就是把redis+tornado的事做了。同时也借鉴了github同行的Demo程序，确实异步任务处理起来比较麻烦，为什么不用成熟的组件呢？

最终的技术架构图：

当然要考虑架构ROI投入产出比，后期会加入web漏洞扫描、web指纹识别、暴力破解、子域名暴力破解、目录文件列举、Web爬虫、SQLi、XSS这些耗时长的任务。

0x03、总结

本篇主要讨论了架构设计方便的问题，下篇会着重在系统搭建、代码实现等方面说明分布式扫描系统的那些事。

原文发布时间为：2018-05-23

本文来自云栖社区合作伙伴“嘶吼网”，了解相关信息可以关注“嘶吼网”。