reorganise realy.Server and make listeners non-singleton

realy/addEvent.go

@@ -0,0 +1,70 @@
+package realy
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+	"strings"
+
+	"realy.lol/event"
+	"realy.lol/normalize"
+	"realy.lol/realy/listeners"
+	"realy.lol/relay"
+	"realy.lol/relay/wrapper"
+	"realy.lol/store"
+)
+
+func (s *Server) addEvent(c cx, rl relay.I, ev *event.T, hr *http.Request, origin st,
+	authedPubkey by) (accepted bo, message by) {
+	if ev == nil {
+		return false, normalize.Invalid.F("empty event")
+	}
+	sto := rl.Storage(c)
+	wrap := &wrapper.Relay{I: sto}
+	advancedSaver, _ := sto.(relay.AdvancedSaver)
+	accept, notice, after := rl.AcceptEvent(c, ev, hr, origin, authedPubkey)
+	if !accept {
+		return false, normalize.Blocked.F(notice)
+	}
+	if ev.Tags.ContainsProtectedMarker() {
+		if len(authedPubkey) == 0 || !equals(ev.PubKey, authedPubkey) {
+			return false,
+				by(fmt.Sprintf("event with relay marker tag '-' (nip-70 protected event) "+
+					"may only be published by matching npub: %0x is not %0x",
+					authedPubkey, ev.PubKey))
+		}
+	}
+	if ev.Kind.IsEphemeral() {
+	} else {
+		if advancedSaver != nil {
+			advancedSaver.BeforeSave(c, ev)
+		}
+		if saveErr := wrap.Publish(c, ev); chk.E(saveErr) {
+			if errors.Is(saveErr, store.ErrDupEvent) {
+				return false, normalize.Error.F(saveErr.Error())
+			}
+			errmsg := saveErr.Error()
+			if listeners.NIP20prefixmatcher.MatchString(errmsg) {
+				if strings.Contains(errmsg, "tombstone") {
+					return false, normalize.Blocked.F("event was deleted, not storing it again")
+				}
+				return false, normalize.Error.F(errmsg)
+			} else {
+				return false, normalize.Error.F("failed to save (%s)", errmsg)
+			}
+		}
+		if advancedSaver != nil {
+			advancedSaver.AfterSave(ev)
+		}
+	}
+	var authRequired bo
+	if ar, ok := rl.(relay.Authenticator); ok {
+		authRequired = ar.AuthEnabled()
+	}
+	if after != nil {
+		after()
+	}
+	s.listeners.NotifyListeners(authRequired, ev)
+	accepted = true
+	return
+}

realy/handleAdmin.go

@@ -0,0 +1,105 @@
+package realy
+
+import (
+	"crypto/subtle"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+
+	"realy.lol/cmd/realy/app"
+	"realy.lol/context"
+	"realy.lol/hex"
+	"realy.lol/sha256"
+)
+
+func (s *Server) auth(r *http.Request) (authed bo) {
+	if s.adminUser == "" || s.adminPass == "" {
+		// disallow this if it hasn't been configured, the default values are empty.
+		return
+	}
+	username, password, ok := r.BasicAuth()
+	if ok {
+		usernameHash := sha256.Sum256(by(username))
+		passwordHash := sha256.Sum256(by(password))
+		expectedUsernameHash := sha256.Sum256(by(s.adminUser))
+		expectedPasswordHash := sha256.Sum256(by(s.adminPass))
+		usernameMatch := subtle.ConstantTimeCompare(usernameHash[:],
+			expectedUsernameHash[:]) == 1
+		passwordMatch := subtle.ConstantTimeCompare(passwordHash[:],
+			expectedPasswordHash[:]) == 1
+		if usernameMatch && passwordMatch {
+			return true
+		}
+	}
+	return
+}
+
+func (s *Server) unauthorized(w http.ResponseWriter) {
+	w.Header().Set("WWW-Authenticate", `Basic realm="restricted", charset="UTF-8"`)
+	http.Error(w, "Unauthorized", http.StatusUnauthorized)
+	fmt.Fprintf(w, "you may have not configured your admin username/password")
+}
+
+func (s *Server) handleAdmin(w http.ResponseWriter, r *http.Request) {
+	switch {
+	case strings.HasPrefix(r.URL.Path, "/export"):
+		if ok := s.auth(r); !ok {
+			s.unauthorized(w)
+			return
+		}
+		log.I.F("export of event data requested on admin port")
+		sto := s.relay.Storage(context.Bg())
+		if strings.Count(r.URL.Path, "/") > 1 {
+			split := strings.Split(r.URL.Path, "/")
+			if len(split) != 3 {
+				fprintf(w, "incorrectly formatted export parameter: '%s'", r.URL.Path)
+				return
+			}
+			switch split[2] {
+			case "users":
+				if rl, ok := s.relay.(*app.Relay); ok {
+					follows := make([]by, 0, len(rl.Followed))
+					for f := range rl.Followed {
+						follows = append(follows, by(f))
+					}
+					sto.Export(s.Ctx, w, follows...)
+				}
+			default:
+				var exportPubkeys []by
+				pubkeys := strings.Split(split[2], "-")
+				for _, pubkey := range pubkeys {
+					pk, err := hex.Dec(pubkey)
+					if err != nil {
+						log.E.F("invalid public key '%s' in parameters", pubkey)
+						continue
+					}
+					exportPubkeys = append(exportPubkeys, pk)
+				}
+				sto.Export(s.Ctx, w, exportPubkeys...)
+			}
+		} else {
+			sto.Export(s.Ctx, w)
+		}
+	case strings.HasPrefix(r.URL.Path, "/import"):
+		if ok := s.auth(r); !ok {
+			s.unauthorized(w)
+			return
+		}
+		log.I.F("import of event data requested on admin port %s", r.RequestURI)
+		sto := s.relay.Storage(context.Bg())
+		read := io.LimitReader(r.Body, r.ContentLength)
+		sto.Import(read)
+	case strings.HasPrefix(r.URL.Path, "/shutdown"):
+		if ok := s.auth(r); !ok {
+			s.unauthorized(w)
+			return
+		}
+		fprintf(w, "shutting down")
+		defer chk.E(r.Body.Close())
+		s.Shutdown()
+	default:
+		fprintf(w, "todo: realy web interface page\n\n")
+		s.handleRelayInfo(w, r)
+	}
+}

realy/handleAuth.go

@@ -0,0 +1,50 @@
+package realy
+
+import (
+	"realy.lol/auth"
+	"realy.lol/envelopes/authenvelope"
+	"realy.lol/envelopes/okenvelope"
+	"realy.lol/normalize"
+	"realy.lol/relay"
+	"realy.lol/web"
+)
+
+func (s *Server) handleAuth(ws *web.Socket, req by) (msg by) {
+	if auther, ok := s.relay.(relay.Authenticator); ok && auther.AuthEnabled() {
+		svcUrl := auther.ServiceUrl(ws.Req())
+		if svcUrl == "" {
+			return
+		}
+		log.T.F("received auth response,%s", req)
+		var err er
+		var rem by
+		env := authenvelope.NewResponse()
+		if rem, err = env.UnmarshalJSON(req); chk.E(err) {
+			return
+		}
+		if len(rem) > 0 {
+			log.I.F("extra '%s'", rem)
+		}
+		var valid bo
+		if valid, err = auth.Validate(env.Event, by(ws.Challenge()), svcUrl); chk.E(err) {
+			if err := okenvelope.NewFrom(env.Event.ID, false,
+				normalize.Error.F(err.Error())).Write(ws); chk.E(err) {
+				return by(err.Error())
+			}
+			return normalize.Error.F(err.Error())
+		} else if !valid {
+			if err = okenvelope.NewFrom(env.Event.ID, false,
+				normalize.Error.F("failed to authenticate")).Write(ws); chk.E(err) {
+				return by(err.Error())
+			}
+			return normalize.Restricted.F("auth response does not validate")
+		} else {
+			if err = okenvelope.NewFrom(env.Event.ID, true, by{}).Write(ws); chk.E(err) {
+				return
+			}
+			log.D.F("%s authed to pubkey,%0x", ws.RealRemote(), env.Event.PubKey)
+			ws.SetAuthed(st(env.Event.PubKey))
+		}
+	}
+	return
+}

realy/handleClose.go

@@ -0,0 +1,23 @@
+package realy
+
+import (
+	"realy.lol/envelopes/closeenvelope"
+	"realy.lol/web"
+)
+
+func (s *Server) handleClose(ws *web.Socket, req by) (note by) {
+	var err er
+	var rem by
+	env := closeenvelope.New()
+	if rem, err = env.UnmarshalJSON(req); chk.E(err) {
+		return by(err.Error())
+	}
+	if len(rem) > 0 {
+		log.I.F("extra '%s'", rem)
+	}
+	if env.ID.String() == "" {
+		return by("CLOSE has no <id>")
+	}
+	s.listeners.RemoveListenerId(ws, env.ID.String())
+	return
+}

realy/handleCount.go

@@ -0,0 +1,112 @@
+package realy
+
+import (
+	"realy.lol/context"
+	"realy.lol/envelopes/authenvelope"
+	"realy.lol/envelopes/closedenvelope"
+	"realy.lol/envelopes/countenvelope"
+	"realy.lol/kind"
+	"realy.lol/normalize"
+	"realy.lol/relay"
+	"realy.lol/store"
+	"realy.lol/tag"
+	"realy.lol/web"
+)
+
+func (s *Server) handleCount(c context.T, ws *web.Socket, req by, store store.I) (msg by) {
+	counter, ok := store.(relay.EventCounter)
+	if !ok {
+		return normalize.Restricted.F("this relay does not support NIP-45")
+	}
+	if ws.AuthRequested() && len(ws.Authed()) == 0 {
+		return
+	}
+	var err er
+	var rem by
+	env := countenvelope.New()
+	if rem, err = env.UnmarshalJSON(req); chk.E(err) {
+		return normalize.Error.F(err.Error())
+	}
+	if len(rem) > 0 {
+		log.I.F("extra '%s'", rem)
+	}
+	if env.Subscription == nil || env.Subscription.String() == "" {
+		return normalize.Error.F("COUNT has no <subscription id>")
+	}
+	allowed := env.Filters
+	if accepter, ok := s.relay.(relay.ReqAcceptor); ok {
+		var accepted bo
+		allowed, accepted = accepter.AcceptReq(c, ws.Req(), env.Subscription.T, env.Filters,
+			by(ws.Authed()))
+		if !accepted || allowed == nil {
+			var auther relay.Authenticator
+			if auther, ok = s.relay.(relay.Authenticator); ok && auther.AuthEnabled() && !ws.AuthRequested() {
+				ws.RequestAuth()
+				if err = closedenvelope.NewFrom(env.Subscription,
+					normalize.AuthRequired.F("auth required for count processing")).Write(ws); chk.E(err) {
+				}
+				log.I.F("requesting auth from client from %s", ws.RealRemote())
+				if err = authenvelope.NewChallengeWith(ws.Challenge()).Write(ws); chk.E(err) {
+					return
+				}
+				return
+			}
+		}
+	}
+	if allowed != env.Filters {
+		defer func() {
+			var auther relay.Authenticator
+			var ok bo
+			if auther, ok = s.relay.(relay.Authenticator); ok && auther.AuthEnabled() && !ws.AuthRequested() {
+				ws.RequestAuth()
+				if err = closedenvelope.NewFrom(env.Subscription,
+					normalize.AuthRequired.F("auth required for request processing")).Write(ws); chk.E(err) {
+				}
+				log.T.F("requesting auth from client from %s, challenge '%s'", ws.RealRemote(),
+					ws.Challenge())
+				if err = authenvelope.NewChallengeWith(ws.Challenge()).Write(ws); chk.E(err) {
+					return
+				}
+				return
+			}
+		}()
+	}
+	var total no
+	var approx bo
+	if allowed != nil {
+		for _, f := range allowed.F {
+			var auther relay.Authenticator
+			if auther, ok = s.relay.(relay.Authenticator); ok && auther.AuthEnabled() {
+				if f.Kinds.Contains(kind.EncryptedDirectMessage) || f.Kinds.Contains(kind.GiftWrap) {
+					senders := f.Authors
+					receivers := f.Tags.GetAll(tag.New("p"))
+					switch {
+					case len(ws.Authed()) == 0:
+						return normalize.Restricted.F("this realy does not serve kind-4 to unauthenticated users," + " does your client implement NIP-42?")
+					case senders.Len() == 1 && receivers.Len() < 2 && equals(senders.F()[0],
+						by(ws.Authed())):
+					case receivers.Len() == 1 && senders.Len() < 2 && equals(receivers.N(0).Value(),
+						by(ws.Authed())):
+					default:
+						return normalize.Restricted.F("authenticated user does not have" + " authorization for requested filters")
+					}
+				}
+			}
+			var count no
+			count, approx, err = counter.CountEvents(c, f)
+			if err != nil {
+				log.E.F("store: %v", err)
+				continue
+			}
+			total += count
+		}
+	}
+	var res *countenvelope.Response
+	if res, err = countenvelope.NewResponseFrom(env.Subscription.T, total, approx); chk.E(err) {
+		return
+	}
+	if err = res.Write(ws); chk.E(err) {
+		return
+	}
+	return
+}