Padreug
4811fcf352
Generalize the AUTH-gated, recipient-only delivery rule from NIP-04 to also cover NIP-17 kind 1059 gift wraps. When the relay is configured to require AUTH for kind 1059, only the AUTH'd recipient named in the event's `p` tag receives it; otherwise gift wraps broadcast like any regular event. - relay/event.py: add `is_seal`, `is_gift_wrap`, `is_private_message` helpers (kinds 13, 1059) - relay/client_connection.py: rename `_is_direct_message_for_other` -> `_is_private_event_for_other`; key off `is_private_message` so the same gating applies to kinds 4 and 1059 - relay/relay.py: advertise NIPs 17, 44, 59 in NIP-11 supported_nips - README: document NIP-17/44/59 transport-level support - tests/test_nip17.py: unit tests for kind classification, AUTH-gated 1059 delivery (recipient vs non-recipient vs unauthenticated), and regression coverage for kind 4 gating NIP-44 (encryption) and NIP-59 (wrap/seal) are client-side concerns; the relay treats payloads as opaque ciphertext and stores kind 1059 like any regular event. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
124 lines
3.6 KiB
Python
124 lines
3.6 KiB
Python
import hashlib
|
|
import json
|
|
from enum import Enum
|
|
|
|
from coincurve import PublicKeyXOnly
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
class NostrEventType(str, Enum):
|
|
EVENT = "EVENT"
|
|
REQ = "REQ"
|
|
CLOSE = "CLOSE"
|
|
AUTH = "AUTH"
|
|
|
|
|
|
class NostrEvent(BaseModel):
|
|
id: str
|
|
relay_id: str
|
|
publisher: str
|
|
pubkey: str
|
|
created_at: int
|
|
kind: int
|
|
tags: list[list[str]] = Field(default=[], no_database=True)
|
|
content: str = ""
|
|
sig: str
|
|
size: int = 0
|
|
|
|
def nostr_dict(self) -> dict:
|
|
_nostr_dict = dict(self)
|
|
_nostr_dict.pop("relay_id")
|
|
_nostr_dict.pop("publisher")
|
|
return _nostr_dict
|
|
|
|
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()
|
|
return hashlib.sha256(data.encode()).hexdigest()
|
|
|
|
@property
|
|
def size_bytes(self) -> int:
|
|
s = json.dumps(self.nostr_dict(), separators=(",", ":"), ensure_ascii=False)
|
|
return len(s.encode())
|
|
|
|
@property
|
|
def is_replaceable_event(self) -> bool:
|
|
return self.kind in [0, 3, 41] or (self.kind >= 10000 and self.kind < 20000)
|
|
|
|
@property
|
|
def is_auth_response_event(self) -> bool:
|
|
return self.kind == 22242
|
|
|
|
@property
|
|
def is_direct_message(self) -> bool:
|
|
return self.kind == 4
|
|
|
|
@property
|
|
def is_delete_event(self) -> bool:
|
|
return self.kind == 5
|
|
|
|
@property
|
|
def is_seal(self) -> bool:
|
|
return self.kind == 13
|
|
|
|
@property
|
|
def is_gift_wrap(self) -> bool:
|
|
return self.kind == 1059
|
|
|
|
@property
|
|
def is_private_message(self) -> bool:
|
|
# Kinds whose payload addresses a single recipient via a `p` tag and is
|
|
# not meant to be broadcast to other subscribers when AUTH is enforced.
|
|
# NIP-04 (kind 4) and NIP-17 (kind 1059 gift wrap).
|
|
return self.is_direct_message or self.is_gift_wrap
|
|
|
|
@property
|
|
def is_regular_event(self) -> bool:
|
|
return self.kind >= 1000 and self.kind < 10000
|
|
|
|
@property
|
|
def is_ephemeral_event(self) -> bool:
|
|
return self.kind >= 20000 and self.kind < 30000
|
|
|
|
@property
|
|
def is_addressable_event(self) -> bool:
|
|
return self.kind >= 30000 and self.kind < 40000
|
|
|
|
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 = 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 serialize_response(self, subscription_id):
|
|
return [NostrEventType.EVENT, subscription_id, self.nostr_dict()]
|
|
|
|
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)
|
|
|
|
def is_direct_message_for_pubkey(self, pubkey: str) -> bool:
|
|
return self.is_direct_message and self.has_tag_value("p", pubkey)
|