ct/l0: preserve the produce-time record crc on the read path

src/v/cloud_topics/level_zero/reader/materialized_extent.cc

@@ -100,14 +100,9 @@ model::record_batch make_raft_data_batch(materialized_extent ext) {
       size() - model::packed_record_batch_header_size);
     auto header = storage::batch_header_from_disk_iobuf(
       std::move(header_bytes));
-    // NOTE: the serialized raft_data batch doesn't have the offset set
-    // so we need to populate it from the placeholder batch. We also need
-    // to make sure that crc is correct.
+
+    // base_offset comes from the placeholder; the serialized batch has none.
     header.base_offset = kafka::offset_cast(ext.meta.base_offset);
-    header.crc = model::crc_record_batch(header, records_bytes);
-    crc::crc32c crc;
-    model::crc_record_batch_header(crc, header);
-    header.header_crc = crc.value();
     model::record_batch batch(
       header,
       std::move(records_bytes),

src/v/cloud_topics/level_zero/stm/placeholder.cc

@@ -9,7 +9,9 @@
  */
 #include "cloud_topics/level_zero/stm/placeholder.h"
 
+#include "model/record.h"
 #include "model/record_batch_types.h"
+#include "model/record_utils.h"
 #include "storage/record_batch_builder.h"
 
 namespace cloud_topics {
@@ -79,13 +81,15 @@ ctp_placeholder parse_placeholder_batch(model::record_batch batch) {
 model::record_batch apply_placeholder_to_batch(
   const model::record_batch_header& placeholder_batch_header,
   model::record_batch uploaded_batch) {
-    // crcs and sizes are set later in reset_size_checksum_metadata
     model::record_batch_header merged_header{
       .header_crc = 0,
       .size_bytes = 0,
       .base_offset = placeholder_batch_header.base_offset,
       .type = uploaded_batch.header().type,
-      .crc = 0,
+      // The produce-time value, which has to reach the consumer unchanged so
+      // that corruption in the records stays detectable. Every field it covers
+      // is reproduced below from the same batch it was computed over.
+      .crc = uploaded_batch.header().crc,
       // We need to use the same attributes for compression and timestamp types
       // which are changed in the placeholder batch
       .attrs = uploaded_batch.header().attrs,
@@ -102,7 +106,11 @@ model::record_batch apply_placeholder_to_batch(
       .record_count = placeholder_batch_header.record_count,
       .ctx = placeholder_batch_header.ctx,
     };
-    merged_header.reset_size_checksum_metadata(uploaded_batch.data());
+    // size_bytes and header_crc describe the merged header, so both are
+    // derived here; .crc is the produce-time value set above and stands.
+    merged_header.size_bytes = model::packed_record_batch_header_size
+                               + uploaded_batch.data().size_bytes();
+    merged_header.header_crc = model::internal_header_only_crc(merged_header);
     return model::record_batch(
       merged_header,
       std::move(uploaded_batch).release_data(),

src/v/cloud_topics/tests/BUILD

@@ -40,6 +40,7 @@ redpanda_cc_gtest(
         "//src/v/cluster:self_test",
         "//src/v/cluster:topic_metrics_watcher",
         "//src/v/config",
+        "//src/v/container:chunked_vector",
         "//src/v/kafka/server/tests:kafka_test_utils",
         "//src/v/model",
         "//src/v/model:batch_builder",

src/v/cloud_topics/tests/end_to_end_test.cc

@@ -10,6 +10,7 @@
 
 #include "cloud_io/tests/s3_imposter.h"
 #include "cloud_topics/level_zero/stm/ctp_stm.h"
+#include "container/chunked_vector.h"
 #include "kafka/server/tests/list_offsets_utils.h"
 #include "kafka/server/tests/produce_consume_utils.h"
 #include "model/batch_builder.h"
@@ -139,6 +140,62 @@ TEST_F(e2e_fixture, test_l0_path) {
     }
 }
 
+// Regression test for an incorrect integity check: previous versions of
+// Redpanda wouldn't detect corrupted data on the L0 path, and would instead
+// re-CRC based on what was in the cloud. Guard against this by flipping a bit
+// in an uploaded object and ensure the client doesn't see it (our client
+// silently drops it).
+TEST_F(e2e_fixture, test_corrupt_l0_object_fails_client_crc_check) {
+    // Disable reconciliation and disable the batch cache to ensure we read
+    // from L0 objects.
+    test_local_cfg.get("cloud_topics_disable_reconciliation_loop")
+      .set_value(true);
+    test_local_cfg.get("disable_batch_cache").set_value(true);
+
+    const ss::sstring marker(64, 'A');
+    auto* producer = make_producer();
+    producer
+      ->produce_to_partition(
+        topic_name, model::partition_id(0), std::vector<kv_t>{{"key", marker}})
+      .get();
+
+    auto puts = get_requests(
+      [&marker](const http_test_utils::request_info& req) {
+          return req.method == "PUT"
+                 && req.content.find(marker) != ss::sstring::npos;
+      });
+    ASSERT_EQ(puts.size(), 1);
+    const auto l0_url = puts.front().url;
+    auto corrupted = puts.front().content;
+    auto flip_at = corrupted.find(marker) + marker.size() / 2;
+    corrupted[flip_at] = static_cast<char>(corrupted[flip_at] ^ 0x01);
+
+    // Republish the object with the corrupted body.
+    const auto key = l0_url.substr(1);
+    remove_expectations(chunked_vector<ss::sstring>::single(key));
+    add_expectations(
+      chunked_vector<expectation>::single(
+        expectation{.url = key, .body = corrupted}));
+
+    auto* consumer = make_consumer();
+    auto records = consumer
+                     ->raw_consume_from_partition(
+                       topic_name, model::partition_id(0), model::offset(0))
+                     .get();
+
+    // Sanity check that we actually got the object from storage.
+    auto gets = get_requests(
+      [&l0_url](const http_test_utils::request_info& req) {
+          return req.method == "GET" && req.url == l0_url;
+      });
+    ASSERT_FALSE(gets.empty()) << "fetch never downloaded the L0 object";
+
+    EXPECT_TRUE(records.empty())
+      << "client accepted " << records.size()
+      << " record(s) from a corrupted L0 object: the read path replaced the "
+         "produce-time record crc with one computed over the corrupted bytes";
+}
+
 TEST_F(e2e_fixture, timequery) {
     lconf().log_message_timestamp_after_max_ms.set_value(
       serde::max_serializable_ms);