GH-1317: bound FlightData frame field length before allocating

[Arawoof06]

What's Changed

ArrowMessage.frame is the DoPut/DoExchange request marshaller, and for each FlightData field it allocates a buffer sized by a length prefix taken straight off the wire before reading any content, so a client can force an arbitrarily large allocation with a tiny frame (an unauthenticated denial of service during request deframing). The descriptor and header paths use new byte[size], which allocates on the JVM heap and bypasses the BufferAllocator limit entirely. A field can never be longer than the bytes still buffered for the message, so checkFieldLength now rejects a negative or over-long size against stream.available() at all four allocation sites before anything is allocated.

Concretely, a ~10-byte HEADER frame whose length prefix encodes 0x7FFFFFFF drives a new byte[2147483647] (2 GiB) heap allocation before the subsequent read fails; with the check it is rejected in O(1) with no allocation. The added test asserts a field declaring more bytes than the frame contains is refused, and that a well-formed field still parses.

Closes #1317.

[github-actions]

Thank you for opening a pull request!

Please label the PR with one or more of:

• bug-fix
• chore
• dependencies
• documentation
• enhancement

Also, add the 'breaking-change' label if appropriate.

See CONTRIBUTING.md for details.

[jbonofre]

available() is not the number of bytes left in the message in general. It is exact for gRPC's uncompressed BufferInputStream, but when the peer compresses messages gRPC hands parse() a decompressing stream, and InflaterInputStream.available() returns 1 until EOF and 0 after.

With this check, every compressed FlightData field longer than 1 byte is rejected as malformed, so DoGet/DoPut/DoExchange fail as soon as a peer uses e.g. withCompression("gzip"). That works today: #742 added the tagFirstByte == -1 guard a few lines up for exactly this stream type.

Could we apply the available() bound only when the stream is io.grpc.KnownLength?

if (size < 0) {
  throw new IOException("Malformed FlightData frame: negative field length " + size);
}
if (stream instanceof KnownLength && size > stream.available()) {
  throw new IOException(...);
}

For other steams the length can't be checked up front, so the read itself needs to be incremental, e.g. stream.readBytes(size) plus a length check instead of new bytes[size] + readFully, and allocating the ArrowBuf once the bytes have arrived. That way the allocation follows the bytes actually received for every stream type.

[jbonofre]

This buffer (and body below) is leaked when a later field is rejected: checkFieldLength throws, and the catch at the end of frame() only wraps and rethrows.

A frame of [app_metadata, N valid bytes][data_header, len=0x7FFFFFFF] leaks N bytes of direct memory per message, which a peer can repeat until the allocator is exhausted. That is the same DoS this PR is addressing. The EOF path had this leak before, but since we are adding an explicit reject path it should release appMetadata and body on failure.

Related and pre-existing: a repeated app_metadata field overwrites the previous buffer without releasing it, unlike the BODY case just below.

Could you release the earlier one here too?

[jbonofre]

Nit: two small things here:

• for size < 0 it reads "field length -1 exceeds 10 bytes remaining", which is misleading when diagnosing a corrupt length prefix. A separate message for the negative case would help.
• String.format without a Locale renders %d with the default locale's digits. Flight SQL JDBC interval strings use the default locale for digits #1311 just moved the JDBC interval formatting Locale.ROOT for this reason. String.format(Locale.ROOT, ...) or plain concatenation would be consistent with that.

[jbonofre]

Nit: the readRawVarint32 + checkFieldLength pair is repeated at four call sites. A single readFieldLength(stream) helper that reads and validates would make it impossible to add a new length-delimited case and forget the check.

[jbonofre]

Both tests go through ByteArrayInputStream, whose available() is exact, so they can't see the compressed-stream case.

Could you add:

• a well-formed frame wrapped in a GZIPInputStream (this fails with the current check)
• a reject case with a valid app_metadata field before the oversized one, so allocator.close() in tearDown catches a leaked buffer. As written nothing is allocated before the bad field, so "rejected before anything is allocated" isn't actually asserted
• a negative length (5-byte varint), and at least one of DESCRIPTOR / BODY.

[jbonofre]

This accepts any Exception who message contains "exceeds". It passes because frame() wraps the IOException in a RuntimeException, so getMessage() is really the cause's toString(). Asserting on the type is sturdier:

final RuntimeException e =
    assertThrows(
        RuntimeException.class,
        () -> marshaller.parse(new ByteArrayInputStream(frame.toByteArray())));
assertInstanceOf(IOException.class, e.getCause());

[jbonofre]

Nit: protobuf already provides this. CodedOutputStream.writeTag(FlightData.DATA_HEADER_FIELD_NUMBER, WireFormat.WIRETYPE_LENGTH_DELIMITED) and writeUInt32NoTag(...) cover the malformed frame, which also removes the need to re-declare HEADER_TAG / APP_METADATA_TAG. The well-formed case can be FlightData.newBuilder().setAppMetadata(...).build().toByteArray(), as the marshaller tests in TestBasicOperation do.