apache / apache/pulsar

[improve][broker] Bound decompressed message size before allocation; release ByteBuf on codec exceptions

Open
#26,442 1 comment 0 reactions 0 assignees View on GitHub
type/enhancement
Dominant language
Java
Stars
15.3k
Forks
3.8k
Avg merge
1d 14h
Merged PRs (30d)
160

Description

### Search before reporting

- [x] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.

### Motivation

When a message is compressed, the decompressed size is taken directly from the uncompressedSize field of MessageMetadata and used to allocate a ByteBuf before decompression. This field is client-controlled, the broker's produce path does not validate it before persisting the message, and most decompression call-sites do not validate it before allocating. A single malformed message can therefore trigger an allocation of up to ~2 GiB (Integer.MAX_VALUE) on any component that later decompresses the entry — even though the compressed payload itself is tiny and passes the frame-size limit.

This is filed as a robustness / defense-in-depth improvement. (Per the documented security model, denial of service by a perimeter-trusted client is outside Pulsar's threat model, so this is not a security report.)

### Solution

1. Bound uncompressedSize at the produce entry point (preferred — prevents the malformed entry from being persisted): in ServerCnx.handleSend / PersistentTopic.publishMessage, parse the metadata once and reject the send when uncompressedSize > maxMessageSize.

2. Defense-in-depth at every codec.decode(..., getUncompressedSize()) call-site: in MessageParser.uncompressPayloadIfNeeded (add an uncompressedSize > maxMessageSize branch alongside the existing payloadSize check at :160), PersistentTopicsBase:3350, PulsarTopicCompactionService:150, AbstractTwoPhaseCompactor:479, RawBatchConverter:69/150. This also covers entries that bypass the produce path (cross-cluster replication from older brokers, direct BookKeeper tampering).

3. Release ByteBuf on codec exceptions: wrap the allocation + (de)compression in try/finally, releasing the buffer on failure, in CompressionCodecLZ4/Zstd/ZLib/Snappy.encode/decode.

### Alternatives

_No response_

### Anything else?

_No response_

### Are you willing to submit a PR?

- [x] I'm willing to submit a PR!

Contributor guide

Open the contributing guide

Research direction

Start with ServerCnx.handleSend and PersistentTopic.publishMessage, then inspect the listed codec.decode call-sites, especially MessageParser.uncompressPayloadIfNeeded at :160. Review the allocation and exception paths in CompressionCodecLZ4, CompressionCodecZstd, CompressionCodecZLib, and CompressionCodecSnappy. Done means oversized uncompressed sizes are rejected before allocation or persistence, and allocated buffers are released when codec operations fail.

Written by the indexing model from the issue text.

Assessment

Tech stack
java
Domain
distributed-systems, performance
Issue type
Bug
Difficulty
5/5
Estimated time
Over a week
Activity status
Active
Clarity
Clearly specified
Newbie friendliness
45/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.