57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
import hashlib
|
|
import json
|
|
from typing import List, Optional
|
|
|
|
from coincurve import PublicKeyXOnly
|
|
from pydantic import BaseModel
|
|
|
|
|
|
class NostrEvent(BaseModel):
|
|
id: str = ""
|
|
pubkey: str
|
|
created_at: int
|
|
kind: int
|
|
tags: List[List[str]] = []
|
|
content: str = ""
|
|
sig: Optional[str] = None
|
|
|
|
def serialize(self) -> List:
|
|
return [0, self.pubkey, self.created_at, self.kind, self.tags, self.content]
|
|
|
|
def serialize_json(self) -> str:
|
|
e = self.serialize()
|
|
return json.dumps(e, separators=(",", ":"), ensure_ascii=False)
|
|
|
|
@property
|
|
def event_id(self) -> str:
|
|
data = self.serialize_json()
|
|
id = hashlib.sha256(data.encode()).hexdigest()
|
|
return id
|
|
|
|
def check_signature(self):
|
|
event_id = self.event_id
|
|
if self.id != event_id:
|
|
raise ValueError(
|
|
f"Invalid event id. Expected: '{event_id}' got '{self.id}'"
|
|
)
|
|
try:
|
|
pub_key = PublicKeyXOnly(bytes.fromhex(self.pubkey))
|
|
except Exception as exc:
|
|
raise ValueError(
|
|
f"Invalid public key: '{self.pubkey}' for event '{self.id}'"
|
|
) from exc
|
|
|
|
valid_signature = self.sig and pub_key.verify(
|
|
bytes.fromhex(self.sig), bytes.fromhex(event_id)
|
|
)
|
|
if not valid_signature:
|
|
raise ValueError(f"Invalid signature: '{self.sig}' for event '{self.id}'")
|
|
|
|
def stringify(self) -> str:
|
|
return json.dumps(dict(self))
|
|
|
|
def tag_values(self, tag_name: str) -> List[str]:
|
|
return [t[1] for t in self.tags if t[0] == tag_name]
|
|
|
|
def has_tag_value(self, tag_name: str, tag_value: str) -> bool:
|
|
return tag_value in self.tag_values(tag_name)
|