1 | package soju
|
---|
2 |
|
---|
3 | import (
|
---|
4 | "context"
|
---|
5 | "crypto/sha256"
|
---|
6 | "encoding/binary"
|
---|
7 | "encoding/hex"
|
---|
8 | "fmt"
|
---|
9 | "math/big"
|
---|
10 | "net"
|
---|
11 | "sort"
|
---|
12 | "strings"
|
---|
13 | "time"
|
---|
14 |
|
---|
15 | "gopkg.in/irc.v3"
|
---|
16 | )
|
---|
17 |
|
---|
18 | type event interface{}
|
---|
19 |
|
---|
20 | type eventUpstreamMessage struct {
|
---|
21 | msg *irc.Message
|
---|
22 | uc *upstreamConn
|
---|
23 | }
|
---|
24 |
|
---|
25 | type eventUpstreamConnectionError struct {
|
---|
26 | net *network
|
---|
27 | err error
|
---|
28 | }
|
---|
29 |
|
---|
30 | type eventUpstreamConnected struct {
|
---|
31 | uc *upstreamConn
|
---|
32 | }
|
---|
33 |
|
---|
34 | type eventUpstreamDisconnected struct {
|
---|
35 | uc *upstreamConn
|
---|
36 | }
|
---|
37 |
|
---|
38 | type eventUpstreamError struct {
|
---|
39 | uc *upstreamConn
|
---|
40 | err error
|
---|
41 | }
|
---|
42 |
|
---|
43 | type eventDownstreamMessage struct {
|
---|
44 | msg *irc.Message
|
---|
45 | dc *downstreamConn
|
---|
46 | }
|
---|
47 |
|
---|
48 | type eventDownstreamConnected struct {
|
---|
49 | dc *downstreamConn
|
---|
50 | }
|
---|
51 |
|
---|
52 | type eventDownstreamDisconnected struct {
|
---|
53 | dc *downstreamConn
|
---|
54 | }
|
---|
55 |
|
---|
56 | type eventChannelDetach struct {
|
---|
57 | uc *upstreamConn
|
---|
58 | name string
|
---|
59 | }
|
---|
60 |
|
---|
61 | type eventBroadcast struct {
|
---|
62 | msg *irc.Message
|
---|
63 | }
|
---|
64 |
|
---|
65 | type eventStop struct{}
|
---|
66 |
|
---|
67 | type eventUserUpdate struct {
|
---|
68 | password *string
|
---|
69 | admin *bool
|
---|
70 | done chan error
|
---|
71 | }
|
---|
72 |
|
---|
73 | type deliveredClientMap map[string]string // client name -> msg ID
|
---|
74 |
|
---|
75 | type deliveredStore struct {
|
---|
76 | m deliveredCasemapMap
|
---|
77 | }
|
---|
78 |
|
---|
79 | func newDeliveredStore() deliveredStore {
|
---|
80 | return deliveredStore{deliveredCasemapMap{newCasemapMap(0)}}
|
---|
81 | }
|
---|
82 |
|
---|
83 | func (ds deliveredStore) HasTarget(target string) bool {
|
---|
84 | return ds.m.Value(target) != nil
|
---|
85 | }
|
---|
86 |
|
---|
87 | func (ds deliveredStore) LoadID(target, clientName string) string {
|
---|
88 | clients := ds.m.Value(target)
|
---|
89 | if clients == nil {
|
---|
90 | return ""
|
---|
91 | }
|
---|
92 | return clients[clientName]
|
---|
93 | }
|
---|
94 |
|
---|
95 | func (ds deliveredStore) StoreID(target, clientName, msgID string) {
|
---|
96 | clients := ds.m.Value(target)
|
---|
97 | if clients == nil {
|
---|
98 | clients = make(deliveredClientMap)
|
---|
99 | ds.m.SetValue(target, clients)
|
---|
100 | }
|
---|
101 | clients[clientName] = msgID
|
---|
102 | }
|
---|
103 |
|
---|
104 | func (ds deliveredStore) ForEachTarget(f func(target string)) {
|
---|
105 | for _, entry := range ds.m.innerMap {
|
---|
106 | f(entry.originalKey)
|
---|
107 | }
|
---|
108 | }
|
---|
109 |
|
---|
110 | func (ds deliveredStore) ForEachClient(f func(clientName string)) {
|
---|
111 | clients := make(map[string]struct{})
|
---|
112 | for _, entry := range ds.m.innerMap {
|
---|
113 | delivered := entry.value.(deliveredClientMap)
|
---|
114 | for clientName := range delivered {
|
---|
115 | clients[clientName] = struct{}{}
|
---|
116 | }
|
---|
117 | }
|
---|
118 |
|
---|
119 | for clientName := range clients {
|
---|
120 | f(clientName)
|
---|
121 | }
|
---|
122 | }
|
---|
123 |
|
---|
124 | type network struct {
|
---|
125 | Network
|
---|
126 | user *user
|
---|
127 | logger Logger
|
---|
128 | stopped chan struct{}
|
---|
129 |
|
---|
130 | conn *upstreamConn
|
---|
131 | channels channelCasemapMap
|
---|
132 | delivered deliveredStore
|
---|
133 | lastError error
|
---|
134 | casemap casemapping
|
---|
135 | }
|
---|
136 |
|
---|
137 | func newNetwork(user *user, record *Network, channels []Channel) *network {
|
---|
138 | logger := &prefixLogger{user.logger, fmt.Sprintf("network %q: ", record.GetName())}
|
---|
139 |
|
---|
140 | m := channelCasemapMap{newCasemapMap(0)}
|
---|
141 | for _, ch := range channels {
|
---|
142 | ch := ch
|
---|
143 | m.SetValue(ch.Name, &ch)
|
---|
144 | }
|
---|
145 |
|
---|
146 | return &network{
|
---|
147 | Network: *record,
|
---|
148 | user: user,
|
---|
149 | logger: logger,
|
---|
150 | stopped: make(chan struct{}),
|
---|
151 | channels: m,
|
---|
152 | delivered: newDeliveredStore(),
|
---|
153 | casemap: casemapRFC1459,
|
---|
154 | }
|
---|
155 | }
|
---|
156 |
|
---|
157 | func (net *network) forEachDownstream(f func(*downstreamConn)) {
|
---|
158 | net.user.forEachDownstream(func(dc *downstreamConn) {
|
---|
159 | if dc.network == nil && !dc.isMultiUpstream {
|
---|
160 | return
|
---|
161 | }
|
---|
162 | if dc.network != nil && dc.network != net {
|
---|
163 | return
|
---|
164 | }
|
---|
165 | f(dc)
|
---|
166 | })
|
---|
167 | }
|
---|
168 |
|
---|
169 | func (net *network) isStopped() bool {
|
---|
170 | select {
|
---|
171 | case <-net.stopped:
|
---|
172 | return true
|
---|
173 | default:
|
---|
174 | return false
|
---|
175 | }
|
---|
176 | }
|
---|
177 |
|
---|
178 | func userIdent(u *User) string {
|
---|
179 | // The ident is a string we will send to upstream servers in clear-text.
|
---|
180 | // For privacy reasons, make sure it doesn't expose any meaningful user
|
---|
181 | // metadata. We just use the base64-encoded hashed ID, so that people don't
|
---|
182 | // start relying on the string being an integer or following a pattern.
|
---|
183 | var b [64]byte
|
---|
184 | binary.LittleEndian.PutUint64(b[:], uint64(u.ID))
|
---|
185 | h := sha256.Sum256(b[:])
|
---|
186 | return hex.EncodeToString(h[:16])
|
---|
187 | }
|
---|
188 |
|
---|
189 | func (net *network) run() {
|
---|
190 | if !net.Enabled {
|
---|
191 | return
|
---|
192 | }
|
---|
193 |
|
---|
194 | var lastTry time.Time
|
---|
195 | backoff := newBackoffer(retryConnectMinDelay, retryConnectMaxDelay, retryConnectJitter)
|
---|
196 | for {
|
---|
197 | if net.isStopped() {
|
---|
198 | return
|
---|
199 | }
|
---|
200 |
|
---|
201 | delay := backoff.Next() - time.Now().Sub(lastTry)
|
---|
202 | if delay > 0 {
|
---|
203 | net.logger.Printf("waiting %v before trying to reconnect to %q", delay.Truncate(time.Second), net.Addr)
|
---|
204 | time.Sleep(delay)
|
---|
205 | }
|
---|
206 | lastTry = time.Now()
|
---|
207 |
|
---|
208 | net.user.srv.metrics.upstreams.Add(1)
|
---|
209 |
|
---|
210 | uc, err := connectToUpstream(context.TODO(), net)
|
---|
211 | if err != nil {
|
---|
212 | net.logger.Printf("failed to connect to upstream server %q: %v", net.Addr, err)
|
---|
213 | net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("failed to connect: %v", err)}
|
---|
214 | net.user.srv.metrics.upstreams.Add(-1)
|
---|
215 | net.user.srv.metrics.upstreamConnectErrorsTotal.Inc()
|
---|
216 | continue
|
---|
217 | }
|
---|
218 |
|
---|
219 | if net.user.srv.Identd != nil {
|
---|
220 | net.user.srv.Identd.Store(uc.RemoteAddr().String(), uc.LocalAddr().String(), userIdent(&net.user.User))
|
---|
221 | }
|
---|
222 |
|
---|
223 | uc.register()
|
---|
224 | if err := uc.runUntilRegistered(); err != nil {
|
---|
225 | text := err.Error()
|
---|
226 | temp := true
|
---|
227 | if regErr, ok := err.(registrationError); ok {
|
---|
228 | text = regErr.Reason()
|
---|
229 | temp = regErr.Temporary()
|
---|
230 | }
|
---|
231 | uc.logger.Printf("failed to register: %v", text)
|
---|
232 | net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("failed to register: %v", text)}
|
---|
233 | uc.Close()
|
---|
234 | net.user.srv.metrics.upstreams.Add(-1)
|
---|
235 | net.user.srv.metrics.upstreamConnectErrorsTotal.Inc()
|
---|
236 | if !temp {
|
---|
237 | return
|
---|
238 | }
|
---|
239 | continue
|
---|
240 | }
|
---|
241 |
|
---|
242 | // TODO: this is racy with net.stopped. If the network is stopped
|
---|
243 | // before the user goroutine receives eventUpstreamConnected, the
|
---|
244 | // connection won't be closed.
|
---|
245 | net.user.events <- eventUpstreamConnected{uc}
|
---|
246 | if err := uc.readMessages(net.user.events); err != nil {
|
---|
247 | uc.logger.Printf("failed to handle messages: %v", err)
|
---|
248 | net.user.events <- eventUpstreamError{uc, fmt.Errorf("failed to handle messages: %v", err)}
|
---|
249 | }
|
---|
250 | uc.Close()
|
---|
251 | net.user.events <- eventUpstreamDisconnected{uc}
|
---|
252 |
|
---|
253 | if net.user.srv.Identd != nil {
|
---|
254 | net.user.srv.Identd.Delete(uc.RemoteAddr().String(), uc.LocalAddr().String())
|
---|
255 | }
|
---|
256 |
|
---|
257 | net.user.srv.metrics.upstreams.Add(-1)
|
---|
258 | backoff.Reset()
|
---|
259 | }
|
---|
260 | }
|
---|
261 |
|
---|
262 | func (net *network) stop() {
|
---|
263 | if !net.isStopped() {
|
---|
264 | close(net.stopped)
|
---|
265 | }
|
---|
266 |
|
---|
267 | if net.conn != nil {
|
---|
268 | net.conn.Close()
|
---|
269 | }
|
---|
270 | }
|
---|
271 |
|
---|
272 | func (net *network) detach(ch *Channel) {
|
---|
273 | if ch.Detached {
|
---|
274 | return
|
---|
275 | }
|
---|
276 |
|
---|
277 | net.logger.Printf("detaching channel %q", ch.Name)
|
---|
278 |
|
---|
279 | ch.Detached = true
|
---|
280 |
|
---|
281 | if net.user.msgStore != nil {
|
---|
282 | nameCM := net.casemap(ch.Name)
|
---|
283 | lastID, err := net.user.msgStore.LastMsgID(&net.Network, nameCM, time.Now())
|
---|
284 | if err != nil {
|
---|
285 | net.logger.Printf("failed to get last message ID for channel %q: %v", ch.Name, err)
|
---|
286 | }
|
---|
287 | ch.DetachedInternalMsgID = lastID
|
---|
288 | }
|
---|
289 |
|
---|
290 | if net.conn != nil {
|
---|
291 | uch := net.conn.channels.Value(ch.Name)
|
---|
292 | if uch != nil {
|
---|
293 | uch.updateAutoDetach(0)
|
---|
294 | }
|
---|
295 | }
|
---|
296 |
|
---|
297 | net.forEachDownstream(func(dc *downstreamConn) {
|
---|
298 | dc.SendMessage(&irc.Message{
|
---|
299 | Prefix: dc.prefix(),
|
---|
300 | Command: "PART",
|
---|
301 | Params: []string{dc.marshalEntity(net, ch.Name), "Detach"},
|
---|
302 | })
|
---|
303 | })
|
---|
304 | }
|
---|
305 |
|
---|
306 | func (net *network) attach(ch *Channel) {
|
---|
307 | if !ch.Detached {
|
---|
308 | return
|
---|
309 | }
|
---|
310 |
|
---|
311 | net.logger.Printf("attaching channel %q", ch.Name)
|
---|
312 |
|
---|
313 | detachedMsgID := ch.DetachedInternalMsgID
|
---|
314 | ch.Detached = false
|
---|
315 | ch.DetachedInternalMsgID = ""
|
---|
316 |
|
---|
317 | var uch *upstreamChannel
|
---|
318 | if net.conn != nil {
|
---|
319 | uch = net.conn.channels.Value(ch.Name)
|
---|
320 |
|
---|
321 | net.conn.updateChannelAutoDetach(ch.Name)
|
---|
322 | }
|
---|
323 |
|
---|
324 | net.forEachDownstream(func(dc *downstreamConn) {
|
---|
325 | dc.SendMessage(&irc.Message{
|
---|
326 | Prefix: dc.prefix(),
|
---|
327 | Command: "JOIN",
|
---|
328 | Params: []string{dc.marshalEntity(net, ch.Name)},
|
---|
329 | })
|
---|
330 |
|
---|
331 | if uch != nil {
|
---|
332 | forwardChannel(dc, uch)
|
---|
333 | }
|
---|
334 |
|
---|
335 | if detachedMsgID != "" {
|
---|
336 | dc.sendTargetBacklog(context.TODO(), net, ch.Name, detachedMsgID)
|
---|
337 | }
|
---|
338 | })
|
---|
339 | }
|
---|
340 |
|
---|
341 | func (net *network) deleteChannel(ctx context.Context, name string) error {
|
---|
342 | ch := net.channels.Value(name)
|
---|
343 | if ch == nil {
|
---|
344 | return fmt.Errorf("unknown channel %q", name)
|
---|
345 | }
|
---|
346 | if net.conn != nil {
|
---|
347 | uch := net.conn.channels.Value(ch.Name)
|
---|
348 | if uch != nil {
|
---|
349 | uch.updateAutoDetach(0)
|
---|
350 | }
|
---|
351 | }
|
---|
352 |
|
---|
353 | if err := net.user.srv.db.DeleteChannel(ctx, ch.ID); err != nil {
|
---|
354 | return err
|
---|
355 | }
|
---|
356 | net.channels.Delete(name)
|
---|
357 | return nil
|
---|
358 | }
|
---|
359 |
|
---|
360 | func (net *network) updateCasemapping(newCasemap casemapping) {
|
---|
361 | net.casemap = newCasemap
|
---|
362 | net.channels.SetCasemapping(newCasemap)
|
---|
363 | net.delivered.m.SetCasemapping(newCasemap)
|
---|
364 | if uc := net.conn; uc != nil {
|
---|
365 | uc.channels.SetCasemapping(newCasemap)
|
---|
366 | for _, entry := range uc.channels.innerMap {
|
---|
367 | uch := entry.value.(*upstreamChannel)
|
---|
368 | uch.Members.SetCasemapping(newCasemap)
|
---|
369 | }
|
---|
370 | uc.monitored.SetCasemapping(newCasemap)
|
---|
371 | }
|
---|
372 | net.forEachDownstream(func(dc *downstreamConn) {
|
---|
373 | dc.monitored.SetCasemapping(newCasemap)
|
---|
374 | })
|
---|
375 | }
|
---|
376 |
|
---|
377 | func (net *network) storeClientDeliveryReceipts(ctx context.Context, clientName string) {
|
---|
378 | if !net.user.hasPersistentMsgStore() {
|
---|
379 | return
|
---|
380 | }
|
---|
381 |
|
---|
382 | var receipts []DeliveryReceipt
|
---|
383 | net.delivered.ForEachTarget(func(target string) {
|
---|
384 | msgID := net.delivered.LoadID(target, clientName)
|
---|
385 | if msgID == "" {
|
---|
386 | return
|
---|
387 | }
|
---|
388 | receipts = append(receipts, DeliveryReceipt{
|
---|
389 | Target: target,
|
---|
390 | InternalMsgID: msgID,
|
---|
391 | })
|
---|
392 | })
|
---|
393 |
|
---|
394 | if err := net.user.srv.db.StoreClientDeliveryReceipts(ctx, net.ID, clientName, receipts); err != nil {
|
---|
395 | net.logger.Printf("failed to store delivery receipts for client %q: %v", clientName, err)
|
---|
396 | }
|
---|
397 | }
|
---|
398 |
|
---|
399 | func (net *network) isHighlight(msg *irc.Message) bool {
|
---|
400 | if msg.Command != "PRIVMSG" && msg.Command != "NOTICE" {
|
---|
401 | return false
|
---|
402 | }
|
---|
403 |
|
---|
404 | text := msg.Params[1]
|
---|
405 |
|
---|
406 | nick := net.Nick
|
---|
407 | if net.conn != nil {
|
---|
408 | nick = net.conn.nick
|
---|
409 | }
|
---|
410 |
|
---|
411 | // TODO: use case-mapping aware comparison here
|
---|
412 | return msg.Prefix.Name != nick && isHighlight(text, nick)
|
---|
413 | }
|
---|
414 |
|
---|
415 | func (net *network) detachedMessageNeedsRelay(ch *Channel, msg *irc.Message) bool {
|
---|
416 | highlight := net.isHighlight(msg)
|
---|
417 | return ch.RelayDetached == FilterMessage || ((ch.RelayDetached == FilterHighlight || ch.RelayDetached == FilterDefault) && highlight)
|
---|
418 | }
|
---|
419 |
|
---|
420 | func (net *network) autoSaveSASLPlain(ctx context.Context, username, password string) {
|
---|
421 | // User may have e.g. EXTERNAL mechanism configured. We do not want to
|
---|
422 | // automatically erase the key pair or any other credentials.
|
---|
423 | if net.SASL.Mechanism != "" && net.SASL.Mechanism != "PLAIN" {
|
---|
424 | return
|
---|
425 | }
|
---|
426 |
|
---|
427 | net.logger.Printf("auto-saving SASL PLAIN credentials with username %q", username)
|
---|
428 | net.SASL.Mechanism = "PLAIN"
|
---|
429 | net.SASL.Plain.Username = username
|
---|
430 | net.SASL.Plain.Password = password
|
---|
431 | if err := net.user.srv.db.StoreNetwork(ctx, net.user.ID, &net.Network); err != nil {
|
---|
432 | net.logger.Printf("failed to save SASL PLAIN credentials: %v", err)
|
---|
433 | }
|
---|
434 | }
|
---|
435 |
|
---|
436 | type user struct {
|
---|
437 | User
|
---|
438 | srv *Server
|
---|
439 | logger Logger
|
---|
440 |
|
---|
441 | events chan event
|
---|
442 | done chan struct{}
|
---|
443 |
|
---|
444 | networks []*network
|
---|
445 | downstreamConns []*downstreamConn
|
---|
446 | msgStore messageStore
|
---|
447 | }
|
---|
448 |
|
---|
449 | func newUser(srv *Server, record *User) *user {
|
---|
450 | logger := &prefixLogger{srv.Logger, fmt.Sprintf("user %q: ", record.Username)}
|
---|
451 |
|
---|
452 | var msgStore messageStore
|
---|
453 | if logPath := srv.Config().LogPath; logPath != "" {
|
---|
454 | msgStore = newFSMessageStore(logPath, record.Username)
|
---|
455 | } else {
|
---|
456 | msgStore = newMemoryMessageStore()
|
---|
457 | }
|
---|
458 |
|
---|
459 | return &user{
|
---|
460 | User: *record,
|
---|
461 | srv: srv,
|
---|
462 | logger: logger,
|
---|
463 | events: make(chan event, 64),
|
---|
464 | done: make(chan struct{}),
|
---|
465 | msgStore: msgStore,
|
---|
466 | }
|
---|
467 | }
|
---|
468 |
|
---|
469 | func (u *user) forEachUpstream(f func(uc *upstreamConn)) {
|
---|
470 | for _, network := range u.networks {
|
---|
471 | if network.conn == nil {
|
---|
472 | continue
|
---|
473 | }
|
---|
474 | f(network.conn)
|
---|
475 | }
|
---|
476 | }
|
---|
477 |
|
---|
478 | func (u *user) forEachDownstream(f func(dc *downstreamConn)) {
|
---|
479 | for _, dc := range u.downstreamConns {
|
---|
480 | f(dc)
|
---|
481 | }
|
---|
482 | }
|
---|
483 |
|
---|
484 | func (u *user) getNetwork(name string) *network {
|
---|
485 | for _, network := range u.networks {
|
---|
486 | if network.Addr == name {
|
---|
487 | return network
|
---|
488 | }
|
---|
489 | if network.Name != "" && network.Name == name {
|
---|
490 | return network
|
---|
491 | }
|
---|
492 | }
|
---|
493 | return nil
|
---|
494 | }
|
---|
495 |
|
---|
496 | func (u *user) getNetworkByID(id int64) *network {
|
---|
497 | for _, net := range u.networks {
|
---|
498 | if net.ID == id {
|
---|
499 | return net
|
---|
500 | }
|
---|
501 | }
|
---|
502 | return nil
|
---|
503 | }
|
---|
504 |
|
---|
505 | func (u *user) run() {
|
---|
506 | defer func() {
|
---|
507 | if u.msgStore != nil {
|
---|
508 | if err := u.msgStore.Close(); err != nil {
|
---|
509 | u.logger.Printf("failed to close message store for user %q: %v", u.Username, err)
|
---|
510 | }
|
---|
511 | }
|
---|
512 | close(u.done)
|
---|
513 | }()
|
---|
514 |
|
---|
515 | networks, err := u.srv.db.ListNetworks(context.TODO(), u.ID)
|
---|
516 | if err != nil {
|
---|
517 | u.logger.Printf("failed to list networks for user %q: %v", u.Username, err)
|
---|
518 | return
|
---|
519 | }
|
---|
520 |
|
---|
521 | sort.Slice(networks, func(i, j int) bool {
|
---|
522 | return networks[i].ID < networks[j].ID
|
---|
523 | })
|
---|
524 |
|
---|
525 | for _, record := range networks {
|
---|
526 | record := record
|
---|
527 | channels, err := u.srv.db.ListChannels(context.TODO(), record.ID)
|
---|
528 | if err != nil {
|
---|
529 | u.logger.Printf("failed to list channels for user %q, network %q: %v", u.Username, record.GetName(), err)
|
---|
530 | continue
|
---|
531 | }
|
---|
532 |
|
---|
533 | network := newNetwork(u, &record, channels)
|
---|
534 | u.networks = append(u.networks, network)
|
---|
535 |
|
---|
536 | if u.hasPersistentMsgStore() {
|
---|
537 | receipts, err := u.srv.db.ListDeliveryReceipts(context.TODO(), record.ID)
|
---|
538 | if err != nil {
|
---|
539 | u.logger.Printf("failed to load delivery receipts for user %q, network %q: %v", u.Username, network.GetName(), err)
|
---|
540 | return
|
---|
541 | }
|
---|
542 |
|
---|
543 | for _, rcpt := range receipts {
|
---|
544 | network.delivered.StoreID(rcpt.Target, rcpt.Client, rcpt.InternalMsgID)
|
---|
545 | }
|
---|
546 | }
|
---|
547 |
|
---|
548 | go network.run()
|
---|
549 | }
|
---|
550 |
|
---|
551 | for e := range u.events {
|
---|
552 | switch e := e.(type) {
|
---|
553 | case eventUpstreamConnected:
|
---|
554 | uc := e.uc
|
---|
555 |
|
---|
556 | uc.network.conn = uc
|
---|
557 |
|
---|
558 | uc.updateAway()
|
---|
559 | uc.updateMonitor()
|
---|
560 |
|
---|
561 | netIDStr := fmt.Sprintf("%v", uc.network.ID)
|
---|
562 | uc.forEachDownstream(func(dc *downstreamConn) {
|
---|
563 | dc.updateSupportedCaps()
|
---|
564 |
|
---|
565 | if !dc.caps["soju.im/bouncer-networks"] {
|
---|
566 | sendServiceNOTICE(dc, fmt.Sprintf("connected to %s", uc.network.GetName()))
|
---|
567 | }
|
---|
568 |
|
---|
569 | dc.updateNick()
|
---|
570 | dc.updateRealname()
|
---|
571 | dc.updateAccount()
|
---|
572 | })
|
---|
573 | u.forEachDownstream(func(dc *downstreamConn) {
|
---|
574 | if dc.caps["soju.im/bouncer-networks-notify"] {
|
---|
575 | dc.SendMessage(&irc.Message{
|
---|
576 | Prefix: dc.srv.prefix(),
|
---|
577 | Command: "BOUNCER",
|
---|
578 | Params: []string{"NETWORK", netIDStr, "state=connected"},
|
---|
579 | })
|
---|
580 | }
|
---|
581 | })
|
---|
582 | uc.network.lastError = nil
|
---|
583 | case eventUpstreamDisconnected:
|
---|
584 | u.handleUpstreamDisconnected(e.uc)
|
---|
585 | case eventUpstreamConnectionError:
|
---|
586 | net := e.net
|
---|
587 |
|
---|
588 | stopped := false
|
---|
589 | select {
|
---|
590 | case <-net.stopped:
|
---|
591 | stopped = true
|
---|
592 | default:
|
---|
593 | }
|
---|
594 |
|
---|
595 | if !stopped && (net.lastError == nil || net.lastError.Error() != e.err.Error()) {
|
---|
596 | net.forEachDownstream(func(dc *downstreamConn) {
|
---|
597 | sendServiceNOTICE(dc, fmt.Sprintf("failed connecting/registering to %s: %v", net.GetName(), e.err))
|
---|
598 | })
|
---|
599 | }
|
---|
600 | net.lastError = e.err
|
---|
601 | case eventUpstreamError:
|
---|
602 | uc := e.uc
|
---|
603 |
|
---|
604 | uc.forEachDownstream(func(dc *downstreamConn) {
|
---|
605 | sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s: %v", uc.network.GetName(), e.err))
|
---|
606 | })
|
---|
607 | uc.network.lastError = e.err
|
---|
608 | case eventUpstreamMessage:
|
---|
609 | msg, uc := e.msg, e.uc
|
---|
610 | if uc.isClosed() {
|
---|
611 | uc.logger.Printf("ignoring message on closed connection: %v", msg)
|
---|
612 | break
|
---|
613 | }
|
---|
614 | if err := uc.handleMessage(context.TODO(), msg); err != nil {
|
---|
615 | uc.logger.Printf("failed to handle message %q: %v", msg, err)
|
---|
616 | }
|
---|
617 | case eventChannelDetach:
|
---|
618 | uc, name := e.uc, e.name
|
---|
619 | c := uc.network.channels.Value(name)
|
---|
620 | if c == nil || c.Detached {
|
---|
621 | continue
|
---|
622 | }
|
---|
623 | uc.network.detach(c)
|
---|
624 | if err := uc.srv.db.StoreChannel(context.TODO(), uc.network.ID, c); err != nil {
|
---|
625 | u.logger.Printf("failed to store updated detached channel %q: %v", c.Name, err)
|
---|
626 | }
|
---|
627 | case eventDownstreamConnected:
|
---|
628 | dc := e.dc
|
---|
629 |
|
---|
630 | if dc.network != nil {
|
---|
631 | dc.monitored.SetCasemapping(dc.network.casemap)
|
---|
632 | }
|
---|
633 |
|
---|
634 | if err := dc.welcome(context.TODO()); err != nil {
|
---|
635 | dc.logger.Printf("failed to handle new registered connection: %v", err)
|
---|
636 | break
|
---|
637 | }
|
---|
638 |
|
---|
639 | u.downstreamConns = append(u.downstreamConns, dc)
|
---|
640 |
|
---|
641 | dc.forEachNetwork(func(network *network) {
|
---|
642 | if network.lastError != nil {
|
---|
643 | sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s: %v", network.GetName(), network.lastError))
|
---|
644 | }
|
---|
645 | })
|
---|
646 |
|
---|
647 | u.forEachUpstream(func(uc *upstreamConn) {
|
---|
648 | uc.updateAway()
|
---|
649 | })
|
---|
650 | case eventDownstreamDisconnected:
|
---|
651 | dc := e.dc
|
---|
652 |
|
---|
653 | for i := range u.downstreamConns {
|
---|
654 | if u.downstreamConns[i] == dc {
|
---|
655 | u.downstreamConns = append(u.downstreamConns[:i], u.downstreamConns[i+1:]...)
|
---|
656 | break
|
---|
657 | }
|
---|
658 | }
|
---|
659 |
|
---|
660 | dc.forEachNetwork(func(net *network) {
|
---|
661 | net.storeClientDeliveryReceipts(context.TODO(), dc.clientName)
|
---|
662 | })
|
---|
663 |
|
---|
664 | u.forEachUpstream(func(uc *upstreamConn) {
|
---|
665 | uc.cancelPendingCommandsByDownstreamID(dc.id)
|
---|
666 | uc.updateAway()
|
---|
667 | uc.updateMonitor()
|
---|
668 | })
|
---|
669 | case eventDownstreamMessage:
|
---|
670 | msg, dc := e.msg, e.dc
|
---|
671 | if dc.isClosed() {
|
---|
672 | dc.logger.Printf("ignoring message on closed connection: %v", msg)
|
---|
673 | break
|
---|
674 | }
|
---|
675 | err := dc.handleMessage(context.TODO(), msg)
|
---|
676 | if ircErr, ok := err.(ircError); ok {
|
---|
677 | ircErr.Message.Prefix = dc.srv.prefix()
|
---|
678 | dc.SendMessage(ircErr.Message)
|
---|
679 | } else if err != nil {
|
---|
680 | dc.logger.Printf("failed to handle message %q: %v", msg, err)
|
---|
681 | dc.Close()
|
---|
682 | }
|
---|
683 | case eventBroadcast:
|
---|
684 | msg := e.msg
|
---|
685 | u.forEachDownstream(func(dc *downstreamConn) {
|
---|
686 | dc.SendMessage(msg)
|
---|
687 | })
|
---|
688 | case eventUserUpdate:
|
---|
689 | // copy the user record because we'll mutate it
|
---|
690 | record := u.User
|
---|
691 |
|
---|
692 | if e.password != nil {
|
---|
693 | record.Password = *e.password
|
---|
694 | }
|
---|
695 | if e.admin != nil {
|
---|
696 | record.Admin = *e.admin
|
---|
697 | }
|
---|
698 |
|
---|
699 | e.done <- u.updateUser(context.TODO(), &record)
|
---|
700 |
|
---|
701 | // If the password was updated, kill all downstream connections to
|
---|
702 | // force them to re-authenticate with the new credentials.
|
---|
703 | if e.password != nil {
|
---|
704 | u.forEachDownstream(func(dc *downstreamConn) {
|
---|
705 | dc.Close()
|
---|
706 | })
|
---|
707 | }
|
---|
708 | case eventStop:
|
---|
709 | u.forEachDownstream(func(dc *downstreamConn) {
|
---|
710 | dc.Close()
|
---|
711 | })
|
---|
712 | for _, n := range u.networks {
|
---|
713 | n.stop()
|
---|
714 |
|
---|
715 | n.delivered.ForEachClient(func(clientName string) {
|
---|
716 | n.storeClientDeliveryReceipts(context.TODO(), clientName)
|
---|
717 | })
|
---|
718 | }
|
---|
719 | return
|
---|
720 | default:
|
---|
721 | panic(fmt.Sprintf("received unknown event type: %T", e))
|
---|
722 | }
|
---|
723 | }
|
---|
724 | }
|
---|
725 |
|
---|
726 | func (u *user) handleUpstreamDisconnected(uc *upstreamConn) {
|
---|
727 | uc.network.conn = nil
|
---|
728 |
|
---|
729 | uc.abortPendingCommands()
|
---|
730 |
|
---|
731 | for _, entry := range uc.channels.innerMap {
|
---|
732 | uch := entry.value.(*upstreamChannel)
|
---|
733 | uch.updateAutoDetach(0)
|
---|
734 | }
|
---|
735 |
|
---|
736 | netIDStr := fmt.Sprintf("%v", uc.network.ID)
|
---|
737 | uc.forEachDownstream(func(dc *downstreamConn) {
|
---|
738 | dc.updateSupportedCaps()
|
---|
739 | })
|
---|
740 |
|
---|
741 | // If the network has been removed, don't send a state change notification
|
---|
742 | found := false
|
---|
743 | for _, net := range u.networks {
|
---|
744 | if net == uc.network {
|
---|
745 | found = true
|
---|
746 | break
|
---|
747 | }
|
---|
748 | }
|
---|
749 | if !found {
|
---|
750 | return
|
---|
751 | }
|
---|
752 |
|
---|
753 | u.forEachDownstream(func(dc *downstreamConn) {
|
---|
754 | if dc.caps["soju.im/bouncer-networks-notify"] {
|
---|
755 | dc.SendMessage(&irc.Message{
|
---|
756 | Prefix: dc.srv.prefix(),
|
---|
757 | Command: "BOUNCER",
|
---|
758 | Params: []string{"NETWORK", netIDStr, "state=disconnected"},
|
---|
759 | })
|
---|
760 | }
|
---|
761 | })
|
---|
762 |
|
---|
763 | if uc.network.lastError == nil {
|
---|
764 | uc.forEachDownstream(func(dc *downstreamConn) {
|
---|
765 | if !dc.caps["soju.im/bouncer-networks"] {
|
---|
766 | sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s", uc.network.GetName()))
|
---|
767 | }
|
---|
768 | })
|
---|
769 | }
|
---|
770 | }
|
---|
771 |
|
---|
772 | func (u *user) addNetwork(network *network) {
|
---|
773 | u.networks = append(u.networks, network)
|
---|
774 |
|
---|
775 | sort.Slice(u.networks, func(i, j int) bool {
|
---|
776 | return u.networks[i].ID < u.networks[j].ID
|
---|
777 | })
|
---|
778 |
|
---|
779 | go network.run()
|
---|
780 | }
|
---|
781 |
|
---|
782 | func (u *user) removeNetwork(network *network) {
|
---|
783 | network.stop()
|
---|
784 |
|
---|
785 | u.forEachDownstream(func(dc *downstreamConn) {
|
---|
786 | if dc.network != nil && dc.network == network {
|
---|
787 | dc.Close()
|
---|
788 | }
|
---|
789 | })
|
---|
790 |
|
---|
791 | for i, net := range u.networks {
|
---|
792 | if net == network {
|
---|
793 | u.networks = append(u.networks[:i], u.networks[i+1:]...)
|
---|
794 | return
|
---|
795 | }
|
---|
796 | }
|
---|
797 |
|
---|
798 | panic("tried to remove a non-existing network")
|
---|
799 | }
|
---|
800 |
|
---|
801 | func (u *user) checkNetwork(record *Network) error {
|
---|
802 | url, err := record.URL()
|
---|
803 | if err != nil {
|
---|
804 | return err
|
---|
805 | }
|
---|
806 | if url.User != nil {
|
---|
807 | return fmt.Errorf("%v:// URL must not have username and password information", url.Scheme)
|
---|
808 | }
|
---|
809 | if url.RawQuery != "" {
|
---|
810 | return fmt.Errorf("%v:// URL must not have query values", url.Scheme)
|
---|
811 | }
|
---|
812 | if url.Fragment != "" {
|
---|
813 | return fmt.Errorf("%v:// URL must not have a fragment", url.Scheme)
|
---|
814 | }
|
---|
815 | switch url.Scheme {
|
---|
816 | case "ircs", "irc+insecure":
|
---|
817 | if url.Host == "" {
|
---|
818 | return fmt.Errorf("%v:// URL must have a host", url.Scheme)
|
---|
819 | }
|
---|
820 | if url.Path != "" {
|
---|
821 | return fmt.Errorf("%v:// URL must not have a path", url.Scheme)
|
---|
822 | }
|
---|
823 | case "irc+unix", "unix":
|
---|
824 | if url.Host != "" {
|
---|
825 | return fmt.Errorf("%v:// URL must not have a host", url.Scheme)
|
---|
826 | }
|
---|
827 | if url.Path == "" {
|
---|
828 | return fmt.Errorf("%v:// URL must have a path", url.Scheme)
|
---|
829 | }
|
---|
830 | default:
|
---|
831 | return fmt.Errorf("unknown URL scheme %q", url.Scheme)
|
---|
832 | }
|
---|
833 |
|
---|
834 | if record.GetName() == "" {
|
---|
835 | return fmt.Errorf("network name cannot be empty")
|
---|
836 | }
|
---|
837 | if strings.HasPrefix(record.GetName(), "-") {
|
---|
838 | // Can be mixed up with flags when sending commands to the service
|
---|
839 | return fmt.Errorf("network name cannot start with a dash character")
|
---|
840 | }
|
---|
841 |
|
---|
842 | for _, net := range u.networks {
|
---|
843 | if net.GetName() == record.GetName() && net.ID != record.ID {
|
---|
844 | return fmt.Errorf("a network with the name %q already exists", record.GetName())
|
---|
845 | }
|
---|
846 | }
|
---|
847 |
|
---|
848 | return nil
|
---|
849 | }
|
---|
850 |
|
---|
851 | func (u *user) createNetwork(ctx context.Context, record *Network) (*network, error) {
|
---|
852 | if record.ID != 0 {
|
---|
853 | panic("tried creating an already-existing network")
|
---|
854 | }
|
---|
855 |
|
---|
856 | if err := u.checkNetwork(record); err != nil {
|
---|
857 | return nil, err
|
---|
858 | }
|
---|
859 |
|
---|
860 | if max := u.srv.Config().MaxUserNetworks; max >= 0 && len(u.networks) >= max {
|
---|
861 | return nil, fmt.Errorf("maximum number of networks reached")
|
---|
862 | }
|
---|
863 |
|
---|
864 | network := newNetwork(u, record, nil)
|
---|
865 | err := u.srv.db.StoreNetwork(ctx, u.ID, &network.Network)
|
---|
866 | if err != nil {
|
---|
867 | return nil, err
|
---|
868 | }
|
---|
869 |
|
---|
870 | u.addNetwork(network)
|
---|
871 |
|
---|
872 | idStr := fmt.Sprintf("%v", network.ID)
|
---|
873 | attrs := getNetworkAttrs(network)
|
---|
874 | u.forEachDownstream(func(dc *downstreamConn) {
|
---|
875 | if dc.caps["soju.im/bouncer-networks-notify"] {
|
---|
876 | dc.SendMessage(&irc.Message{
|
---|
877 | Prefix: dc.srv.prefix(),
|
---|
878 | Command: "BOUNCER",
|
---|
879 | Params: []string{"NETWORK", idStr, attrs.String()},
|
---|
880 | })
|
---|
881 | }
|
---|
882 | })
|
---|
883 |
|
---|
884 | return network, nil
|
---|
885 | }
|
---|
886 |
|
---|
887 | func (u *user) updateNetwork(ctx context.Context, record *Network) (*network, error) {
|
---|
888 | if record.ID == 0 {
|
---|
889 | panic("tried updating a new network")
|
---|
890 | }
|
---|
891 |
|
---|
892 | // If the realname is reset to the default, just wipe the per-network
|
---|
893 | // setting
|
---|
894 | if record.Realname == u.Realname {
|
---|
895 | record.Realname = ""
|
---|
896 | }
|
---|
897 |
|
---|
898 | if err := u.checkNetwork(record); err != nil {
|
---|
899 | return nil, err
|
---|
900 | }
|
---|
901 |
|
---|
902 | network := u.getNetworkByID(record.ID)
|
---|
903 | if network == nil {
|
---|
904 | panic("tried updating a non-existing network")
|
---|
905 | }
|
---|
906 |
|
---|
907 | if err := u.srv.db.StoreNetwork(ctx, u.ID, record); err != nil {
|
---|
908 | return nil, err
|
---|
909 | }
|
---|
910 |
|
---|
911 | // Most network changes require us to re-connect to the upstream server
|
---|
912 |
|
---|
913 | channels := make([]Channel, 0, network.channels.Len())
|
---|
914 | for _, entry := range network.channels.innerMap {
|
---|
915 | ch := entry.value.(*Channel)
|
---|
916 | channels = append(channels, *ch)
|
---|
917 | }
|
---|
918 |
|
---|
919 | updatedNetwork := newNetwork(u, record, channels)
|
---|
920 |
|
---|
921 | // If we're currently connected, disconnect and perform the necessary
|
---|
922 | // bookkeeping
|
---|
923 | if network.conn != nil {
|
---|
924 | network.stop()
|
---|
925 | // Note: this will set network.conn to nil
|
---|
926 | u.handleUpstreamDisconnected(network.conn)
|
---|
927 | }
|
---|
928 |
|
---|
929 | // Patch downstream connections to use our fresh updated network
|
---|
930 | u.forEachDownstream(func(dc *downstreamConn) {
|
---|
931 | if dc.network != nil && dc.network == network {
|
---|
932 | dc.network = updatedNetwork
|
---|
933 | }
|
---|
934 | })
|
---|
935 |
|
---|
936 | // We need to remove the network after patching downstream connections,
|
---|
937 | // otherwise they'll get closed
|
---|
938 | u.removeNetwork(network)
|
---|
939 |
|
---|
940 | // The filesystem message store needs to be notified whenever the network
|
---|
941 | // is renamed
|
---|
942 | fsMsgStore, isFS := u.msgStore.(*fsMessageStore)
|
---|
943 | if isFS && updatedNetwork.GetName() != network.GetName() {
|
---|
944 | if err := fsMsgStore.RenameNetwork(&network.Network, &updatedNetwork.Network); err != nil {
|
---|
945 | network.logger.Printf("failed to update FS message store network name to %q: %v", updatedNetwork.GetName(), err)
|
---|
946 | }
|
---|
947 | }
|
---|
948 |
|
---|
949 | // This will re-connect to the upstream server
|
---|
950 | u.addNetwork(updatedNetwork)
|
---|
951 |
|
---|
952 | // TODO: only broadcast attributes that have changed
|
---|
953 | idStr := fmt.Sprintf("%v", updatedNetwork.ID)
|
---|
954 | attrs := getNetworkAttrs(updatedNetwork)
|
---|
955 | u.forEachDownstream(func(dc *downstreamConn) {
|
---|
956 | if dc.caps["soju.im/bouncer-networks-notify"] {
|
---|
957 | dc.SendMessage(&irc.Message{
|
---|
958 | Prefix: dc.srv.prefix(),
|
---|
959 | Command: "BOUNCER",
|
---|
960 | Params: []string{"NETWORK", idStr, attrs.String()},
|
---|
961 | })
|
---|
962 | }
|
---|
963 | })
|
---|
964 |
|
---|
965 | return updatedNetwork, nil
|
---|
966 | }
|
---|
967 |
|
---|
968 | func (u *user) deleteNetwork(ctx context.Context, id int64) error {
|
---|
969 | network := u.getNetworkByID(id)
|
---|
970 | if network == nil {
|
---|
971 | panic("tried deleting a non-existing network")
|
---|
972 | }
|
---|
973 |
|
---|
974 | if err := u.srv.db.DeleteNetwork(ctx, network.ID); err != nil {
|
---|
975 | return err
|
---|
976 | }
|
---|
977 |
|
---|
978 | u.removeNetwork(network)
|
---|
979 |
|
---|
980 | idStr := fmt.Sprintf("%v", network.ID)
|
---|
981 | u.forEachDownstream(func(dc *downstreamConn) {
|
---|
982 | if dc.caps["soju.im/bouncer-networks-notify"] {
|
---|
983 | dc.SendMessage(&irc.Message{
|
---|
984 | Prefix: dc.srv.prefix(),
|
---|
985 | Command: "BOUNCER",
|
---|
986 | Params: []string{"NETWORK", idStr, "*"},
|
---|
987 | })
|
---|
988 | }
|
---|
989 | })
|
---|
990 |
|
---|
991 | return nil
|
---|
992 | }
|
---|
993 |
|
---|
994 | func (u *user) updateUser(ctx context.Context, record *User) error {
|
---|
995 | if u.ID != record.ID {
|
---|
996 | panic("ID mismatch when updating user")
|
---|
997 | }
|
---|
998 |
|
---|
999 | realnameUpdated := u.Realname != record.Realname
|
---|
1000 | if err := u.srv.db.StoreUser(ctx, record); err != nil {
|
---|
1001 | return fmt.Errorf("failed to update user %q: %v", u.Username, err)
|
---|
1002 | }
|
---|
1003 | u.User = *record
|
---|
1004 |
|
---|
1005 | if realnameUpdated {
|
---|
1006 | // Re-connect to networks which use the default realname
|
---|
1007 | var needUpdate []Network
|
---|
1008 | for _, net := range u.networks {
|
---|
1009 | if net.Realname == "" {
|
---|
1010 | needUpdate = append(needUpdate, net.Network)
|
---|
1011 | }
|
---|
1012 | }
|
---|
1013 |
|
---|
1014 | var netErr error
|
---|
1015 | for _, net := range needUpdate {
|
---|
1016 | if _, err := u.updateNetwork(ctx, &net); err != nil {
|
---|
1017 | netErr = err
|
---|
1018 | }
|
---|
1019 | }
|
---|
1020 | if netErr != nil {
|
---|
1021 | return netErr
|
---|
1022 | }
|
---|
1023 | }
|
---|
1024 |
|
---|
1025 | return nil
|
---|
1026 | }
|
---|
1027 |
|
---|
1028 | func (u *user) stop() {
|
---|
1029 | u.events <- eventStop{}
|
---|
1030 | <-u.done
|
---|
1031 | }
|
---|
1032 |
|
---|
1033 | func (u *user) hasPersistentMsgStore() bool {
|
---|
1034 | if u.msgStore == nil {
|
---|
1035 | return false
|
---|
1036 | }
|
---|
1037 | _, isMem := u.msgStore.(*memoryMessageStore)
|
---|
1038 | return !isMem
|
---|
1039 | }
|
---|
1040 |
|
---|
1041 | // localAddrForHost returns the local address to use when connecting to host.
|
---|
1042 | // A nil address is returned when the OS should automatically pick one.
|
---|
1043 | func (u *user) localTCPAddrForHost(ctx context.Context, host string) (*net.TCPAddr, error) {
|
---|
1044 | upstreamUserIPs := u.srv.Config().UpstreamUserIPs
|
---|
1045 | if len(upstreamUserIPs) == 0 {
|
---|
1046 | return nil, nil
|
---|
1047 | }
|
---|
1048 |
|
---|
1049 | ips, err := net.DefaultResolver.LookupIP(ctx, "ip", host)
|
---|
1050 | if err != nil {
|
---|
1051 | return nil, err
|
---|
1052 | }
|
---|
1053 |
|
---|
1054 | wantIPv6 := false
|
---|
1055 | for _, ip := range ips {
|
---|
1056 | if ip.To4() == nil {
|
---|
1057 | wantIPv6 = true
|
---|
1058 | break
|
---|
1059 | }
|
---|
1060 | }
|
---|
1061 |
|
---|
1062 | var ipNet *net.IPNet
|
---|
1063 | for _, in := range upstreamUserIPs {
|
---|
1064 | if wantIPv6 == (in.IP.To4() == nil) {
|
---|
1065 | ipNet = in
|
---|
1066 | break
|
---|
1067 | }
|
---|
1068 | }
|
---|
1069 | if ipNet == nil {
|
---|
1070 | return nil, nil
|
---|
1071 | }
|
---|
1072 |
|
---|
1073 | var ipInt big.Int
|
---|
1074 | ipInt.SetBytes(ipNet.IP)
|
---|
1075 | ipInt.Add(&ipInt, big.NewInt(u.ID+1))
|
---|
1076 | ip := net.IP(ipInt.Bytes())
|
---|
1077 | if !ipNet.Contains(ip) {
|
---|
1078 | return nil, fmt.Errorf("IP network %v too small", ipNet)
|
---|
1079 | }
|
---|
1080 |
|
---|
1081 | return &net.TCPAddr{IP: ip}, nil
|
---|
1082 | }
|
---|