source: code/trunk/upstream.go@ 286

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

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