[claude] Timmy Nostr identity — keypair, profile, relay presence (#856) (#1325)

Co-authored-by: Claude (Opus 4.6) <claude@hermes.local>
Co-committed-by: Claude (Opus 4.6) <claude@hermes.local>

tests/unit/test_nostr_event.py

@@ -0,0 +1,177 @@
+"""Unit tests for infrastructure.nostr.event."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import time
+
+import pytest
+
+from infrastructure.nostr.event import (
+    _event_hash,
+    build_event,
+    schnorr_sign,
+    schnorr_verify,
+)
+from infrastructure.nostr.keypair import generate_keypair
+
+
+class TestSchorrSign:
+    def test_returns_64_bytes(self):
+        kp = generate_keypair()
+        msg = b"\x00" * 32
+        sig = schnorr_sign(msg, kp.privkey_bytes)
+        assert len(sig) == 64
+
+    def test_different_msg_different_sig(self):
+        kp = generate_keypair()
+        sig1 = schnorr_sign(b"\x01" * 32, kp.privkey_bytes)
+        sig2 = schnorr_sign(b"\x02" * 32, kp.privkey_bytes)
+        assert sig1 != sig2
+
+    def test_raises_on_wrong_msg_length(self):
+        kp = generate_keypair()
+        with pytest.raises(ValueError, match="32 bytes"):
+            schnorr_sign(b"too short", kp.privkey_bytes)
+
+    def test_raises_on_wrong_key_length(self):
+        msg = b"\x00" * 32
+        with pytest.raises(ValueError, match="32 bytes"):
+            schnorr_sign(msg, b"too short")
+
+    def test_nondeterministic_due_to_randomness(self):
+        # BIP-340 uses auxiliary randomness; repeated calls produce different sigs
+        kp = generate_keypair()
+        msg = b"\x42" * 32
+        sig1 = schnorr_sign(msg, kp.privkey_bytes)
+        sig2 = schnorr_sign(msg, kp.privkey_bytes)
+        # With different random nonces these should differ (astronomically unlikely to collide)
+        # We just verify both are valid
+        assert schnorr_verify(msg, kp.pubkey_bytes, sig1)
+        assert schnorr_verify(msg, kp.pubkey_bytes, sig2)
+
+
+class TestSchnorrVerify:
+    def test_valid_signature_verifies(self):
+        kp = generate_keypair()
+        msg = hashlib.sha256(b"hello nostr").digest()
+        sig = schnorr_sign(msg, kp.privkey_bytes)
+        assert schnorr_verify(msg, kp.pubkey_bytes, sig) is True
+
+    def test_wrong_pubkey_fails(self):
+        kp1 = generate_keypair()
+        kp2 = generate_keypair()
+        msg = b"\x00" * 32
+        sig = schnorr_sign(msg, kp1.privkey_bytes)
+        assert schnorr_verify(msg, kp2.pubkey_bytes, sig) is False
+
+    def test_tampered_sig_fails(self):
+        kp = generate_keypair()
+        msg = b"\x00" * 32
+        sig = bytearray(schnorr_sign(msg, kp.privkey_bytes))
+        sig[0] ^= 0xFF
+        assert schnorr_verify(msg, kp.pubkey_bytes, bytes(sig)) is False
+
+    def test_tampered_msg_fails(self):
+        kp = generate_keypair()
+        msg = b"\x00" * 32
+        sig = schnorr_sign(msg, kp.privkey_bytes)
+        bad_msg = b"\xFF" * 32
+        assert schnorr_verify(bad_msg, kp.pubkey_bytes, sig) is False
+
+    def test_wrong_lengths_return_false(self):
+        kp = generate_keypair()
+        msg = b"\x00" * 32
+        sig = schnorr_sign(msg, kp.privkey_bytes)
+        assert schnorr_verify(msg[:16], kp.pubkey_bytes, sig) is False
+        assert schnorr_verify(msg, kp.pubkey_bytes[:16], sig) is False
+        assert schnorr_verify(msg, kp.pubkey_bytes, sig[:32]) is False
+
+    def test_never_raises(self):
+        # Should return False for any garbage input, not raise
+        assert schnorr_verify(b"x", b"y", b"z") is False
+
+
+class TestEventHash:
+    def test_returns_32_bytes(self):
+        h = _event_hash("aabbcc", 0, 1, [], "")
+        assert len(h) == 32
+
+    def test_deterministic(self):
+        h1 = _event_hash("aa", 1, 1, [], "hello")
+        h2 = _event_hash("aa", 1, 1, [], "hello")
+        assert h1 == h2
+
+    def test_different_content_different_hash(self):
+        h1 = _event_hash("aa", 1, 1, [], "hello")
+        h2 = _event_hash("aa", 1, 1, [], "world")
+        assert h1 != h2
+
+
+class TestBuildEvent:
+    def test_returns_required_fields(self):
+        kp = generate_keypair()
+        ev = build_event(kind=1, content="hello", keypair=kp)
+        assert set(ev) >= {"id", "pubkey", "created_at", "kind", "tags", "content", "sig"}
+
+    def test_kind_matches(self):
+        kp = generate_keypair()
+        ev = build_event(kind=0, content="{}", keypair=kp)
+        assert ev["kind"] == 0
+
+    def test_pubkey_matches_keypair(self):
+        kp = generate_keypair()
+        ev = build_event(kind=1, content="x", keypair=kp)
+        assert ev["pubkey"] == kp.pubkey_hex
+
+    def test_id_is_64_char_hex(self):
+        kp = generate_keypair()
+        ev = build_event(kind=1, content="x", keypair=kp)
+        assert len(ev["id"]) == 64
+        assert all(c in "0123456789abcdef" for c in ev["id"])
+
+    def test_sig_is_128_char_hex(self):
+        kp = generate_keypair()
+        ev = build_event(kind=1, content="x", keypair=kp)
+        assert len(ev["sig"]) == 128
+        assert all(c in "0123456789abcdef" for c in ev["sig"])
+
+    def test_signature_verifies(self):
+        kp = generate_keypair()
+        ev = build_event(kind=1, content="test", keypair=kp)
+        sig_bytes = bytes.fromhex(ev["sig"])
+        id_bytes = bytes.fromhex(ev["id"])
+        assert schnorr_verify(id_bytes, kp.pubkey_bytes, sig_bytes)
+
+    def test_id_matches_canonical_hash(self):
+        kp = generate_keypair()
+        ts = int(time.time())
+        ev = build_event(kind=1, content="hi", keypair=kp, created_at=ts)
+        expected_hash = _event_hash(kp.pubkey_hex, ts, 1, [], "hi").hex()
+        assert ev["id"] == expected_hash
+
+    def test_custom_tags(self):
+        kp = generate_keypair()
+        tags = [["t", "gaming"], ["r", "wss://relay.example.com"]]
+        ev = build_event(kind=1, content="x", keypair=kp, tags=tags)
+        assert ev["tags"] == tags
+
+    def test_default_tags_empty(self):
+        kp = generate_keypair()
+        ev = build_event(kind=1, content="x", keypair=kp)
+        assert ev["tags"] == []
+
+    def test_custom_created_at(self):
+        kp = generate_keypair()
+        ts = 1700000000
+        ev = build_event(kind=1, content="x", keypair=kp, created_at=ts)
+        assert ev["created_at"] == ts
+
+    def test_kind0_profile_content_is_json(self):
+        kp = generate_keypair()
+        profile = {"name": "Timmy", "about": "test"}
+        ev = build_event(kind=0, content=json.dumps(profile), keypair=kp)
+        assert ev["kind"] == 0
+        parsed = json.loads(ev["content"])
+        assert parsed["name"] == "Timmy"

tests/unit/test_nostr_identity.py

@@ -0,0 +1,272 @@
+"""Unit tests for infrastructure.nostr.identity."""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from infrastructure.nostr.identity import AnnounceResult, NostrIdentityManager
+from infrastructure.nostr.keypair import generate_keypair
+
+
+@pytest.fixture()
+def manager():
+    return NostrIdentityManager()
+
+
+@pytest.fixture()
+def kp():
+    return generate_keypair()
+
+
+class TestAnnounceResult:
+    def test_any_relay_ok_false_when_empty(self):
+        r = AnnounceResult()
+        assert r.any_relay_ok is False
+
+    def test_any_relay_ok_true_when_one_ok(self):
+        r = AnnounceResult(relay_results={"wss://a": True, "wss://b": False})
+        assert r.any_relay_ok is True
+
+    def test_to_dict_keys(self):
+        r = AnnounceResult(kind_0_ok=True, relay_results={"wss://a": True})
+        d = r.to_dict()
+        assert set(d) == {"kind_0", "kind_31990", "relays"}
+
+
+class TestGetKeypair:
+    def test_returns_none_when_no_privkey(self, manager):
+        mock_settings = MagicMock(nostr_privkey="")
+        with patch("infrastructure.nostr.identity.settings", mock_settings):
+            assert manager.get_keypair() is None
+
+    def test_returns_keypair_when_configured(self, manager, kp):
+        mock_settings = MagicMock(nostr_privkey=kp.privkey_hex)
+        with patch("infrastructure.nostr.identity.settings", mock_settings):
+            result = manager.get_keypair()
+        assert result is not None
+        assert result.pubkey_hex == kp.pubkey_hex
+
+    def test_returns_none_on_invalid_key(self, manager):
+        mock_settings = MagicMock(nostr_privkey="not_a_valid_key")
+        with patch("infrastructure.nostr.identity.settings", mock_settings):
+            assert manager.get_keypair() is None
+
+
+class TestGetRelayUrls:
+    def test_empty_string_returns_empty_list(self, manager):
+        mock_settings = MagicMock(nostr_relays="")
+        with patch("infrastructure.nostr.identity.settings", mock_settings):
+            assert manager.get_relay_urls() == []
+
+    def test_single_relay(self, manager):
+        mock_settings = MagicMock(nostr_relays="wss://relay.damus.io")
+        with patch("infrastructure.nostr.identity.settings", mock_settings):
+            urls = manager.get_relay_urls()
+        assert urls == ["wss://relay.damus.io"]
+
+    def test_multiple_relays(self, manager):
+        mock_settings = MagicMock(nostr_relays="wss://a.com,wss://b.com, wss://c.com ")
+        with patch("infrastructure.nostr.identity.settings", mock_settings):
+            urls = manager.get_relay_urls()
+        assert urls == ["wss://a.com", "wss://b.com", "wss://c.com"]
+
+
+class TestBuildProfileEvent:
+    def test_kind_is_0(self, manager, kp):
+        mock_settings = MagicMock(
+            nostr_profile_name="Timmy",
+            nostr_profile_about="",
+            nostr_profile_picture="",
+            nostr_nip05="",
+        )
+        with patch("infrastructure.nostr.identity.settings", mock_settings):
+            ev = manager.build_profile_event(kp)
+        assert ev["kind"] == 0
+
+    def test_content_contains_name(self, manager, kp):
+        mock_settings = MagicMock(
+            nostr_profile_name="Timmy",
+            nostr_profile_about="A great AI agent",
+            nostr_profile_picture="",
+            nostr_nip05="",
+        )
+        with patch("infrastructure.nostr.identity.settings", mock_settings):
+            ev = manager.build_profile_event(kp)
+        profile = json.loads(ev["content"])
+        assert profile["name"] == "Timmy"
+
+    def test_nip05_included_when_set(self, manager, kp):
+        mock_settings = MagicMock(
+            nostr_profile_name="Timmy",
+            nostr_profile_about="",
+            nostr_profile_picture="",
+            nostr_nip05="timmy@tower.local",
+        )
+        with patch("infrastructure.nostr.identity.settings", mock_settings):
+            ev = manager.build_profile_event(kp)
+        profile = json.loads(ev["content"])
+        assert profile["nip05"] == "timmy@tower.local"
+
+    def test_nip05_omitted_when_empty(self, manager, kp):
+        mock_settings = MagicMock(
+            nostr_profile_name="Timmy",
+            nostr_profile_about="",
+            nostr_profile_picture="",
+            nostr_nip05="",
+        )
+        with patch("infrastructure.nostr.identity.settings", mock_settings):
+            ev = manager.build_profile_event(kp)
+        profile = json.loads(ev["content"])
+        assert "nip05" not in profile
+
+    def test_default_name_when_blank(self, manager, kp):
+        mock_settings = MagicMock(
+            nostr_profile_name="",
+            nostr_profile_about="",
+            nostr_profile_picture="",
+            nostr_nip05="",
+        )
+        with patch("infrastructure.nostr.identity.settings", mock_settings):
+            ev = manager.build_profile_event(kp)
+        profile = json.loads(ev["content"])
+        assert profile["name"] == "Timmy"  # default
+
+
+class TestBuildCapabilityEvent:
+    def test_kind_is_31990(self, manager, kp):
+        mock_settings = MagicMock(
+            nostr_profile_name="Timmy",
+            nostr_profile_about="",
+            nostr_profile_picture="",
+            nostr_nip05="",
+        )
+        with patch("infrastructure.nostr.identity.settings", mock_settings):
+            ev = manager.build_capability_event(kp)
+        assert ev["kind"] == 31990
+
+    def test_has_d_tag(self, manager, kp):
+        mock_settings = MagicMock(
+            nostr_profile_name="Timmy",
+            nostr_profile_about="",
+            nostr_profile_picture="",
+            nostr_nip05="",
+        )
+        with patch("infrastructure.nostr.identity.settings", mock_settings):
+            ev = manager.build_capability_event(kp)
+        d_tags = [t for t in ev["tags"] if t[0] == "d"]
+        assert d_tags
+        assert d_tags[0][1] == "timmy-mission-control"
+
+    def test_content_is_json(self, manager, kp):
+        mock_settings = MagicMock(
+            nostr_profile_name="Timmy",
+            nostr_profile_about="",
+            nostr_profile_picture="",
+            nostr_nip05="",
+        )
+        with patch("infrastructure.nostr.identity.settings", mock_settings):
+            ev = manager.build_capability_event(kp)
+        parsed = json.loads(ev["content"])
+        assert "name" in parsed
+        assert "capabilities" in parsed
+
+
+class TestAnnounce:
+    @pytest.mark.asyncio
+    async def test_returns_empty_result_when_no_privkey(self, manager):
+        mock_settings = MagicMock(
+            nostr_privkey="",
+            nostr_relays="",
+            nostr_profile_name="Timmy",
+            nostr_profile_about="",
+            nostr_profile_picture="",
+            nostr_nip05="",
+        )
+        with patch("infrastructure.nostr.identity.settings", mock_settings):
+            result = await manager.announce()
+        assert result.kind_0_ok is False
+        assert result.kind_31990_ok is False
+
+    @pytest.mark.asyncio
+    async def test_returns_empty_result_when_no_relays(self, manager, kp):
+        mock_settings = MagicMock(
+            nostr_privkey=kp.privkey_hex,
+            nostr_relays="",
+            nostr_profile_name="Timmy",
+            nostr_profile_about="",
+            nostr_profile_picture="",
+            nostr_nip05="",
+        )
+        with patch("infrastructure.nostr.identity.settings", mock_settings):
+            result = await manager.announce()
+        assert result.kind_0_ok is False
+
+    @pytest.mark.asyncio
+    async def test_publishes_kind0_and_kind31990(self, manager, kp):
+        mock_settings = MagicMock(
+            nostr_privkey=kp.privkey_hex,
+            nostr_relays="wss://relay.test",
+            nostr_profile_name="Timmy",
+            nostr_profile_about="Test agent",
+            nostr_profile_picture="",
+            nostr_nip05="timmy@test",
+        )
+        with (
+            patch("infrastructure.nostr.identity.settings", mock_settings),
+            patch(
+                "infrastructure.nostr.identity.publish_to_relays",
+                new=AsyncMock(return_value={"wss://relay.test": True}),
+            ) as mock_publish,
+        ):
+            result = await manager.announce()
+
+        assert mock_publish.call_count == 2  # kind 0 + kind 31990
+        assert result.kind_0_ok is True
+        assert result.kind_31990_ok is True
+        assert result.relay_results["wss://relay.test"] is True
+
+    @pytest.mark.asyncio
+    async def test_degrades_gracefully_on_relay_failure(self, manager, kp):
+        mock_settings = MagicMock(
+            nostr_privkey=kp.privkey_hex,
+            nostr_relays="wss://relay.test",
+            nostr_profile_name="Timmy",
+            nostr_profile_about="",
+            nostr_profile_picture="",
+            nostr_nip05="",
+        )
+        with (
+            patch("infrastructure.nostr.identity.settings", mock_settings),
+            patch(
+                "infrastructure.nostr.identity.publish_to_relays",
+                new=AsyncMock(return_value={"wss://relay.test": False}),
+            ),
+        ):
+            result = await manager.announce()
+
+        assert result.kind_0_ok is False
+        assert result.kind_31990_ok is False
+
+    @pytest.mark.asyncio
+    async def test_never_raises_on_exception(self, manager, kp):
+        mock_settings = MagicMock(
+            nostr_privkey=kp.privkey_hex,
+            nostr_relays="wss://relay.test",
+            nostr_profile_name="Timmy",
+            nostr_profile_about="",
+            nostr_profile_picture="",
+            nostr_nip05="",
+        )
+        with (
+            patch("infrastructure.nostr.identity.settings", mock_settings),
+            patch(
+                "infrastructure.nostr.identity.publish_to_relays",
+                new=AsyncMock(side_effect=Exception("relay exploded")),
+            ),
+        ):
+            # Must not raise
+            result = await manager.announce()
+        assert isinstance(result, AnnounceResult)

tests/unit/test_nostr_keypair.py

@@ -0,0 +1,126 @@
+"""Unit tests for infrastructure.nostr.keypair."""
+
+from __future__ import annotations
+
+import pytest
+
+from infrastructure.nostr.keypair import (
+    NostrKeypair,
+    _bech32_decode,
+    _bech32_encode,
+    generate_keypair,
+    load_keypair,
+    pubkey_from_privkey,
+)
+
+
+class TestGenerateKeypair:
+    def test_returns_nostr_keypair(self):
+        kp = generate_keypair()
+        assert isinstance(kp, NostrKeypair)
+
+    def test_privkey_hex_is_64_chars(self):
+        kp = generate_keypair()
+        assert len(kp.privkey_hex) == 64
+        assert all(c in "0123456789abcdef" for c in kp.privkey_hex)
+
+    def test_pubkey_hex_is_64_chars(self):
+        kp = generate_keypair()
+        assert len(kp.pubkey_hex) == 64
+        assert all(c in "0123456789abcdef" for c in kp.pubkey_hex)
+
+    def test_nsec_starts_with_nsec1(self):
+        kp = generate_keypair()
+        assert kp.nsec.startswith("nsec1")
+
+    def test_npub_starts_with_npub1(self):
+        kp = generate_keypair()
+        assert kp.npub.startswith("npub1")
+
+    def test_two_keypairs_are_different(self):
+        kp1 = generate_keypair()
+        kp2 = generate_keypair()
+        assert kp1.privkey_hex != kp2.privkey_hex
+        assert kp1.pubkey_hex != kp2.pubkey_hex
+
+    def test_privkey_bytes_matches_hex(self):
+        kp = generate_keypair()
+        assert kp.privkey_bytes == bytes.fromhex(kp.privkey_hex)
+
+    def test_pubkey_bytes_matches_hex(self):
+        kp = generate_keypair()
+        assert kp.pubkey_bytes == bytes.fromhex(kp.pubkey_hex)
+
+
+class TestLoadKeypair:
+    def test_round_trip_via_privkey_hex(self):
+        kp1 = generate_keypair()
+        kp2 = load_keypair(privkey_hex=kp1.privkey_hex)
+        assert kp2.privkey_hex == kp1.privkey_hex
+        assert kp2.pubkey_hex == kp1.pubkey_hex
+
+    def test_round_trip_via_nsec(self):
+        kp1 = generate_keypair()
+        kp2 = load_keypair(nsec=kp1.nsec)
+        assert kp2.privkey_hex == kp1.privkey_hex
+        assert kp2.pubkey_hex == kp1.pubkey_hex
+
+    def test_raises_if_both_supplied(self):
+        kp = generate_keypair()
+        with pytest.raises(ValueError, match="either"):
+            load_keypair(privkey_hex=kp.privkey_hex, nsec=kp.nsec)
+
+    def test_raises_if_neither_supplied(self):
+        with pytest.raises(ValueError, match="either"):
+            load_keypair()
+
+    def test_raises_on_invalid_hex(self):
+        with pytest.raises((ValueError, Exception)):
+            load_keypair(privkey_hex="zzzz")
+
+    def test_raises_on_wrong_length_hex(self):
+        with pytest.raises(ValueError):
+            load_keypair(privkey_hex="deadbeef")  # too short
+
+    def test_raises_on_wrong_hrp_bech32(self):
+        kp = generate_keypair()
+        # npub is bech32 but with hrp "npub", not "nsec"
+        with pytest.raises(ValueError):
+            load_keypair(nsec=kp.npub)
+
+    def test_npub_derived_from_privkey(self):
+        kp1 = generate_keypair()
+        kp2 = load_keypair(privkey_hex=kp1.privkey_hex)
+        assert kp2.npub == kp1.npub
+
+
+class TestPubkeyFromPrivkey:
+    def test_derives_correct_pubkey(self):
+        kp = generate_keypair()
+        derived = pubkey_from_privkey(kp.privkey_hex)
+        assert derived == kp.pubkey_hex
+
+    def test_is_deterministic(self):
+        kp = generate_keypair()
+        assert pubkey_from_privkey(kp.privkey_hex) == pubkey_from_privkey(kp.privkey_hex)
+
+
+class TestBech32:
+    def test_encode_decode_round_trip(self):
+        data = bytes(range(32))
+        encoded = _bech32_encode("test", data)
+        hrp, decoded = _bech32_decode(encoded)
+        assert hrp == "test"
+        assert decoded == data
+
+    def test_invalid_checksum_raises(self):
+        kp = generate_keypair()
+        mangled = kp.npub[:-1] + ("q" if kp.npub[-1] != "q" else "p")
+        with pytest.raises(ValueError, match="checksum"):
+            _bech32_decode(mangled)
+
+    def test_npub_roundtrip(self):
+        kp = generate_keypair()
+        hrp, pub = _bech32_decode(kp.npub)
+        assert hrp == "npub"
+        assert pub.hex() == kp.pubkey_hex