source: code/trunk/user.go@ 735

Last change on this file since 735 was 735, checked in by contact, 4 years ago

Add exponential backoff when re-connecting to upstream

The first reconnection attempt waits for 1min, the second the 2min,
and so on up to 10min. There's a 1min jitter so that multiple failed
connections don't try to reconnect at the exact same time.

Closes: https://todo.sr.ht/~emersion/soju/161

File size: 25.2 KB
RevLine 
[101]1package soju
2
3import (
[652]4 "context"
[385]5 "crypto/sha256"
6 "encoding/binary"
[395]7 "encoding/hex"
[218]8 "fmt"
[705]9 "math/big"
10 "net"
[101]11 "time"
[103]12
13 "gopkg.in/irc.v3"
[101]14)
15
[165]16type event interface{}
17
18type eventUpstreamMessage struct {
[103]19 msg *irc.Message
20 uc *upstreamConn
21}
22
[218]23type eventUpstreamConnectionError struct {
24 net *network
25 err error
26}
27
[196]28type eventUpstreamConnected struct {
29 uc *upstreamConn
30}
31
[179]32type eventUpstreamDisconnected struct {
33 uc *upstreamConn
34}
35
[218]36type eventUpstreamError struct {
37 uc *upstreamConn
38 err error
39}
40
[165]41type eventDownstreamMessage struct {
[103]42 msg *irc.Message
43 dc *downstreamConn
44}
45
[166]46type eventDownstreamConnected struct {
47 dc *downstreamConn
48}
49
[167]50type eventDownstreamDisconnected struct {
51 dc *downstreamConn
52}
53
[435]54type eventChannelDetach struct {
55 uc *upstreamConn
56 name string
57}
58
[563]59type eventBroadcast struct {
60 msg *irc.Message
61}
62
[376]63type eventStop struct{}
64
[625]65type eventUserUpdate struct {
66 password *string
67 admin *bool
68 done chan error
69}
70
[480]71type deliveredClientMap map[string]string // client name -> msg ID
72
[485]73type deliveredStore struct {
74 m deliveredCasemapMap
75}
76
77func newDeliveredStore() deliveredStore {
78 return deliveredStore{deliveredCasemapMap{newCasemapMap(0)}}
79}
80
81func (ds deliveredStore) HasTarget(target string) bool {
82 return ds.m.Value(target) != nil
83}
84
85func (ds deliveredStore) LoadID(target, clientName string) string {
86 clients := ds.m.Value(target)
87 if clients == nil {
88 return ""
89 }
90 return clients[clientName]
91}
92
93func (ds deliveredStore) StoreID(target, clientName, msgID string) {
94 clients := ds.m.Value(target)
95 if clients == nil {
96 clients = make(deliveredClientMap)
97 ds.m.SetValue(target, clients)
98 }
99 clients[clientName] = msgID
100}
101
102func (ds deliveredStore) ForEachTarget(f func(target string)) {
103 for _, entry := range ds.m.innerMap {
104 f(entry.originalKey)
105 }
106}
107
[489]108func (ds deliveredStore) ForEachClient(f func(clientName string)) {
109 clients := make(map[string]struct{})
110 for _, entry := range ds.m.innerMap {
111 delivered := entry.value.(deliveredClientMap)
112 for clientName := range delivered {
113 clients[clientName] = struct{}{}
114 }
115 }
116
117 for clientName := range clients {
118 f(clientName)
119 }
120}
121
[101]122type network struct {
123 Network
[202]124 user *user
[501]125 logger Logger
[202]126 stopped chan struct{}
[131]127
[482]128 conn *upstreamConn
129 channels channelCasemapMap
[485]130 delivered deliveredStore
[482]131 lastError error
132 casemap casemapping
[101]133}
134
[267]135func newNetwork(user *user, record *Network, channels []Channel) *network {
[501]136 logger := &prefixLogger{user.logger, fmt.Sprintf("network %q: ", record.GetName())}
137
[478]138 m := channelCasemapMap{newCasemapMap(0)}
[267]139 for _, ch := range channels {
[283]140 ch := ch
[478]141 m.SetValue(ch.Name, &ch)
[267]142 }
143
[101]144 return &network{
[482]145 Network: *record,
146 user: user,
[501]147 logger: logger,
[482]148 stopped: make(chan struct{}),
149 channels: m,
[485]150 delivered: newDeliveredStore(),
[482]151 casemap: casemapRFC1459,
[101]152 }
153}
154
[218]155func (net *network) forEachDownstream(f func(*downstreamConn)) {
156 net.user.forEachDownstream(func(dc *downstreamConn) {
[693]157 if dc.network == nil && !dc.isMultiUpstream {
[532]158 return
159 }
[218]160 if dc.network != nil && dc.network != net {
161 return
162 }
163 f(dc)
164 })
165}
166
[311]167func (net *network) isStopped() bool {
168 select {
169 case <-net.stopped:
170 return true
171 default:
172 return false
173 }
174}
175
[385]176func userIdent(u *User) string {
177 // The ident is a string we will send to upstream servers in clear-text.
178 // For privacy reasons, make sure it doesn't expose any meaningful user
179 // metadata. We just use the base64-encoded hashed ID, so that people don't
180 // start relying on the string being an integer or following a pattern.
181 var b [64]byte
182 binary.LittleEndian.PutUint64(b[:], uint64(u.ID))
183 h := sha256.Sum256(b[:])
[395]184 return hex.EncodeToString(h[:16])
[385]185}
186
[101]187func (net *network) run() {
[542]188 if !net.Enabled {
189 return
190 }
191
[101]192 var lastTry time.Time
[735]193 backoff := newBackoffer(retryConnectMinDelay, retryConnectMaxDelay, retryConnectJitter)
[101]194 for {
[311]195 if net.isStopped() {
[202]196 return
197 }
198
[735]199 delay := backoff.Next() - time.Now().Sub(lastTry)
200 if delay > 0 {
[501]201 net.logger.Printf("waiting %v before trying to reconnect to %q", delay.Truncate(time.Second), net.Addr)
[101]202 time.Sleep(delay)
203 }
204 lastTry = time.Now()
205
[733]206 net.user.srv.metrics.upstreams.Add(1)
207
[732]208 uc, err := connectToUpstream(context.TODO(), net)
[101]209 if err != nil {
[501]210 net.logger.Printf("failed to connect to upstream server %q: %v", net.Addr, err)
[218]211 net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("failed to connect: %v", err)}
[733]212 net.user.srv.metrics.upstreams.Add(-1)
[734]213 net.user.srv.metrics.upstreamConnectErrorsTotal.Inc()
[101]214 continue
215 }
216
[385]217 if net.user.srv.Identd != nil {
218 net.user.srv.Identd.Store(uc.RemoteAddr().String(), uc.LocalAddr().String(), userIdent(&net.user.User))
219 }
220
[101]221 uc.register()
[197]222 if err := uc.runUntilRegistered(); err != nil {
[399]223 text := err.Error()
224 if regErr, ok := err.(registrationError); ok {
225 text = string(regErr)
226 }
227 uc.logger.Printf("failed to register: %v", text)
228 net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("failed to register: %v", text)}
[197]229 uc.Close()
[733]230 net.user.srv.metrics.upstreams.Add(-1)
[734]231 net.user.srv.metrics.upstreamConnectErrorsTotal.Inc()
[197]232 continue
233 }
[101]234
[311]235 // TODO: this is racy with net.stopped. If the network is stopped
236 // before the user goroutine receives eventUpstreamConnected, the
237 // connection won't be closed.
[196]238 net.user.events <- eventUpstreamConnected{uc}
[165]239 if err := uc.readMessages(net.user.events); err != nil {
[101]240 uc.logger.Printf("failed to handle messages: %v", err)
[218]241 net.user.events <- eventUpstreamError{uc, fmt.Errorf("failed to handle messages: %v", err)}
[101]242 }
243 uc.Close()
[179]244 net.user.events <- eventUpstreamDisconnected{uc}
[385]245
246 if net.user.srv.Identd != nil {
247 net.user.srv.Identd.Delete(uc.RemoteAddr().String(), uc.LocalAddr().String())
248 }
[710]249
250 net.user.srv.metrics.upstreams.Add(-1)
[735]251 backoff.Reset()
[101]252 }
253}
254
[309]255func (net *network) stop() {
[311]256 if !net.isStopped() {
[202]257 close(net.stopped)
258 }
259
[279]260 if net.conn != nil {
261 net.conn.Close()
[202]262 }
263}
264
[435]265func (net *network) detach(ch *Channel) {
266 if ch.Detached {
267 return
[267]268 }
[497]269
[501]270 net.logger.Printf("detaching channel %q", ch.Name)
[435]271
[497]272 ch.Detached = true
273
274 if net.user.msgStore != nil {
275 nameCM := net.casemap(ch.Name)
[666]276 lastID, err := net.user.msgStore.LastMsgID(&net.Network, nameCM, time.Now())
[497]277 if err != nil {
[501]278 net.logger.Printf("failed to get last message ID for channel %q: %v", ch.Name, err)
[497]279 }
280 ch.DetachedInternalMsgID = lastID
281 }
282
[435]283 if net.conn != nil {
[478]284 uch := net.conn.channels.Value(ch.Name)
285 if uch != nil {
[435]286 uch.updateAutoDetach(0)
287 }
[222]288 }
[284]289
[435]290 net.forEachDownstream(func(dc *downstreamConn) {
291 dc.SendMessage(&irc.Message{
292 Prefix: dc.prefix(),
293 Command: "PART",
294 Params: []string{dc.marshalEntity(net, ch.Name), "Detach"},
295 })
296 })
297}
[284]298
[435]299func (net *network) attach(ch *Channel) {
300 if !ch.Detached {
301 return
302 }
[497]303
[501]304 net.logger.Printf("attaching channel %q", ch.Name)
[284]305
[497]306 detachedMsgID := ch.DetachedInternalMsgID
307 ch.Detached = false
308 ch.DetachedInternalMsgID = ""
309
[435]310 var uch *upstreamChannel
311 if net.conn != nil {
[478]312 uch = net.conn.channels.Value(ch.Name)
[284]313
[435]314 net.conn.updateChannelAutoDetach(ch.Name)
315 }
[284]316
[435]317 net.forEachDownstream(func(dc *downstreamConn) {
318 dc.SendMessage(&irc.Message{
319 Prefix: dc.prefix(),
320 Command: "JOIN",
321 Params: []string{dc.marshalEntity(net, ch.Name)},
322 })
323
324 if uch != nil {
325 forwardChannel(dc, uch)
[284]326 }
327
[497]328 if detachedMsgID != "" {
[701]329 dc.sendTargetBacklog(context.TODO(), net, ch.Name, detachedMsgID)
[495]330 }
[435]331 })
[222]332}
333
[676]334func (net *network) deleteChannel(ctx context.Context, name string) error {
[478]335 ch := net.channels.Value(name)
336 if ch == nil {
[416]337 return fmt.Errorf("unknown channel %q", name)
338 }
[435]339 if net.conn != nil {
[478]340 uch := net.conn.channels.Value(ch.Name)
341 if uch != nil {
[435]342 uch.updateAutoDetach(0)
343 }
344 }
345
[676]346 if err := net.user.srv.db.DeleteChannel(ctx, ch.ID); err != nil {
[267]347 return err
348 }
[478]349 net.channels.Delete(name)
[267]350 return nil
[222]351}
352
[478]353func (net *network) updateCasemapping(newCasemap casemapping) {
354 net.casemap = newCasemap
355 net.channels.SetCasemapping(newCasemap)
[485]356 net.delivered.m.SetCasemapping(newCasemap)
[684]357 if uc := net.conn; uc != nil {
358 uc.channels.SetCasemapping(newCasemap)
359 for _, entry := range uc.channels.innerMap {
[478]360 uch := entry.value.(*upstreamChannel)
361 uch.Members.SetCasemapping(newCasemap)
362 }
[684]363 uc.monitored.SetCasemapping(newCasemap)
[478]364 }
[684]365 net.forEachDownstream(func(dc *downstreamConn) {
366 dc.monitored.SetCasemapping(newCasemap)
367 })
[478]368}
369
[489]370func (net *network) storeClientDeliveryReceipts(clientName string) {
371 if !net.user.hasPersistentMsgStore() {
372 return
373 }
374
375 var receipts []DeliveryReceipt
376 net.delivered.ForEachTarget(func(target string) {
377 msgID := net.delivered.LoadID(target, clientName)
378 if msgID == "" {
379 return
380 }
381 receipts = append(receipts, DeliveryReceipt{
382 Target: target,
383 InternalMsgID: msgID,
384 })
385 })
386
[652]387 if err := net.user.srv.db.StoreClientDeliveryReceipts(context.TODO(), net.ID, clientName, receipts); err != nil {
[501]388 net.logger.Printf("failed to store delivery receipts for client %q: %v", clientName, err)
[489]389 }
390}
391
[499]392func (net *network) isHighlight(msg *irc.Message) bool {
393 if msg.Command != "PRIVMSG" && msg.Command != "NOTICE" {
394 return false
395 }
396
397 text := msg.Params[1]
398
399 nick := net.Nick
400 if net.conn != nil {
401 nick = net.conn.nick
402 }
403
404 // TODO: use case-mapping aware comparison here
405 return msg.Prefix.Name != nick && isHighlight(text, nick)
406}
407
408func (net *network) detachedMessageNeedsRelay(ch *Channel, msg *irc.Message) bool {
409 highlight := net.isHighlight(msg)
410 return ch.RelayDetached == FilterMessage || ((ch.RelayDetached == FilterHighlight || ch.RelayDetached == FilterDefault) && highlight)
411}
412
[724]413func (net *network) autoSaveSASLPlain(ctx context.Context, username, password string) {
414 // User may have e.g. EXTERNAL mechanism configured. We do not want to
415 // automatically erase the key pair or any other credentials.
416 if net.SASL.Mechanism != "" && net.SASL.Mechanism != "PLAIN" {
417 return
418 }
419
420 net.logger.Printf("auto-saving SASL PLAIN credentials with username %q", username)
421 net.SASL.Mechanism = "PLAIN"
422 net.SASL.Plain.Username = username
423 net.SASL.Plain.Password = password
424 if err := net.user.srv.db.StoreNetwork(ctx, net.user.ID, &net.Network); err != nil {
425 net.logger.Printf("failed to save SASL PLAIN credentials: %v", err)
426 }
427}
428
[101]429type user struct {
430 User
[493]431 srv *Server
432 logger Logger
[101]433
[165]434 events chan event
[377]435 done chan struct{}
[103]436
[101]437 networks []*network
438 downstreamConns []*downstreamConn
[439]439 msgStore messageStore
[101]440}
441
442func newUser(srv *Server, record *User) *user {
[493]443 logger := &prefixLogger{srv.Logger, fmt.Sprintf("user %q: ", record.Username)}
444
[439]445 var msgStore messageStore
[691]446 if logPath := srv.Config().LogPath; logPath != "" {
447 msgStore = newFSMessageStore(logPath, record.Username)
[442]448 } else {
449 msgStore = newMemoryMessageStore()
[423]450 }
451
[101]452 return &user{
[489]453 User: *record,
454 srv: srv,
[493]455 logger: logger,
[489]456 events: make(chan event, 64),
457 done: make(chan struct{}),
458 msgStore: msgStore,
[101]459 }
460}
461
462func (u *user) forEachNetwork(f func(*network)) {
463 for _, network := range u.networks {
464 f(network)
465 }
466}
467
468func (u *user) forEachUpstream(f func(uc *upstreamConn)) {
469 for _, network := range u.networks {
[279]470 if network.conn == nil {
[101]471 continue
472 }
[279]473 f(network.conn)
[101]474 }
475}
476
477func (u *user) forEachDownstream(f func(dc *downstreamConn)) {
478 for _, dc := range u.downstreamConns {
479 f(dc)
480 }
481}
482
483func (u *user) getNetwork(name string) *network {
484 for _, network := range u.networks {
485 if network.Addr == name {
486 return network
487 }
[201]488 if network.Name != "" && network.Name == name {
489 return network
490 }
[101]491 }
492 return nil
493}
494
[313]495func (u *user) getNetworkByID(id int64) *network {
496 for _, net := range u.networks {
497 if net.ID == id {
498 return net
499 }
500 }
501 return nil
502}
503
[101]504func (u *user) run() {
[423]505 defer func() {
506 if u.msgStore != nil {
507 if err := u.msgStore.Close(); err != nil {
[493]508 u.logger.Printf("failed to close message store for user %q: %v", u.Username, err)
[423]509 }
510 }
511 close(u.done)
512 }()
[377]513
[652]514 networks, err := u.srv.db.ListNetworks(context.TODO(), u.ID)
[101]515 if err != nil {
[493]516 u.logger.Printf("failed to list networks for user %q: %v", u.Username, err)
[101]517 return
518 }
519
520 for _, record := range networks {
[283]521 record := record
[652]522 channels, err := u.srv.db.ListChannels(context.TODO(), record.ID)
[267]523 if err != nil {
[493]524 u.logger.Printf("failed to list channels for user %q, network %q: %v", u.Username, record.GetName(), err)
[359]525 continue
[267]526 }
527
528 network := newNetwork(u, &record, channels)
[101]529 u.networks = append(u.networks, network)
530
[489]531 if u.hasPersistentMsgStore() {
[652]532 receipts, err := u.srv.db.ListDeliveryReceipts(context.TODO(), record.ID)
[489]533 if err != nil {
[493]534 u.logger.Printf("failed to load delivery receipts for user %q, network %q: %v", u.Username, network.GetName(), err)
[489]535 return
536 }
537
538 for _, rcpt := range receipts {
539 network.delivered.StoreID(rcpt.Target, rcpt.Client, rcpt.InternalMsgID)
540 }
541 }
542
[101]543 go network.run()
544 }
[103]545
[165]546 for e := range u.events {
547 switch e := e.(type) {
[196]548 case eventUpstreamConnected:
[198]549 uc := e.uc
[199]550
551 uc.network.conn = uc
552
[198]553 uc.updateAway()
[684]554 uc.updateMonitor()
[218]555
[532]556 netIDStr := fmt.Sprintf("%v", uc.network.ID)
[218]557 uc.forEachDownstream(func(dc *downstreamConn) {
[276]558 dc.updateSupportedCaps()
[296]559
[543]560 if !dc.caps["soju.im/bouncer-networks"] {
561 sendServiceNOTICE(dc, fmt.Sprintf("connected to %s", uc.network.GetName()))
562 }
563
564 dc.updateNick()
565 dc.updateRealname()
[722]566 dc.updateAccount()
[543]567 })
568 u.forEachDownstream(func(dc *downstreamConn) {
[535]569 if dc.caps["soju.im/bouncer-networks-notify"] {
[532]570 dc.SendMessage(&irc.Message{
571 Prefix: dc.srv.prefix(),
572 Command: "BOUNCER",
[544]573 Params: []string{"NETWORK", netIDStr, "state=connected"},
[532]574 })
575 }
[218]576 })
577 uc.network.lastError = nil
[179]578 case eventUpstreamDisconnected:
[313]579 u.handleUpstreamDisconnected(e.uc)
580 case eventUpstreamConnectionError:
581 net := e.net
[199]582
[313]583 stopped := false
584 select {
585 case <-net.stopped:
586 stopped = true
587 default:
[179]588 }
[199]589
[313]590 if !stopped && (net.lastError == nil || net.lastError.Error() != e.err.Error()) {
[218]591 net.forEachDownstream(func(dc *downstreamConn) {
[223]592 sendServiceNOTICE(dc, fmt.Sprintf("failed connecting/registering to %s: %v", net.GetName(), e.err))
[218]593 })
594 }
595 net.lastError = e.err
596 case eventUpstreamError:
597 uc := e.uc
598
599 uc.forEachDownstream(func(dc *downstreamConn) {
[223]600 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s: %v", uc.network.GetName(), e.err))
[218]601 })
602 uc.network.lastError = e.err
[165]603 case eventUpstreamMessage:
604 msg, uc := e.msg, e.uc
[175]605 if uc.isClosed() {
[133]606 uc.logger.Printf("ignoring message on closed connection: %v", msg)
607 break
608 }
[103]609 if err := uc.handleMessage(msg); err != nil {
610 uc.logger.Printf("failed to handle message %q: %v", msg, err)
611 }
[435]612 case eventChannelDetach:
613 uc, name := e.uc, e.name
[478]614 c := uc.network.channels.Value(name)
615 if c == nil || c.Detached {
[435]616 continue
617 }
618 uc.network.detach(c)
[652]619 if err := uc.srv.db.StoreChannel(context.TODO(), uc.network.ID, c); err != nil {
[493]620 u.logger.Printf("failed to store updated detached channel %q: %v", c.Name, err)
[435]621 }
[166]622 case eventDownstreamConnected:
623 dc := e.dc
[168]624
[684]625 if dc.network != nil {
626 dc.monitored.SetCasemapping(dc.network.casemap)
627 }
628
[701]629 if err := dc.welcome(context.TODO()); err != nil {
[168]630 dc.logger.Printf("failed to handle new registered connection: %v", err)
631 break
632 }
633
[166]634 u.downstreamConns = append(u.downstreamConns, dc)
[198]635
[467]636 dc.forEachNetwork(func(network *network) {
637 if network.lastError != nil {
638 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s: %v", network.GetName(), network.lastError))
639 }
640 })
641
[198]642 u.forEachUpstream(func(uc *upstreamConn) {
643 uc.updateAway()
644 })
[167]645 case eventDownstreamDisconnected:
646 dc := e.dc
[204]647
[167]648 for i := range u.downstreamConns {
649 if u.downstreamConns[i] == dc {
650 u.downstreamConns = append(u.downstreamConns[:i], u.downstreamConns[i+1:]...)
651 break
652 }
653 }
[198]654
[489]655 dc.forEachNetwork(func(net *network) {
656 net.storeClientDeliveryReceipts(dc.clientName)
657 })
658
[198]659 u.forEachUpstream(func(uc *upstreamConn) {
660 uc.updateAway()
[684]661 uc.updateMonitor()
[198]662 })
[165]663 case eventDownstreamMessage:
664 msg, dc := e.msg, e.dc
[133]665 if dc.isClosed() {
666 dc.logger.Printf("ignoring message on closed connection: %v", msg)
667 break
668 }
[704]669 err := dc.handleMessage(context.TODO(), msg)
[103]670 if ircErr, ok := err.(ircError); ok {
671 ircErr.Message.Prefix = dc.srv.prefix()
672 dc.SendMessage(ircErr.Message)
673 } else if err != nil {
674 dc.logger.Printf("failed to handle message %q: %v", msg, err)
675 dc.Close()
676 }
[563]677 case eventBroadcast:
678 msg := e.msg
679 u.forEachDownstream(func(dc *downstreamConn) {
680 dc.SendMessage(msg)
681 })
[625]682 case eventUserUpdate:
683 // copy the user record because we'll mutate it
684 record := u.User
685
686 if e.password != nil {
687 record.Password = *e.password
688 }
689 if e.admin != nil {
690 record.Admin = *e.admin
691 }
692
[676]693 e.done <- u.updateUser(context.TODO(), &record)
[625]694
695 // If the password was updated, kill all downstream connections to
696 // force them to re-authenticate with the new credentials.
697 if e.password != nil {
698 u.forEachDownstream(func(dc *downstreamConn) {
699 dc.Close()
700 })
701 }
[376]702 case eventStop:
703 u.forEachDownstream(func(dc *downstreamConn) {
704 dc.Close()
705 })
706 for _, n := range u.networks {
707 n.stop()
[489]708
709 n.delivered.ForEachClient(func(clientName string) {
710 n.storeClientDeliveryReceipts(clientName)
711 })
[376]712 }
713 return
[165]714 default:
[494]715 panic(fmt.Sprintf("received unknown event type: %T", e))
[103]716 }
717 }
[101]718}
719
[313]720func (u *user) handleUpstreamDisconnected(uc *upstreamConn) {
721 uc.network.conn = nil
722
[682]723 uc.endPendingCommands()
[313]724
[478]725 for _, entry := range uc.channels.innerMap {
726 uch := entry.value.(*upstreamChannel)
[435]727 uch.updateAutoDetach(0)
728 }
729
[532]730 netIDStr := fmt.Sprintf("%v", uc.network.ID)
[313]731 uc.forEachDownstream(func(dc *downstreamConn) {
732 dc.updateSupportedCaps()
[543]733 })
[583]734
735 // If the network has been removed, don't send a state change notification
736 found := false
737 for _, net := range u.networks {
738 if net == uc.network {
739 found = true
740 break
741 }
742 }
743 if !found {
744 return
745 }
746
[543]747 u.forEachDownstream(func(dc *downstreamConn) {
[535]748 if dc.caps["soju.im/bouncer-networks-notify"] {
[532]749 dc.SendMessage(&irc.Message{
750 Prefix: dc.srv.prefix(),
751 Command: "BOUNCER",
[544]752 Params: []string{"NETWORK", netIDStr, "state=disconnected"},
[532]753 })
754 }
[313]755 })
756
757 if uc.network.lastError == nil {
758 uc.forEachDownstream(func(dc *downstreamConn) {
[532]759 if !dc.caps["soju.im/bouncer-networks"] {
760 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s", uc.network.GetName()))
761 }
[313]762 })
763 }
764}
765
766func (u *user) addNetwork(network *network) {
767 u.networks = append(u.networks, network)
768 go network.run()
769}
770
771func (u *user) removeNetwork(network *network) {
772 network.stop()
773
774 u.forEachDownstream(func(dc *downstreamConn) {
775 if dc.network != nil && dc.network == network {
776 dc.Close()
777 }
778 })
779
780 for i, net := range u.networks {
781 if net == network {
782 u.networks = append(u.networks[:i], u.networks[i+1:]...)
783 return
784 }
785 }
786
787 panic("tried to remove a non-existing network")
788}
789
[500]790func (u *user) checkNetwork(record *Network) error {
[731]791 url, err := record.URL()
792 if err != nil {
793 return err
794 }
795 if url.User != nil {
796 return fmt.Errorf("%v:// URL must not have username and password information", url.Scheme)
797 }
798 if url.RawQuery != "" {
799 return fmt.Errorf("%v:// URL must not have query values", url.Scheme)
800 }
801 if url.Fragment != "" {
802 return fmt.Errorf("%v:// URL must not have a fragment", url.Scheme)
803 }
804 switch url.Scheme {
805 case "ircs", "irc+insecure":
806 if url.Host == "" {
807 return fmt.Errorf("%v:// URL must have a host", url.Scheme)
808 }
809 if url.Path != "" {
810 return fmt.Errorf("%v:// URL must not have a path", url.Scheme)
811 }
812 case "irc+unix", "unix":
813 if url.Host != "" {
814 return fmt.Errorf("%v:// URL must not have a host", url.Scheme)
815 }
816 if url.Path == "" {
817 return fmt.Errorf("%v:// URL must have a path", url.Scheme)
818 }
819 default:
820 return fmt.Errorf("unknown URL scheme %q", url.Scheme)
821 }
822
[500]823 for _, net := range u.networks {
824 if net.GetName() == record.GetName() && net.ID != record.ID {
825 return fmt.Errorf("a network with the name %q already exists", record.GetName())
826 }
827 }
[731]828
[500]829 return nil
830}
831
[676]832func (u *user) createNetwork(ctx context.Context, record *Network) (*network, error) {
[313]833 if record.ID != 0 {
[144]834 panic("tried creating an already-existing network")
835 }
836
[500]837 if err := u.checkNetwork(record); err != nil {
838 return nil, err
839 }
840
[691]841 if max := u.srv.Config().MaxUserNetworks; max >= 0 && len(u.networks) >= max {
[612]842 return nil, fmt.Errorf("maximum number of networks reached")
843 }
844
[313]845 network := newNetwork(u, record, nil)
[676]846 err := u.srv.db.StoreNetwork(ctx, u.ID, &network.Network)
[101]847 if err != nil {
848 return nil, err
849 }
[144]850
[313]851 u.addNetwork(network)
[144]852
[532]853 idStr := fmt.Sprintf("%v", network.ID)
[535]854 attrs := getNetworkAttrs(network)
[532]855 u.forEachDownstream(func(dc *downstreamConn) {
[535]856 if dc.caps["soju.im/bouncer-networks-notify"] {
[532]857 dc.SendMessage(&irc.Message{
858 Prefix: dc.srv.prefix(),
859 Command: "BOUNCER",
[535]860 Params: []string{"NETWORK", idStr, attrs.String()},
[532]861 })
862 }
863 })
864
[101]865 return network, nil
866}
[202]867
[676]868func (u *user) updateNetwork(ctx context.Context, record *Network) (*network, error) {
[313]869 if record.ID == 0 {
870 panic("tried updating a new network")
871 }
[202]872
[568]873 // If the realname is reset to the default, just wipe the per-network
874 // setting
875 if record.Realname == u.Realname {
876 record.Realname = ""
877 }
878
[500]879 if err := u.checkNetwork(record); err != nil {
880 return nil, err
881 }
882
[313]883 network := u.getNetworkByID(record.ID)
884 if network == nil {
885 panic("tried updating a non-existing network")
886 }
887
[676]888 if err := u.srv.db.StoreNetwork(ctx, u.ID, record); err != nil {
[313]889 return nil, err
890 }
891
892 // Most network changes require us to re-connect to the upstream server
893
[478]894 channels := make([]Channel, 0, network.channels.Len())
895 for _, entry := range network.channels.innerMap {
896 ch := entry.value.(*Channel)
[313]897 channels = append(channels, *ch)
898 }
899
900 updatedNetwork := newNetwork(u, record, channels)
901
902 // If we're currently connected, disconnect and perform the necessary
903 // bookkeeping
904 if network.conn != nil {
905 network.stop()
906 // Note: this will set network.conn to nil
907 u.handleUpstreamDisconnected(network.conn)
908 }
909
910 // Patch downstream connections to use our fresh updated network
911 u.forEachDownstream(func(dc *downstreamConn) {
912 if dc.network != nil && dc.network == network {
913 dc.network = updatedNetwork
[202]914 }
[313]915 })
[202]916
[313]917 // We need to remove the network after patching downstream connections,
918 // otherwise they'll get closed
919 u.removeNetwork(network)
[202]920
[644]921 // The filesystem message store needs to be notified whenever the network
922 // is renamed
923 fsMsgStore, isFS := u.msgStore.(*fsMessageStore)
924 if isFS && updatedNetwork.GetName() != network.GetName() {
[666]925 if err := fsMsgStore.RenameNetwork(&network.Network, &updatedNetwork.Network); err != nil {
[644]926 network.logger.Printf("failed to update FS message store network name to %q: %v", updatedNetwork.GetName(), err)
927 }
928 }
929
[313]930 // This will re-connect to the upstream server
931 u.addNetwork(updatedNetwork)
932
[535]933 // TODO: only broadcast attributes that have changed
934 idStr := fmt.Sprintf("%v", updatedNetwork.ID)
935 attrs := getNetworkAttrs(updatedNetwork)
936 u.forEachDownstream(func(dc *downstreamConn) {
937 if dc.caps["soju.im/bouncer-networks-notify"] {
938 dc.SendMessage(&irc.Message{
939 Prefix: dc.srv.prefix(),
940 Command: "BOUNCER",
941 Params: []string{"NETWORK", idStr, attrs.String()},
942 })
943 }
944 })
[532]945
[313]946 return updatedNetwork, nil
947}
948
[676]949func (u *user) deleteNetwork(ctx context.Context, id int64) error {
[313]950 network := u.getNetworkByID(id)
951 if network == nil {
952 panic("tried deleting a non-existing network")
[202]953 }
954
[676]955 if err := u.srv.db.DeleteNetwork(ctx, network.ID); err != nil {
[313]956 return err
957 }
958
959 u.removeNetwork(network)
[532]960
961 idStr := fmt.Sprintf("%v", network.ID)
962 u.forEachDownstream(func(dc *downstreamConn) {
[535]963 if dc.caps["soju.im/bouncer-networks-notify"] {
[532]964 dc.SendMessage(&irc.Message{
965 Prefix: dc.srv.prefix(),
966 Command: "BOUNCER",
967 Params: []string{"NETWORK", idStr, "*"},
968 })
969 }
970 })
971
[313]972 return nil
[202]973}
[252]974
[676]975func (u *user) updateUser(ctx context.Context, record *User) error {
[572]976 if u.ID != record.ID {
977 panic("ID mismatch when updating user")
978 }
[376]979
[572]980 realnameUpdated := u.Realname != record.Realname
[676]981 if err := u.srv.db.StoreUser(ctx, record); err != nil {
[568]982 return fmt.Errorf("failed to update user %q: %v", u.Username, err)
983 }
[572]984 u.User = *record
[568]985
[572]986 if realnameUpdated {
987 // Re-connect to networks which use the default realname
988 var needUpdate []Network
989 u.forEachNetwork(func(net *network) {
990 if net.Realname == "" {
991 needUpdate = append(needUpdate, net.Network)
992 }
993 })
[568]994
[572]995 var netErr error
996 for _, net := range needUpdate {
[676]997 if _, err := u.updateNetwork(ctx, &net); err != nil {
[572]998 netErr = err
999 }
[568]1000 }
[572]1001 if netErr != nil {
1002 return netErr
1003 }
[568]1004 }
1005
[572]1006 return nil
[568]1007}
1008
[376]1009func (u *user) stop() {
1010 u.events <- eventStop{}
[377]1011 <-u.done
[376]1012}
[489]1013
1014func (u *user) hasPersistentMsgStore() bool {
1015 if u.msgStore == nil {
1016 return false
1017 }
1018 _, isMem := u.msgStore.(*memoryMessageStore)
1019 return !isMem
1020}
[705]1021
1022// localAddrForHost returns the local address to use when connecting to host.
1023// A nil address is returned when the OS should automatically pick one.
[732]1024func (u *user) localTCPAddrForHost(ctx context.Context, host string) (*net.TCPAddr, error) {
[705]1025 upstreamUserIPs := u.srv.Config().UpstreamUserIPs
1026 if len(upstreamUserIPs) == 0 {
1027 return nil, nil
1028 }
1029
[732]1030 ips, err := net.DefaultResolver.LookupIP(ctx, "ip", host)
[705]1031 if err != nil {
1032 return nil, err
1033 }
1034
1035 wantIPv6 := false
1036 for _, ip := range ips {
1037 if ip.To4() == nil {
1038 wantIPv6 = true
1039 break
1040 }
1041 }
1042
1043 var ipNet *net.IPNet
1044 for _, in := range upstreamUserIPs {
1045 if wantIPv6 == (in.IP.To4() == nil) {
1046 ipNet = in
1047 break
1048 }
1049 }
1050 if ipNet == nil {
1051 return nil, nil
1052 }
1053
1054 var ipInt big.Int
1055 ipInt.SetBytes(ipNet.IP)
1056 ipInt.Add(&ipInt, big.NewInt(u.ID+1))
1057 ip := net.IP(ipInt.Bytes())
1058 if !ipNet.Contains(ip) {
1059 return nil, fmt.Errorf("IP network %v too small", ipNet)
1060 }
1061
1062 return &net.TCPAddr{IP: ip}, nil
1063}
Note: See TracBrowser for help on using the repository browser.