source: code/trunk/user.go@ 398

Last change on this file since 398 was 398, checked in by contact, 5 years ago

Implement rate limiting for upstream messages

Allow up to 10 outgoing messages in a burst, then throttle to 1 message
each 2 seconds.

Closes: ​https://todo.sr.ht/~emersion/soju/87
File size: 13.6 KB
Line 
1package soju
2
3import (
4 "crypto/sha256"
5 "encoding/binary"
6 "encoding/hex"
7 "fmt"
8 "time"
9
10 "gopkg.in/irc.v3"
11)
12
13type event interface{}
14
15type eventUpstreamMessage struct {
16 msg *irc.Message
17 uc *upstreamConn
18}
19
20type eventUpstreamConnectionError struct {
21 net *network
22 err error
23}
24
25type eventUpstreamConnected struct {
26 uc *upstreamConn
27}
28
29type eventUpstreamDisconnected struct {
30 uc *upstreamConn
31}
32
33type eventUpstreamError struct {
34 uc *upstreamConn
35 err error
36}
37
38type eventDownstreamMessage struct {
39 msg *irc.Message
40 dc *downstreamConn
41}
42
43type eventDownstreamConnected struct {
44 dc *downstreamConn
45}
46
47type eventDownstreamDisconnected struct {
48 dc *downstreamConn
49}
50
51type eventStop struct{}
52
53type networkHistory struct {
54 offlineClients map[string]uint64 // indexed by client name
55 ring *Ring // can be nil if there are no offline clients
56}
57
58type network struct {
59 Network
60 user *user
61 stopped chan struct{}
62
63 conn *upstreamConn
64 channels map[string]*Channel
65 history map[string]*networkHistory // indexed by entity
66 offlineClients map[string]struct{} // indexed by client name
67 lastError error
68}
69
70func newNetwork(user *user, record *Network, channels []Channel) *network {
71 m := make(map[string]*Channel, len(channels))
72 for _, ch := range channels {
73 ch := ch
74 m[ch.Name] = &ch
75 }
76
77 return &network{
78 Network: *record,
79 user: user,
80 stopped: make(chan struct{}),
81 channels: m,
82 history: make(map[string]*networkHistory),
83 offlineClients: make(map[string]struct{}),
84 }
85}
86
87func (net *network) forEachDownstream(f func(*downstreamConn)) {
88 net.user.forEachDownstream(func(dc *downstreamConn) {
89 if dc.network != nil && dc.network != net {
90 return
91 }
92 f(dc)
93 })
94}
95
96func (net *network) isStopped() bool {
97 select {
98 case <-net.stopped:
99 return true
100 default:
101 return false
102 }
103}
104
105func userIdent(u *User) string {
106 // The ident is a string we will send to upstream servers in clear-text.
107 // For privacy reasons, make sure it doesn't expose any meaningful user
108 // metadata. We just use the base64-encoded hashed ID, so that people don't
109 // start relying on the string being an integer or following a pattern.
110 var b [64]byte
111 binary.LittleEndian.PutUint64(b[:], uint64(u.ID))
112 h := sha256.Sum256(b[:])
113 return hex.EncodeToString(h[:16])
114}
115
116func (net *network) run() {
117 var lastTry time.Time
118 for {
119 if net.isStopped() {
120 return
121 }
122
123 if dur := time.Now().Sub(lastTry); dur < retryConnectDelay {
124 delay := retryConnectDelay - dur
125 net.user.srv.Logger.Printf("waiting %v before trying to reconnect to %q", delay.Truncate(time.Second), net.Addr)
126 time.Sleep(delay)
127 }
128 lastTry = time.Now()
129
130 uc, err := connectToUpstream(net)
131 if err != nil {
132 net.user.srv.Logger.Printf("failed to connect to upstream server %q: %v", net.Addr, err)
133 net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("failed to connect: %v", err)}
134 continue
135 }
136
137 if net.user.srv.Identd != nil {
138 net.user.srv.Identd.Store(uc.RemoteAddr().String(), uc.LocalAddr().String(), userIdent(&net.user.User))
139 }
140
141 uc.register()
142 if err := uc.runUntilRegistered(); err != nil {
143 uc.logger.Printf("failed to register: %v", err)
144 net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("failed to register: %v", err)}
145 uc.Close()
146 continue
147 }
148
149 // TODO: this is racy with net.stopped. If the network is stopped
150 // before the user goroutine receives eventUpstreamConnected, the
151 // connection won't be closed.
152 net.user.events <- eventUpstreamConnected{uc}
153 if err := uc.readMessages(net.user.events); err != nil {
154 uc.logger.Printf("failed to handle messages: %v", err)
155 net.user.events <- eventUpstreamError{uc, fmt.Errorf("failed to handle messages: %v", err)}
156 }
157 uc.Close()
158 net.user.events <- eventUpstreamDisconnected{uc}
159
160 if net.user.srv.Identd != nil {
161 net.user.srv.Identd.Delete(uc.RemoteAddr().String(), uc.LocalAddr().String())
162 }
163 }
164}
165
166func (net *network) stop() {
167 if !net.isStopped() {
168 close(net.stopped)
169 }
170
171 if net.conn != nil {
172 net.conn.Close()
173 }
174}
175
176func (net *network) createUpdateChannel(ch *Channel) error {
177 if current, ok := net.channels[ch.Name]; ok {
178 ch.ID = current.ID // update channel if it already exists
179 }
180 if err := net.user.srv.db.StoreChannel(net.ID, ch); err != nil {
181 return err
182 }
183 prev := net.channels[ch.Name]
184 net.channels[ch.Name] = ch
185
186 if prev != nil && prev.Detached != ch.Detached {
187 history := net.history[ch.Name]
188 if ch.Detached {
189 net.user.srv.Logger.Printf("network %q: detaching channel %q", net.GetName(), ch.Name)
190 net.forEachDownstream(func(dc *downstreamConn) {
191 net.offlineClients[dc.clientName] = struct{}{}
192 if history != nil {
193 history.offlineClients[dc.clientName] = history.ring.Cur()
194 }
195
196 dc.SendMessage(&irc.Message{
197 Prefix: dc.prefix(),
198 Command: "PART",
199 Params: []string{dc.marshalEntity(net, ch.Name), "Detach"},
200 })
201 })
202 } else {
203 net.user.srv.Logger.Printf("network %q: attaching channel %q", net.GetName(), ch.Name)
204
205 var uch *upstreamChannel
206 if net.conn != nil {
207 uch = net.conn.channels[ch.Name]
208 }
209
210 net.forEachDownstream(func(dc *downstreamConn) {
211 dc.SendMessage(&irc.Message{
212 Prefix: dc.prefix(),
213 Command: "JOIN",
214 Params: []string{dc.marshalEntity(net, ch.Name)},
215 })
216
217 if uch != nil {
218 forwardChannel(dc, uch)
219 }
220
221 if history != nil {
222 dc.sendNetworkHistory(net)
223 }
224 })
225 }
226 }
227
228 return nil
229}
230
231func (net *network) deleteChannel(name string) error {
232 if err := net.user.srv.db.DeleteChannel(net.ID, name); err != nil {
233 return err
234 }
235 delete(net.channels, name)
236 return nil
237}
238
239type user struct {
240 User
241 srv *Server
242
243 events chan event
244 done chan struct{}
245
246 networks []*network
247 downstreamConns []*downstreamConn
248
249 // LIST commands in progress
250 pendingLISTs []pendingLIST
251}
252
253type pendingLIST struct {
254 downstreamID uint64
255 // list of per-upstream LIST commands not yet sent or completed
256 pendingCommands map[int64]*irc.Message
257}
258
259func newUser(srv *Server, record *User) *user {
260 return &user{
261 User: *record,
262 srv: srv,
263 events: make(chan event, 64),
264 done: make(chan struct{}),
265 }
266}
267
268func (u *user) forEachNetwork(f func(*network)) {
269 for _, network := range u.networks {
270 f(network)
271 }
272}
273
274func (u *user) forEachUpstream(f func(uc *upstreamConn)) {
275 for _, network := range u.networks {
276 if network.conn == nil {
277 continue
278 }
279 f(network.conn)
280 }
281}
282
283func (u *user) forEachDownstream(f func(dc *downstreamConn)) {
284 for _, dc := range u.downstreamConns {
285 f(dc)
286 }
287}
288
289func (u *user) getNetwork(name string) *network {
290 for _, network := range u.networks {
291 if network.Addr == name {
292 return network
293 }
294 if network.Name != "" && network.Name == name {
295 return network
296 }
297 }
298 return nil
299}
300
301func (u *user) getNetworkByID(id int64) *network {
302 for _, net := range u.networks {
303 if net.ID == id {
304 return net
305 }
306 }
307 return nil
308}
309
310func (u *user) run() {
311 defer close(u.done)
312
313 networks, err := u.srv.db.ListNetworks(u.Username)
314 if err != nil {
315 u.srv.Logger.Printf("failed to list networks for user %q: %v", u.Username, err)
316 return
317 }
318
319 for _, record := range networks {
320 record := record
321 channels, err := u.srv.db.ListChannels(record.ID)
322 if err != nil {
323 u.srv.Logger.Printf("failed to list channels for user %q, network %q: %v", u.Username, record.GetName(), err)
324 continue
325 }
326
327 network := newNetwork(u, &record, channels)
328 u.networks = append(u.networks, network)
329
330 go network.run()
331 }
332
333 for e := range u.events {
334 switch e := e.(type) {
335 case eventUpstreamConnected:
336 uc := e.uc
337
338 uc.network.conn = uc
339
340 uc.updateAway()
341
342 uc.forEachDownstream(func(dc *downstreamConn) {
343 dc.updateSupportedCaps()
344 sendServiceNOTICE(dc, fmt.Sprintf("connected to %s", uc.network.GetName()))
345
346 dc.updateNick()
347 })
348 uc.network.lastError = nil
349 case eventUpstreamDisconnected:
350 u.handleUpstreamDisconnected(e.uc)
351 case eventUpstreamConnectionError:
352 net := e.net
353
354 stopped := false
355 select {
356 case <-net.stopped:
357 stopped = true
358 default:
359 }
360
361 if !stopped && (net.lastError == nil || net.lastError.Error() != e.err.Error()) {
362 net.forEachDownstream(func(dc *downstreamConn) {
363 sendServiceNOTICE(dc, fmt.Sprintf("failed connecting/registering to %s: %v", net.GetName(), e.err))
364 })
365 }
366 net.lastError = e.err
367 case eventUpstreamError:
368 uc := e.uc
369
370 uc.forEachDownstream(func(dc *downstreamConn) {
371 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s: %v", uc.network.GetName(), e.err))
372 })
373 uc.network.lastError = e.err
374 case eventUpstreamMessage:
375 msg, uc := e.msg, e.uc
376 if uc.isClosed() {
377 uc.logger.Printf("ignoring message on closed connection: %v", msg)
378 break
379 }
380 if err := uc.handleMessage(msg); err != nil {
381 uc.logger.Printf("failed to handle message %q: %v", msg, err)
382 }
383 case eventDownstreamConnected:
384 dc := e.dc
385
386 if err := dc.welcome(); err != nil {
387 dc.logger.Printf("failed to handle new registered connection: %v", err)
388 break
389 }
390
391 u.downstreamConns = append(u.downstreamConns, dc)
392
393 u.forEachUpstream(func(uc *upstreamConn) {
394 uc.updateAway()
395 })
396
397 dc.updateSupportedCaps()
398 case eventDownstreamDisconnected:
399 dc := e.dc
400
401 for i := range u.downstreamConns {
402 if u.downstreamConns[i] == dc {
403 u.downstreamConns = append(u.downstreamConns[:i], u.downstreamConns[i+1:]...)
404 break
405 }
406 }
407
408 // Save history if we're the last client with this name
409 skipHistory := make(map[*network]bool)
410 u.forEachDownstream(func(conn *downstreamConn) {
411 if dc.clientName == conn.clientName {
412 skipHistory[conn.network] = true
413 }
414 })
415
416 dc.forEachNetwork(func(net *network) {
417 if skipHistory[net] || skipHistory[nil] {
418 return
419 }
420
421 net.offlineClients[dc.clientName] = struct{}{}
422 for target, history := range net.history {
423 if ch, ok := net.channels[target]; ok && ch.Detached {
424 continue
425 }
426 history.offlineClients[dc.clientName] = history.ring.Cur()
427 }
428 })
429
430 u.forEachUpstream(func(uc *upstreamConn) {
431 uc.updateAway()
432 })
433 case eventDownstreamMessage:
434 msg, dc := e.msg, e.dc
435 if dc.isClosed() {
436 dc.logger.Printf("ignoring message on closed connection: %v", msg)
437 break
438 }
439 err := dc.handleMessage(msg)
440 if ircErr, ok := err.(ircError); ok {
441 ircErr.Message.Prefix = dc.srv.prefix()
442 dc.SendMessage(ircErr.Message)
443 } else if err != nil {
444 dc.logger.Printf("failed to handle message %q: %v", msg, err)
445 dc.Close()
446 }
447 case eventStop:
448 u.forEachDownstream(func(dc *downstreamConn) {
449 dc.Close()
450 })
451 for _, n := range u.networks {
452 n.stop()
453 }
454 return
455 default:
456 u.srv.Logger.Printf("received unknown event type: %T", e)
457 }
458 }
459}
460
461func (u *user) handleUpstreamDisconnected(uc *upstreamConn) {
462 uc.network.conn = nil
463
464 for _, ml := range uc.messageLoggers {
465 if err := ml.Close(); err != nil {
466 uc.logger.Printf("failed to close message logger: %v", err)
467 }
468 }
469
470 uc.endPendingLISTs(true)
471
472 uc.forEachDownstream(func(dc *downstreamConn) {
473 dc.updateSupportedCaps()
474 })
475
476 if uc.network.lastError == nil {
477 uc.forEachDownstream(func(dc *downstreamConn) {
478 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s", uc.network.GetName()))
479 })
480 }
481}
482
483func (u *user) addNetwork(network *network) {
484 u.networks = append(u.networks, network)
485 go network.run()
486}
487
488func (u *user) removeNetwork(network *network) {
489 network.stop()
490
491 u.forEachDownstream(func(dc *downstreamConn) {
492 if dc.network != nil && dc.network == network {
493 dc.Close()
494 }
495 })
496
497 for i, net := range u.networks {
498 if net == network {
499 u.networks = append(u.networks[:i], u.networks[i+1:]...)
500 return
501 }
502 }
503
504 panic("tried to remove a non-existing network")
505}
506
507func (u *user) createNetwork(record *Network) (*network, error) {
508 if record.ID != 0 {
509 panic("tried creating an already-existing network")
510 }
511
512 network := newNetwork(u, record, nil)
513 err := u.srv.db.StoreNetwork(u.Username, &network.Network)
514 if err != nil {
515 return nil, err
516 }
517
518 u.addNetwork(network)
519
520 return network, nil
521}
522
523func (u *user) updateNetwork(record *Network) (*network, error) {
524 if record.ID == 0 {
525 panic("tried updating a new network")
526 }
527
528 network := u.getNetworkByID(record.ID)
529 if network == nil {
530 panic("tried updating a non-existing network")
531 }
532
533 if err := u.srv.db.StoreNetwork(u.Username, record); err != nil {
534 return nil, err
535 }
536
537 // Most network changes require us to re-connect to the upstream server
538
539 channels := make([]Channel, 0, len(network.channels))
540 for _, ch := range network.channels {
541 channels = append(channels, *ch)
542 }
543
544 updatedNetwork := newNetwork(u, record, channels)
545
546 // If we're currently connected, disconnect and perform the necessary
547 // bookkeeping
548 if network.conn != nil {
549 network.stop()
550 // Note: this will set network.conn to nil
551 u.handleUpstreamDisconnected(network.conn)
552 }
553
554 // Patch downstream connections to use our fresh updated network
555 u.forEachDownstream(func(dc *downstreamConn) {
556 if dc.network != nil && dc.network == network {
557 dc.network = updatedNetwork
558 }
559 })
560
561 // We need to remove the network after patching downstream connections,
562 // otherwise they'll get closed
563 u.removeNetwork(network)
564
565 // This will re-connect to the upstream server
566 u.addNetwork(updatedNetwork)
567
568 return updatedNetwork, nil
569}
570
571func (u *user) deleteNetwork(id int64) error {
572 network := u.getNetworkByID(id)
573 if network == nil {
574 panic("tried deleting a non-existing network")
575 }
576
577 if err := u.srv.db.DeleteNetwork(network.ID); err != nil {
578 return err
579 }
580
581 u.removeNetwork(network)
582 return nil
583}
584
585func (u *user) updatePassword(hashed string) error {
586 u.User.Password = hashed
587 return u.srv.db.StoreUser(&u.User)
588}
589
590func (u *user) stop() {
591 u.events <- eventStop{}
592 <-u.done
593}
Note: See TracBrowser for help on using the repository browser.