source: code/trunk/user.go@ 752

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

Use more descriptive errors when aborting pending commands

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