source: code/trunk/user.go@ 734

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

Add soju_upstream_connect_errors_total metric

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