hardening read of corrupt oplog (#98)

* Add extra validation of oplog entries when reading oplog

1 files changed, 50 insertions, 24 deletions

diff --git a/src/zenserver/projectstore/projectstore.cpp b/src/zenserver/projectstore/projectstore.cpp
index beac1ab97..a542a6581 100644
--- a/src/zenserver/projectstore/projectstore.cpp
+++ b/src/zenserver/projectstore/projectstore.cpp
@@ -298,10 +298,11 @@ struct ProjectStore::OplogStorage : public RefCounted
 
 		ZEN_INFO("replaying log for '{}'", m_OplogStoragePath);
 
+		uint64_t OpsBlockSize = m_OpBlobs.FileSize();
+
 		Stopwatch Timer;
 
-		uint64_t InvalidEntries	  = 0;
-		uint64_t TombstoneEntries = 0;
+		uint64_t InvalidEntries = 0;
 
 		std::vector<OplogEntry> OpLogEntries;
 		std::vector<size_t>		OplogOrder;
@@ -316,20 +317,29 @@ struct ProjectStore::OplogStorage : public RefCounted
 						if (auto It = LatestKeys.find(LogEntry.OpKeyHash); It == LatestKeys.end())
 						{
 							ZEN_SCOPED_WARN("found tombstone referencing unknown key {}", LogEntry.OpKeyHash);
-						}
-					}
-					else
-					{
-						if (LogEntry.OpCoreSize == 0)
-						{
 							++InvalidEntries;
 							return;
 						}
+					}
+					else if (LogEntry.OpCoreSize == 0)
+					{
+						ZEN_SCOPED_WARN("skipping zero size op {}", LogEntry.OpKeyHash);
+						++InvalidEntries;
+						return;
+					}
+					else if (LogEntry.OpLsn == 0)
+					{
+						ZEN_SCOPED_WARN("skipping zero lsn op {}", LogEntry.OpKeyHash);
+						++InvalidEntries;
+						return;
+					}
 
-						const uint64_t OpFileOffset = LogEntry.OpCoreOffset * m_OpsAlign;
-						m_NextOpsOffset =
-							Max(m_NextOpsOffset.load(std::memory_order_relaxed), RoundUp(OpFileOffset + LogEntry.OpCoreSize, m_OpsAlign));
-						m_MaxLsn = Max(m_MaxLsn.load(std::memory_order_relaxed), LogEntry.OpLsn);
+					const uint64_t OpFileOffset = LogEntry.OpCoreOffset * m_OpsAlign;
+					if ((OpFileOffset + LogEntry.OpCoreSize) > OpsBlockSize)
+					{
+						ZEN_SCOPED_WARN("skipping out of bounds op {}", LogEntry.OpKeyHash);
+						++InvalidEntries;
+						return;
 					}
 
 					if (auto It = LatestKeys.find(LogEntry.OpKeyHash); It != LatestKeys.end())
@@ -338,7 +348,7 @@ struct ProjectStore::OplogStorage : public RefCounted
 
 						if (LogEntry.IsTombstone() && Entry.IsTombstone())
 						{
-							ZEN_SCOPED_WARN("found double tombstone - '{}'", LogEntry.OpKeyHash);
+							ZEN_SCOPED_WARN("found double tombstone - {}", LogEntry.OpKeyHash);
 						}
 
 						Entry = LogEntry;
@@ -360,8 +370,13 @@ struct ProjectStore::OplogStorage : public RefCounted
 			return LhsEntry.OpCoreOffset < RhsEntry.OpCoreOffset;
 		});
 
+		uint64_t TombstoneEntries = 0;
+
 		BasicFileBuffer OpBlobsBuffer(m_OpBlobs, 65536);
 
+		uint32_t MaxOpLsn		  = m_MaxLsn.load(std::memory_order_relaxed);
+		uint64_t NextOpFileOffset = m_NextOpsOffset.load(std::memory_order_relaxed);
+
 		for (size_t OplogOrderIndex : OplogOrder)
 		{
 			const OplogEntry& LogEntry = OpLogEntries[OplogOrderIndex];
@@ -375,16 +390,28 @@ struct ProjectStore::OplogStorage : public RefCounted
 				// Verify checksum, ignore op data if incorrect
 
 				auto VerifyAndHandleOp = [&](MemoryView OpBufferView) {
-					const uint32_t OpCoreHash = uint32_t(XXH3_64bits(OpBufferView.GetData(), LogEntry.OpCoreSize) & 0xffffFFFF);
+					const uint32_t ExpectedOpCoreHash = LogEntry.OpCoreHash;
+					const uint32_t OpCoreHash		  = uint32_t(XXH3_64bits(OpBufferView.GetData(), LogEntry.OpCoreSize) & 0xffffFFFF);
 
-					if (OpCoreHash == LogEntry.OpCoreHash)
+					if (OpCoreHash != ExpectedOpCoreHash)
 					{
-						Handler(CbObjectView(OpBufferView.GetData()), LogEntry);
+						ZEN_WARN("skipping bad checksum op - {}. Expected: {}, found: {}",
+								 LogEntry.OpKeyHash,
+								 ExpectedOpCoreHash,
+								 OpCoreHash);
+					}
+					else if (CbValidateError Err = ValidateCompactBinary(OpBufferView, CbValidateMode::Default);
+							 Err != CbValidateError::None)
+					{
+						ZEN_WARN("skipping invalid format op - {}. Error: '{}'", LogEntry.OpKeyHash, ToString(Err));
 					}
 					else
 					{
-						ZEN_WARN("skipping oplog entry with bad checksum!");
-						InvalidEntries++;
+						Handler(CbObjectView(OpBufferView.GetData()), LogEntry);
+						MaxOpLsn = Max(MaxOpLsn, LogEntry.OpLsn);
+						const uint64_t EntryNextOpFileOffset =
+							RoundUp((LogEntry.OpCoreOffset * m_OpsAlign) + LogEntry.OpCoreSize, m_OpsAlign);
+						NextOpFileOffset = Max(NextOpFileOffset, EntryNextOpFileOffset);
 					}
 				};
 
@@ -404,16 +431,15 @@ struct ProjectStore::OplogStorage : public RefCounted
 			}
 		}
 
-		if (InvalidEntries)
-		{
-			ZEN_WARN("ignored {} invalid oplog entries", InvalidEntries);
-		}
+		m_NextOpsOffset = NextOpFileOffset;
+		m_MaxLsn		= MaxOpLsn;
 
-		ZEN_INFO("oplog replay completed in {} - Max LSN# {}, Next offset: {}, {} tombstones",
+		ZEN_INFO("oplog replay completed in {} - Max LSN# {}, Next offset: {}, {} tombstones, {} invalid entries",
 				 NiceTimeSpanMs(Timer.GetElapsedTimeMs()),
 				 m_MaxLsn.load(),
 				 m_NextOpsOffset.load(),
-				 TombstoneEntries);
+				 TombstoneEntries,
+				 InvalidEntries);
 	}
 
 	void ReplayLogEntries(const std::span<OplogEntryAddress> Entries, std::function<void(CbObjectView)>&& Handler)