mirror of
https://github.com/fiatjaf/khatru.git
synced 2026-04-07 14:06:51 +02:00
Compare commits
9 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
33545587b6 | ||
|
|
214371f8bd | ||
|
|
fbb40f3b74 | ||
|
|
d97a2f1cf2 | ||
|
|
c9a7d60543 | ||
|
|
2bb6d4d29a | ||
|
|
2292ce4a30 | ||
|
|
2ae219a34c | ||
|
|
8c9394993b |
132
adding.go
132
adding.go
@@ -11,34 +11,46 @@ import (
|
||||
|
||||
// AddEvent sends an event through then normal add pipeline, as if it was received from a websocket.
|
||||
func (rl *Relay) AddEvent(ctx context.Context, evt *nostr.Event) (skipBroadcast bool, writeError error) {
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
|
||||
if evt == nil {
|
||||
return false, errors.New("error: event is nil")
|
||||
}
|
||||
|
||||
if nostr.IsEphemeralKind(evt.Kind) {
|
||||
return false, rl.handleEphemeral(ctx, evt)
|
||||
} else {
|
||||
return rl.handleNormal(ctx, evt)
|
||||
}
|
||||
}
|
||||
|
||||
func (rl *Relay) handleNormal(ctx context.Context, evt *nostr.Event) (skipBroadcast bool, writeError error) {
|
||||
for _, reject := range rl.RejectEvent {
|
||||
if reject, msg := reject(ctx, evt); reject {
|
||||
if msg == "" {
|
||||
return false, errors.New("blocked: no reason")
|
||||
return true, errors.New("blocked: no reason")
|
||||
} else {
|
||||
return false, errors.New(nostr.NormalizeOKMessage(msg, "blocked"))
|
||||
return true, errors.New(nostr.NormalizeOKMessage(msg, "blocked"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if 20000 <= evt.Kind && evt.Kind < 30000 {
|
||||
// do not store ephemeral events
|
||||
for _, oee := range rl.OnEphemeralEvent {
|
||||
oee(ctx, evt)
|
||||
// will store
|
||||
// regular kinds are just saved directly
|
||||
if nostr.IsRegularKind(evt.Kind) {
|
||||
for _, store := range rl.StoreEvent {
|
||||
if err := store(ctx, evt); err != nil {
|
||||
switch err {
|
||||
case eventstore.ErrDupEvent:
|
||||
return true, nil
|
||||
default:
|
||||
return false, fmt.Errorf("%s", nostr.NormalizeOKMessage(err.Error(), "error"))
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// will store
|
||||
// regular kinds are just saved directly
|
||||
if nostr.IsRegularKind(evt.Kind) {
|
||||
for _, store := range rl.StoreEvent {
|
||||
if err := store(ctx, evt); err != nil {
|
||||
// otherwise it's a replaceable -- so we'll use the replacer functions if we have any
|
||||
if len(rl.ReplaceEvent) > 0 {
|
||||
for _, repl := range rl.ReplaceEvent {
|
||||
if err := repl(ctx, evt); err != nil {
|
||||
switch err {
|
||||
case eventstore.ErrDupEvent:
|
||||
return true, nil
|
||||
@@ -48,68 +60,54 @@ func (rl *Relay) AddEvent(ctx context.Context, evt *nostr.Event) (skipBroadcast
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// otherwise it's a replaceable -- so we'll use the replacer functions if we have any
|
||||
if len(rl.ReplaceEvent) > 0 {
|
||||
for _, repl := range rl.ReplaceEvent {
|
||||
if err := repl(ctx, evt); err != nil {
|
||||
switch err {
|
||||
// otherwise do it the manual way
|
||||
filter := nostr.Filter{Limit: 1, Kinds: []int{evt.Kind}, Authors: []string{evt.PubKey}}
|
||||
if nostr.IsAddressableKind(evt.Kind) {
|
||||
// when addressable, add the "d" tag to the filter
|
||||
filter.Tags = nostr.TagMap{"d": []string{evt.Tags.GetD()}}
|
||||
}
|
||||
|
||||
// now we fetch old events and delete them
|
||||
shouldStore := true
|
||||
for _, query := range rl.QueryEvents {
|
||||
ch, err := query(ctx, filter)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
for previous := range ch {
|
||||
if isOlder(previous, evt) {
|
||||
for _, del := range rl.DeleteEvent {
|
||||
del(ctx, previous)
|
||||
}
|
||||
} else {
|
||||
// we found a more recent event, so we won't delete it and also will not store this new one
|
||||
shouldStore = false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// store
|
||||
if shouldStore {
|
||||
for _, store := range rl.StoreEvent {
|
||||
if saveErr := store(ctx, evt); saveErr != nil {
|
||||
switch saveErr {
|
||||
case eventstore.ErrDupEvent:
|
||||
return true, nil
|
||||
default:
|
||||
return false, fmt.Errorf("%s", nostr.NormalizeOKMessage(err.Error(), "error"))
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// otherwise do it the manual way
|
||||
filter := nostr.Filter{Limit: 1, Kinds: []int{evt.Kind}, Authors: []string{evt.PubKey}}
|
||||
if nostr.IsAddressableKind(evt.Kind) {
|
||||
// when addressable, add the "d" tag to the filter
|
||||
filter.Tags = nostr.TagMap{"d": []string{evt.Tags.GetD()}}
|
||||
}
|
||||
|
||||
// now we fetch old events and delete them
|
||||
shouldStore := true
|
||||
for _, query := range rl.QueryEvents {
|
||||
ch, err := query(ctx, filter)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
for previous := range ch {
|
||||
if isOlder(previous, evt) {
|
||||
for _, del := range rl.DeleteEvent {
|
||||
del(ctx, previous)
|
||||
}
|
||||
} else {
|
||||
// we found a more recent event, so we won't delete it and also will not store this new one
|
||||
shouldStore = false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// store
|
||||
if shouldStore {
|
||||
for _, store := range rl.StoreEvent {
|
||||
if saveErr := store(ctx, evt); saveErr != nil {
|
||||
switch saveErr {
|
||||
case eventstore.ErrDupEvent:
|
||||
return true, nil
|
||||
default:
|
||||
return false, fmt.Errorf("%s", nostr.NormalizeOKMessage(saveErr.Error(), "error"))
|
||||
}
|
||||
return false, fmt.Errorf("%s", nostr.NormalizeOKMessage(saveErr.Error(), "error"))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for _, ons := range rl.OnEventSaved {
|
||||
ons(ctx, evt)
|
||||
}
|
||||
|
||||
// track event expiration if applicable
|
||||
rl.expirationManager.trackEvent(evt)
|
||||
}
|
||||
|
||||
for _, ons := range rl.OnEventSaved {
|
||||
ons(ctx, evt)
|
||||
}
|
||||
|
||||
// track event expiration if applicable
|
||||
rl.expirationManager.trackEvent(evt)
|
||||
|
||||
return false, nil
|
||||
}
|
||||
|
||||
@@ -25,7 +25,7 @@ func (bs BlossomServer) handleUploadCheck(w http.ResponseWriter, r *http.Request
|
||||
blossomError(w, "missing \"Authorization\" header", 401)
|
||||
return
|
||||
}
|
||||
if auth.Tags.GetFirst([]string{"t", "upload"}) == nil {
|
||||
if auth.Tags.FindWithValue("t", "upload") == nil {
|
||||
blossomError(w, "invalid \"Authorization\" event \"t\" tag", 403)
|
||||
return
|
||||
}
|
||||
@@ -59,7 +59,7 @@ func (bs BlossomServer) handleUpload(w http.ResponseWriter, r *http.Request) {
|
||||
blossomError(w, "missing \"Authorization\" header", 401)
|
||||
return
|
||||
}
|
||||
if auth.Tags.GetFirst([]string{"t", "upload"}) == nil {
|
||||
if auth.Tags.FindWithValue("t", "upload") == nil {
|
||||
blossomError(w, "invalid \"Authorization\" event \"t\" tag", 403)
|
||||
return
|
||||
}
|
||||
@@ -163,13 +163,13 @@ func (bs BlossomServer) handleGetBlob(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
// if there is one, we check if it has the extra requirements
|
||||
if auth != nil {
|
||||
if auth.Tags.GetFirst([]string{"t", "get"}) == nil {
|
||||
if auth.Tags.FindWithValue("t", "get") == nil {
|
||||
blossomError(w, "invalid \"Authorization\" event \"t\" tag", 403)
|
||||
return
|
||||
}
|
||||
|
||||
if auth.Tags.GetFirst([]string{"x", hhash}) == nil &&
|
||||
auth.Tags.GetFirst([]string{"server", bs.ServiceURL}) == nil {
|
||||
if auth.Tags.FindWithValue("x", hhash) == nil &&
|
||||
auth.Tags.FindWithValue("server", bs.ServiceURL) == nil {
|
||||
blossomError(w, "invalid \"Authorization\" event \"x\" or \"server\" tag", 403)
|
||||
return
|
||||
}
|
||||
@@ -239,7 +239,7 @@ func (bs BlossomServer) handleList(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
// if there is one, we check if it has the extra requirements
|
||||
if auth != nil {
|
||||
if auth.Tags.GetFirst([]string{"t", "list"}) == nil {
|
||||
if auth.Tags.FindWithValue("t", "list") == nil {
|
||||
blossomError(w, "invalid \"Authorization\" event \"t\" tag", 403)
|
||||
return
|
||||
}
|
||||
@@ -283,7 +283,7 @@ func (bs BlossomServer) handleDelete(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
if auth != nil {
|
||||
if auth.Tags.GetFirst([]string{"t", "delete"}) == nil {
|
||||
if auth.Tags.FindWithValue("t", "delete") == nil {
|
||||
blossomError(w, "invalid \"Authorization\" event \"t\" tag", 403)
|
||||
return
|
||||
}
|
||||
@@ -296,8 +296,8 @@ func (bs BlossomServer) handleDelete(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
hhash = hhash[1:]
|
||||
if auth.Tags.GetFirst([]string{"x", hhash}) == nil &&
|
||||
auth.Tags.GetFirst([]string{"server", bs.ServiceURL}) == nil {
|
||||
if auth.Tags.FindWithValue("x", hhash) == nil &&
|
||||
auth.Tags.FindWithValue("server", bs.ServiceURL) == nil {
|
||||
blossomError(w, "invalid \"Authorization\" event \"x\" or \"server\" tag", 403)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -6,6 +6,6 @@ import (
|
||||
|
||||
// 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 (rl *Relay) BroadcastEvent(evt *nostr.Event) {
|
||||
rl.notifyListeners(evt)
|
||||
func (rl *Relay) BroadcastEvent(evt *nostr.Event) int {
|
||||
return rl.notifyListeners(evt)
|
||||
}
|
||||
|
||||
@@ -39,6 +39,7 @@ func (rl *Relay) handleDeleteRequest(ctx context.Context, evt *nostr.Event) erro
|
||||
continue
|
||||
}
|
||||
|
||||
ctx := context.WithValue(ctx, internalCallKey, struct{}{})
|
||||
for _, query := range rl.QueryEvents {
|
||||
ch, err := query(ctx, f)
|
||||
if err != nil {
|
||||
@@ -66,6 +67,9 @@ func (rl *Relay) handleDeleteRequest(ctx context.Context, evt *nostr.Event) erro
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// if it was tracked to be expired that is not needed anymore
|
||||
rl.expirationManager.removeEvent(target.ID)
|
||||
} else {
|
||||
// fail and stop here
|
||||
return fmt.Errorf("blocked: %s", msg)
|
||||
|
||||
@@ -47,7 +47,7 @@ router.Route().
|
||||
return true
|
||||
case event.Kind <= 12 && event.Kind >= 9:
|
||||
return true
|
||||
case event.Tags.GetFirst([]string{"h", ""}) != nil:
|
||||
case event.Tags.Find("h") != nil:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
|
||||
26
ephemeral.go
Normal file
26
ephemeral.go
Normal file
@@ -0,0 +1,26 @@
|
||||
package khatru
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
|
||||
"github.com/nbd-wtf/go-nostr"
|
||||
)
|
||||
|
||||
func (rl *Relay) handleEphemeral(ctx context.Context, evt *nostr.Event) error {
|
||||
for _, reject := range rl.RejectEvent {
|
||||
if reject, msg := reject(ctx, evt); reject {
|
||||
if msg == "" {
|
||||
return errors.New("blocked: no reason")
|
||||
} else {
|
||||
return errors.New(nostr.NormalizeOKMessage(msg, "blocked"))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for _, oee := range rl.OnEphemeralEvent {
|
||||
oee(ctx, evt)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -13,7 +13,7 @@ func main() {
|
||||
relay := khatru.NewRelay()
|
||||
|
||||
db := lmdb.LMDBBackend{Path: "/tmp/khatru-lmdb-tmp"}
|
||||
os.MkdirAll(db.Path, 0755)
|
||||
os.MkdirAll(db.Path, 0o755)
|
||||
if err := db.Init(); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
@@ -16,7 +16,7 @@ func main() {
|
||||
relay := khatru.NewRelay()
|
||||
|
||||
db := lmdb.LMDBBackend{Path: "/tmp/exclusive"}
|
||||
os.MkdirAll(db.Path, 0755)
|
||||
os.MkdirAll(db.Path, 0o755)
|
||||
if err := db.Init(); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
@@ -52,7 +52,7 @@ func main() {
|
||||
return slices.Contains(filter.Kinds, 1) && slices.Contains(filter.Tags["t"], "spam")
|
||||
}).
|
||||
Event(func(event *nostr.Event) bool {
|
||||
return event.Kind == 1 && event.Tags.GetFirst([]string{"t", "spam"}) != nil
|
||||
return event.Kind == 1 && event.Tags.FindWithValue("t", "spam") != nil
|
||||
}).
|
||||
Relay(r2)
|
||||
|
||||
|
||||
@@ -73,6 +73,7 @@ func (em *expirationManager) initialScan(ctx context.Context) {
|
||||
defer em.mu.Unlock()
|
||||
|
||||
// query all events
|
||||
ctx = context.WithValue(ctx, internalCallKey, struct{}{})
|
||||
for _, query := range em.relay.QueryEvents {
|
||||
ch, err := query(ctx, nostr.Filter{})
|
||||
if err != nil {
|
||||
@@ -107,6 +108,7 @@ func (em *expirationManager) checkExpiredEvents(ctx context.Context) {
|
||||
|
||||
heap.Pop(&em.events)
|
||||
|
||||
ctx := context.WithValue(ctx, internalCallKey, struct{}{})
|
||||
for _, query := range em.relay.QueryEvents {
|
||||
ch, err := query(ctx, nostr.Filter{IDs: []string{next.id}})
|
||||
if err != nil {
|
||||
@@ -133,3 +135,16 @@ func (em *expirationManager) trackEvent(evt *nostr.Event) {
|
||||
em.mu.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
func (em *expirationManager) removeEvent(id string) {
|
||||
em.mu.Lock()
|
||||
defer em.mu.Unlock()
|
||||
|
||||
// Find and remove the event from the heap
|
||||
for i := 0; i < len(em.events); i++ {
|
||||
if em.events[i].id == id {
|
||||
heap.Remove(&em.events, i)
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
6
go.mod
6
go.mod
@@ -8,7 +8,7 @@ require (
|
||||
github.com/fiatjaf/eventstore v0.16.2
|
||||
github.com/liamg/magic v0.0.1
|
||||
github.com/mailru/easyjson v0.9.0
|
||||
github.com/nbd-wtf/go-nostr v0.51.7
|
||||
github.com/nbd-wtf/go-nostr v0.51.8
|
||||
github.com/puzpuzpuz/xsync/v3 v3.5.1
|
||||
github.com/rs/cors v1.11.1
|
||||
github.com/stretchr/testify v1.10.0
|
||||
@@ -22,11 +22,11 @@ require (
|
||||
github.com/aquasecurity/esquery v0.2.0 // indirect
|
||||
github.com/btcsuite/btcd/btcec/v2 v2.3.4 // indirect
|
||||
github.com/btcsuite/btcd/chaincfg/chainhash v1.1.0 // indirect
|
||||
github.com/bytedance/sonic v1.13.1 // indirect
|
||||
github.com/bytedance/sonic v1.13.2 // indirect
|
||||
github.com/bytedance/sonic/loader v0.2.4 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||
github.com/cloudwego/base64x v0.1.5 // indirect
|
||||
github.com/coder/websocket v1.8.12 // indirect
|
||||
github.com/coder/websocket v1.8.13 // indirect
|
||||
github.com/davecgh/go-spew v1.1.1 // indirect
|
||||
github.com/decred/dcrd/crypto/blake256 v1.1.0 // indirect
|
||||
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0 // indirect
|
||||
|
||||
7
go.sum
7
go.sum
@@ -14,12 +14,15 @@ github.com/aquasecurity/esquery v0.2.0 h1:9WWXve95TE8hbm3736WB7nS6Owl8UGDeu+0jiy
|
||||
github.com/aquasecurity/esquery v0.2.0/go.mod h1:VU+CIFR6C+H142HHZf9RUkp4Eedpo9UrEKeCQHWf9ao=
|
||||
github.com/bep/debounce v1.2.1 h1:v67fRdBA9UQu2NhLFXrSg0Brw7CexQekrBwDMM8bzeY=
|
||||
github.com/bep/debounce v1.2.1/go.mod h1:H8yggRPQKLUhUoqrJC1bO2xNya7vanpDl7xR3ISbCJ0=
|
||||
github.com/btcsuite/btcd v0.24.2 h1:aLmxPguqxza+4ag8R1I2nnJjSu2iFn/kqtHTIImswcY=
|
||||
github.com/btcsuite/btcd/btcec/v2 v2.3.4 h1:3EJjcN70HCu/mwqlUsGK8GcNVyLVxFDlWurTXGPFfiQ=
|
||||
github.com/btcsuite/btcd/btcec/v2 v2.3.4/go.mod h1:zYzJ8etWJQIv1Ogk7OzpWjowwOdXY1W/17j2MW85J04=
|
||||
github.com/btcsuite/btcd/chaincfg/chainhash v1.1.0 h1:59Kx4K6lzOW5w6nFlA0v5+lk/6sjybR934QNHSJZPTQ=
|
||||
github.com/btcsuite/btcd/chaincfg/chainhash v1.1.0/go.mod h1:7SFka0XMvUgj3hfZtydOrQY2mwhPclbT2snogU7SQQc=
|
||||
github.com/bytedance/sonic v1.13.1 h1:Jyd5CIvdFnkOWuKXr+wm4Nyk2h0yAFsr8ucJgEasO3g=
|
||||
github.com/bytedance/sonic v1.13.1/go.mod h1:o68xyaF9u2gvVBuGHPlUVCy+ZfmNNO5ETf1+KgkJhz4=
|
||||
github.com/bytedance/sonic v1.13.2 h1:8/H1FempDZqC4VqjptGo14QQlJx8VdZJegxs6wwfqpQ=
|
||||
github.com/bytedance/sonic v1.13.2/go.mod h1:o68xyaF9u2gvVBuGHPlUVCy+ZfmNNO5ETf1+KgkJhz4=
|
||||
github.com/bytedance/sonic/loader v0.1.1/go.mod h1:ncP89zfokxS5LZrJxl5z0UJcsk4M4yY2JpfqGeCtNLU=
|
||||
github.com/bytedance/sonic/loader v0.2.4 h1:ZWCw4stuXUsn1/+zQDqeE7JKP+QO47tz7QCNan80NzY=
|
||||
github.com/bytedance/sonic/loader v0.2.4/go.mod h1:N8A3vUdtUebEY2/VQC0MyhYeKUFosQU6FxH2JmUe6VI=
|
||||
@@ -33,6 +36,8 @@ github.com/cloudwego/iasm v0.2.0/go.mod h1:8rXZaNYT2n95jn+zTI1sDr+IgcD2GVs0nlbbQ
|
||||
github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
|
||||
github.com/coder/websocket v1.8.12 h1:5bUXkEPPIbewrnkU8LTCLVaxi4N4J8ahufH2vlo4NAo=
|
||||
github.com/coder/websocket v1.8.12/go.mod h1:LNVeNrXQZfe5qhS9ALED3uA+l5pPqvwXg3CKoDBB2gs=
|
||||
github.com/coder/websocket v1.8.13 h1:f3QZdXy7uGVz+4uCJy2nTZyM0yTBj8yANEHhqlXZ9FE=
|
||||
github.com/coder/websocket v1.8.13/go.mod h1:LNVeNrXQZfe5qhS9ALED3uA+l5pPqvwXg3CKoDBB2gs=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
@@ -131,6 +136,8 @@ github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9G
|
||||
github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
|
||||
github.com/nbd-wtf/go-nostr v0.51.7 h1:dGjtaaFQ1kA3H+vF8wt9a9WYl54K8C0JmVDf4cp+a4A=
|
||||
github.com/nbd-wtf/go-nostr v0.51.7/go.mod h1:d6+DfvMWYG5pA3dmNMBJd6WCHVDDhkXbHqvfljf0Gzg=
|
||||
github.com/nbd-wtf/go-nostr v0.51.8 h1:CIoS+YqChcm4e1L1rfMZ3/mIwTz4CwApM2qx7MHNzmE=
|
||||
github.com/nbd-wtf/go-nostr v0.51.8/go.mod h1:d6+DfvMWYG5pA3dmNMBJd6WCHVDDhkXbHqvfljf0Gzg=
|
||||
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
|
||||
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
|
||||
69
handlers.go
69
handlers.go
@@ -6,6 +6,7 @@ import (
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -17,6 +18,7 @@ import (
|
||||
"github.com/nbd-wtf/go-nostr/nip42"
|
||||
"github.com/nbd-wtf/go-nostr/nip45"
|
||||
"github.com/nbd-wtf/go-nostr/nip45/hyperloglog"
|
||||
"github.com/nbd-wtf/go-nostr/nip70"
|
||||
"github.com/nbd-wtf/go-nostr/nip77"
|
||||
"github.com/nbd-wtf/go-nostr/nip77/negentropy"
|
||||
"github.com/puzpuzpuz/xsync/v3"
|
||||
@@ -140,6 +142,7 @@ func (rl *Relay) HandleWebsocket(w http.ResponseWriter, r *http.Request) {
|
||||
continue
|
||||
}
|
||||
|
||||
// this is safe because ReadMessage() will always create a new slice
|
||||
message := unsafe.String(unsafe.SliceData(msgb), len(msgb))
|
||||
|
||||
// parse messages sequentially otherwise sonic breaks
|
||||
@@ -175,28 +178,31 @@ func (rl *Relay) HandleWebsocket(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
// check NIP-70 protected
|
||||
for _, v := range env.Event.Tags {
|
||||
if len(v) == 1 && v[0] == "-" {
|
||||
msg := "must be published by event author"
|
||||
authed := GetAuthed(ctx)
|
||||
if authed == "" {
|
||||
RequestAuth(ctx)
|
||||
ws.WriteJSON(nostr.OKEnvelope{
|
||||
EventID: env.Event.ID,
|
||||
OK: false,
|
||||
Reason: "auth-required: " + msg,
|
||||
})
|
||||
return
|
||||
}
|
||||
if authed != env.Event.PubKey {
|
||||
ws.WriteJSON(nostr.OKEnvelope{
|
||||
EventID: env.Event.ID,
|
||||
OK: false,
|
||||
Reason: "blocked: " + msg,
|
||||
})
|
||||
return
|
||||
}
|
||||
if nip70.IsProtected(env.Event) {
|
||||
authed := GetAuthed(ctx)
|
||||
if authed == "" {
|
||||
RequestAuth(ctx)
|
||||
ws.WriteJSON(nostr.OKEnvelope{
|
||||
EventID: env.Event.ID,
|
||||
OK: false,
|
||||
Reason: "auth-required: must be published by authenticated event author",
|
||||
})
|
||||
return
|
||||
} else if authed != env.Event.PubKey {
|
||||
ws.WriteJSON(nostr.OKEnvelope{
|
||||
EventID: env.Event.ID,
|
||||
OK: false,
|
||||
Reason: "blocked: must be published by event author",
|
||||
})
|
||||
return
|
||||
}
|
||||
} else if nip70.HasEmbeddedProtected(env.Event) {
|
||||
ws.WriteJSON(nostr.OKEnvelope{
|
||||
EventID: env.Event.ID,
|
||||
OK: false,
|
||||
Reason: "blocked: can't repost nip70 protected",
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
srl := rl
|
||||
@@ -211,9 +217,12 @@ func (rl *Relay) HandleWebsocket(w http.ResponseWriter, r *http.Request) {
|
||||
if env.Event.Kind == 5 {
|
||||
// this always returns "blocked: " whenever it returns an error
|
||||
writeErr = srl.handleDeleteRequest(ctx, &env.Event)
|
||||
} else if nostr.IsEphemeralKind(env.Event.Kind) {
|
||||
// this will also always return a prefixed reason
|
||||
writeErr = srl.handleEphemeral(ctx, &env.Event)
|
||||
} else {
|
||||
// this will also always return a prefixed reason
|
||||
skipBroadcast, writeErr = srl.AddEvent(ctx, &env.Event)
|
||||
skipBroadcast, writeErr = srl.handleNormal(ctx, &env.Event)
|
||||
}
|
||||
|
||||
var reason string
|
||||
@@ -223,9 +232,20 @@ func (rl *Relay) HandleWebsocket(w http.ResponseWriter, r *http.Request) {
|
||||
ovw(ctx, &env.Event)
|
||||
}
|
||||
if !skipBroadcast {
|
||||
srl.notifyListeners(&env.Event)
|
||||
n := srl.notifyListeners(&env.Event)
|
||||
|
||||
// the number of notified listeners matters in ephemeral events
|
||||
if nostr.IsEphemeralKind(env.Event.Kind) {
|
||||
if n == 0 {
|
||||
ok = false
|
||||
reason = "mute: no one was listening for this"
|
||||
} else {
|
||||
reason = "broadcasted to " + strconv.Itoa(n) + " listeners"
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
ok = false
|
||||
reason = writeErr.Error()
|
||||
if strings.HasPrefix(reason, "auth-required:") {
|
||||
RequestAuth(ctx)
|
||||
@@ -240,14 +260,13 @@ func (rl *Relay) HandleWebsocket(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
var total int64
|
||||
var hll *hyperloglog.HyperLogLog
|
||||
uneligibleForHLL := false
|
||||
|
||||
srl := rl
|
||||
if rl.getSubRelayFromFilter != nil {
|
||||
srl = rl.getSubRelayFromFilter(env.Filter)
|
||||
}
|
||||
|
||||
if offset := nip45.HyperLogLogEventPubkeyOffsetForFilter(env.Filter); offset != -1 && !uneligibleForHLL {
|
||||
if offset := nip45.HyperLogLogEventPubkeyOffsetForFilter(env.Filter); offset != -1 {
|
||||
total, hll = srl.handleCountRequestWithHLL(ctx, ws, env.Filter, offset)
|
||||
} else {
|
||||
total = srl.handleCountRequest(ctx, ws, env.Filter)
|
||||
|
||||
@@ -132,15 +132,20 @@ func (rl *Relay) removeClientAndListeners(ws *WebSocket) {
|
||||
delete(rl.clients, ws)
|
||||
}
|
||||
|
||||
func (rl *Relay) notifyListeners(event *nostr.Event) {
|
||||
// returns how many listeners were notified
|
||||
func (rl *Relay) notifyListeners(event *nostr.Event) int {
|
||||
count := 0
|
||||
listenersloop:
|
||||
for _, listener := range rl.listeners {
|
||||
if listener.filter.Matches(event) {
|
||||
for _, pb := range rl.PreventBroadcast {
|
||||
if pb(listener.ws, event) {
|
||||
return
|
||||
continue listenersloop
|
||||
}
|
||||
}
|
||||
listener.ws.WriteJSON(nostr.EventEnvelope{SubscriptionID: &listener.id, Event: *event})
|
||||
count++
|
||||
}
|
||||
}
|
||||
return count
|
||||
}
|
||||
|
||||
4
nip86.go
4
nip86.go
@@ -86,10 +86,10 @@ func (rl *Relay) HandleNIP86(w http.ResponseWriter, r *http.Request) {
|
||||
goto respond
|
||||
}
|
||||
|
||||
if uTag := evt.Tags.GetFirst([]string{"u", ""}); uTag == nil || rl.getBaseURL(r) != (*uTag)[1] {
|
||||
if uTag := evt.Tags.Find("u"); uTag == nil || rl.getBaseURL(r) != uTag[1] {
|
||||
resp.Error = "invalid 'u' tag"
|
||||
goto respond
|
||||
} else if pht := evt.Tags.GetFirst([]string{"payload", hex.EncodeToString(payloadHash[:])}); pht == nil {
|
||||
} else if pht := evt.Tags.FindWithValue("payload", hex.EncodeToString(payloadHash[:])); pht == nil {
|
||||
resp.Error = "invalid auth event payload hash"
|
||||
goto respond
|
||||
} else if evt.CreatedAt < nostr.Now()-30 {
|
||||
|
||||
@@ -110,7 +110,7 @@ func RejectEventsWithBase64Media(ctx context.Context, evt *nostr.Event) (bool, s
|
||||
}
|
||||
|
||||
func OnlyAllowNIP70ProtectedEvents(ctx context.Context, event *nostr.Event) (reject bool, msg string) {
|
||||
if nip70.IsProtected(event) {
|
||||
if nip70.IsProtected(*event) {
|
||||
return false, ""
|
||||
}
|
||||
return true, "blocked: we only accept events protected with the nip70 \"-\" tag"
|
||||
|
||||
@@ -2,7 +2,6 @@ package policies
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"slices"
|
||||
|
||||
"github.com/fiatjaf/khatru"
|
||||
|
||||
7
utils.go
7
utils.go
@@ -10,6 +10,7 @@ const (
|
||||
wsKey = iota
|
||||
subscriptionIdKey
|
||||
nip86HeaderAuthKey
|
||||
internalCallKey
|
||||
)
|
||||
|
||||
func RequestAuth(ctx context.Context) {
|
||||
@@ -40,6 +41,12 @@ func GetAuthed(ctx context.Context) string {
|
||||
return ""
|
||||
}
|
||||
|
||||
// IsInternalCall returns true when a call to QueryEvents, for example, is being made because of a deletion
|
||||
// or expiration request.
|
||||
func IsInternalCall(ctx context.Context) bool {
|
||||
return ctx.Value(internalCallKey) != nil
|
||||
}
|
||||
|
||||
func GetIP(ctx context.Context) string {
|
||||
conn := GetConnection(ctx)
|
||||
if conn == nil {
|
||||
|
||||
10
websocket.go
10
websocket.go
@@ -33,12 +33,14 @@ type WebSocket struct {
|
||||
|
||||
func (ws *WebSocket) WriteJSON(any any) error {
|
||||
ws.mutex.Lock()
|
||||
defer ws.mutex.Unlock()
|
||||
return ws.conn.WriteJSON(any)
|
||||
err := ws.conn.WriteJSON(any)
|
||||
ws.mutex.Unlock()
|
||||
return err
|
||||
}
|
||||
|
||||
func (ws *WebSocket) WriteMessage(t int, b []byte) error {
|
||||
ws.mutex.Lock()
|
||||
defer ws.mutex.Unlock()
|
||||
return ws.conn.WriteMessage(t, b)
|
||||
err := ws.conn.WriteMessage(t, b)
|
||||
ws.mutex.Unlock()
|
||||
return err
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user