soju/user.go
Simon Ser b49552ed5b Don't send bouncer-networks updates for removed networks
We may asynchronously receive an upstream's connection status after
the network has been removed.

Sending an update is pretty bad since it's interpreted by clients
as a new network.
2023-10-03 12:01:59 +02:00

1306 lines
32 KiB
Go

package soju
import (
"context"
"crypto/sha256"
"encoding/binary"
"encoding/hex"
"errors"
"fmt"
"math/big"
"net"
"sort"
"strings"
"sync/atomic"
"time"
"git.sr.ht/~emersion/soju/xirc"
"github.com/SherClockHolmes/webpush-go"
"gopkg.in/irc.v4"
"git.sr.ht/~emersion/soju/database"
"git.sr.ht/~emersion/soju/msgstore"
)
type UserUpdateFunc func(record *database.User) error
type event interface{}
type eventUpstreamMessage struct {
msg *irc.Message
uc *upstreamConn
}
type eventUpstreamConnectionError struct {
net *network
err error
}
type eventUpstreamConnected struct {
uc *upstreamConn
}
type eventUpstreamDisconnected struct {
uc *upstreamConn
}
type eventUpstreamError struct {
uc *upstreamConn
err error
}
type eventDownstreamMessage struct {
msg *irc.Message
dc *downstreamConn
}
type eventDownstreamConnected struct {
dc *downstreamConn
}
type eventDownstreamDisconnected struct {
dc *downstreamConn
}
type eventChannelDetach struct {
uc *upstreamConn
name string
}
type eventBroadcast struct {
msg *irc.Message
}
type eventStop struct{}
type eventUserUpdate struct {
password *string
admin *bool
enabled *bool
done chan error
}
type eventTryRegainNick struct {
uc *upstreamConn
nick string
}
type eventUserRun struct {
params []string
print chan string
ret chan error
}
type deliveredClientMap map[string]string // client name -> msg ID
type deliveredStore struct {
m xirc.CaseMappingMap[deliveredClientMap]
}
func newDeliveredStore(cm xirc.CaseMapping) deliveredStore {
return deliveredStore{xirc.NewCaseMappingMap[deliveredClientMap](cm)}
}
func (ds deliveredStore) HasTarget(target string) bool {
return ds.m.Get(target) != nil
}
func (ds deliveredStore) LoadID(target, clientName string) string {
clients := ds.m.Get(target)
if clients == nil {
return ""
}
return clients[clientName]
}
func (ds deliveredStore) StoreID(target, clientName, msgID string) {
clients := ds.m.Get(target)
if clients == nil {
clients = make(deliveredClientMap)
ds.m.Set(target, clients)
}
clients[clientName] = msgID
}
func (ds deliveredStore) ForEachTarget(f func(target string)) {
ds.m.ForEach(func(name string, _ deliveredClientMap) {
f(name)
})
}
func (ds deliveredStore) ForEachClient(f func(clientName string)) {
clients := make(map[string]struct{})
ds.m.ForEach(func(name string, delivered deliveredClientMap) {
for clientName := range delivered {
clients[clientName] = struct{}{}
}
})
for clientName := range clients {
f(clientName)
}
}
type network struct {
database.Network
user *user
logger Logger
stopped chan struct{}
conn *upstreamConn
channels xirc.CaseMappingMap[*database.Channel]
delivered deliveredStore
pushTargets xirc.CaseMappingMap[time.Time]
lastError error
casemap xirc.CaseMapping
}
func newNetwork(user *user, record *database.Network, channels []database.Channel) *network {
logger := &prefixLogger{user.logger, fmt.Sprintf("network %q: ", record.GetName())}
// Initialize maps with the most strict case-mapping to avoid collisions:
// we don't know which case-mapping will be used by the upstream server yet
cm := xirc.CaseMappingASCII
m := xirc.NewCaseMappingMap[*database.Channel](cm)
for _, ch := range channels {
ch := ch
m.Set(ch.Name, &ch)
}
return &network{
Network: *record,
user: user,
logger: logger,
stopped: make(chan struct{}),
channels: m,
delivered: newDeliveredStore(cm),
pushTargets: xirc.NewCaseMappingMap[time.Time](cm),
casemap: stdCaseMapping,
}
}
func (net *network) forEachDownstream(f func(*downstreamConn)) {
for _, dc := range net.user.downstreamConns {
if dc.network != net {
continue
}
f(dc)
}
}
func (net *network) isStopped() bool {
select {
case <-net.stopped:
return true
default:
return false
}
}
func (net *network) equalCasemap(a, b string) bool {
return net.casemap(a) == net.casemap(b)
}
func userIdent(u *database.User) string {
// The ident is a string we will send to upstream servers in clear-text.
// For privacy reasons, make sure it doesn't expose any meaningful user
// metadata. We just use the base64-encoded hashed ID, so that people don't
// start relying on the string being an integer or following a pattern.
var b [64]byte
binary.LittleEndian.PutUint64(b[:], uint64(u.ID))
h := sha256.Sum256(b[:])
return hex.EncodeToString(h[:16])
}
func (net *network) runConn(ctx context.Context) error {
net.user.srv.metrics.upstreams.Add(1)
defer net.user.srv.metrics.upstreams.Add(-1)
done := ctx.Done()
ctx, cancel := context.WithTimeout(ctx, time.Minute)
defer cancel()
uc, err := connectToUpstream(ctx, net)
if err != nil {
return fmt.Errorf("failed to connect: %w", err)
}
defer uc.Close()
// The context is cancelled by the caller when the network is stopped.
go func() {
<-done
uc.Close()
}()
if net.user.srv.Identd != nil {
net.user.srv.Identd.Store(uc.RemoteAddr().String(), uc.LocalAddr().String(), userIdent(&net.user.User))
defer net.user.srv.Identd.Delete(uc.RemoteAddr().String(), uc.LocalAddr().String())
}
// TODO: this is racy, we're not running in the user goroutine yet
// uc.register accesses user/network DB records
uc.register(ctx)
if err := uc.runUntilRegistered(ctx); err != nil {
return fmt.Errorf("failed to register: %w", err)
}
net.user.events <- eventUpstreamConnected{uc}
defer func() {
net.user.events <- eventUpstreamDisconnected{uc}
}()
if err := uc.readMessages(net.user.events); err != nil {
return fmt.Errorf("failed to handle messages: %w", err)
}
return nil
}
func (net *network) run() {
if !net.user.Enabled || !net.Enabled {
return
}
ctx, cancel := context.WithCancel(context.TODO())
go func() {
<-net.stopped
cancel()
}()
var lastTry time.Time
backoff := newBackoffer(retryConnectMinDelay, retryConnectMaxDelay, retryConnectJitter)
for {
if net.isStopped() {
return
}
delay := backoff.Next() - time.Now().Sub(lastTry)
if delay > 0 {
net.logger.Printf("waiting %v before trying to reconnect to %q", delay.Truncate(time.Second), net.Addr)
time.Sleep(delay)
}
lastTry = time.Now()
if err := net.runConn(ctx); err != nil {
text := err.Error()
temp := true
var regErr registrationError
if errors.As(err, &regErr) {
text = "failed to register: " + regErr.Reason()
temp = regErr.Temporary()
}
net.logger.Printf("connection error to %q: %v", net.Addr, text)
net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("connection error: %v", err)}
net.user.srv.metrics.upstreamConnectErrorsTotal.Inc()
if !temp {
return
}
} else {
backoff.Reset()
}
}
}
func (net *network) stop() {
if !net.isStopped() {
close(net.stopped)
}
}
func (net *network) detach(ch *database.Channel) {
if ch.Detached {
return
}
net.logger.Printf("detaching channel %q", ch.Name)
ch.Detached = true
if net.user.msgStore != nil {
nameCM := net.casemap(ch.Name)
lastID, err := net.user.msgStore.LastMsgID(&net.Network, nameCM, time.Now())
if err != nil {
net.logger.Printf("failed to get last message ID for channel %q: %v", ch.Name, err)
}
ch.DetachedInternalMsgID = lastID
}
if net.conn != nil {
uch := net.conn.channels.Get(ch.Name)
if uch != nil {
uch.updateAutoDetach(0)
}
}
net.forEachDownstream(func(dc *downstreamConn) {
dc.SendMessage(context.TODO(), &irc.Message{
Prefix: dc.prefix(),
Command: "PART",
Params: []string{ch.Name, "Detach"},
})
})
}
func (net *network) attach(ctx context.Context, ch *database.Channel) {
if !ch.Detached {
return
}
net.logger.Printf("attaching channel %q", ch.Name)
detachedMsgID := ch.DetachedInternalMsgID
ch.Detached = false
ch.DetachedInternalMsgID = ""
var uch *upstreamChannel
if net.conn != nil {
uch = net.conn.channels.Get(ch.Name)
net.conn.updateChannelAutoDetach(ch.Name)
}
net.forEachDownstream(func(dc *downstreamConn) {
dc.SendMessage(ctx, &irc.Message{
Prefix: dc.prefix(),
Command: "JOIN",
Params: []string{ch.Name},
})
if uch != nil {
forwardChannel(ctx, dc, uch)
}
if detachedMsgID != "" {
dc.sendTargetBacklog(ctx, net, ch.Name, detachedMsgID)
}
})
}
func (net *network) deleteChannel(ctx context.Context, name string) error {
ch := net.channels.Get(name)
if ch == nil {
return fmt.Errorf("unknown channel %q", name)
}
if net.conn != nil {
uch := net.conn.channels.Get(ch.Name)
if uch != nil {
uch.updateAutoDetach(0)
}
}
if err := net.user.srv.db.DeleteChannel(ctx, ch.ID); err != nil {
return err
}
net.channels.Del(name)
return nil
}
func (net *network) updateCasemapping(newCasemap xirc.CaseMapping) {
net.casemap = newCasemap
net.channels.SetCaseMapping(newCasemap)
net.delivered.m.SetCaseMapping(newCasemap)
net.pushTargets.SetCaseMapping(newCasemap)
if uc := net.conn; uc != nil {
uc.channels.SetCaseMapping(newCasemap)
uc.channels.ForEach(func(_ string, uch *upstreamChannel) {
uch.Members.SetCaseMapping(newCasemap)
})
uc.users.SetCaseMapping(newCasemap)
uc.monitored.SetCaseMapping(newCasemap)
}
net.forEachDownstream(func(dc *downstreamConn) {
dc.updateCasemapping()
})
}
func (net *network) storeClientDeliveryReceipts(ctx context.Context, clientName string) {
if !net.user.hasPersistentMsgStore() {
return
}
var receipts []database.DeliveryReceipt
net.delivered.ForEachTarget(func(target string) {
msgID := net.delivered.LoadID(target, clientName)
if msgID == "" {
return
}
receipts = append(receipts, database.DeliveryReceipt{
Target: target,
InternalMsgID: msgID,
})
})
if err := net.user.srv.db.StoreClientDeliveryReceipts(ctx, net.ID, clientName, receipts); err != nil {
net.logger.Printf("failed to store delivery receipts for client %q: %v", clientName, err)
}
}
func (net *network) isHighlight(msg *irc.Message) bool {
if msg.Command != "PRIVMSG" && msg.Command != "NOTICE" {
return false
}
text := msg.Params[1]
nick := database.GetNick(&net.user.User, &net.Network)
if net.conn != nil {
nick = net.conn.nick
}
// TODO: use case-mapping aware comparison here
return msg.Prefix.Name != nick && isHighlight(text, nick)
}
func (net *network) detachedMessageNeedsRelay(ch *database.Channel, msg *irc.Message) bool {
highlight := net.isHighlight(msg)
return ch.RelayDetached == database.FilterMessage || ((ch.RelayDetached == database.FilterHighlight || ch.RelayDetached == database.FilterDefault) && highlight)
}
func (net *network) autoSaveSASLPlain(ctx context.Context, username, password string) {
// User may have e.g. EXTERNAL mechanism configured. We do not want to
// automatically erase the key pair or any other credentials.
if net.SASL.Mechanism != "" && net.SASL.Mechanism != "PLAIN" {
return
}
net.logger.Printf("auto-saving SASL PLAIN credentials with username %q", username)
net.SASL.Mechanism = "PLAIN"
net.SASL.Plain.Username = username
net.SASL.Plain.Password = password
if err := net.user.srv.db.StoreNetwork(ctx, net.user.ID, &net.Network); err != nil {
net.logger.Printf("failed to save SASL PLAIN credentials: %v", err)
}
}
// broadcastWebPush broadcasts a Web Push message for the given IRC message.
//
// Broadcasting the message to all Web Push endpoints might take a while, so
// callers should call this function in a new goroutine.
func (net *network) broadcastWebPush(msg *irc.Message) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
defer cancel()
subs, err := net.user.srv.db.ListWebPushSubscriptions(ctx, net.user.ID, net.ID)
if err != nil {
net.logger.Printf("failed to list Web push subscriptions: %v", err)
return
}
for _, sub := range subs {
err := net.user.srv.sendWebPush(ctx, &webpush.Subscription{
Endpoint: sub.Endpoint,
Keys: webpush.Keys{
Auth: sub.Keys.Auth,
P256dh: sub.Keys.P256DH,
},
}, sub.Keys.VAPID, msg)
if err == errWebPushSubscriptionExpired {
if err := net.user.srv.db.DeleteWebPushSubscription(ctx, sub.ID); err != nil {
net.logger.Printf("failed to delete expired Web Push subscription %q: %v", sub.Endpoint, err)
} else {
net.logger.Debugf("deleted expired Web Push subscription %q", sub.Endpoint)
}
} else if err != nil {
net.logger.Printf("failed to send Web push notification to endpoint %q: %v", sub.Endpoint, err)
// If it failed for any reason and is old, delete it
if time.Since(sub.UpdatedAt) > webpushPruneSubscriptionDelay {
if err := net.user.srv.db.DeleteWebPushSubscription(ctx, sub.ID); err != nil {
net.logger.Printf("failed to delete pruned Web Push subscription %q: %v", sub.Endpoint, err)
} else {
net.logger.Printf("deleted pruned Web Push subscription %q", sub.Endpoint)
}
}
}
}
}
type user struct {
database.User
srv *Server
logger Logger
events chan event
done chan struct{}
numDownstreamConns atomic.Int64
networks []*network
downstreamConns []*downstreamConn
msgStore msgstore.Store
}
func newUser(srv *Server, record *database.User) *user {
logger := &prefixLogger{srv.Logger, fmt.Sprintf("user %q: ", record.Username)}
var msgStore msgstore.Store
switch srv.Config().MsgStoreDriver {
case "fs":
msgStore = msgstore.NewFSStore(srv.Config().MsgStorePath, record)
case "db":
msgStore = msgstore.NewDBStore(srv.db)
case "memory":
msgStore = msgstore.NewMemoryStore()
}
return &user{
User: *record,
srv: srv,
logger: logger,
events: make(chan event, 64),
done: make(chan struct{}),
msgStore: msgStore,
}
}
func (u *user) forEachUpstream(f func(uc *upstreamConn)) {
for _, network := range u.networks {
if network.conn == nil {
continue
}
f(network.conn)
}
}
func (u *user) getNetwork(name string) *network {
for _, network := range u.networks {
if network.Addr == name {
return network
}
if network.Name != "" && network.Name == name {
return network
}
}
return nil
}
func (u *user) getNetworkByID(id int64) *network {
for _, net := range u.networks {
if net.ID == id {
return net
}
}
return nil
}
func (u *user) run() {
defer func() {
if u.msgStore != nil {
if err := u.msgStore.Close(); err != nil {
u.logger.Printf("failed to close message store for user %q: %v", u.Username, err)
}
}
close(u.done)
}()
networks, err := u.srv.db.ListNetworks(context.TODO(), u.ID)
if err != nil {
u.logger.Printf("failed to list networks for user %q: %v", u.Username, err)
return
}
sort.Slice(networks, func(i, j int) bool {
return networks[i].ID < networks[j].ID
})
for _, record := range networks {
record := record
channels, err := u.srv.db.ListChannels(context.TODO(), record.ID)
if err != nil {
u.logger.Printf("failed to list channels for user %q, network %q: %v", u.Username, record.GetName(), err)
continue
}
network := newNetwork(u, &record, channels)
u.networks = append(u.networks, network)
if u.hasPersistentMsgStore() {
receipts, err := u.srv.db.ListDeliveryReceipts(context.TODO(), record.ID)
if err != nil {
u.logger.Printf("failed to load delivery receipts for user %q, network %q: %v", u.Username, network.GetName(), err)
return
}
for _, rcpt := range receipts {
network.delivered.StoreID(rcpt.Target, rcpt.Client, rcpt.InternalMsgID)
}
}
go network.run()
}
for e := range u.events {
switch e := e.(type) {
case eventUpstreamConnected:
uc := e.uc
uc.network.conn = uc
uc.updateAway()
uc.updateMonitor()
ctx := context.TODO()
uc.forEachDownstream(func(dc *downstreamConn) {
dc.updateSupportedCaps(ctx)
if !dc.caps.IsEnabled("soju.im/bouncer-networks") {
sendServiceNOTICE(dc, fmt.Sprintf("connected to %s", uc.network.GetName()))
}
dc.updateNick(ctx)
dc.updateHost(ctx)
dc.updateRealname(ctx)
dc.updateAccount(ctx)
dc.updateCasemapping()
})
u.notifyBouncerNetworkState(uc.network.ID, irc.Tags{
"state": "connected",
"error": "",
})
uc.network.lastError = nil
case eventUpstreamDisconnected:
u.handleUpstreamDisconnected(e.uc)
case eventUpstreamConnectionError:
net := e.net
stopped := false
select {
case <-net.stopped:
stopped = true
default:
}
if !stopped && (net.lastError == nil || net.lastError.Error() != e.err.Error()) {
net.forEachDownstream(func(dc *downstreamConn) {
sendServiceNOTICE(dc, fmt.Sprintf("failed connecting/registering to %s: %v", net.GetName(), e.err))
})
}
net.lastError = e.err
u.notifyBouncerNetworkState(net.ID, irc.Tags{
"error": net.lastError.Error(),
})
case eventUpstreamError:
uc := e.uc
uc.forEachDownstream(func(dc *downstreamConn) {
sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s: %v", uc.network.GetName(), e.err))
})
uc.network.lastError = e.err
u.notifyBouncerNetworkState(uc.network.ID, irc.Tags{
"error": uc.network.lastError.Error(),
})
case eventUpstreamMessage:
msg, uc := e.msg, e.uc
if uc.isClosed() {
uc.logger.Printf("ignoring message on closed connection: %v", msg)
break
}
if err := uc.handleMessage(context.TODO(), msg); err != nil {
uc.logger.Printf("failed to handle message %q: %v", msg, err)
}
case eventChannelDetach:
uc, name := e.uc, e.name
c := uc.network.channels.Get(name)
if c == nil || c.Detached {
continue
}
uc.network.detach(c)
if err := uc.srv.db.StoreChannel(context.TODO(), uc.network.ID, c); err != nil {
u.logger.Printf("failed to store updated detached channel %q: %v", c.Name, err)
}
case eventDownstreamConnected:
dc := e.dc
ctx := context.TODO()
if dc.network != nil {
dc.monitored.SetCaseMapping(dc.network.casemap)
}
if !u.Enabled && u.srv.Config().EnableUsersOnAuth {
err := u.updateUser(ctx, func(record *database.User) error {
record.Enabled = true
return nil
})
if err != nil {
dc.logger.Printf("failed to enable user after successful authentication: %v", err)
}
}
if !u.Enabled {
dc.SendMessage(ctx, &irc.Message{
Command: "ERROR",
Params: []string{"This bouncer account is disabled"},
})
// TODO: close dc after the error message is sent
break
}
if err := dc.welcome(ctx, u); err != nil {
if ircErr, ok := err.(ircError); ok {
msg := ircErr.Message.Copy()
msg.Prefix = dc.srv.prefix()
dc.SendMessage(ctx, msg)
} else {
dc.SendMessage(ctx, &irc.Message{
Command: "ERROR",
Params: []string{"Internal server error"},
})
}
dc.logger.Printf("failed to handle new registered connection: %v", err)
// TODO: close dc after the error message is sent
break
}
u.downstreamConns = append(u.downstreamConns, dc)
u.numDownstreamConns.Add(1)
dc.forEachNetwork(func(network *network) {
if network.lastError != nil {
sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s: %v", network.GetName(), network.lastError))
}
})
u.forEachUpstream(func(uc *upstreamConn) {
uc.updateAway()
})
u.bumpDownstreamInteractionTime(ctx)
case eventDownstreamDisconnected:
dc := e.dc
ctx := context.TODO()
for i := range u.downstreamConns {
if u.downstreamConns[i] == dc {
u.downstreamConns = append(u.downstreamConns[:i], u.downstreamConns[i+1:]...)
u.numDownstreamConns.Add(-1)
break
}
}
dc.forEachNetwork(func(net *network) {
net.storeClientDeliveryReceipts(ctx, dc.clientName)
})
u.forEachUpstream(func(uc *upstreamConn) {
uc.cancelPendingCommandsByDownstreamID(dc.id)
uc.updateAway()
uc.updateMonitor()
})
u.bumpDownstreamInteractionTime(ctx)
case eventDownstreamMessage:
msg, dc := e.msg, e.dc
if dc.isClosed() {
dc.logger.Printf("ignoring message on closed connection: %v", msg)
break
}
err := dc.handleMessage(context.TODO(), msg)
if ircErr, ok := err.(ircError); ok {
ircErr.Message.Prefix = dc.srv.prefix()
dc.SendMessage(context.TODO(), ircErr.Message)
} else if err != nil {
dc.logger.Printf("failed to handle message %q: %v", msg, err)
dc.Close()
}
case eventBroadcast:
msg := e.msg
for _, dc := range u.downstreamConns {
dc.SendMessage(context.TODO(), msg)
}
case eventUserUpdate:
e.done <- u.updateUser(context.TODO(), func(record *database.User) error {
if e.password != nil {
record.Password = *e.password
}
if e.admin != nil {
record.Admin = *e.admin
}
if e.enabled != nil {
record.Enabled = *e.enabled
}
return nil
})
// If the password was updated, kill all downstream connections to
// force them to re-authenticate with the new credentials.
if e.password != nil {
for _, dc := range u.downstreamConns {
dc.Close()
}
}
case eventTryRegainNick:
e.uc.tryRegainNick(e.nick)
case eventUserRun:
ctx := context.TODO()
err := handleServiceCommand(&serviceContext{
Context: ctx,
user: u,
srv: u.srv,
admin: u.Admin,
print: func(text string) {
// Avoid blocking on e.print in case our context is canceled.
// This is a no-op right now because we use context.TODO(),
// but might be useful later when we add timeouts.
select {
case <-ctx.Done():
case e.print <- text:
}
},
}, e.params)
select {
case <-ctx.Done():
case e.ret <- err:
}
case eventStop:
for _, dc := range u.downstreamConns {
dc.Close()
}
for _, n := range u.networks {
n.stop()
n.delivered.ForEachClient(func(clientName string) {
n.storeClientDeliveryReceipts(context.TODO(), clientName)
})
}
return
default:
panic(fmt.Sprintf("received unknown event type: %T", e))
}
}
}
func (u *user) handleUpstreamDisconnected(uc *upstreamConn) {
uc.network.conn = nil
uc.stopRegainNickTimer()
uc.abortPendingCommands()
uc.channels.ForEach(func(_ string, uch *upstreamChannel) {
uch.updateAutoDetach(0)
})
uc.forEachDownstream(func(dc *downstreamConn) {
dc.updateSupportedCaps(context.TODO())
})
// If the network has been removed, don't send a state change notification
found := false
for _, net := range u.networks {
if net == uc.network {
found = true
break
}
}
if !found {
return
}
u.notifyBouncerNetworkState(uc.network.ID, irc.Tags{"state": "disconnected"})
if uc.network.lastError == nil {
uc.forEachDownstream(func(dc *downstreamConn) {
if !dc.caps.IsEnabled("soju.im/bouncer-networks") {
sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s", uc.network.GetName()))
}
})
}
}
func (u *user) notifyBouncerNetworkState(netID int64, attrs irc.Tags) {
// Don't send state updates for removed networks
found := false
for _, net := range u.networks {
if net.ID == netID {
found = true
break
}
}
if !found {
return
}
netIDStr := fmt.Sprintf("%v", netID)
for _, dc := range u.downstreamConns {
if dc.caps.IsEnabled("soju.im/bouncer-networks-notify") {
dc.SendMessage(context.TODO(), &irc.Message{
Prefix: dc.srv.prefix(),
Command: "BOUNCER",
Params: []string{"NETWORK", netIDStr, attrs.String()},
})
}
}
}
func (u *user) addNetwork(network *network) {
u.networks = append(u.networks, network)
sort.Slice(u.networks, func(i, j int) bool {
return u.networks[i].ID < u.networks[j].ID
})
go network.run()
}
func (u *user) removeNetwork(network *network) {
network.stop()
for _, dc := range u.downstreamConns {
if dc.network != nil && dc.network == network {
dc.Close()
}
}
for i, net := range u.networks {
if net == network {
u.networks = append(u.networks[:i], u.networks[i+1:]...)
return
}
}
panic("tried to remove a non-existing network")
}
func (u *user) checkNetwork(record *database.Network) error {
url, err := record.URL()
if err != nil {
return err
}
if url.User != nil {
return fmt.Errorf("%v:// URL must not have username and password information", url.Scheme)
}
if url.RawQuery != "" {
return fmt.Errorf("%v:// URL must not have query values", url.Scheme)
}
if url.Fragment != "" {
return fmt.Errorf("%v:// URL must not have a fragment", url.Scheme)
}
switch url.Scheme {
case "ircs", "irc+insecure":
if url.Host == "" {
return fmt.Errorf("%v:// URL must have a host", url.Scheme)
}
if url.Path != "" {
return fmt.Errorf("%v:// URL must not have a path", url.Scheme)
}
case "irc+unix", "unix":
if url.Host != "" {
return fmt.Errorf("%v:// URL must not have a host", url.Scheme)
}
if url.Path == "" {
return fmt.Errorf("%v:// URL must have a path", url.Scheme)
}
default:
return fmt.Errorf("unknown URL scheme %q", url.Scheme)
}
if record.GetName() == "" {
return fmt.Errorf("network name cannot be empty")
}
if strings.HasPrefix(record.GetName(), "-") {
// Can be mixed up with flags when sending commands to the service
return fmt.Errorf("network name cannot start with a dash character")
}
for _, net := range u.networks {
if net.GetName() == record.GetName() && net.ID != record.ID {
return fmt.Errorf("a network with the name %q already exists", record.GetName())
}
}
return nil
}
func (u *user) createNetwork(ctx context.Context, record *database.Network) (*network, error) {
if record.ID != 0 {
panic("tried creating an already-existing network")
}
if err := u.checkNetwork(record); err != nil {
return nil, err
}
if max := u.srv.Config().MaxUserNetworks; max >= 0 && len(u.networks) >= max {
return nil, fmt.Errorf("maximum number of networks reached")
}
network := newNetwork(u, record, nil)
err := u.srv.db.StoreNetwork(ctx, u.ID, &network.Network)
if err != nil {
return nil, err
}
u.addNetwork(network)
attrs := getNetworkAttrs(network)
u.notifyBouncerNetworkState(network.ID, attrs)
return network, nil
}
func (u *user) updateNetwork(ctx context.Context, record *database.Network) (*network, error) {
if record.ID == 0 {
panic("tried updating a new network")
}
// If the nickname/realname is reset to the default, just wipe the
// per-network setting
if record.Nick == u.Nick {
record.Nick = ""
}
if record.Realname == u.Realname {
record.Realname = ""
}
if err := u.checkNetwork(record); err != nil {
return nil, err
}
network := u.getNetworkByID(record.ID)
if network == nil {
panic("tried updating a non-existing network")
}
if err := u.srv.db.StoreNetwork(ctx, u.ID, record); err != nil {
return nil, err
}
// Most network changes require us to re-connect to the upstream server
channels := make([]database.Channel, 0, network.channels.Len())
network.channels.ForEach(func(_ string, ch *database.Channel) {
channels = append(channels, *ch)
})
updatedNetwork := newNetwork(u, record, channels)
// If we're currently connected, disconnect and perform the necessary
// bookkeeping
network.stop()
if network.conn != nil {
// Note: this will set network.conn to nil
u.handleUpstreamDisconnected(network.conn)
}
// Patch downstream connections to use our fresh updated network
for _, dc := range u.downstreamConns {
if dc.network != nil && dc.network == network {
dc.network = updatedNetwork
}
}
// We need to remove the network after patching downstream connections,
// otherwise they'll get closed
u.removeNetwork(network)
// The filesystem message store needs to be notified whenever the network
// is renamed
renameNetMsgStore, ok := u.msgStore.(msgstore.RenameNetworkStore)
if ok && updatedNetwork.GetName() != network.GetName() {
if err := renameNetMsgStore.RenameNetwork(&network.Network, &updatedNetwork.Network); err != nil {
network.logger.Printf("failed to update message store network name to %q: %v", updatedNetwork.GetName(), err)
}
}
// This will re-connect to the upstream server
u.addNetwork(updatedNetwork)
// TODO: only broadcast attributes that have changed
attrs := getNetworkAttrs(updatedNetwork)
u.notifyBouncerNetworkState(updatedNetwork.ID, attrs)
return updatedNetwork, nil
}
func (u *user) deleteNetwork(ctx context.Context, id int64) error {
network := u.getNetworkByID(id)
if network == nil {
panic("tried deleting a non-existing network")
}
if err := u.srv.db.DeleteNetwork(ctx, network.ID); err != nil {
return err
}
u.removeNetwork(network)
idStr := fmt.Sprintf("%v", network.ID)
for _, dc := range u.downstreamConns {
if dc.caps.IsEnabled("soju.im/bouncer-networks-notify") {
dc.SendMessage(ctx, &irc.Message{
Prefix: dc.srv.prefix(),
Command: "BOUNCER",
Params: []string{"NETWORK", idStr, "*"},
})
}
}
return nil
}
func (u *user) updateUser(ctx context.Context, update UserUpdateFunc) error {
record := u.User // copy
if err := update(&record); err != nil {
return err
}
nickUpdated := u.Nick != record.Nick
realnameUpdated := u.Realname != record.Realname
enabledUpdated := u.Enabled != record.Enabled
if err := u.srv.db.StoreUser(ctx, &record); err != nil {
return fmt.Errorf("failed to update user %q: %v", u.Username, err)
}
u.User = record
if nickUpdated {
for _, net := range u.networks {
if net.Nick != "" {
continue
}
if uc := net.conn; uc != nil {
uc.SendMessage(ctx, &irc.Message{
Command: "NICK",
Params: []string{database.GetNick(&u.User, &net.Network)},
})
}
}
}
if realnameUpdated || enabledUpdated {
// Re-connect to networks which use the default realname
var needUpdate []database.Network
for _, net := range u.networks {
// If only the realname was updated, maybe we can skip the
// re-connect
if realnameUpdated && !enabledUpdated {
// If this network has a custom realname set, no need to
// re-connect: the user-wide realname remains unused
if net.Realname != "" {
continue
}
// We only need to call updateNetwork for upstreams that don't
// support setname
if uc := net.conn; uc != nil && uc.caps.IsEnabled("setname") {
uc.SendMessage(ctx, &irc.Message{
Command: "SETNAME",
Params: []string{database.GetRealname(&u.User, &net.Network)},
})
continue
}
}
needUpdate = append(needUpdate, net.Network)
}
var netErr error
for _, net := range needUpdate {
if _, err := u.updateNetwork(ctx, &net); err != nil {
netErr = err
}
}
if netErr != nil {
return netErr
}
}
if !u.Enabled {
// TODO: send an error message before disconnecting
for _, dc := range u.downstreamConns {
dc.Close()
}
}
return nil
}
func (u *user) stop(ctx context.Context) error {
select {
case <-u.done:
return nil // already stopped
case u.events <- eventStop{}:
// we've requested to stop, let's wait for the user goroutine to exit
case <-ctx.Done():
return ctx.Err()
}
select {
case <-u.done:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
func (u *user) hasPersistentMsgStore() bool {
if u.msgStore == nil {
return false
}
return !msgstore.IsMemoryStore(u.msgStore)
}
func (u *user) FormatServerTime(t time.Time) string {
if u.msgStore != nil && msgstore.IsFSStore(u.msgStore) {
// The FS message store truncates message timestamps to the second,
// so truncate them here to get consistent timestamps.
t = t.Truncate(time.Second)
}
return xirc.FormatServerTime(t)
}
// localAddrForHost returns the local address to use when connecting to host.
// A nil address is returned when the OS should automatically pick one.
func (u *user) localTCPAddrForHost(ctx context.Context, host string) (*net.TCPAddr, error) {
upstreamUserIPs := u.srv.Config().UpstreamUserIPs
if len(upstreamUserIPs) == 0 {
return nil, nil
}
ips, err := net.DefaultResolver.LookupIP(ctx, "ip", host)
if err != nil {
return nil, err
}
wantIPv6 := false
for _, ip := range ips {
if ip.To4() == nil {
wantIPv6 = true
break
}
}
var ipNet *net.IPNet
for _, in := range upstreamUserIPs {
if wantIPv6 == (in.IP.To4() == nil) {
ipNet = in
break
}
}
if ipNet == nil {
return nil, nil
}
var ipInt big.Int
ipInt.SetBytes(ipNet.IP)
ipInt.Add(&ipInt, big.NewInt(u.ID+1))
ip := net.IP(ipInt.Bytes())
if !ipNet.Contains(ip) {
return nil, fmt.Errorf("IP network %v too small", ipNet)
}
return &net.TCPAddr{IP: ip}, nil
}
func (u *user) bumpDownstreamInteractionTime(ctx context.Context) {
err := u.updateUser(ctx, func(record *database.User) error {
record.DownstreamInteractedAt = time.Now()
return nil
})
if err != nil {
u.logger.Printf("failed to bump downstream interaction time: %v", err)
}
}