server_subs.mx raw
1 package server
2
3 import (
4 "bytes"
5 "runtime"
6 "time"
7
8 "git.smesh.lol/nostr/pkg/envelope"
9 "git.smesh.lol/nostr/pkg/filter"
10 "git.smesh.lol/morly/pkg/relay/tree"
11 )
12
13 // --- Subscription and Nostr message handling ---
14
15 func (s *Server) handleReq(fd int32, msg []byte) {
16 c := s.conns[fd]
17 if c == nil {
18 return
19 }
20 // Everything this subscription keeps - the id, the filters and the raw
21 // request - outlives this call in c.subs, so parse and build it while
22 // borrowing the root arena. Copying the filter slice header out of this
23 // frame is not enough: the array it points at would stay behind, and the
24 // receiver's sovereign compaction walk later follows that pointer into
25 // memory the frame has already released.
26 prev := runtime.CurrentArena()
27 runtime.SovereignSetArena(runtime.RootArena())
28 _, rem, _ := envelope.Identify(msg)
29 filter.ClearTaint()
30 var req envelope.Req
31 if _, err := req.Unmarshal(rem); err != nil {
32 runtime.SovereignRestoreArena(prev)
33 return
34 }
35 if filter.IsTainted() {
36 runtime.SovereignRestoreArena(prev)
37 c.jailed = true
38 return
39 }
40 id := string(req.Subscription)
41 cfg := s.cfg
42 authed := len(c.authedPubkey) > 0
43 wl := s.t.ConnIsWhitelisted(fd)
44
45 if cfg.AuthRequired && !authed && !wl {
46 cl := &envelope.Closed{
47 Subscription: []byte(id),
48 Reason: []byte("auth-required: authentication required"),
49 }
50 runtime.SovereignRestoreArena(prev)
51 s.t.SendWS(fd, cl.Marshal(nil))
52 return
53 }
54
55 maxSubs := cfg.MaxSubscriptions
56 if maxSubs > 0 {
57 if _, exists := c.subs[id]; !exists && len(c.subs) >= maxSubs {
58 cl := &envelope.Closed{
59 Subscription: []byte(id),
60 Reason: []byte("error: too many subscriptions"),
61 }
62 runtime.SovereignRestoreArena(prev)
63 s.t.SendWS(fd, cl.Marshal(nil))
64 return
65 }
66 }
67 filters := req.Filters
68 reqCopy := []byte{:len(msg)}
69 copy(reqCopy, msg)
70 c.subs[id] = &sub{id: id, filters: filters, rawReq: reqCopy}
71 runtime.SovereignRestoreArena(prev)
72 s.bcast.OnSubscribe(int32(fd), []byte(id), msg)
73
74 needsFilter := cfg.RelayURL != "" && !cfg.PrivilegedOpen && !wl
75 limit := cfg.QueryResultLimit
76 if limit <= 0 {
77 limit = 256
78 }
79 // Backfill is a database query: root hands the whole REQ frame to the
80 // database-engine node, which returns ready-to-send EVENT frames plus the
81 // EOSE. Root stays free while the query runs.
82 subID := req.Subscription
83 var authed []byte
84 if len(c.authedPubkey) > 0 {
85 authed = []byte{:len(c.authedPubkey)}
86 copy(authed, c.authedPubkey)
87 }
88 if !s.dbSend(tree.Request{
89 Op: tree.OpHistory,
90 ConnID: fd,
91 Limit: limit,
92 Filtered: needsFilter,
93 NIP70: cfg.NIP70Enforce,
94 Marmot: cfg.MarmotOpen,
95 SubID: subID,
96 Filter: msg,
97 AuthedPubkey: authed,
98 }) {
99 s.t.SendWS(fd, (&envelope.EOSE{Subscription: subID}).Marshal(nil))
100 }
101 }
102
103 func (s *Server) handleClose(fd int32, msg []byte) {
104 c := s.conns[fd]
105 if c == nil {
106 return
107 }
108 _, rem, _ := envelope.Identify(msg)
109 var cl envelope.Close
110 if _, err := cl.Unmarshal(rem); err != nil {
111 return
112 }
113 delete(c.subs, string(cl.ID))
114 s.bcast.OnUnsubscribe(int32(fd), cl.ID)
115 }
116
117 func (s *Server) handleCount(fd int32, msg []byte) {
118 c := s.conns[fd]
119 if c == nil {
120 return
121 }
122 if s.cfg.AuthRequired && len(c.authedPubkey) == 0 && !s.t.ConnIsWhitelisted(fd) {
123 return
124 }
125 _, rem, ierr := envelope.Identify(msg)
126 if ierr != nil {
127 return
128 }
129 filter.ClearTaint()
130 var cr envelope.CountRequest
131 if _, err := cr.Unmarshal(rem); err != nil {
132 return
133 }
134 if filter.IsTainted() {
135 c.jailed = true
136 return
137 }
138 s.dbSend(tree.Request{Op: tree.OpCount, ConnID: fd, Filter: msg})
139 }
140
141 func (s *Server) handleAuth(fd int32, msg []byte) {
142 c := s.conns[fd]
143 if c == nil {
144 return
145 }
146 _, rem, _ := envelope.Identify(msg)
147 var auth envelope.AuthResponse
148 if _, aerr := auth.Unmarshal(rem); aerr != nil || auth.Event == nil {
149 bad := &envelope.OK{
150 EventID: []byte{:32},
151 OK: false,
152 Reason: []byte("error: failed to parse auth event"),
153 }
154 s.t.SendWS(fd, bad.Marshal(nil))
155 return
156 }
157 if auth.Event.Kind != 22242 {
158 s.authFail(fd, auth.Event.ID, "error: wrong event kind for auth")
159 return
160 }
161 ct := auth.Event.Tags.GetFirst([]byte("challenge"))
162 if ct == nil || !bytes.Equal(ct.Value(), c.challenge) {
163 s.authFail(fd, auth.Event.ID, "error: wrong challenge")
164 return
165 }
166 rt := auth.Event.Tags.GetFirst([]byte("relay"))
167 if rt == nil || len(rt.Value()) == 0 {
168 s.authFail(fd, auth.Event.ID, "error: missing relay tag")
169 return
170 }
171 if !relayURLMatch([]byte(s.cfg.RelayURL), rt.Value()) {
172 s.authFail(fd, auth.Event.ID, "error: relay URL mismatch")
173 return
174 }
175 now := time.Now().Unix()
176 if auth.Event.CreatedAt > now+600 || auth.Event.CreatedAt < now-600 {
177 s.authFail(fd, auth.Event.ID, "error: timestamp out of range")
178 return
179 }
180 valid, err := auth.Event.Verify()
181 if err != nil || !valid {
182 s.authFail(fd, auth.Event.ID, "error: invalid signature")
183 return
184 }
185 // The authed pubkey outlives this call - it stays on the connection state
186 // for the life of the connection - so copy it into the root arena.
187 authPrev := runtime.CurrentArena()
188 runtime.SovereignSetArena(runtime.RootArena())
189 c.authedPubkey = []byte{:len(auth.Event.Pubkey)}
190 copy(c.authedPubkey, auth.Event.Pubkey)
191 runtime.SovereignRestoreArena(authPrev)
192 s.bcast.OnAuth(int32(fd), c.authedPubkey)
193 ok := &envelope.OK{EventID: auth.Event.ID, OK: true}
194 s.t.SendWS(fd, ok.Marshal(nil))
195 }
196
197 func (s *Server) authFail(fd int32, id []byte, reason string) {
198 ok := &envelope.OK{EventID: id, OK: false, Reason: []byte(reason)}
199 s.t.SendWS(fd, ok.Marshal(nil))
200 }
201
202 func relayURLMatch(expected, found []byte) (ok bool) {
203 if bytes.Equal(expected, found) {
204 return true
205 }
206 e := makeCopy(expected)
207 f := makeCopy(found)
208 toLower(e)
209 toLower(f)
210 if len(e) > 0 && e[len(e)-1] == '/' {
211 e = e[:len(e)-1]
212 }
213 if len(f) > 0 && f[len(f)-1] == '/' {
214 f = f[:len(f)-1]
215 }
216 return bytes.Equal(stripScheme(e), stripScheme(f))
217 }
218
219 func stripScheme(u []byte) (buf []byte) {
220 if bytes.HasPrefix(u, []byte("wss://")) {
221 return u[6:]
222 }
223 if bytes.HasPrefix(u, []byte("ws://")) {
224 return u[5:]
225 }
226 return u
227 }
228