source: code/trunk/user.go@ 733

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

Fix upstream gauge metric

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