MB-29040: [2/2] Sanitise delete with value via DcpConsumer

1) Allow a wider range of datatypes to be received by adjusting the
mcbp validator.

2) Update the DcpConsumer so that delete with values are checked and
sanitised, this means even a deleted marked as 'xattr' needs checking
because the source may be sending an incorrect xattr with raw body +
user xattrs which will need deleting.

3) Add a DcpConsumer test which checks we strip the faulty input.

Change-Id: I219f21df9a63bc6b1c004fa382bd1f32f94a3e90
Reviewed-on: http://review.couchbase.org/93041
Reviewed-by: Dave Rigby <[email protected]>
Tested-by: Build Bot <[email protected]>

Author: jimwwalker

daemon/mcbp_validators.cc

@@ -383,24 +383,38 @@ static protocol_binary_response_status dcp_mutation_validator(const Cookie& cook
     return verify_common_dcp_restrictions(cookie);
 }
 
+/// @return true if the datatype is valid for a deletion
+static bool valid_dcp_delete_datatype(protocol_binary_datatype_t datatype) {
+    // MB-29040: Allowing xattr + JSON. A bug in the producer means
+    // it may send XATTR|JSON (with snappy possible). These are now allowed
+    // so rebalance won't be failed and the consumer will sanitise the faulty
+    // documents.
+    std::array<const protocol_binary_datatype_t, 5> valid = {
+            {PROTOCOL_BINARY_RAW_BYTES,
+             PROTOCOL_BINARY_DATATYPE_XATTR,
+             PROTOCOL_BINARY_DATATYPE_XATTR | PROTOCOL_BINARY_DATATYPE_SNAPPY,
+             PROTOCOL_BINARY_DATATYPE_XATTR | PROTOCOL_BINARY_DATATYPE_JSON,
+             PROTOCOL_BINARY_DATATYPE_XATTR | PROTOCOL_BINARY_DATATYPE_SNAPPY |
+                     PROTOCOL_BINARY_DATATYPE_JSON}};
+    for (auto d : valid) {
+        if (datatype == d) {
+            return true;
+        }
+    }
+    return false;
+}
+
 static protocol_binary_response_status dcp_deletion_validator(const Cookie& cookie)
 {
     auto req = static_cast<protocol_binary_request_dcp_deletion*>(
             cookie.getPacketAsVoidPtr());
-    const auto datatype = req->message.header.request.datatype;
 
     if (req->message.header.request.magic != PROTOCOL_BINARY_REQ ||
         req->message.header.request.keylen == 0) {
         return PROTOCOL_BINARY_RESPONSE_EINVAL;
     }
 
-    // Check datatype - only allow raw, or iff XATTRs are enabled XATTR (with or
-    // without snappy)
-    if (!(mcbp::datatype::is_raw(datatype) ||
-          ((datatype == PROTOCOL_BINARY_DATATYPE_XATTR ||
-            datatype == (PROTOCOL_BINARY_DATATYPE_XATTR |
-                         PROTOCOL_BINARY_DATATYPE_SNAPPY)) &&
-           may_accept_xattr(cookie)))) {
+    if (!valid_dcp_delete_datatype(req->message.header.request.datatype)) {
         return PROTOCOL_BINARY_RESPONSE_EINVAL;
     }
 

engines/ep/src/dcp/consumer.cc

@@ -543,6 +543,16 @@ ENGINE_ERROR_CODE DcpConsumer::deletion(uint32_t opaque,
                                   revSeqno));
         item->setDeleted();
 
+        // MB-29040: Producer may send deleted doc with value that still has
+        // the user xattrs and the body. Fix up that mistake by running the
+        // expiry hook which will correctly process the document
+        if (mcbp::datatype::is_xattr(datatype) && value.size()) {
+            auto vb = engine_.getVBucket(vbucket);
+            if (vb) {
+                engine_.getKVBucket()->runPreExpiryHook(*vb, *item);
+            }
+        }
+
         std::unique_ptr<ExtendedMetaData> emd;
         if (meta.size() > 0) {
             emd = std::make_unique<ExtendedMetaData>(meta.data(), meta.size());

engines/ep/src/kv_bucket.cc

@@ -531,6 +531,22 @@ void KVBucket::getValue(Item& it) {
     }
 }
 
+void KVBucket::runPreExpiryHook(VBucket& vb, Item& it) {
+    it.decompressValue(); // A no-op for already decompressed items
+    auto info =
+            it.toItemInfo(vb.failovers->getLatestUUID(), vb.getHLCEpochSeqno());
+    if (engine.getServerApi()->document->pre_expiry(info)) {
+        // The payload is modified and contains data we should use
+        it.replaceValue(Blob::New(static_cast<char*>(info.value[0].iov_base),
+                                  info.value[0].iov_len));
+        it.setDataType(info.datatype);
+    } else {
+        // Make the document empty and raw
+        it.replaceValue(Blob::New(0));
+        it.setDataType(PROTOCOL_BINARY_RAW_BYTES);
+    }
+}
+
 void KVBucket::deleteExpiredItem(Item& it,
                                  time_t startTime,
                                  ExpireBy source) {
@@ -546,19 +562,7 @@ void KVBucket::deleteExpiredItem(Item& it,
         // Process positive seqnos (ignoring special *temp* items) and only
         // those items with a value
         if (it.getBySeqno() >= 0 && it.getNBytes()) {
-            auto info = it.toItemInfo(vb->failovers->getLatestUUID(),
-                                      vb->getHLCEpochSeqno());
-            if (engine.getServerApi()->document->pre_expiry(info)) {
-                // The payload is modified and contains data we should use
-                it.replaceValue(
-                        Blob::New(static_cast<char*>(info.value[0].iov_base),
-                                  info.value[0].iov_len));
-                it.setDataType(info.datatype);
-            } else {
-                // Make the document empty and raw
-                it.replaceValue(Blob::New(0));
-                it.setDataType(PROTOCOL_BINARY_RAW_BYTES);
-            }
+            runPreExpiryHook(*vb, it);
         }
 
         // Obtain reader access to the VB state change lock so that

engines/ep/src/kv_bucket.h

@@ -572,6 +572,18 @@ class KVBucket : public KVBucketIface {
                                              VBucket::id_type vbucket,
                                              const char** msg);
 
+    /**
+     * Run the server-api pre-expiry hook against the Item - the function
+     * may (if the pre-expiry hook dictates) mutate the Item value so that
+     * xattrs and the value are removed. The method doesn't care for the Item
+     * state (i.e. isDeleted) and the callers should be passing expired/deleted
+     * items only.
+     *
+     * @param vb The vbucket it belongs to
+     * @param it A reference to the Item to run the hook against and possibly
+     *        mutate.
+     */
+    void runPreExpiryHook(VBucket& vb, Item& it);
     void deleteExpiredItem(Item& it, time_t startTime, ExpireBy source);
     void deleteExpiredItems(std::list<Item>&, ExpireBy);
 

engines/ep/tests/module_tests/evp_store_single_threaded_test.cc

@@ -2521,6 +2521,102 @@ TEST_F(MB_29287, dataloss_hole) {
     EXPECT_EQ(1, vb->checkpointManager->getNumOfCursors());
 }
 
+class XattrCompressedTest
+    : public SingleThreadedEPBucketTest,
+      public ::testing::WithParamInterface<::testing::tuple<bool, bool>> {
+public:
+    bool isXattrSystem() const {
+        return ::testing::get<0>(GetParam());
+    }
+    bool isSnappy() const {
+        return ::testing::get<1>(GetParam());
+    }
+};
+
+// Create a replica VB and consumer, then send it an xattr value which should
+// of been stripped at the source, but wasn't because of MB29040. Then check
+// the consumer sanitises the document. Run the test with user/system xattrs
+// and snappy on/off
+TEST_P(XattrCompressedTest, MB_29040_sanitise_input) {
+    setVBucketStateAndRunPersistTask(vbid, vbucket_state_replica);
+
+    auto consumer = std::make_shared<MockDcpConsumer>(
+            *engine, cookie, "MB_29040_sanitise_input");
+    int opaque = 1;
+    ASSERT_EQ(ENGINE_SUCCESS, consumer->addStream(opaque, vbid, /*flags*/ 0));
+
+    std::string body;
+    if (!isXattrSystem()) {
+        body.assign("value");
+    }
+    auto value = createXattrValue(body, isXattrSystem(), isSnappy());
+
+    // Send deletion in a single seqno snapshot
+    int64_t bySeqno = 1;
+    EXPECT_EQ(ENGINE_SUCCESS,
+              consumer->snapshotMarker(
+                      opaque, vbid, bySeqno, bySeqno, MARKER_FLAG_CHK));
+
+    cb::const_byte_buffer valueBuf{
+            reinterpret_cast<const uint8_t*>(value.data()), value.size()};
+    EXPECT_EQ(
+            ENGINE_SUCCESS,
+            consumer->deletion(
+                    opaque,
+                    {"key", DocNamespace::DefaultCollection},
+                    valueBuf,
+                    /*priv_bytes*/ 0,
+                    PROTOCOL_BINARY_DATATYPE_XATTR |
+                            (isSnappy() ? PROTOCOL_BINARY_DATATYPE_SNAPPY : 0),
+                    /*cas*/ 3,
+                    vbid,
+                    bySeqno,
+                    /*revSeqno*/ 0,
+                    /*meta*/ {}));
+
+    EXPECT_EQ(std::make_pair(false, size_t(1)),
+              getEPBucket().flushVBucket(vbid));
+
+    ASSERT_EQ(ENGINE_SUCCESS, consumer->closeStream(opaque, vbid));
+
+    // Switch to active
+    setVBucketStateAndRunPersistTask(vbid, vbucket_state_active);
+
+    get_options_t options = static_cast<get_options_t>(
+            QUEUE_BG_FETCH | HONOR_STATES | TRACK_REFERENCE | DELETE_TEMP |
+            HIDE_LOCKED_CAS | TRACK_STATISTICS | GET_DELETED_VALUE);
+    auto gv = store->get(
+            {"key", DocNamespace::DefaultCollection}, vbid, cookie, options);
+    EXPECT_EQ(ENGINE_EWOULDBLOCK, gv.getStatus());
+
+    // Manually run the bgfetch task.
+    MockGlobalTask mockTask(engine->getTaskable(), TaskId::MultiBGFetcherTask);
+    store->getVBucket(vbid)->getShard()->getBgFetcher()->run(&mockTask);
+    gv = store->get({"key", DocNamespace::DefaultCollection},
+                    vbid,
+                    cookie,
+                    GET_DELETED_VALUE);
+    ASSERT_EQ(ENGINE_SUCCESS, gv.getStatus());
+
+    // This is the only system key test_helpers::createXattrValue gives us
+    cb::xattr::Blob blob;
+    blob.set("_sync", "{\"cas\":\"0xdeadbeefcafefeed\"}");
+
+    EXPECT_TRUE(gv.item->isDeleted());
+    EXPECT_EQ(0, gv.item->getFlags());
+    EXPECT_EQ(3, gv.item->getCas());
+    EXPECT_EQ(isXattrSystem() ? blob.size() : 0,
+              gv.item->getValue()->valueSize());
+    EXPECT_EQ(isXattrSystem() ? PROTOCOL_BINARY_DATATYPE_XATTR
+                              : PROTOCOL_BINARY_RAW_BYTES,
+              gv.item->getDataType());
+}
+
 INSTANTIATE_TEST_CASE_P(XattrSystemUserTest,
                         XattrSystemUserTest,
                         ::testing::Bool(), );
+
+INSTANTIATE_TEST_CASE_P(XattrCompressedTest,
+                        XattrCompressedTest,
+                        ::testing::Combine(::testing::Bool(),
+                                           ::testing::Bool()), );

tests/mcbp/mcbp_test.cc

@@ -1956,13 +1956,10 @@ TEST_P(DcpDeletionValidatorTest, ValidDatatype) {
 
 TEST_P(DcpDeletionValidatorTest, InvalidDatatype) {
     using cb::mcbp::Datatype;
-    const std::array<uint8_t, 5> datatypes = {
+    const std::array<uint8_t, 3> datatypes = {
             {uint8_t(Datatype::JSON),
              uint8_t(Datatype::Snappy),
-             uint8_t(Datatype::Snappy) | uint8_t(Datatype::JSON),
-             uint8_t(Datatype::Xattr) | uint8_t(Datatype::JSON),
-             (uint8_t(Datatype::Xattr) | uint8_t(Datatype::Snappy) |
-              uint8_t(Datatype::JSON))}};
+             uint8_t(Datatype::Snappy) | uint8_t(Datatype::JSON)}};
 
     for (auto invalid : datatypes) {
         header.request.datatype = invalid;