Context Navigation

downstream.go@ 166

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

Add eventDownstreamConnected

In a later commit, we'll be able to move part of downstreamConn.register
into the user goroutine to prevent races.

References: https://todo.sr.ht/~emersion/soju/22

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

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

Context Navigation

source: code/trunk/downstream.go@ 166

Download in other formats: