source: code/trunk/upstream.go@ 194

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

Request server-time cap

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