EncryptedDirectMessage class; simplify Event

README.md

@@ -11,6 +11,7 @@ public_key = private_key.public_key
 print(f"Private key: {private_key.bech32()}")
 print(f"Public key: {public_key.bech32()}")
 ```
+
 **Connect to relays**
 ```python
 import json
@@ -30,6 +31,7 @@ while relay_manager.message_pool.has_notices():
 
 relay_manager.close_connections()
 ```
+
 **Publish to relays**
 ```python
 import json 
@@ -48,14 +50,46 @@ time.sleep(1.25) # allow the connections to open
 
 private_key = PrivateKey()
 
-event = Event(private_key.public_key.hex(), "Hello Nostr")
+event = Event("Hello Nostr")
 private_key.sign_event(event)
 
 relay_manager.publish_event(event)
 time.sleep(1) # allow the messages to send
 
 relay_manager.close_connections()
 ```
+
+**Reply to a note**
+```python
+from nostr.event import Event
+
+reply = Event(
+  content="Hey, that's a great point!",
+)
+
+# create 'e' tag reference to the note you're replying to
+reply.add_event_ref(original_note_id)
+
+# create 'p' tag reference to the pubkey you're replying to
+reply.add_pubkey_ref(original_note_author_pubkey)
+
+private_key.sign_event(reply)
+relay_manager.publish_event(reply)
+```
+
+**Send a DM**
+```python
+from nostr.event import EncryptedDirectMessage
+
+dm = EncryptedDirectMessage(
+  recipient_pubkey=recipient_pubkey,
+  cleartext_content="Secret message!"
+)
+private_key.sign_event(dm)
+relay_manager.publish_event(dm)
+```
+
+
 **Receive events from relays**
 ```python
 import json
@@ -112,7 +146,6 @@ delegation = Delegation(
 identity_pk.sign_delegation(delegation)
 
 event = Event(
-    delegatee_pk.public_key.hex(),
     "Hello, NIP-26!",
     tags=[delegation.get_tag()],
 )

nostr/event.py

@@ -1,6 +1,8 @@
 import time
 import json
+from dataclasses import dataclass
 from enum import IntEnum
+from typing import List
 from secp256k1 import PrivateKey, PublicKey
 from hashlib import sha256
 
@@ -16,41 +18,65 @@ class EventKind(IntEnum):
     DELETE = 5
 
 
-class Event():
-    def __init__(
-            self, 
-            public_key: str, 
-            content: str, 
-            created_at: int = None, 
-            kind: int=EventKind.TEXT_NOTE, 
-            tags: "list[list[str]]"=[], 
-            id: str=None, 
-            signature: str=None) -> None:
-        if not isinstance(content, str):
+
+@dataclass
+class Event:
+    public_key: str = None
+    content: str = None
+    created_at: int = None
+    kind: int = EventKind.TEXT_NOTE
+    tags: List[List[str]] = None
+    id: str = None
+    signature: str = None
+
+
+    def __post_init__(self):
+        if self.content is not None and not isinstance(self.content, str):
+            # DMs initialize content to None but all other kinds should pass in a str
             raise TypeError("Argument 'content' must be of type str")
 
-        self.public_key = public_key
-        self.content = content
-        self.created_at = created_at or int(time.time())
-        self.kind = kind
-        self.tags = tags
-        self.signature = signature
-        self.id = id or Event.compute_id(self.public_key, self.created_at, self.kind, self.tags, self.content)
+        if self.created_at is None:
+            self.created_at = int(time.time())
+
+        # Can't initialize the nested type above w/out more complex factory, so doing it here
+        if self.tags is None:
+            self.tags = []
+
+        if self.id is None:
+            self.compute_id()
+
 
     @staticmethod
-    def serialize(public_key: str, created_at: int, kind: int, tags: "list[list[str]]", content: str) -> bytes:
+    def serialize(public_key: str, created_at: int, kind: int, tags: List[List[str]], content: str) -> bytes:
         data = [0, public_key, created_at, kind, tags, content]
         data_str = json.dumps(data, separators=(',', ':'), ensure_ascii=False)
         return data_str.encode()
 
-    @staticmethod
-    def compute_id(public_key: str, created_at: int, kind: int, tags: "list[list[str]]", content: str) -> str:
-        return sha256(Event.serialize(public_key, created_at, kind, tags, content)).hexdigest()
+
+    def compute_id(self):
+        self.id = sha256(Event.serialize(self.public_key, self.created_at, self.kind, self.tags, self.content)).hexdigest()
+
+
+    def add_pubkey_ref(self, pubkey:str):
+        """ Adds a reference to a pubkey as a 'p' tag """
+        self.tags.append(['p', pubkey])
+        self.compute_id()
+
+
+    def add_event_ref(self, event_id:str):
+        """ Adds a reference to an event_id as an 'e' tag """
+        self.tags.append(['e', event_id])
+        self.compute_id()
+
 
     def verify(self) -> bool:
         pub_key = PublicKey(bytes.fromhex("02" + self.public_key), True) # add 02 for schnorr (bip340)
-        event_id = Event.compute_id(self.public_key, self.created_at, self.kind, self.tags, self.content)
-        return pub_key.schnorr_verify(bytes.fromhex(event_id), bytes.fromhex(self.signature), None, raw=True)
+
+        # Always recompute id just in case something changed
+        self.compute_id()
+
+        return pub_key.schnorr_verify(bytes.fromhex(self.id), bytes.fromhex(self.signature), None, raw=True)
+
 
     def to_message(self) -> str:
         return json.dumps(
@@ -67,3 +93,29 @@ def to_message(self) -> str:
                 }
             ]
         )
+
+
+
+@dataclass
+class EncryptedDirectMessage(Event):
+    recipient_pubkey: str = None
+    cleartext_content: str = None
+    reference_event_id: str = None
+
+
+    def __post_init__(self):
+        if self.content is not None:
+            raise Exception("Encrypted DMs cannot use the `content` field; use `cleartext_content` instead.")
+
+        if self.recipient_pubkey is None:
+            raise Exception("Must specify a recipient_pubkey.")
+
+        self.kind = EventKind.ENCRYPTED_DIRECT_MESSAGE
+        super().__post_init__()
+
+        # Must specify the DM recipient's pubkey in a 'p' tag
+        self.add_pubkey_ref(self.recipient_pubkey)
+
+        # Optionally specify a reference event (DM) this is a reply to
+        if self.reference_event_id:
+            self.add_event_ref(self.reference_event_id)

nostr/key.py

@@ -7,7 +7,7 @@
 from hashlib import sha256
 
 from nostr.delegation import Delegation
-from nostr.event import Event
+from nostr.event import EncryptedDirectMessage, Event, EventKind
 from . import bech32
 
 
@@ -77,6 +77,10 @@ def encrypt_message(self, message: str, public_key_hex: str) -> str:
         encrypted_message = encryptor.update(padded_data) + encryptor.finalize()
 
         return f"{base64.b64encode(encrypted_message).decode()}?iv={base64.b64encode(iv).decode()}"
+
+    def encrypt_dm(self, dm: EncryptedDirectMessage) -> None:
+        encrypted_message = self.encrypt_message(message=dm.cleartext_content, public_key_hex=dm.recipient_pubkey)
+        dm.content = encrypted_message
 
     def decrypt_message(self, encoded_message: str, public_key_hex: str) -> str:
         encoded_data = encoded_message.split('?iv=')
@@ -100,6 +104,11 @@ def sign_message_hash(self, hash: bytes) -> str:
         return sig.hex()
 
     def sign_event(self, event: Event) -> None:
+        if event.kind == EventKind.ENCRYPTED_DIRECT_MESSAGE and event.content is None:
+            self.encrypt_dm(event)
+        if not event.public_key:
+            event.public_key = self.public_key.hex()
+        event.compute_id()
         event.signature = self.sign_message_hash(bytes.fromhex(event.id))
 
     def sign_delegation(self, delegation: Delegation) -> None:
@@ -108,19 +117,6 @@ def sign_delegation(self, delegation: Delegation) -> None:
     def __eq__(self, other):
         return self.raw_secret == other.raw_secret
 
-def mine_vanity_key(prefix: str = None, suffix: str = None) -> PrivateKey:
-    if prefix is None and suffix is None:
-        raise ValueError("Expected at least one of 'prefix' or 'suffix' arguments")
-
-    while True:
-        sk = PrivateKey()
-        if prefix is not None and not sk.public_key.bech32()[5:5+len(prefix)] == prefix:
-            continue
-        if suffix is not None and not sk.public_key.bech32()[-len(suffix):] == suffix:
-            continue
-        break
-
-    return sk
 
 ffi = FFI()
 @ffi.callback("int (unsigned char *, const unsigned char *, const unsigned char *, void *)")

test/test_event.py

@@ -1,14 +1,57 @@
-from nostr.event import Event
-from nostr.key import PrivateKey
+import pytest
 import time
+from nostr.event import Event, EncryptedDirectMessage
+from nostr.key import PrivateKey
+
+
+
+class TestEvent:
+    def test_event_default_time(self):
+        """
+            ensure created_at default value reflects the time at Event object instantiation
+            see: https://github.com/jeffthibault/python-nostr/issues/23
+        """
+        event1 = Event(content='test event')
+        time.sleep(1.5)
+        event2 = Event(content='test event')
+        assert event1.created_at < event2.created_at
+
+
+    def test_add_event_ref(self):
+        some_event_id = "some_event_id"
+        event = Event(content="Adding an 'e' tag")
+        event.add_event_ref(some_event_id)
+        assert ['e', some_event_id] in event.tags
+
+
+    def test_add_pubkey_ref(self):
+        some_pubkey = "some_pubkey"
+        event = Event(content="Adding a 'p' tag")
+        event.add_pubkey_ref(some_pubkey)
+        assert ['p', some_pubkey] in event.tags
+
+
+
+class TestEncryptedDirectMessage:
+    def setup_class(self):
+        self.sender_pk = PrivateKey()
+        self.sender_pubkey = self.sender_pk.public_key.hex()
+        self.recipient_pk = PrivateKey()
+        self.recipient_pubkey = self.recipient_pk.public_key.hex()
+
+
+    def test_content_field_not_allowed(self):
+        """ Should not let users instantiate a new DM with `content` field data """
+        with pytest.raises(Exception) as e:
+            EncryptedDirectMessage(recipient_pubkey=self.recipient_pubkey, content="My message!")
+
+        assert "cannot use" in str(e)
+
 
-def test_event_default_time():
-    """
-        ensure created_at default value reflects the time at Event object instantiation
-        see: https://github.com/jeffthibault/python-nostr/issues/23
-    """
-    public_key = PrivateKey().public_key.hex()
-    event1 = Event(public_key=public_key, content='test event')
-    time.sleep(1.5)
-    event2 = Event(public_key=public_key, content='test event')
-    assert event1.created_at < event2.created_at
+    def test_recipient_p_tag(self):
+        """ Should generate recipient 'p' tag """
+        dm = EncryptedDirectMessage(
+            recipient_pubkey=self.recipient_pubkey,
+            cleartext_content="Secret message!"
+        )
+        assert ['p', self.recipient_pubkey] in dm.tags