source: code/trunk/upstream.go@ 82

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

Pass-through NICK messages for other users
File size: 9.9 KB
Line 
1package jounce
2
3import (
4 "crypto/tls"
5 "fmt"
6 "io"
7 "net"
8 "strconv"
9 "strings"
10 "time"
11
12 "gopkg.in/irc.v3"
13)
14
15type upstreamChannel struct {
16 Name string
17 conn *upstreamConn
18 Topic string
19 TopicWho string
20 TopicTime time.Time
21 Status channelStatus
22 modes modeSet
23 Members map[string]membership
24 complete bool
25}
26
27type upstreamConn struct {
28 network *network
29 logger Logger
30 net net.Conn
31 irc *irc.Conn
32 srv *Server
33 user *user
34 messages chan<- *irc.Message
35 ring *Ring
36
37 serverName string
38 availableUserModes string
39 availableChannelModes string
40 channelModesWithParam string
41
42 registered bool
43 nick string
44 username string
45 realname string
46 closed bool
47 modes modeSet
48 channels map[string]*upstreamChannel
49 history map[string]uint64
50}
51
52func connectToUpstream(network *network) (*upstreamConn, error) {
53 logger := &prefixLogger{network.user.srv.Logger, fmt.Sprintf("upstream %q: ", network.Addr)}
54
55 addr := network.Addr
56 if !strings.ContainsRune(addr, ':') {
57 addr = addr + ":6697"
58 }
59
60 logger.Printf("connecting to TLS server at address %q", addr)
61 netConn, err := tls.Dial("tcp", addr, nil)
62 if err != nil {
63 return nil, fmt.Errorf("failed to dial %q: %v", addr, err)
64 }
65
66 setKeepAlive(netConn)
67
68 msgs := make(chan *irc.Message, 64)
69 uc := &upstreamConn{
70 network: network,
71 logger: logger,
72 net: netConn,
73 irc: irc.NewConn(netConn),
74 srv: network.user.srv,
75 user: network.user,
76 messages: msgs,
77 ring: NewRing(network.user.srv.RingCap),
78 channels: make(map[string]*upstreamChannel),
79 history: make(map[string]uint64),
80 }
81
82 go func() {
83 for msg := range msgs {
84 if uc.srv.Debug {
85 uc.logger.Printf("sent: %v", msg)
86 }
87 if err := uc.irc.WriteMessage(msg); err != nil {
88 uc.logger.Printf("failed to write message: %v", err)
89 }
90 }
91 if err := uc.net.Close(); err != nil {
92 uc.logger.Printf("failed to close connection: %v", err)
93 } else {
94 uc.logger.Printf("connection closed")
95 }
96 }()
97
98 return uc, nil
99}
100
101func (uc *upstreamConn) Close() error {
102 if uc.closed {
103 return fmt.Errorf("upstream connection already closed")
104 }
105 close(uc.messages)
106 uc.closed = true
107 return nil
108}
109
110func (uc *upstreamConn) forEachDownstream(f func(*downstreamConn)) {
111 uc.user.forEachDownstream(func(dc *downstreamConn) {
112 if dc.network != nil && dc.network != uc.network {
113 return
114 }
115 f(dc)
116 })
117}
118
119func (uc *upstreamConn) getChannel(name string) (*upstreamChannel, error) {
120 ch, ok := uc.channels[name]
121 if !ok {
122 return nil, fmt.Errorf("unknown channel %q", name)
123 }
124 return ch, nil
125}
126
127func (uc *upstreamConn) handleMessage(msg *irc.Message) error {
128 switch msg.Command {
129 case "PING":
130 uc.SendMessage(&irc.Message{
131 Command: "PONG",
132 Params: msg.Params,
133 })
134 return nil
135 case "MODE":
136 if msg.Prefix == nil {
137 return fmt.Errorf("missing prefix")
138 }
139
140 var name, modeStr string
141 if err := parseMessageParams(msg, &name, &modeStr); err != nil {
142 return err
143 }
144
145 if name == msg.Prefix.Name { // user mode change
146 if name != uc.nick {
147 return fmt.Errorf("received MODE message for unknow nick %q", name)
148 }
149 return uc.modes.Apply(modeStr)
150 } else { // channel mode change
151 ch, err := uc.getChannel(name)
152 if err != nil {
153 return err
154 }
155 if err := ch.modes.Apply(modeStr); err != nil {
156 return err
157 }
158
159 uc.forEachDownstream(func(dc *downstreamConn) {
160 dc.SendMessage(&irc.Message{
161 Prefix: dc.marshalUserPrefix(uc, msg.Prefix),
162 Command: "MODE",
163 Params: []string{dc.marshalChannel(uc, name), modeStr},
164 })
165 })
166 }
167 case "NOTICE":
168 uc.logger.Print(msg)
169 case irc.RPL_WELCOME:
170 uc.registered = true
171 uc.logger.Printf("connection registered")
172
173 channels, err := uc.srv.db.ListChannels(uc.network.ID)
174 if err != nil {
175 uc.logger.Printf("failed to list channels from database: %v", err)
176 break
177 }
178
179 for _, ch := range channels {
180 uc.SendMessage(&irc.Message{
181 Command: "JOIN",
182 Params: []string{ch.Name},
183 })
184 }
185 case irc.RPL_MYINFO:
186 if err := parseMessageParams(msg, nil, &uc.serverName, nil, &uc.availableUserModes, &uc.availableChannelModes); err != nil {
187 return err
188 }
189 if len(msg.Params) > 5 {
190 uc.channelModesWithParam = msg.Params[5]
191 }
192 case "NICK":
193 var newNick string
194 if err := parseMessageParams(msg, &newNick); err != nil {
195 return err
196 }
197
198 if msg.Prefix.Name == uc.nick {
199 uc.logger.Printf("changed nick from %q to %q", uc.nick, newNick)
200 uc.nick = newNick
201 }
202
203 for _, ch := range uc.channels {
204 if membership, ok := ch.Members[msg.Prefix.Name]; ok {
205 delete(ch.Members, msg.Prefix.Name)
206 ch.Members[newNick] = membership
207 }
208 }
209
210 if msg.Prefix.Name != uc.nick {
211 uc.forEachDownstream(func(dc *downstreamConn) {
212 dc.SendMessage(&irc.Message{
213 Prefix: dc.marshalUserPrefix(uc, msg.Prefix),
214 Command: "NICK",
215 Params: []string{newNick},
216 })
217 })
218 }
219 case "JOIN":
220 if msg.Prefix == nil {
221 return fmt.Errorf("expected a prefix")
222 }
223
224 var channels string
225 if err := parseMessageParams(msg, &channels); err != nil {
226 return err
227 }
228
229 for _, ch := range strings.Split(channels, ",") {
230 if msg.Prefix.Name == uc.nick {
231 uc.logger.Printf("joined channel %q", ch)
232 uc.channels[ch] = &upstreamChannel{
233 Name: ch,
234 conn: uc,
235 Members: make(map[string]membership),
236 }
237 } else {
238 ch, err := uc.getChannel(ch)
239 if err != nil {
240 return err
241 }
242 ch.Members[msg.Prefix.Name] = 0
243 }
244
245 uc.forEachDownstream(func(dc *downstreamConn) {
246 dc.SendMessage(&irc.Message{
247 Prefix: dc.marshalUserPrefix(uc, msg.Prefix),
248 Command: "JOIN",
249 Params: []string{dc.marshalChannel(uc, ch)},
250 })
251 })
252 }
253 case "PART":
254 if msg.Prefix == nil {
255 return fmt.Errorf("expected a prefix")
256 }
257
258 var channels string
259 if err := parseMessageParams(msg, &channels); err != nil {
260 return err
261 }
262
263 for _, ch := range strings.Split(channels, ",") {
264 if msg.Prefix.Name == uc.nick {
265 uc.logger.Printf("parted channel %q", ch)
266 delete(uc.channels, ch)
267 } else {
268 ch, err := uc.getChannel(ch)
269 if err != nil {
270 return err
271 }
272 delete(ch.Members, msg.Prefix.Name)
273 }
274
275 uc.forEachDownstream(func(dc *downstreamConn) {
276 dc.SendMessage(&irc.Message{
277 Prefix: dc.marshalUserPrefix(uc, msg.Prefix),
278 Command: "PART",
279 Params: []string{dc.marshalChannel(uc, ch)},
280 })
281 })
282 }
283 case irc.RPL_TOPIC, irc.RPL_NOTOPIC:
284 var name, topic string
285 if err := parseMessageParams(msg, nil, &name, &topic); err != nil {
286 return err
287 }
288 ch, err := uc.getChannel(name)
289 if err != nil {
290 return err
291 }
292 if msg.Command == irc.RPL_TOPIC {
293 ch.Topic = topic
294 } else {
295 ch.Topic = ""
296 }
297 case "TOPIC":
298 var name string
299 if err := parseMessageParams(msg, &name); err != nil {
300 return err
301 }
302 ch, err := uc.getChannel(name)
303 if err != nil {
304 return err
305 }
306 if len(msg.Params) > 1 {
307 ch.Topic = msg.Params[1]
308 } else {
309 ch.Topic = ""
310 }
311 uc.forEachDownstream(func(dc *downstreamConn) {
312 params := []string{dc.marshalChannel(uc, name)}
313 if ch.Topic != "" {
314 params = append(params, ch.Topic)
315 }
316 dc.SendMessage(&irc.Message{
317 Prefix: dc.marshalUserPrefix(uc, msg.Prefix),
318 Command: "TOPIC",
319 Params: params,
320 })
321 })
322 case rpl_topicwhotime:
323 var name, who, timeStr string
324 if err := parseMessageParams(msg, nil, &name, &who, &timeStr); err != nil {
325 return err
326 }
327 ch, err := uc.getChannel(name)
328 if err != nil {
329 return err
330 }
331 ch.TopicWho = who
332 sec, err := strconv.ParseInt(timeStr, 10, 64)
333 if err != nil {
334 return fmt.Errorf("failed to parse topic time: %v", err)
335 }
336 ch.TopicTime = time.Unix(sec, 0)
337 case irc.RPL_NAMREPLY:
338 var name, statusStr, members string
339 if err := parseMessageParams(msg, nil, &statusStr, &name, &members); err != nil {
340 return err
341 }
342 ch, err := uc.getChannel(name)
343 if err != nil {
344 return err
345 }
346
347 status, err := parseChannelStatus(statusStr)
348 if err != nil {
349 return err
350 }
351 ch.Status = status
352
353 for _, s := range strings.Split(members, " ") {
354 membership, nick := parseMembershipPrefix(s)
355 ch.Members[nick] = membership
356 }
357 case irc.RPL_ENDOFNAMES:
358 var name string
359 if err := parseMessageParams(msg, nil, &name); err != nil {
360 return err
361 }
362 ch, err := uc.getChannel(name)
363 if err != nil {
364 return err
365 }
366
367 if ch.complete {
368 return fmt.Errorf("received unexpected RPL_ENDOFNAMES")
369 }
370 ch.complete = true
371
372 uc.forEachDownstream(func(dc *downstreamConn) {
373 forwardChannel(dc, ch)
374 })
375 case "PRIVMSG":
376 if err := parseMessageParams(msg, nil, nil); err != nil {
377 return err
378 }
379 uc.ring.Produce(msg)
380 case irc.RPL_YOURHOST, irc.RPL_CREATED:
381 // Ignore
382 case irc.RPL_LUSERCLIENT, irc.RPL_LUSEROP, irc.RPL_LUSERUNKNOWN, irc.RPL_LUSERCHANNELS, irc.RPL_LUSERME:
383 // Ignore
384 case irc.RPL_MOTDSTART, irc.RPL_MOTD, irc.RPL_ENDOFMOTD:
385 // Ignore
386 case rpl_localusers, rpl_globalusers:
387 // Ignore
388 case irc.RPL_STATSVLINE, irc.RPL_STATSPING, irc.RPL_STATSBLINE, irc.RPL_STATSDLINE:
389 // Ignore
390 default:
391 uc.logger.Printf("unhandled upstream message: %v", msg)
392 }
393 return nil
394}
395
396func (uc *upstreamConn) register() {
397 uc.nick = uc.network.Nick
398 uc.username = uc.network.Username
399 if uc.username == "" {
400 uc.username = uc.nick
401 }
402 uc.realname = uc.network.Realname
403 if uc.realname == "" {
404 uc.realname = uc.nick
405 }
406
407 uc.SendMessage(&irc.Message{
408 Command: "NICK",
409 Params: []string{uc.nick},
410 })
411 uc.SendMessage(&irc.Message{
412 Command: "USER",
413 Params: []string{uc.username, "0", "*", uc.realname},
414 })
415}
416
417func (uc *upstreamConn) readMessages() error {
418 for {
419 msg, err := uc.irc.ReadMessage()
420 if err == io.EOF {
421 break
422 } else if err != nil {
423 return fmt.Errorf("failed to read IRC command: %v", err)
424 }
425
426 if uc.srv.Debug {
427 uc.logger.Printf("received: %v", msg)
428 }
429
430 if err := uc.handleMessage(msg); err != nil {
431 uc.logger.Printf("failed to handle message %q: %v", msg, err)
432 }
433 }
434
435 return nil
436}
437
438func (uc *upstreamConn) SendMessage(msg *irc.Message) {
439 uc.messages <- msg
440}
Note: See TracBrowser for help on using the repository browser.