khatru

README

khatru, a relay framework

Khatru makes it easy to write very very custom relays:

• custom event or filter acceptance policies
• custom AUTH handlers
• custom storage and pluggable databases
• custom webpages and other HTTP handlers

Here's a sample:

package main

import (
	"context"
	"fmt"
	"log"
	"net/http"

	"github.com/fiatjaf/khatru"
	"github.com/nbd-wtf/go-nostr"
)

func main() {
	// create the relay instance
	relay := khatru.NewRelay()

	// set up some basic properties (will be returned on the NIP-11 endpoint)
	relay.Info.Name = "my relay"
	relay.Info.PubKey = "79be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798"
	relay.Info.Description = "this is my custom relay"
	relay.Info.Icon = "https://external-content.duckduckgo.com/iu/?u=https%3A%2F%2Fliquipedia.net%2Fcommons%2Fimages%2F3%2F35%2FSCProbe.jpg&f=1&nofb=1&ipt=0cbbfef25bce41da63d910e86c3c343e6c3b9d63194ca9755351bb7c2efa3359&ipo=images"

	// you must bring your own storage scheme -- if you want to have any
	store := make(map[string]*nostr.Event, 120)

	// set up the basic relay functions
	relay.StoreEvent = append(relay.StoreEvent,
		func(ctx context.Context, event *nostr.Event) error {
			store[event.ID] = event
			return nil
		},
	)
	relay.QueryEvents = append(relay.QueryEvents,
		func(ctx context.Context, filter nostr.Filter) (chan *nostr.Event, error) {
			ch := make(chan *nostr.Event)
			go func() {
				for _, evt := range store {
					if filter.Matches(evt) {
						ch <- evt
					}
				}
				close(ch)
			}()
			return ch, nil
		},
	)
	relay.DeleteEvent = append(relay.DeleteEvent,
		func(ctx context.Context, event *nostr.Event) error {
			delete(store, event.ID)
			return nil
		},
	)

	// there are many other configurable things you can set
	relay.RejectEvent = append(relay.RejectEvent,
		// built-in policies
		policies.ValidateKind,

		// define your own policies
		policies.PreventLargeTags(100),
		func(ctx context.Context, event *nostr.Event) (reject bool, msg string) {
			if event.PubKey == "fa984bd7dbb282f07e16e7ae87b26a2a7b9b90b7246a44771f0cf5ae58018f52" {
				return true, "we don't allow this person to write here"
			}
			return false, "" // anyone else can
		},
	)

	// you can request auth by rejecting an event or a request with the prefix "auth-required: "
	relay.RejectFilter = append(relay.RejectFilter,
		// built-in policies
		policies.NoComplexFilters,

		// define your own policies
		func(ctx context.Context, filter nostr.Filter) (reject bool, msg string) {
			if pubkey := khatru.GetAuthed(ctx); pubkey != "" {
				log.Printf("request from %s\n", pubkey)
				return false, ""
			}
			return true, "auth-required: only authenticated users can read from this relay"
			// (this will cause an AUTH message to be sent and then a CLOSED message such that clients can
			//  authenticate and then request again)
		},
	)
	// check the docs for more goodies!

	mux := relay.Router()
	// set up other http handlers
	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("content-type", "text/html")
		fmt.Fprintf(w, `<b>welcome</b> to my relay!`)
	})

	// start the server
	fmt.Println("running on :3334")
	http.ListenAndServe(":3334", relay)
}

But I don't want to write my own database!

Fear no more. Using the https://github.com/fiatjaf/eventstore module you get a bunch of compatible databases out of the box and you can just plug them into your relay. For example, sqlite:

	db := sqlite3.SQLite3Backend{DatabaseURL: "/tmp/khatru-sqlite-tmp"}
	if err := db.Init(); err != nil {
		panic(err)
	}

	relay.StoreEvent = append(relay.StoreEvent, db.SaveEvent)
	relay.QueryEvents = append(relay.QueryEvents, db.QueryEvents)
	relay.CountEvents = append(relay.CountEvents, db.CountEvents)
	relay.DeleteEvent = append(relay.DeleteEvent, db.DeleteEvent)
	relay.ReplaceEvent = append(relay.ReplaceEvent, db.ReplaceEvent)

But I don't want to write a bunch of custom policies!

Fear no more. We have a bunch of common policies written in the github.com/fiatjaf/khatru/policies package and also a handpicked selection of base sane defaults, which you can apply with:

	policies.ApplySaneDefaults(relay)

Contributions to this are very much welcomed.

Documentation

Index

• Variables
• func GetAuthed(ctx context.Context) string
• func GetIP(ctx context.Context) string
• func GetIPFromRequest(r *http.Request) string
• func GetSubscriptionID(ctx context.Context) string
• func IsInternalCall(ctx context.Context) bool
• func RequestAuth(ctx context.Context)
• type NegentropySession
• type Relay
  • func NewRelay() *Relay
  • func (rl *Relay) AddEvent(ctx context.Context, evt *nostr.Event) (skipBroadcast bool, writeError error)
  • func (rl *Relay) BroadcastEvent(evt *nostr.Event)
  • func (rl *Relay) GetListeningFilters() []nostr.Filter
  • func (rl *Relay) HandleNIP11(w http.ResponseWriter, r *http.Request)
  • func (rl *Relay) HandleNIP86(w http.ResponseWriter, r *http.Request)
  • func (rl *Relay) HandleWebsocket(w http.ResponseWriter, r *http.Request)
  • func (rl *Relay) Router() *http.ServeMux
  • func (rl *Relay) ServeHTTP(w http.ResponseWriter, r *http.Request)
  • func (rl *Relay) SetRouter(mux *http.ServeMux)
  • func (rl *Relay) Shutdown(ctx context.Context)
  • func (rl *Relay) Start(host string, port int, started ...chan bool) error
• type RelayManagementAPI
• type Route
• type Router
  • func NewRouter() *Router
  • func (rr *Router) Route() routeBuilder
• type WebSocket
  • func GetConnection(ctx context.Context) *WebSocket
  • func (ws *WebSocket) WriteJSON(any any) error
  • func (ws *WebSocket) WriteMessage(t int, b []byte) error

Variables

var ErrSubscriptionClosedByClient = errors.New("subscription closed by client")

Functions

func GetAuthed

func GetAuthed(ctx context.Context) string

func GetIP

func GetIP(ctx context.Context) string

func GetIPFromRequest

func GetIPFromRequest(r *http.Request) string

func GetSubscriptionID

func GetSubscriptionID(ctx context.Context) string

func IsInternalCall

func IsInternalCall(ctx context.Context) bool

IsInternalCall returns true when a call to QueryEvents, for example, is being made because of a deletion or expiration request.

func RequestAuth

func RequestAuth(ctx context.Context)

Types

type NegentropySession

type NegentropySession struct {
	// contains filtered or unexported fields
}

type Relay

type Relay struct {
	// setting this variable overwrites the hackish workaround we do to try to figure out our own base URL
	ServiceURL string

	// hooks that will be called at various times
	RejectEvent               []func(ctx context.Context, event *nostr.Event) (reject bool, msg string)
	OverwriteDeletionOutcome  []func(ctx context.Context, target *nostr.Event, deletion *nostr.Event) (acceptDeletion bool, msg string)
	StoreEvent                []func(ctx context.Context, event *nostr.Event) error
	ReplaceEvent              []func(ctx context.Context, event *nostr.Event) error
	DeleteEvent               []func(ctx context.Context, event *nostr.Event) error
	OnEventSaved              []func(ctx context.Context, event *nostr.Event)
	OnEphemeralEvent          []func(ctx context.Context, event *nostr.Event)
	RejectFilter              []func(ctx context.Context, filter nostr.Filter) (reject bool, msg string)
	RejectCountFilter         []func(ctx context.Context, filter nostr.Filter) (reject bool, msg string)
	OverwriteFilter           []func(ctx context.Context, filter *nostr.Filter)
	QueryEvents               []func(ctx context.Context, filter nostr.Filter) (chan *nostr.Event, error)
	CountEvents               []func(ctx context.Context, filter nostr.Filter) (int64, error)
	CountEventsHLL            []func(ctx context.Context, filter nostr.Filter, offset int) (int64, *hyperloglog.HyperLogLog, error)
	RejectConnection          []func(r *http.Request) bool
	OnConnect                 []func(ctx context.Context)
	OnDisconnect              []func(ctx context.Context)
	OverwriteRelayInformation []func(ctx context.Context, r *http.Request, info nip11.RelayInformationDocument) nip11.RelayInformationDocument
	OverwriteResponseEvent    []func(ctx context.Context, event *nostr.Event)
	PreventBroadcast          []func(ws *WebSocket, event *nostr.Event) bool

	// setting up handlers here will enable these methods
	ManagementAPI RelayManagementAPI

	// editing info will affect the NIP-11 responses
	Info *nip11.RelayInformationDocument

	// Default logger, as set by NewServer, is a stdlib logger prefixed with "[khatru-relay] ",
	// outputting to stderr.
	Log *log.Logger

	// set this to true to support negentropy
	Negentropy bool

	// in case you call Server.Start
	Addr string

	// websocket options
	WriteWait      time.Duration // Time allowed to write a message to the peer.
	PongWait       time.Duration // Time allowed to read the next pong message from the peer.
	PingPeriod     time.Duration // Send pings to peer with this period. Must be less than pongWait.
	MaxMessageSize int64         // Maximum message size allowed from peer.
	// contains filtered or unexported fields
}

func NewRelay

func NewRelay() *Relay

func (*Relay) AddEvent

func (rl *Relay) AddEvent(ctx context.Context, evt *nostr.Event) (skipBroadcast bool, writeError error)

AddEvent sends an event through then normal add pipeline, as if it was received from a websocket.

func (*Relay) BroadcastEvent

func (rl *Relay) BroadcastEvent(evt *nostr.Event)

BroadcastEvent emits an event to all listeners whose filters' match, skipping all filters and actions it also doesn't attempt to store the event or trigger any reactions or callbacks

func (*Relay) GetListeningFilters

func (rl *Relay) GetListeningFilters() []nostr.Filter

func (*Relay) HandleNIP11

func (rl *Relay) HandleNIP11(w http.ResponseWriter, r *http.Request)

func (*Relay) HandleNIP86

func (rl *Relay) HandleNIP86(w http.ResponseWriter, r *http.Request)

func (*Relay) HandleWebsocket

func (rl *Relay) HandleWebsocket(w http.ResponseWriter, r *http.Request)

func (*Relay) Router

func (rl *Relay) Router() *http.ServeMux

func (*Relay) ServeHTTP

func (rl *Relay) ServeHTTP(w http.ResponseWriter, r *http.Request)

ServeHTTP implements http.Handler interface.

func (*Relay) SetRouter

func (rl *Relay) SetRouter(mux *http.ServeMux)

func (*Relay) Shutdown

func (rl *Relay) Shutdown(ctx context.Context)

Shutdown sends a websocket close control message to all connected clients.

func (*Relay) Start

func (rl *Relay) Start(host string, port int, started ...chan bool) error

Start creates an http server and starts listening on given host and port.

type RelayManagementAPI

type RelayManagementAPI struct {
	RejectAPICall []func(ctx context.Context, mp nip86.MethodParams) (reject bool, msg string)

	BanPubKey                   func(ctx context.Context, pubkey string, reason string) error
	ListBannedPubKeys           func(ctx context.Context) ([]nip86.PubKeyReason, error)
	AllowPubKey                 func(ctx context.Context, pubkey string, reason string) error
	ListAllowedPubKeys          func(ctx context.Context) ([]nip86.PubKeyReason, error)
	ListEventsNeedingModeration func(ctx context.Context) ([]nip86.IDReason, error)
	AllowEvent                  func(ctx context.Context, id string, reason string) error
	BanEvent                    func(ctx context.Context, id string, reason string) error
	ListBannedEvents            func(ctx context.Context) ([]nip86.IDReason, error)
	ListAllowedEvents           func(ctx context.Context) ([]nip86.IDReason, error)
	ChangeRelayName             func(ctx context.Context, name string) error
	ChangeRelayDescription      func(ctx context.Context, desc string) error
	ChangeRelayIcon             func(ctx context.Context, icon string) error
	AllowKind                   func(ctx context.Context, kind int) error
	DisallowKind                func(ctx context.Context, kind int) error
	ListAllowedKinds            func(ctx context.Context) ([]int, error)
	ListDisAllowedKinds         func(ctx context.Context) ([]int, error)
	BlockIP                     func(ctx context.Context, ip net.IP, reason string) error
	UnblockIP                   func(ctx context.Context, ip net.IP, reason string) error
	ListBlockedIPs              func(ctx context.Context) ([]nip86.IPReason, error)
	Stats                       func(ctx context.Context) (nip86.Response, error)
	GrantAdmin                  func(ctx context.Context, pubkey string, methods []string) error
	RevokeAdmin                 func(ctx context.Context, pubkey string, methods []string) error
	Generic                     func(ctx context.Context, request nip86.Request) (nip86.Response, error)
}

type Route

type Route struct {
	// contains filtered or unexported fields
}

type Router

type Router struct{ *Relay }

func NewRouter

func NewRouter() *Router

func (*Router) Route

func (rr *Router) Route() routeBuilder

type WebSocket

type WebSocket struct {

	// original request
	Request *http.Request

	// this Context will be canceled whenever the connection is closed from the client side or server-side.
	Context context.Context

	// nip42
	Challenge       string
	AuthedPublicKey string
	Authed          chan struct{}
	// contains filtered or unexported fields
}

func GetConnection

func GetConnection(ctx context.Context) *WebSocket

func (*WebSocket) WriteJSON

func (ws *WebSocket) WriteJSON(any any) error

func (*WebSocket) WriteMessage

func (ws *WebSocket) WriteMessage(t int, b []byte) error