Context Navigation

source: code/trunk/downstream.go@ 204

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

Simplify ring consumer goroutine

Since network.history is now only accessed from the user goroutine, a
lock becomes unnecessary.

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

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

Download in other formats: