From 5143bd81e82c35ce686dd40860ec2aebe30aaf22 Mon Sep 17 00:00:00 2001 From: Jacek Lewandowski Date: Fri, 26 May 2023 09:29:37 +0200 Subject: [PATCH] Track the amount of read data per row If an sstable is corrupted in a nasty way, we may read invalid cell sizes and try to read much more data for a row than we should. In rare scenarios this can lead even to OOMs. This simple fix adds tracking and limiting the amount of data that is read per row. Row has its size stored in preamble which can be used as a limit. If the deserialization code tries to read more than that, it will simply fail with EOF which will prevent more serious problems later. Patch by Jacek Lewandowski; reviewed by Berenguer Blasi and Maxwell Guo for CASSANDRA-18513 # Conflicts: # src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java --- CHANGES.txt | 28 +++- .../org/apache/cassandra/db/TypeSizes.java | 1 + .../db/rows/UnfilteredSerializer.java | 9 +- .../io/util/TrackedDataInputPlus.java | 62 ++++++-- .../db/rows/UnfilteredSerializerTest.java | 135 ++++++++++++++++++ 5 files changed, 218 insertions(+), 17 deletions(-) create mode 100644 test/unit/org/apache/cassandra/db/rows/UnfilteredSerializerTest.java diff --git a/CHANGES.txt b/CHANGES.txt index 4b761c6d19..e0ffe9061e 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0.11 + * Track the amount of read data per row (CASSANDRA-18513) * Fix Down nodes counter in nodetool describecluster (CASSANDRA-18512) * Remove unnecessary shuffling of GossipDigests in Gossiper#makeRandomGossipDigest (CASSANDRA-18546) Merged from 3.11: @@ -7,6 +8,7 @@ Merged from 3.0: * Suppress CVE-2023-2976 (CASSANDRA-18562) * Remove dh_python use in Debian packaging (CASSANDRA-18558) + 4.0.10 * Improve nodetool enable{audit,fullquery}log (CASSANDRA-18550) * Report network cache info in nodetool (CASSANDRa-18400) @@ -28,6 +30,7 @@ Merged from 3.0: * Do not remove SSTables when cause of FSReadError is OutOfMemoryError while using best_effort disk failure policy (CASSANDRA-18336) * Do not remove truncated_at entry in system.local while dropping an index (CASSANDRA-18105) + 4.0.9 * Update zstd-jni library to version 1.5.5 (CASSANDRA-18429) * Backport CASSANDRA-17205 to 4.0 branch - Remove self-reference in SSTableTidier (CASSANDRA-18332) @@ -45,6 +48,7 @@ Merged from 3.0: * Fix the ordering of sstables when running sstableupgrade tool (CASSANDRA-18143) * Fix default file system error handler for disk_failure_policy die (CASSANDRA-18294) + 4.0.8 * Log warning message on aggregation queries without key or on multiple keys (CASSANDRA-18219) * Fix the output of FQL dump tool to properly separate entries (CASSANDRA-18215) @@ -81,6 +85,7 @@ Merged from 3.0: * Suppress CVE-2022-41854 and similar (CASSANDRA-18083) * Fix running Ant rat targets without git (CASSANDRA-17974) + 4.0.7 * Remove empty cq4 files in log directory to not fail the startup of BinLog (CASSANDRA-17933) * Fix multiple BufferPool bugs (CASSANDRA-16681) @@ -103,6 +108,7 @@ Merged from 3.0: * Fix scrubber falling into infinite loop when the last partition is broken (CASSANDRA-17862) * Fix resetting schema (CASSANDRA-17819) + 4.0.6 * Fix race condition on updating cdc size and advancing to next segment (CASSANDRA-17792) * Add 'noboolean' rpm build for older distros like CentOS7 (CASSANDRA-17765) @@ -119,6 +125,7 @@ Merged from 3.0: * Improve libjemalloc resolution in bin/cassandra (CASSANDRA-15767) * Fix restarting of services on gossipping-only member (CASSANDRA-17752) + 4.0.5 * Utilise BTree improvements to reduce garbage and improve throughput (CASSANDRA-15511) * Make sure existing delayed tasks in StreamTransferTask cannot prevent clean shutdown (CASSANDRA-17706) @@ -137,6 +144,7 @@ Merged from 3.0: * Suppress CVE-2022-24823 (CASSANDRA-17633) * fsync TOC and digest files (CASSANDRA-10709) + 4.0.4 * Optimise BTree build,update and transform operations (CASSANDRA-15510) * Clean up schema migration coordinator and tests (CASSANDRA-17533) @@ -179,11 +187,13 @@ Merged from 3.0: * Lazy transaction log replica creation allows incorrect replica content divergence during anticompaction (CASSANDRA-17273) * LeveledCompactionStrategy disk space check improvements (CASSANDRA-17272) + 4.0.3 * Deprecate otc_coalescing_strategy, otc_coalescing_window_us, otc_coalescing_enough_coalesced_messages, otc_backlog_expiration_interval_ms (CASSANDRA-17377) * Improve start up processing of Incremental Repair information read from system.repairs (CASSANDRA-17342) + 4.0.2 * Extend operator control over the UDF threading model for CVE-2021-44521 (CASSANDRA-17352) * Full Java 11 support (CASSANDRA-16894) @@ -243,6 +253,7 @@ Merged from 3.0: * Make the addition of regular column to COMPACT tables throw an InvalidRequestException (CASSANDRA-14564) * Race in CompactionExecutorTest (CASSANDRA-17239) + 4.0.1 * Tolerate missing DNS entry when completing a host replacement (CASSANDRA-16873) * Harden PrunableArrayQueue against Pruner implementations that might throw exceptions (CASSANDRA-16866) @@ -279,6 +290,7 @@ Merged from 3.0: * Ensure java executable is on the path (CASSANDRA-14325) * Clean transaction log leftovers at the beginning of sstablelevelreset and sstableofflinerelevel (CASSANDRA-12519) + 4.0.0 * Avoid signaling DigestResolver until the minimum number of responses are guaranteed to be visible (CASSANDRA-16807) * Fix pre-4.0 FWD_FRM parameter serializer (CASSANDRA-16808) @@ -288,6 +300,7 @@ Merged from 3.0: Merged from 3.11: Merged from 3.0: + 4.0-rc2 * Improved password obfuscation (CASSANDRA-16801) * Avoid memoizing the wrong min cluster version during upgrades (CASSANDRA-16759) @@ -323,6 +336,7 @@ Merged from 3.0: * Prevent loss of commit log data when moving sstables between nodes (CASSANDRA-16619) * Fix materialized view builders inserting truncated data (CASSANDRA-16567) + 4.0-rc1 * Allow for setting buffer max capacity to increase it dynamically as needed (CASSANDRA-16524) * Harden internode message resource limit accounting against serialization failures (CASSANDRA-16616) @@ -404,6 +418,7 @@ Merged from 2.2: * Fix centos packaging for arm64, >=4.0 rpm's now require python3 (CASSANDRA-16477) * Make TokenMetadata's ring version increments atomic (CASSANDRA-16286) + 4.0-beta4 * DROP COMPACT STORAGE should invalidate prepared statements still using CompactTableMetadata (CASSANDRA-16361) * Update default num_tokens to 16 and allocate_tokens_for_local_replication_factor to 3 (CASSANDRA-13701) @@ -451,6 +466,7 @@ Merged from 3.0: Merged from 2.2: * Fix the histogram merge of the table metrics (CASSANDRA-16259) + 4.0-beta3 * Segregate Network and Chunk Cache BufferPools and Recirculate Partially Freed Chunks (CASSANDRA-15229) * Fail truncation requests when they fail on a replica (CASSANDRA-16208) @@ -493,6 +509,7 @@ Merged from 2.2: * Automatically drop compact storage on tables for which it is safe (CASSANDRA-16048) * Fixed NullPointerException for COMPACT STORAGE tables with null clustering (CASSANDRA-16241) + 4.0-beta2 * Add addition incremental repair visibility to nodetool repair_admin (CASSANDRA-14939) * Always access system properties and environment variables via the new CassandraRelevantProperties and CassandraRelevantEnv classes (CASSANDRA-15876) @@ -531,6 +548,7 @@ Merged from 2.2: Merged from 2.1: * Only allow strings to be passed to JMX authentication (CASSANDRA-16077) + 4.0-beta1 * Remove BackPressureStrategy (CASSANDRA-15375) * Improve messaging on indexing frozen collections (CASSANDRA-15908) @@ -620,6 +638,7 @@ Merged from 2.2: Merged from 2.1: * Fix writing of snapshot manifest when the table has table-backed secondary indexes (CASSANDRA-10968) + 4.0-alpha4 * Add client request size server metrics (CASSANDRA-15704) * Add additional logging around FileUtils and compaction leftover cleanup (CASSANDRA-15705) @@ -676,6 +695,7 @@ Merged from 2.1: * Fix Red Hat init script on newer systemd versions (CASSANDRA-15273) * Allow EXTRA_CLASSPATH to work on tar/source installations (CASSANDRA-15567) + 4.0-alpha3 * Restore monotonic read consistency guarantees for blocking read repair (CASSANDRA-14740) * Separate exceptions for CAS write timeout exceptions caused by contention and unkown result (CASSANDRA-15350) @@ -727,6 +747,7 @@ Merged from 2.2: * In-JVM DTest: Support NodeTool in dtest (CASSANDRA-15429) * Added data modeling documentation (CASSANDRA-15443) + 4.0-alpha2 * Fix SASI non-literal string comparisons (range operators) (CASSANDRA-15169) * Upgrade Guava to 27, and to java-driver 3.6.0 (from 3.4.0-SNAPSHOT) (CASSANDRA-14655) @@ -1118,6 +1139,7 @@ Merged from 3.0: * Multi-version in-JVM dtests (CASSANDRA-14937) * Allow instance class loaders to be garbage collected for inJVM dtest (CASSANDRA-15170) + 3.11.6 * Fix bad UDT sstable metadata serialization headers written by C* 3.0 on upgrade and in sstablescrub (CASSANDRA-15035) * Fix nodetool compactionstats showing extra pending task for TWCS - patch implemented (CASSANDRA-15409) @@ -1148,6 +1170,7 @@ Merged from 2.2: * In-JVM DTest: Support NodeTool in dtest (CASSANDRA-15429) * Fix NativeLibrary.tryOpenDirectory callers for Windows (CASSANDRA-15426) + 3.11.5 * Fix cassandra-env.sh to use $CASSANDRA_CONF to find cassandra-jaas.config (CASSANDRA-14305) * Fixed nodetool cfstats printing index name twice (CASSANDRA-14903) @@ -2284,7 +2307,6 @@ Merged from 2.1: * Sane default (200Mbps) for inter-DC streaming througput (CASSANDRA-8708) - 3.2 * Make sure tokens don't exist in several data directories (CASSANDRA-6696) * Add requireAuthorization method to IAuthorizer (CASSANDRA-10852) @@ -7317,7 +7339,6 @@ Merged from 0.8: - Similarly, merged batch_insert_super into batch_insert. - 0.4.0 beta * On-disk data format has changed to allow billions of keys/rows per node instead of only millions @@ -7354,7 +7375,6 @@ Merged from 0.8: * Rename configuration "table" to "keyspace" * Moved to crash-only design; no more shutdown (just kill the process) * Lots of bug fixes - Full list of issues resolved in 0.4 is at https://issues.apache.org/jira/secure/IssueNavigator.jspa?reset=true&&pid=12310865&fixfor=12313862&resolution=1&sorter/field=issuekey&sorter/order=DESC @@ -7385,3 +7405,5 @@ Full list of issues resolved in 0.4 is at https://issues.apache.org/jira/secure/ * Combined blocking and non-blocking versions of insert APIs * Added FlushPeriodInMinutes configuration parameter to force flushing of infrequently-updated ColumnFamilies + + diff --git a/src/java/org/apache/cassandra/db/TypeSizes.java b/src/java/org/apache/cassandra/db/TypeSizes.java index aed18a6c04..8df83b6ba2 100644 --- a/src/java/org/apache/cassandra/db/TypeSizes.java +++ b/src/java/org/apache/cassandra/db/TypeSizes.java @@ -29,6 +29,7 @@ public final class TypeSizes public static final int BOOL_SIZE = 1; public static final int BYTE_SIZE = 1; + public static final int CHAR_SIZE = 2; public static final int SHORT_SIZE = 2; public static final int INT_SIZE = 4; public static final int LONG_SIZE = 8; diff --git a/src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java b/src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java index 1c0dcd498b..b7a9fc8448 100644 --- a/src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java +++ b/src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java @@ -28,6 +28,7 @@ import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputBuffer; import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.io.util.FileDataInput; +import org.apache.cassandra.io.util.TrackedDataInputPlus; import org.apache.cassandra.utils.SearchIterator; import org.apache.cassandra.utils.WrappedException; @@ -578,8 +579,9 @@ public class UnfilteredSerializer if (header.isForSSTable()) { - in.readUnsignedVInt(); // Skip row size + long rowSize = in.readUnsignedVInt(); in.readUnsignedVInt(); // previous unfiltered size + in = new TrackedDataInputPlus(in, rowSize); } LivenessInfo rowLiveness = LivenessInfo.EMPTY; @@ -600,13 +602,14 @@ public class UnfilteredSerializer try { + DataInputPlus finalIn = in; columns.apply(column -> { try { if (column.isSimple()) - readSimpleColumn(column, in, header, helper, builder, livenessInfo); + readSimpleColumn(column, finalIn, header, helper, builder, livenessInfo); else - readComplexColumn(column, in, header, helper, hasComplexDeletion, builder, livenessInfo); + readComplexColumn(column, finalIn, header, helper, hasComplexDeletion, builder, livenessInfo); } catch (IOException e) { diff --git a/src/java/org/apache/cassandra/io/util/TrackedDataInputPlus.java b/src/java/org/apache/cassandra/io/util/TrackedDataInputPlus.java index dc5bbb6599..d319a86e4f 100644 --- a/src/java/org/apache/cassandra/io/util/TrackedDataInputPlus.java +++ b/src/java/org/apache/cassandra/io/util/TrackedDataInputPlus.java @@ -19,19 +19,37 @@ package org.apache.cassandra.io.util; import java.io.DataInput; import java.io.DataInputStream; +import java.io.EOFException; import java.io.IOException; +import net.nicoulaj.compilecommand.annotations.Inline; +import org.apache.cassandra.db.TypeSizes; + /** * This class is to track bytes read from given DataInput */ public class TrackedDataInputPlus implements DataInputPlus, BytesReadTracker { private long bytesRead; + private final long limit; final DataInput source; + /** + * Create a TrackedDataInputPlus from given DataInput with no limit of bytes to read + */ public TrackedDataInputPlus(DataInput source) + { + this(source, -1); + } + + /** + * Create a TrackedDataInputPlus from given DataInput with limit of bytes to read. If limit is reached + * {@link IOException} will be thrown when trying to read more bytes. + */ + public TrackedDataInputPlus(DataInput source, long limit) { this.source = source; + this.limit = limit; } public long getBytesRead() @@ -49,55 +67,63 @@ public class TrackedDataInputPlus implements DataInputPlus, BytesReadTracker public boolean readBoolean() throws IOException { + checkCanRead(TypeSizes.BOOL_SIZE); boolean bool = source.readBoolean(); - bytesRead += 1; + bytesRead += TypeSizes.BOOL_SIZE; return bool; } public byte readByte() throws IOException { + checkCanRead(TypeSizes.BYTE_SIZE); byte b = source.readByte(); - bytesRead += 1; + bytesRead += TypeSizes.BYTE_SIZE; return b; } public char readChar() throws IOException { + checkCanRead(TypeSizes.CHAR_SIZE); char c = source.readChar(); - bytesRead += 2; + bytesRead += TypeSizes.CHAR_SIZE; return c; } public double readDouble() throws IOException { + checkCanRead(TypeSizes.DOUBLE_SIZE); double d = source.readDouble(); - bytesRead += 8; + bytesRead += TypeSizes.DOUBLE_SIZE; return d; } public float readFloat() throws IOException { + checkCanRead(TypeSizes.FLOAT_SIZE); float f = source.readFloat(); - bytesRead += 4; + bytesRead += TypeSizes.FLOAT_SIZE; return f; } public void readFully(byte[] b, int off, int len) throws IOException { + checkCanRead(len); source.readFully(b, off, len); bytesRead += len; } public void readFully(byte[] b) throws IOException { + checkCanRead(b.length); source.readFully(b); bytesRead += b.length; } public int readInt() throws IOException { + checkCanRead(TypeSizes.INT_SIZE); int i = source.readInt(); - bytesRead += 4; + bytesRead += TypeSizes.INT_SIZE; return i; } @@ -110,15 +136,17 @@ public class TrackedDataInputPlus implements DataInputPlus, BytesReadTracker public long readLong() throws IOException { + checkCanRead(TypeSizes.LONG_SIZE); long l = source.readLong(); - bytesRead += 8; + bytesRead += TypeSizes.LONG_SIZE; return l; } public short readShort() throws IOException { + checkCanRead(TypeSizes.SHORT_SIZE); short s = source.readShort(); - bytesRead += 2; + bytesRead += TypeSizes.SHORT_SIZE; return s; } @@ -129,22 +157,34 @@ public class TrackedDataInputPlus implements DataInputPlus, BytesReadTracker public int readUnsignedByte() throws IOException { + checkCanRead(TypeSizes.BYTE_SIZE); int i = source.readUnsignedByte(); - bytesRead += 1; + bytesRead += TypeSizes.BYTE_SIZE; return i; } public int readUnsignedShort() throws IOException { + checkCanRead(TypeSizes.SHORT_SIZE); int i = source.readUnsignedShort(); - bytesRead += 2; + bytesRead += TypeSizes.SHORT_SIZE; return i; } public int skipBytes(int n) throws IOException { - int skipped = source.skipBytes(n); + int skipped = source.skipBytes(limit < 0 ? n : (int) Math.min(limit - bytesRead, n)); bytesRead += skipped; return skipped; } + + @Inline + private void checkCanRead(int size) throws IOException + { + if (limit >= 0 && bytesRead + size > limit) + { + skipBytes((int) (limit - bytesRead)); + throw new EOFException("EOF after " + (limit - bytesRead) + " bytes out of " + size); + } + } } diff --git a/test/unit/org/apache/cassandra/db/rows/UnfilteredSerializerTest.java b/test/unit/org/apache/cassandra/db/rows/UnfilteredSerializerTest.java new file mode 100644 index 0000000000..835e613b9e --- /dev/null +++ b/test/unit/org/apache/cassandra/db/rows/UnfilteredSerializerTest.java @@ -0,0 +1,135 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.db.rows; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Random; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Function; + +import org.junit.BeforeClass; +import org.junit.Test; + +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.Clustering; +import org.apache.cassandra.db.SerializationHeader; +import org.apache.cassandra.db.filter.ColumnFilter; +import org.apache.cassandra.db.marshal.BytesType; +import org.apache.cassandra.db.marshal.IntegerType; +import org.apache.cassandra.io.util.DataInputBuffer; +import org.apache.cassandra.io.util.DataOutputBuffer; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.schema.TableMetadata; + +import static org.assertj.core.api.Assertions.assertThatIOException; +import static org.junit.Assert.assertEquals; + +public class UnfilteredSerializerTest +{ + private static TableMetadata md; + + @BeforeClass + public static void beforeClass() + { + DatabaseDescriptor.daemonInitialization(); + + md = TableMetadata.builder("ks", "cf") + .addPartitionKeyColumn("pk", IntegerType.instance) + .addRegularColumn("v1", BytesType.instance) + .addRegularColumn("v2", BytesType.instance) + .build(); + } + + @Test + public void testRowSerDe() throws IOException + { + // test serialization and deserialization of a row body + testRowBodySerDe(10, Function.identity()); + } + + @Test + public void testRowSerDeWithCorruption() throws IOException + { + // test serialization and deserialization of a row body when the row is corrupted in the way that the actual + // row content is larger than the row size serialized in the preamble + ByteBuffer largeRow = getSerializedRow(50); + assertThatIOException().isThrownBy(() -> testRowBodySerDe(10, buf -> replaceRowContent(buf, largeRow))) + .withMessageMatching("EOF after \\d+ bytes out of 50"); + } + + public static void testRowBodySerDe(int cellSize, Function transform) throws IOException + { + Random random = new Random(); + ByteBuffer data1 = ByteBuffer.allocate(cellSize); + ByteBuffer data2 = ByteBuffer.allocate(cellSize); + random.nextBytes(data1.array()); + random.nextBytes(data2.array()); + + Row.Builder builder = BTreeRow.sortedBuilder(); + builder.newRow(Clustering.EMPTY); + builder.addCell(BufferCell.live(md.regularColumns().getSimple(0), TimeUnit.MILLISECONDS.toMicros(System.currentTimeMillis()), data1.duplicate())); + builder.addCell(BufferCell.live(md.regularColumns().getSimple(1), TimeUnit.MILLISECONDS.toMicros(System.currentTimeMillis()), data2.duplicate())); + Row writtenRow = builder.build(); + + try (DataOutputBuffer out = new DataOutputBuffer()) + { + UnfilteredSerializer.serializer.serialize(writtenRow, new SerializationHelper(SerializationHeader.makeWithoutStats(md)), out, 0, MessagingService.current_version); + out.flush(); + try (DataInputBuffer in = new DataInputBuffer(transform.apply(out.asNewBuffer()), false)) + { + builder = BTreeRow.sortedBuilder(); + DeserializationHelper helper = new DeserializationHelper(md, MessagingService.current_version, DeserializationHelper.Flag.LOCAL, ColumnFilter.all(md)); + Unfiltered readRow = UnfilteredSerializer.serializer.deserialize(in, SerializationHeader.makeWithoutStats(md), helper, builder); + assertEquals(writtenRow, readRow); + } + } + } + + private ByteBuffer getSerializedRow(int cellSize) throws IOException + { + AtomicReference rowData = new AtomicReference<>(); + testRowBodySerDe(cellSize, buf -> { + rowData.set(buf.duplicate()); + return buf; + }); + return rowData.get(); + } + + private static ByteBuffer replaceRowContent(ByteBuffer original, ByteBuffer replacement) + { + try (DataOutputBuffer out = new DataOutputBuffer(); + DataInputBuffer origIn = new DataInputBuffer(original, true); + DataInputBuffer replIn = new DataInputBuffer(replacement, false)) + { + out.writeByte(origIn.readByte()); // flag + out.writeUnsignedVInt(origIn.readUnsignedVInt()); // row size + replIn.readByte(); // skip flag + replIn.readUnsignedVInt(); // skip row size + out.write(replacement); + out.close(); + return out.asNewBuffer(); + } + catch (IOException ex) + { + throw new RuntimeException(ex); + } + } +}