"""Nostr protocol implementation. https://github.com/nostr-protocol/nostr https://github.com/nostr-protocol/nips/blob/master/01.md https://github.com/nostr-protocol/nips#list Nostr Object key ids are NIP-21 nostr:... URIs. https://nips.nostr.com/21 """ import logging from google.cloud import ndb from google.cloud.ndb.query import OR from granary import as1 import granary.nostr from granary.nostr import ( bech32_prefix_for, id_to_uri, KIND_PROFILE, KIND_RELAYS, nip05_to_npub, uri_to_id, ) from oauth_dropins.webutil import flask_util from oauth_dropins.webutil import util from oauth_dropins.webutil.flask_util import get_required_param from oauth_dropins.webutil import util from oauth_dropins.webutil.util import add, json_dumps, json_loads from requests import RequestException import secp256k1 from websockets.exceptions import ConnectionClosedOK from websockets.sync.client import connect from werkzeug.exceptions import NotFound import common from common import ( DOMAIN_RE, DOMAINS, error, USER_AGENT, ) from flask_app import app import ids from models import Object, PROTOCOLS, Target, User from protocol import Protocol import web logger = logging.getLogger(__name__) class Nostr(User, Protocol): """Nostr class. Key id is NIP-21 nostr:npub... bech32 npub URI. https://github.com/nostr-protocol/nips/blob/master/19.md """ ABBREV = 'nostr' PHRASE = 'Nostr' LOGO_HTML = '' CONTENT_TYPE = 'application/json' HAS_COPIES = True DEFAULT_TARGET = 'wss://nos.lol' REQUIRES_AVATAR = True REQUIRES_NAME = True DEFAULT_ENABLED_PROTOCOLS = ('web',) SUPPORTED_AS1_TYPES = frozenset( tuple(as1.ACTOR_TYPES) + tuple(as1.POST_TYPES) + ('post', 'delete', 'undo') # no update/edit (I think?) + ('follow', 'like', 'share', 'stop-following') ) SUPPORTS_DMS = False # NIP-17 HTML_PROFILES = False relays = ndb.KeyProperty(kind='Object') """NIP-65 kind 10002 event with this user's relays.""" valid_nip05 = ndb.StringProperty() """NIP-05 identifier that we've resolved and verified.""" def _pre_put_hook(self): """Validates that the id is a bech32-encoded nostr:npub id.""" assert self.key.id().startswith('nostr:npub'), self.key.id() return super()._pre_put_hook() def hex_pubkey(self): """Returns the user's hex-encoded Nostr public secp256k1 key. Returns: str: """ return uri_to_id(self.key.id()) def npub(self): """Returns the user's bech32-encoded ActivityPub public secp256k1 key. Returns: str: """ return self.key.id().removeprefix('nostr:') @ndb.ComputedProperty def handle(self): """Returns the NIP-05 identity from the user's profile event.""" if nip05 := self.nip_05(): return nip05.removeprefix('_@') elif self.key: return self.key.id().removeprefix('nostr:') @ndb.ComputedProperty def status(self): if not self.obj or not self.obj.as1: return 'no-profile' if not self.valid_nip05 or self.valid_nip05 != self.nip_05(): return 'no-nip05' return super().status def nip_05(self): if self.obj and self.obj.nostr and self.obj.nostr.get('kind') == KIND_PROFILE: content = json_loads(self.obj.nostr.get('content', '{}')) if nip05 := content.get('nip05'): return nip05 def web_url(self): if self.obj_key: return granary.nostr.Nostr.user_url( self.obj_key.id().removeprefix("nostr:")) @classmethod def owns_id(cls, id): return id.startswith('nostr:') or bool(granary.nostr.is_bech32(id)) @classmethod def owns_handle(cls, handle, allow_internal=False): if not handle: return False # TODO: implement allow_internal? if (handle.startswith('npub') or cls.is_user_at_domain(handle, allow_internal=True)): return True if web.is_valid_domain(handle): return None # could be a _@ NIP-05 return False @classmethod def handle_to_id(cls, handle): if cls.owns_handle(handle) is False: return None elif handle.startswith('npub'): return handle return granary.nostr.nip05_to_npub(handle) @classmethod def bridged_web_url_for(cls, user, fallback=False): if not isinstance(user, cls) and user.obj: if nprofile := user.obj.get_copy(cls): return granary.nostr.Nostr.user_url(nprofile) @classmethod def target_for(cls, obj, shared=False): """Returns the first NIP-65 relay for the given object's author.""" if obj and (id := as1.get_owner(obj.as1)) and id.startswith('nostr:npub'): if user := Nostr.get_or_create(id, allow_opt_out=True): if user.relays and (relays := user.relays.get()): if relays.nostr: for tag in relays.nostr.get('tags', []): if tag[0] == 'r' and (len(tag) == 2 or tag[2] == 'write'): return tag[1] @classmethod def create_for(cls, user): """Creates a Nostr profile for a non-Nostr user. Args: user (models.User) """ assert not isinstance(user, cls) if npub := user.get_copy(cls): return # generate keypair if necessary, store npub as copy in user if not user.nostr_key_bytes: logger.info(f'generating Nostr keypair for {user.key}') privkey = secp256k1.PrivateKey() user.nostr_key_bytes = privkey.private_key logger.info(f'adding Nostr copy user {user.npub()} for {user.key}') user.add('copies', Target(uri='nostr:' + user.npub(), protocol='nostr')) user.put() if user.obj and any(copy.protocol == 'nostr' for copy in user.obj.copies): return # create Nostr profile (kind 0 event) if necessary if not user.obj or not user.obj.as1: user.reload_profile() if user.obj and not user.obj.get_copy(cls): cls.send(user.obj, cls.DEFAULT_TARGET, from_user=user) def reload_profile(self, **kwargs): """Reloads this user's kind 0 profile, NIP-65 relay list, and NIP-05 id. https://nips.nostr.com/1#kinds https://nips.nostr.com/65 https://nips.nostr.com/5 """ client = granary.nostr.Nostr() relay = self.target_for(self.obj) or self.DEFAULT_TARGET logger.debug(f'connecting to {relay}') with connect(relay, open_timeout=util.HTTP_TIMEOUT, close_timeout=util.HTTP_TIMEOUT) as websocket: events = client.query(websocket, { 'authors': [self.hex_pubkey()], 'kinds': [KIND_PROFILE, KIND_RELAYS], }) profile = relays = None for event in events: kind = event.get('kind') obj = Object(id=id_to_uri('nevent', event['id']), nostr=event, source_protocol='nostr') if kind == KIND_PROFILE and not profile: profile = obj self.obj_key = profile.put() elif kind == KIND_RELAYS and not relays: relays = obj self.relays = relays.put() if profile and relays: break # check NIP-05 self.valid_nip05 = None if nip05 := self.nip_05(): try: if nip05_to_npub(nip05) == self.npub(): self.valid_nip05 = nip05 except BaseException as e: code, _ = util.interpret_http_exception(e) if not code: logger.info(e) self.put() @classmethod def set_username(to_cls, user, username): """check NIP-05 DNS, then update profile event with nip05?""" if not user.is_enabled(to_cls): raise ValueError("First, you'll need to bridge your account into Nostr by following this account.") npub = user.get_copy(to_cls) username = username.removeprefix('@') # TODO logger.info(f'Setting Nostr NIP-05 for {user.key.id()} to {username}') raise NotImplementedError() @classmethod def fetch(cls, obj, **kwargs): """Fetches a Nostr event from a relay. Args: obj (models.Object): with the id to fetch. Fills data into the ``nostr`` property. kwargs: ignored Returns: bool: True if the object was fetched and populated successfully, False otherwise """ uri = obj.key.id() if not cls.owns_id(uri): logger.info(f"Nostr can't fetch {uri}") return False bech32_id = uri.removeprefix('nostr:') is_profile = bech32_id.startswith('npub') or bech32_id.startswith('nprofile') hex_id = uri_to_id(uri) filter = ({'authors': [hex_id], 'kinds': [KIND_PROFILE]} if is_profile else {'ids': [hex_id]}) client = granary.nostr.Nostr() relay = cls.target_for(obj) logger.debug(f'connecting to {relay}') with connect(relay, open_timeout=util.HTTP_TIMEOUT, close_timeout=util.HTTP_TIMEOUT) as websocket: events = client.query(websocket, filter) if not events: return False obj.nostr = events[0] return True @classmethod def _convert(to_cls, obj, from_user=None): """Converts a :class:`models.Object` to a Nostr event. Args: obj (models.Object) from_user (models.User): user this object is from Returns: dict: JSON Nostr event """ obj_as1 = obj.as1 translated = to_cls.translate_ids(obj_as1) # find first relay (target) for referenced user (follow of, in reply to, # repost of) if as1.object_type(obj_as1) in as1.CRUD_VERBS: obj_as1 = as1.get_object(obj_as1) remote_relay = '' if remote_obj := granary.nostr.Nostr().base_object(obj_as1): if id := remote_obj.get('id'): if id.startswith('nostr:npub'): obj = Object(our_as1={'objectType': 'person', 'id': id}) else: obj = Nostr.load(id) remote_relay = to_cls.target_for(obj) # convert! privkey = (from_user.nsec() if from_user and from_user.nostr_key_bytes else None) return granary.nostr.from_as1(translated, privkey=privkey, remote_relay=remote_relay) @classmethod def send(to_cls, obj, relay_url, from_user=None, **kwargs): """Sends an event to a relay.""" assert from_user assert from_user.nostr_key_bytes event = to_cls.convert(obj, from_user=from_user) assert event.get('pubkey') == from_user.hex_pubkey(), event assert event.get('sig'), event logger.debug(f'connecting to {relay_url}') with connect(relay_url, open_timeout=util.HTTP_TIMEOUT, close_timeout=util.HTTP_TIMEOUT) as websocket: try: msg = ['EVENT', event] logger.debug(f'{websocket.remote_address} <= {event}') websocket.send(json_dumps(msg)) resp = websocket.recv(timeout=util.HTTP_TIMEOUT) logger.debug(f'{websocket.remote_address} => {resp}') except ConnectionClosedOK as cc: logger.warning(cc) return False obj.copies = [copy for copy in obj.copies if copy.protocol != 'nostr'] uri = id_to_uri(bech32_prefix_for(event), event['id']) obj.add('copies', Target(uri=uri, protocol=to_cls.LABEL)) obj.put() return True @app.get('/.well-known/nostr.json') @flask_util.headers(common.CACHE_CONTROL) def nip_05(): """NIP-05 endpoint that serves handles for users bridged into Nostr. https://nips.nostr.com/5 Query params: name (str): should only contain a-z0-9-_. Returns a JSON object with: names: {: } relays: optional, {: [relay urls]} """ name = get_required_param('name') if (proto := Protocol.for_request()) and proto != Nostr: user = proto.query(OR(proto.handle == name, proto.handle_as_domain == name, proto.key == ndb.Key(proto, name), )).get() if user and user.is_enabled(Nostr): if npub := user.get_copy(Nostr): return { 'names': {name: uri_to_id(npub)}, } raise NotFound()