source: code/trunk/user.go@ 400

Last change on this file since 400 was 399, checked in by contact, 5 years ago

Improve registration error messages

  • Don't print the raw IRC message, since we already show the original error message
  • Avoid double-printing "registration failed"
File size: 13.7 KB
Line 
1package soju
2
3import (
4 "crypto/sha256"
5 "encoding/binary"
6 "encoding/hex"
7 "fmt"
8 "time"
9
10 "gopkg.in/irc.v3"
11)
12
13type event interface{}
14
15type eventUpstreamMessage struct {
16 msg *irc.Message
17 uc *upstreamConn
18}
19
20type eventUpstreamConnectionError struct {
21 net *network
22 err error
23}
24
25type eventUpstreamConnected struct {
26 uc *upstreamConn
27}
28
29type eventUpstreamDisconnected struct {
30 uc *upstreamConn
31}
32
33type eventUpstreamError struct {
34 uc *upstreamConn
35 err error
36}
37
38type eventDownstreamMessage struct {
39 msg *irc.Message
40 dc *downstreamConn
41}
42
43type eventDownstreamConnected struct {
44 dc *downstreamConn
45}
46
47type eventDownstreamDisconnected struct {
48 dc *downstreamConn
49}
50
51type eventStop struct{}
52
53type networkHistory struct {
54 offlineClients map[string]uint64 // indexed by client name
55 ring *Ring // can be nil if there are no offline clients
56}
57
58type network struct {
59 Network
60 user *user
61 stopped chan struct{}
62
63 conn *upstreamConn
64 channels map[string]*Channel
65 history map[string]*networkHistory // indexed by entity
66 offlineClients map[string]struct{} // indexed by client name
67 lastError error
68}
69
70func newNetwork(user *user, record *Network, channels []Channel) *network {
71 m := make(map[string]*Channel, len(channels))
72 for _, ch := range channels {
73 ch := ch
74 m[ch.Name] = &ch
75 }
76
77 return &network{
78 Network: *record,
79 user: user,
80 stopped: make(chan struct{}),
81 channels: m,
82 history: make(map[string]*networkHistory),
83 offlineClients: make(map[string]struct{}),
84 }
85}
86
87func (net *network) forEachDownstream(f func(*downstreamConn)) {
88 net.user.forEachDownstream(func(dc *downstreamConn) {
89 if dc.network != nil && dc.network != net {
90 return
91 }
92 f(dc)
93 })
94}
95
96func (net *network) isStopped() bool {
97 select {
98 case <-net.stopped:
99 return true
100 default:
101 return false
102 }
103}
104
105func userIdent(u *User) string {
106 // The ident is a string we will send to upstream servers in clear-text.
107 // For privacy reasons, make sure it doesn't expose any meaningful user
108 // metadata. We just use the base64-encoded hashed ID, so that people don't
109 // start relying on the string being an integer or following a pattern.
110 var b [64]byte
111 binary.LittleEndian.PutUint64(b[:], uint64(u.ID))
112 h := sha256.Sum256(b[:])
113 return hex.EncodeToString(h[:16])
114}
115
116func (net *network) run() {
117 var lastTry time.Time
118 for {
119 if net.isStopped() {
120 return
121 }
122
123 if dur := time.Now().Sub(lastTry); dur < retryConnectDelay {
124 delay := retryConnectDelay - dur
125 net.user.srv.Logger.Printf("waiting %v before trying to reconnect to %q", delay.Truncate(time.Second), net.Addr)
126 time.Sleep(delay)
127 }
128 lastTry = time.Now()
129
130 uc, err := connectToUpstream(net)
131 if err != nil {
132 net.user.srv.Logger.Printf("failed to connect to upstream server %q: %v", net.Addr, err)
133 net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("failed to connect: %v", err)}
134 continue
135 }
136
137 if net.user.srv.Identd != nil {
138 net.user.srv.Identd.Store(uc.RemoteAddr().String(), uc.LocalAddr().String(), userIdent(&net.user.User))
139 }
140
141 uc.register()
142 if err := uc.runUntilRegistered(); err != nil {
143 text := err.Error()
144 if regErr, ok := err.(registrationError); ok {
145 text = string(regErr)
146 }
147 uc.logger.Printf("failed to register: %v", text)
148 net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("failed to register: %v", text)}
149 uc.Close()
150 continue
151 }
152
153 // TODO: this is racy with net.stopped. If the network is stopped
154 // before the user goroutine receives eventUpstreamConnected, the
155 // connection won't be closed.
156 net.user.events <- eventUpstreamConnected{uc}
157 if err := uc.readMessages(net.user.events); err != nil {
158 uc.logger.Printf("failed to handle messages: %v", err)
159 net.user.events <- eventUpstreamError{uc, fmt.Errorf("failed to handle messages: %v", err)}
160 }
161 uc.Close()
162 net.user.events <- eventUpstreamDisconnected{uc}
163
164 if net.user.srv.Identd != nil {
165 net.user.srv.Identd.Delete(uc.RemoteAddr().String(), uc.LocalAddr().String())
166 }
167 }
168}
169
170func (net *network) stop() {
171 if !net.isStopped() {
172 close(net.stopped)
173 }
174
175 if net.conn != nil {
176 net.conn.Close()
177 }
178}
179
180func (net *network) createUpdateChannel(ch *Channel) error {
181 if current, ok := net.channels[ch.Name]; ok {
182 ch.ID = current.ID // update channel if it already exists
183 }
184 if err := net.user.srv.db.StoreChannel(net.ID, ch); err != nil {
185 return err
186 }
187 prev := net.channels[ch.Name]
188 net.channels[ch.Name] = ch
189
190 if prev != nil && prev.Detached != ch.Detached {
191 history := net.history[ch.Name]
192 if ch.Detached {
193 net.user.srv.Logger.Printf("network %q: detaching channel %q", net.GetName(), ch.Name)
194 net.forEachDownstream(func(dc *downstreamConn) {
195 net.offlineClients[dc.clientName] = struct{}{}
196 if history != nil {
197 history.offlineClients[dc.clientName] = history.ring.Cur()
198 }
199
200 dc.SendMessage(&irc.Message{
201 Prefix: dc.prefix(),
202 Command: "PART",
203 Params: []string{dc.marshalEntity(net, ch.Name), "Detach"},
204 })
205 })
206 } else {
207 net.user.srv.Logger.Printf("network %q: attaching channel %q", net.GetName(), ch.Name)
208
209 var uch *upstreamChannel
210 if net.conn != nil {
211 uch = net.conn.channels[ch.Name]
212 }
213
214 net.forEachDownstream(func(dc *downstreamConn) {
215 dc.SendMessage(&irc.Message{
216 Prefix: dc.prefix(),
217 Command: "JOIN",
218 Params: []string{dc.marshalEntity(net, ch.Name)},
219 })
220
221 if uch != nil {
222 forwardChannel(dc, uch)
223 }
224
225 if history != nil {
226 dc.sendNetworkHistory(net)
227 }
228 })
229 }
230 }
231
232 return nil
233}
234
235func (net *network) deleteChannel(name string) error {
236 if err := net.user.srv.db.DeleteChannel(net.ID, name); err != nil {
237 return err
238 }
239 delete(net.channels, name)
240 return nil
241}
242
243type user struct {
244 User
245 srv *Server
246
247 events chan event
248 done chan struct{}
249
250 networks []*network
251 downstreamConns []*downstreamConn
252
253 // LIST commands in progress
254 pendingLISTs []pendingLIST
255}
256
257type pendingLIST struct {
258 downstreamID uint64
259 // list of per-upstream LIST commands not yet sent or completed
260 pendingCommands map[int64]*irc.Message
261}
262
263func newUser(srv *Server, record *User) *user {
264 return &user{
265 User: *record,
266 srv: srv,
267 events: make(chan event, 64),
268 done: make(chan struct{}),
269 }
270}
271
272func (u *user) forEachNetwork(f func(*network)) {
273 for _, network := range u.networks {
274 f(network)
275 }
276}
277
278func (u *user) forEachUpstream(f func(uc *upstreamConn)) {
279 for _, network := range u.networks {
280 if network.conn == nil {
281 continue
282 }
283 f(network.conn)
284 }
285}
286
287func (u *user) forEachDownstream(f func(dc *downstreamConn)) {
288 for _, dc := range u.downstreamConns {
289 f(dc)
290 }
291}
292
293func (u *user) getNetwork(name string) *network {
294 for _, network := range u.networks {
295 if network.Addr == name {
296 return network
297 }
298 if network.Name != "" && network.Name == name {
299 return network
300 }
301 }
302 return nil
303}
304
305func (u *user) getNetworkByID(id int64) *network {
306 for _, net := range u.networks {
307 if net.ID == id {
308 return net
309 }
310 }
311 return nil
312}
313
314func (u *user) run() {
315 defer close(u.done)
316
317 networks, err := u.srv.db.ListNetworks(u.Username)
318 if err != nil {
319 u.srv.Logger.Printf("failed to list networks for user %q: %v", u.Username, err)
320 return
321 }
322
323 for _, record := range networks {
324 record := record
325 channels, err := u.srv.db.ListChannels(record.ID)
326 if err != nil {
327 u.srv.Logger.Printf("failed to list channels for user %q, network %q: %v", u.Username, record.GetName(), err)
328 continue
329 }
330
331 network := newNetwork(u, &record, channels)
332 u.networks = append(u.networks, network)
333
334 go network.run()
335 }
336
337 for e := range u.events {
338 switch e := e.(type) {
339 case eventUpstreamConnected:
340 uc := e.uc
341
342 uc.network.conn = uc
343
344 uc.updateAway()
345
346 uc.forEachDownstream(func(dc *downstreamConn) {
347 dc.updateSupportedCaps()
348 sendServiceNOTICE(dc, fmt.Sprintf("connected to %s", uc.network.GetName()))
349
350 dc.updateNick()
351 })
352 uc.network.lastError = nil
353 case eventUpstreamDisconnected:
354 u.handleUpstreamDisconnected(e.uc)
355 case eventUpstreamConnectionError:
356 net := e.net
357
358 stopped := false
359 select {
360 case <-net.stopped:
361 stopped = true
362 default:
363 }
364
365 if !stopped && (net.lastError == nil || net.lastError.Error() != e.err.Error()) {
366 net.forEachDownstream(func(dc *downstreamConn) {
367 sendServiceNOTICE(dc, fmt.Sprintf("failed connecting/registering to %s: %v", net.GetName(), e.err))
368 })
369 }
370 net.lastError = e.err
371 case eventUpstreamError:
372 uc := e.uc
373
374 uc.forEachDownstream(func(dc *downstreamConn) {
375 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s: %v", uc.network.GetName(), e.err))
376 })
377 uc.network.lastError = e.err
378 case eventUpstreamMessage:
379 msg, uc := e.msg, e.uc
380 if uc.isClosed() {
381 uc.logger.Printf("ignoring message on closed connection: %v", msg)
382 break
383 }
384 if err := uc.handleMessage(msg); err != nil {
385 uc.logger.Printf("failed to handle message %q: %v", msg, err)
386 }
387 case eventDownstreamConnected:
388 dc := e.dc
389
390 if err := dc.welcome(); err != nil {
391 dc.logger.Printf("failed to handle new registered connection: %v", err)
392 break
393 }
394
395 u.downstreamConns = append(u.downstreamConns, dc)
396
397 u.forEachUpstream(func(uc *upstreamConn) {
398 uc.updateAway()
399 })
400
401 dc.updateSupportedCaps()
402 case eventDownstreamDisconnected:
403 dc := e.dc
404
405 for i := range u.downstreamConns {
406 if u.downstreamConns[i] == dc {
407 u.downstreamConns = append(u.downstreamConns[:i], u.downstreamConns[i+1:]...)
408 break
409 }
410 }
411
412 // Save history if we're the last client with this name
413 skipHistory := make(map[*network]bool)
414 u.forEachDownstream(func(conn *downstreamConn) {
415 if dc.clientName == conn.clientName {
416 skipHistory[conn.network] = true
417 }
418 })
419
420 dc.forEachNetwork(func(net *network) {
421 if skipHistory[net] || skipHistory[nil] {
422 return
423 }
424
425 net.offlineClients[dc.clientName] = struct{}{}
426 for target, history := range net.history {
427 if ch, ok := net.channels[target]; ok && ch.Detached {
428 continue
429 }
430 history.offlineClients[dc.clientName] = history.ring.Cur()
431 }
432 })
433
434 u.forEachUpstream(func(uc *upstreamConn) {
435 uc.updateAway()
436 })
437 case eventDownstreamMessage:
438 msg, dc := e.msg, e.dc
439 if dc.isClosed() {
440 dc.logger.Printf("ignoring message on closed connection: %v", msg)
441 break
442 }
443 err := dc.handleMessage(msg)
444 if ircErr, ok := err.(ircError); ok {
445 ircErr.Message.Prefix = dc.srv.prefix()
446 dc.SendMessage(ircErr.Message)
447 } else if err != nil {
448 dc.logger.Printf("failed to handle message %q: %v", msg, err)
449 dc.Close()
450 }
451 case eventStop:
452 u.forEachDownstream(func(dc *downstreamConn) {
453 dc.Close()
454 })
455 for _, n := range u.networks {
456 n.stop()
457 }
458 return
459 default:
460 u.srv.Logger.Printf("received unknown event type: %T", e)
461 }
462 }
463}
464
465func (u *user) handleUpstreamDisconnected(uc *upstreamConn) {
466 uc.network.conn = nil
467
468 for _, ml := range uc.messageLoggers {
469 if err := ml.Close(); err != nil {
470 uc.logger.Printf("failed to close message logger: %v", err)
471 }
472 }
473
474 uc.endPendingLISTs(true)
475
476 uc.forEachDownstream(func(dc *downstreamConn) {
477 dc.updateSupportedCaps()
478 })
479
480 if uc.network.lastError == nil {
481 uc.forEachDownstream(func(dc *downstreamConn) {
482 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s", uc.network.GetName()))
483 })
484 }
485}
486
487func (u *user) addNetwork(network *network) {
488 u.networks = append(u.networks, network)
489 go network.run()
490}
491
492func (u *user) removeNetwork(network *network) {
493 network.stop()
494
495 u.forEachDownstream(func(dc *downstreamConn) {
496 if dc.network != nil && dc.network == network {
497 dc.Close()
498 }
499 })
500
501 for i, net := range u.networks {
502 if net == network {
503 u.networks = append(u.networks[:i], u.networks[i+1:]...)
504 return
505 }
506 }
507
508 panic("tried to remove a non-existing network")
509}
510
511func (u *user) createNetwork(record *Network) (*network, error) {
512 if record.ID != 0 {
513 panic("tried creating an already-existing network")
514 }
515
516 network := newNetwork(u, record, nil)
517 err := u.srv.db.StoreNetwork(u.Username, &network.Network)
518 if err != nil {
519 return nil, err
520 }
521
522 u.addNetwork(network)
523
524 return network, nil
525}
526
527func (u *user) updateNetwork(record *Network) (*network, error) {
528 if record.ID == 0 {
529 panic("tried updating a new network")
530 }
531
532 network := u.getNetworkByID(record.ID)
533 if network == nil {
534 panic("tried updating a non-existing network")
535 }
536
537 if err := u.srv.db.StoreNetwork(u.Username, record); err != nil {
538 return nil, err
539 }
540
541 // Most network changes require us to re-connect to the upstream server
542
543 channels := make([]Channel, 0, len(network.channels))
544 for _, ch := range network.channels {
545 channels = append(channels, *ch)
546 }
547
548 updatedNetwork := newNetwork(u, record, channels)
549
550 // If we're currently connected, disconnect and perform the necessary
551 // bookkeeping
552 if network.conn != nil {
553 network.stop()
554 // Note: this will set network.conn to nil
555 u.handleUpstreamDisconnected(network.conn)
556 }
557
558 // Patch downstream connections to use our fresh updated network
559 u.forEachDownstream(func(dc *downstreamConn) {
560 if dc.network != nil && dc.network == network {
561 dc.network = updatedNetwork
562 }
563 })
564
565 // We need to remove the network after patching downstream connections,
566 // otherwise they'll get closed
567 u.removeNetwork(network)
568
569 // This will re-connect to the upstream server
570 u.addNetwork(updatedNetwork)
571
572 return updatedNetwork, nil
573}
574
575func (u *user) deleteNetwork(id int64) error {
576 network := u.getNetworkByID(id)
577 if network == nil {
578 panic("tried deleting a non-existing network")
579 }
580
581 if err := u.srv.db.DeleteNetwork(network.ID); err != nil {
582 return err
583 }
584
585 u.removeNetwork(network)
586 return nil
587}
588
589func (u *user) updatePassword(hashed string) error {
590 u.User.Password = hashed
591 return u.srv.db.StoreUser(&u.User)
592}
593
594func (u *user) stop() {
595 u.events <- eventStop{}
596 <-u.done
597}
Note: See TracBrowser for help on using the repository browser.