source: code/trunk/upstream.go@ 285

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

Add support for detached channels

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