soju/user.go
Simon Ser 6f01bd86c3 downstream: pass context to SendMessage
Just like upstream does.
2023-04-06 13:23:20 +02:00

1294 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 := net.Nick
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().LogDriver {
case "fs":
msgStore = msgstore.NewFSStore(srv.Config().LogPath, 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) {
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)
}
}