mirror of
https://github.com/apple/foundationdb.git
synced 2025-06-02 19:25:52 +08:00
EaR - Misc fixes found using end-to-end integration testing (#9806)
* EaR - Misc fixes found using end-to-end integration testing Description Major changes proposed includes: 1. RESTClient filtering of trailing `/`(s) characters from input URI resource path 2. Avoid EKP exponential backup given RESTClient supports exponential backoffs retries for all retryable errors. 3. Memory allocation optimizations: 3.1. BaseCipher key management using Standalone semantics in KMSConnector interface endpoints 3.2. Optimize memcpy while looking encryption-keys in EKP endpoints 4. Avoid delay while starting EKP, given its criticality during cluster recovery. 5. Update BlobCipher to handle variable size BaseCipher buffer 6. Improved logging Testing Setup: 1. External KMS server to supply encryption keys (inhouse) 2. Create cluster with: cluster_aware & domain_aware config * Fix EncryptionOps test Description Testing * EaR - Misc fixes found using end-to-end integration testing Description Major changes: 1. Cleanup EKP driven exponential backup files. 2. Update EKP not to use #1. Testing * EaR - Misc fixes found using end-to-end integration testing Description Address review comments Testing * Fix AES 256 key length value Description Testing * Address review comments Description Testing
This commit is contained in:
Ata E Husain Bohra
GitHub
parent
3164cadc6f
commit
3f6fcada45
@ -39,7 +39,6 @@
|
||||
#include "flow/serialize.h"
|
||||
#include "flow/Trace.h"
|
||||
#include "flow/UnitTest.h"
|
||||
#include "flow/xxhash.h"
|
||||
|
||||
#include <chrono>
|
||||
#include <cstring>
|
||||
@ -57,8 +56,7 @@
|
||||
#include <io.h>
|
||||
#endif
|
||||
|
||||
#define BLOB_CIPHER_DEBUG false
|
||||
#define BLOB_CIPHER_SERIALIZATION_CHECKS false
|
||||
#define BLOB_CIPHER_DEBUG DEBUG_ENCRYPT_KEY_CIPHER
|
||||
|
||||
namespace {
|
||||
void validateEncryptHeaderFlagVersion(const int flagsVersion) {
|
||||
@ -377,7 +375,7 @@ BlobCipherKey::BlobCipherKey(const EncryptCipherDomainId& domainId,
|
||||
BlobCipherKey::BlobCipherKey(const EncryptCipherDomainId& domainId,
|
||||
const EncryptCipherBaseKeyId& baseCiphId,
|
||||
const uint8_t* baseCiph,
|
||||
int baseCiphLen,
|
||||
const int baseCiphLen,
|
||||
const EncryptCipherRandomSalt& salt,
|
||||
const int64_t refreshAt,
|
||||
const int64_t expireAt) {
|
||||
@ -386,15 +384,14 @@ BlobCipherKey::BlobCipherKey(const EncryptCipherDomainId& domainId,
|
||||
|
||||
void BlobCipherKey::initKey(const EncryptCipherDomainId& domainId,
|
||||
const uint8_t* baseCiph,
|
||||
int baseCiphLen,
|
||||
const int baseCiphLen,
|
||||
const EncryptCipherBaseKeyId& baseCiphId,
|
||||
const EncryptCipherRandomSalt& salt,
|
||||
const int64_t refreshAt,
|
||||
const int64_t expireAt) {
|
||||
// Set the base encryption key properties
|
||||
baseCipher = std::make_unique<uint8_t[]>(AES_256_KEY_LENGTH);
|
||||
memset(baseCipher.get(), 0, AES_256_KEY_LENGTH);
|
||||
memcpy(baseCipher.get(), baseCiph, std::min<int>(baseCiphLen, AES_256_KEY_LENGTH));
|
||||
baseCipher = std::make_unique<uint8_t[]>(baseCiphLen);
|
||||
memcpy(baseCipher.get(), baseCiph, baseCiphLen);
|
||||
baseCipherLen = baseCiphLen;
|
||||
baseCipherId = baseCiphId;
|
||||
// Set the encryption domain for the base encryption key
|
||||
@ -479,6 +476,7 @@ Reference<BlobCipherKey> BlobCipherKeyIdCache::insertBaseCipherKey(const Encrypt
|
||||
const int64_t refreshAt,
|
||||
const int64_t expireAt) {
|
||||
ASSERT_GT(baseCipherId, INVALID_ENCRYPT_CIPHER_KEY_ID);
|
||||
ASSERT_GT(baseCipherLen, 0);
|
||||
|
||||
// BaseCipherKeys are immutable, given the routine invocation updates 'latestCipher',
|
||||
// ensure no key-tampering is done
|
||||
@ -501,9 +499,17 @@ Reference<BlobCipherKey> BlobCipherKeyIdCache::insertBaseCipherKey(const Encrypt
|
||||
}
|
||||
}
|
||||
|
||||
// Logging only tracks newly inserted cipher in the cache
|
||||
// Approach limits the logging to two instances when new cipher gets added to the cache, two
|
||||
// possible scenarios could be:
|
||||
// 1. Cold start - cache getting warmed up
|
||||
// 2. New cipher - new Tenant and/or KMS driven key-rotation
|
||||
// Frequency of the log is governed by KMS driven `refreshAt` interval which is usually a long duration (days if
|
||||
// not months)
|
||||
TraceEvent(SevInfo, "BlobCipherKeyInsertBaseCipherKeyLatest")
|
||||
.detail("DomainId", domainId)
|
||||
.detail("BaseCipherId", baseCipherId)
|
||||
.detail("BaseCipherLen", baseCipherLen)
|
||||
.detail("RefreshAt", refreshAt)
|
||||
.detail("ExpireAt", expireAt);
|
||||
|
||||
@ -528,6 +534,7 @@ Reference<BlobCipherKey> BlobCipherKeyIdCache::insertBaseCipherKey(const Encrypt
|
||||
const int64_t expireAt) {
|
||||
ASSERT_NE(baseCipherId, INVALID_ENCRYPT_CIPHER_KEY_ID);
|
||||
ASSERT_NE(salt, INVALID_ENCRYPT_RANDOM_SALT);
|
||||
ASSERT_GT(baseCipherLen, 0);
|
||||
|
||||
BlobCipherKeyIdCacheKey cacheKey = getCacheKey(baseCipherId, salt);
|
||||
|
||||
@ -551,9 +558,16 @@ Reference<BlobCipherKey> BlobCipherKeyIdCache::insertBaseCipherKey(const Encrypt
|
||||
}
|
||||
}
|
||||
|
||||
// Logging only tracks newly inserted cipher in the cache
|
||||
// possible scenarios could be:
|
||||
// 1. Cold start - cache getting warmed up
|
||||
// 2. New cipher - new Tenant and/or KMS driven key-rotation
|
||||
// Frequency of the log is governed by KMS driven `refreshAt` interval which is usually a long duration (days if
|
||||
// not months)
|
||||
TraceEvent(SevInfo, "BlobCipherKeyInsertBaseCipherKey")
|
||||
.detail("DomainId", domainId)
|
||||
.detail("BaseCipherId", baseCipherId)
|
||||
.detail("BaseCipherLen", baseCipherLen)
|
||||
.detail("Salt", salt)
|
||||
.detail("RefreshAt", refreshAt)
|
||||
.detail("ExpireAt", expireAt);
|
||||
@ -1790,8 +1804,8 @@ public:
|
||||
const EncryptCipherBaseKeyId& kId,
|
||||
const int64_t rAt,
|
||||
const int64_t eAt)
|
||||
: domainId(dId), len(deterministicRandom()->randomInt(AES_256_KEY_LENGTH / 2, AES_256_KEY_LENGTH + 1)),
|
||||
keyId(kId), key(std::make_unique<uint8_t[]>(len)), refreshAt(rAt), expireAt(eAt) {
|
||||
: domainId(dId), len(deterministicRandom()->randomInt(4, 128)), keyId(kId), key(std::make_unique<uint8_t[]>(len)),
|
||||
refreshAt(rAt), expireAt(eAt) {
|
||||
deterministicRandom()->randomBytes(key.get(), len);
|
||||
}
|
||||
};
|
||||
|
||||
@ -36,7 +36,6 @@
|
||||
#include "flow/Trace.h"
|
||||
#include "flow/serialize.h"
|
||||
#include "flow/UnitTest.h"
|
||||
#include "flow/xxhash.h"
|
||||
|
||||
#include "fmt/format.h"
|
||||
|
||||
@ -1878,7 +1877,7 @@ const EncryptCipherBaseKeyId encryptBaseCipherId = deterministicRandom()->random
|
||||
const EncryptCipherRandomSalt encryptSalt = deterministicRandom()->randomUInt64();
|
||||
|
||||
Standalone<StringRef> getBaseCipher() {
|
||||
Standalone<StringRef> baseCipher = makeString(AES_256_KEY_LENGTH);
|
||||
Standalone<StringRef> baseCipher = makeString(deterministicRandom()->randomInt(4, 256));
|
||||
deterministicRandom()->randomBytes(mutateString(baseCipher), baseCipher.size());
|
||||
return baseCipher;
|
||||
}
|
||||
|
||||
@ -174,10 +174,11 @@ ACTOR Future<Reference<HTTP::Response>> doRequest_impl(Reference<RESTClient> cli
|
||||
statsPtr->requests_failed++;
|
||||
|
||||
// All errors in err are potentially retryable as well as certain HTTP response codes...
|
||||
bool retryable = err.present() || r->code == HTTP::HTTP_STATUS_CODE_INTERNAL_SERVER_ERROR ||
|
||||
r->code == HTTP::HTTP_STATUS_CODE_BAD_GATEWAY ||
|
||||
r->code == HTTP::HTTP_STATUS_CODE_SERVICE_UNAVAILABLE ||
|
||||
r->code == HTTP::HTTP_STATUS_CODE_TOO_MANY_REQUESTS;
|
||||
bool retryable =
|
||||
err.present() || r->code == HTTP::HTTP_STATUS_CODE_INTERNAL_SERVER_ERROR ||
|
||||
r->code == HTTP::HTTP_STATUS_CODE_BAD_GATEWAY || r->code == HTTP::HTTP_STATUS_CODE_BAD_GATEWAY ||
|
||||
r->code == HTTP::HTTP_STATUS_CODE_SERVICE_UNAVAILABLE ||
|
||||
r->code == HTTP::HTTP_STATUS_CODE_TOO_MANY_REQUESTS || r->code == HTTP::HTTP_STATUS_CODE_TIMEOUT;
|
||||
|
||||
// But only if our previous attempt was not the last allowable try.
|
||||
retryable = retryable && (thisTry < maxTries);
|
||||
|
||||
@ -591,7 +591,7 @@ public:
|
||||
BlobCipherKey(const EncryptCipherDomainId& domainId,
|
||||
const EncryptCipherBaseKeyId& baseCiphId,
|
||||
const uint8_t* baseCiph,
|
||||
int baseCiphLen,
|
||||
const int baseCiphLen,
|
||||
const int64_t refreshAt,
|
||||
int64_t expireAt);
|
||||
BlobCipherKey(const EncryptCipherDomainId& domainId,
|
||||
@ -659,7 +659,7 @@ private:
|
||||
|
||||
void initKey(const EncryptCipherDomainId& domainId,
|
||||
const uint8_t* baseCiph,
|
||||
int baseCiphLen,
|
||||
const int baseCiphLen,
|
||||
const EncryptCipherBaseKeyId& baseCiphId,
|
||||
const EncryptCipherRandomSalt& salt,
|
||||
const int64_t refreshAt,
|
||||
|
||||
@ -102,18 +102,17 @@ struct EKPBaseCipherDetails {
|
||||
constexpr static FileIdentifier file_identifier = 2149615;
|
||||
int64_t encryptDomainId;
|
||||
uint64_t baseCipherId;
|
||||
StringRef baseCipherKey;
|
||||
Standalone<StringRef> baseCipherKey;
|
||||
int64_t refreshAt;
|
||||
int64_t expireAt;
|
||||
|
||||
EKPBaseCipherDetails()
|
||||
: encryptDomainId(0), baseCipherId(0), baseCipherKey(StringRef()), refreshAt(0), expireAt(-1) {}
|
||||
explicit EKPBaseCipherDetails(int64_t dId, uint64_t id, StringRef key, Arena& arena)
|
||||
: encryptDomainId(dId), baseCipherId(id), baseCipherKey(StringRef(arena, key)),
|
||||
refreshAt(std::numeric_limits<int64_t>::max()), expireAt(std::numeric_limits<int64_t>::max()) {}
|
||||
explicit EKPBaseCipherDetails(int64_t dId, uint64_t id, StringRef key, Arena& arena, int64_t refAt, int64_t expAt)
|
||||
: encryptDomainId(dId), baseCipherId(id), baseCipherKey(StringRef(arena, key)), refreshAt(refAt),
|
||||
expireAt(expAt) {}
|
||||
: encryptDomainId(0), baseCipherId(0), baseCipherKey(Standalone<StringRef>()), refreshAt(0), expireAt(-1) {}
|
||||
explicit EKPBaseCipherDetails(int64_t dId, uint64_t id, Standalone<StringRef> key)
|
||||
: encryptDomainId(dId), baseCipherId(id), baseCipherKey(key), refreshAt(std::numeric_limits<int64_t>::max()),
|
||||
expireAt(std::numeric_limits<int64_t>::max()) {}
|
||||
explicit EKPBaseCipherDetails(int64_t dId, uint64_t id, Standalone<StringRef> key, int64_t refAt, int64_t expAt)
|
||||
: encryptDomainId(dId), baseCipherId(id), baseCipherKey(key), refreshAt(refAt), expireAt(expAt) {}
|
||||
|
||||
template <class Ar>
|
||||
void serialize(Ar& ar) {
|
||||
@ -123,7 +122,6 @@ struct EKPBaseCipherDetails {
|
||||
|
||||
struct EKPGetBaseCipherKeysByIdsReply {
|
||||
constexpr static FileIdentifier file_identifier = 9485259;
|
||||
Arena arena;
|
||||
std::vector<EKPBaseCipherDetails> baseCipherDetails;
|
||||
int numHits;
|
||||
Optional<Error> error;
|
||||
@ -132,7 +130,7 @@ struct EKPGetBaseCipherKeysByIdsReply {
|
||||
|
||||
template <class Ar>
|
||||
void serialize(Ar& ar) {
|
||||
serializer(ar, baseCipherDetails, numHits, error, arena);
|
||||
serializer(ar, baseCipherDetails, numHits, error);
|
||||
}
|
||||
};
|
||||
|
||||
@ -160,7 +158,6 @@ struct EKPGetBaseCipherKeysRequestInfo {
|
||||
|
||||
struct EKPGetBaseCipherKeysByIdsRequest {
|
||||
constexpr static FileIdentifier file_identifier = 4930263;
|
||||
Arena arena;
|
||||
std::vector<EKPGetBaseCipherKeysRequestInfo> baseCipherInfos;
|
||||
Optional<UID> debugId;
|
||||
ReplyPromise<EKPGetBaseCipherKeysByIdsReply> reply;
|
||||
@ -169,13 +166,12 @@ struct EKPGetBaseCipherKeysByIdsRequest {
|
||||
|
||||
template <class Ar>
|
||||
void serialize(Ar& ar) {
|
||||
serializer(ar, baseCipherInfos, debugId, reply, arena);
|
||||
serializer(ar, baseCipherInfos, debugId, reply);
|
||||
}
|
||||
};
|
||||
|
||||
struct EKPGetLatestBaseCipherKeysReply {
|
||||
constexpr static FileIdentifier file_identifier = 4831583;
|
||||
Arena arena;
|
||||
std::vector<EKPBaseCipherDetails> baseCipherDetails;
|
||||
int numHits;
|
||||
Optional<Error> error;
|
||||
@ -186,7 +182,7 @@ struct EKPGetLatestBaseCipherKeysReply {
|
||||
|
||||
template <class Ar>
|
||||
void serialize(Ar& ar) {
|
||||
serializer(ar, baseCipherDetails, numHits, error, arena);
|
||||
serializer(ar, baseCipherDetails, numHits, error);
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
@ -82,10 +82,12 @@ constexpr int HTTP_STATUS_CODE_ACCEPTED = 202;
|
||||
constexpr int HTTP_STATUS_CODE_NO_CONTENT = 204;
|
||||
constexpr int HTTP_STATUS_CODE_UNAUTHORIZED = 401;
|
||||
constexpr int HTTP_STATUS_CODE_NOT_ACCEPTABLE = 406;
|
||||
constexpr int HTTP_STATUS_CODE_TIMEOUT = 408;
|
||||
constexpr int HTTP_STATUS_CODE_TOO_MANY_REQUESTS = 429;
|
||||
constexpr int HTTP_STATUS_CODE_INTERNAL_SERVER_ERROR = 500;
|
||||
constexpr int HTTP_STATUS_CODE_BAD_GATEWAY = 502;
|
||||
constexpr int HTTP_STATUS_CODE_SERVICE_UNAVAILABLE = 503;
|
||||
constexpr int HTTP_STATUS_GATEWAY_TIMEOUT = 504;
|
||||
|
||||
constexpr int HTTP_RETRYAFTER_DELAY_SECS = 300;
|
||||
|
||||
|
||||
@ -2243,10 +2243,9 @@ ACTOR Future<Void> monitorConsistencyScan(ClusterControllerData* self) {
|
||||
}
|
||||
}
|
||||
|
||||
ACTOR Future<Void> startEncryptKeyProxy(ClusterControllerData* self, double waitTime) {
|
||||
ACTOR Future<Void> startEncryptKeyProxy(ClusterControllerData* self) {
|
||||
// If master fails at the same time, give it a chance to clear master PID.
|
||||
// Also wait to avoid too many consecutive recruits in a small time window.
|
||||
wait(delay(waitTime));
|
||||
wait(delay(0.0));
|
||||
|
||||
TraceEvent("CCEKP_Start", self->id).log();
|
||||
loop {
|
||||
@ -2327,18 +2326,21 @@ ACTOR Future<Void> monitorEncryptKeyProxy(ClusterControllerData* self) {
|
||||
state SingletonRecruitThrottler recruitThrottler;
|
||||
loop {
|
||||
if (self->db.serverInfo->get().encryptKeyProxy.present() && !self->recruitEncryptKeyProxy.get()) {
|
||||
choose {
|
||||
loop choose {
|
||||
when(wait(waitFailureClient(self->db.serverInfo->get().encryptKeyProxy.get().waitFailure,
|
||||
SERVER_KNOBS->ENCRYPT_KEY_PROXY_FAILURE_TIME))) {
|
||||
TraceEvent("CCEKP_Died", self->id);
|
||||
const auto& encryptKeyProxy = self->db.serverInfo->get().encryptKeyProxy;
|
||||
EncryptKeyProxySingleton(encryptKeyProxy).halt(*self, encryptKeyProxy.get().locality.processId());
|
||||
self->db.clearInterf(ProcessClass::EncryptKeyProxyClass);
|
||||
TraceEvent("CCEKP_Died", self->id);
|
||||
break;
|
||||
}
|
||||
when(wait(self->recruitEncryptKeyProxy.onChange())) {
|
||||
break;
|
||||
}
|
||||
when(wait(self->recruitEncryptKeyProxy.onChange())) {}
|
||||
}
|
||||
} else {
|
||||
wait(startEncryptKeyProxy(self, recruitThrottler.newRecruitment()));
|
||||
wait(startEncryptKeyProxy(self));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -22,7 +22,6 @@
|
||||
#include "fdbclient/EncryptKeyProxyInterface.h"
|
||||
|
||||
#include "fdbrpc/Locality.h"
|
||||
#include "fdbserver/EncryptKeyProxy.actor.h"
|
||||
#include "fdbserver/KmsConnector.h"
|
||||
#include "fdbserver/KmsConnectorInterface.h"
|
||||
#include "fdbserver/Knobs.h"
|
||||
@ -288,6 +287,10 @@ public:
|
||||
EncryptBaseCipherDomainIdKeyIdCacheKey cacheKey = getBaseCipherDomainIdKeyIdCacheKey(domainId, baseCipherId);
|
||||
baseCipherDomainIdKeyIdCache[cacheKey] =
|
||||
EncryptBaseCipherKey(domainId, baseCipherId, baseCipherKey, refreshAtTS, expireAtTS);
|
||||
TraceEvent("InsertIntoBaseCipherIdCache")
|
||||
.detail("DomId", domainId)
|
||||
.detail("BaseCipherId", baseCipherId)
|
||||
.detail("BaseCipherLen", baseCipherKey.size());
|
||||
}
|
||||
|
||||
void insertIntoBlobMetadataCache(const BlobMetadataDomainId domainId,
|
||||
@ -328,7 +331,7 @@ ACTOR Future<Void> getCipherKeysByBaseCipherKeyIds(Reference<EncryptKeyProxyData
|
||||
boost::hash<std::pair<EncryptCipherDomainId, EncryptCipherBaseKeyId>>>
|
||||
lookupCipherInfoMap;
|
||||
|
||||
state std::vector<EKPBaseCipherDetails> cachedCipherDetails;
|
||||
state int numHits = 0;
|
||||
state EKPGetBaseCipherKeysByIdsRequest keysByIds = req;
|
||||
state EKPGetBaseCipherKeysByIdsReply keyIdsReply;
|
||||
state Optional<TraceEvent> dbgTrace =
|
||||
@ -360,8 +363,9 @@ ACTOR Future<Void> getCipherKeysByBaseCipherKeyIds(Reference<EncryptKeyProxyData
|
||||
ekpProxyData->getBaseCipherDomainIdKeyIdCacheKey(item.domainId, item.baseCipherId);
|
||||
const auto itr = ekpProxyData->baseCipherDomainIdKeyIdCache.find(cacheKey);
|
||||
if (itr != ekpProxyData->baseCipherDomainIdKeyIdCache.end() && itr->second.isValid()) {
|
||||
cachedCipherDetails.emplace_back(
|
||||
itr->second.domainId, itr->second.baseCipherId, itr->second.baseCipherKey, keyIdsReply.arena);
|
||||
keyIdsReply.baseCipherDetails.emplace_back(
|
||||
itr->second.domainId, itr->second.baseCipherId, itr->second.baseCipherKey);
|
||||
numHits++;
|
||||
|
||||
if (dbgTrace.present()) {
|
||||
// {encryptId, baseCipherId} forms a unique tuple across encryption domains
|
||||
@ -375,7 +379,9 @@ ACTOR Future<Void> getCipherKeysByBaseCipherKeyIds(Reference<EncryptKeyProxyData
|
||||
}
|
||||
}
|
||||
|
||||
ekpProxyData->baseCipherKeyIdCacheHits += cachedCipherDetails.size();
|
||||
ASSERT_EQ(keyIdsReply.baseCipherDetails.size(), numHits);
|
||||
|
||||
ekpProxyData->baseCipherKeyIdCacheHits += numHits;
|
||||
ekpProxyData->baseCipherKeyIdCacheMisses += lookupCipherInfoMap.size();
|
||||
|
||||
if (!lookupCipherInfoMap.empty()) {
|
||||
@ -386,23 +392,11 @@ ACTOR Future<Void> getCipherKeysByBaseCipherKeyIds(Reference<EncryptKeyProxyData
|
||||
}
|
||||
keysByIdsReq.debugId = keysByIds.debugId;
|
||||
state double startTime = now();
|
||||
std::function<Future<KmsConnLookupEKsByKeyIdsRep>()> keysByIdsRepF = [&]() {
|
||||
return kmsConnectorInf.ekLookupByIds.getReply(keysByIdsReq);
|
||||
};
|
||||
std::function<void()> retryTrace = [&]() {
|
||||
for (const auto& item : keysByIdsReq.encryptKeyInfos) {
|
||||
TraceEvent(SevDebug, "GetCipherKeysByKeyIdsRetry")
|
||||
.suppressFor(30)
|
||||
.detail("DomainId", item.domainId);
|
||||
}
|
||||
};
|
||||
KmsConnLookupEKsByKeyIdsRep keysByIdsRep =
|
||||
wait(kmsReqWithExponentialBackoff(keysByIdsRepF, retryTrace, "GetCipherKeysByKeyIds"_sr));
|
||||
KmsConnLookupEKsByKeyIdsRep keysByIdsRep = wait(kmsConnectorInf.ekLookupByIds.getReply(keysByIdsReq));
|
||||
ekpProxyData->kmsLookupByIdsReqLatency.addMeasurement(now() - startTime);
|
||||
|
||||
for (const auto& item : keysByIdsRep.cipherKeyDetails) {
|
||||
keyIdsReply.baseCipherDetails.emplace_back(
|
||||
item.encryptDomainId, item.encryptKeyId, item.encryptKey, keyIdsReply.arena);
|
||||
keyIdsReply.baseCipherDetails.emplace_back(item.encryptDomainId, item.encryptKeyId, item.encryptKey);
|
||||
}
|
||||
|
||||
// Record the fetched cipher details to the local cache for the future references
|
||||
@ -448,11 +442,7 @@ ACTOR Future<Void> getCipherKeysByBaseCipherKeyIds(Reference<EncryptKeyProxyData
|
||||
}
|
||||
}
|
||||
|
||||
// Append cached cipherKeyDetails to the result-set
|
||||
keyIdsReply.baseCipherDetails.insert(
|
||||
keyIdsReply.baseCipherDetails.end(), cachedCipherDetails.begin(), cachedCipherDetails.end());
|
||||
|
||||
keyIdsReply.numHits = cachedCipherDetails.size();
|
||||
keyIdsReply.numHits = numHits;
|
||||
keysByIds.reply.send(keyIdsReply);
|
||||
|
||||
CODE_PROBE(!lookupCipherInfoMap.empty(), "EKP fetch cipherKeys by KeyId from KMS");
|
||||
@ -465,10 +455,9 @@ ACTOR Future<Void> getLatestCipherKeys(Reference<EncryptKeyProxyData> ekpProxyDa
|
||||
EKPGetLatestBaseCipherKeysRequest req) {
|
||||
// Scan the cached cipher-keys and filter our baseCipherIds locally cached
|
||||
// for the rest, reachout to KMS to fetch the required details
|
||||
state std::vector<EKPBaseCipherDetails> cachedCipherDetails;
|
||||
state int numHits = 0;
|
||||
state EKPGetLatestBaseCipherKeysRequest latestKeysReq = req;
|
||||
state EKPGetLatestBaseCipherKeysReply latestCipherReply;
|
||||
state Arena& arena = latestCipherReply.arena;
|
||||
state Optional<TraceEvent> dbgTrace =
|
||||
latestKeysReq.debugId.present() ? TraceEvent("GetByDomIds", ekpProxyData->myId) : Optional<TraceEvent>();
|
||||
|
||||
@ -499,12 +488,12 @@ ACTOR Future<Void> getLatestCipherKeys(Reference<EncryptKeyProxyData> ekpProxyDa
|
||||
for (const auto domainId : dedupedDomainIds) {
|
||||
const auto itr = ekpProxyData->baseCipherDomainIdCache.find(domainId);
|
||||
if (itr != ekpProxyData->baseCipherDomainIdCache.end() && itr->second.isValid()) {
|
||||
cachedCipherDetails.emplace_back(domainId,
|
||||
itr->second.baseCipherId,
|
||||
itr->second.baseCipherKey,
|
||||
arena,
|
||||
itr->second.refreshAt,
|
||||
itr->second.expireAt);
|
||||
latestCipherReply.baseCipherDetails.emplace_back(domainId,
|
||||
itr->second.baseCipherId,
|
||||
itr->second.baseCipherKey,
|
||||
itr->second.refreshAt,
|
||||
itr->second.expireAt);
|
||||
numHits++;
|
||||
|
||||
if (dbgTrace.present()) {
|
||||
// {encryptDomainId, baseCipherId} forms a unique tuple across encryption domains
|
||||
@ -520,7 +509,9 @@ ACTOR Future<Void> getLatestCipherKeys(Reference<EncryptKeyProxyData> ekpProxyDa
|
||||
}
|
||||
}
|
||||
|
||||
ekpProxyData->baseCipherDomainIdCacheHits += cachedCipherDetails.size();
|
||||
ASSERT_EQ(numHits, latestCipherReply.baseCipherDetails.size());
|
||||
|
||||
ekpProxyData->baseCipherDomainIdCacheHits += numHits;
|
||||
ekpProxyData->baseCipherDomainIdCacheMisses += lookupCipherDomainIds.size();
|
||||
|
||||
if (!lookupCipherDomainIds.empty()) {
|
||||
@ -532,16 +523,8 @@ ACTOR Future<Void> getLatestCipherKeys(Reference<EncryptKeyProxyData> ekpProxyDa
|
||||
keysByDomainIdReq.debugId = latestKeysReq.debugId;
|
||||
|
||||
state double startTime = now();
|
||||
std::function<Future<KmsConnLookupEKsByDomainIdsRep>()> keysByDomainIdRepF = [&]() {
|
||||
return kmsConnectorInf.ekLookupByDomainIds.getReply(keysByDomainIdReq);
|
||||
};
|
||||
std::function<void()> retryTrace = [&]() {
|
||||
for (const auto& item : keysByDomainIdReq.encryptDomainIds) {
|
||||
TraceEvent(SevDebug, "GetLatestCipherKeysRetry").suppressFor(30).detail("DomainId", item);
|
||||
}
|
||||
};
|
||||
KmsConnLookupEKsByDomainIdsRep keysByDomainIdRep =
|
||||
wait(kmsReqWithExponentialBackoff(keysByDomainIdRepF, retryTrace, "GetLatestCipherKeys"_sr));
|
||||
wait(kmsConnectorInf.ekLookupByDomainIds.getReply(keysByDomainIdReq));
|
||||
ekpProxyData->kmsLookupByDomainIdsReqLatency.addMeasurement(now() - startTime);
|
||||
|
||||
for (auto& item : keysByDomainIdRep.cipherKeyDetails) {
|
||||
@ -550,7 +533,6 @@ ACTOR Future<Void> getLatestCipherKeys(Reference<EncryptKeyProxyData> ekpProxyDa
|
||||
latestCipherReply.baseCipherDetails.emplace_back(item.encryptDomainId,
|
||||
item.encryptKeyId,
|
||||
item.encryptKey,
|
||||
arena,
|
||||
validityTS.refreshAtTS,
|
||||
validityTS.expAtTS);
|
||||
|
||||
@ -588,12 +570,7 @@ ACTOR Future<Void> getLatestCipherKeys(Reference<EncryptKeyProxyData> ekpProxyDa
|
||||
}
|
||||
}
|
||||
|
||||
for (auto& item : cachedCipherDetails) {
|
||||
latestCipherReply.baseCipherDetails.emplace_back(
|
||||
item.encryptDomainId, item.baseCipherId, item.baseCipherKey, arena);
|
||||
}
|
||||
|
||||
latestCipherReply.numHits = cachedCipherDetails.size();
|
||||
latestCipherReply.numHits = numHits;
|
||||
latestKeysReq.reply.send(latestCipherReply);
|
||||
|
||||
CODE_PROBE(!lookupCipherDomainIds.empty(), "EKP fetch latest cipherKeys from KMS");
|
||||
@ -657,15 +634,7 @@ ACTOR Future<Void> refreshEncryptionKeysImpl(Reference<EncryptKeyProxyData> ekpP
|
||||
}
|
||||
|
||||
state double startTime = now();
|
||||
std::function<Future<KmsConnLookupEKsByDomainIdsRep>()> repF = [&]() {
|
||||
return kmsConnectorInf.ekLookupByDomainIds.getReply(req);
|
||||
};
|
||||
std::function<void()> retryTrace = [&]() {
|
||||
for (const auto& item : req.encryptDomainIds) {
|
||||
TraceEvent(SevDebug, "RefreshEKsRetry").suppressFor(30).detail("DomainId", item);
|
||||
}
|
||||
};
|
||||
KmsConnLookupEKsByDomainIdsRep rep = wait(kmsReqWithExponentialBackoff(repF, retryTrace, "RefreshEKs"_sr));
|
||||
KmsConnLookupEKsByDomainIdsRep rep = wait(kmsConnectorInf.ekLookupByDomainIds.getReply(req));
|
||||
ekpProxyData->kmsLookupByDomainIdsReqLatency.addMeasurement(now() - startTime);
|
||||
for (const auto& item : rep.cipherKeyDetails) {
|
||||
const auto itr = ekpProxyData->baseCipherDomainIdCache.find(item.encryptDomainId);
|
||||
@ -761,16 +730,7 @@ ACTOR Future<Void> getLatestBlobMetadata(Reference<EncryptKeyProxyData> ekpProxy
|
||||
ekpProxyData->blobMetadataCacheMisses += kmsReq.domainIds.size();
|
||||
try {
|
||||
state double startTime = now();
|
||||
std::function<Future<KmsConnBlobMetadataRep>()> kmsRepF = [&]() {
|
||||
return kmsConnectorInf.blobMetadataReq.getReply(kmsReq);
|
||||
};
|
||||
std::function<void()> retryTrace = [&]() {
|
||||
for (const auto& item : kmsReq.domainIds) {
|
||||
TraceEvent(SevDebug, "GetLatestBlobMetadataRetry").suppressFor(30).detail("DomainId", item);
|
||||
}
|
||||
};
|
||||
KmsConnBlobMetadataRep kmsRep =
|
||||
wait(kmsReqWithExponentialBackoff(kmsRepF, retryTrace, "GetLatestBlobMetadata"_sr));
|
||||
KmsConnBlobMetadataRep kmsRep = wait(kmsConnectorInf.blobMetadataReq.getReply(kmsReq));
|
||||
ekpProxyData->kmsBlobMetadataReqLatency.addMeasurement(now() - startTime);
|
||||
metadataDetails.arena().dependsOn(kmsRep.metadataDetails.arena());
|
||||
|
||||
@ -833,15 +793,7 @@ ACTOR Future<Void> refreshBlobMetadataCore(Reference<EncryptKeyProxyData> ekpPro
|
||||
}
|
||||
|
||||
startTime = now();
|
||||
std::function<Future<KmsConnBlobMetadataRep>()> repF = [&]() {
|
||||
return kmsConnectorInf.blobMetadataReq.getReply(req);
|
||||
};
|
||||
std::function<void()> retryTrace = [&]() {
|
||||
for (const auto& item : req.domainIds) {
|
||||
TraceEvent(SevDebug, "RefreshBlobMetadataRetry").suppressFor(30).detail("DomainId", item);
|
||||
}
|
||||
};
|
||||
KmsConnBlobMetadataRep rep = wait(kmsReqWithExponentialBackoff(repF, retryTrace, "RefreshBlobMetadata"_sr));
|
||||
KmsConnBlobMetadataRep rep = wait(kmsConnectorInf.blobMetadataReq.getReply(req));
|
||||
ekpProxyData->kmsBlobMetadataReqLatency.addMeasurement(now() - startTime);
|
||||
for (auto& item : rep.metadataDetails) {
|
||||
ekpProxyData->insertIntoBlobMetadataCache(item.domainId, item);
|
||||
@ -941,48 +893,3 @@ ACTOR Future<Void> encryptKeyProxyServer(EncryptKeyProxyInterface ekpInterface,
|
||||
|
||||
return Void();
|
||||
}
|
||||
|
||||
TEST_CASE("/EncryptKeyProxy/ExponentialBackoff") {
|
||||
state int callCount = 0;
|
||||
state bool alwaysThrow = false;
|
||||
state std::vector<Error> errors = { encrypt_keys_fetch_failed(), timed_out(), connection_failed() };
|
||||
state std::function<Future<bool>()> repF = [&]() {
|
||||
if (callCount > 1 && !alwaysThrow) {
|
||||
return true;
|
||||
}
|
||||
callCount += 1;
|
||||
throw deterministicRandom()->randomChoice(errors);
|
||||
};
|
||||
state std::function<void()> retryTrace = [&]() {};
|
||||
bool resp = wait(kmsReqWithExponentialBackoff(repF, retryTrace, "TestEKPBackoff1"_sr));
|
||||
ASSERT(resp);
|
||||
ASSERT_EQ(callCount, 2);
|
||||
|
||||
// Exceeding retry limit should result in failure
|
||||
IKnobCollection::getMutableGlobalKnobCollection().setKnob("ekp_kms_connection_retries",
|
||||
KnobValueRef::create(int{ 3 }));
|
||||
callCount = 0;
|
||||
alwaysThrow = true;
|
||||
errors = { timed_out() };
|
||||
try {
|
||||
wait(success(kmsReqWithExponentialBackoff(repF, retryTrace, "TestEKPBackoff2"_sr)));
|
||||
ASSERT(false);
|
||||
} catch (Error& e) {
|
||||
ASSERT_EQ(e.code(), error_code_timed_out);
|
||||
ASSERT_EQ(callCount, FLOW_KNOBS->EKP_KMS_CONNECTION_RETRIES + 1);
|
||||
}
|
||||
|
||||
// A non-retryable error should throw immediately
|
||||
callCount = 0;
|
||||
alwaysThrow = false;
|
||||
errors = { encrypt_key_not_found() };
|
||||
try {
|
||||
wait(success(kmsReqWithExponentialBackoff(repF, retryTrace, "TestEKPBackoff3"_sr)));
|
||||
ASSERT(false);
|
||||
} catch (Error& e) {
|
||||
ASSERT_EQ(e.code(), error_code_encrypt_key_not_found);
|
||||
ASSERT_EQ(callCount, 1);
|
||||
}
|
||||
|
||||
return Void();
|
||||
}
|
||||
|
||||
@ -19,6 +19,7 @@
|
||||
*/
|
||||
#include "fdbserver/IPager.h"
|
||||
|
||||
#include "fdbclient/BlobCipher.h"
|
||||
#include "flow/IRandom.h"
|
||||
#include "flow/UnitTest.h"
|
||||
#include <limits>
|
||||
@ -33,13 +34,14 @@ TEST_CASE("/fdbserver/IPager/ArenaPage/PageContentChecksum") {
|
||||
deterministicRandom()->randomBytes(page->mutateData(), page->dataSize());
|
||||
PhysicalPageID pageID = deterministicRandom()->randomUInt32();
|
||||
if (encodingType == AESEncryption || encodingType == AESEncryptionWithAuth) {
|
||||
uint8_t cipherKeyBytes[AES_256_KEY_LENGTH];
|
||||
deterministicRandom()->randomBytes(cipherKeyBytes, AES_256_KEY_LENGTH);
|
||||
const int cipherBytesLen = deterministicRandom()->randomInt(4, (4 * AES_256_KEY_LENGTH) + 1);
|
||||
uint8_t cipherKeyBytes[cipherBytesLen];
|
||||
deterministicRandom()->randomBytes(cipherKeyBytes, cipherBytesLen);
|
||||
Reference<BlobCipherKey> cipherKey =
|
||||
makeReference<BlobCipherKey>(0 /*domainId*/,
|
||||
1 /*baseCipherId*/,
|
||||
cipherKeyBytes,
|
||||
AES_256_KEY_LENGTH,
|
||||
&cipherKeyBytes[0],
|
||||
cipherBytesLen,
|
||||
std::numeric_limits<int64_t>::max() /*refreshAt*/,
|
||||
std::numeric_limits<int64_t>::max() /*expireAt*/
|
||||
);
|
||||
|
||||
@ -20,6 +20,7 @@
|
||||
|
||||
#include "fdbserver/RESTKmsConnector.h"
|
||||
|
||||
#include "fdbclient/BlobCipher.h"
|
||||
#include "fdbclient/FDBTypes.h"
|
||||
#include "fdbclient/RESTClient.h"
|
||||
|
||||
@ -259,12 +260,16 @@ ACTOR Future<Void> parseDiscoverKmsUrlFile(Reference<RESTKmsConnectorCtx> ctx, s
|
||||
std::string url;
|
||||
while (std::getline(ss, url, DISCOVER_URL_FILE_URL_SEP)) {
|
||||
std::string trimedUrl = boost::trim_copy(url);
|
||||
// Remove the trailing '/'(s)
|
||||
while (!trimedUrl.empty() && trimedUrl.ends_with('/')) {
|
||||
trimedUrl.pop_back();
|
||||
}
|
||||
if (trimedUrl.empty()) {
|
||||
// Empty URL, ignore and continue
|
||||
continue;
|
||||
}
|
||||
TraceEvent("RESTParseDiscoverKmsUrlsAddUrl", ctx->uid).detail("Url", url);
|
||||
ctx->kmsUrlHeap.emplace(std::make_shared<KmsUrlCtx>(url));
|
||||
TraceEvent("RESTParseDiscoverKmsUrlsAddUrl", ctx->uid).detail("OrgUrl", url).detail("TrimUrl", trimedUrl);
|
||||
ctx->kmsUrlHeap.emplace(std::make_shared<KmsUrlCtx>(trimedUrl));
|
||||
}
|
||||
|
||||
return Void();
|
||||
@ -451,6 +456,7 @@ Standalone<VectorRef<EncryptCipherKeyDetailsRef>> parseEncryptCipherResponse(Ref
|
||||
TraceEvent event("RESTParseEncryptCipherResponse", ctx->uid);
|
||||
event.detail("DomainId", domainId);
|
||||
event.detail("BaseCipherId", baseCipherId);
|
||||
event.detail("BaseCipherLen", cipher.size());
|
||||
if (refreshAfterSec.present()) {
|
||||
event.detail("RefreshAt", refreshAfterSec.get());
|
||||
}
|
||||
@ -826,10 +832,10 @@ ACTOR Future<Void> fetchEncryptionKeysByKeyIds(Reference<RESTKmsConnectorCtx> ct
|
||||
std::function<Standalone<VectorRef<EncryptCipherKeyDetailsRef>>(Reference<RESTKmsConnectorCtx>,
|
||||
Reference<HTTP::Response>)>
|
||||
f = &parseEncryptCipherResponse;
|
||||
Standalone<VectorRef<EncryptCipherKeyDetailsRef>> result = wait(kmsRequestImpl(
|
||||
ctx, SERVER_KNOBS->REST_KMS_CONNECTOR_GET_ENCRYPTION_KEYS_ENDPOINT, requestBodyRef, std::move(f)));
|
||||
reply.cipherKeyDetails = result;
|
||||
reply.arena.dependsOn(result.arena());
|
||||
wait(store(
|
||||
reply.cipherKeyDetails,
|
||||
kmsRequestImpl(
|
||||
ctx, SERVER_KNOBS->REST_KMS_CONNECTOR_GET_ENCRYPTION_KEYS_ENDPOINT, requestBodyRef, std::move(f))));
|
||||
req.reply.send(reply);
|
||||
} catch (Error& e) {
|
||||
TraceEvent("RESTLookupEKsByKeyIdsFailed", ctx->uid).error(e);
|
||||
@ -909,10 +915,11 @@ ACTOR Future<Void> fetchEncryptionKeysByDomainIds(Reference<RESTKmsConnectorCtx>
|
||||
Reference<HTTP::Response>)>
|
||||
f = &parseEncryptCipherResponse;
|
||||
|
||||
Standalone<VectorRef<EncryptCipherKeyDetailsRef>> result = wait(kmsRequestImpl(
|
||||
ctx, SERVER_KNOBS->REST_KMS_CONNECTOR_GET_LATEST_ENCRYPTION_KEYS_ENDPOINT, requestBodyRef, std::move(f)));
|
||||
reply.cipherKeyDetails = result;
|
||||
reply.arena.dependsOn(result.arena());
|
||||
wait(store(reply.cipherKeyDetails,
|
||||
kmsRequestImpl(ctx,
|
||||
SERVER_KNOBS->REST_KMS_CONNECTOR_GET_LATEST_ENCRYPTION_KEYS_ENDPOINT,
|
||||
requestBodyRef,
|
||||
std::move(f))));
|
||||
req.reply.send(reply);
|
||||
} catch (Error& e) {
|
||||
TraceEvent("RESTLookupEKsByDomainIdsFailed", ctx->uid).error(e);
|
||||
@ -1017,7 +1024,7 @@ ACTOR Future<Void> procureValidationTokensFromFiles(Reference<RESTKmsConnectorCt
|
||||
TraceEvent("RESTValidationToken", ctx->uid).detail("DetailsStr", details);
|
||||
|
||||
state std::unordered_map<std::string, std::string> tokenFilePathMap;
|
||||
while (!details.empty()) {
|
||||
loop {
|
||||
StringRef name = detailsRef.eat(TOKEN_NAME_FILE_SEP);
|
||||
if (name.empty()) {
|
||||
break;
|
||||
@ -1772,9 +1779,18 @@ ACTOR Future<Void> testParseDiscoverKmsUrlFile(Reference<RESTKmsConnectorCtx> ct
|
||||
ASSERT(fileExists(tmpFile->getFileName()));
|
||||
|
||||
state std::unordered_set<std::string> urls;
|
||||
urls.emplace("https://127.0.0.1/foo");
|
||||
urls.emplace("https://127.0.0.1/foo1");
|
||||
urls.emplace("https://127.0.0.1/foo2");
|
||||
urls.emplace("https://127.0.0.1/foo ");
|
||||
urls.emplace(" https://127.0.0.1/foo1");
|
||||
urls.emplace(" https://127.0.0.1/foo2 ");
|
||||
urls.emplace("https://127.0.0.1/foo3/");
|
||||
urls.emplace("https://127.0.0.1/foo4///");
|
||||
|
||||
state std::unordered_set<std::string> compareUrls;
|
||||
compareUrls.emplace("https://127.0.0.1/foo");
|
||||
compareUrls.emplace("https://127.0.0.1/foo1");
|
||||
compareUrls.emplace("https://127.0.0.1/foo2");
|
||||
compareUrls.emplace("https://127.0.0.1/foo3");
|
||||
compareUrls.emplace("https://127.0.0.1/foo4");
|
||||
|
||||
std::string content;
|
||||
for (auto& url : urls) {
|
||||
@ -1789,7 +1805,7 @@ ACTOR Future<Void> testParseDiscoverKmsUrlFile(Reference<RESTKmsConnectorCtx> ct
|
||||
std::shared_ptr<KmsUrlCtx> urlCtx = ctx->kmsUrlHeap.top();
|
||||
ctx->kmsUrlHeap.pop();
|
||||
|
||||
ASSERT(urls.find(urlCtx->url) != urls.end());
|
||||
ASSERT(compareUrls.find(urlCtx->url) != compareUrls.end());
|
||||
ASSERT_EQ(urlCtx->nFailedResponses, 0);
|
||||
ASSERT_EQ(urlCtx->nRequests, 0);
|
||||
ASSERT_EQ(urlCtx->nResponseParseFailures, 0);
|
||||
|
||||
@ -29,6 +29,7 @@
|
||||
|
||||
#include "flow/ActorCollection.h"
|
||||
#include "flow/Arena.h"
|
||||
#include "flow/CodeProbe.h"
|
||||
#include "flow/EncryptUtils.h"
|
||||
#include "flow/Error.h"
|
||||
#include "flow/FastRef.h"
|
||||
@ -42,6 +43,7 @@
|
||||
|
||||
#include "fmt/format.h"
|
||||
|
||||
#include <cstring>
|
||||
#include <limits>
|
||||
#include <memory>
|
||||
#include <unordered_map>
|
||||
@ -49,12 +51,36 @@
|
||||
|
||||
#include "flow/actorcompiler.h" // This must be the last #include.
|
||||
|
||||
#define DEBUG_SIM_KEY_CIPHER DEBUG_ENCRYPT_KEY_CIPHER
|
||||
|
||||
using SimEncryptKey = std::string;
|
||||
struct SimEncryptKeyCtx {
|
||||
static const int minCipherLen = 4;
|
||||
static const int maxCipherLen = 256;
|
||||
|
||||
EncryptCipherBaseKeyId id;
|
||||
SimEncryptKey key;
|
||||
int keyLen;
|
||||
|
||||
explicit SimEncryptKeyCtx(EncryptCipherBaseKeyId kId, const char* data) : id(kId), key(data, AES_256_KEY_LENGTH) {}
|
||||
explicit SimEncryptKeyCtx(EncryptCipherBaseKeyId kId, const char* data, const int dataLen)
|
||||
: id(kId), key(data, dataLen), keyLen(dataLen) {
|
||||
if (DEBUG_SIM_KEY_CIPHER) {
|
||||
TraceEvent(SevDebug, "SimKmsKeyCtxInit").detail("BaseCipherId", kId).detail("BaseCipherLen", dataLen);
|
||||
}
|
||||
}
|
||||
|
||||
static int getKeyLen(const EncryptCipherBaseKeyId id) {
|
||||
ASSERT_GT(id, INVALID_ENCRYPT_CIPHER_KEY_ID);
|
||||
|
||||
int ret = AES_256_KEY_LENGTH;
|
||||
if ((id % 2) == 0) {
|
||||
ret += (id % AES_256_KEY_LENGTH);
|
||||
}
|
||||
CODE_PROBE(ret == AES_256_KEY_LENGTH, "BaseCipherKeyLen AES_256_KEY_LENGTH");
|
||||
CODE_PROBE(ret != AES_256_KEY_LENGTH, "BaseCipherKeyLen variable length");
|
||||
|
||||
return ret;
|
||||
}
|
||||
};
|
||||
|
||||
// The credentials may be allowed to change, but the storage locations and partitioning cannot change, even across
|
||||
@ -71,14 +97,26 @@ struct SimKmsConnectorContext : NonCopyable, ReferenceCounted<SimKmsConnectorCon
|
||||
// Construct encryption keyStore.
|
||||
// Note the keys generated must be the same after restart.
|
||||
for (int i = 1; i <= maxEncryptionKeys; i++) {
|
||||
const int keyLen = SimEncryptKeyCtx::getKeyLen(i);
|
||||
uint8_t key[keyLen];
|
||||
uint8_t digest[AUTH_TOKEN_HMAC_SHA_SIZE];
|
||||
|
||||
// TODO: Allow baseCipherKeyLen < AES_256_KEY_LENGTH
|
||||
ASSERT_EQ(AES_256_KEY_LENGTH, AUTH_TOKEN_HMAC_SHA_SIZE);
|
||||
computeAuthToken({ { reinterpret_cast<const uint8_t*>(&i), sizeof(i) } },
|
||||
SHA_KEY,
|
||||
AES_256_KEY_LENGTH,
|
||||
&digest[0],
|
||||
EncryptAuthTokenAlgo::ENCRYPT_HEADER_AUTH_TOKEN_ALGO_HMAC_SHA,
|
||||
AUTH_TOKEN_HMAC_SHA_SIZE);
|
||||
simEncryptKeyStore[i] = std::make_unique<SimEncryptKeyCtx>(i, reinterpret_cast<const char*>(&digest[0]));
|
||||
memcpy(&key[0], &digest[0], std::min(keyLen, AUTH_TOKEN_HMAC_SHA_SIZE));
|
||||
// Simulate variable length 'baseCipher' returned by external KMS
|
||||
if (keyLen > AUTH_TOKEN_HMAC_SHA_SIZE) {
|
||||
// pad it with known value
|
||||
memset(&key[AUTH_TOKEN_HMAC_SHA_SIZE], 'b', (keyLen - AUTH_TOKEN_HMAC_SHA_SIZE));
|
||||
}
|
||||
simEncryptKeyStore[i] =
|
||||
std::make_unique<SimEncryptKeyCtx>(i, reinterpret_cast<const char*>(&key[0]), keyLen);
|
||||
}
|
||||
}
|
||||
};
|
||||
@ -134,6 +172,14 @@ ACTOR Future<Void> ekLookupByIds(Reference<SimKmsConnectorContext> ctx,
|
||||
dbgKIdTrace.get().detail(
|
||||
getEncryptDbgTraceKey(ENCRYPT_DBG_TRACE_RESULT_PREFIX, item.domainId.get(), itr->first), "");
|
||||
}
|
||||
|
||||
if (DEBUG_SIM_KEY_CIPHER) {
|
||||
TraceEvent("SimKmsEKLookupByKeyId")
|
||||
.detail("DomId", item.domainId.get())
|
||||
.detail("BaseCipherId", item.baseCipherId)
|
||||
.detail("BaseCipherLen", itr->second->keyLen)
|
||||
.detail("BaseCipherKeyLen", itr->second->keyLen);
|
||||
}
|
||||
} else {
|
||||
TraceEvent("SimKmsEKLookupByIdsKeyNotFound")
|
||||
.detail("DomId", item.domainId)
|
||||
@ -185,6 +231,14 @@ ACTOR Future<Void> ekLookupByDomainIds(Reference<SimKmsConnectorContext> ctx,
|
||||
// {encryptId, baseCipherId} forms a unique tuple across encryption domains
|
||||
dbgDIdTrace.get().detail(getEncryptDbgTraceKey(ENCRYPT_DBG_TRACE_RESULT_PREFIX, domainId, keyId), "");
|
||||
}
|
||||
if (DEBUG_SIM_KEY_CIPHER) {
|
||||
TraceEvent("SimKmsEKLookupByDomainId")
|
||||
.detail("DomId", domainId)
|
||||
.detail("BaseCipherId", itr->second->id)
|
||||
.detail("BaseCipherLen", itr->second->keyLen)
|
||||
.detail("BaseCipherKeyLen", itr->second->keyLen);
|
||||
}
|
||||
|
||||
} else {
|
||||
TraceEvent("SimKmsEKLookupByDomainIdKeyNotFound").detail("DomId", domainId);
|
||||
success = false;
|
||||
@ -294,9 +348,10 @@ ACTOR Future<Void> testRunWorkload(KmsConnectorInterface inf, uint32_t nEncrypti
|
||||
}
|
||||
KmsConnLookupEKsByDomainIdsRep domainIdsRep = wait(inf.ekLookupByDomainIds.getReply(domainIdsReq));
|
||||
for (auto& element : domainIdsRep.cipherKeyDetails) {
|
||||
domainIdKeyMap.emplace(
|
||||
element.encryptDomainId,
|
||||
std::make_unique<SimEncryptKeyCtx>(element.encryptKeyId, element.encryptKey.toString().c_str()));
|
||||
domainIdKeyMap.emplace(element.encryptDomainId,
|
||||
std::make_unique<SimEncryptKeyCtx>(element.encryptKeyId,
|
||||
element.encryptKey.toString().c_str(),
|
||||
element.encryptKey.size()));
|
||||
}
|
||||
|
||||
// randomly pick any domainId and validate if lookupByKeyId result matches
|
||||
|
||||
@ -1,81 +0,0 @@
|
||||
/*
|
||||
* EncryptKeyProxy.actor.h
|
||||
*
|
||||
* This source file is part of the FoundationDB open source project
|
||||
*
|
||||
* Copyright 2013-2022 Apple Inc. and the FoundationDB project authors
|
||||
*
|
||||
* Licensed 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.
|
||||
*/
|
||||
|
||||
#pragma once
|
||||
|
||||
// When actually compiled (NO_INTELLISENSE), include the generated version of this file. In intellisense use the source
|
||||
// version.
|
||||
#if defined(NO_INTELLISENSE) && !defined(ENCRYPT_KEY_PROXY_ACTOR_G_H)
|
||||
#define ENCRYPT_KEY_PROXY_ACTOR_G_H
|
||||
#include "fdbserver/EncryptKeyProxy.actor.g.h"
|
||||
#elif !defined(ENCRYPT_KEY_PROXY_ACTOR_H)
|
||||
#define ENCRYPT_KEY_PROXY_ACTOR_H
|
||||
#include "flow/flow.h"
|
||||
#include "flow/actorcompiler.h" // This must be the last #include.
|
||||
|
||||
namespace EncryptKeyProxy {
|
||||
|
||||
inline bool canRetryWith(Error e) {
|
||||
// The below are the only errors that should be retried, all others should throw immediately
|
||||
switch (e.code()) {
|
||||
case error_code_encrypt_keys_fetch_failed:
|
||||
case error_code_timed_out:
|
||||
case error_code_connection_failed:
|
||||
return true;
|
||||
default:
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
} // namespace EncryptKeyProxy
|
||||
|
||||
ACTOR template <class T>
|
||||
Future<T> kmsReqWithExponentialBackoff(std::function<Future<T>()> func,
|
||||
std::function<void()> retryTrace,
|
||||
StringRef funcName) {
|
||||
state int numRetries = 0;
|
||||
state double kmsBackoff = FLOW_KNOBS->EKP_KMS_CONNECTION_BACKOFF;
|
||||
TraceEvent(SevDebug, "KMSRequestStart").suppressFor(30).detail("Function", funcName);
|
||||
|
||||
loop {
|
||||
try {
|
||||
T val = wait(func());
|
||||
return val;
|
||||
} catch (Error& e) {
|
||||
TraceEvent("KMSRequestReceivedError").detail("Function", funcName).detail("ErrorCode", e.code());
|
||||
if (!EncryptKeyProxy::canRetryWith(e)) {
|
||||
throw e;
|
||||
}
|
||||
if (numRetries >= FLOW_KNOBS->EKP_KMS_CONNECTION_RETRIES) {
|
||||
TraceEvent(SevWarnAlways, "KMSRequestRetryLimitExceeded")
|
||||
.detail("Function", funcName)
|
||||
.detail("ErrorCode", e.code());
|
||||
throw e;
|
||||
}
|
||||
retryTrace();
|
||||
numRetries++;
|
||||
wait(delay(kmsBackoff));
|
||||
kmsBackoff = kmsBackoff * 2; // exponential backoff
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#include "flow/unactorcompiler.h"
|
||||
#endif
|
||||
@ -75,16 +75,15 @@ struct EncryptCipherKeyDetailsRef {
|
||||
Optional<int64_t> refreshAfterSec;
|
||||
Optional<int64_t> expireAfterSec;
|
||||
|
||||
EncryptCipherKeyDetailsRef() {}
|
||||
EncryptCipherKeyDetailsRef()
|
||||
: encryptDomainId(INVALID_ENCRYPT_DOMAIN_ID), encryptKeyId(INVALID_ENCRYPT_CIPHER_KEY_ID),
|
||||
encryptKey(StringRef()) {}
|
||||
explicit EncryptCipherKeyDetailsRef(Arena& arena,
|
||||
EncryptCipherDomainId dId,
|
||||
EncryptCipherBaseKeyId keyId,
|
||||
StringRef key)
|
||||
: encryptDomainId(dId), encryptKeyId(keyId), encryptKey(StringRef(arena, key)),
|
||||
refreshAfterSec(Optional<int64_t>()), expireAfterSec(Optional<int64_t>()) {}
|
||||
explicit EncryptCipherKeyDetailsRef(EncryptCipherDomainId dId, EncryptCipherBaseKeyId keyId, StringRef key)
|
||||
: encryptDomainId(dId), encryptKeyId(keyId), encryptKey(key), refreshAfterSec(Optional<int64_t>()),
|
||||
expireAfterSec(Optional<int64_t>()) {}
|
||||
explicit EncryptCipherKeyDetailsRef(Arena& arena,
|
||||
EncryptCipherDomainId dId,
|
||||
EncryptCipherBaseKeyId keyId,
|
||||
@ -93,13 +92,6 @@ struct EncryptCipherKeyDetailsRef {
|
||||
Optional<int64_t> expAfterSec)
|
||||
: encryptDomainId(dId), encryptKeyId(keyId), encryptKey(StringRef(arena, key)), refreshAfterSec(refAfterSec),
|
||||
expireAfterSec(expAfterSec) {}
|
||||
explicit EncryptCipherKeyDetailsRef(EncryptCipherDomainId dId,
|
||||
EncryptCipherBaseKeyId keyId,
|
||||
StringRef key,
|
||||
Optional<int64_t> refAfterSec,
|
||||
Optional<int64_t> expAfterSec)
|
||||
: encryptDomainId(dId), encryptKeyId(keyId), encryptKey(key), refreshAfterSec(refAfterSec),
|
||||
expireAfterSec(expAfterSec) {}
|
||||
|
||||
bool operator==(const EncryptCipherKeyDetailsRef& toCompare) {
|
||||
return encryptDomainId == toCompare.encryptDomainId && encryptKeyId == toCompare.encryptKeyId &&
|
||||
@ -115,7 +107,7 @@ struct EncryptCipherKeyDetailsRef {
|
||||
struct KmsConnLookupEKsByKeyIdsRep {
|
||||
constexpr static FileIdentifier file_identifier = 2313778;
|
||||
Arena arena;
|
||||
VectorRef<EncryptCipherKeyDetailsRef> cipherKeyDetails;
|
||||
Standalone<VectorRef<EncryptCipherKeyDetailsRef>> cipherKeyDetails;
|
||||
|
||||
KmsConnLookupEKsByKeyIdsRep() {}
|
||||
|
||||
@ -172,7 +164,7 @@ struct KmsConnLookupEKsByKeyIdsReq {
|
||||
struct KmsConnLookupEKsByDomainIdsRep {
|
||||
constexpr static FileIdentifier file_identifier = 3009025;
|
||||
Arena arena;
|
||||
VectorRef<EncryptCipherKeyDetailsRef> cipherKeyDetails;
|
||||
Standalone<VectorRef<EncryptCipherKeyDetailsRef>> cipherKeyDetails;
|
||||
|
||||
KmsConnLookupEKsByDomainIdsRep() {}
|
||||
|
||||
|
||||
@ -2499,6 +2499,10 @@ ACTOR Future<Void> workerServer(Reference<IClusterConnectionRecord> connRecord,
|
||||
} else {
|
||||
startRole(Role::ENCRYPT_KEY_PROXY, recruited.id(), interf.id());
|
||||
DUMPTOKEN(recruited.waitFailure);
|
||||
DUMPTOKEN(recruited.haltEncryptKeyProxy);
|
||||
DUMPTOKEN(recruited.getBaseCipherKeysByIds);
|
||||
DUMPTOKEN(recruited.getLatestBaseCipherKeys);
|
||||
DUMPTOKEN(recruited.getLatestBlobMetadata);
|
||||
|
||||
Future<Void> encryptKeyProxyProcess = encryptKeyProxyServer(recruited, dbInfo);
|
||||
errorForwarders.add(forwardError(
|
||||
|
||||
@ -119,6 +119,8 @@ struct EncryptionOpsWorkload : TestWorkload {
|
||||
int maxBufSize;
|
||||
std::unique_ptr<uint8_t[]> buff;
|
||||
int enableTTLTest;
|
||||
int minBaseCipherLen;
|
||||
int maxBaseCipherLen;
|
||||
|
||||
std::unique_ptr<WorkloadMetrics> metrics;
|
||||
|
||||
@ -143,6 +145,11 @@ struct EncryptionOpsWorkload : TestWorkload {
|
||||
|
||||
metrics = std::make_unique<WorkloadMetrics>();
|
||||
|
||||
minBaseCipherLen = deterministicRandom()->randomInt(4, 11);
|
||||
maxBaseCipherLen = deterministicRandom()->randomInt(AES_256_KEY_LENGTH, (4 * AES_256_KEY_LENGTH) + 1);
|
||||
|
||||
ASSERT_LT(minBaseCipherLen, maxBaseCipherLen);
|
||||
|
||||
if (wcx.clientId == 0 && mode == 1) {
|
||||
enableTTLTest = true;
|
||||
}
|
||||
@ -168,9 +175,9 @@ struct EncryptionOpsWorkload : TestWorkload {
|
||||
throw internal_error();
|
||||
}
|
||||
|
||||
static void generateRandomBaseCipher(const int maxLen, uint8_t* buff, int* retLen) {
|
||||
static void generateRandomBaseCipher(const int minLen, const int maxLen, uint8_t* buff, int* retLen) {
|
||||
memset(buff, 0, maxLen);
|
||||
*retLen = deterministicRandom()->randomInt(maxLen / 2, maxLen);
|
||||
*retLen = deterministicRandom()->randomInt(minLen, maxLen);
|
||||
deterministicRandom()->randomBytes(buff, *retLen);
|
||||
}
|
||||
|
||||
@ -179,11 +186,12 @@ struct EncryptionOpsWorkload : TestWorkload {
|
||||
|
||||
TraceEvent("SetupCipherEssentialsStart").detail("MinDomainId", minDomainId).detail("MaxDomainId", maxDomainId);
|
||||
|
||||
uint8_t buff[AES_256_KEY_LENGTH];
|
||||
uint8_t buff[maxBaseCipherLen];
|
||||
std::vector<Reference<BlobCipherKey>> cipherKeys;
|
||||
int cipherLen = 0;
|
||||
for (EncryptCipherDomainId id = minDomainId; id <= maxDomainId; id++) {
|
||||
generateRandomBaseCipher(AES_256_KEY_LENGTH, &buff[0], &cipherLen);
|
||||
generateRandomBaseCipher(minBaseCipherLen, maxBaseCipherLen, &buff[0], &cipherLen);
|
||||
|
||||
cipherKeyCache->insertCipherKey(id,
|
||||
minBaseCipherId,
|
||||
buff,
|
||||
@ -191,7 +199,7 @@ struct EncryptionOpsWorkload : TestWorkload {
|
||||
std::numeric_limits<int64_t>::max(),
|
||||
std::numeric_limits<int64_t>::max());
|
||||
|
||||
ASSERT(cipherLen > 0 && cipherLen <= AES_256_KEY_LENGTH);
|
||||
ASSERT(cipherLen > 0 && cipherLen <= maxBaseCipherLen);
|
||||
|
||||
cipherKeys = cipherKeyCache->getAllCiphers(id);
|
||||
ASSERT_EQ(cipherKeys.size(), 1);
|
||||
@ -199,7 +207,7 @@ struct EncryptionOpsWorkload : TestWorkload {
|
||||
|
||||
// insert the Encrypt Header cipherKey; record cipherDetails as getLatestCipher() may not work with multiple
|
||||
// test clients
|
||||
generateRandomBaseCipher(AES_256_KEY_LENGTH, &buff[0], &cipherLen);
|
||||
generateRandomBaseCipher(minBaseCipherLen, maxBaseCipherLen, &buff[0], &cipherLen);
|
||||
cipherKeyCache->insertCipherKey(ENCRYPT_HEADER_DOMAIN_ID,
|
||||
headerBaseCipherId,
|
||||
buff,
|
||||
@ -241,10 +249,15 @@ struct EncryptionOpsWorkload : TestWorkload {
|
||||
Reference<BlobCipherKey> cipherKey = cipherKeyCache->getLatestCipherKey(encryptDomainId);
|
||||
*nextBaseCipherId = cipherKey->getBaseCipherId() + 1;
|
||||
|
||||
generateRandomBaseCipher(AES_256_KEY_LENGTH, baseCipher, baseCipherLen);
|
||||
generateRandomBaseCipher(minBaseCipherLen, maxBaseCipherLen, baseCipher, baseCipherLen);
|
||||
|
||||
ASSERT(*baseCipherLen > 0 && *baseCipherLen <= AES_256_KEY_LENGTH);
|
||||
TraceEvent("UpdateBaseCipher").detail("DomainId", encryptDomainId).detail("BaseCipherId", *nextBaseCipherId);
|
||||
ASSERT(*baseCipherLen > 0 && *baseCipherLen <= maxBaseCipherLen);
|
||||
TraceEvent("UpdateBaseCipher")
|
||||
.detail("DomainId", encryptDomainId)
|
||||
.detail("BaseCipherId", *nextBaseCipherId)
|
||||
.detail("BaseCipherLen", *baseCipherLen)
|
||||
.detail("ExistingBaseCipherId", cipherKey->getBaseCipherId())
|
||||
.detail("ExistingBaseCipherLen", cipherKey->getBaseCipherLen());
|
||||
}
|
||||
|
||||
Reference<BlobCipherKey> getEncryptionKey(const EncryptCipherDomainId& domainId,
|
||||
@ -431,7 +444,7 @@ struct EncryptionOpsWorkload : TestWorkload {
|
||||
}
|
||||
|
||||
void testBlobCipherKeyCacheOps() {
|
||||
uint8_t baseCipher[AES_256_KEY_LENGTH];
|
||||
uint8_t baseCipher[maxBaseCipherLen];
|
||||
int baseCipherLen = 0;
|
||||
EncryptCipherBaseKeyId nextBaseCipherId;
|
||||
|
||||
|
||||
@ -313,10 +313,6 @@ void FlowKnobs::initialize(Randomize randomize, IsSimulated isSimulated) {
|
||||
// Refer to EncryptUtil::EncryptAuthTokenAlgo for more details
|
||||
init( ENCRYPT_HEADER_AUTH_TOKEN_ENABLED, false ); if ( randomize && BUGGIFY ) { ENCRYPT_HEADER_AUTH_TOKEN_ENABLED = !ENCRYPT_HEADER_AUTH_TOKEN_ENABLED; }
|
||||
init( ENCRYPT_HEADER_AUTH_TOKEN_ALGO, 0 ); if ( randomize && ENCRYPT_HEADER_AUTH_TOKEN_ENABLED ) { ENCRYPT_HEADER_AUTH_TOKEN_ALGO = getRandomAuthTokenAlgo(); }
|
||||
// start exponential backoff at 5s when reaching out to the KMS from EKP
|
||||
init( EKP_KMS_CONNECTION_BACKOFF, 5.0 );
|
||||
// number of times to retry KMS requests from the EKP (roughly attempt to reach out to the KMS for a total of 5 minutes)
|
||||
init( EKP_KMS_CONNECTION_RETRIES, 6 );
|
||||
init( ENCRYPT_INPLACE_ENABLED, false ); if ( randomize && BUGGIFY ) { ENCRYPT_INPLACE_ENABLED = true; }
|
||||
|
||||
// REST Client
|
||||
|
||||
@ -23,6 +23,7 @@
|
||||
#pragma once
|
||||
|
||||
#include "flow/Arena.h"
|
||||
#include "flow/xxhash.h"
|
||||
|
||||
#include <cstdint>
|
||||
#include <limits>
|
||||
@ -30,6 +31,8 @@
|
||||
#include <string_view>
|
||||
#include <unordered_set>
|
||||
|
||||
#define DEBUG_ENCRYPT_KEY_CIPHER false
|
||||
|
||||
constexpr const int AUTH_TOKEN_HMAC_SHA_SIZE = 32;
|
||||
constexpr const int AUTH_TOKEN_AES_CMAC_SIZE = 16;
|
||||
constexpr const int AUTH_TOKEN_MAX_SIZE = AUTH_TOKEN_HMAC_SHA_SIZE;
|
||||
|
||||
@ -375,8 +375,6 @@ public:
|
||||
double ENCRYPT_KEY_CACHE_LOGGING_SKETCH_ACCURACY;
|
||||
bool ENCRYPT_HEADER_AUTH_TOKEN_ENABLED;
|
||||
int ENCRYPT_HEADER_AUTH_TOKEN_ALGO;
|
||||
double EKP_KMS_CONNECTION_BACKOFF;
|
||||
int EKP_KMS_CONNECTION_RETRIES;
|
||||
bool ENCRYPT_INPLACE_ENABLED; // Encrypt the page inplace
|
||||
|
||||
// RESTClient
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user