端口扫描系统实践心得

摘要

端口扫描对任何一名网络安全从业者来说都不陌生,但作为一名小白,在甲方做扫描系统时踩了不少坑,在网络上找相关资料时没有发现太多相关的文章,于是想写下这篇文章和大家分享一下代码,顺便讨教一下主机存活判断和指纹识别的问题,欢迎大佬们批评和指正。

0×00 目的

对于外网,能够监控对外开放端口情况,并及时的发现向外暴露的高危端口,以便安全人员进行响应处理。对于内网,日常 的端口扫描以及指纹识别,不仅能够帮助梳理公司资产,并且能够帮助进行后续内网的漏洞扫描。

0×01 存活主机判断

开始做端口扫描时,所考虑的第一步便是存活主机判断。最初的设想便是使用nmap的-sP参数,对IP地址进行存活判断。

代码如下:

def ip_alive_check(ip_str): cmd = "/usr/bin/nmap -sP "+ip_str
   output = os.popen(cmd).readlines()
   flag = False for line in list(output): if not line: continue if str(line).lower().find("1 host up") >= 0:
           flag = True break return flag 

但随后便发现这个办法存在问题,引用Nmap官方文档如下:

-sP选项在默认情况下, 发送一个ICMP回声请求和一个TCP报文到80端口。

如果非特权用户执行,就发送一个SYN报文 (用connect()系统调用)到目标机的80端口。

当特权用户扫描局域网上的目标机时,会发送ARP请求(-PR), ,除非使用了–send-ip选项。

-sP选项可以和除-P0)之外的任何发现探测类型-P* 选项结合使用以达到更大的灵活性。

一旦使用了任何探测类型和端口选项,默认的探测(ACK和回应请求)就被覆盖了。 

当防守严密的防火墙位于运行Nmap的源主机和目标网络之间时, 推荐使用那些高级选项。

否则,当防火墙捕获并丢弃探测包或者响应包时,一些主机就不能被探测到。

抓包如下:

局域网环境:

非root用户

image.pngimage.png

nmap通过向目标IP的80端口和443端口分别发送SYN包来判断主机是否存活,由于目标主机的80和443端口均未开放,所以均返回RST包

nmap扫描结果:0 hosts up

root用户

image.pngimage.png

nmap发送ARP请求并得到响应

nmap扫描结果:1 hosts up

非局域网环境:

非root用户

image.pngimage.png

同样的,nmap向目标主机的80和443端口发送SYN包,通过返回的确认包得到目标主机存活。所以扫描结果为:1 hosts up。

root用户

image.png

image.png

这次,nmap不仅向目标主机的80和443端口发送了SYN包,还向目标主机发送了ICMP Echo请求以及Timestamp请求,nmap会综合这四种方式的响应情况来判断目标主机是否存活。显然这次的扫描结果为:1 hosts up。

通过对nmap -sP参数的分析便可得知,实际上对存活主机的判断并不准确。许多主机的防火墙会过滤掉ICMP包,而且80和443端口也不一定会保证对外开放。

而nmap官方文档中提到的高级选项在实际的使用中也都不能保证准确性,所以对于存活主机的判断,一直没有找到比较好的解决办法,在实际的扫描中便没有用上这一步骤,还请各路大佬指点指点有没有什么成本比较低的解决方案。

0×02 Masscan扫描端口

直接对全端口使用nmap进行扫描速度较慢,所以选择使用号称,三分钟扫遍全网的masscan。

masscan采用的是无状态的扫描技术即无需关心TCP状态,不占用系统TCP/IP协议栈资源,忘记syn,ack,fin,timewait ,不进行会话组包,而nmap则是需要记录TCP/IP的状态,并且OS能够处理的TCP/IP连接数存在上限,这就导致了nmap扫描的速度不如masscan。

代码如下:

class Masscan(object): def __init__(self, args): self.masscan_bin = config.MASSCAN_BIN # Masscan路径 例如:/usr/bin/masscan self.result_xml = '/tmp/masscan/'+args['hosts'] # 暂存的masscan扫描结果名称 self.rate = config.MASSCAN_RATE # 发包速率,例如:10000 self.retries = config.MASSCAN_RETRIES # 发送重试的次数 例如:3 self.wait = config.MASSCAN_WAIT # 指定发送完包之后的等待时间,例如:5 self.ports = args['ports'] # 端口 self.hosts = args['hosts'] # IP def scan(self): cmd = 'mkdir -p /tmp/masscan/' os.system(cmd)
        command = ( '{masscan_bin} -oX {result_xml} --rate={rate} --retries={retries} --wait={wait} -p {ports} {hosts}' ).format(
            masscan_bin=self.masscan_bin,
            result_xml=self.result_xml,
            rate=self.rate,
            retries=self.retries,
            wait=self.wait,
            hosts=self.hosts,
            ports=self.ports
        )

        process = subprocess.Popen(
            command,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            shell=True )
        logger.info(b'\nStarting masscan,the command is '+str(command)) try:
            _, stderr = process.communicate() if not stderr.startswith(b'\nStarting masscan'):
                logger.failure('Masscan Error\n{}'.format(stderr))
                os._exit(1) except KeyboardInterrupt:
            logger.failure('User aborted')
            os._exit(1) def parse_result_xml(self, d_ip): result = {} try:
            xml_size = os.path.getsize(self.result_xml) if xml_size > 0 and xml_size < 10000:
                tree = ET.parse(self.result_xml)
                root = tree.getroot() for host in root.iter('host'):
                    ip = host.find('address').attrib['addr']
                    port = host.find('ports').find('port').attrib['portid'] if result.setdefault(ip):
                        result[ip].append(port) else:
                        result[ip] = [port] elif xml_size >= 10000:
                result = {d_ip: ['1-65535']} else:
                result = {} except Exception as e:
            logger.info('----------')
            logger.info(str(e))
            logger.info('ParseError!!!')
            logger.info('----------')
            logger.info(self.result_xml) pass cmd_rm = 'rm -rf ' + self.result_xml
        os.system(cmd_rm) return result 

在使用masscan得注意速率问题,在带宽有限的情况下,速率过高则会导致丢包的情况发生从而导致扫描结果漏报。具体的速率根据实际的带宽情况慢慢调教即可。

在缓存masscan的扫描结果时,选择了直接写在tmp目录下,解析完后再删除。也可以使用redis进行缓存。

在实际的测试中发现了一个问题,masscan在扫描时,可能是因为目标主机防火墙的抗DDos功能,对masscan所发送的SYN包均会回复ACK包,所以masscan会误报部分IP开放特别大量端口的情况。选择了对xml文件的大小加了个判断,如果过大,直接将结果置为1-65535,扔给nmap重新扫一下。

0×03 Nmap扫描识别指纹

Masscan虽然扫描速度够快,但是在指纹识别这一块却是远远比不了nmap,于是在masscan扫描完成后,使用nmap对端口进行指纹识别,以及确认结果以防止masscan误报(实际上masscan的误报好像挺少的)。

代码如下:

class Nmap(object): def __init__(self, masscan_result): self.nm = nmap.PortScanner(nmap_search_path=(config.NMAP_BIN,)) # config.NMAP_BIN:nmap的路径,例如:/usr/bin/nmap self.nmap_args = config.NMAP_ARGS # nmap 扫描时的参数 例如:-Pn -sV -sS --host-timeout 1200 self.targets = []
        self.result = [] for host, ports in masscan_result.items():
            self.targets.append({host: ','.join(ports)}) def scan(self, args): for target in self.targets: for host, ports in target.items(): try:
                    self.nm.scan(host, ports, self.nmap_args) if host not in self.nm.all_hosts(): continue if self.nm[host].has_key('tcp'): for port, data in self.nm[host]['tcp'].items():
                            state = data['state']
                            product = data['product']
                            name = data['name']
                            ip = args['hosts']
                            address = args['address'] if product:
                                service = product else:
                                service = name
                            version = data['version'] if state == "open":
                                x = save_it(ip, port, address, service, version)
                                x.detect_new_port() # 储存结果时,进行一下判断,看是否是新增端口 if self.nm[host].has_key('udp'): for port, data in self.nm[host]['udp'].items():
                            state = data['state']
                            product = data['product']
                            name = data['name']
                            ip = args['hosts']
                            address = args['address'] if product:
                                service = product else:
                                service = name
                            version = data['version'] if state == "open":
                                x = save_it(ip, port, address, service, version)
                                x.detect_new_port() # 储存结果时,进行一下判断,看是否是新增端口 except Exception as e:
                    logger.info('the exception i nmap.scan is ' + str(e)) continue 

在设置nmap的扫描参数时,别忘了带上-Pn或者-P0跳过判断主机存活的步骤,因为nmap默认是会先对主机进行存活判断再进行扫描,可能会因为误判而导致漏扫。

0×04 告警

告警部分则是和扫描一样,使用django的celery进行定时任务,所以扫描和告警存在一定的时间间隔,于是在告警前便使用socket对库中的扫描结果进行一下验证。

def detect_port(ip, port): s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(1) try:
        s.connect((ip, int(port)))
        s.shutdown(2) return True except Exception as e: return False 

0×05 后续

多进程则是由celery所依赖的billiard库实现

from billiard import Pool

p = Pool(5) for ip in ip_data:
    p.apply_async(port_scan, args=(ip, log_name,))
p.close()
p.join() 

如果IP数量较大,可以将端口扫描、资产发现、漏洞扫描等集成起来,做成agent,搭配rabbitmq实现分布式的扫描系统。

扫描时别忘了避开交换机和打印机等比较容易脆弱的设备,在实际进行扫描时就曾遇见过某型号打印机存在缺陷,一扫就自己疯狂打印(都吓到了晚上正在加班的同事),最后无奈只能避开。

经过一段时间的使用,带宽足够,速率合适的情况下,masscan扫描的准确性还挺不错的。但即使nmap的指纹库已经较为丰富,在识别web应用程序、中间件这些时,还是有些不够用,不便于后续的漏洞扫描。

目前评论:0 条

发表评论