You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
462 lines
11 KiB
Go
462 lines
11 KiB
Go
package bxmpp
|
|
|
|
import (
|
|
"bytes"
|
|
"crypto/tls"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"net/url"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/42wim/matterbridge/bridge"
|
|
"github.com/42wim/matterbridge/bridge/config"
|
|
"github.com/42wim/matterbridge/bridge/helper"
|
|
"github.com/jpillora/backoff"
|
|
"github.com/matterbridge/go-xmpp"
|
|
"github.com/rs/xid"
|
|
)
|
|
|
|
type Bxmpp struct {
|
|
*bridge.Config
|
|
|
|
startTime time.Time
|
|
xc *xmpp.Client
|
|
xmppMap map[string]string
|
|
connected bool
|
|
sync.RWMutex
|
|
|
|
avatarAvailability map[string]bool
|
|
avatarMap map[string]string
|
|
}
|
|
|
|
func New(cfg *bridge.Config) bridge.Bridger {
|
|
return &Bxmpp{
|
|
Config: cfg,
|
|
xmppMap: make(map[string]string),
|
|
avatarAvailability: make(map[string]bool),
|
|
avatarMap: make(map[string]string),
|
|
}
|
|
}
|
|
|
|
func (b *Bxmpp) Connect() error {
|
|
b.Log.Infof("Connecting %s", b.GetString("Server"))
|
|
if err := b.createXMPP(); err != nil {
|
|
b.Log.Debugf("%#v", err)
|
|
return err
|
|
}
|
|
|
|
b.Log.Info("Connection succeeded")
|
|
go b.manageConnection()
|
|
return nil
|
|
}
|
|
|
|
func (b *Bxmpp) Disconnect() error {
|
|
return nil
|
|
}
|
|
|
|
func (b *Bxmpp) JoinChannel(channel config.ChannelInfo) error {
|
|
if channel.Options.Key != "" {
|
|
b.Log.Debugf("using key %s for channel %s", channel.Options.Key, channel.Name)
|
|
b.xc.JoinProtectedMUC(channel.Name+"@"+b.GetString("Muc"), b.GetString("Nick"), channel.Options.Key, xmpp.NoHistory, 0, nil)
|
|
} else {
|
|
b.xc.JoinMUCNoHistory(channel.Name+"@"+b.GetString("Muc"), b.GetString("Nick"))
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (b *Bxmpp) Send(msg config.Message) (string, error) {
|
|
// should be fixed by using a cache instead of dropping
|
|
if !b.Connected() {
|
|
return "", fmt.Errorf("bridge %s not connected, dropping message %#v to bridge", b.Account, msg)
|
|
}
|
|
// ignore delete messages
|
|
if msg.Event == config.EventMsgDelete {
|
|
return "", nil
|
|
}
|
|
|
|
b.Log.Debugf("=> Receiving %#v", msg)
|
|
|
|
if msg.Event == config.EventAvatarDownload {
|
|
return b.cacheAvatar(&msg), nil
|
|
}
|
|
|
|
// Make a action /me of the message, prepend the username with it.
|
|
// https://xmpp.org/extensions/xep-0245.html
|
|
if msg.Event == config.EventUserAction {
|
|
msg.Username = "/me " + msg.Username
|
|
}
|
|
|
|
// Upload a file (in XMPP case send the upload URL because XMPP has no native upload support).
|
|
var err error
|
|
if msg.Extra != nil {
|
|
for _, rmsg := range helper.HandleExtra(&msg, b.General) {
|
|
b.Log.Debugf("=> Sending attachement message %#v", rmsg)
|
|
if b.GetString("WebhookURL") != "" {
|
|
err = b.postSlackCompatibleWebhook(msg)
|
|
} else {
|
|
_, err = b.xc.Send(xmpp.Chat{
|
|
Type: "groupchat",
|
|
Remote: rmsg.Channel + "@" + b.GetString("Muc"),
|
|
Text: rmsg.Username + rmsg.Text,
|
|
})
|
|
}
|
|
|
|
if err != nil {
|
|
b.Log.WithError(err).Error("Unable to send message with share URL.")
|
|
}
|
|
}
|
|
if len(msg.Extra["file"]) > 0 {
|
|
return "", b.handleUploadFile(&msg)
|
|
}
|
|
}
|
|
|
|
if b.GetString("WebhookURL") != "" {
|
|
b.Log.Debugf("Sending message using Webhook")
|
|
err := b.postSlackCompatibleWebhook(msg)
|
|
if err != nil {
|
|
b.Log.Errorf("Failed to send message using webhook: %s", err)
|
|
return "", err
|
|
}
|
|
|
|
return "", nil
|
|
}
|
|
|
|
// Post normal message.
|
|
var msgReplaceID string
|
|
msgID := xid.New().String()
|
|
if msg.ID != "" {
|
|
msgReplaceID = msg.ID
|
|
}
|
|
b.Log.Debugf("=> Sending message %#v", msg)
|
|
if _, err := b.xc.Send(xmpp.Chat{
|
|
Type: "groupchat",
|
|
Remote: msg.Channel + "@" + b.GetString("Muc"),
|
|
Text: msg.Username + msg.Text,
|
|
ID: msgID,
|
|
ReplaceID: msgReplaceID,
|
|
}); err != nil {
|
|
return "", err
|
|
}
|
|
return msgID, nil
|
|
}
|
|
|
|
func (b *Bxmpp) postSlackCompatibleWebhook(msg config.Message) error {
|
|
type XMPPWebhook struct {
|
|
Username string `json:"username"`
|
|
Text string `json:"text"`
|
|
}
|
|
webhookBody, err := json.Marshal(XMPPWebhook{
|
|
Username: msg.Username,
|
|
Text: msg.Text,
|
|
})
|
|
if err != nil {
|
|
b.Log.Errorf("Failed to marshal webhook: %s", err)
|
|
return err
|
|
}
|
|
|
|
resp, err := http.Post(b.GetString("WebhookURL")+"/"+url.QueryEscape(msg.Channel), "application/json", bytes.NewReader(webhookBody))
|
|
if err != nil {
|
|
b.Log.Errorf("Failed to POST webhook: %s", err)
|
|
return err
|
|
}
|
|
|
|
resp.Body.Close()
|
|
return nil
|
|
}
|
|
|
|
func (b *Bxmpp) createXMPP() error {
|
|
var serverName string
|
|
switch {
|
|
case !b.GetBool("Anonymous"):
|
|
if !strings.Contains(b.GetString("Jid"), "@") {
|
|
return fmt.Errorf("the Jid %s doesn't contain an @", b.GetString("Jid"))
|
|
}
|
|
serverName = strings.Split(b.GetString("Jid"), "@")[1]
|
|
case !strings.Contains(b.GetString("Server"), ":"):
|
|
serverName = strings.Split(b.GetString("Server"), ":")[0]
|
|
default:
|
|
serverName = b.GetString("Server")
|
|
}
|
|
|
|
tc := &tls.Config{
|
|
ServerName: serverName,
|
|
InsecureSkipVerify: b.GetBool("SkipTLSVerify"), // nolint: gosec
|
|
}
|
|
|
|
xmpp.DebugWriter = b.Log.Writer()
|
|
|
|
options := xmpp.Options{
|
|
Host: b.GetString("Server"),
|
|
User: b.GetString("Jid"),
|
|
Password: b.GetString("Password"),
|
|
NoTLS: true,
|
|
StartTLS: !b.GetBool("NoTLS"),
|
|
TLSConfig: tc,
|
|
Debug: b.GetBool("debug"),
|
|
Session: true,
|
|
Status: "",
|
|
StatusMessage: "",
|
|
Resource: "",
|
|
InsecureAllowUnencryptedAuth: b.GetBool("NoTLS"),
|
|
}
|
|
var err error
|
|
b.xc, err = options.NewClient()
|
|
return err
|
|
}
|
|
|
|
func (b *Bxmpp) manageConnection() {
|
|
b.setConnected(true)
|
|
initial := true
|
|
bf := &backoff.Backoff{
|
|
Min: time.Second,
|
|
Max: 5 * time.Minute,
|
|
Jitter: true,
|
|
}
|
|
|
|
// Main connection loop. Each iteration corresponds to a successful
|
|
// connection attempt and the subsequent handling of the connection.
|
|
for {
|
|
if initial {
|
|
initial = false
|
|
} else {
|
|
b.Remote <- config.Message{
|
|
Username: "system",
|
|
Text: "rejoin",
|
|
Channel: "",
|
|
Account: b.Account,
|
|
Event: config.EventRejoinChannels,
|
|
}
|
|
}
|
|
|
|
if err := b.handleXMPP(); err != nil {
|
|
b.Log.WithError(err).Error("Disconnected.")
|
|
b.setConnected(false)
|
|
}
|
|
|
|
// Reconnection loop using an exponential back-off strategy. We
|
|
// only break out of the loop if we have successfully reconnected.
|
|
for {
|
|
d := bf.Duration()
|
|
b.Log.Infof("Reconnecting in %s.", d)
|
|
time.Sleep(d)
|
|
|
|
b.Log.Infof("Reconnecting now.")
|
|
if err := b.createXMPP(); err == nil {
|
|
b.setConnected(true)
|
|
bf.Reset()
|
|
break
|
|
}
|
|
b.Log.Warn("Failed to reconnect.")
|
|
}
|
|
}
|
|
}
|
|
|
|
func (b *Bxmpp) xmppKeepAlive() chan bool {
|
|
done := make(chan bool)
|
|
go func() {
|
|
ticker := time.NewTicker(90 * time.Second)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-ticker.C:
|
|
b.Log.Debugf("PING")
|
|
if err := b.xc.PingC2S("", ""); err != nil {
|
|
b.Log.Debugf("PING failed %#v", err)
|
|
}
|
|
case <-done:
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
return done
|
|
}
|
|
|
|
func (b *Bxmpp) handleXMPP() error {
|
|
b.startTime = time.Now()
|
|
|
|
done := b.xmppKeepAlive()
|
|
defer close(done)
|
|
|
|
for {
|
|
m, err := b.xc.Recv()
|
|
if err != nil {
|
|
// An error together with AvatarData is non-fatal
|
|
switch m.(type) {
|
|
case xmpp.AvatarData:
|
|
continue
|
|
default:
|
|
return err
|
|
}
|
|
}
|
|
|
|
switch v := m.(type) {
|
|
case xmpp.Chat:
|
|
if v.Type == "groupchat" {
|
|
b.Log.Debugf("== Receiving %#v", v)
|
|
|
|
// Skip invalid messages.
|
|
if b.skipMessage(v) {
|
|
continue
|
|
}
|
|
|
|
var event string
|
|
if strings.Contains(v.Text, "has set the subject to:") {
|
|
event = config.EventTopicChange
|
|
}
|
|
|
|
available, sok := b.avatarAvailability[v.Remote]
|
|
avatar := ""
|
|
if !sok {
|
|
b.Log.Debugf("Requesting avatar data")
|
|
b.avatarAvailability[v.Remote] = false
|
|
b.xc.AvatarRequestData(v.Remote)
|
|
} else if available {
|
|
avatar = getAvatar(b.avatarMap, v.Remote, b.General)
|
|
}
|
|
|
|
msgID := v.ID
|
|
if v.ReplaceID != "" {
|
|
msgID = v.ReplaceID
|
|
}
|
|
rmsg := config.Message{
|
|
Username: b.parseNick(v.Remote),
|
|
Text: v.Text,
|
|
Channel: b.parseChannel(v.Remote),
|
|
Account: b.Account,
|
|
Avatar: avatar,
|
|
UserID: v.Remote,
|
|
ID: msgID,
|
|
Event: event,
|
|
}
|
|
|
|
// Check if we have an action event.
|
|
var ok bool
|
|
rmsg.Text, ok = b.replaceAction(rmsg.Text)
|
|
if ok {
|
|
rmsg.Event = config.EventUserAction
|
|
}
|
|
|
|
b.Log.Debugf("<= Sending message from %s on %s to gateway", rmsg.Username, b.Account)
|
|
b.Log.Debugf("<= Message is %#v", rmsg)
|
|
b.Remote <- rmsg
|
|
}
|
|
case xmpp.AvatarData:
|
|
b.handleDownloadAvatar(v)
|
|
b.avatarAvailability[v.From] = true
|
|
b.Log.Debugf("Avatar for %s is now available", v.From)
|
|
case xmpp.Presence:
|
|
// Do nothing.
|
|
}
|
|
}
|
|
}
|
|
|
|
func (b *Bxmpp) replaceAction(text string) (string, bool) {
|
|
if strings.HasPrefix(text, "/me ") {
|
|
return strings.Replace(text, "/me ", "", -1), true
|
|
}
|
|
return text, false
|
|
}
|
|
|
|
// handleUploadFile handles native upload of files
|
|
func (b *Bxmpp) handleUploadFile(msg *config.Message) error {
|
|
var urlDesc string
|
|
|
|
for _, file := range msg.Extra["file"] {
|
|
fileInfo := file.(config.FileInfo)
|
|
if fileInfo.Comment != "" {
|
|
msg.Text += fileInfo.Comment + ": "
|
|
}
|
|
if fileInfo.URL != "" {
|
|
msg.Text = fileInfo.URL
|
|
if fileInfo.Comment != "" {
|
|
msg.Text = fileInfo.Comment + ": " + fileInfo.URL
|
|
urlDesc = fileInfo.Comment
|
|
}
|
|
}
|
|
if _, err := b.xc.Send(xmpp.Chat{
|
|
Type: "groupchat",
|
|
Remote: msg.Channel + "@" + b.GetString("Muc"),
|
|
Text: msg.Username + msg.Text,
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
|
|
if fileInfo.URL != "" {
|
|
if _, err := b.xc.SendOOB(xmpp.Chat{
|
|
Type: "groupchat",
|
|
Remote: msg.Channel + "@" + b.GetString("Muc"),
|
|
Ooburl: fileInfo.URL,
|
|
Oobdesc: urlDesc,
|
|
}); err != nil {
|
|
b.Log.WithError(err).Warn("Failed to send share URL.")
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (b *Bxmpp) parseNick(remote string) string {
|
|
s := strings.Split(remote, "@")
|
|
if len(s) > 1 {
|
|
s = strings.Split(s[1], "/")
|
|
if len(s) == 2 {
|
|
return s[1] // nick
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func (b *Bxmpp) parseChannel(remote string) string {
|
|
s := strings.Split(remote, "@")
|
|
if len(s) >= 2 {
|
|
return s[0] // channel
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// skipMessage skips messages that need to be skipped
|
|
func (b *Bxmpp) skipMessage(message xmpp.Chat) bool {
|
|
// skip messages from ourselves
|
|
if b.parseNick(message.Remote) == b.GetString("Nick") {
|
|
return true
|
|
}
|
|
|
|
// skip empty messages
|
|
if message.Text == "" {
|
|
return true
|
|
}
|
|
|
|
// skip subject messages
|
|
if strings.Contains(message.Text, "</subject>") {
|
|
return true
|
|
}
|
|
|
|
// do not show subjects on connect #732
|
|
if strings.Contains(message.Text, "has set the subject to:") && time.Since(b.startTime) < time.Second*5 {
|
|
return true
|
|
}
|
|
|
|
// Ignore messages posted by our webhook
|
|
if b.GetString("WebhookURL") != "" && strings.Contains(message.ID, "webhookbot") {
|
|
return true
|
|
}
|
|
|
|
// skip delayed messages
|
|
return !message.Stamp.IsZero() && time.Since(message.Stamp).Minutes() > 5
|
|
}
|
|
|
|
func (b *Bxmpp) setConnected(state bool) {
|
|
b.Lock()
|
|
b.connected = state
|
|
defer b.Unlock()
|
|
}
|
|
|
|
func (b *Bxmpp) Connected() bool {
|
|
b.RLock()
|
|
defer b.RUnlock()
|
|
return b.connected
|
|
}
|