Context Navigation

source: code/trunk/downstream.go@ 179

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

Introduce eventUpstreamDisconnected

This allows us to perform cleanup actions in the user goroutine. This
removes the need for pendingLISTsLock.

File size: 31.1 KB

Rev	Line
[98]	1	package soju
[13]	2
	3	import (
[91]	4	"crypto/tls"
[112]	5	"encoding/base64"
[13]	6	"fmt"
	7	"io"
	8	"net"
[108]	9	"strconv"
[39]	10	"strings"
[105]	11	"sync"
[91]	12	"time"
[13]	13
[112]	14	"github.com/emersion/go-sasl"
[85]	15	"golang.org/x/crypto/bcrypt"
[13]	16	"gopkg.in/irc.v3"
	17	)
	18
	19	type ircError struct {
	20	Message *irc.Message
	21	}
	22
[85]	23	func (err ircError) Error() string {
	24	return err.Message.String()
	25	}
	26
[13]	27	func newUnknownCommandError(cmd string) ircError {
	28	return ircError{&irc.Message{
	29	Command: irc.ERR_UNKNOWNCOMMAND,
	30	Params: []string{
	31	"*",
	32	cmd,
	33	"Unknown command",
	34	},
	35	}}
	36	}
	37
	38	func newNeedMoreParamsError(cmd string) ircError {
	39	return ircError{&irc.Message{
	40	Command: irc.ERR_NEEDMOREPARAMS,
	41	Params: []string{
	42	"*",
	43	cmd,
	44	"Not enough parameters",
	45	},
	46	}}
	47	}
	48
[85]	49	var errAuthFailed = ircError{&irc.Message{
	50	Command: irc.ERR_PASSWDMISMATCH,
	51	Params: []string{"*", "Invalid username or password"},
	52	}}
[13]	53
[104]	54	type ringMessage struct {
[69]	55	consumer *RingConsumer
	56	upstreamConn *upstreamConn
	57	}
	58
[13]	59	type downstreamConn struct {
[154]	60	id uint64
[69]	61	net net.Conn
	62	irc *irc.Conn
	63	srv *Server
	64	logger Logger
[102]	65	outgoing chan *irc.Message
[104]	66	ringMessages chan ringMessage
[69]	67	closed chan struct{}
[22]	68
[100]	69	registered bool
	70	user *user
	71	nick string
	72	username string
	73	rawUsername string
[168]	74	networkName string
[100]	75	realname string
[141]	76	hostname string
[100]	77	password string // empty after authentication
	78	network *network // can be nil
[105]	79
[108]	80	negociatingCaps bool
	81	capVersion int
	82	caps map[string]bool
	83
[112]	84	saslServer sasl.Server
	85
[105]	86	lock sync.Mutex
	87	ourMessages map[*irc.Message]struct{}
[13]	88	}
	89
[154]	90	func newDownstreamConn(srv Server, netConn net.Conn, id uint64) downstreamConn {
[55]	91	dc := &downstreamConn{
[154]	92	id: id,
[69]	93	net: netConn,
	94	irc: irc.NewConn(netConn),
	95	srv: srv,
	96	logger: &prefixLogger{srv.Logger, fmt.Sprintf("downstream %q: ", netConn.RemoteAddr())},
[102]	97	outgoing: make(chan *irc.Message, 64),
[104]	98	ringMessages: make(chan ringMessage),
[69]	99	closed: make(chan struct{}),
[108]	100	caps: make(map[string]bool),
[105]	101	ourMessages: make(map[*irc.Message]struct{}),
[22]	102	}
[141]	103	dc.hostname = netConn.RemoteAddr().String()
	104	if host, _, err := net.SplitHostPort(dc.hostname); err == nil {
	105	dc.hostname = host
	106	}
[26]	107
	108	go func() {
[56]	109	if err := dc.writeMessages(); err != nil {
	110	dc.logger.Printf("failed to write message: %v", err)
[26]	111	}
[55]	112	if err := dc.net.Close(); err != nil {
	113	dc.logger.Printf("failed to close connection: %v", err)
[45]	114	} else {
[55]	115	dc.logger.Printf("connection closed")
[45]	116	}
[26]	117	}()
	118
[130]	119	dc.logger.Printf("new connection")
[55]	120	return dc
[22]	121	}
	122
[55]	123	func (dc downstreamConn) prefix() irc.Prefix {
[27]	124	return &irc.Prefix{
[55]	125	Name: dc.nick,
	126	User: dc.username,
[141]	127	Host: dc.hostname,
[27]	128	}
	129	}
	130
[90]	131	func (dc downstreamConn) forEachNetwork(f func(network)) {
	132	if dc.network != nil {
	133	f(dc.network)
	134	} else {
	135	dc.user.forEachNetwork(f)
	136	}
	137	}
	138
[73]	139	func (dc downstreamConn) forEachUpstream(f func(upstreamConn)) {
	140	dc.user.forEachUpstream(func(uc *upstreamConn) {
[77]	141	if dc.network != nil && uc.network != dc.network {
[73]	142	return
	143	}
	144	f(uc)
	145	})
	146	}
	147
[89]	148	// upstream returns the upstream connection, if any. If there are zero or if
	149	// there are multiple upstream connections, it returns nil.
	150	func (dc downstreamConn) upstream() upstreamConn {
	151	if dc.network == nil {
	152	return nil
	153	}
[136]	154	return dc.network.upstream()
[89]	155	}
	156
[129]	157	func (dc downstreamConn) marshalEntity(uc upstreamConn, entity string) string {
	158	if uc.isChannel(entity) {
	159	return dc.marshalChannel(uc, entity)
[119]	160	}
[129]	161	return dc.marshalNick(uc, entity)
[119]	162	}
	163
	164	func (dc downstreamConn) marshalChannel(uc upstreamConn, name string) string {
[130]	165	if dc.network != nil {
[119]	166	return name
	167	}
	168	return name + "/" + uc.network.GetName()
	169	}
	170
[127]	171	func (dc downstreamConn) unmarshalEntity(name string) (upstreamConn, string, error) {
[89]	172	if uc := dc.upstream(); uc != nil {
	173	return uc, name, nil
	174	}
	175
[127]	176	var conn *upstreamConn
[119]	177	if i := strings.LastIndexByte(name, '/'); i >= 0 {
[127]	178	network := name[i+1:]
[119]	179	name = name[:i]
	180
	181	dc.forEachUpstream(func(uc *upstreamConn) {
	182	if network != uc.network.GetName() {
	183	return
	184	}
	185	conn = uc
	186	})
	187	}
	188
[127]	189	if conn == nil {
[73]	190	return nil, "", ircError{&irc.Message{
	191	Command: irc.ERR_NOSUCHCHANNEL,
	192	Params: []string{name, "No such channel"},
	193	}}
[69]	194	}
[127]	195	return conn, name, nil
[69]	196	}
	197
	198	func (dc downstreamConn) marshalNick(uc upstreamConn, nick string) string {
	199	if nick == uc.nick {
	200	return dc.nick
	201	}
[130]	202	if dc.network != nil {
[119]	203	return nick
	204	}
	205	return nick + "/" + uc.network.GetName()
[69]	206	}
	207
	208	func (dc downstreamConn) marshalUserPrefix(uc upstreamConn, prefix irc.Prefix) irc.Prefix {
	209	if prefix.Name == uc.nick {
	210	return dc.prefix()
	211	}
[130]	212	if dc.network != nil {
[119]	213	return prefix
	214	}
	215	return &irc.Prefix{
	216	Name: prefix.Name + "/" + uc.network.GetName(),
	217	User: prefix.User,
	218	Host: prefix.Host,
	219	}
[69]	220	}
	221
[57]	222	func (dc *downstreamConn) isClosed() bool {
	223	select {
	224	case <-dc.closed:
	225	return true
	226	default:
	227	return false
	228	}
	229	}
	230
[165]	231	func (dc *downstreamConn) readMessages(ch chan<- event) error {
[22]	232	for {
[55]	233	msg, err := dc.irc.ReadMessage()
[22]	234	if err == io.EOF {
	235	break
	236	} else if err != nil {
	237	return fmt.Errorf("failed to read IRC command: %v", err)
	238	}
	239
[64]	240	if dc.srv.Debug {
	241	dc.logger.Printf("received: %v", msg)
	242	}
	243
[165]	244	ch <- eventDownstreamMessage{msg, dc}
[22]	245	}
	246
[45]	247	return nil
[22]	248	}
	249
[56]	250	func (dc *downstreamConn) writeMessages() error {
[57]	251	for {
	252	var err error
	253	var closed bool
	254	select {
[102]	255	case msg := <-dc.outgoing:
[64]	256	if dc.srv.Debug {
	257	dc.logger.Printf("sent: %v", msg)
	258	}
[57]	259	err = dc.irc.WriteMessage(msg)
[104]	260	case ringMessage := <-dc.ringMessages:
	261	consumer, uc := ringMessage.consumer, ringMessage.upstreamConn
[57]	262	for {
	263	msg := consumer.Peek()
	264	if msg == nil {
	265	break
	266	}
[105]	267
	268	dc.lock.Lock()
	269	_, ours := dc.ourMessages[msg]
	270	delete(dc.ourMessages, msg)
	271	dc.lock.Unlock()
	272	if ours {
	273	// The message comes from our connection, don't echo it
	274	// back
[113]	275	consumer.Consume()
[105]	276	continue
	277	}
	278
[69]	279	msg = msg.Copy()
	280	switch msg.Command {
	281	case "PRIVMSG":
[119]	282	msg.Prefix = dc.marshalUserPrefix(uc, msg.Prefix)
	283	msg.Params[0] = dc.marshalEntity(uc, msg.Params[0])
[69]	284	default:
	285	panic("expected to consume a PRIVMSG message")
	286	}
[64]	287	if dc.srv.Debug {
	288	dc.logger.Printf("sent: %v", msg)
	289	}
[57]	290	err = dc.irc.WriteMessage(msg)
	291	if err != nil {
	292	break
	293	}
	294	consumer.Consume()
	295	}
	296	case <-dc.closed:
	297	closed = true
	298	}
	299	if err != nil {
[56]	300	return err
	301	}
[57]	302	if closed {
	303	break
	304	}
[56]	305	}
	306	return nil
	307	}
	308
[55]	309	func (dc *downstreamConn) Close() error {
[57]	310	if dc.isClosed() {
[26]	311	return fmt.Errorf("downstream connection already closed")
	312	}
[40]	313
[57]	314	close(dc.closed)
[45]	315	return nil
[13]	316	}
	317
[55]	318	func (dc downstreamConn) SendMessage(msg irc.Message) {
[102]	319	dc.outgoing <- msg
[54]	320	}
	321
[55]	322	func (dc downstreamConn) handleMessage(msg irc.Message) error {
[13]	323	switch msg.Command {
[28]	324	case "QUIT":
[55]	325	return dc.Close()
[13]	326	default:
[55]	327	if dc.registered {
	328	return dc.handleMessageRegistered(msg)
[13]	329	} else {
[55]	330	return dc.handleMessageUnregistered(msg)
[13]	331	}
	332	}
	333	}
	334
[55]	335	func (dc downstreamConn) handleMessageUnregistered(msg irc.Message) error {
[13]	336	switch msg.Command {
	337	case "NICK":
[117]	338	var nick string
	339	if err := parseMessageParams(msg, &nick); err != nil {
[43]	340	return err
[13]	341	}
[117]	342	if nick == serviceNick {
	343	return ircError{&irc.Message{
	344	Command: irc.ERR_NICKNAMEINUSE,
	345	Params: []string{dc.nick, nick, "Nickname reserved for bouncer service"},
	346	}}
	347	}
	348	dc.nick = nick
[13]	349	case "USER":
[117]	350	if err := parseMessageParams(msg, &dc.rawUsername, nil, nil, &dc.realname); err != nil {
[43]	351	return err
[13]	352	}
[85]	353	case "PASS":
	354	if err := parseMessageParams(msg, &dc.password); err != nil {
	355	return err
	356	}
[108]	357	case "CAP":
	358	var subCmd string
	359	if err := parseMessageParams(msg, &subCmd); err != nil {
	360	return err
	361	}
	362	if err := dc.handleCapCommand(subCmd, msg.Params[1:]); err != nil {
	363	return err
	364	}
[112]	365	case "AUTHENTICATE":
	366	if !dc.caps["sasl"] {
	367	return ircError{&irc.Message{
[125]	368	Command: irc.ERR_SASLFAIL,
[112]	369	Params: []string{"*", "AUTHENTICATE requires the \"sasl\" capability to be enabled"},
	370	}}
	371	}
	372	if len(msg.Params) == 0 {
	373	return ircError{&irc.Message{
[125]	374	Command: irc.ERR_SASLFAIL,
[112]	375	Params: []string{"*", "Missing AUTHENTICATE argument"},
	376	}}
	377	}
	378	if dc.nick == "" {
	379	return ircError{&irc.Message{
[125]	380	Command: irc.ERR_SASLFAIL,
[112]	381	Params: []string{"*", "Expected NICK command before AUTHENTICATE"},
	382	}}
	383	}
	384
	385	var resp []byte
	386	if dc.saslServer == nil {
	387	mech := strings.ToUpper(msg.Params[0])
	388	switch mech {
	389	case "PLAIN":
	390	dc.saslServer = sasl.NewPlainServer(sasl.PlainAuthenticator(func(identity, username, password string) error {
	391	return dc.authenticate(username, password)
	392	}))
	393	default:
	394	return ircError{&irc.Message{
[125]	395	Command: irc.ERR_SASLFAIL,
[112]	396	Params: []string{"*", fmt.Sprintf("Unsupported SASL mechanism %q", mech)},
	397	}}
	398	}
	399	} else if msg.Params[0] == "*" {
	400	dc.saslServer = nil
	401	return ircError{&irc.Message{
[125]	402	Command: irc.ERR_SASLABORTED,
[112]	403	Params: []string{"*", "SASL authentication aborted"},
	404	}}
	405	} else if msg.Params[0] == "+" {
	406	resp = nil
	407	} else {
	408	// TODO: multi-line messages
	409	var err error
	410	resp, err = base64.StdEncoding.DecodeString(msg.Params[0])
	411	if err != nil {
	412	dc.saslServer = nil
	413	return ircError{&irc.Message{
[125]	414	Command: irc.ERR_SASLFAIL,
[112]	415	Params: []string{"*", "Invalid base64-encoded response"},
	416	}}
	417	}
	418	}
	419
	420	challenge, done, err := dc.saslServer.Next(resp)
	421	if err != nil {
	422	dc.saslServer = nil
	423	if ircErr, ok := err.(ircError); ok && ircErr.Message.Command == irc.ERR_PASSWDMISMATCH {
	424	return ircError{&irc.Message{
[125]	425	Command: irc.ERR_SASLFAIL,
[112]	426	Params: []string{"*", ircErr.Message.Params[1]},
	427	}}
	428	}
	429	dc.SendMessage(&irc.Message{
	430	Prefix: dc.srv.prefix(),
[125]	431	Command: irc.ERR_SASLFAIL,
[112]	432	Params: []string{"*", "SASL error"},
	433	})
	434	return fmt.Errorf("SASL authentication failed: %v", err)
	435	} else if done {
	436	dc.saslServer = nil
	437	dc.SendMessage(&irc.Message{
	438	Prefix: dc.srv.prefix(),
[125]	439	Command: irc.RPL_LOGGEDIN,
[112]	440	Params: []string{dc.nick, dc.nick, dc.user.Username, "You are now logged in"},
	441	})
	442	dc.SendMessage(&irc.Message{
	443	Prefix: dc.srv.prefix(),
[125]	444	Command: irc.RPL_SASLSUCCESS,
[112]	445	Params: []string{dc.nick, "SASL authentication successful"},
	446	})
	447	} else {
	448	challengeStr := "+"
[135]	449	if len(challenge) > 0 {
[112]	450	challengeStr = base64.StdEncoding.EncodeToString(challenge)
	451	}
	452
	453	// TODO: multi-line messages
	454	dc.SendMessage(&irc.Message{
	455	Prefix: dc.srv.prefix(),
	456	Command: "AUTHENTICATE",
	457	Params: []string{challengeStr},
	458	})
	459	}
[13]	460	default:
[55]	461	dc.logger.Printf("unhandled message: %v", msg)
[13]	462	return newUnknownCommandError(msg.Command)
	463	}
[108]	464	if dc.rawUsername != "" && dc.nick != "" && !dc.negociatingCaps {
[55]	465	return dc.register()
[13]	466	}
	467	return nil
	468	}
	469
[108]	470	func (dc *downstreamConn) handleCapCommand(cmd string, args []string) error {
[111]	471	cmd = strings.ToUpper(cmd)
	472
[108]	473	replyTo := dc.nick
	474	if !dc.registered {
	475	replyTo = "*"
	476	}
	477
	478	switch cmd {
	479	case "LS":
	480	if len(args) > 0 {
	481	var err error
	482	if dc.capVersion, err = strconv.Atoi(args[0]); err != nil {
	483	return err
	484	}
	485	}
	486
	487	var caps []string
[112]	488	if dc.capVersion >= 302 {
[108]	489	caps = append(caps, "sasl=PLAIN")
	490	} else {
	491	caps = append(caps, "sasl")
[112]	492	}
[108]	493
	494	// TODO: multi-line replies
	495	dc.SendMessage(&irc.Message{
	496	Prefix: dc.srv.prefix(),
	497	Command: "CAP",
	498	Params: []string{replyTo, "LS", strings.Join(caps, " ")},
	499	})
	500
	501	if !dc.registered {
	502	dc.negociatingCaps = true
	503	}
	504	case "LIST":
	505	var caps []string
	506	for name := range dc.caps {
	507	caps = append(caps, name)
	508	}
	509
	510	// TODO: multi-line replies
	511	dc.SendMessage(&irc.Message{
	512	Prefix: dc.srv.prefix(),
	513	Command: "CAP",
	514	Params: []string{replyTo, "LIST", strings.Join(caps, " ")},
	515	})
	516	case "REQ":
	517	if len(args) == 0 {
	518	return ircError{&irc.Message{
	519	Command: err_invalidcapcmd,
	520	Params: []string{replyTo, cmd, "Missing argument in CAP REQ command"},
	521	}}
	522	}
	523
	524	caps := strings.Fields(args[0])
	525	ack := true
	526	for _, name := range caps {
	527	name = strings.ToLower(name)
	528	enable := !strings.HasPrefix(name, "-")
	529	if !enable {
	530	name = strings.TrimPrefix(name, "-")
	531	}
	532
	533	enabled := dc.caps[name]
	534	if enable == enabled {
	535	continue
	536	}
	537
	538	switch name {
[112]	539	case "sasl":
	540	dc.caps[name] = enable
[108]	541	default:
	542	ack = false
	543	}
	544	}
	545
	546	reply := "NAK"
	547	if ack {
	548	reply = "ACK"
	549	}
	550	dc.SendMessage(&irc.Message{
	551	Prefix: dc.srv.prefix(),
	552	Command: "CAP",
	553	Params: []string{replyTo, reply, args[0]},
	554	})
	555	case "END":
	556	dc.negociatingCaps = false
	557	default:
	558	return ircError{&irc.Message{
	559	Command: err_invalidcapcmd,
	560	Params: []string{replyTo, cmd, "Unknown CAP command"},
	561	}}
	562	}
	563	return nil
	564	}
	565
[91]	566	func sanityCheckServer(addr string) error {
	567	dialer := net.Dialer{Timeout: 30 * time.Second}
	568	conn, err := tls.DialWithDialer(&dialer, "tcp", addr, nil)
	569	if err != nil {
	570	return err
	571	}
	572	return conn.Close()
	573	}
	574
[112]	575	func unmarshalUsername(rawUsername string) (username, network string) {
	576	username = rawUsername
[73]	577	if i := strings.LastIndexAny(username, "/@"); i >= 0 {
[112]	578	network = username[i+1:]
[73]	579	}
	580	if i := strings.IndexAny(username, "/@"); i >= 0 {
	581	username = username[:i]
	582	}
[112]	583	return username, network
	584	}
[73]	585
[168]	586	func (dc *downstreamConn) authenticate(username, password string) error {
	587	username, networkName := unmarshalUsername(username)
	588
[173]	589	u, err := dc.srv.db.GetUser(username)
	590	if err != nil {
	591	dc.logger.Printf("failed authentication for %q: %v", username, err)
[168]	592	return errAuthFailed
	593	}
	594
[173]	595	err = bcrypt.CompareHashAndPassword([]byte(u.Password), []byte(password))
[168]	596	if err != nil {
	597	dc.logger.Printf("failed authentication for %q: %v", username, err)
	598	return errAuthFailed
	599	}
	600
[173]	601	dc.user = dc.srv.getUser(username)
	602	if dc.user == nil {
	603	dc.logger.Printf("failed authentication for %q: user not active", username)
	604	return errAuthFailed
	605	}
[168]	606	dc.networkName = networkName
	607	return nil
	608	}
	609
	610	func (dc *downstreamConn) register() error {
	611	if dc.registered {
	612	return fmt.Errorf("tried to register twice")
	613	}
	614
	615	password := dc.password
	616	dc.password = ""
	617	if dc.user == nil {
	618	if err := dc.authenticate(dc.rawUsername, password); err != nil {
	619	return err
	620	}
	621	}
	622
	623	if dc.networkName == "" {
	624	_, dc.networkName = unmarshalUsername(dc.rawUsername)
	625	}
	626
	627	dc.registered = true
	628	dc.username = dc.user.Username
	629	dc.logger.Printf("registration complete for user %q", dc.username)
	630	return nil
	631	}
	632
	633	func (dc *downstreamConn) loadNetwork() error {
	634	if dc.networkName == "" {
[112]	635	return nil
	636	}
[85]	637
[168]	638	network := dc.user.getNetwork(dc.networkName)
[112]	639	if network == nil {
[168]	640	addr := dc.networkName
[112]	641	if !strings.ContainsRune(addr, ':') {
	642	addr = addr + ":6697"
	643	}
	644
	645	dc.logger.Printf("trying to connect to new network %q", addr)
	646	if err := sanityCheckServer(addr); err != nil {
	647	dc.logger.Printf("failed to connect to %q: %v", addr, err)
	648	return ircError{&irc.Message{
	649	Command: irc.ERR_PASSWDMISMATCH,
[168]	650	Params: []string{"*", fmt.Sprintf("Failed to connect to %q", dc.networkName)},
[112]	651	}}
	652	}
	653
[168]	654	dc.logger.Printf("auto-saving network %q", dc.networkName)
[112]	655	var err error
[120]	656	network, err = dc.user.createNetwork(&Network{
[168]	657	Addr: dc.networkName,
[120]	658	Nick: dc.nick,
	659	})
[112]	660	if err != nil {
	661	return err
	662	}
	663	}
	664
	665	dc.network = network
	666	return nil
	667	}
	668
[168]	669	func (dc *downstreamConn) welcome() error {
	670	if dc.user == nil \|\| !dc.registered {
	671	panic("tried to welcome an unregistered connection")
[37]	672	}
	673
[168]	674	// TODO: doing this might take some time. We should do it in dc.register
	675	// instead, but we'll potentially be adding a new network and this must be
	676	// done in the user goroutine.
	677	if err := dc.loadNetwork(); err != nil {
	678	return err
[85]	679	}
	680
[166]	681	firstDownstream := len(dc.user.downstreamConns) == 0
[40]	682
[55]	683	dc.SendMessage(&irc.Message{
	684	Prefix: dc.srv.prefix(),
[13]	685	Command: irc.RPL_WELCOME,
[98]	686	Params: []string{dc.nick, "Welcome to soju, " + dc.nick},
[54]	687	})
[55]	688	dc.SendMessage(&irc.Message{
	689	Prefix: dc.srv.prefix(),
[13]	690	Command: irc.RPL_YOURHOST,
[55]	691	Params: []string{dc.nick, "Your host is " + dc.srv.Hostname},
[54]	692	})
[55]	693	dc.SendMessage(&irc.Message{
	694	Prefix: dc.srv.prefix(),
[13]	695	Command: irc.RPL_CREATED,
[55]	696	Params: []string{dc.nick, "Who cares when the server was created?"},
[54]	697	})
[55]	698	dc.SendMessage(&irc.Message{
	699	Prefix: dc.srv.prefix(),
[13]	700	Command: irc.RPL_MYINFO,
[98]	701	Params: []string{dc.nick, dc.srv.Hostname, "soju", "aiwroO", "OovaimnqpsrtklbeI"},
[54]	702	})
[93]	703	// TODO: RPL_ISUPPORT
[55]	704	dc.SendMessage(&irc.Message{
	705	Prefix: dc.srv.prefix(),
[13]	706	Command: irc.ERR_NOMOTD,
[55]	707	Params: []string{dc.nick, "No MOTD"},
[54]	708	})
[13]	709
[73]	710	dc.forEachUpstream(func(uc *upstreamConn) {
[30]	711	for _, ch := range uc.channels {
	712	if ch.complete {
[132]	713	dc.SendMessage(&irc.Message{
	714	Prefix: dc.prefix(),
	715	Command: "JOIN",
	716	Params: []string{dc.marshalChannel(ch.conn, ch.Name)},
	717	})
	718
[55]	719	forwardChannel(dc, ch)
[30]	720	}
	721	}
[143]	722	})
[50]	723
[143]	724	dc.forEachNetwork(func(net *network) {
[144]	725	// TODO: need to take dc.network into account when deciding whether or
	726	// not to load history
	727	dc.runNetwork(net, firstDownstream)
	728	})
[57]	729
[144]	730	return nil
	731	}
	732
	733	// runNetwork starts listening for messages coming from the network's ring
	734	// buffer.
	735	//
	736	// It panics if the network is not suitable for the downstream connection.
	737	func (dc downstreamConn) runNetwork(net network, loadHistory bool) {
	738	if dc.network != nil && net != dc.network {
	739	panic("network not suitable for downstream connection")
	740	}
	741
	742	historyName := dc.rawUsername
	743
	744	var seqPtr *uint64
	745	if loadHistory {
	746	net.lock.Lock()
	747	seq, ok := net.history[historyName]
	748	net.lock.Unlock()
	749	if ok {
	750	seqPtr = &seq
[50]	751	}
[144]	752	}
[57]	753
[144]	754	consumer, ch := net.ring.NewConsumer(seqPtr)
	755	go func() {
	756	for {
	757	var closed bool
	758	select {
	759	case <-ch:
	760	uc := net.upstream()
	761	if uc == nil {
	762	dc.logger.Printf("ignoring messages for upstream %q: upstream is disconnected", net.Addr)
[57]	763	break
	764	}
[144]	765	dc.ringMessages <- ringMessage{consumer, uc}
	766	case <-dc.closed:
	767	closed = true
[57]	768	}
[144]	769	if closed {
	770	break
	771	}
	772	}
[57]	773
[144]	774	seq := consumer.Close()
[57]	775
[167]	776	net.lock.Lock()
	777	net.history[historyName] = seq
	778	net.lock.Unlock()
[144]	779	}()
[13]	780	}
	781
[103]	782	func (dc *downstreamConn) runUntilRegistered() error {
	783	for !dc.registered {
	784	msg, err := dc.irc.ReadMessage()
[106]	785	if err != nil {
[103]	786	return fmt.Errorf("failed to read IRC command: %v", err)
	787	}
	788
[110]	789	if dc.srv.Debug {
	790	dc.logger.Printf("received: %v", msg)
	791	}
	792
[103]	793	err = dc.handleMessage(msg)
	794	if ircErr, ok := err.(ircError); ok {
	795	ircErr.Message.Prefix = dc.srv.prefix()
	796	dc.SendMessage(ircErr.Message)
	797	} else if err != nil {
	798	return fmt.Errorf("failed to handle IRC command %q: %v", msg, err)
	799	}
	800	}
	801
	802	return nil
	803	}
	804
[55]	805	func (dc downstreamConn) handleMessageRegistered(msg irc.Message) error {
[13]	806	switch msg.Command {
[111]	807	case "CAP":
	808	var subCmd string
	809	if err := parseMessageParams(msg, &subCmd); err != nil {
	810	return err
	811	}
	812	if err := dc.handleCapCommand(subCmd, msg.Params[1:]); err != nil {
	813	return err
	814	}
[107]	815	case "PING":
	816	dc.SendMessage(&irc.Message{
	817	Prefix: dc.srv.prefix(),
	818	Command: "PONG",
	819	Params: msg.Params,
	820	})
	821	return nil
[42]	822	case "USER":
[13]	823	return ircError{&irc.Message{
	824	Command: irc.ERR_ALREADYREGISTERED,
[55]	825	Params: []string{dc.nick, "You may not reregister"},
[13]	826	}}
[42]	827	case "NICK":
[90]	828	var nick string
	829	if err := parseMessageParams(msg, &nick); err != nil {
	830	return err
	831	}
	832
	833	var err error
	834	dc.forEachNetwork(func(n *network) {
	835	if err != nil {
	836	return
	837	}
	838	n.Nick = nick
	839	err = dc.srv.db.StoreNetwork(dc.user.Username, &n.Network)
	840	})
	841	if err != nil {
	842	return err
	843	}
	844
[73]	845	dc.forEachUpstream(func(uc *upstreamConn) {
[60]	846	uc.SendMessage(msg)
[42]	847	})
[146]	848	case "JOIN":
	849	var namesStr string
	850	if err := parseMessageParams(msg, &namesStr); err != nil {
[48]	851	return err
	852	}
	853
[146]	854	var keys []string
	855	if len(msg.Params) > 1 {
	856	keys = strings.Split(msg.Params[1], ",")
	857	}
	858
	859	for i, name := range strings.Split(namesStr, ",") {
[145]	860	uc, upstreamName, err := dc.unmarshalEntity(name)
	861	if err != nil {
[158]	862	return err
[145]	863	}
[48]	864
[146]	865	var key string
	866	if len(keys) > i {
	867	key = keys[i]
	868	}
	869
	870	params := []string{upstreamName}
	871	if key != "" {
	872	params = append(params, key)
	873	}
[145]	874	uc.SendMessage(&irc.Message{
[146]	875	Command: "JOIN",
	876	Params: params,
[145]	877	})
[89]	878
[146]	879	err = dc.srv.db.StoreChannel(uc.network.ID, &Channel{
	880	Name: upstreamName,
	881	Key: key,
	882	})
	883	if err != nil {
	884	dc.logger.Printf("failed to create channel %q in DB: %v", upstreamName, err)
[89]	885	}
	886	}
[146]	887	case "PART":
	888	var namesStr string
	889	if err := parseMessageParams(msg, &namesStr); err != nil {
	890	return err
	891	}
	892
	893	var reason string
	894	if len(msg.Params) > 1 {
	895	reason = msg.Params[1]
	896	}
	897
	898	for _, name := range strings.Split(namesStr, ",") {
	899	uc, upstreamName, err := dc.unmarshalEntity(name)
	900	if err != nil {
[158]	901	return err
[146]	902	}
	903
	904	params := []string{upstreamName}
	905	if reason != "" {
	906	params = append(params, reason)
	907	}
	908	uc.SendMessage(&irc.Message{
	909	Command: "PART",
	910	Params: params,
	911	})
	912
	913	if err := dc.srv.db.DeleteChannel(uc.network.ID, upstreamName); err != nil {
	914	dc.logger.Printf("failed to delete channel %q in DB: %v", upstreamName, err)
	915	}
	916	}
[159]	917	case "KICK":
	918	var channelStr, userStr string
	919	if err := parseMessageParams(msg, &channelStr, &userStr); err != nil {
	920	return err
	921	}
	922
	923	channels := strings.Split(channelStr, ",")
	924	users := strings.Split(userStr, ",")
	925
	926	var reason string
	927	if len(msg.Params) > 2 {
	928	reason = msg.Params[2]
	929	}
	930
	931	if len(channels) != 1 && len(channels) != len(users) {
	932	return ircError{&irc.Message{
	933	Command: irc.ERR_BADCHANMASK,
	934	Params: []string{dc.nick, channelStr, "Bad channel mask"},
	935	}}
	936	}
	937
	938	for i, user := range users {
	939	var channel string
	940	if len(channels) == 1 {
	941	channel = channels[0]
	942	} else {
	943	channel = channels[i]
	944	}
	945
	946	ucChannel, upstreamChannel, err := dc.unmarshalEntity(channel)
	947	if err != nil {
	948	return err
	949	}
	950
	951	ucUser, upstreamUser, err := dc.unmarshalEntity(user)
	952	if err != nil {
	953	return err
	954	}
	955
	956	if ucChannel != ucUser {
	957	return ircError{&irc.Message{
	958	Command: irc.ERR_USERNOTINCHANNEL,
	959	Params: []string{dc.nick, user, channel, "They aren't on that channel"},
	960	}}
	961	}
	962	uc := ucChannel
	963
	964	params := []string{upstreamChannel, upstreamUser}
	965	if reason != "" {
	966	params = append(params, reason)
	967	}
	968	uc.SendMessage(&irc.Message{
	969	Command: "KICK",
	970	Params: params,
	971	})
	972	}
[69]	973	case "MODE":
[46]	974	var name string
	975	if err := parseMessageParams(msg, &name); err != nil {
	976	return err
	977	}
	978
	979	var modeStr string
	980	if len(msg.Params) > 1 {
	981	modeStr = msg.Params[1]
	982	}
	983
[139]	984	if name == dc.nick {
[46]	985	if modeStr != "" {
[73]	986	dc.forEachUpstream(func(uc *upstreamConn) {
[69]	987	uc.SendMessage(&irc.Message{
	988	Command: "MODE",
	989	Params: []string{uc.nick, modeStr},
	990	})
[46]	991	})
	992	} else {
[55]	993	dc.SendMessage(&irc.Message{
	994	Prefix: dc.srv.prefix(),
[46]	995	Command: irc.RPL_UMODEIS,
[129]	996	Params: []string{dc.nick, ""}, // TODO
[54]	997	})
[46]	998	}
[139]	999	return nil
[46]	1000	}
[139]	1001
	1002	uc, upstreamName, err := dc.unmarshalEntity(name)
	1003	if err != nil {
	1004	return err
	1005	}
	1006
	1007	if !uc.isChannel(upstreamName) {
	1008	return ircError{&irc.Message{
	1009	Command: irc.ERR_USERSDONTMATCH,
	1010	Params: []string{dc.nick, "Cannot change mode for other users"},
	1011	}}
	1012	}
	1013
	1014	if modeStr != "" {
	1015	params := []string{upstreamName, modeStr}
	1016	params = append(params, msg.Params[2:]...)
	1017	uc.SendMessage(&irc.Message{
	1018	Command: "MODE",
	1019	Params: params,
	1020	})
	1021	} else {
	1022	ch, ok := uc.channels[upstreamName]
	1023	if !ok {
	1024	return ircError{&irc.Message{
	1025	Command: irc.ERR_NOSUCHCHANNEL,
	1026	Params: []string{dc.nick, name, "No such channel"},
	1027	}}
	1028	}
	1029
	1030	if ch.modes == nil {
	1031	// we haven't received the initial RPL_CHANNELMODEIS yet
	1032	// ignore the request, we will broadcast the modes later when we receive RPL_CHANNELMODEIS
	1033	return nil
	1034	}
	1035
	1036	modeStr, modeParams := ch.modes.Format()
	1037	params := []string{dc.nick, name, modeStr}
	1038	params = append(params, modeParams...)
	1039
	1040	dc.SendMessage(&irc.Message{
	1041	Prefix: dc.srv.prefix(),
	1042	Command: irc.RPL_CHANNELMODEIS,
	1043	Params: params,
	1044	})
[162]	1045	if ch.creationTime != "" {
	1046	dc.SendMessage(&irc.Message{
	1047	Prefix: dc.srv.prefix(),
	1048	Command: rpl_creationtime,
	1049	Params: []string{dc.nick, name, ch.creationTime},
	1050	})
	1051	}
[139]	1052	}
[160]	1053	case "TOPIC":
	1054	var channel string
	1055	if err := parseMessageParams(msg, &channel); err != nil {
	1056	return err
	1057	}
	1058
	1059	uc, upstreamChannel, err := dc.unmarshalEntity(channel)
	1060	if err != nil {
	1061	return err
	1062	}
	1063
	1064	if len(msg.Params) > 1 { // setting topic
	1065	topic := msg.Params[1]
	1066	uc.SendMessage(&irc.Message{
	1067	Command: "TOPIC",
	1068	Params: []string{upstreamChannel, topic},
	1069	})
	1070	} else { // getting topic
	1071	ch, ok := uc.channels[upstreamChannel]
	1072	if !ok {
	1073	return ircError{&irc.Message{
	1074	Command: irc.ERR_NOSUCHCHANNEL,
	1075	Params: []string{dc.nick, upstreamChannel, "No such channel"},
	1076	}}
	1077	}
	1078	sendTopic(dc, ch)
	1079	}
[177]	1080	case "LIST":
	1081	// TODO: support ELIST when supported by all upstreams
	1082
	1083	pl := pendingLIST{
	1084	downstreamID: dc.id,
	1085	pendingCommands: make(map[int64]*irc.Message),
	1086	}
	1087	var upstreamChannels map[int64][]string
	1088	if len(msg.Params) > 0 {
	1089	upstreamChannels = make(map[int64][]string)
	1090	channels := strings.Split(msg.Params[0], ",")
	1091	for _, channel := range channels {
	1092	uc, upstreamChannel, err := dc.unmarshalEntity(channel)
	1093	if err != nil {
	1094	return err
	1095	}
	1096	upstreamChannels[uc.network.ID] = append(upstreamChannels[uc.network.ID], upstreamChannel)
	1097	}
	1098	}
	1099
	1100	dc.user.pendingLISTs = append(dc.user.pendingLISTs, pl)
	1101	dc.forEachUpstream(func(uc *upstreamConn) {
	1102	var params []string
	1103	if upstreamChannels != nil {
	1104	if channels, ok := upstreamChannels[uc.network.ID]; ok {
	1105	params = []string{strings.Join(channels, ",")}
	1106	} else {
	1107	return
	1108	}
	1109	}
	1110	pl.pendingCommands[uc.network.ID] = &irc.Message{
	1111	Command: "LIST",
	1112	Params: params,
	1113	}
	1114	uc.trySendList(dc.id)
	1115	})
[140]	1116	case "NAMES":
	1117	if len(msg.Params) == 0 {
	1118	dc.SendMessage(&irc.Message{
	1119	Prefix: dc.srv.prefix(),
	1120	Command: irc.RPL_ENDOFNAMES,
	1121	Params: []string{dc.nick, "*", "End of /NAMES list"},
	1122	})
	1123	return nil
	1124	}
	1125
	1126	channels := strings.Split(msg.Params[0], ",")
	1127	for _, channel := range channels {
	1128	uc, upstreamChannel, err := dc.unmarshalEntity(channel)
	1129	if err != nil {
	1130	return err
	1131	}
	1132
	1133	ch, ok := uc.channels[upstreamChannel]
	1134	if ok {
	1135	sendNames(dc, ch)
	1136	} else {
	1137	// NAMES on a channel we have not joined, ask upstream
[176]	1138	uc.SendMessageLabeled(dc.id, &irc.Message{
[140]	1139	Command: "NAMES",
	1140	Params: []string{upstreamChannel},
	1141	})
	1142	}
	1143	}
[127]	1144	case "WHO":
	1145	if len(msg.Params) == 0 {
	1146	// TODO: support WHO without parameters
	1147	dc.SendMessage(&irc.Message{
	1148	Prefix: dc.srv.prefix(),
	1149	Command: irc.RPL_ENDOFWHO,
[140]	1150	Params: []string{dc.nick, "*", "End of /WHO list"},
[127]	1151	})
	1152	return nil
	1153	}
	1154
	1155	// TODO: support WHO masks
	1156	entity := msg.Params[0]
	1157
[142]	1158	if entity == dc.nick {
	1159	// TODO: support AWAY (H/G) in self WHO reply
	1160	dc.SendMessage(&irc.Message{
	1161	Prefix: dc.srv.prefix(),
	1162	Command: irc.RPL_WHOREPLY,
	1163	Params: []string{dc.nick, "*", dc.username, dc.hostname, dc.srv.Hostname, dc.nick, "H", "0 " + dc.realname},
	1164	})
	1165	dc.SendMessage(&irc.Message{
	1166	Prefix: dc.srv.prefix(),
	1167	Command: irc.RPL_ENDOFWHO,
	1168	Params: []string{dc.nick, dc.nick, "End of /WHO list"},
	1169	})
	1170	return nil
	1171	}
	1172
[127]	1173	uc, upstreamName, err := dc.unmarshalEntity(entity)
	1174	if err != nil {
	1175	return err
	1176	}
	1177
	1178	var params []string
	1179	if len(msg.Params) == 2 {
	1180	params = []string{upstreamName, msg.Params[1]}
	1181	} else {
	1182	params = []string{upstreamName}
	1183	}
	1184
[176]	1185	uc.SendMessageLabeled(dc.id, &irc.Message{
[127]	1186	Command: "WHO",
	1187	Params: params,
	1188	})
[128]	1189	case "WHOIS":
	1190	if len(msg.Params) == 0 {
	1191	return ircError{&irc.Message{
	1192	Command: irc.ERR_NONICKNAMEGIVEN,
	1193	Params: []string{dc.nick, "No nickname given"},
	1194	}}
	1195	}
	1196
	1197	var target, mask string
	1198	if len(msg.Params) == 1 {
	1199	target = ""
	1200	mask = msg.Params[0]
	1201	} else {
	1202	target = msg.Params[0]
	1203	mask = msg.Params[1]
	1204	}
	1205	// TODO: support multiple WHOIS users
	1206	if i := strings.IndexByte(mask, ','); i >= 0 {
	1207	mask = mask[:i]
	1208	}
	1209
[142]	1210	if mask == dc.nick {
	1211	dc.SendMessage(&irc.Message{
	1212	Prefix: dc.srv.prefix(),
	1213	Command: irc.RPL_WHOISUSER,
	1214	Params: []string{dc.nick, dc.nick, dc.username, dc.hostname, "*", dc.realname},
	1215	})
	1216	dc.SendMessage(&irc.Message{
	1217	Prefix: dc.srv.prefix(),
	1218	Command: irc.RPL_WHOISSERVER,
	1219	Params: []string{dc.nick, dc.nick, dc.srv.Hostname, "soju"},
	1220	})
	1221	dc.SendMessage(&irc.Message{
	1222	Prefix: dc.srv.prefix(),
	1223	Command: irc.RPL_ENDOFWHOIS,
	1224	Params: []string{dc.nick, dc.nick, "End of /WHOIS list"},
	1225	})
	1226	return nil
	1227	}
	1228
[128]	1229	// TODO: support WHOIS masks
	1230	uc, upstreamNick, err := dc.unmarshalEntity(mask)
	1231	if err != nil {
	1232	return err
	1233	}
	1234
	1235	var params []string
	1236	if target != "" {
	1237	params = []string{target, upstreamNick}
	1238	} else {
	1239	params = []string{upstreamNick}
	1240	}
	1241
[176]	1242	uc.SendMessageLabeled(dc.id, &irc.Message{
[128]	1243	Command: "WHOIS",
	1244	Params: params,
	1245	})
[58]	1246	case "PRIVMSG":
	1247	var targetsStr, text string
	1248	if err := parseMessageParams(msg, &targetsStr, &text); err != nil {
	1249	return err
	1250	}
	1251
	1252	for _, name := range strings.Split(targetsStr, ",") {
[117]	1253	if name == serviceNick {
	1254	handleServicePRIVMSG(dc, text)
	1255	continue
	1256	}
	1257
[127]	1258	uc, upstreamName, err := dc.unmarshalEntity(name)
[58]	1259	if err != nil {
	1260	return err
	1261	}
	1262
[95]	1263	if upstreamName == "NickServ" {
	1264	dc.handleNickServPRIVMSG(uc, text)
	1265	}
	1266
[69]	1267	uc.SendMessage(&irc.Message{
[58]	1268	Command: "PRIVMSG",
[69]	1269	Params: []string{upstreamName, text},
[60]	1270	})
[105]	1271
[113]	1272	echoMsg := &irc.Message{
	1273	Prefix: &irc.Prefix{
	1274	Name: uc.nick,
	1275	User: uc.username,
	1276	},
[114]	1277	Command: "PRIVMSG",
[113]	1278	Params: []string{upstreamName, text},
	1279	}
[105]	1280	dc.lock.Lock()
[113]	1281	dc.ourMessages[echoMsg] = struct{}{}
[105]	1282	dc.lock.Unlock()
	1283
[143]	1284	uc.network.ring.Produce(echoMsg)
[58]	1285	}
[164]	1286	case "NOTICE":
	1287	var targetsStr, text string
	1288	if err := parseMessageParams(msg, &targetsStr, &text); err != nil {
	1289	return err
	1290	}
	1291
	1292	for _, name := range strings.Split(targetsStr, ",") {
	1293	uc, upstreamName, err := dc.unmarshalEntity(name)
	1294	if err != nil {
	1295	return err
	1296	}
	1297
	1298	uc.SendMessage(&irc.Message{
	1299	Command: "NOTICE",
	1300	Params: []string{upstreamName, text},
	1301	})
	1302	}
[163]	1303	case "INVITE":
	1304	var user, channel string
	1305	if err := parseMessageParams(msg, &user, &channel); err != nil {
	1306	return err
	1307	}
	1308
	1309	ucChannel, upstreamChannel, err := dc.unmarshalEntity(channel)
	1310	if err != nil {
	1311	return err
	1312	}
	1313
	1314	ucUser, upstreamUser, err := dc.unmarshalEntity(user)
	1315	if err != nil {
	1316	return err
	1317	}
	1318
	1319	if ucChannel != ucUser {
	1320	return ircError{&irc.Message{
	1321	Command: irc.ERR_USERNOTINCHANNEL,
	1322	Params: []string{dc.nick, user, channel, "They aren't on that channel"},
	1323	}}
	1324	}
	1325	uc := ucChannel
	1326
[176]	1327	uc.SendMessageLabeled(dc.id, &irc.Message{
[163]	1328	Command: "INVITE",
	1329	Params: []string{upstreamUser, upstreamChannel},
	1330	})
[13]	1331	default:
[55]	1332	dc.logger.Printf("unhandled message: %v", msg)
[13]	1333	return newUnknownCommandError(msg.Command)
	1334	}
[42]	1335	return nil
[13]	1336	}
[95]	1337
	1338	func (dc downstreamConn) handleNickServPRIVMSG(uc upstreamConn, text string) {
	1339	username, password, ok := parseNickServCredentials(text, uc.nick)
	1340	if !ok {
	1341	return
	1342	}
	1343
	1344	dc.logger.Printf("auto-saving NickServ credentials with username %q", username)
	1345	n := uc.network
	1346	n.SASL.Mechanism = "PLAIN"
	1347	n.SASL.Plain.Username = username
	1348	n.SASL.Plain.Password = password
	1349	if err := dc.srv.db.StoreNetwork(dc.user.Username, &n.Network); err != nil {
	1350	dc.logger.Printf("failed to save NickServ credentials: %v", err)
	1351	}
	1352	}
	1353
	1354	func parseNickServCredentials(text, nick string) (username, password string, ok bool) {
	1355	fields := strings.Fields(text)
	1356	if len(fields) < 2 {
	1357	return "", "", false
	1358	}
	1359	cmd := strings.ToUpper(fields[0])
	1360	params := fields[1:]
	1361	switch cmd {
	1362	case "REGISTER":
	1363	username = nick
	1364	password = params[0]
	1365	case "IDENTIFY":
	1366	if len(params) == 1 {
	1367	username = nick
	1368	} else {
	1369	username = params[0]
	1370	}
	1371	password = params[1]
	1372	}
	1373	return username, password, true
	1374	}

Note: See TracBrowser for help on using the repository browser.

Download in other formats: