feat: Add verify_hash function for events.

Author: goloroden

eventsourcingdb/event/event.py

@@ -1,5 +1,7 @@
 from dataclasses import dataclass
 from datetime import datetime
+import json
+from hashlib import sha256
 from typing import Any, TypeVar
 
 from ..errors.internal_error import InternalError
@@ -94,6 +96,32 @@ def parse(unknown_object: dict) -> "Event":
             trace_state=trace_state,
         )
 
+    def verify_hash(self) -> None:
+        metadata = "|".join([
+            self.spec_version,
+            self.event_id,
+            self.predecessor_hash,
+            self.time.isoformat(sep="T"),
+            self.source,
+            self.subject,
+            self.type,
+            self.data_content_type,
+        ])
+
+        metadata_bytes = metadata.encode("utf-8")
+        data_bytes = json.dumps(self.data).encode("utf-8")
+
+        metadata_hash = sha256(metadata_bytes).hexdigest()
+        data_hash = sha256(data_bytes).hexdigest()
+
+        final_hash = sha256()
+        final_hash.update(metadata_hash.encode("utf-8"))
+        final_hash.update(data_hash.encode("utf-8"))
+        final_hash_hex = final_hash.hexdigest()
+
+        if final_hash_hex != self.hash:
+            raise ValidationError("Failed to verify hash.")
+
     def to_json(self) -> dict[str, Any]:
         json = {
             "specversion": self.spec_version,

tests/event/test_verify_hash.py

@@ -0,0 +1,29 @@
+import pytest
+
+from eventsourcingdb import EventCandidate
+
+from ..conftest import TestData
+from ..shared.database import Database
+
+
+class TestVerifyHash:
+    @staticmethod
+    @pytest.mark.asyncio
+    async def test_verifies_the_event_hash(
+        database: Database,
+        test_data: TestData,
+    ) -> None:
+        client = database.get_client()
+
+        written_events = await client.write_events(
+            [
+                EventCandidate(
+                    source=test_data.TEST_SOURCE_STRING, subject="/test", type="io.eventsourcingdb.test", data={"value": 23}
+                )
+            ],
+        )
+
+        assert len(written_events) == 1
+
+        written_event = written_events[0]
+        written_event.verify_hash()