feat(v2): hand-rolled NIP-44 v2 crypto + reference-vector tests (#29 v1)

LNbits ships only NIP-04 (AES-CBC) in lnbits.utils.nostr.encrypt_content,
but the locked design at #29 (paired with lamassu-next#56) wires kind-30078
cassette config with NIP-44 v2 content per the privacy-by-default
architecture (dcd0874). Hand-rolling rather than adding a Python lib dep
per the plan-approval (option A) — keeps the impl auditable inline and
avoids pulling in a non-trivial dep tree.

nip44.py covers the full envelope:
  - get_conversation_key — ECDH x-coord + HKDF-extract with salt b"nip44-v2"
  - encrypt_with_conversation_key / decrypt_with_conversation_key — low-level,
    nonce-controllable for testing pinned vectors
  - encrypt_for / decrypt_from — high-level pair-keyed API (the shape app
    code reaches for)
  - _pad / _unpad — NIP-44 v2 length-prefixed padding scheme
  - HMAC-SHA256 verification on nonce || ciphertext, constant-time compare
    via hmac.compare_digest
  - Typed errors (Nip44VersionError / Nip44MacError / Nip44LengthError)
    so callers can distinguish tamper from corruption from spec mismatch

Stack: coincurve for ECDH (already a transitive lnbits dep), cryptography
for ChaCha20 + HKDF-expand (also already there). No new pyproject deps.

34 tests in tests/test_nip44_v2.py, three layers:
  1. Pinned reference vector — conversation_key for (sec=1, sec=2) matches
     the canonical paulmillr/nip44 published value
     (c41c775356fd92eadc63ff5a0dc1da211b268cbea22316767095b2871ea1412d).
     Regression-fails loudly if key derivation drifts.
  2. Round-trip + tamper detection — encrypt/decrypt across plaintext
     lengths (1, 32, 33, 1000, 5000, 65535 bytes); flipped MAC byte;
     flipped ciphertext byte; flipped nonce byte; wrong recipient privkey;
     version-byte rejection; padding-formula spot checks.
  3. Cross-impl byte-compat — placeholder test_decrypts_bitspire_sample
     marked @pytest.mark.skip, pending bitspire posting a sample event
     encrypted on their nostr-tools side to the coord log (per the
     2026-05-30T15:55Z entry). Wire that fixture and unskip when posted.

Total: 132 passed, 1 skipped (cross-test fixture pending), 1 pre-existing
async-plugin failure unchanged.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Padreug 2026-05-30 18:10:30 +02:00
commit da07bae554
2 changed files with 543 additions and 0 deletions

271
nip44.py Normal file
View file

@ -0,0 +1,271 @@
"""
NIP-44 v2 versioned encrypted payloads (https://github.com/nostr-protocol/nips/blob/master/44.md).
Hand-rolled because lnbits ships only NIP-04 (AES-CBC) in `lnbits.utils.nostr.encrypt_content`,
and the locked design at aiolabs/satmachineadmin#29 (paired with lamassu-next#56) wires
cassette config over kind-30078 with NIP-44 v2 encrypted content. Adding a Python NIP-44
v2 lib dep was an option per the plan; chose the hand-roll path to stay dep-light and
keep the impl auditable inline.
Two safety nets keep this honest:
1. tests/test_nip44_v2.py runs reference vectors + round-trip + tamper-detection.
2. bitspire posts a sample event encrypted on their nostr-tools side to the coord log;
test_decrypts_bitspire_sample_event_from_coord_log cross-checks our impl against
theirs by decrypting that event with a known privkey.
Wire format (per spec):
payload = base64( 0x02 || nonce (32B) || ciphertext (var) || mac (32B) )
Key derivation:
conversation_key = HKDF-extract(salt=b"nip44-v2", IKM=ecdh_shared_x) # 32B PRK, stable per pair
per-message:
nonce = csprng(32 bytes)
temp = HKDF-expand(PRK=conversation_key, info=nonce, L=76)
chacha_key = temp[0:32]
chacha_nonce = temp[32:44]
hmac_key = temp[44:76]
Padding scheme (NIP-44 v2 length-prefixed, variable-chunk):
padded = uint16_be(len(plaintext)) || plaintext || zeros
such that 2 + padded_data_len matches a fixed step.
"""
from __future__ import annotations
import base64
import hashlib
import hmac as hmac_stdlib
import os
import struct
from typing import Optional
import coincurve
from cryptography.hazmat.primitives import hashes, hmac
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms
from cryptography.hazmat.primitives.kdf.hkdf import HKDFExpand
# Spec constants.
_VERSION = 0x02
_HKDF_SALT = b"nip44-v2"
_MIN_PLAINTEXT_LEN = 1
_MAX_PLAINTEXT_LEN = 65535
_NONCE_LEN = 32
_MAC_LEN = 32
_MIN_PAYLOAD_LEN = 1 + _NONCE_LEN + (2 + 32) + _MAC_LEN # version + nonce + min padded + mac
_MAX_PAYLOAD_LEN = 1 + _NONCE_LEN + (2 + 65536) + _MAC_LEN
class Nip44Error(Exception):
"""Generic NIP-44 v2 envelope error. Subclasses distinguish failure modes."""
class Nip44VersionError(Nip44Error):
"""First payload byte was not 0x02. Could be a NIP-04 envelope, a v1 NIP-44, or garbage."""
class Nip44MacError(Nip44Error):
"""HMAC verification failed — payload was tampered, wrong conversation key, or corrupted in transit."""
class Nip44LengthError(Nip44Error):
"""Plaintext or payload length outside the spec-allowed range, or padding header lies."""
# =============================================================================
# Padding (NIP-44 v2)
# =============================================================================
def _calc_padded_len(plaintext_len: int) -> int:
"""Per NIP-44 v2 padding scheme:
if L <= 32: padded_len = 32
else: chunk = max(32, next_power_2(L-1) // 8); padded_len = chunk * ((L-1) // chunk + 1)
"""
if plaintext_len <= 32:
return 32
next_power = 1 << (plaintext_len - 1).bit_length()
chunk = max(32, next_power // 8)
return chunk * ((plaintext_len - 1) // chunk + 1)
def _pad(plaintext: bytes) -> bytes:
"""Prefix uint16_be length + plaintext + zero-fill to the NIP-44 v2 boundary."""
n = len(plaintext)
if n < _MIN_PLAINTEXT_LEN or n > _MAX_PLAINTEXT_LEN:
raise Nip44LengthError(
f"plaintext length {n} outside [{_MIN_PLAINTEXT_LEN}, {_MAX_PLAINTEXT_LEN}]"
)
padded_data_len = _calc_padded_len(n)
zeros = b"\x00" * (padded_data_len - n)
return struct.pack(">H", n) + plaintext + zeros
def _unpad(padded: bytes) -> bytes:
"""Strip the uint16_be length prefix and zero padding. Validates that the
declared length is consistent with the padded payload (rejects a forged
length prefix that would slice past the buffer or imply a different
padded_data_len than what we received)."""
if len(padded) < 2:
raise Nip44LengthError("padded payload too short to hold length prefix")
declared_len = struct.unpack(">H", padded[0:2])[0]
if declared_len < _MIN_PLAINTEXT_LEN or declared_len > _MAX_PLAINTEXT_LEN:
raise Nip44LengthError(f"declared plaintext length {declared_len} out of range")
if len(padded) != 2 + _calc_padded_len(declared_len):
raise Nip44LengthError(
f"padded buffer length {len(padded)} doesn't match the calculated padding "
f"for declared length {declared_len}"
)
return padded[2 : 2 + declared_len]
# =============================================================================
# Conversation + message-key derivation
# =============================================================================
def get_conversation_key(privkey_hex: str, pubkey_hex: str) -> bytes:
"""Derive the per-pair stable conversation key (PRK) used for all messages
between sender (privkey) and recipient (pubkey).
Steps:
shared_x = ECDH(privkey, pubkey).x # 32 bytes, x-coordinate
prk = HKDF-extract(salt=b"nip44-v2", IKM=shared_x)
coincurve's `.multiply(secret).format(compressed=True)[1:]` strips the
leading 0x02/0x03 parity byte to return the raw x-coord same trick
`lnbits.utils.nostr.encrypt_content` uses for NIP-04.
"""
sender = coincurve.PrivateKey(bytes.fromhex(privkey_hex))
recipient_pub = coincurve.PublicKey(b"\x02" + bytes.fromhex(pubkey_hex))
shared_x = recipient_pub.multiply(sender.secret).format(compressed=True)[1:]
# HKDF-extract is HMAC-SHA256(key=salt, msg=ikm) per RFC 5869.
return hmac_stdlib.new(_HKDF_SALT, shared_x, hashlib.sha256).digest()
def _derive_message_keys(
conversation_key: bytes, nonce: bytes
) -> tuple[bytes, bytes, bytes]:
"""Per-message key expansion: HKDF-expand(PRK=conversation_key, info=nonce, L=76).
Returns (chacha_key 32B, chacha_nonce 12B, hmac_key 32B)."""
hkdf = HKDFExpand(algorithm=hashes.SHA256(), length=76, info=nonce)
okm = hkdf.derive(conversation_key)
return okm[0:32], okm[32:44], okm[44:76]
def _hmac_aad(hmac_key: bytes, nonce: bytes, ciphertext: bytes) -> bytes:
"""HMAC-SHA256(key=hmac_key, msg=nonce || ciphertext). Returns 32-byte MAC."""
h = hmac.HMAC(hmac_key, hashes.SHA256())
h.update(nonce)
h.update(ciphertext)
return h.finalize()
def _chacha20(key: bytes, nonce: bytes, data: bytes) -> bytes:
"""ChaCha20 stream cipher (symmetric: encrypt == decrypt). Used both directions.
The `cryptography` lib's `algorithms.ChaCha20(key, nonce)` expects a
16-byte nonce arg: a 4-byte little-endian initial counter prefix +
12-byte actual nonce. NIP-44 v2 starts the counter at 0 and uses the
HKDF-derived 12-byte chacha_nonce, so we prefix four zero bytes here.
"""
if len(nonce) != 12:
raise Nip44LengthError(
f"chacha_nonce must be 12 bytes (NIP-44 v2), got {len(nonce)}"
)
cipher = Cipher(algorithms.ChaCha20(key, b"\x00\x00\x00\x00" + nonce), mode=None)
return cipher.encryptor().update(data)
# =============================================================================
# Public API — low-level (nonce-controllable for testability)
# =============================================================================
def encrypt_with_conversation_key(
plaintext: str,
conversation_key: bytes,
*,
nonce: Optional[bytes] = None,
) -> str:
"""Encrypt `plaintext` under a precomputed `conversation_key` (32B PRK).
`nonce` is 32 random bytes when omitted (the production path). Tests pass
it explicitly to assert pinned reference vectors.
Returns the base64-encoded payload string suitable as a Nostr event's
`content` field for kind-30078 (and any other kind that uses NIP-44 v2).
"""
if nonce is None:
nonce = os.urandom(_NONCE_LEN)
elif len(nonce) != _NONCE_LEN:
raise Nip44LengthError(f"nonce must be exactly {_NONCE_LEN} bytes")
padded = _pad(plaintext.encode("utf-8"))
chacha_key, chacha_nonce, hmac_key = _derive_message_keys(conversation_key, nonce)
ciphertext = _chacha20(chacha_key, chacha_nonce, padded)
mac = _hmac_aad(hmac_key, nonce, ciphertext)
return base64.b64encode(
bytes([_VERSION]) + nonce + ciphertext + mac
).decode("ascii")
def decrypt_with_conversation_key(payload_b64: str, conversation_key: bytes) -> str:
"""Decrypt a NIP-44 v2 payload using a precomputed `conversation_key`.
Raises:
Nip44VersionError payload's first byte isn't 0x02
Nip44LengthError payload too short / too long / declared length lies
Nip44MacError HMAC verification failed (tamper, wrong key, corruption)
"""
try:
raw = base64.b64decode(payload_b64, validate=True)
except Exception as exc: # noqa: BLE001 — we want any base64 failure surfaced uniformly
raise Nip44LengthError(f"payload is not valid base64: {exc}") from exc
if len(raw) < _MIN_PAYLOAD_LEN or len(raw) > _MAX_PAYLOAD_LEN:
raise Nip44LengthError(f"payload length {len(raw)} outside valid range")
if raw[0] != _VERSION:
raise Nip44VersionError(f"unsupported NIP-44 version: 0x{raw[0]:02x}")
nonce = raw[1 : 1 + _NONCE_LEN]
mac_received = raw[-_MAC_LEN:]
ciphertext = raw[1 + _NONCE_LEN : -_MAC_LEN]
chacha_key, chacha_nonce, hmac_key = _derive_message_keys(conversation_key, nonce)
mac_expected = _hmac_aad(hmac_key, nonce, ciphertext)
# constant-time compare to avoid timing-leak in MAC verification
if not hmac_stdlib.compare_digest(mac_received, mac_expected):
raise Nip44MacError("HMAC verification failed")
padded = _chacha20(chacha_key, chacha_nonce, ciphertext)
plaintext_bytes = _unpad(padded)
return plaintext_bytes.decode("utf-8")
# =============================================================================
# Public API — high-level (pair-keyed, the call shape app code reaches for)
# =============================================================================
def encrypt_for(
plaintext: str,
sender_privkey_hex: str,
recipient_pubkey_hex: str,
*,
nonce: Optional[bytes] = None,
) -> str:
"""Encrypt `plaintext` from the sender (holding the privkey) to the recipient
(identified by pubkey). The recipient can decrypt with `decrypt_from(
payload, recipient_privkey_hex, sender_pubkey_hex)` symmetric on the
conversation key, which is the same derived value from either side."""
conversation_key = get_conversation_key(sender_privkey_hex, recipient_pubkey_hex)
return encrypt_with_conversation_key(plaintext, conversation_key, nonce=nonce)
def decrypt_from(
payload_b64: str, recipient_privkey_hex: str, sender_pubkey_hex: str
) -> str:
"""Decrypt a payload that the recipient (holding the privkey) received from
the sender (identified by pubkey)."""
conversation_key = get_conversation_key(recipient_privkey_hex, sender_pubkey_hex)
return decrypt_with_conversation_key(payload_b64, conversation_key)