Context Navigation

source: code/trunk/downstream.go@ 202

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

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

Download in other formats: