Compare commits

...

6 Commits

Author SHA1 Message Date
fiatjaf
7a3eb6fb08 plugins.PreventLargeTags() 2023-11-13 16:26:27 -03:00
fiatjaf
1abeab4851 filter.Limit is zero by default, so we will short-circuit on negative limits. 2023-11-13 10:55:17 -03:00
fiatjaf
ecb4fc66f8 shortcut querying when limit == 0 2023-11-12 09:27:36 -03:00
fiatjaf
af24bd2132 fix eose waitgroup on early reject. 2023-11-11 21:15:10 -03:00
fiatjaf
4905a46ccd make filter rejection actually work and move logic to a separate file. 2023-11-11 21:08:39 -03:00
fiatjaf
21e0c559f7 simplify event rejection code. 2023-11-11 18:00:04 -03:00
4 changed files with 106 additions and 83 deletions

View File

@@ -13,22 +13,15 @@ func (rl *Relay) AddEvent(ctx context.Context, evt *nostr.Event) error {
return fmt.Errorf("event is nil")
}
msg := ""
rejecting := false
for _, reject := range rl.RejectEvent {
rejecting, msg = reject(ctx, evt)
if rejecting {
break
if reject, msg := reject(ctx, evt); reject {
if msg == "" {
msg = "no reason"
}
return fmt.Errorf(msg)
}
}
if rejecting {
if msg == "" {
msg = "no reason"
}
return fmt.Errorf(msg)
}
if 20000 <= evt.Kind && evt.Kind < 30000 {
// do not store ephemeral events
} else {

View File

@@ -7,6 +7,7 @@ import (
"encoding/hex"
"encoding/json"
"net/http"
"strings"
"sync"
"time"
@@ -165,30 +166,7 @@ func (rl *Relay) HandleWebsocket(w http.ResponseWriter, r *http.Request) {
ws.WriteJSON(nostr.NoticeEnvelope("failed to decode filter"))
continue
}
filter := filters[i]
// overwrite the filter (for example, to eliminate some kinds or tags that we know we don't support)
for _, ovw := range rl.OverwriteCountFilter {
ovw(ctx, &filter)
}
// then check if we'll reject this filter
for _, reject := range rl.RejectCountFilter {
if rejecting, msg := reject(ctx, filter); rejecting {
ws.WriteJSON(nostr.NoticeEnvelope(msg))
continue
}
}
// run the functions to count (generally it will be just one)
for _, count := range rl.CountEvents {
res, err := count(ctx, filter)
if err != nil {
ws.WriteJSON(nostr.NoticeEnvelope(err.Error()))
}
total += res
}
total += rl.handleCountRequest(ctx, ws, filters[i])
}
ws.WriteJSON([]interface{}{"COUNT", id, map[string]int64{"count": total}})
@@ -205,57 +183,13 @@ func (rl *Relay) HandleWebsocket(w http.ResponseWriter, r *http.Request) {
eose.Add(len(request[2:]))
for i, filterReq := range request[2:] {
if err := json.Unmarshal(
filterReq,
&filters[i],
); err != nil {
if err := json.Unmarshal(filterReq, &filters[i]); err != nil {
ws.WriteJSON(nostr.NoticeEnvelope("failed to decode filter"))
eose.Done()
continue
}
filter := filters[i]
// overwrite the filter (for example, to eliminate some kinds or
// that we know we don't support)
for _, ovw := range rl.OverwriteFilter {
ovw(ctx, &filter)
}
// then check if we'll reject this filter (we apply this after overwriting
// because we may, for example, remove some things from the incoming filters
// that we know we don't support, and then if the end result is an empty
// filter we can just reject it)
for _, reject := range rl.RejectFilter {
if rejecting, msg := reject(ctx, filter); rejecting {
ws.WriteJSON(nostr.NoticeEnvelope(msg))
continue
}
}
// run the functions to query events (generally just one,
// but we might be fetching stuff from multiple places)
eose.Add(len(rl.QueryEvents))
for _, query := range rl.QueryEvents {
ch, err := query(ctx, filter)
if err != nil {
ws.WriteJSON(nostr.NoticeEnvelope(err.Error()))
eose.Done()
continue
}
go func(ch chan *nostr.Event) {
for event := range ch {
for _, ovw := range rl.OverwriteResponseEvent {
ovw(ctx, event)
}
ws.WriteJSON(nostr.EventEnvelope{SubscriptionID: &id, Event: *event})
}
eose.Done()
}(ch)
}
eose.Done()
go rl.handleRequest(ctx, id, &eose, ws, filters[i])
}
go func() {
@@ -306,7 +240,7 @@ func (rl *Relay) HandleWebsocket(w http.ResponseWriter, r *http.Request) {
case <-ticker.C:
err := ws.WriteMessage(websocket.PingMessage, nil)
if err != nil {
if err.Error() != "use of closed network connection" {
if !strings.HasSuffix(err.Error(), "use of closed network connection") {
rl.Log.Printf("error writing ping: %v; closing websocket\n", err)
}
return

View File

@@ -23,6 +23,20 @@ func PreventTooManyIndexableTags(max int) func(context.Context, *nostr.Event) (b
}
}
// PreventLargeTags rejects events that have indexable tag values greater than maxTagValueLen.
func PreventLargeTags(maxTagValueLen int) func(context.Context, *nostr.Event) (bool, string) {
return func(ctx context.Context, event *nostr.Event) (reject bool, msg string) {
for _, tag := range event.Tags {
if len(tag) > 1 && len(tag[0]) == 1 {
if len(tag[1]) > maxTagValueLen {
return true, "event contains too large tags"
}
}
}
return false, ""
}
}
// RestrictToSpecifiedKinds returns a function that can be used as a RejectFilter that will reject
// any events with kinds different than the specified ones.
func RestrictToSpecifiedKinds(kinds ...uint16) func(context.Context, *nostr.Event) (bool, string) {

82
serve-req.go Normal file
View File

@@ -0,0 +1,82 @@
package khatru
import (
"context"
"sync"
"github.com/nbd-wtf/go-nostr"
)
func (rl *Relay) handleRequest(ctx context.Context, id string, eose *sync.WaitGroup, ws *WebSocket, filter nostr.Filter) {
defer eose.Done()
// overwrite the filter (for example, to eliminate some kinds or
// that we know we don't support)
for _, ovw := range rl.OverwriteFilter {
ovw(ctx, &filter)
}
if filter.Limit < 0 {
return
}
// then check if we'll reject this filter (we apply this after overwriting
// because we may, for example, remove some things from the incoming filters
// that we know we don't support, and then if the end result is an empty
// filter we can just reject it)
for _, reject := range rl.RejectFilter {
if reject, msg := reject(ctx, filter); reject {
ws.WriteJSON(nostr.NoticeEnvelope(msg))
return
}
}
// run the functions to query events (generally just one,
// but we might be fetching stuff from multiple places)
eose.Add(len(rl.QueryEvents))
for _, query := range rl.QueryEvents {
ch, err := query(ctx, filter)
if err != nil {
ws.WriteJSON(nostr.NoticeEnvelope(err.Error()))
eose.Done()
continue
}
go func(ch chan *nostr.Event) {
for event := range ch {
for _, ovw := range rl.OverwriteResponseEvent {
ovw(ctx, event)
}
ws.WriteJSON(nostr.EventEnvelope{SubscriptionID: &id, Event: *event})
}
eose.Done()
}(ch)
}
}
func (rl *Relay) handleCountRequest(ctx context.Context, ws *WebSocket, filter nostr.Filter) int64 {
// overwrite the filter (for example, to eliminate some kinds or tags that we know we don't support)
for _, ovw := range rl.OverwriteCountFilter {
ovw(ctx, &filter)
}
// then check if we'll reject this filter
for _, reject := range rl.RejectCountFilter {
if rejecting, msg := reject(ctx, filter); rejecting {
ws.WriteJSON(nostr.NoticeEnvelope(msg))
return 0
}
}
// run the functions to count (generally it will be just one)
var subtotal int64 = 0
for _, count := range rl.CountEvents {
res, err := count(ctx, filter)
if err != nil {
ws.WriteJSON(nostr.NoticeEnvelope(err.Error()))
}
subtotal += res
}
return subtotal
}