内网主机资产扫描那些事
本项目仅进行内网主机资产整理，无漏洞利用、攻击性行为，请使用者遵守当地相关法律，勿用于非授权测试，勿用于未授权扫描，如作他用所承受的法律责任一概与作者无关，下载使用即代表使用者同意上述观点。
需求分析
实现一个自动化的内网资产扫描器，那么首先需要实现如下功能：
- 开放端口扫描
- 运行服务检测
- 主机部署网站探测
- 自动生成报表
- 报表结果可视化
如上需求分析，需要一个快速的端口扫描工具实现资产扫描，需要前端一些可视化的js库完成数据集可视化。
端口扫描有如下选择：
Nmap
优点：模块丰富，功能齐全，支持多种不同方式扫描端口，拥有完备的主机服务漏洞扫描插件
缺点：量级较重，迁移麻烦，速度较慢，不够灵活，插件适配性接口不方便维护Masscan
优点：速度快，速度快，速度快
缺点：存在误报率，linux下编译masscan对新手不友好自己写
优点：灵活，轻便，自定义功能
缺点：重复造轮子，速度和准确率相对较低，意味着不能吃到nmap提供的漏洞扫描红利可视化库有如下选择：
Seaborn
优点：基于Matplotlib的一个库，图形界面多，操作简便
缺点：图形适合科学数据研究，在前端显示内网资产个人觉得不是很完美Matplotlib
优点：python可视化库的鼻祖，超级强大，等级上如同上面的nmap
缺点：上手难度较大，打包成exe体积更大
Pandas
优点：对数据分析友好，上手难度不大
缺点：个人觉得不是很好看，不能算缺点pyecharts+bootstrap设计布局
优点：极易上手，简单快捷使用
缺点：对不擅长js的同学来说，想要改动图形难度很大
考虑到时间成本，代码体积等等因素，最终敲定的结果是masscan+pyecharts+bootstrap
手脚架搭建
环境准备
敲定使用masscan+pyecharts后，需要安装masscan，Windows用户可以在GitHub下载成品，Linux用户使用命令：
apt-get install masscan
或者下载源码手动编译。
相关库准备
需要准备如下库进行轮子改造工程：
python-masscan:masscan对接python的一个库
requests:网络请求
socket:端口连接，获取banner
re:正则匹配数据
random:随机字符串
time:时间模块
urllib:主要对url清洗主API设计
按照功能，将端口扫描独立成一个大类，数据清洗独立成一个函数，其他代码的做一些传承工作即可。
扫描流程设计
按照传入的参数，进行端口服务检测，数据清洗，生成结果报表。
输出结果设计
输出的结果需要做到数据的可视化，使用饼图展示。以及每台主机的详细资产数据。
粗略的输出结果如下图所示：
代码编写
如上整体的扫描策略，扫描流程，数据结构，现在可以开始设计单个的功能函数。
端口扫描
接口具体设计如下
输入接口：
网段 192.168.0.0/24
独立IP 192.168.1.1
调用方法：
ip = '192.168.11.0/24'
a = IpInfoScan(ip)
res = a.GetResult()
print(res)
or:
ip = '192.168.11.5'
a = IpInfoScan(ip)
res = a.GetResult()
print(res)
输出结果：
{
ip:192.168.0.1,
alive:True,
ports:[22,80,8888,3306]
server:['ssh','http','https','mysql'],
services:{22:ssh,80:http.....},
urls:{'http://192.168.0.1:80':后台管理系统}
time:2019-11-20-13:15
}
代码如下：
# -*- coding:utf-8 -*-
import re
from urllib.parse import urlparse
import masscan
import requests
import socket
import datetime
filenames = '-'.join(str(datetime.datetime.now()).replace(' ','-').replace(':','-').split('.')[0].split('-'))+'.txt'
from concurrent.futures import ThreadPoolExecutor
requests.packages.urllib3.disable_warnings()
Alive_Status = [200,301,302,400,404]
def get_title(r):
# 该函数用来获取网页的标题
title = '获取失败'
try:
title_pattern = b'<title>(.*?)</title>'
title = re.search(title_pattern, r, re.S | re.I).group(1)
try:
title = title.decode().replace('
', '').strip()
return title
except:
try:
title = title.decode('gbk').replace('
', '').strip()
return title
except:
return title
except:
return title
finally:
return title
def Requests(url):
# 该函数用来发起网络请求
headers = {'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/71.0.3578.98 Safari/537.36'}
url1 = 'http://'+url
url2 = 'https://'+url
title = '获取失败'
title1 = '获取失败'
title2 = '获取失败'
content1 = None
content2 = None
try:
r = requests.get(url='http://'+url,headers=headers,verify=False,timeout=20)
if b'text/html' in r.content or b'<title>' in r.content or b'</html>' in r.content:
content1 = r.content
if r.status_code in Alive_Status:
u = urlparse(str(r.url))
title1 = get_title(r.content)
url1 = u.scheme + '://' + u.netloc
except Exception as e:
pass
try:
r = requests.get(url='https://'+url,headers=headers,verify=False,timeout=20)
if b'text/html' in r.content or b'<title>' in r.content or b'</html>' in r.content:
content2 = r.content
if r.status_code in Alive_Status:
u = urlparse(str(r.url))
title2 = get_title(r.content)
url2 = u.scheme + '://' + u.netloc
except Exception as e:
pass
if title1 != '获取失败':
return {url1: title1}
if title2 != '获取失败':
return {url2: title2}
if content1 != None:
return {url1:title}
if content2 != None:
return {url2:title}
def Get_Alive_Url(urls):
'''
如果想要获取 IP 段内存活web服务
hosts = IPy.IP('192.168.1.0/24')
urls = []
for host in hosts:
urls.append('http://{}:{}'.format(host,80))
urls.append('https://{}:{}'.format(host,443))
Get_Alive_Url(urls)
返回结果是一个列表，列表内数据为字典 多个自带你 {网址：标题}
'''
with ThreadPoolExecutor(max_workers=8) as p:
future_tasks = [p.submit(Requests, i) for i in urls]
result = [obj.result() for obj in future_tasks if obj.result() is not None]
return result
from tinydb import TinyDB, where
from tinydb.storages import JSONStorage
from tinydb.middlewares import CachingMiddleware
from collections import namedtuple
import os
Port = namedtuple("Port", ["name", "port", "protocol", "description"])
__BASE_PATH__ = os.path.dirname(os.path.abspath(__file__))
__DATABASE_PATH__ = os.path.join(__BASE_PATH__, 'ports.json')
__DB__ = TinyDB(__DATABASE_PATH__, storage=CachingMiddleware(JSONStorage))
def GetPortInfo(port, like=False):
"""
判断端口服务，传入参数为 字符串类型的数字
返回服务名称 'http'，没有则返回 '检测失效'
"""
where_field = "port" if port.isdigit() else "name"
if like:
ports = __DB__.search(where(where_field).search(port))
else:
ports = __DB__.search(where(where_field) == port)
try:
return ports[0]['name'] # flake8: noqa (F812)
except:
return '识别端口异常'
class IpInfoScan:
def __init__(self,ip):
self.ip = ip
# 传入的数据是网段哦 192.168.0.0/24
self.Banner = {b'http': [b'^HTTP/.*
Server: Apache/2',b'HTTP/'], b'ssh': [b'^SSH-.*openssh'], b'netbios': [b'ÂÂ'], b'backdoor-fxsvc': [b'^500 Not Loged in'], b'backdoor-shell': [b'^sh[$#]'], b'bachdoor-shell': [b'[a-z]*sh: .* command not found'], b'backdoor-cmdshell': [b'^Microsoft Windows .* Copyright .*>'], b'db2': [b'.*SQLDB2RA'], b'db2jds': [b'^N'], b'dell-openmanage': [b'^N
'], b'finger': [b'finger: GET: '], b'ftp': [b'^220 .* UserGate'], b'http-iis': [b'^<h1>Bad Request .Invalid URL.</h1>'], b'http-jserv': [b'^HTTP/.*Cookie.*JServSessionId'], b'http-tomcat': [b'.*Servlet-Engine'], b'http-weblogic': [b'^HTTP/.*Cookie.*WebLogicSession'], b'http-vnc': [b'^HTTP/.*RealVNC/'], b'ldap': [b'^0E'], b'smb': [b'^.ÿSMBr.*'], b'msrdp': [b'^Ã4'], b'msrdp-proxy': [b'^nmproxy: Procotol byte is not 8
$'], b'msrpc': [b'
....$'], b'mssql': [b';MSSQLSERVER;'], b'telnet': [b'^ÿþ'], b'mysql': [b"whost '"], b'mysql-blocked': [b'^\('], b'mysql-secured': [b'this MySQL'], b'mongodb': [b'^.*version.....([\.\d]+)'], b'nagiosd': [b'Sorry, you \(.*are not among the allowed hosts...'], b'nessus': [b'< NTP 1.2 >
User:'], b'oracle-tns-listener': [b'\(ADDRESS=\(PROTOCOL='], b'oracle-dbsnmp': [b'^'], b'oracle-https': [b'^220- ora'], b'oracle-rmi': [b'^N '], b'postgres': [b'^EFATAL'], b'rlogin': [b'^Permission denied.
'], b'rpc-nfs': [b'^'], b'rpc': [b'^Â'], b'rsync': [b'^@RSYNCD:.*'], b'smux': [b'^A'], b'snmp-public': [b'public¢'], b'snmp': [b'A'], b'socks': [b'^[-]'], b'ssl': [b'^.....'], b'sybase': [b'^'], b'tftp': [b'^[]'], b'uucp': [b'^login: password: '], b'vnc': [b'^RFB.*'], b'webmin': [b'^0\.0\.0\.0:.*:[0-9]'], b'websphere-javaw': [b'^
']}
def GetOpenPort(self):
HostInfos = {}
try:
mas = masscan.PortScanner()
mas.scan(self.ip,ports='21,22,23,25,80,81,88,8080,8888,999,9999,7000,1433,1521,3306,3389,6379,7001,27017,27018')
# 这里简单的扫一下普通端口即可
Results = mas.scan_result['scan']
AliveHosts = list(Results.keys())
if AliveHosts != []:
for k, v in Results.items():
HostInfos[str(k)] = list(v['tcp'].keys())
return HostInfos
except Exception as e:
pass
return HostInfos
def GetOneIPorts(self,ip):
try:
mas = masscan.PortScanner()
mas.scan(ip)
OpenPorts = mas.scan_result['scan'][ip]['tcp'].keys()
except:
return None
return {ip:OpenPorts}
def GetBannerServer(self,ip,port):
try:
s = socket.socket()
s.settimeout(0.5)
s.connect((ip,int(port)))
s.send(b'langzi
')
SocketRecv = (s.recv(1024))
for k,v in self.Banner.items():
for b in v:
banner = re.search(b,SocketRecv,re.I)
if banner:
return k.decode()
return '获取失败'
except Exception as e:
return '获取失败'
finally:
s.close()
def GetPoerInfos(self,ip,lis):
# 传入参数为 开放的端口列表 [80,8888,3389]
PortInfos = {}
for li in lis:
server = self.GetBannerServer(ip,li)
if server == '获取失败':
server = self.GetBannerServer(ip, li)
PortInfos[str(li)] = server
if PortInfos != {}:
for k,v in PortInfos.items():
if v == '获取失败':
PortInfos[k] = GetPortInfo(str(k))
return PortInfos
def GetResult(self):
results = []
print('[{}] 端口扫描 : {}'.format(str(datetime.datetime.now()).split('.')[0], self.ip))
if '-' in self.ip or '/' in self.ip:
openports = self.GetOpenPort()
else:
openports = self.GetOneIPorts(self.ip)
#openports = [80,3389]
if openports != {} and openports != None:
for k,v in openports.items():
retuls = {}
print('[{}] 主机 {} 开放端口 {}个'.format(str(datetime.datetime.now()).split('.')[0], k,len(v)))
res = self.GetPoerInfos(k,v)
# {'80': 'http', '3389': 'ms-wbt-server'}
urls = []
for port in v:
urls.append('{}:{}'.format(k, port))
AliveUrls = Get_Alive_Url(urls)
retuls['ip']=k
retuls['alive']=True
retuls['ports']=list(res.keys())
retuls['server']=list(res.values())
retuls['services']=res
retuls['urls']=AliveUrls
retuls['time']=str(datetime.datetime.now()).replace(' ','-').replace(':','-').split('.')[0]
results.append(retuls)
return results
if __name__ == '__main__':
# 该类的调用方法如下
ip = '192.168.5.0/24'
a = IpInfoScan(ip)
res = a.GetResult()
print(res)数据可视化
使用pyecharts的饼图做主题演示，演示代码如下：
def run():
inner_x_data = ["存活主机", "开放端口", "部署网站"]
inner_y_data = [60,120,53]
inner_data_pair = [list(z) for z in zip(inner_x_data, inner_y_data)]
mid_x_data = ["22", "80", "3306", "21", "3389", "1521", "6379"]
mid_y_data = [335, 310, 234, 135, 1048, 251, 147]
mid_data_pair = [list(z) for z in zip(mid_x_data, mid_y_data)]
outer_x_data = ["ssh", "http", "mysql", "ftp", "msc", "oralce", "redis"]
outer_y_data = [335, 310, 234, 135, 1048, 251, 147]
outer_data_pair = [list(z) for z in zip(outer_x_data, outer_y_data)]
c=(
Pie(init_opts=opts.InitOpts(width="1600px", height="800px"))
.add(
series_name="总体资产",
data_pair=inner_data_pair,
radius=[0, "20%"],
label_opts=opts.LabelOpts(position="inner",formatter="{b}:{c}个"),
)
.add(
series_name="开放端口",
data_pair=mid_data_pair,
radius=["25%", "50%"],
label_opts=opts.LabelOpts(position="inner",formatter="端口:{b}|总数:{c}"),
)
.add(
series_name="部署服务",
radius=["55%", "80%"],
data_pair=outer_data_pair,
label_opts=opts.LabelOpts(formatter="{a}:{b}|占比:{d}%"),
)
.set_global_opts(legend_opts=opts.LegendOpts(pos_left="mid", orient="vertical"))
.set_series_opts(
tooltip_opts=opts.TooltipOpts(
trigger="item", formatter="{a} <br/>{b}: {c} ({d}%)"
)
)
.render("test.html")
)
run()生成的案例图如下：
数据清洗
如上，核心功能已经完成，接下来需要将扫描的结果进行可视化处理，最后将两者的结果在前端显示出来，值得注意的是，为了自己更加方便的获取数据结果，数据清洗需要有如下需求：
- 结构图简洁明了获取主体资产部署情况
- 结构图显示主机存活数量，端口开放总数，运行服务，部署多少网站
- 点击相关标签就能获取该端口开放的全部主机IP
- 每台主机资产的详情需要展示，应该包含主机部署网站，开放端口，运行服务。
- 提供日志功能。
代码如下：
def CleanData(IPdata,txtfile,htmlfile,Portfolio):
Btn_Class = ['btn btn-danger', 'btn btn-warning', 'btn btn-info', 'btn btn-primary', 'btn btn-default',
'btn btn-success']
AllResultFiles = set()
AllResultFiles.add('AliveHosts')
AllResultFiles.add('AliveUrls')
for i in IPdata:
service = (i.get('services'))
for k, v in service.items():
AllResultFiles.add(k)
AllResultFiles.add(v)
ImgData = WriteImgTxt(IPdata,txtfile)
with open('../'+htmlfile,'a+',encoding='utf-8')as a:
a.write('''
<html lang="en">
<head>
<meta charset="UTF-8">
<title>网络资产拓扑-LangNetworkTopology3</title>
<link rel="stylesheet" href='{0}/static/bootstrap-theme.min.css'>
<link rel="stylesheet" href='{0}/static/bootstrap.min.css'>
<script type="text/javascript" src="{0}/static/echarts.min.js"></script>
<script type="text/javascript" src="{0}/static/echarts-wordcloud.min.js"></script>
</head>
<h1> 主机资产分布图</h1><hr/>
'''.format(os.path.abspath('')))
a.write(ImgData)
a.write('''
<hr />
<h1> 主机资产整理表</h1><hr />
<div class='btn-toolbar'>
''')
for file in AllResultFiles:
a.write('''<a href="{}.txt" target="_blank"><button class="{}">{}</button></a>'''.format(os.path.join(os.path.abspath(''),Portfolio,file).replace('/','\'),random.choice(Btn_Class),file))
a.write('''
</div>
<hr />
<h1> 主机资产详情表</h1><hr />
<div class="col-md-3">
''')
portips = {}
serips = {}
for i in IPdata:
ports = i.get('ports')
servs = i.get('server')
for port in ports:
portips[port] = []
for serv in servs:
serips[serv] = []
for i in IPdata:
ports = i.get('ports')
servs = i.get('server')
for port in ports:
portips[port].append(i.get('ip'))
for serv in servs:
serips[serv].append(i.get('ip'))
for k, v in serips.items():
a.write('''
<div class="panel panel-default">
<div class="panel-body">
服务:{} 运行主机
</div></div>
'''.format(k))
for vv in v:
a.write('''
<div class="panel-footer">
{}
</div>
'''.format(vv))
with open(os.path.join(Portfolio, k) + '.txt', 'a+', encoding='utf-8')as b:
b.write(vv+'
')
a.write('<hr>')
a.write('</div><div class="col-md-6">')
for k in IPdata:
ip = k.get('ip')
ports = '|'.join(k.get('ports'))
service = '|'.join(k.get('server'))
weburls = k.get('urls')
if weburls == []:
weburl = '无部署网站'
else:
web = []
for i in weburls:
for k1,v in i.items():
weburl = '<a href="{}" target="_blank">{}</a>'.format(k1,v)
web.append(weburl)
weburl = '<br>'.join(web)
timen = k.get('time')
a.write('''
<div class="panel panel-primary">
<div class="panel-heading">
主机:{}资产详情
</div>
<div class="panel-body">
<table class="table">
<tr><td>当前主机</td><td>{}</td></tr>
<tr><td>开放端口</td><td>{}</td></tr>
<tr><td>运行服务</td><td>{}</td></tr>
<tr><td>部署网站</td><td>{}</td></tr>
<tr><td>发现时间</td><td>{}</td></tr>
</table>
</div>
</div>
'''.format(ip,ip,ports,service,weburl,timen))
a.write('''
</div>
</div>
</div><div class="col-md-3">''')
for k,v in portips.items():
a.write('''
<div class="panel panel-default">
<div class="panel-body">
端口:{} 开放主机
</div></div>
'''.format(k))
for vv in v:
a.write('''
<div class="panel-footer">
{}
</div>
'''.format(vv))
with open(os.path.join(Portfolio, k) + '.txt', 'a+', encoding='utf-8')as b:
b.write(vv+'
')
a.write('<hr>')
a.write('</div></body></html>')传承代码
到这里，主体的功能函数已经完成，接下来就是将造好的小部件搭载轮子上就完成了。不过需要考虑许多因素，比如提供接口让局域网管理员可以设置扫描的端口，扫描的进程数，扫描每秒的发包量等等。这些内容可以提供一个输入点，让管理员输入设置即可。
具体代码如下：
# -*- coding:utf-8 -*-
from __future__ import unicode_literals
import string
import sys
import time
import pyecharts.options as opts
from pyecharts.charts import Pie
import re
from urllib.parse import urlparse
import masscan
import requests
import socket
import datetime
import os
import random
from concurrent.futures import ThreadPoolExecutor
requests.packages.urllib3.disable_warnings()
# from pyecharts.charts import Page, WordCloud
Portfolio = 'CleanData/'+'-'.join(str(datetime.datetime.now()).replace(' ','-').replace(':','-').split('.')[0].split('-'))
os.makedirs(Portfolio)
ImgTxt = '-'.join(str(datetime.datetime.now()).replace(' ','-').replace(':','-').split('.')[0].split('-'))+'.txt'
ImgHtml = '-'.join(str(datetime.datetime.now()).replace(' ','-').replace(':','-').split('.')[0].split('-'))+'.html'
def Log(x):
with open('../LangNetWorkTopoLog.txt','a+',encoding='utf-8')as a:
a.write(str( '-'.join(str(datetime.datetime.now()).replace(' ','-').replace(':','-').split('.')[0].split('-')))+' '+x+'
')
Alive_Status = [200,204,206,301,302,304,401,402,403,404,500,501,502,503]
def get_title(r):
title = '获取失败'
try:
title_pattern = b'<title>(.*?)</title>'
title = re.search(title_pattern, r, re.S | re.I).group(1)
try:
title = title.decode().replace('
', '').strip()
return title
except:
try:
title = title.decode('gbk').replace('
', '').strip()
return title
except:
return title
except:
return title
finally:
return title
def Requests(url):
headers = {'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/71.0.3578.98 Safari/537.36'}
url1 = 'http://'+url
url2 = 'https://'+url
title = '获取失败'
title1 = '获取失败'
title2 = '获取失败'
content1 = None
content2 = None
try:
r = requests.get(url='http://'+url,headers=headers,verify=False,timeout=20)
if b'text/html' in r.content or b'<title>' in r.content or b'</html>' in r.content:
content1 = r.content
if r.status_code in Alive_Status:
u = urlparse(str(r.url))
title1 = get_title(r.content)
url1 = u.scheme + '://' + u.netloc
except Exception as e:
pass
try:
r = requests.get(url='https://'+url,headers=headers,verify=False,timeout=20)
if b'text/html' in r.content or b'<title>' in r.content or b'</html>' in r.content:
content2 = r.content
if r.status_code in Alive_Status:
u = urlparse(str(r.url))
title2 = get_title(r.content)
url2 = u.scheme + '://' + u.netloc
except Exception as e:
pass
if title1 != '获取失败':
return {url1: title1}
if title2 != '获取失败':
return {url2: title2}
if content1 != None:
return {url1:title}
if content2 != None:
return {url2:title}
def Get_Alive_Url(urls):
'''
如果想要获取 IP 段内存活web服务
hosts = IPy.IP('192.168.1.0/24')
urls = []
for host in hosts:
urls.append('http://{}:{}'.format(host,80))
urls.append('https://{}:{}'.format(host,443))
Get_Alive_Url(urls)
返回结果是一个列表，列表内数据为字典 多个自带你 {网址：标题}
'''
with ThreadPoolExecutor(max_workers=8) as p:
future_tasks = [p.submit(Requests, i) for i in urls]
result = [obj.result() for obj in future_tasks if obj.result() is not None]
return result
from tinydb import TinyDB, where
from tinydb.storages import JSONStorage
from tinydb.middlewares import CachingMiddleware
from collections import namedtuple
import os
Port = namedtuple("Port", ["name", "port", "protocol", "description"])
__BASE_PATH__ = os.path.dirname(os.path.abspath(__file__))
__DATABASE_PATH__ = os.path.join(__BASE_PATH__, 'ports.json')
__DB__ = TinyDB(__DATABASE_PATH__, storage=CachingMiddleware(JSONStorage))
def GetPortInfo(port, like=False):
"""
判断端口服务，传入参数为 字符串类型的数字
返回服务名称 'http'，没有则返回 '检测失效'
"""
where_field = "port" if port.isdigit() else "name"
if like:
ports = __DB__.search(where(where_field).search(port))
else:
ports = __DB__.search(where(where_field) == port)
try:
return ports[0]['name'] # flake8: noqa (F812)
except:
return '识别端口异常'
class IpInfoScan:
def __init__(self,ip):
self.ip = ip
# 传入的数据是网段哦 192.168.0.0/24
self.Banner = {b'http': [b'^HTTP/.*
Server: Apache/2',b'HTTP/'], b'ssh': [b'^SSH-.*openssh'], b'netbios': [b'ÂÂ'], b'backdoor-fxsvc': [b'^500 Not Loged in'], b'backdoor-shell': [b'^sh[$#]'], b'bachdoor-shell': [b'[a-z]*sh: .* command not found'], b'backdoor-cmdshell': [b'^Microsoft Windows .* Copyright .*>'], b'db2': [b'.*SQLDB2RA'], b'db2jds': [b'^N'], b'dell-openmanage': [b'^N
'], b'finger': [b'finger: GET: '], b'ftp': [b'^220 .* UserGate'], b'http-iis': [b'^<h1>Bad Request .Invalid URL.</h1>'], b'http-jserv': [b'^HTTP/.*Cookie.*JServSessionId'], b'http-tomcat': [b'.*Servlet-Engine'], b'http-weblogic': [b'^HTTP/.*Cookie.*WebLogicSession'], b'http-vnc': [b'^HTTP/.*RealVNC/'], b'ldap': [b'^0E'], b'smb': [b'^.ÿSMBr.*'], b'msrdp': [b'^Ã4'], b'msrdp-proxy': [b'^nmproxy: Procotol byte is not 8
$'], b'msrpc': [b'
....$'], b'mssql': [b';MSSQLSERVER;'], b'telnet': [b'^ÿþ'], b'mysql': [b"whost '"], b'mysql-blocked': [b'^\('], b'mysql-secured': [b'this MySQL'], b'mongodb': [b'^.*version.....([\.\d]+)'], b'nagiosd': [b'Sorry, you \(.*are not among the allowed hosts...'], b'nessus': [b'< NTP 1.2 >
User:'], b'oracle-tns-listener': [b'\(ADDRESS=\(PROTOCOL='], b'oracle-dbsnmp': [b'^'], b'oracle-https': [b'^220- ora'], b'oracle-rmi': [b'^N '], b'postgres': [b'^EFATAL'], b'rlogin': [b'^Permission denied.
'], b'rpc-nfs': [b'^'], b'rpc': [b'^Â'], b'rsync': [b'^@RSYNCD:.*'], b'smux': [b'^A'], b'snmp-public': [b'public¢'], b'snmp': [b'A'], b'socks': [b'^[-]'], b'ssl': [b'^.....'], b'sybase': [b'^'], b'tftp': [b'^[]'], b'uucp': [b'^login: password: '], b'vnc': [b'^RFB.*'], b'webmin': [b'^0\.0\.0\.0:.*:[0-9]'], b'websphere-javaw': [b'^
']}
def GetOpenPort(self,inport,rate):
HostInfos = {}
try:
mas = masscan.PortScanner()
#mas.scan(self.ip,ports='21,22,23,25,80,81,88,8080,8888,999,9999,7000,1433,1521,3306,3389,6379,7001,27017,27018')
# 这里简单的扫一下普通端口即可
mas.scan(self.ip, ports=inport, arguments='--rate {}'.format(rate))
# if inport == '0':
# mas.scan(self.ip,arguments='--rate {}'.format(rate))
# else:
# mas.scan(self.ip,ports=inport,arguments='--rate {}'.format(rate))
Results = mas.scan_result['scan']
AliveHosts = list(Results.keys())
if AliveHosts != []:
for k, v in Results.items():
HostInfos[str(k)] = list(v['tcp'].keys())
return HostInfos
except Exception as e:
Log('扫描IP出现异常:{}'.format(str(e)))
pass
return HostInfos
def GetOneIPorts(self,ip,inport,rate):
try:
mas = masscan.PortScanner()
mas.scan(self.ip, ports=inport, arguments='--rate {}'.format(rate))
# if inport == '0':
# mas.scan(self.ip,arguments='--rate {}'.format(rate))
# else:
# mas.scan(self.ip,ports=inport,arguments='--rate {}'.format(rate))
OpenPorts = mas.scan_result['scan'][ip]['tcp'].keys()
except Exception as e:
Log('获取扫描IP端口结果异常:{}'.format(str(e)))
return None
return {ip:OpenPorts}
def GetBannerServer(self,ip,port):
try:
s = socket.socket()
s.settimeout(0.5)
s.connect((ip,int(port)))
s.send(b'langzi
')
SocketRecv = (s.recv(1024))
for k,v in self.Banner.items():
for b in v:
banner = re.search(b,SocketRecv,re.I)
if banner:
return k.decode()
return '获取失败'
except Exception as e:
# Log('向端口发起连接异常:{}'.format(str(e)))
return '获取失败'
finally:
s.close()
def GetPoerInfos(self,ip,lis):
# 传入参数为 开放的端口列表 [80,8888,3389]
PortInfos = {}
for li in lis:
server = self.GetBannerServer(ip,li)
if server == '获取失败':
server = self.GetBannerServer(ip, li)
PortInfos[str(li)] = server
if PortInfos != {}:
for k,v in PortInfos.items():
if v == '获取失败':
PortInfos[k] = GetPortInfo(str(k))
return PortInfos
def GetResult(self,inport,rate):
results = []
print('[{}] 端口扫描 : {}'.format(str(datetime.datetime.now()).split('.')[0], self.ip))
Log('开始扫描IP:{}'.format(self.ip))
if '-' in self.ip or '/' in self.ip:
openports = self.GetOpenPort(inport,rate)
else:
openports = self.GetOneIPorts(self.ip,inport,rate)
#openports = [80,3389]
if openports != {} and openports != None:
for k,v in openports.items():
retuls = {}
print('[{}] 主机 {} 开放端口 {} 个'.format(str(datetime.datetime.now()).split('.')[0], k.ljust(15),len(v)))
Log('主机 {} 开放端口 {} '.format(k,str(v)))
with open(os.path.join(Portfolio, 'AliveHosts') + '.txt', 'a+', encoding='utf-8')as b:
b.write(k + '
')
res = self.GetPoerInfos(k,v)
# {'80': 'http', '3389': 'ms-wbt-server'}
urls = []
for port in v:
urls.append('{}:{}'.format(k, port))
AliveUrls = Get_Alive_Url(urls)
retuls['ip']=k
retuls['alive']=True
retuls['ports']=list(res.keys())
retuls['server']=list(res.values())
retuls['services']=res
retuls['urls']=AliveUrls
if AliveUrls != []:
for urls in AliveUrls:
for u,t in urls.items():
with open(os.path.join(Portfolio, 'AliveUrls') + '.txt', 'a+', encoding='utf-8')as b:
b.write(u + '
')
retuls['time']=str(datetime.datetime.now()).replace(' ','-').replace(':','-').split('.')[0]
results.append(retuls)
Log(str(results))
return results
def WriteImgTxt(IPdata,filename):
alivehosts = len(IPdata)
openports = 0
weburls = 0
portdict = {}
servicedict = {}
for i in IPdata:
weburls += (len(i.get('urls')))
openports += (len(i.get('ports')))
service = (i.get('services'))
for k, v in service.items():
portdict[k] = 0
servicedict[v.upper()] = 0
for i in IPdata:
service = (i.get('services'))
for k, v in service.items():
portdict[k] = portdict[k] + 1
servicedict[v.upper()] = servicedict[v.upper()] + 1
inner_x_data = ["存活主机", "开放端口", "部署网站"]
inner_y_data = [alivehosts, openports, weburls]
inner_data_pair = [list(z) for z in zip(inner_x_data, inner_y_data)]
mid_data_pair = list(portdict.items())
outer_data_pair = list(servicedict.items())
c=(
Pie(init_opts=opts.InitOpts(width="2200px", height="900px"))
.add(
series_name="总体资产",
data_pair=inner_data_pair,
radius=[0, "20%"],
label_opts=opts.LabelOpts(position="inner",formatter="{b}:{c}个"),
)
.add(
series_name="开放端口",
data_pair=mid_data_pair,
radius=["25%", "50%"],
label_opts=opts.LabelOpts(position="inner",formatter="端口:{b}|总数:{c}"),
)
.add(
series_name="部署服务",
radius=["55%", "80%"],
data_pair=outer_data_pair,
label_opts=opts.LabelOpts(formatter="{a}:{b}|占比:{d}%"),
)
.set_global_opts(legend_opts=opts.LegendOpts(pos_left="mid", orient="vertical"))
.set_series_opts(
tooltip_opts=opts.TooltipOpts(
trigger="item", formatter="{a} <br/>{b}: {c} ({d}%)"
)
)
.render(filename)
)
# words = mid_data_pair + outer_data_pair
# c = (
# WordCloud(init_opts=opts.InitOpts(width="1200px", height="800px"))
# .add("", words, word_size_range=[30, 80])
# # .set_global_opts(title_opts=opts.TitleOpts(title="WordCloud-基本示例"))
# ).render('test.txt')
if os.path.exists(filename):
with open(filename, 'r', encoding='utf-8')as a:
res1 = re.search('(<body>.*?</body>)', a.read(), re.S | re.I).group(1)
os.remove(filename)
return res1
# with open('test.txt', 'r', encoding='utf-8')as a:
# res2 = re.search('(<body>.*?</body>)', a.read(), re.S | re.I).group(1)
# os.remove('test.txt')
# res3 = '<div class="col-sm-6">{}</div><div class="col-sm-6">{}</div>'.format(res1,res2)
# return res3
else:
Log('生成效果图失败')
def CleanData(IPdata,txtfile,htmlfile):
Btn_Class = ['btn btn-danger', 'btn btn-warning', 'btn btn-info', 'btn btn-primary', 'btn btn-default',
'btn btn-success']
AllResultFiles = set()
AllResultFiles.add('AliveHosts')
AllResultFiles.add('AliveUrls')
for i in IPdata:
service = (i.get('services'))
for k, v in service.items():
AllResultFiles.add(k)
AllResultFiles.add(v)
ImgData = WriteImgTxt(IPdata,txtfile)
with open('../'+htmlfile,'a+',encoding='utf-8')as a:
a.write('''
<html lang="en">
<head>
<meta charset="UTF-8">
<title>网络资产拓扑-LangNetworkTopology3</title>
<link rel="stylesheet" href='{0}/static/bootstrap-theme.min.css'>
<link rel="stylesheet" href='{0}/static/bootstrap.min.css'>
<script type="text/javascript" src="{0}/static/echarts.min.js"></script>
<script type="text/javascript" src="{0}/static/echarts-wordcloud.min.js"></script>
</head>
<h1> 主机资产分布图</h1><hr/>
'''.format(os.path.abspath('')))
a.write(ImgData)
a.write('''
<hr />
<h1> 主机资产整理表</h1><hr />
<div class='btn-toolbar'>
''')
for file in AllResultFiles:
a.write('''<a href="{}.txt" target="_blank"><button class="{}">{}</button></a>'''.format(os.path.join(os.path.abspath(''),Portfolio,file).replace('/','\'),random.choice(Btn_Class),file))
a.write('''
</div>
<hr />
<h1> 主机资产详情表</h1><hr />
<div class="col-md-3">
''')
portips = {}
serips = {}
for i in IPdata:
ports = i.get('ports')
servs = i.get('server')
for port in ports:
portips[port] = []
for serv in servs:
serips[serv] = []
for i in IPdata:
ports = i.get('ports')
servs = i.get('server')
for port in ports:
portips[port].append(i.get('ip'))
for serv in servs:
serips[serv].append(i.get('ip'))
for k, v in serips.items():
a.write('''
<div class="panel panel-default">
<div class="panel-body">
服务:{} 运行主机
</div></div>
'''.format(k))
for vv in v:
a.write('''
<div class="panel-footer">
{}
</div>
'''.format(vv))
with open(os.path.join(Portfolio, k) + '.txt', 'a+', encoding='utf-8')as b:
b.write(vv+'
')
a.write('<hr>')
a.write('</div><div class="col-md-6">')
for k in IPdata:
ip = k.get('ip')
ports = '|'.join(k.get('ports'))
service = '|'.join(k.get('server'))
weburls = k.get('urls')
if weburls == []:
weburl = '无部署网站'
else:
web = []
for i in weburls:
for k1,v in i.items():
weburl = '<a href="{}" target="_blank">{}</a>'.format(k1,v)
web.append(weburl)
weburl = '<br>'.join(web)
timen = k.get('time')
a.write('''
<div class="panel panel-primary">
<div class="panel-heading">
主机:{}资产详情
</div>
<div class="panel-body">
<table class="table">
<tr><td>当前主机</td><td>{}</td></tr>
<tr><td>开放端口</td><td>{}</td></tr>
<tr><td>运行服务</td><td>{}</td></tr>
<tr><td>部署网站</td><td>{}</td></tr>
<tr><td>发现时间</td><td>{}</td></tr>
</table>
</div>
</div>
'''.format(ip,ip,ports,service,weburl,timen))
a.write('''
</div>
</div>
</div><div class="col-md-3">''')
for k,v in portips.items():
a.write('''
<div class="panel panel-default">
<div class="panel-body">
端口:{} 开放主机
</div></div>
'''.format(k))
for vv in v:
a.write('''
<div class="panel-footer">
{}
</div>
'''.format(vv))
with open(os.path.join(Portfolio, k) + '.txt', 'a+', encoding='utf-8')as b:
b.write(vv+'
')
a.write('<hr>')
a.write('</div></body></html>')
if __name__ == '__main__':
list_jindu = string.ascii_letters + string.digits + '.' + '_' + ' '+'['+']'+'*'
jindu = ' [*] LangNetworkTopology3 Console Start...'
jindud = ''
for xx in jindu:
for x in list_jindu:
sys.stdout.write(jindud + "
")
if xx == x:
jindud = jindud + x
sys.stdout.write(jindud + "
")
time.sleep(0.01)
break
else:
sys.stdout.write(jindud + x + "
")
time.sleep(0.01)
sys.stdout.flush()
sys.stdout.write(jindud + "
")
sys.stdout.write(jindud + '
')
print('''
_ _
| | (_)
| | __ _ _ __ __ _ _____
| | / _` | '_ / _` |_ / |
| |___| (_| | | | | (_| |/ /| |
|________,_|_| |_|__, /___|_|
__/ |
|___/
''')
time.sleep(5)
inp = input('导入IP文本:')
ips = [x.replace('
','').strip() for x in open(inp,'r',encoding='utf-8').readlines()]
por = input('输入扫描端口(21,22,80-888,6379,27017):')
rat = input('设置每秒发包量(1000-5000):')
try:
if 0<int(rat)<500000:
pass
except:
print('发包量设置错误')
time.sleep(600)
res = []
if por == '0':
por = '2375,1098,135,50030,27018,873,514,8888,6002,4444,9110,4899,9200,1435,7000,27019,8161,11211,1521,8093,3306,137,999,4950,1099,50070,6371,88,7003,1434,89,9999,513,87,2601,8009,9300,5632,1080,9043,512,8649,6000,22,5900,9001,2049,9990,6001,8089,50000,81,53,888,2439,9111,8088,1423,8873,23,8083,1527,1001,21,80,6003,525,3888,9000,30015,1433,389,27017,2888,8000,2638,2181,7001,111,6372,25,4445,3389,139,5631,8080,6379,445,7002,161,2100'
start_time = time.time()
TIME = str(int(str(time.time() - start_time).split('.')[0]) / 60).split('.')[0] + '分钟'
for ip in ips:
a = IpInfoScan(ip)
res.extend(a.GetResult(por.replace('，',',').replace(' ',',').replace(',,',','),rat))
if res == []:
print('
扫描完毕~无存活IP~')
else:
CleanData(IPdata=res,txtfile=ImgTxt,htmlfile=ImgHtml)
print('
扫描完毕~耗时:{}~
结果保存在:{}'.format(TIME,os.path.join(os.path.abspath('..'),ImgHtml)))
while 1:
time.sleep(500)使用实例
将内网的主机IP保存在一个文本内：
192.168.8.0/24
10.152.168.0/24
10.16.26.3
10.26.36.0/24
192.168.0.2
192.168.0.6
192.168.0.15
192.168.0.16
192.168.0.17
192.168.0.12
192.168.0.19
接下来直接启动主程序，按照提示导入IP文本，设置扫描端口(输入0使用默认端口扫描)，设置每秒的发包量，设置扫描进程数。即可开始扫描~
生成结果
结果自动保存在当前目录下，以html文件的形式展示结果
具体的每台主机资产都详细展示出来
点击相关数据的标签，可以查看该数据的全部相关主机
源码地址
https://github.com/LangziFun/LangNetworkTopology3
博客：langzi.fun微
本文参与 腾讯云自媒体分享计划，分享自微信公众号。
原始发表：2020-04-14，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读
目录
相关产品与服务
腾讯云 BI
腾讯云 BI（Business Intelligence，BI）提供从数据源接入、数据建模到数据可视化分析全流程的BI能力，帮助经营者快速获取决策数据依据。系统采用敏捷自助式设计，使用者仅需通过简单拖拽即可完成原本复杂的报表开发过程，并支持报表的分享、推送等企业协作场景。