Session 2: Matrix client with E2EE and room-scoped handler
Adapted from Pete: password login + device persistence, cryptohelper with cross-signing bootstrap, sync loop with auto-join on invite, and PostThreadedReply for command responses. Messages outside allowed_rooms are dropped.
This commit is contained in:
prosolis
316
internal/matrix/client.go
Normal file
316
internal/matrix/client.go
Normal file
@@ -0,0 +1,316 @@
|
||||
package matrix
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"bellhop/internal/config"
|
||||
|
||||
"maunium.net/go/mautrix"
|
||||
"maunium.net/go/mautrix/crypto/cryptohelper"
|
||||
"maunium.net/go/mautrix/event"
|
||||
"maunium.net/go/mautrix/id"
|
||||
)
|
||||
|
||||
// MessageHandler is called when a text message is received in an allowlisted room.
|
||||
type MessageHandler func(roomID id.RoomID, eventID id.EventID, sender id.UserID, body string)
|
||||
|
||||
type deviceInfo struct {
|
||||
AccessToken string `json:"access_token"`
|
||||
DeviceID string `json:"device_id"`
|
||||
UserID string `json:"user_id"`
|
||||
}
|
||||
|
||||
type Client struct {
|
||||
mx *mautrix.Client
|
||||
allowedRooms map[id.RoomID]struct{}
|
||||
userID id.UserID
|
||||
cfg config.MatrixConfig
|
||||
crypto *cryptohelper.CryptoHelper
|
||||
|
||||
mu sync.RWMutex
|
||||
onMessage MessageHandler
|
||||
cancelSync context.CancelFunc
|
||||
}
|
||||
|
||||
// New creates a Matrix client with password login, device persistence, and E2EE.
|
||||
func New(cfg config.MatrixConfig) (*Client, error) {
|
||||
if err := os.MkdirAll(cfg.DataDir, 0o755); err != nil {
|
||||
return nil, fmt.Errorf("create data dir: %w", err)
|
||||
}
|
||||
|
||||
devicePath := filepath.Join(cfg.DataDir, "device.json")
|
||||
|
||||
device, err := loadDevice(devicePath)
|
||||
if err != nil {
|
||||
slog.Info("no existing device found, will login fresh")
|
||||
}
|
||||
|
||||
var mx *mautrix.Client
|
||||
|
||||
if device != nil {
|
||||
if isTokenValid(cfg.Homeserver, device.AccessToken) {
|
||||
slog.Info("existing device credentials valid", "device_id", device.DeviceID)
|
||||
mx, err = mautrix.NewClient(cfg.Homeserver, id.UserID(device.UserID), device.AccessToken)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("create client with existing token: %w", err)
|
||||
}
|
||||
mx.DeviceID = id.DeviceID(device.DeviceID)
|
||||
} else {
|
||||
slog.Warn("existing device credentials invalid, logging in again")
|
||||
device = nil
|
||||
}
|
||||
}
|
||||
|
||||
if device == nil {
|
||||
mx, err = mautrix.NewClient(cfg.Homeserver, "", "")
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("create client: %w", err)
|
||||
}
|
||||
|
||||
resp, err := mx.Login(context.Background(), &mautrix.ReqLogin{
|
||||
Type: mautrix.AuthTypePassword,
|
||||
Identifier: mautrix.UserIdentifier{
|
||||
Type: mautrix.IdentifierTypeUser,
|
||||
User: cfg.UserID,
|
||||
},
|
||||
Password: cfg.Password,
|
||||
InitialDeviceDisplayName: cfg.DisplayName,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("login: %w", err)
|
||||
}
|
||||
|
||||
mx.AccessToken = resp.AccessToken
|
||||
mx.UserID = resp.UserID
|
||||
mx.DeviceID = resp.DeviceID
|
||||
|
||||
device = &deviceInfo{
|
||||
AccessToken: resp.AccessToken,
|
||||
DeviceID: string(resp.DeviceID),
|
||||
UserID: string(resp.UserID),
|
||||
}
|
||||
if err := saveDevice(devicePath, device); err != nil {
|
||||
slog.Warn("failed to save device info", "err", err)
|
||||
}
|
||||
|
||||
slog.Info("logged in successfully", "user_id", resp.UserID, "device_id", resp.DeviceID)
|
||||
}
|
||||
|
||||
cryptoDBPath := filepath.Join(cfg.DataDir, "crypto.db")
|
||||
ch, err := cryptohelper.NewCryptoHelper(mx, []byte(cfg.PickleKey), cryptoDBPath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("init crypto helper: %w", err)
|
||||
}
|
||||
|
||||
// Surface decrypt failures — the default silent no-op hides missing-megolm-session
|
||||
// errors, which look like the bot ignoring commands.
|
||||
ch.DecryptErrorCallback = func(evt *event.Event, err error) {
|
||||
slog.Warn("matrix: failed to decrypt incoming event",
|
||||
"room", evt.RoomID, "event_id", evt.ID, "sender", evt.Sender, "err", err)
|
||||
}
|
||||
|
||||
ch.LoginAs = &mautrix.ReqLogin{
|
||||
Type: mautrix.AuthTypePassword,
|
||||
Identifier: mautrix.UserIdentifier{
|
||||
Type: mautrix.IdentifierTypeUser,
|
||||
User: cfg.UserID,
|
||||
},
|
||||
Password: cfg.Password,
|
||||
InitialDeviceDisplayName: cfg.DisplayName,
|
||||
}
|
||||
|
||||
if err := ch.Init(context.Background()); err != nil {
|
||||
return nil, fmt.Errorf("crypto helper init: %w", err)
|
||||
}
|
||||
|
||||
mx.Crypto = ch
|
||||
|
||||
mach := ch.Machine()
|
||||
existingKeys, err := mach.GetOwnCrossSigningPublicKeys(context.Background())
|
||||
if err != nil {
|
||||
slog.Warn("cross-signing: failed to fetch existing keys", "err", err)
|
||||
}
|
||||
if existingKeys == nil || existingKeys.MasterKey == "" {
|
||||
_, _, err = mach.GenerateAndUploadCrossSigningKeys(context.Background(), func(ui *mautrix.RespUserInteractive) interface{} {
|
||||
return map[string]interface{}{
|
||||
"type": mautrix.AuthTypePassword,
|
||||
"identifier": map[string]interface{}{
|
||||
"type": mautrix.IdentifierTypeUser,
|
||||
"user": cfg.UserID,
|
||||
},
|
||||
"password": cfg.Password,
|
||||
"session": ui.Session,
|
||||
}
|
||||
}, "")
|
||||
if err != nil {
|
||||
slog.Warn("cross-signing: key upload failed", "err", err)
|
||||
} else {
|
||||
slog.Info("cross-signing: keys uploaded")
|
||||
}
|
||||
if err := mach.SignOwnDevice(context.Background(), mach.OwnIdentity()); err != nil {
|
||||
slog.Warn("cross-signing: sign own device failed", "err", err)
|
||||
}
|
||||
if err := mach.SignOwnMasterKey(context.Background()); err != nil {
|
||||
slog.Warn("cross-signing: sign master key failed", "err", err)
|
||||
}
|
||||
} else {
|
||||
slog.Info("cross-signing: already configured, skipping bootstrap")
|
||||
}
|
||||
|
||||
slog.Info("E2EE initialized", "device_id", mx.DeviceID)
|
||||
|
||||
allowed := make(map[id.RoomID]struct{}, len(cfg.AllowedRooms))
|
||||
for _, r := range cfg.AllowedRooms {
|
||||
allowed[id.RoomID(r)] = struct{}{}
|
||||
}
|
||||
|
||||
return &Client{
|
||||
mx: mx,
|
||||
allowedRooms: allowed,
|
||||
userID: mx.UserID,
|
||||
cfg: cfg,
|
||||
crypto: ch,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// SetMessageHandler registers a callback for text messages in allowlisted rooms.
|
||||
func (c *Client) SetMessageHandler(fn MessageHandler) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.onMessage = fn
|
||||
}
|
||||
|
||||
// Start begins the Matrix sync loop in the background.
|
||||
func (c *Client) Start(ctx context.Context) {
|
||||
syncer := c.mx.Syncer.(*mautrix.DefaultSyncer)
|
||||
|
||||
syncer.OnEventType(event.EventMessage, func(ctx context.Context, evt *event.Event) {
|
||||
if evt.Sender == c.userID {
|
||||
return
|
||||
}
|
||||
if _, ok := c.allowedRooms[evt.RoomID]; !ok {
|
||||
return
|
||||
}
|
||||
msg := evt.Content.AsMessage()
|
||||
if msg == nil || msg.MsgType != event.MsgText {
|
||||
return
|
||||
}
|
||||
c.mu.RLock()
|
||||
handler := c.onMessage
|
||||
c.mu.RUnlock()
|
||||
if handler != nil {
|
||||
handler(evt.RoomID, evt.ID, evt.Sender, msg.Body)
|
||||
}
|
||||
})
|
||||
|
||||
// Auto-join on invite to any room — allowlist still gates message handling.
|
||||
syncer.OnEventType(event.StateMember, func(ctx context.Context, evt *event.Event) {
|
||||
if evt.GetStateKey() == string(c.userID) && evt.Content.AsMember().Membership == event.MembershipInvite {
|
||||
if _, err := c.mx.JoinRoomByID(ctx, evt.RoomID); err != nil {
|
||||
slog.Error("failed to auto-join room", "room", evt.RoomID, "err", err)
|
||||
} else {
|
||||
slog.Info("auto-joined room", "room", evt.RoomID)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
syncCtx, cancel := context.WithCancel(ctx)
|
||||
c.cancelSync = cancel
|
||||
|
||||
go func() {
|
||||
for {
|
||||
err := c.mx.SyncWithContext(syncCtx)
|
||||
if syncCtx.Err() != nil {
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
slog.Error("matrix sync stopped, restarting in 5s", "err", err)
|
||||
} else {
|
||||
slog.Warn("matrix sync returned without error, restarting in 5s")
|
||||
}
|
||||
select {
|
||||
case <-time.After(5 * time.Second):
|
||||
case <-syncCtx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// Stop halts the sync loop and closes the crypto store.
|
||||
func (c *Client) Stop() {
|
||||
if c.cancelSync != nil {
|
||||
c.cancelSync()
|
||||
}
|
||||
if c.crypto != nil {
|
||||
if err := c.crypto.Close(); err != nil {
|
||||
slog.Error("failed to close crypto helper", "err", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// PostThreadedReply sends a plain/HTML message into the thread rooted at rootEventID.
|
||||
// IsFallingBack + m.in_reply_to gives non-thread-aware clients a sensible quoted reply.
|
||||
func (c *Client) PostThreadedReply(roomID id.RoomID, rootEventID id.EventID, plain, htmlBody string) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
|
||||
content := &event.MessageEventContent{
|
||||
MsgType: event.MsgText,
|
||||
Body: plain,
|
||||
Format: event.FormatHTML,
|
||||
FormattedBody: htmlBody,
|
||||
RelatesTo: &event.RelatesTo{
|
||||
Type: event.RelThread,
|
||||
EventID: rootEventID,
|
||||
InReplyTo: &event.InReplyTo{EventID: rootEventID},
|
||||
IsFallingBack: true,
|
||||
},
|
||||
}
|
||||
_, err := c.mx.SendMessageEvent(ctx, roomID, event.EventMessage, content)
|
||||
return err
|
||||
}
|
||||
|
||||
func loadDevice(path string) (*deviceInfo, error) {
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var info deviceInfo
|
||||
if err := json.Unmarshal(data, &info); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &info, nil
|
||||
}
|
||||
|
||||
func saveDevice(path string, info *deviceInfo) error {
|
||||
data, err := json.MarshalIndent(info, "", " ")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(path, data, 0o600)
|
||||
}
|
||||
|
||||
func isTokenValid(homeserver, accessToken string) bool {
|
||||
req, err := http.NewRequest("GET", homeserver+"/_matrix/client/v3/account/whoami", nil)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
req.Header.Set("Authorization", "Bearer "+accessToken)
|
||||
|
||||
client := &http.Client{Timeout: 10 * time.Second}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
return resp.StatusCode == http.StatusOK
|
||||
}
|
||||
Reference in New Issue
Block a user