Audit all AuthZ token usages

fdbrpc/TokenCache.actor.cpp

@@ -8,6 +8,7 @@
 #include "flow/network.h"
 
 #include <boost/unordered_map.hpp>
+#include <boost/unordered_set.hpp>
 
 #include <fmt/format.h>
 #include <list>
@@ -123,20 +124,65 @@ TEST_CASE("/fdbrpc/authz/LRUCache") {
 	return Void();
 }
 
-struct TokenCacheImpl {
-	struct CacheEntry {
-		Arena arena;
-		VectorRef<TenantNameRef> tenants;
-		double expirationTime = 0.0;
-	};
+struct CacheEntry {
+	Arena arena;
+	VectorRef<TenantNameRef> tenants;
+	Optional<StringRef> tokenId;
+	double expirationTime = 0.0;
+};
 
+struct AuditEntry {
+	NetworkAddress address;
+	Optional<Standalone<StringRef>> tokenId;
+	explicit AuditEntry(NetworkAddress const& address, CacheEntry const& cacheEntry)
+	  : address(address),
+	    tokenId(cacheEntry.tokenId.present() ? Standalone<StringRef>(cacheEntry.tokenId.get(), cacheEntry.arena)
+	                                         : Optional<Standalone<StringRef>>()) {}
+};
+
+bool operator==(AuditEntry const& lhs, AuditEntry const& rhs) {
+	return (lhs.address == rhs.address) && (lhs.tokenId.present() == rhs.tokenId.present()) &&
+	       (!lhs.tokenId.present() || lhs.tokenId.get() == rhs.tokenId.get());
+}
+
+std::size_t hash_value(AuditEntry const& value) {
+	std::size_t seed = 0;
+	boost::hash_combine(seed, value.address);
+	if (value.tokenId.present()) {
+		boost::hash_combine(seed, value.tokenId.get());
+	}
+	return seed;
+}
+
+struct TokenCacheImpl {
 	LRUCache<StringRef, CacheEntry> cache;
-	TokenCacheImpl() : cache(FLOW_KNOBS->TOKEN_CACHE_SIZE) {}
+	boost::unordered_set<AuditEntry> usedTokens;
+	Future<Void> auditor;
+	TokenCacheImpl();
 
 	bool validate(TenantNameRef tenant, StringRef token);
 	bool validateAndAdd(double currentTime, StringRef token, NetworkAddress const& peer);
 };
 
+ACTOR Future<Void> tokenCacheAudit(TokenCacheImpl* self) {
+	state boost::unordered_set<AuditEntry> audits;
+	state boost::unordered_set<AuditEntry>::iterator iter;
+	loop {
+		wait(delay(FLOW_KNOBS->AUDIT_TIME_WINDOW));
+		audits.swap(self->usedTokens);
+		for (iter = audits.begin(); iter != audits.end(); ++iter) {
+			CODE_PROBE(true, "Audit Logging Running");
+			TraceEvent("AuditTokenUsed").detail("Client", iter->address).detail("TokenId", iter->tokenId).log();
+			wait(yield());
+		}
+		audits.clear();
+	}
+}
+
+TokenCacheImpl::TokenCacheImpl() : cache(FLOW_KNOBS->TOKEN_CACHE_SIZE) {
+	auditor = tokenCacheAudit(this);
+}
+
 TokenCache::TokenCache() : impl(new TokenCacheImpl()) {}
 TokenCache::~TokenCache() {
 	delete impl;
@@ -212,6 +258,9 @@ bool TokenCacheImpl::validateAndAdd(double currentTime, StringRef token, Network
 		for (auto tenant : t.tenants.get()) {
 			c.tenants.push_back_deep(c.arena, tenant);
 		}
+		if (t.tokenId.present()) {
+			c.tokenId = StringRef(c.arena, t.tokenId.get());
+		}
 		cache.insert(StringRef(c.arena, token), c);
 		return true;
 	}
@@ -250,6 +299,8 @@ bool TokenCacheImpl::validate(TenantNameRef name, StringRef token) {
 		TraceEvent(SevWarn, "TenantTokenMismatch").detail("From", peer).detail("Tenant", name.toString());
 		return false;
 	}
+	// autit logging
+	usedTokens.insert(AuditEntry(peer, *cachedEntry.get()));
 	return true;
 }
 

flow/Knobs.cpp

@@ -136,6 +136,7 @@ void FlowKnobs::initialize(Randomize randomize, IsSimulated isSimulated) {
 	init( PUBLIC_KEY_FILE_MAX_SIZE,                    1024 * 1024 );
 	init( PUBLIC_KEY_FILE_REFRESH_INTERVAL_SECONDS,             30 );
 	init( MAX_CACHED_EXPIRED_TOKENS,                          1024 );
+	init( AUDIT_TIME_WINDOW,                                   5.0 );
 
 	//AsyncFileCached
 	init( PAGE_CACHE_4K,                                   2LL<<30 );

flow/include/flow/Knobs.h

@@ -200,6 +200,7 @@ public:
 	int PUBLIC_KEY_FILE_MAX_SIZE;
 	int PUBLIC_KEY_FILE_REFRESH_INTERVAL_SECONDS;
 	int MAX_CACHED_EXPIRED_TOKENS;
+	double AUDIT_TIME_WINDOW;
 
 	// AsyncFileCached
 	int64_t PAGE_CACHE_4K;