source: code/trunk/upstream.go@ 180

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

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