source: code/trunk/upstream.go@ 234

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

Remove downstreamConn.ringConsumers

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