Context Navigation

source: code/trunk/downstream.go@ 98

Last change on this file since 98 was 98, checked in by contact, 5 years ago
Rename project to soju
File size: 14.6 KB

Rev	Line
[98]	1	package soju
[13]	2
	3	import (
[91]	4	"crypto/tls"
[13]	5	"fmt"
	6	"io"
	7	"net"
[39]	8	"strings"
[91]	9	"time"
[13]	10
[85]	11	"golang.org/x/crypto/bcrypt"
[13]	12	"gopkg.in/irc.v3"
	13	)
	14
	15	type ircError struct {
	16	Message *irc.Message
	17	}
	18
[85]	19	func (err ircError) Error() string {
	20	return err.Message.String()
	21	}
	22
[13]	23	func newUnknownCommandError(cmd string) ircError {
	24	return ircError{&irc.Message{
	25	Command: irc.ERR_UNKNOWNCOMMAND,
	26	Params: []string{
	27	"*",
	28	cmd,
	29	"Unknown command",
	30	},
	31	}}
	32	}
	33
	34	func newNeedMoreParamsError(cmd string) ircError {
	35	return ircError{&irc.Message{
	36	Command: irc.ERR_NEEDMOREPARAMS,
	37	Params: []string{
	38	"*",
	39	cmd,
	40	"Not enough parameters",
	41	},
	42	}}
	43	}
	44
[85]	45	var errAuthFailed = ircError{&irc.Message{
	46	Command: irc.ERR_PASSWDMISMATCH,
	47	Params: []string{"*", "Invalid username or password"},
	48	}}
[13]	49
[69]	50	type consumption struct {
	51	consumer *RingConsumer
	52	upstreamConn *upstreamConn
	53	}
	54
[13]	55	type downstreamConn struct {
[69]	56	net net.Conn
	57	irc *irc.Conn
	58	srv *Server
	59	logger Logger
	60	messages chan *irc.Message
	61	consumptions chan consumption
	62	closed chan struct{}
[22]	63
[13]	64	registered bool
[37]	65	user *user
[13]	66	nick string
	67	username string
	68	realname string
[85]	69	password string // empty after authentication
[77]	70	network *network // can be nil
[13]	71	}
	72
[22]	73	func newDownstreamConn(srv Server, netConn net.Conn) downstreamConn {
[55]	74	dc := &downstreamConn{
[69]	75	net: netConn,
	76	irc: irc.NewConn(netConn),
	77	srv: srv,
	78	logger: &prefixLogger{srv.Logger, fmt.Sprintf("downstream %q: ", netConn.RemoteAddr())},
	79	messages: make(chan *irc.Message, 64),
	80	consumptions: make(chan consumption),
	81	closed: make(chan struct{}),
[22]	82	}
[26]	83
	84	go func() {
[56]	85	if err := dc.writeMessages(); err != nil {
	86	dc.logger.Printf("failed to write message: %v", err)
[26]	87	}
[55]	88	if err := dc.net.Close(); err != nil {
	89	dc.logger.Printf("failed to close connection: %v", err)
[45]	90	} else {
[55]	91	dc.logger.Printf("connection closed")
[45]	92	}
[26]	93	}()
	94
[55]	95	return dc
[22]	96	}
	97
[55]	98	func (dc downstreamConn) prefix() irc.Prefix {
[27]	99	return &irc.Prefix{
[55]	100	Name: dc.nick,
	101	User: dc.username,
[27]	102	// TODO: fill the host?
	103	}
	104	}
	105
[69]	106	func (dc downstreamConn) marshalChannel(uc upstreamConn, name string) string {
	107	return name
	108	}
	109
[90]	110	func (dc downstreamConn) forEachNetwork(f func(network)) {
	111	if dc.network != nil {
	112	f(dc.network)
	113	} else {
	114	dc.user.forEachNetwork(f)
	115	}
	116	}
	117
[73]	118	func (dc downstreamConn) forEachUpstream(f func(upstreamConn)) {
	119	dc.user.forEachUpstream(func(uc *upstreamConn) {
[77]	120	if dc.network != nil && uc.network != dc.network {
[73]	121	return
	122	}
	123	f(uc)
	124	})
	125	}
	126
[89]	127	// upstream returns the upstream connection, if any. If there are zero or if
	128	// there are multiple upstream connections, it returns nil.
	129	func (dc downstreamConn) upstream() upstreamConn {
	130	if dc.network == nil {
	131	return nil
	132	}
	133
	134	var upstream *upstreamConn
	135	dc.forEachUpstream(func(uc *upstreamConn) {
	136	upstream = uc
	137	})
	138	return upstream
	139	}
	140
[69]	141	func (dc downstreamConn) unmarshalChannel(name string) (upstreamConn, string, error) {
[89]	142	if uc := dc.upstream(); uc != nil {
	143	return uc, name, nil
	144	}
	145
[73]	146	// TODO: extract network name from channel name if dc.upstream == nil
	147	var channel *upstreamChannel
	148	var err error
	149	dc.forEachUpstream(func(uc *upstreamConn) {
	150	if err != nil {
	151	return
	152	}
	153	if ch, ok := uc.channels[name]; ok {
	154	if channel != nil {
	155	err = fmt.Errorf("ambiguous channel name %q", name)
	156	} else {
	157	channel = ch
	158	}
	159	}
	160	})
	161	if channel == nil {
	162	return nil, "", ircError{&irc.Message{
	163	Command: irc.ERR_NOSUCHCHANNEL,
	164	Params: []string{name, "No such channel"},
	165	}}
[69]	166	}
[73]	167	return channel.conn, channel.Name, nil
[69]	168	}
	169
	170	func (dc downstreamConn) marshalNick(uc upstreamConn, nick string) string {
	171	if nick == uc.nick {
	172	return dc.nick
	173	}
	174	return nick
	175	}
	176
	177	func (dc downstreamConn) marshalUserPrefix(uc upstreamConn, prefix irc.Prefix) irc.Prefix {
	178	if prefix.Name == uc.nick {
	179	return dc.prefix()
	180	}
	181	return prefix
	182	}
	183
[57]	184	func (dc *downstreamConn) isClosed() bool {
	185	select {
	186	case <-dc.closed:
	187	return true
	188	default:
	189	return false
	190	}
	191	}
	192
[55]	193	func (dc *downstreamConn) readMessages() error {
	194	dc.logger.Printf("new connection")
[22]	195
	196	for {
[55]	197	msg, err := dc.irc.ReadMessage()
[22]	198	if err == io.EOF {
	199	break
	200	} else if err != nil {
	201	return fmt.Errorf("failed to read IRC command: %v", err)
	202	}
	203
[64]	204	if dc.srv.Debug {
	205	dc.logger.Printf("received: %v", msg)
	206	}
	207
[55]	208	err = dc.handleMessage(msg)
[22]	209	if ircErr, ok := err.(ircError); ok {
[55]	210	ircErr.Message.Prefix = dc.srv.prefix()
	211	dc.SendMessage(ircErr.Message)
[22]	212	} else if err != nil {
	213	return fmt.Errorf("failed to handle IRC command %q: %v", msg.Command, err)
	214	}
	215
[57]	216	if dc.isClosed() {
[22]	217	return nil
	218	}
	219	}
	220
[45]	221	return nil
[22]	222	}
	223
[56]	224	func (dc *downstreamConn) writeMessages() error {
[57]	225	for {
	226	var err error
	227	var closed bool
	228	select {
	229	case msg := <-dc.messages:
[64]	230	if dc.srv.Debug {
	231	dc.logger.Printf("sent: %v", msg)
	232	}
[57]	233	err = dc.irc.WriteMessage(msg)
[69]	234	case consumption := <-dc.consumptions:
	235	consumer, uc := consumption.consumer, consumption.upstreamConn
[57]	236	for {
	237	msg := consumer.Peek()
	238	if msg == nil {
	239	break
	240	}
[69]	241	msg = msg.Copy()
	242	switch msg.Command {
	243	case "PRIVMSG":
	244	// TODO: detect whether it's a user or a channel
	245	msg.Params[0] = dc.marshalChannel(uc, msg.Params[0])
	246	default:
	247	panic("expected to consume a PRIVMSG message")
	248	}
[64]	249	if dc.srv.Debug {
	250	dc.logger.Printf("sent: %v", msg)
	251	}
[57]	252	err = dc.irc.WriteMessage(msg)
	253	if err != nil {
	254	break
	255	}
	256	consumer.Consume()
	257	}
	258	case <-dc.closed:
	259	closed = true
	260	}
	261	if err != nil {
[56]	262	return err
	263	}
[57]	264	if closed {
	265	break
	266	}
[56]	267	}
	268	return nil
	269	}
	270
[55]	271	func (dc *downstreamConn) Close() error {
[57]	272	if dc.isClosed() {
[26]	273	return fmt.Errorf("downstream connection already closed")
	274	}
[40]	275
[55]	276	if u := dc.user; u != nil {
[40]	277	u.lock.Lock()
	278	for i := range u.downstreamConns {
[55]	279	if u.downstreamConns[i] == dc {
[40]	280	u.downstreamConns = append(u.downstreamConns[:i], u.downstreamConns[i+1:]...)
[63]	281	break
[40]	282	}
	283	}
	284	u.lock.Unlock()
[13]	285	}
[40]	286
[57]	287	close(dc.closed)
[45]	288	return nil
[13]	289	}
	290
[55]	291	func (dc downstreamConn) SendMessage(msg irc.Message) {
	292	dc.messages <- msg
[54]	293	}
	294
[55]	295	func (dc downstreamConn) handleMessage(msg irc.Message) error {
[13]	296	switch msg.Command {
[28]	297	case "QUIT":
[55]	298	return dc.Close()
[13]	299	case "PING":
[55]	300	dc.SendMessage(&irc.Message{
	301	Prefix: dc.srv.prefix(),
[13]	302	Command: "PONG",
[68]	303	Params: msg.Params,
[54]	304	})
[26]	305	return nil
[13]	306	default:
[55]	307	if dc.registered {
	308	return dc.handleMessageRegistered(msg)
[13]	309	} else {
[55]	310	return dc.handleMessageUnregistered(msg)
[13]	311	}
	312	}
	313	}
	314
[55]	315	func (dc downstreamConn) handleMessageUnregistered(msg irc.Message) error {
[13]	316	switch msg.Command {
	317	case "NICK":
[55]	318	if err := parseMessageParams(msg, &dc.nick); err != nil {
[43]	319	return err
[13]	320	}
	321	case "USER":
[43]	322	var username string
[55]	323	if err := parseMessageParams(msg, &username, nil, nil, &dc.realname); err != nil {
[43]	324	return err
[13]	325	}
[55]	326	dc.username = "~" + username
[85]	327	case "PASS":
	328	if err := parseMessageParams(msg, &dc.password); err != nil {
	329	return err
	330	}
[13]	331	default:
[55]	332	dc.logger.Printf("unhandled message: %v", msg)
[13]	333	return newUnknownCommandError(msg.Command)
	334	}
[55]	335	if dc.username != "" && dc.nick != "" {
	336	return dc.register()
[13]	337	}
	338	return nil
	339	}
	340
[91]	341	func sanityCheckServer(addr string) error {
	342	dialer := net.Dialer{Timeout: 30 * time.Second}
	343	conn, err := tls.DialWithDialer(&dialer, "tcp", addr, nil)
	344	if err != nil {
	345	return err
	346	}
	347	return conn.Close()
	348	}
	349
[55]	350	func (dc *downstreamConn) register() error {
[73]	351	username := strings.TrimPrefix(dc.username, "~")
[77]	352	var networkName string
[73]	353	if i := strings.LastIndexAny(username, "/@"); i >= 0 {
[77]	354	networkName = username[i+1:]
[73]	355	}
	356	if i := strings.IndexAny(username, "/@"); i >= 0 {
	357	username = username[:i]
	358	}
	359
[85]	360	password := dc.password
	361	dc.password = ""
	362
[73]	363	u := dc.srv.getUser(username)
[38]	364	if u == nil {
[85]	365	dc.logger.Printf("failed authentication for %q: unknown username", username)
	366	return errAuthFailed
[37]	367	}
	368
[85]	369	err := bcrypt.CompareHashAndPassword([]byte(u.Password), []byte(password))
	370	if err != nil {
	371	dc.logger.Printf("failed authentication for %q: %v", username, err)
	372	return errAuthFailed
	373	}
	374
[88]	375	var network *network
[77]	376	if networkName != "" {
[88]	377	network = u.getNetwork(networkName)
	378	if network == nil {
[91]	379	addr := networkName
	380	if !strings.ContainsRune(addr, ':') {
	381	addr = addr + ":6697"
	382	}
	383
[95]	384	dc.logger.Printf("trying to connect to new network %q", addr)
[91]	385	if err := sanityCheckServer(addr); err != nil {
	386	dc.logger.Printf("failed to connect to %q: %v", addr, err)
	387	return ircError{&irc.Message{
	388	Command: irc.ERR_PASSWDMISMATCH,
	389	Params: []string{"*", fmt.Sprintf("Failed to connect to %q", networkName)},
	390	}}
	391	}
	392
[95]	393	dc.logger.Printf("auto-saving network %q", networkName)
[91]	394	network, err = u.createNetwork(networkName, dc.nick)
	395	if err != nil {
	396	return err
	397	}
[73]	398	}
	399	}
	400
[55]	401	dc.registered = true
	402	dc.user = u
[88]	403	dc.network = network
[13]	404
[40]	405	u.lock.Lock()
[57]	406	firstDownstream := len(u.downstreamConns) == 0
[55]	407	u.downstreamConns = append(u.downstreamConns, dc)
[40]	408	u.lock.Unlock()
	409
[55]	410	dc.SendMessage(&irc.Message{
	411	Prefix: dc.srv.prefix(),
[13]	412	Command: irc.RPL_WELCOME,
[98]	413	Params: []string{dc.nick, "Welcome to soju, " + dc.nick},
[54]	414	})
[55]	415	dc.SendMessage(&irc.Message{
	416	Prefix: dc.srv.prefix(),
[13]	417	Command: irc.RPL_YOURHOST,
[55]	418	Params: []string{dc.nick, "Your host is " + dc.srv.Hostname},
[54]	419	})
[55]	420	dc.SendMessage(&irc.Message{
	421	Prefix: dc.srv.prefix(),
[13]	422	Command: irc.RPL_CREATED,
[55]	423	Params: []string{dc.nick, "Who cares when the server was created?"},
[54]	424	})
[55]	425	dc.SendMessage(&irc.Message{
	426	Prefix: dc.srv.prefix(),
[13]	427	Command: irc.RPL_MYINFO,
[98]	428	Params: []string{dc.nick, dc.srv.Hostname, "soju", "aiwroO", "OovaimnqpsrtklbeI"},
[54]	429	})
[93]	430	// TODO: RPL_ISUPPORT
[55]	431	dc.SendMessage(&irc.Message{
	432	Prefix: dc.srv.prefix(),
[13]	433	Command: irc.ERR_NOMOTD,
[55]	434	Params: []string{dc.nick, "No MOTD"},
[54]	435	})
[13]	436
[73]	437	dc.forEachUpstream(func(uc *upstreamConn) {
[30]	438	// TODO: fix races accessing upstream connection data
	439	for _, ch := range uc.channels {
	440	if ch.complete {
[55]	441	forwardChannel(dc, ch)
[30]	442	}
	443	}
[50]	444
[73]	445	historyName := dc.username
[57]	446
	447	var seqPtr *uint64
	448	if firstDownstream {
	449	seq, ok := uc.history[historyName]
	450	if ok {
	451	seqPtr = &seq
[50]	452	}
	453	}
[57]	454
[59]	455	consumer, ch := uc.ring.NewConsumer(seqPtr)
[57]	456	go func() {
	457	for {
	458	var closed bool
	459	select {
	460	case <-ch:
[69]	461	dc.consumptions <- consumption{consumer, uc}
[57]	462	case <-dc.closed:
	463	closed = true
	464	}
	465	if closed {
	466	break
	467	}
	468	}
	469
	470	seq := consumer.Close()
	471
	472	dc.user.lock.Lock()
	473	lastDownstream := len(dc.user.downstreamConns) == 0
	474	dc.user.lock.Unlock()
	475
	476	if lastDownstream {
	477	uc.history[historyName] = seq
	478	}
	479	}()
[39]	480	})
[50]	481
[13]	482	return nil
	483	}
	484
[55]	485	func (dc downstreamConn) handleMessageRegistered(msg irc.Message) error {
[13]	486	switch msg.Command {
[42]	487	case "USER":
[13]	488	return ircError{&irc.Message{
	489	Command: irc.ERR_ALREADYREGISTERED,
[55]	490	Params: []string{dc.nick, "You may not reregister"},
[13]	491	}}
[42]	492	case "NICK":
[90]	493	var nick string
	494	if err := parseMessageParams(msg, &nick); err != nil {
	495	return err
	496	}
	497
	498	var err error
	499	dc.forEachNetwork(func(n *network) {
	500	if err != nil {
	501	return
	502	}
	503	n.Nick = nick
	504	err = dc.srv.db.StoreNetwork(dc.user.Username, &n.Network)
	505	})
	506	if err != nil {
	507	return err
	508	}
	509
[73]	510	dc.forEachUpstream(func(uc *upstreamConn) {
[60]	511	uc.SendMessage(msg)
[42]	512	})
[69]	513	case "JOIN", "PART":
[48]	514	var name string
	515	if err := parseMessageParams(msg, &name); err != nil {
	516	return err
	517	}
	518
[69]	519	uc, upstreamName, err := dc.unmarshalChannel(name)
	520	if err != nil {
	521	return ircError{&irc.Message{
	522	Command: irc.ERR_NOSUCHCHANNEL,
	523	Params: []string{name, err.Error()},
	524	}}
[48]	525	}
	526
[69]	527	uc.SendMessage(&irc.Message{
	528	Command: msg.Command,
	529	Params: []string{upstreamName},
	530	})
[89]	531
	532	switch msg.Command {
	533	case "JOIN":
	534	err := dc.srv.db.StoreChannel(uc.network.ID, &Channel{
	535	Name: upstreamName,
	536	})
	537	if err != nil {
	538	dc.logger.Printf("failed to create channel %q in DB: %v", upstreamName, err)
	539	}
	540	case "PART":
	541	if err := dc.srv.db.DeleteChannel(uc.network.ID, upstreamName); err != nil {
	542	dc.logger.Printf("failed to delete channel %q in DB: %v", upstreamName, err)
	543	}
	544	}
[69]	545	case "MODE":
	546	if msg.Prefix == nil {
	547	return fmt.Errorf("missing prefix")
[49]	548	}
	549
[46]	550	var name string
	551	if err := parseMessageParams(msg, &name); err != nil {
	552	return err
	553	}
	554
	555	var modeStr string
	556	if len(msg.Params) > 1 {
	557	modeStr = msg.Params[1]
	558	}
	559
	560	if msg.Prefix.Name != name {
[69]	561	uc, upstreamName, err := dc.unmarshalChannel(name)
[46]	562	if err != nil {
	563	return err
	564	}
	565
	566	if modeStr != "" {
[69]	567	uc.SendMessage(&irc.Message{
	568	Command: "MODE",
	569	Params: []string{upstreamName, modeStr},
	570	})
[46]	571	} else {
[69]	572	ch, ok := uc.channels[upstreamName]
	573	if !ok {
	574	return ircError{&irc.Message{
	575	Command: irc.ERR_NOSUCHCHANNEL,
	576	Params: []string{name, "No such channel"},
	577	}}
	578	}
	579
[55]	580	dc.SendMessage(&irc.Message{
	581	Prefix: dc.srv.prefix(),
[46]	582	Command: irc.RPL_CHANNELMODEIS,
[69]	583	Params: []string{name, string(ch.modes)},
[54]	584	})
[46]	585	}
	586	} else {
[55]	587	if name != dc.nick {
[46]	588	return ircError{&irc.Message{
	589	Command: irc.ERR_USERSDONTMATCH,
[55]	590	Params: []string{dc.nick, "Cannot change mode for other users"},
[46]	591	}}
	592	}
	593
	594	if modeStr != "" {
[73]	595	dc.forEachUpstream(func(uc *upstreamConn) {
[69]	596	uc.SendMessage(&irc.Message{
	597	Command: "MODE",
	598	Params: []string{uc.nick, modeStr},
	599	})
[46]	600	})
	601	} else {
[55]	602	dc.SendMessage(&irc.Message{
	603	Prefix: dc.srv.prefix(),
[46]	604	Command: irc.RPL_UMODEIS,
	605	Params: []string{""}, // TODO
[54]	606	})
[46]	607	}
	608	}
[58]	609	case "PRIVMSG":
	610	var targetsStr, text string
	611	if err := parseMessageParams(msg, &targetsStr, &text); err != nil {
	612	return err
	613	}
	614
	615	for _, name := range strings.Split(targetsStr, ",") {
[69]	616	uc, upstreamName, err := dc.unmarshalChannel(name)
[58]	617	if err != nil {
	618	return err
	619	}
	620
[95]	621	if upstreamName == "NickServ" {
	622	dc.handleNickServPRIVMSG(uc, text)
	623	}
	624
[69]	625	uc.SendMessage(&irc.Message{
[58]	626	Command: "PRIVMSG",
[69]	627	Params: []string{upstreamName, text},
[60]	628	})
[58]	629	}
[13]	630	default:
[55]	631	dc.logger.Printf("unhandled message: %v", msg)
[13]	632	return newUnknownCommandError(msg.Command)
	633	}
[42]	634	return nil
[13]	635	}
[95]	636
	637	func (dc downstreamConn) handleNickServPRIVMSG(uc upstreamConn, text string) {
	638	username, password, ok := parseNickServCredentials(text, uc.nick)
	639	if !ok {
	640	return
	641	}
	642
	643	dc.logger.Printf("auto-saving NickServ credentials with username %q", username)
	644	n := uc.network
	645	n.SASL.Mechanism = "PLAIN"
	646	n.SASL.Plain.Username = username
	647	n.SASL.Plain.Password = password
	648	if err := dc.srv.db.StoreNetwork(dc.user.Username, &n.Network); err != nil {
	649	dc.logger.Printf("failed to save NickServ credentials: %v", err)
	650	}
	651	}
	652
	653	func parseNickServCredentials(text, nick string) (username, password string, ok bool) {
	654	fields := strings.Fields(text)
	655	if len(fields) < 2 {
	656	return "", "", false
	657	}
	658	cmd := strings.ToUpper(fields[0])
	659	params := fields[1:]
	660	switch cmd {
	661	case "REGISTER":
	662	username = nick
	663	password = params[0]
	664	case "IDENTIFY":
	665	if len(params) == 1 {
	666	username = nick
	667	} else {
	668	username = params[0]
	669	}
	670	password = params[1]
	671	}
	672	return username, password, true
	673	}

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

Download in other formats: