source: code/trunk/user.go@ 821

Last change on this file since 821 was 804, checked in by koizumi.aoi, 2 years ago

Drunk as I like

Signed-off-by: Aoi K <koizumi.aoi@…>

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