EaR: Configurable encryption support for Tlog mutations (#9394)

* EaR: Configurable encryption support for TLog mutations

Description

  diff-1 : Address review comments

Major changes includes:
1. Update the code involved in ensuring Tlog mutation encryption to be
compliant with "configurable encryption" feature.
2. Update ENABLE_CONFIGURABLE_ENCRYPTION flag to be 'true' by default
and BUGGIFY it.

Testing

devRunCorrectness - 100K

fdbclient/BackupAgentBase.actor.cpp

@@ -262,7 +262,7 @@ bool validTenantAccess(std::map<int64_t, TenantName>* tenantMap,
 	}
 	int64_t tenantId = TenantInfo::INVALID_TENANT;
 	if (m.isEncrypted()) {
-		tenantId = m.encryptionHeader()->cipherTextDetails.encryptDomainId;
+		tenantId = m.encryptDomainId();
 	} else {
 		tenantId = TenantAPI::extractTenantIdFromMutation(m);
 	}
@@ -383,12 +383,18 @@ ACTOR static Future<Void> decodeBackupLogValue(Arena* arena,
 			// Decrypt mutation ref if encrypted
 			if (logValue.isEncrypted()) {
 				encryptedLogValue = logValue;
-				state EncryptCipherDomainId domainId = logValue.encryptionHeader()->cipherTextDetails.encryptDomainId;
+				state EncryptCipherDomainId domainId = logValue.encryptDomainId();
 				Reference<AsyncVar<ClientDBInfo> const> dbInfo = cx->clientInfo;
 				try {
-					TextAndHeaderCipherKeys cipherKeys =
+					if (CLIENT_KNOBS->ENABLE_CONFIGURABLE_ENCRYPTION) {
-					    wait(getEncryptCipherKeys(dbInfo, *logValue.encryptionHeader(), BlobCipherMetrics::RESTORE));
+						TextAndHeaderCipherKeys cipherKeys = wait(getEncryptCipherKeys(
-					logValue = logValue.decrypt(cipherKeys, tempArena, BlobCipherMetrics::BACKUP);
+						    dbInfo, logValue.configurableEncryptionHeader(), BlobCipherMetrics::RESTORE));
+						logValue = logValue.decrypt(cipherKeys, tempArena, BlobCipherMetrics::RESTORE);
+					} else {
+						TextAndHeaderCipherKeys cipherKeys = wait(
+						    getEncryptCipherKeys(dbInfo, *logValue.encryptionHeader(), BlobCipherMetrics::RESTORE));
+						logValue = logValue.decrypt(cipherKeys, tempArena, BlobCipherMetrics::RESTORE);
+					}
 				} catch (Error& e) {
 					// It's possible a tenant was deleted and the encrypt key fetch failed
 					TraceEvent(SevWarnAlways, "MutationLogRestoreEncryptKeyFetchFailed")

fdbclient/BlobCipher.cpp

@@ -211,6 +211,10 @@ EncryptAuthTokenMode BlobCipherEncryptHeaderRef::getAuthTokenMode() const {
 	    flags);
 }
 
+EncryptCipherDomainId BlobCipherEncryptHeaderRef::getDomainId() const {
+	return std::visit([](auto& h) { return h.v1.cipherTextDetails.encryptDomainId; }, algoHeader);
+}
+
 void BlobCipherEncryptHeaderRef::validateEncryptionHeaderDetails(const BlobCipherDetails& textCipherDetails,
                                                                  const BlobCipherDetails& headerCipherDetails,
                                                                  const StringRef& ivRef) const {

fdbclient/ClientKnobs.cpp

@@ -24,6 +24,7 @@
 #include "fdbclient/Tenant.h"
 #include "flow/IRandom.h"
 #include "flow/UnitTest.h"
+#include "flow/flow.h"
 
 #define init(...) KNOB_FN(__VA_ARGS__, INIT_ATOMIC_KNOB, INIT_KNOB)(__VA_ARGS__)
 
@@ -304,7 +305,7 @@ void ClientKnobs::initialize(Randomize randomize) {
 	init( ENABLE_ENCRYPTION_CPU_TIME_LOGGING,       false );
 	init( SIMULATION_ENABLE_SNAPSHOT_ENCRYPTION_CHECKS,        true );
 	init( SIMULATION_EKP_TENANT_IDS_TO_DROP,         "-1" );
-	init( ENABLE_CONFIGURABLE_ENCRYPTION,           true );
+	init( ENABLE_CONFIGURABLE_ENCRYPTION,            true );
 	init( ENCRYPT_HEADER_FLAGS_VERSION,                 1 );
 	init( ENCRYPT_HEADER_AES_CTR_NO_AUTH_VERSION,       1 );
 	init( ENCRYPT_HEADER_AES_CTR_AES_CMAC_AUTH_VERSION, 1 );

fdbclient/include/fdbclient/BlobCipher.h

@@ -213,7 +213,7 @@ struct EncryptHeaderCipherDetails {
 	BlobCipherDetails textCipherDetails;
 	Optional<BlobCipherDetails> headerCipherDetails;
 
-	EncryptHeaderCipherDetails() {}
+	EncryptHeaderCipherDetails() = default;
 	EncryptHeaderCipherDetails(const BlobCipherDetails& tCipherDetails) : textCipherDetails(tCipherDetails) {}
 	EncryptHeaderCipherDetails(const BlobCipherDetails& tCipherDetails, const BlobCipherDetails& hCipherDetails)
 	  : textCipherDetails(tCipherDetails), headerCipherDetails(hCipherDetails) {}
@@ -494,6 +494,7 @@ struct BlobCipherEncryptHeaderRef {
 	const uint8_t* getIV() const;
 	const EncryptHeaderCipherDetails getCipherDetails() const;
 	EncryptAuthTokenMode getAuthTokenMode() const;
+	EncryptCipherDomainId getDomainId() const;
 
 	void validateEncryptionHeaderDetails(const BlobCipherDetails& textCipherDetails,
 	                                     const BlobCipherDetails& headerCipherDetails,

fdbclient/include/fdbclient/CommitTransaction.h

@@ -30,6 +30,8 @@
 #include "flow/EncryptUtils.h"
 #include "flow/Knobs.h"
 
+#include <unordered_set>
+
 // The versioned message has wire format : -1, version, messages
 static const int32_t VERSION_HEADER = -1;
 
@@ -143,6 +145,38 @@ struct MutationRef {
 		return reinterpret_cast<const BlobCipherEncryptHeader*>(param1.begin());
 	}
 
+	const BlobCipherEncryptHeaderRef configurableEncryptionHeader() const {
+		ASSERT(isEncrypted());
+		return BlobCipherEncryptHeaderRef::fromStringRef(param1);
+	}
+
+	EncryptCipherDomainId encryptDomainId() const {
+		ASSERT(isEncrypted());
+		return CLIENT_KNOBS->ENABLE_CONFIGURABLE_ENCRYPTION ? configurableEncryptionHeader().getDomainId()
+		                                                    : encryptionHeader()->cipherTextDetails.encryptDomainId;
+	}
+
+	void updateEncryptCipherDetails(std::unordered_set<BlobCipherDetails>& cipherDetails) {
+		ASSERT(isEncrypted());
+
+		if (CLIENT_KNOBS->ENABLE_CONFIGURABLE_ENCRYPTION) {
+			BlobCipherEncryptHeaderRef header = configurableEncryptionHeader();
+			EncryptHeaderCipherDetails details = header.getCipherDetails();
+			ASSERT(details.textCipherDetails.isValid());
+			cipherDetails.insert(details.textCipherDetails);
+			if (details.headerCipherDetails.present()) {
+				ASSERT(details.headerCipherDetails.get().isValid());
+				cipherDetails.insert(details.headerCipherDetails.get());
+			}
+		} else {
+			const BlobCipherEncryptHeader* header = encryptionHeader();
+			cipherDetails.insert(header->cipherTextDetails);
+			if (header->cipherHeaderDetails.isValid()) {
+				cipherDetails.insert(header->cipherHeaderDetails);
+			}
+		}
+	}
+
 	MutationRef encrypt(TextAndHeaderCipherKeys cipherKeys,
 	                    Arena& arena,
 	                    BlobCipherMetrics::UsageType usageType) const {
@@ -150,6 +184,7 @@ struct MutationRef {
 		deterministicRandom()->randomBytes(iv, AES_256_IV_LENGTH);
 		BinaryWriter bw(AssumeVersion(ProtocolVersion::withEncryptionAtRest()));
 		bw << *this;
+
 		EncryptBlobCipherAes265Ctr cipher(
 		    cipherKeys.cipherTextKey,
 		    cipherKeys.cipherHeaderKey,
@@ -157,11 +192,22 @@ struct MutationRef {
 		    AES_256_IV_LENGTH,
 		    getEncryptAuthTokenMode(EncryptAuthTokenMode::ENCRYPT_HEADER_AUTH_TOKEN_MODE_SINGLE),
 		    usageType);
-		BlobCipherEncryptHeader* header = new (arena) BlobCipherEncryptHeader;
+
-		StringRef headerRef(reinterpret_cast<const uint8_t*>(header), sizeof(BlobCipherEncryptHeader));
+		StringRef serializedHeader;
-		StringRef payload =
+		StringRef payload;
-		    cipher.encrypt(static_cast<const uint8_t*>(bw.getData()), bw.getLength(), header, arena)->toStringRef();
+		if (CLIENT_KNOBS->ENABLE_CONFIGURABLE_ENCRYPTION) {
-		return MutationRef(Encrypted, headerRef, payload);
+			BlobCipherEncryptHeaderRef header;
+			payload = cipher.encrypt(static_cast<const uint8_t*>(bw.getData()), bw.getLength(), &header, arena);
+			Standalone<StringRef> headerStr = BlobCipherEncryptHeaderRef::toStringRef(header);
+			arena.dependsOn(headerStr.arena());
+			serializedHeader = headerStr;
+		} else {
+			BlobCipherEncryptHeader* header = new (arena) BlobCipherEncryptHeader;
+			serializedHeader = StringRef(reinterpret_cast<const uint8_t*>(header), sizeof(BlobCipherEncryptHeader));
+			payload =
+			    cipher.encrypt(static_cast<const uint8_t*>(bw.getData()), bw.getLength(), header, arena)->toStringRef();
+		}
+		return MutationRef(Encrypted, serializedHeader, payload);
 	}
 
 	MutationRef encrypt(const std::unordered_map<EncryptCipherDomainId, Reference<BlobCipherKey>>& cipherKeys,
@@ -183,6 +229,7 @@ struct MutationRef {
 		deterministicRandom()->randomBytes(iv, AES_256_IV_LENGTH);
 		BinaryWriter bw(AssumeVersion(ProtocolVersion::withEncryptionAtRest()));
 		bw << *this;
+
 		EncryptBlobCipherAes265Ctr cipher(
 		    textCipherKey,
 		    headerCipherKey,
@@ -190,11 +237,22 @@ struct MutationRef {
 		    AES_256_IV_LENGTH,
 		    getEncryptAuthTokenMode(EncryptAuthTokenMode::ENCRYPT_HEADER_AUTH_TOKEN_MODE_SINGLE),
 		    usageType);
-		BlobCipherEncryptHeader* header = new (arena) BlobCipherEncryptHeader;
+
-		StringRef headerRef(reinterpret_cast<const uint8_t*>(header), sizeof(BlobCipherEncryptHeader));
+		if (CLIENT_KNOBS->ENABLE_CONFIGURABLE_ENCRYPTION) {
-		StringRef payload =
+			BlobCipherEncryptHeaderRef header;
-		    cipher.encrypt(static_cast<const uint8_t*>(bw.getData()), bw.getLength(), header, arena)->toStringRef();
+			StringRef payload =
-		return MutationRef(Encrypted, headerRef, payload);
+			    cipher.encrypt(static_cast<const uint8_t*>(bw.getData()), bw.getLength(), &header, arena);
+			Standalone<StringRef> serializedHeader = BlobCipherEncryptHeaderRef::toStringRef(header);
+			arena.dependsOn(serializedHeader.arena());
+			return MutationRef(Encrypted, serializedHeader, payload);
+		} else {
+			BlobCipherEncryptHeader* header = new (arena) BlobCipherEncryptHeader;
+			StringRef serializedHeader =
+			    StringRef(reinterpret_cast<const uint8_t*>(header), sizeof(BlobCipherEncryptHeader));
+			StringRef payload =
+			    cipher.encrypt(static_cast<const uint8_t*>(bw.getData()), bw.getLength(), header, arena)->toStringRef();
+			return MutationRef(Encrypted, serializedHeader, payload);
+		}
 	}
 
 	MutationRef encryptMetadata(const std::unordered_map<EncryptCipherDomainId, Reference<BlobCipherKey>>& cipherKeys,
@@ -207,9 +265,18 @@ struct MutationRef {
 	                    Arena& arena,
 	                    BlobCipherMetrics::UsageType usageType,
 	                    StringRef* buf = nullptr) const {
-		const BlobCipherEncryptHeader* header = encryptionHeader();
+		StringRef plaintext;
-		DecryptBlobCipherAes256Ctr cipher(cipherKeys.cipherTextKey, cipherKeys.cipherHeaderKey, header->iv, usageType);
+		if (CLIENT_KNOBS->ENABLE_CONFIGURABLE_ENCRYPTION) {
-		StringRef plaintext = cipher.decrypt(param2.begin(), param2.size(), *header, arena)->toStringRef();
+			const BlobCipherEncryptHeaderRef header = configurableEncryptionHeader();
+			DecryptBlobCipherAes256Ctr cipher(
+			    cipherKeys.cipherTextKey, cipherKeys.cipherHeaderKey, header.getIV(), usageType);
+			plaintext = cipher.decrypt(param2.begin(), param2.size(), header, arena);
+		} else {
+			const BlobCipherEncryptHeader* header = encryptionHeader();
+			DecryptBlobCipherAes256Ctr cipher(
+			    cipherKeys.cipherTextKey, cipherKeys.cipherHeaderKey, header->iv, usageType);
+			plaintext = cipher.decrypt(param2.begin(), param2.size(), *header, arena)->toStringRef();
+		}
 		if (buf != nullptr) {
 			*buf = plaintext;
 		}
@@ -229,7 +296,6 @@ struct MutationRef {
 
 	TextAndHeaderCipherKeys getCipherKeys(
 	    const std::unordered_map<BlobCipherDetails, Reference<BlobCipherKey>>& cipherKeys) const {
-		const BlobCipherEncryptHeader* header = encryptionHeader();
 		auto getCipherKey = [&](const BlobCipherDetails& details) -> Reference<BlobCipherKey> {
 			if (!details.isValid()) {
 				return {};
@@ -239,8 +305,22 @@ struct MutationRef {
 			return iter->second;
 		};
 		TextAndHeaderCipherKeys textAndHeaderKeys;
-		textAndHeaderKeys.cipherHeaderKey = getCipherKey(header->cipherHeaderDetails);
+		if (CLIENT_KNOBS->ENABLE_CONFIGURABLE_ENCRYPTION) {
-		textAndHeaderKeys.cipherTextKey = getCipherKey(header->cipherTextDetails);
+			const BlobCipherEncryptHeaderRef header = configurableEncryptionHeader();
+			EncryptHeaderCipherDetails cipherDetails = header.getCipherDetails();
+			ASSERT(cipherDetails.textCipherDetails.isValid());
+			textAndHeaderKeys.cipherTextKey = getCipherKey(cipherDetails.textCipherDetails);
+			if (cipherDetails.headerCipherDetails.present()) {
+				ASSERT(cipherDetails.headerCipherDetails.get().isValid());
+				textAndHeaderKeys.cipherHeaderKey = getCipherKey(cipherDetails.headerCipherDetails.get());
+			} else {
+				ASSERT(!FLOW_KNOBS->ENCRYPT_HEADER_AUTH_TOKEN_ENABLED);
+			}
+		} else {
+			const BlobCipherEncryptHeader* header = encryptionHeader();
+			textAndHeaderKeys.cipherHeaderKey = getCipherKey(header->cipherHeaderDetails);
+			textAndHeaderKeys.cipherTextKey = getCipherKey(header->cipherTextDetails);
+		}
 		return textAndHeaderKeys;
 	}
 

fdbclient/include/fdbclient/GetEncryptCipherKeys.actor.h

@@ -56,7 +56,7 @@ Future<Void> onEncryptKeyProxyChange(Reference<AsyncVar<T> const> db) {
 			break;
 		}
 	}
-	TraceEvent("GetEncryptCipherKeys_EncryptKeyProxyChanged")
+	TraceEvent("GetEncryptCipherKeysEncryptKeyProxyChanged")
 	    .detail("PreviousProxyId", previousProxyId.orDefault(UID()))
 	    .detail("CurrentProxyId", currentProxyId.orDefault(UID()));
 	return Void();
@@ -69,19 +69,19 @@ Future<EKPGetLatestBaseCipherKeysReply> getUncachedLatestEncryptCipherKeys(Refer
 	Optional<EncryptKeyProxyInterface> proxy = db->get().encryptKeyProxy;
 	if (!proxy.present()) {
 		// Wait for onEncryptKeyProxyChange.
-		TraceEvent("GetLatestEncryptCipherKeys_EncryptKeyProxyNotPresent").detail("UsageType", toString(usageType));
+		TraceEvent("GetLatestEncryptCipherKeysEncryptKeyProxyNotPresent").detail("UsageType", toString(usageType));
 		return Never();
 	}
 	request.reply.reset();
 	try {
 		EKPGetLatestBaseCipherKeysReply reply = wait(proxy.get().getLatestBaseCipherKeys.getReply(request));
 		if (reply.error.present()) {
-			TraceEvent(SevWarn, "GetLatestEncryptCipherKeys_RequestFailed").error(reply.error.get());
+			TraceEvent(SevWarn, "GetLatestEncryptCipherKeysRequestFailed").error(reply.error.get());
 			throw encrypt_keys_fetch_failed();
 		}
 		return reply;
 	} catch (Error& e) {
-		TraceEvent("GetLatestEncryptCipherKeys_CaughtError").error(e);
+		TraceEvent("GetLatestEncryptCipherKeysCaughtError").error(e);
 		if (e.code() == error_code_broken_promise) {
 			// Wait for onEncryptKeyProxyChange.
 			return Never();
@@ -103,7 +103,7 @@ Future<std::unordered_map<EncryptCipherDomainId, Reference<BlobCipherKey>>> getL
 	state EKPGetLatestBaseCipherKeysRequest request;
 
 	if (!db.isValid()) {
-		TraceEvent(SevError, "GetLatestEncryptCipherKeys_ServerDBInfoNotAvailable");
+		TraceEvent(SevError, "GetLatestEncryptCipherKeysServerDBInfoNotAvailable");
 		throw encrypt_ops_error();
 	}
 
@@ -140,7 +140,7 @@ Future<std::unordered_map<EncryptCipherDomainId, Reference<BlobCipherKey>>> getL
 			// Check for any missing cipher keys.
 			for (auto domainId : request.encryptDomainIds) {
 				if (cipherKeys.count(domainId) == 0) {
-					TraceEvent(SevWarn, "GetLatestEncryptCipherKeys_KeyMissing").detail("DomainId", domainId);
+					TraceEvent(SevWarn, "GetLatestEncryptCipherKeysKeyMissing").detail("DomainId", domainId);
 					throw encrypt_key_not_found();
 				}
 			}
@@ -176,14 +176,14 @@ Future<EKPGetBaseCipherKeysByIdsReply> getUncachedEncryptCipherKeys(Reference<As
 	Optional<EncryptKeyProxyInterface> proxy = db->get().encryptKeyProxy;
 	if (!proxy.present()) {
 		// Wait for onEncryptKeyProxyChange.
-		TraceEvent("GetEncryptCipherKeys_EncryptKeyProxyNotPresent").detail("UsageType", toString(usageType));
+		TraceEvent("GetEncryptCipherKeysEncryptKeyProxyNotPresent").detail("UsageType", toString(usageType));
 		return Never();
 	}
 	request.reply.reset();
 	try {
 		EKPGetBaseCipherKeysByIdsReply reply = wait(proxy.get().getBaseCipherKeysByIds.getReply(request));
 		if (reply.error.present()) {
-			TraceEvent(SevWarn, "GetEncryptCipherKeys_RequestFailed").error(reply.error.get());
+			TraceEvent(SevWarn, "GetEncryptCipherKeysRequestFailed").error(reply.error.get());
 			throw encrypt_keys_fetch_failed();
 		}
 		if (g_network && g_network->isSimulated() && usageType == BlobCipherMetrics::RESTORE) {
@@ -192,7 +192,7 @@ Future<EKPGetBaseCipherKeysByIdsReply> getUncachedEncryptCipherKeys(Reference<As
 			if (!tenantIdsToDrop.count(TenantInfo::INVALID_TENANT)) {
 				for (auto& baseCipherInfo : request.baseCipherInfos) {
 					if (tenantIdsToDrop.count(baseCipherInfo.domainId)) {
-						TraceEvent("GetEncryptCipherKeys_SimulatedError").detail("DomainId", baseCipherInfo.domainId);
+						TraceEvent("GetEncryptCipherKeysSimulatedError").detail("DomainId", baseCipherInfo.domainId);
 						throw encrypt_keys_fetch_failed();
 					}
 				}
@@ -200,7 +200,7 @@ Future<EKPGetBaseCipherKeysByIdsReply> getUncachedEncryptCipherKeys(Reference<As
 		}
 		return reply;
 	} catch (Error& e) {
-		TraceEvent("GetEncryptCipherKeys_CaughtError").error(e);
+		TraceEvent("GetEncryptCipherKeysCaughtError").error(e);
 		if (e.code() == error_code_broken_promise) {
 			// Wait for onEncryptKeyProxyChange.
 			return Never();
@@ -225,7 +225,7 @@ Future<std::unordered_map<BlobCipherDetails, Reference<BlobCipherKey>>> getEncry
 	state EKPGetBaseCipherKeysByIdsRequest request;
 
 	if (!db.isValid()) {
-		TraceEvent(SevError, "GetEncryptCipherKeys_ServerDBInfoNotAvailable");
+		TraceEvent(SevError, "GetEncryptCipherKeysServerDBInfoNotAvailable");
 		throw encrypt_ops_error();
 	}
 
@@ -262,7 +262,7 @@ Future<std::unordered_map<BlobCipherDetails, Reference<BlobCipherKey>>> getEncry
 				BaseCipherIndex baseIdx = std::make_pair(details.encryptDomainId, details.baseCipherId);
 				const auto& itr = baseCipherKeys.find(baseIdx);
 				if (itr == baseCipherKeys.end()) {
-					TraceEvent(SevError, "GetEncryptCipherKeys_KeyMissing")
+					TraceEvent(SevError, "GetEncryptCipherKeysKeyMissing")
 					    .detail("DomainId", details.encryptDomainId)
 					    .detail("BaseCipherId", details.baseCipherId);
 					throw encrypt_key_not_found();

fdbserver/BackupWorker.actor.cpp

@@ -105,11 +105,7 @@ struct VersionedMessage {
 			ArenaReader reader(arena, message, AssumeVersion(ProtocolVersion::withEncryptionAtRest()));
 			MutationRef m;
 			reader >> m;
-			const BlobCipherEncryptHeader* header = m.encryptionHeader();
+			m.updateEncryptCipherDetails(cipherDetails);
-			cipherDetails.insert(header->cipherTextDetails);
-			if (header->cipherHeaderDetails.isValid()) {
-				cipherDetails.insert(header->cipherHeaderDetails);
-			}
 		}
 	}
 };

fdbserver/CommitProxyServer.actor.cpp

@@ -1691,6 +1691,8 @@ ACTOR Future<WriteMutationRefVar> writeMutationEncryptedMutation(CommitBatchCont
                                                                  Arena* arena) {
 	state MutationRef encryptedMutation = encryptedMutationOpt->get();
 	state const BlobCipherEncryptHeader* header;
+	state BlobCipherEncryptHeaderRef headerRef;
+	state MutationRef decryptedMutation;
 
 	static_assert(TenantInfo::INVALID_TENANT == INVALID_ENCRYPT_DOMAIN_ID);
 	ASSERT(self->pProxyCommitData->encryptMode.isEncryptionEnabled());
@@ -1698,9 +1700,15 @@ ACTOR Future<WriteMutationRefVar> writeMutationEncryptedMutation(CommitBatchCont
 
 	ASSERT(encryptedMutation.isEncrypted());
 	Reference<AsyncVar<ServerDBInfo> const> dbInfo = self->pProxyCommitData->db;
-	header = encryptedMutation.encryptionHeader();
+	if (CLIENT_KNOBS->ENABLE_CONFIGURABLE_ENCRYPTION) {
-	TextAndHeaderCipherKeys cipherKeys = wait(getEncryptCipherKeys(dbInfo, *header, BlobCipherMetrics::TLOG));
+		headerRef = encryptedMutation.configurableEncryptionHeader();
-	MutationRef decryptedMutation = encryptedMutation.decrypt(cipherKeys, *arena, BlobCipherMetrics::TLOG);
+		TextAndHeaderCipherKeys cipherKeys = wait(getEncryptCipherKeys(dbInfo, headerRef, BlobCipherMetrics::TLOG));
+		decryptedMutation = encryptedMutation.decrypt(cipherKeys, *arena, BlobCipherMetrics::TLOG);
+	} else {
+		header = encryptedMutation.encryptionHeader();
+		TextAndHeaderCipherKeys cipherKeys = wait(getEncryptCipherKeys(dbInfo, *header, BlobCipherMetrics::TLOG));
+		decryptedMutation = encryptedMutation.decrypt(cipherKeys, *arena, BlobCipherMetrics::TLOG);
+	}
 
 	ASSERT(decryptedMutation.type == mutation->type);
 	ASSERT(decryptedMutation.param1 == mutation->param1);

fdbserver/RestoreLoader.actor.cpp

@@ -372,13 +372,10 @@ void handleRestoreSysInfoRequest(const RestoreSysInfoRequest& req, Reference<Res
 
 ACTOR static Future<MutationRef> _decryptMutation(MutationRef mutation, Database cx, Arena* arena) {
 	ASSERT(mutation.isEncrypted());
+
 	Reference<AsyncVar<ClientDBInfo> const> dbInfo = cx->clientInfo;
-	state const BlobCipherEncryptHeader* header = mutation.encryptionHeader();
 	std::unordered_set<BlobCipherDetails> cipherDetails;
-	cipherDetails.insert(header->cipherTextDetails);
+	mutation.updateEncryptCipherDetails(cipherDetails);
-	if (header->cipherHeaderDetails.isValid()) {
-		cipherDetails.insert(header->cipherHeaderDetails);
-	}
 	std::unordered_map<BlobCipherDetails, Reference<BlobCipherKey>> getCipherKeysResult =
 	    wait(getEncryptCipherKeys(dbInfo, cipherDetails, BlobCipherMetrics::BACKUP));
 	return mutation.decrypt(getCipherKeysResult, *arena, BlobCipherMetrics::BACKUP);

fdbserver/SimulatedCluster.actor.cpp

@@ -1302,6 +1302,7 @@ ACTOR Future<Void> restartSimulatedSystem(std::vector<Future<Void>>* systemActor
 			g_knobs.setKnob("enable_tlog_encryption", KnobValueRef::create(bool{ false }));
 			g_knobs.setKnob("enable_storage_server_encryption", KnobValueRef::create(bool{ false }));
 			g_knobs.setKnob("enable_blob_granule_encryption", KnobValueRef::create(bool{ false }));
+			g_knobs.setKnob("enable_configurable_encryption", KnobValueRef::create(bool{ false }));
 			TraceEvent(SevDebug, "DisableEncryption");
 		}
 		*pConnString = conn;

fdbserver/StorageCache.actor.cpp

@@ -1925,11 +1925,7 @@ ACTOR Future<Void> pullAsyncData(StorageCacheData* data) {
 						cloneReader >> msg;
 						if (msg.isEncrypted()) {
 							if (!cipherKeys.present()) {
-								const BlobCipherEncryptHeader* header = msg.encryptionHeader();
+								msg.updateEncryptCipherDetails(cipherDetails);
-								cipherDetails.insert(header->cipherTextDetails);
-								if (header->cipherHeaderDetails.isValid()) {
-									cipherDetails.insert(header->cipherHeaderDetails);
-								}
 								collectingCipherKeys = true;
 							} else {
 								msg = msg.decrypt(cipherKeys.get(), cloneReader.arena(), BlobCipherMetrics::TLOG);

fdbserver/fdbserver.actor.cpp

@@ -2270,6 +2270,11 @@ int main(int argc, char* argv[]) {
 				g_knobs.setKnob("encrypt_header_auth_token_algo",
 				                KnobValue::create((int)ini.GetLongValue(
 				                    "META", "encryptHeaderAuthTokenAlgo", FLOW_KNOBS->ENCRYPT_HEADER_AUTH_TOKEN_ALGO)));
+				g_knobs.setKnob("enable_configurable_encryption",
+				                KnobValue::create(ini.GetBoolValue("META",
+				                                                   "enableConfigurableEncryption",
+				                                                   CLIENT_KNOBS->ENABLE_CONFIGURABLE_ENCRYPTION)));
+
 				g_knobs.setKnob(
 				    "shard_encode_location_metadata",
 				    KnobValue::create(ini.GetBoolValue("META", "enableShardEncodeLocationMetadata", false)));

fdbserver/storageserver.actor.cpp

@@ -159,6 +159,7 @@ bool canReplyWith(Error e) {
 		return false;
 	}
 }
+
 } // namespace
 
 #define PERSIST_PREFIX "\xff\xff"
@@ -3020,11 +3021,7 @@ ACTOR Future<std::pair<ChangeFeedStreamReply, bool>> getChangeFeedMutations(Stor
 			if (doFilterMutations || !req.encrypted) {
 				for (auto& m : decodedMutations.back().first) {
 					if (m.isEncrypted()) {
-						const BlobCipherEncryptHeader* header = m.encryptionHeader();
+						m.updateEncryptCipherDetails(cipherDetails);
-						cipherDetails.insert(header->cipherTextDetails);
-						if (header->cipherHeaderDetails.isValid()) {
-							cipherDetails.insert(header->cipherHeaderDetails);
-						}
 					}
 				}
 			}
@@ -9257,11 +9254,7 @@ ACTOR Future<Void> update(StorageServer* data, bool* pReceivedUpdate) {
 					}
 					if (msg.isEncrypted()) {
 						if (!cipherKeys.present()) {
-							const BlobCipherEncryptHeader* header = msg.encryptionHeader();
+							msg.updateEncryptCipherDetails(cipherDetails);
-							cipherDetails.insert(header->cipherTextDetails);
-							if (header->cipherHeaderDetails.isValid()) {
-								cipherDetails.insert(header->cipherHeaderDetails);
-							}
 							collectingCipherKeys = true;
 						} else {
 							msg = msg.decrypt(cipherKeys.get(), eager.arena, BlobCipherMetrics::TLOG);

fdbserver/workloads/SaveAndKill.actor.cpp

@@ -85,6 +85,7 @@ struct SaveAndKillWorkload : TestWorkload {
 		ini.SetBoolValue("META", "enableStorageServerEncryption", SERVER_KNOBS->ENABLE_STORAGE_SERVER_ENCRYPTION);
 		ini.SetBoolValue("META", "enableBlobGranuleEncryption", SERVER_KNOBS->ENABLE_BLOB_GRANULE_ENCRYPTION);
 		ini.SetBoolValue("META", "enableShardEncodeLocationMetadata", SERVER_KNOBS->SHARD_ENCODE_LOCATION_METADATA);
+		ini.SetBoolValue("META", "enableConfigurableEncryption", CLIENT_KNOBS->ENABLE_CONFIGURABLE_ENCRYPTION);
 
 		ini.SetBoolValue("META", "encryptHeaderAuthTokenEnabled", FLOW_KNOBS->ENCRYPT_HEADER_AUTH_TOKEN_ENABLED);
 		ini.SetLongValue("META", "encryptHeaderAuthTokenAlgo", FLOW_KNOBS->ENCRYPT_HEADER_AUTH_TOKEN_ALGO);

tests/TestRunner/local_cluster.py

@@ -114,6 +114,7 @@ logdir = {logdir}
 {bg_knob_line}
 {encrypt_knob_line1}
 {encrypt_knob_line2}
+{encrypt_knob_line3}
 {tls_config}
 {authz_public_key_config}
 {custom_config}
@@ -256,6 +257,7 @@ logdir = {logdir}
             bg_knob_line = ""
             encrypt_knob_line1 = ""
             encrypt_knob_line2 = ""
+            encrypt_knob_line3 = ""
             if self.use_legacy_conf_syntax:
                 conf_template = conf_template.replace("-", "_")
             if self.blob_granules_enabled:
@@ -263,6 +265,7 @@ logdir = {logdir}
             if self.enable_encryption_at_rest:
                 encrypt_knob_line1 = "knob_enable_encryption=true"
                 encrypt_knob_line2 = "knob_kms_connector_type=FDBPerfKmsConnector"
+                encrypt_knob_line3 = "knob_enable_configurable_encryption=true"
             f.write(
                 conf_template.format(
                     etcdir=self.etc,
@@ -273,6 +276,7 @@ logdir = {logdir}
                     bg_knob_line=bg_knob_line,
                     encrypt_knob_line1=encrypt_knob_line1,
                     encrypt_knob_line2=encrypt_knob_line2,
+                    encrypt_knob_line3=encrypt_knob_line3,
                     tls_config=self.tls_conf_string(),
                     authz_public_key_config=self.authz_public_key_conf_string(),
                     optional_tls=":tls" if self.tls_config is not None else "",