经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置：技术经验 » 其他 » 网络安全 » 查看文章
redis未授权访问批量验证脚本编写[python]
来源：cnblogs　　作者：Feather·Lee　　时间：2019/3/4 9:14:51　　对本文有异议

漏洞简介

简单来说，redis是一个数据库。在默认的配置下，redis绑定在0.0.0.0:6379，也就是说，如果服务器有公网ip，可以通过访问其公网ip的6379端口来操作redis。最为致命的是，redis默认是没有密码验证的，可以免密码登录操作，攻击者可以通过操作redis进一步控制服务器。

漏洞的危害

  1. 无密码验证登录redis后，可读取、删除、更改数据
  2. 攻击者可以通过redis读写文件，植入后门
  3. 如果redis以root权限运行，攻击者可以写入ssh公钥文件，然后即可远程ssh登录服务器
    ...

漏洞修复

修复方案大概有以下几种：

  1. 把redis绑定在127.0.0.1即本地上
  2. 配置登录验证
  3. 防火墙设置白名单，拒绝不信任的连接
    ...

本文主要讲解验证脚本的编写，故不再过多阐述漏洞原理、利用等细节。

验证方式

登录redis后，执行info命令，可以获得类似下面的信息：

  1. # Server
  2. redis_version:5.0.3
  3. redis_git_sha1:00000000
  4. redis_git_dirty:0
  5. redis_build_id:68e47d9309ff01ae
  6. redis_mode:standalone
  7. ...

如果登录失败，是不可以执行命令的，所以我们可以向目标ip的6379（redis默认端口）发起连接，发送info命令，只要得到的响应中存在上面信息中的某些独特的字符串，如redis_version，我们就认为目标存在redis未授权访问漏洞。

代码如下：

  1. ...
  2. sock = socket.socket()  # 创建套接字
  3. try:
  4.     sock.connect((ip, 6379))  # 连接
  5.     sock.send(payload)  # 发送info命令
  6.     response = sock.recv(1024).decode()  # 接收响应数据
  7.     if 'redis_version' in response:
  8.         result = True  # 存在漏洞
  9.     else:
  10.         result = False  # 不存在漏洞
  11. except (socket.error, socket.timeout)：
  12.     # 连接失败，可能端口6379未开放，或者被拦截，此时认为漏洞不存在
  13.     result = False
  14. ...

好了，现在的关键就在：如何发送info命令？

python有操作redis的第三方库，可以很方便的操作redis。然而，我们并不使用这些第三方库，归根结底，发送info命令其实是发送了一个可以让redis服务识别的特定的数据而已，只要我们知道这个数据是什么，我们就可以使用info命令了。

下面我们就来分析，redis是如何发送info命令的。

截获info命令

我们需要搭建一个redis环境，使用抓包工具来截获使用info命令时redis发送的数据，为了方便，我使用了linux系统的命令netcat、tee来充当抓包工具，读者可以自己在linux系统下搭建redis环境尝试。

我们使用netcat连接到本地的redis服务，然后使用另一个netcat进程监听127.0.0.1:9000，将接受的连接发来的数据，重定向至连接到redis服务的netcat进程的输入，即可完成连接的转发，我们在这两个netcat经常之间，使用tee来截获数据，流程大致如下：

流程

在实际的操作中，我们还需要使用命名管道来实现双向通信，否则客户端无法接受到redis的登录响应就会阻塞，无法发生命令。

具体操作如下：

  1. 启动redis服务
  2. 创建两个管道文件：pipe1、pipe2
    shell $ mkfifo pipe1 $ mkfifo pipe2
  3. 启动一个netcat进程监听在本地的9000端口上，以pipe1作为输入，输出重定向到tee进程，tee进程负责将数据输出到pipe2和屏幕上：
    shell $ ncat -l 127.0.0.1 9000 < pipe1 | tee pipe2
  4. 启动另一个终端，启动netcat进程，负责连接redis服务，以pipe2作为输入，输出重定向到pipe1中：
    shell $ ncat 127.0.0.1 6379 < pipe2 > pipe1
  5. 用redis客户端连接本地9000端口：
    shell $ redis-cli -h 127.0.0.1 -p 9000
  6. 在redis-cli中发送info命令，tee进程在终端上的输出即为整个过程需要发送给redis的数据

附上动图：
获取payload

可以知道payload为：

  1. *1
  2. $4
  3. info

写成python字节串就是：b'*1\r\n$4\r\ninfo\r\n' （'\r\n'是换行符）

所以我们只需使用socket发送以上字节串即可达到同样的效果

编写验证poc

验证漏洞的代码如下：

  1. def poc(url):
  2.     url = url2ip(url)  # 将url转换成ip地址
  3.     if url:
  4.         port = int(url.split(':', -1)) if ':' in url else 6379 # redis默认端口是6379
  5.         host = url.split(':')[0]
  7.         payload = b'b'*1\r\n$4\r\ninfo\r\n' # 发送的数据
  9.         s = socket.socket()
  10.         socket.setdefaulttimeout(3)  # 设置超时时间
  11.         try:
  12.             s.connect((host, port))
  13.             s.send(payload)  # 发送info命令
  14.             response = s.recv(1024).decode()
  15.             s.close()
  17.             if response and 'redis_version' in data:
  18.                 return True,'%s:%s'%(host,port)
  19.         except (socket.error, socket.timeout):
  20.             pass
  21.     
  22.     return False, None

其中url转换成ip地址的函数如下：

  1. def url2ip(url):
  2.     """
  3.     url转换成ip
  4.     argument: url
  5.     return: 形如www.a.com:80格式的字符串 若转换失败则返回None
  6.     """
  8.     try:
  9.         if not url.startswith('http://') and not url.startswith('https://'):
  10.             url = 'http://' + url
  11.         ip = urlparse(url).netloc
  12.         return ip
  13.     except (ValueError, socket.gaierror):
  14.         pass
  16.     return None

处理输入

我们把要验证漏洞的目标放在一个文件里，每一行为一个目标，现在来编写一个函数，读取文件，将所有目标放到一个队列里，代码如下：

  1. def create_queue(file_name):
  2.     """
  3.     创建数据队列
  4.     argument: file_name -> 输入文件名
  5.     return: data,total 数据队列,数据总数
  6.     """
  7.     total = 0
  8.     data = Queue()
  9.     for line in open(file_name):
  10.         url = line.strip()
  11.         if url:
  12.             # 跳过空白的行
  13.             data.put(url)
  14.             total += 1
  16.     data.put(None)  # 结束标记
  17.     return data,total

创建多个线程

我们的start_jobs函数用于启动多个线程来验证目标，其代码如下：

  1. def start_jobs(data, num):
  2.     """
  3.     启动所有工作线程
  4.     argument: data -> 数据队列 num -> 线程数
  5.     """
  6.     is_alive = [True]
  7.     def job():
  8.         """工作线程"""
  9.         while is_alive[0]:
  10.             try:
  11.                 url = data.get()
  12.                 if url == None:
  13.                     # 遇到结束标记
  14.                     break
  15.                 code, result = poc(url)  # 验证漏洞
  16.                 if code:
  17.                     print(result)  # 存在漏洞
  18.             except:
  19.                 is_alive[0] = False
  20.         data.put(None)  # 结束标记
  21.                 
  22.     jobs = [ Thread(target=job) for i in range(num) ]  # 创建多个线程
  23.     for j in jobs:
  24.         j.setDaemon(True)
  25.         j.start()  # 启动线程
  27.     for j in jobs:
  28.         j.join()  # 等待线程退出

编写主程序框架

现在我们需要一个主函数来控制整个流程，代码很简单：

  1. def main():
  2.     import sys
  3.     if len(sys.argv) != 3:
  4.         print('Usage: python %s inputFile numOfThread' % sys.argv[0])
  5.         return
  6.     file_name = sys.argv[1]  # 输入文件
  7.     num = int(sys.argv[2])  # 线程数
  8.     data, total = create_queue(file_name)  # 创建数据队列
  9.     print('total: %s' % total)
  10.     begin = time()
  11.     start_jobs(data, num)  # 启动工作线程
  12.     end = time()
  13.     print('spent %ss' % str(end-begin))
  14.     
  16. if __name__ == '__main__':
  17.     main()

使用方法

现在假设输入文件名为input.txt，脚本文件名为redis_unauth.py，使用16个线程来批量验证漏洞，我们可以启动以下命令：

  1. $ python redis_unauth.py input.txt 16

完整代码

只是一个小脚本，就没必要放到github上了，这里直接贴出，需要的读者可以复制：

  1. #!/usr/python3
  3. '''
  4. created by feather
  5. '''
  7. import socket
  8. from threading import Thread
  9. from queue import Queue
  10. from time import sleep,time
  11. from urllib.parse import urlparse
  13. def poc(url):
  14.     url = url2ip(url)  # 将url转换成ip地址
  15.     if url:
  16.         port = int(url.split(':', -1)) if ':' in url else 6379 # redis默认端口是6379
  17.         host = url.split(':')[0]
  18.         payload = b'*1\r\n$4\r\ninfo\r\n' # 发送的数据
  19.         s = socket.socket()      
  20.         socket.setdefaulttimeout(3)  # 设置超时时间
  21.         try:
  22.             s.connect((host, port))
  23.             s.send(payload)  # 发送info命令
  24.             response = s.recv(1024).decode()
  25.             s.close()
  26.         
  27.             if response and 'redis_version' in response:
  28.                 return True,'%s:%s'%(host,port)
  29.         except (socket.error, socket.timeout):
  30.             pass
  31.     
  32.     return False, None
  34. def url2ip(url):
  35.     """
  36.     url转换成ip
  37.     argument: url
  38.     return: 形如www.a.com:80格式的字符串 若转换失败则返回None
  39.     """
  41.     try:
  42.         if not url.startswith('http://') and not url.startswith('https://'):
  43.             url = 'http://' + url
  44.         ip = urlparse(url).netloc
  45.         return ip
  46.     except (ValueError, socket.gaierror):
  47.         pass
  49.     return None
  50.  
  51. def create_queue(file_name):
  52.     """
  53.     创建数据队列
  54.     argument: file_name -> 输入文件名
  55.     return: data,total 数据队列,数据总数
  56.     """
  57.     total = 0
  58.     data = Queue()
  59.     for line in open(file_name):
  60.         url = line.strip()
  61.         if url:
  62.             # 跳过空白的行
  63.             data.put(url)
  64.             total += 1
  66.     data.put(None)  # 结束标记
  67.     return data,total
  69. def start_jobs(data, num):
  70.     """
  71.     启动所有工作线程
  72.     argument: data -> 数据队列 num -> 线程数
  73.     """
  74.     is_alive = [True]
  75.     def job():
  76.         """工作线程"""
  77.         while is_alive[0]:
  78.             try:
  79.                 url = data.get()
  80.                 if url == None:
  81.                     # 遇到结束标记
  82.                     break
  83.                 code, result = poc(url)  # 验证漏洞
  84.                 if code:
  85.                     print(result)  # 存在漏洞
  86.             except:
  87.                 is_alive[0] = False
  88.         data.put(None)  # 结束标记
  89.                 
  90.     jobs = [ Thread(target=job) for i in range(num) ]  # 创建多个线程
  91.     for j in jobs:
  92.         j.setDaemon(True)
  93.         j.start()  # 启动线程
  95.     for j in jobs:
  96.         j.join()  # 等待线程退出
  98. def main():
  99.     import sys
  100.     if len(sys.argv) != 3:
  101.         print('Usage: python %s inputFile numOfThread' % sys.argv[0])
  102.         return
  103.     file_name = sys.argv[1]  # 输入文件
  104.     num = int(sys.argv[2])  # 线程数
  105.     data, total = create_queue(file_name)  # 创建数据队列
  106.     print('total: %s' % total)
  107.     begin = time()
  108.     start_jobs(data, num)  # 启动工作线程
  109.     end = time()
  110.     print('spent %ss' % str(end-begin))
  111.     
  113. if __name__ == '__main__':
  114.     main()

原文链接：http://www.cnblogs.com/featherl/p/10461935.html

 友情链接：直通硅谷 　点职佳 　北美留学生论坛

网络安全热门文章

  • AI: Web: 2 Vulnhub Walkthrough - APT-101
  • 各类黑客大马默认密码后门
  • 蓝凌OA前台任意文件读取漏洞利用
  • 银河麒麟SP2 auditd服务内存泄露问题
  • confluence 挖矿木马应急响应 - APT-101
  • 信息安全题目整理（1）
  • 时间同步（NTP/PTP）技术原理介绍
  • apache log4j2 漏洞复现linux【纯详版】
  • Kali proxychains
  • 使用tor实现对网络的匿名访问

    • 网络安全推荐文章

  • 异构数据源数据同步 → 从源码分析 DataX 敏感信息的加解密
  • FFmpeg开发笔记（三十八）APP如何访问SRS推流的RTMP直播地址
  • Vulnhub-ICA01
  • 劫持TLS绕过canary && 堆和栈的灵活转换
  • 第二届黄河流域网络安全技能挑战赛Web_wirteup
  • 深入理解高级加密标准（Advanced Encryption Standard）
  • 渗透测试-信息收集
  • RC4Drop加密技术：原理、实践与安全性探究
  • 知名压缩软件 xz 被植入后门，黑客究竟是如何做到的？
  • 《熬夜整理》保姆级系列教程-玩转Wireshark抓包神器教程(1)-初识Wireshark - 北京-宏哥

    • 本站QQ群：前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

    W3xue 的所有内容仅供测试，对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
    关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接（目前9元/月）请联系QQ：27243702 沸活量
    皖ICP备17017327号-2　皖公网安备34020702000426号