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