source: code/trunk/upstream.go@ 179

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

Introduce eventUpstreamDisconnected

This allows us to perform cleanup actions in the user goroutine. This
removes the need for pendingLISTsLock.
File size: 35.9 KB
Line 
1package soju
2
3import (
4 "crypto/tls"
5 "encoding/base64"
6 "errors"
7 "fmt"
8 "io"
9 "net"
10 "os"
11 "path/filepath"
12 "strconv"
13 "strings"
14 "time"
15
16 "github.com/emersion/go-sasl"
17 "gopkg.in/irc.v3"
18)
19
20type upstreamChannel struct {
21 Name string
22 conn *upstreamConn
23 Topic string
24 TopicWho string
25 TopicTime time.Time
26 Status channelStatus
27 modes channelModes
28 creationTime string
29 Members map[string]*membership
30 complete bool
31}
32
33type upstreamConn struct {
34 network *network
35 logger Logger
36 net net.Conn
37 irc *irc.Conn
38 srv *Server
39 user *user
40 outgoing chan<- *irc.Message
41 closed chan struct{}
42
43 serverName string
44 availableUserModes string
45 availableChannelModes map[byte]channelModeType
46 availableChannelTypes string
47 availableMemberships []membership
48
49 registered bool
50 nick string
51 username string
52 realname string
53 modes userModes
54 channels map[string]*upstreamChannel
55 caps map[string]string
56 batches map[string]batch
57
58 tagsSupported bool
59 labelsSupported bool
60 nextLabelID uint64
61
62 saslClient sasl.Client
63 saslStarted bool
64
65 // set of LIST commands in progress, per downstream
66 pendingLISTDownstreamSet map[uint64]struct{}
67
68 logs map[string]entityLog
69}
70
71type entityLog struct {
72 name string
73 file *os.File
74}
75
76func connectToUpstream(network *network) (*upstreamConn, error) {
77 logger := &prefixLogger{network.user.srv.Logger, fmt.Sprintf("upstream %q: ", network.Addr)}
78
79 addr := network.Addr
80 if !strings.ContainsRune(addr, ':') {
81 addr = addr + ":6697"
82 }
83
84 logger.Printf("connecting to TLS server at address %q", addr)
85 netConn, err := tls.Dial("tcp", addr, nil)
86 if err != nil {
87 return nil, fmt.Errorf("failed to dial %q: %v", addr, err)
88 }
89
90 setKeepAlive(netConn)
91
92 outgoing := make(chan *irc.Message, 64)
93 uc := &upstreamConn{
94 network: network,
95 logger: logger,
96 net: netConn,
97 irc: irc.NewConn(netConn),
98 srv: network.user.srv,
99 user: network.user,
100 outgoing: outgoing,
101 channels: make(map[string]*upstreamChannel),
102 caps: make(map[string]string),
103 batches: make(map[string]batch),
104 availableChannelTypes: stdChannelTypes,
105 availableChannelModes: stdChannelModes,
106 availableMemberships: stdMemberships,
107 pendingLISTDownstreamSet: make(map[uint64]struct{}),
108 logs: make(map[string]entityLog),
109 }
110
111 go func() {
112 for {
113 var closed bool
114 select {
115 case msg := <-outgoing:
116 if uc.srv.Debug {
117 uc.logger.Printf("sent: %v", msg)
118 }
119 if err := uc.irc.WriteMessage(msg); err != nil {
120 uc.logger.Printf("failed to write message: %v", err)
121 }
122 case <-uc.closed:
123 closed = true
124 }
125 if closed {
126 break
127 }
128 }
129 if err := uc.net.Close(); err != nil {
130 uc.logger.Printf("failed to close connection: %v", err)
131 } else {
132 uc.logger.Printf("connection closed")
133 }
134 }()
135
136 return uc, nil
137}
138
139func (uc *upstreamConn) isClosed() bool {
140 select {
141 case <-uc.closed:
142 return true
143 default:
144 return false
145 }
146}
147
148func (uc *upstreamConn) Close() error {
149 if uc.isClosed() {
150 return fmt.Errorf("upstream connection already closed")
151 }
152 close(uc.closed)
153 return nil
154}
155
156func (uc *upstreamConn) forEachDownstream(f func(*downstreamConn)) {
157 uc.user.forEachDownstream(func(dc *downstreamConn) {
158 if dc.network != nil && dc.network != uc.network {
159 return
160 }
161 f(dc)
162 })
163}
164
165func (uc *upstreamConn) forEachDownstreamByID(id uint64, f func(*downstreamConn)) {
166 uc.forEachDownstream(func(dc *downstreamConn) {
167 if id != 0 && id != dc.id {
168 return
169 }
170 f(dc)
171 })
172}
173
174func (uc *upstreamConn) getChannel(name string) (*upstreamChannel, error) {
175 ch, ok := uc.channels[name]
176 if !ok {
177 return nil, fmt.Errorf("unknown channel %q", name)
178 }
179 return ch, nil
180}
181
182func (uc *upstreamConn) isChannel(entity string) bool {
183 if i := strings.IndexByte(uc.availableChannelTypes, entity[0]); i >= 0 {
184 return true
185 }
186 return false
187}
188
189func (uc *upstreamConn) getPendingList() *pendingLIST {
190 for _, pl := range uc.user.pendingLISTs {
191 if _, ok := pl.pendingCommands[uc.network.ID]; !ok {
192 continue
193 }
194 return &pl
195 }
196 return nil
197}
198
199func (uc *upstreamConn) endPendingLists(all bool) (found bool) {
200 found = false
201 for i := 0; i < len(uc.user.pendingLISTs); i++ {
202 pl := uc.user.pendingLISTs[i]
203 if _, ok := pl.pendingCommands[uc.network.ID]; !ok {
204 continue
205 }
206 delete(pl.pendingCommands, uc.network.ID)
207 if len(pl.pendingCommands) == 0 {
208 uc.user.pendingLISTs = append(uc.user.pendingLISTs[:i], uc.user.pendingLISTs[i+1:]...)
209 i--
210 uc.forEachDownstreamByID(pl.downstreamID, func(dc *downstreamConn) {
211 dc.SendMessage(&irc.Message{
212 Prefix: dc.srv.prefix(),
213 Command: irc.RPL_LISTEND,
214 Params: []string{dc.nick, "End of /LIST"},
215 })
216 })
217 }
218 found = true
219 if !all {
220 delete(uc.pendingLISTDownstreamSet, pl.downstreamID)
221 uc.user.forEachUpstream(func(uc *upstreamConn) {
222 uc.trySendList(pl.downstreamID)
223 })
224 return
225 }
226 }
227 return
228}
229
230func (uc *upstreamConn) trySendList(downstreamID uint64) {
231 // must be called with a lock in uc.user.pendingLISTsLock
232
233 if _, ok := uc.pendingLISTDownstreamSet[downstreamID]; ok {
234 // a LIST command is already pending
235 // we will try again when that command is completed
236 return
237 }
238
239 for _, pl := range uc.user.pendingLISTs {
240 if pl.downstreamID != downstreamID {
241 continue
242 }
243 // this is the first pending LIST command list of the downstream
244 listCommand, ok := pl.pendingCommands[uc.network.ID]
245 if !ok {
246 // there is no command for this upstream in these LIST commands
247 // do not send anything
248 continue
249 }
250 // there is a command for this upstream in these LIST commands
251 // send it now
252
253 uc.SendMessageLabeled(downstreamID, listCommand)
254
255 uc.pendingLISTDownstreamSet[downstreamID] = struct{}{}
256 return
257 }
258}
259
260func (uc *upstreamConn) parseMembershipPrefix(s string) (membership *membership, nick string) {
261 for _, m := range uc.availableMemberships {
262 if m.Prefix == s[0] {
263 return &m, s[1:]
264 }
265 }
266 return nil, s
267}
268
269func (uc *upstreamConn) handleMessage(msg *irc.Message) error {
270 var label string
271 if l, ok := msg.GetTag("label"); ok {
272 label = l
273 }
274
275 var msgBatch *batch
276 if batchName, ok := msg.GetTag("batch"); ok {
277 b, ok := uc.batches[batchName]
278 if !ok {
279 return fmt.Errorf("unexpected batch reference: batch was not defined: %q", batchName)
280 }
281 msgBatch = &b
282 if label == "" {
283 label = msgBatch.Label
284 }
285 }
286
287 var downstreamID uint64 = 0
288 if label != "" {
289 var labelOffset uint64
290 n, err := fmt.Sscanf(label, "sd-%d-%d", &downstreamID, &labelOffset)
291 if err == nil && n < 2 {
292 err = errors.New("not enough arguments")
293 }
294 if err != nil {
295 return fmt.Errorf("unexpected message label: invalid downstream reference for label %q: %v", label, err)
296 }
297 }
298
299 switch msg.Command {
300 case "PING":
301 uc.SendMessage(&irc.Message{
302 Command: "PONG",
303 Params: msg.Params,
304 })
305 return nil
306 case "NOTICE":
307 uc.logger.Print(msg)
308
309 if msg.Prefix.User == "" && msg.Prefix.Host == "" { // server message
310 uc.forEachDownstream(func(dc *downstreamConn) {
311 dc.SendMessage(msg)
312 })
313 } else { // regular user NOTICE
314 var nick, text string
315 if err := parseMessageParams(msg, &nick, &text); err != nil {
316 return err
317 }
318
319 target := nick
320 if nick == uc.nick {
321 target = msg.Prefix.Name
322 }
323 uc.AppendLog(target, "<%s> %s", msg.Prefix.Name, text)
324
325 uc.forEachDownstream(func(dc *downstreamConn) {
326 dc.SendMessage(&irc.Message{
327 Prefix: dc.marshalUserPrefix(uc, msg.Prefix),
328 Command: "NOTICE",
329 Params: []string{dc.marshalEntity(uc, nick), text},
330 })
331 })
332 }
333 case "CAP":
334 var subCmd string
335 if err := parseMessageParams(msg, nil, &subCmd); err != nil {
336 return err
337 }
338 subCmd = strings.ToUpper(subCmd)
339 subParams := msg.Params[2:]
340 switch subCmd {
341 case "LS":
342 if len(subParams) < 1 {
343 return newNeedMoreParamsError(msg.Command)
344 }
345 caps := strings.Fields(subParams[len(subParams)-1])
346 more := len(subParams) >= 2 && msg.Params[len(subParams)-2] == "*"
347
348 for _, s := range caps {
349 kv := strings.SplitN(s, "=", 2)
350 k := strings.ToLower(kv[0])
351 var v string
352 if len(kv) == 2 {
353 v = kv[1]
354 }
355 uc.caps[k] = v
356 }
357
358 if more {
359 break // wait to receive all capabilities
360 }
361
362 requestCaps := make([]string, 0, 16)
363 for _, c := range []string{"message-tags", "batch", "labeled-response"} {
364 if _, ok := uc.caps[c]; ok {
365 requestCaps = append(requestCaps, c)
366 }
367 }
368
369 if uc.requestSASL() {
370 requestCaps = append(requestCaps, "sasl")
371 }
372
373 if len(requestCaps) > 0 {
374 uc.SendMessage(&irc.Message{
375 Command: "CAP",
376 Params: []string{"REQ", strings.Join(requestCaps, " ")},
377 })
378 }
379
380 if uc.requestSASL() {
381 break // we'll send CAP END after authentication is completed
382 }
383
384 uc.SendMessage(&irc.Message{
385 Command: "CAP",
386 Params: []string{"END"},
387 })
388 case "ACK", "NAK":
389 if len(subParams) < 1 {
390 return newNeedMoreParamsError(msg.Command)
391 }
392 caps := strings.Fields(subParams[0])
393
394 for _, name := range caps {
395 if err := uc.handleCapAck(strings.ToLower(name), subCmd == "ACK"); err != nil {
396 return err
397 }
398 }
399
400 if uc.saslClient == nil {
401 uc.SendMessage(&irc.Message{
402 Command: "CAP",
403 Params: []string{"END"},
404 })
405 }
406 default:
407 uc.logger.Printf("unhandled message: %v", msg)
408 }
409 case "AUTHENTICATE":
410 if uc.saslClient == nil {
411 return fmt.Errorf("received unexpected AUTHENTICATE message")
412 }
413
414 // TODO: if a challenge is 400 bytes long, buffer it
415 var challengeStr string
416 if err := parseMessageParams(msg, &challengeStr); err != nil {
417 uc.SendMessage(&irc.Message{
418 Command: "AUTHENTICATE",
419 Params: []string{"*"},
420 })
421 return err
422 }
423
424 var challenge []byte
425 if challengeStr != "+" {
426 var err error
427 challenge, err = base64.StdEncoding.DecodeString(challengeStr)
428 if err != nil {
429 uc.SendMessage(&irc.Message{
430 Command: "AUTHENTICATE",
431 Params: []string{"*"},
432 })
433 return err
434 }
435 }
436
437 var resp []byte
438 var err error
439 if !uc.saslStarted {
440 _, resp, err = uc.saslClient.Start()
441 uc.saslStarted = true
442 } else {
443 resp, err = uc.saslClient.Next(challenge)
444 }
445 if err != nil {
446 uc.SendMessage(&irc.Message{
447 Command: "AUTHENTICATE",
448 Params: []string{"*"},
449 })
450 return err
451 }
452
453 // TODO: send response in multiple chunks if >= 400 bytes
454 var respStr = "+"
455 if resp != nil {
456 respStr = base64.StdEncoding.EncodeToString(resp)
457 }
458
459 uc.SendMessage(&irc.Message{
460 Command: "AUTHENTICATE",
461 Params: []string{respStr},
462 })
463 case irc.RPL_LOGGEDIN:
464 var account string
465 if err := parseMessageParams(msg, nil, nil, &account); err != nil {
466 return err
467 }
468 uc.logger.Printf("logged in with account %q", account)
469 case irc.RPL_LOGGEDOUT:
470 uc.logger.Printf("logged out")
471 case irc.ERR_NICKLOCKED, irc.RPL_SASLSUCCESS, irc.ERR_SASLFAIL, irc.ERR_SASLTOOLONG, irc.ERR_SASLABORTED:
472 var info string
473 if err := parseMessageParams(msg, nil, &info); err != nil {
474 return err
475 }
476 switch msg.Command {
477 case irc.ERR_NICKLOCKED:
478 uc.logger.Printf("invalid nick used with SASL authentication: %v", info)
479 case irc.ERR_SASLFAIL:
480 uc.logger.Printf("SASL authentication failed: %v", info)
481 case irc.ERR_SASLTOOLONG:
482 uc.logger.Printf("SASL message too long: %v", info)
483 }
484
485 uc.saslClient = nil
486 uc.saslStarted = false
487
488 uc.SendMessage(&irc.Message{
489 Command: "CAP",
490 Params: []string{"END"},
491 })
492 case irc.RPL_WELCOME:
493 uc.registered = true
494 uc.logger.Printf("connection registered")
495
496 channels, err := uc.srv.db.ListChannels(uc.network.ID)
497 if err != nil {
498 uc.logger.Printf("failed to list channels from database: %v", err)
499 break
500 }
501
502 for _, ch := range channels {
503 params := []string{ch.Name}
504 if ch.Key != "" {
505 params = append(params, ch.Key)
506 }
507 uc.SendMessage(&irc.Message{
508 Command: "JOIN",
509 Params: params,
510 })
511 }
512 case irc.RPL_MYINFO:
513 if err := parseMessageParams(msg, nil, &uc.serverName, nil, &uc.availableUserModes, nil); err != nil {
514 return err
515 }
516 case irc.RPL_ISUPPORT:
517 if err := parseMessageParams(msg, nil, nil); err != nil {
518 return err
519 }
520 for _, token := range msg.Params[1 : len(msg.Params)-1] {
521 negate := false
522 parameter := token
523 value := ""
524 if strings.HasPrefix(token, "-") {
525 negate = true
526 token = token[1:]
527 } else {
528 if i := strings.IndexByte(token, '='); i >= 0 {
529 parameter = token[:i]
530 value = token[i+1:]
531 }
532 }
533 if !negate {
534 switch parameter {
535 case "CHANMODES":
536 parts := strings.SplitN(value, ",", 5)
537 if len(parts) < 4 {
538 return fmt.Errorf("malformed ISUPPORT CHANMODES value: %v", value)
539 }
540 modes := make(map[byte]channelModeType)
541 for i, mt := range []channelModeType{modeTypeA, modeTypeB, modeTypeC, modeTypeD} {
542 for j := 0; j < len(parts[i]); j++ {
543 mode := parts[i][j]
544 modes[mode] = mt
545 }
546 }
547 uc.availableChannelModes = modes
548 case "CHANTYPES":
549 uc.availableChannelTypes = value
550 case "PREFIX":
551 if value == "" {
552 uc.availableMemberships = nil
553 } else {
554 if value[0] != '(' {
555 return fmt.Errorf("malformed ISUPPORT PREFIX value: %v", value)
556 }
557 sep := strings.IndexByte(value, ')')
558 if sep < 0 || len(value) != sep*2 {
559 return fmt.Errorf("malformed ISUPPORT PREFIX value: %v", value)
560 }
561 memberships := make([]membership, len(value)/2-1)
562 for i := range memberships {
563 memberships[i] = membership{
564 Mode: value[i+1],
565 Prefix: value[sep+i+1],
566 }
567 }
568 uc.availableMemberships = memberships
569 }
570 }
571 } else {
572 // TODO: handle ISUPPORT negations
573 }
574 }
575 case "BATCH":
576 var tag string
577 if err := parseMessageParams(msg, &tag); err != nil {
578 return err
579 }
580
581 if strings.HasPrefix(tag, "+") {
582 tag = tag[1:]
583 if _, ok := uc.batches[tag]; ok {
584 return fmt.Errorf("unexpected BATCH reference tag: batch was already defined: %q", tag)
585 }
586 var batchType string
587 if err := parseMessageParams(msg, nil, &batchType); err != nil {
588 return err
589 }
590 label := label
591 if label == "" && msgBatch != nil {
592 label = msgBatch.Label
593 }
594 uc.batches[tag] = batch{
595 Type: batchType,
596 Params: msg.Params[2:],
597 Outer: msgBatch,
598 Label: label,
599 }
600 } else if strings.HasPrefix(tag, "-") {
601 tag = tag[1:]
602 if _, ok := uc.batches[tag]; !ok {
603 return fmt.Errorf("unknown BATCH reference tag: %q", tag)
604 }
605 delete(uc.batches, tag)
606 } else {
607 return fmt.Errorf("unexpected BATCH reference tag: missing +/- prefix: %q", tag)
608 }
609 case "NICK":
610 if msg.Prefix == nil {
611 return fmt.Errorf("expected a prefix")
612 }
613
614 var newNick string
615 if err := parseMessageParams(msg, &newNick); err != nil {
616 return err
617 }
618
619 if msg.Prefix.Name == uc.nick {
620 uc.logger.Printf("changed nick from %q to %q", uc.nick, newNick)
621 uc.nick = newNick
622 }
623
624 for _, ch := range uc.channels {
625 if membership, ok := ch.Members[msg.Prefix.Name]; ok {
626 delete(ch.Members, msg.Prefix.Name)
627 ch.Members[newNick] = membership
628 uc.AppendLog(ch.Name, "*** %s is now known as %s", msg.Prefix.Name, newNick)
629 }
630 }
631
632 if msg.Prefix.Name != uc.nick {
633 uc.forEachDownstream(func(dc *downstreamConn) {
634 dc.SendMessage(&irc.Message{
635 Prefix: dc.marshalUserPrefix(uc, msg.Prefix),
636 Command: "NICK",
637 Params: []string{newNick},
638 })
639 })
640 }
641 case "JOIN":
642 if msg.Prefix == nil {
643 return fmt.Errorf("expected a prefix")
644 }
645
646 var channels string
647 if err := parseMessageParams(msg, &channels); err != nil {
648 return err
649 }
650
651 for _, ch := range strings.Split(channels, ",") {
652 if msg.Prefix.Name == uc.nick {
653 uc.logger.Printf("joined channel %q", ch)
654 uc.channels[ch] = &upstreamChannel{
655 Name: ch,
656 conn: uc,
657 Members: make(map[string]*membership),
658 }
659
660 uc.SendMessage(&irc.Message{
661 Command: "MODE",
662 Params: []string{ch},
663 })
664 } else {
665 ch, err := uc.getChannel(ch)
666 if err != nil {
667 return err
668 }
669 ch.Members[msg.Prefix.Name] = nil
670 }
671
672 uc.AppendLog(ch, "*** Joins: %s (%s@%s)", msg.Prefix.Name, msg.Prefix.User, msg.Prefix.Host)
673
674 uc.forEachDownstream(func(dc *downstreamConn) {
675 dc.SendMessage(&irc.Message{
676 Prefix: dc.marshalUserPrefix(uc, msg.Prefix),
677 Command: "JOIN",
678 Params: []string{dc.marshalChannel(uc, ch)},
679 })
680 })
681 }
682 case "PART":
683 if msg.Prefix == nil {
684 return fmt.Errorf("expected a prefix")
685 }
686
687 var channels string
688 if err := parseMessageParams(msg, &channels); err != nil {
689 return err
690 }
691
692 var reason string
693 if len(msg.Params) > 1 {
694 reason = msg.Params[1]
695 }
696
697 for _, ch := range strings.Split(channels, ",") {
698 if msg.Prefix.Name == uc.nick {
699 uc.logger.Printf("parted channel %q", ch)
700 delete(uc.channels, ch)
701 } else {
702 ch, err := uc.getChannel(ch)
703 if err != nil {
704 return err
705 }
706 delete(ch.Members, msg.Prefix.Name)
707 }
708
709 uc.AppendLog(ch, "*** Parts: %s (%s@%s) (%s)", msg.Prefix.Name, msg.Prefix.User, msg.Prefix.Host, reason)
710
711 uc.forEachDownstream(func(dc *downstreamConn) {
712 dc.SendMessage(&irc.Message{
713 Prefix: dc.marshalUserPrefix(uc, msg.Prefix),
714 Command: "PART",
715 Params: []string{dc.marshalChannel(uc, ch)},
716 })
717 })
718 }
719 case "KICK":
720 if msg.Prefix == nil {
721 return fmt.Errorf("expected a prefix")
722 }
723
724 var channel, user string
725 if err := parseMessageParams(msg, &channel, &user); err != nil {
726 return err
727 }
728
729 var reason string
730 if len(msg.Params) > 2 {
731 reason = msg.Params[1]
732 }
733
734 if user == uc.nick {
735 uc.logger.Printf("kicked from channel %q by %s", channel, msg.Prefix.Name)
736 delete(uc.channels, channel)
737 } else {
738 ch, err := uc.getChannel(channel)
739 if err != nil {
740 return err
741 }
742 delete(ch.Members, user)
743 }
744
745 uc.AppendLog(channel, "*** %s was kicked by %s (%s)", user, msg.Prefix.Name, reason)
746
747 uc.forEachDownstream(func(dc *downstreamConn) {
748 params := []string{dc.marshalChannel(uc, channel), dc.marshalNick(uc, user)}
749 if reason != "" {
750 params = append(params, reason)
751 }
752 dc.SendMessage(&irc.Message{
753 Prefix: dc.marshalUserPrefix(uc, msg.Prefix),
754 Command: "KICK",
755 Params: params,
756 })
757 })
758 case "QUIT":
759 if msg.Prefix == nil {
760 return fmt.Errorf("expected a prefix")
761 }
762
763 var reason string
764 if len(msg.Params) > 0 {
765 reason = msg.Params[0]
766 }
767
768 if msg.Prefix.Name == uc.nick {
769 uc.logger.Printf("quit")
770 }
771
772 for _, ch := range uc.channels {
773 if _, ok := ch.Members[msg.Prefix.Name]; ok {
774 delete(ch.Members, msg.Prefix.Name)
775
776 uc.AppendLog(ch.Name, "*** Quits: %s (%s@%s) (%s)", msg.Prefix.Name, msg.Prefix.User, msg.Prefix.Host, reason)
777 }
778 }
779
780 if msg.Prefix.Name != uc.nick {
781 uc.forEachDownstream(func(dc *downstreamConn) {
782 dc.SendMessage(&irc.Message{
783 Prefix: dc.marshalUserPrefix(uc, msg.Prefix),
784 Command: "QUIT",
785 Params: msg.Params,
786 })
787 })
788 }
789 case irc.RPL_TOPIC, irc.RPL_NOTOPIC:
790 var name, topic string
791 if err := parseMessageParams(msg, nil, &name, &topic); err != nil {
792 return err
793 }
794 ch, err := uc.getChannel(name)
795 if err != nil {
796 return err
797 }
798 if msg.Command == irc.RPL_TOPIC {
799 ch.Topic = topic
800 } else {
801 ch.Topic = ""
802 }
803 case "TOPIC":
804 var name string
805 if err := parseMessageParams(msg, &name); err != nil {
806 return err
807 }
808 ch, err := uc.getChannel(name)
809 if err != nil {
810 return err
811 }
812 if len(msg.Params) > 1 {
813 ch.Topic = msg.Params[1]
814 } else {
815 ch.Topic = ""
816 }
817 uc.forEachDownstream(func(dc *downstreamConn) {
818 params := []string{dc.marshalChannel(uc, name)}
819 if ch.Topic != "" {
820 params = append(params, ch.Topic)
821 }
822 dc.SendMessage(&irc.Message{
823 Prefix: dc.marshalUserPrefix(uc, msg.Prefix),
824 Command: "TOPIC",
825 Params: params,
826 })
827 })
828 case "MODE":
829 var name, modeStr string
830 if err := parseMessageParams(msg, &name, &modeStr); err != nil {
831 return err
832 }
833
834 if !uc.isChannel(name) { // user mode change
835 if name != uc.nick {
836 return fmt.Errorf("received MODE message for unknown nick %q", name)
837 }
838 return uc.modes.Apply(modeStr)
839 // TODO: notify downstreams about user mode change?
840 } else { // channel mode change
841 ch, err := uc.getChannel(name)
842 if err != nil {
843 return err
844 }
845
846 if ch.modes != nil {
847 if err := ch.modes.Apply(uc.availableChannelModes, modeStr, msg.Params[2:]...); err != nil {
848 return err
849 }
850 }
851
852 modeMsg := modeStr
853 for _, v := range msg.Params[2:] {
854 modeMsg += " " + v
855 }
856 uc.AppendLog(ch.Name, "*** %s sets mode: %s", msg.Prefix.Name, modeMsg)
857
858 uc.forEachDownstream(func(dc *downstreamConn) {
859 params := []string{dc.marshalChannel(uc, name), modeStr}
860 params = append(params, msg.Params[2:]...)
861
862 dc.SendMessage(&irc.Message{
863 Prefix: dc.marshalUserPrefix(uc, msg.Prefix),
864 Command: "MODE",
865 Params: params,
866 })
867 })
868 }
869 case irc.RPL_UMODEIS:
870 if err := parseMessageParams(msg, nil); err != nil {
871 return err
872 }
873 modeStr := ""
874 if len(msg.Params) > 1 {
875 modeStr = msg.Params[1]
876 }
877
878 uc.modes = ""
879 if err := uc.modes.Apply(modeStr); err != nil {
880 return err
881 }
882 // TODO: send RPL_UMODEIS to downstream connections when applicable
883 case irc.RPL_CHANNELMODEIS:
884 var channel string
885 if err := parseMessageParams(msg, nil, &channel); err != nil {
886 return err
887 }
888 modeStr := ""
889 if len(msg.Params) > 2 {
890 modeStr = msg.Params[2]
891 }
892
893 ch, err := uc.getChannel(channel)
894 if err != nil {
895 return err
896 }
897
898 firstMode := ch.modes == nil
899 ch.modes = make(map[byte]string)
900 if err := ch.modes.Apply(uc.availableChannelModes, modeStr, msg.Params[3:]...); err != nil {
901 return err
902 }
903 if firstMode {
904 modeStr, modeParams := ch.modes.Format()
905
906 uc.forEachDownstream(func(dc *downstreamConn) {
907 params := []string{dc.nick, dc.marshalChannel(uc, channel), modeStr}
908 params = append(params, modeParams...)
909
910 dc.SendMessage(&irc.Message{
911 Prefix: dc.srv.prefix(),
912 Command: irc.RPL_CHANNELMODEIS,
913 Params: params,
914 })
915 })
916 }
917 case rpl_creationtime:
918 var channel, creationTime string
919 if err := parseMessageParams(msg, nil, &channel, &creationTime); err != nil {
920 return err
921 }
922
923 ch, err := uc.getChannel(channel)
924 if err != nil {
925 return err
926 }
927
928 firstCreationTime := ch.creationTime == ""
929 ch.creationTime = creationTime
930 if firstCreationTime {
931 uc.forEachDownstream(func(dc *downstreamConn) {
932 dc.SendMessage(&irc.Message{
933 Prefix: dc.srv.prefix(),
934 Command: rpl_creationtime,
935 Params: []string{dc.nick, channel, creationTime},
936 })
937 })
938 }
939 case rpl_topicwhotime:
940 var name, who, timeStr string
941 if err := parseMessageParams(msg, nil, &name, &who, &timeStr); err != nil {
942 return err
943 }
944 ch, err := uc.getChannel(name)
945 if err != nil {
946 return err
947 }
948 ch.TopicWho = who
949 sec, err := strconv.ParseInt(timeStr, 10, 64)
950 if err != nil {
951 return fmt.Errorf("failed to parse topic time: %v", err)
952 }
953 ch.TopicTime = time.Unix(sec, 0)
954 case irc.RPL_LIST:
955 var channel, clients, topic string
956 if err := parseMessageParams(msg, nil, &channel, &clients, &topic); err != nil {
957 return err
958 }
959
960 pl := uc.getPendingList()
961 if pl == nil {
962 return fmt.Errorf("unexpected RPL_LIST: no matching pending LIST")
963 }
964
965 uc.forEachDownstreamByID(pl.downstreamID, func(dc *downstreamConn) {
966 dc.SendMessage(&irc.Message{
967 Prefix: dc.srv.prefix(),
968 Command: irc.RPL_LIST,
969 Params: []string{dc.nick, dc.marshalChannel(uc, channel), clients, topic},
970 })
971 })
972 case irc.RPL_LISTEND:
973 ok := uc.endPendingLists(false)
974 if !ok {
975 return fmt.Errorf("unexpected RPL_LISTEND: no matching pending LIST")
976 }
977 case irc.RPL_NAMREPLY:
978 var name, statusStr, members string
979 if err := parseMessageParams(msg, nil, &statusStr, &name, &members); err != nil {
980 return err
981 }
982
983 ch, ok := uc.channels[name]
984 if !ok {
985 // NAMES on a channel we have not joined, forward to downstream
986 uc.forEachDownstreamByID(downstreamID, func(dc *downstreamConn) {
987 channel := dc.marshalChannel(uc, name)
988 members := splitSpace(members)
989 for i, member := range members {
990 membership, nick := uc.parseMembershipPrefix(member)
991 members[i] = membership.String() + dc.marshalNick(uc, nick)
992 }
993 memberStr := strings.Join(members, " ")
994
995 dc.SendMessage(&irc.Message{
996 Prefix: dc.srv.prefix(),
997 Command: irc.RPL_NAMREPLY,
998 Params: []string{dc.nick, statusStr, channel, memberStr},
999 })
1000 })
1001 return nil
1002 }
1003
1004 status, err := parseChannelStatus(statusStr)
1005 if err != nil {
1006 return err
1007 }
1008 ch.Status = status
1009
1010 for _, s := range splitSpace(members) {
1011 membership, nick := uc.parseMembershipPrefix(s)
1012 ch.Members[nick] = membership
1013 }
1014 case irc.RPL_ENDOFNAMES:
1015 var name string
1016 if err := parseMessageParams(msg, nil, &name); err != nil {
1017 return err
1018 }
1019
1020 ch, ok := uc.channels[name]
1021 if !ok {
1022 // NAMES on a channel we have not joined, forward to downstream
1023 uc.forEachDownstreamByID(downstreamID, func(dc *downstreamConn) {
1024 channel := dc.marshalChannel(uc, name)
1025
1026 dc.SendMessage(&irc.Message{
1027 Prefix: dc.srv.prefix(),
1028 Command: irc.RPL_ENDOFNAMES,
1029 Params: []string{dc.nick, channel, "End of /NAMES list"},
1030 })
1031 })
1032 return nil
1033 }
1034
1035 if ch.complete {
1036 return fmt.Errorf("received unexpected RPL_ENDOFNAMES")
1037 }
1038 ch.complete = true
1039
1040 uc.forEachDownstream(func(dc *downstreamConn) {
1041 forwardChannel(dc, ch)
1042 })
1043 case irc.RPL_WHOREPLY:
1044 var channel, username, host, server, nick, mode, trailing string
1045 if err := parseMessageParams(msg, nil, &channel, &username, &host, &server, &nick, &mode, &trailing); err != nil {
1046 return err
1047 }
1048
1049 parts := strings.SplitN(trailing, " ", 2)
1050 if len(parts) != 2 {
1051 return fmt.Errorf("received malformed RPL_WHOREPLY: wrong trailing parameter: %s", trailing)
1052 }
1053 realname := parts[1]
1054 hops, err := strconv.Atoi(parts[0])
1055 if err != nil {
1056 return fmt.Errorf("received malformed RPL_WHOREPLY: wrong hop count: %s", parts[0])
1057 }
1058 hops++
1059
1060 trailing = strconv.Itoa(hops) + " " + realname
1061
1062 uc.forEachDownstreamByID(downstreamID, func(dc *downstreamConn) {
1063 channel := channel
1064 if channel != "*" {
1065 channel = dc.marshalChannel(uc, channel)
1066 }
1067 nick := dc.marshalNick(uc, nick)
1068 dc.SendMessage(&irc.Message{
1069 Prefix: dc.srv.prefix(),
1070 Command: irc.RPL_WHOREPLY,
1071 Params: []string{dc.nick, channel, username, host, server, nick, mode, trailing},
1072 })
1073 })
1074 case irc.RPL_ENDOFWHO:
1075 var name string
1076 if err := parseMessageParams(msg, nil, &name); err != nil {
1077 return err
1078 }
1079
1080 uc.forEachDownstreamByID(downstreamID, func(dc *downstreamConn) {
1081 name := name
1082 if name != "*" {
1083 // TODO: support WHO masks
1084 name = dc.marshalEntity(uc, name)
1085 }
1086 dc.SendMessage(&irc.Message{
1087 Prefix: dc.srv.prefix(),
1088 Command: irc.RPL_ENDOFWHO,
1089 Params: []string{dc.nick, name, "End of /WHO list"},
1090 })
1091 })
1092 case irc.RPL_WHOISUSER:
1093 var nick, username, host, realname string
1094 if err := parseMessageParams(msg, nil, &nick, &username, &host, nil, &realname); err != nil {
1095 return err
1096 }
1097
1098 uc.forEachDownstreamByID(downstreamID, func(dc *downstreamConn) {
1099 nick := dc.marshalNick(uc, nick)
1100 dc.SendMessage(&irc.Message{
1101 Prefix: dc.srv.prefix(),
1102 Command: irc.RPL_WHOISUSER,
1103 Params: []string{dc.nick, nick, username, host, "*", realname},
1104 })
1105 })
1106 case irc.RPL_WHOISSERVER:
1107 var nick, server, serverInfo string
1108 if err := parseMessageParams(msg, nil, &nick, &server, &serverInfo); err != nil {
1109 return err
1110 }
1111
1112 uc.forEachDownstreamByID(downstreamID, func(dc *downstreamConn) {
1113 nick := dc.marshalNick(uc, nick)
1114 dc.SendMessage(&irc.Message{
1115 Prefix: dc.srv.prefix(),
1116 Command: irc.RPL_WHOISSERVER,
1117 Params: []string{dc.nick, nick, server, serverInfo},
1118 })
1119 })
1120 case irc.RPL_WHOISOPERATOR:
1121 var nick string
1122 if err := parseMessageParams(msg, nil, &nick); err != nil {
1123 return err
1124 }
1125
1126 uc.forEachDownstreamByID(downstreamID, func(dc *downstreamConn) {
1127 nick := dc.marshalNick(uc, nick)
1128 dc.SendMessage(&irc.Message{
1129 Prefix: dc.srv.prefix(),
1130 Command: irc.RPL_WHOISOPERATOR,
1131 Params: []string{dc.nick, nick, "is an IRC operator"},
1132 })
1133 })
1134 case irc.RPL_WHOISIDLE:
1135 var nick string
1136 if err := parseMessageParams(msg, nil, &nick, nil); err != nil {
1137 return err
1138 }
1139
1140 uc.forEachDownstreamByID(downstreamID, func(dc *downstreamConn) {
1141 nick := dc.marshalNick(uc, nick)
1142 params := []string{dc.nick, nick}
1143 params = append(params, msg.Params[2:]...)
1144 dc.SendMessage(&irc.Message{
1145 Prefix: dc.srv.prefix(),
1146 Command: irc.RPL_WHOISIDLE,
1147 Params: params,
1148 })
1149 })
1150 case irc.RPL_WHOISCHANNELS:
1151 var nick, channelList string
1152 if err := parseMessageParams(msg, nil, &nick, &channelList); err != nil {
1153 return err
1154 }
1155 channels := splitSpace(channelList)
1156
1157 uc.forEachDownstreamByID(downstreamID, func(dc *downstreamConn) {
1158 nick := dc.marshalNick(uc, nick)
1159 channelList := make([]string, len(channels))
1160 for i, channel := range channels {
1161 prefix, channel := uc.parseMembershipPrefix(channel)
1162 channel = dc.marshalChannel(uc, channel)
1163 channelList[i] = prefix.String() + channel
1164 }
1165 channels := strings.Join(channelList, " ")
1166 dc.SendMessage(&irc.Message{
1167 Prefix: dc.srv.prefix(),
1168 Command: irc.RPL_WHOISCHANNELS,
1169 Params: []string{dc.nick, nick, channels},
1170 })
1171 })
1172 case irc.RPL_ENDOFWHOIS:
1173 var nick string
1174 if err := parseMessageParams(msg, nil, &nick); err != nil {
1175 return err
1176 }
1177
1178 uc.forEachDownstreamByID(downstreamID, func(dc *downstreamConn) {
1179 nick := dc.marshalNick(uc, nick)
1180 dc.SendMessage(&irc.Message{
1181 Prefix: dc.srv.prefix(),
1182 Command: irc.RPL_ENDOFWHOIS,
1183 Params: []string{dc.nick, nick, "End of /WHOIS list"},
1184 })
1185 })
1186 case "PRIVMSG":
1187 if msg.Prefix == nil {
1188 return fmt.Errorf("expected a prefix")
1189 }
1190
1191 var nick, text string
1192 if err := parseMessageParams(msg, &nick, &text); err != nil {
1193 return err
1194 }
1195
1196 if msg.Prefix.Name == serviceNick {
1197 uc.logger.Printf("skipping PRIVMSG from soju's service: %v", msg)
1198 break
1199 }
1200 if nick == serviceNick {
1201 uc.logger.Printf("skipping PRIVMSG to soju's service: %v", msg)
1202 break
1203 }
1204
1205 target := nick
1206 if nick == uc.nick {
1207 target = msg.Prefix.Name
1208 }
1209 uc.AppendLog(target, "<%s> %s", msg.Prefix.Name, text)
1210
1211 uc.network.ring.Produce(msg)
1212 case "INVITE":
1213 var nick string
1214 var channel string
1215 if err := parseMessageParams(msg, &nick, &channel); err != nil {
1216 return err
1217 }
1218
1219 uc.forEachDownstream(func(dc *downstreamConn) {
1220 dc.SendMessage(&irc.Message{
1221 Prefix: dc.marshalUserPrefix(uc, msg.Prefix),
1222 Command: "INVITE",
1223 Params: []string{dc.marshalNick(uc, nick), dc.marshalChannel(uc, channel)},
1224 })
1225 })
1226 case irc.RPL_INVITING:
1227 var nick string
1228 var channel string
1229 if err := parseMessageParams(msg, &nick, &channel); err != nil {
1230 return err
1231 }
1232
1233 uc.forEachDownstreamByID(downstreamID, func(dc *downstreamConn) {
1234 dc.SendMessage(&irc.Message{
1235 Prefix: dc.srv.prefix(),
1236 Command: irc.RPL_INVITING,
1237 Params: []string{dc.nick, dc.marshalNick(uc, nick), dc.marshalChannel(uc, channel)},
1238 })
1239 })
1240 case irc.ERR_UNKNOWNCOMMAND, irc.RPL_TRYAGAIN:
1241 var command, reason string
1242 if err := parseMessageParams(msg, nil, &command, &reason); err != nil {
1243 return err
1244 }
1245
1246 if command == "LIST" {
1247 ok := uc.endPendingLists(false)
1248 if !ok {
1249 return fmt.Errorf("unexpected response for LIST: %q: no matching pending LIST", msg.Command)
1250 }
1251 uc.forEachDownstreamByID(downstreamID, func(dc *downstreamConn) {
1252 dc.SendMessage(&irc.Message{
1253 Prefix: uc.srv.prefix(),
1254 Command: msg.Command,
1255 Params: []string{dc.nick, "LIST", reason},
1256 })
1257 })
1258 }
1259 case "TAGMSG":
1260 // TODO: relay to downstream connections that accept message-tags
1261 case "ACK":
1262 // Ignore
1263 case irc.RPL_YOURHOST, irc.RPL_CREATED:
1264 // Ignore
1265 case irc.RPL_LUSERCLIENT, irc.RPL_LUSEROP, irc.RPL_LUSERUNKNOWN, irc.RPL_LUSERCHANNELS, irc.RPL_LUSERME:
1266 // Ignore
1267 case irc.RPL_MOTDSTART, irc.RPL_MOTD, irc.RPL_ENDOFMOTD:
1268 // Ignore
1269 case irc.RPL_LISTSTART:
1270 // Ignore
1271 case rpl_localusers, rpl_globalusers:
1272 // Ignore
1273 case irc.RPL_STATSVLINE, rpl_statsping, irc.RPL_STATSBLINE, irc.RPL_STATSDLINE:
1274 // Ignore
1275 default:
1276 uc.logger.Printf("unhandled message: %v", msg)
1277 }
1278 return nil
1279}
1280
1281func splitSpace(s string) []string {
1282 return strings.FieldsFunc(s, func(r rune) bool {
1283 return r == ' '
1284 })
1285}
1286
1287func (uc *upstreamConn) register() {
1288 uc.nick = uc.network.Nick
1289 uc.username = uc.network.Username
1290 if uc.username == "" {
1291 uc.username = uc.nick
1292 }
1293 uc.realname = uc.network.Realname
1294 if uc.realname == "" {
1295 uc.realname = uc.nick
1296 }
1297
1298 uc.SendMessage(&irc.Message{
1299 Command: "CAP",
1300 Params: []string{"LS", "302"},
1301 })
1302
1303 if uc.network.Pass != "" {
1304 uc.SendMessage(&irc.Message{
1305 Command: "PASS",
1306 Params: []string{uc.network.Pass},
1307 })
1308 }
1309
1310 uc.SendMessage(&irc.Message{
1311 Command: "NICK",
1312 Params: []string{uc.nick},
1313 })
1314 uc.SendMessage(&irc.Message{
1315 Command: "USER",
1316 Params: []string{uc.username, "0", "*", uc.realname},
1317 })
1318}
1319
1320func (uc *upstreamConn) requestSASL() bool {
1321 if uc.network.SASL.Mechanism == "" {
1322 return false
1323 }
1324
1325 v, ok := uc.caps["sasl"]
1326 if !ok {
1327 return false
1328 }
1329 if v != "" {
1330 mechanisms := strings.Split(v, ",")
1331 found := false
1332 for _, mech := range mechanisms {
1333 if strings.EqualFold(mech, uc.network.SASL.Mechanism) {
1334 found = true
1335 break
1336 }
1337 }
1338 if !found {
1339 return false
1340 }
1341 }
1342
1343 return true
1344}
1345
1346func (uc *upstreamConn) handleCapAck(name string, ok bool) error {
1347 auth := &uc.network.SASL
1348 switch name {
1349 case "sasl":
1350 if !ok {
1351 uc.logger.Printf("server refused to acknowledge the SASL capability")
1352 return nil
1353 }
1354
1355 switch auth.Mechanism {
1356 case "PLAIN":
1357 uc.logger.Printf("starting SASL PLAIN authentication with username %q", auth.Plain.Username)
1358 uc.saslClient = sasl.NewPlainClient("", auth.Plain.Username, auth.Plain.Password)
1359 default:
1360 return fmt.Errorf("unsupported SASL mechanism %q", name)
1361 }
1362
1363 uc.SendMessage(&irc.Message{
1364 Command: "AUTHENTICATE",
1365 Params: []string{auth.Mechanism},
1366 })
1367 case "message-tags":
1368 uc.tagsSupported = ok
1369 case "labeled-response":
1370 uc.labelsSupported = ok
1371 }
1372 return nil
1373}
1374
1375func (uc *upstreamConn) readMessages(ch chan<- event) error {
1376 for {
1377 msg, err := uc.irc.ReadMessage()
1378 if err == io.EOF {
1379 break
1380 } else if err != nil {
1381 return fmt.Errorf("failed to read IRC command: %v", err)
1382 }
1383
1384 if uc.srv.Debug {
1385 uc.logger.Printf("received: %v", msg)
1386 }
1387
1388 ch <- eventUpstreamMessage{msg, uc}
1389 }
1390
1391 return nil
1392}
1393
1394func (uc *upstreamConn) SendMessage(msg *irc.Message) {
1395 uc.outgoing <- msg
1396}
1397
1398func (uc *upstreamConn) SendMessageLabeled(downstreamID uint64, msg *irc.Message) {
1399 if uc.labelsSupported {
1400 if msg.Tags == nil {
1401 msg.Tags = make(map[string]irc.TagValue)
1402 }
1403 msg.Tags["label"] = irc.TagValue(fmt.Sprintf("sd-%d-%d", downstreamID, uc.nextLabelID))
1404 uc.nextLabelID++
1405 }
1406 uc.SendMessage(msg)
1407}
1408
1409// TODO: handle moving logs when a network name changes, when support for this is added
1410func (uc *upstreamConn) AppendLog(entity string, format string, a ...interface{}) {
1411 if uc.srv.LogPath == "" {
1412 return
1413 }
1414 // TODO: enforce maximum open file handles (LRU cache of file handles)
1415 // TODO: handle non-monotonic clock behaviour
1416 now := time.Now()
1417 year, month, day := now.Date()
1418 name := fmt.Sprintf("%04d-%02d-%02d.log", year, month, day)
1419 log, ok := uc.logs[entity]
1420 if !ok || log.name != name {
1421 if ok {
1422 log.file.Close()
1423 delete(uc.logs, entity)
1424 }
1425 // TODO: handle/forbid network/entity names with illegal path characters
1426 dir := filepath.Join(uc.srv.LogPath, uc.user.Username, uc.network.Name, entity)
1427 if err := os.MkdirAll(dir, 0600); err != nil {
1428 uc.logger.Printf("failed to log message: could not create logs directory %q: %v", dir, err)
1429 return
1430 }
1431 path := filepath.Join(dir, name)
1432 f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0600)
1433 if err != nil {
1434 uc.logger.Printf("failed to log message: could not open or create log file %q: %v", path, err)
1435 return
1436 }
1437 log = entityLog{
1438 name: name,
1439 file: f,
1440 }
1441 uc.logs[entity] = log
1442 }
1443
1444 format = "[%02d:%02d:%02d] " + format + "\n"
1445 args := []interface{}{now.Hour(), now.Minute(), now.Second()}
1446 args = append(args, a...)
1447
1448 if _, err := fmt.Fprintf(log.file, format, args...); err != nil {
1449 uc.logger.Printf("failed to log message to %q: %v", log.name, err)
1450 }
1451}
Note: See TracBrowser for help on using the repository browser.