hardening read of corrupt oplog (#98)

* Add extra validation of oplog entries when reading oplog

2 files changed, 51 insertions, 24 deletions

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5d48e9d7c..9a1c6ba32 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,7 @@
 - Bugfix: Fix ASSERT that would trigger in GC under certain conditions if source block was empty
 - Improvement: `zen workspace-share create` now resolves relative root paths to absolute paths
 - Improvement: Add better output/logging when failing to initialize shared mutex
+- Improvement: Add hardening to gracefully handle malformed oplogs in project store
 
 ## 5.5.3
 - Feature: New 'workspaces' service which allows a user to share a local folder via zenserver. A workspace can have mulitple workspace shares and they provie an HTTP API that is compatible with the project oplog HTTP API. Workspaces and shares are preserved between runs. Workspaces feature is disabled by default - enable with `--workspaces-enabled` option when launching zenserver.
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)