diff --git a/internal/bot/appservice.go b/internal/bot/appservice.go index 9c53f4a..889026b 100644 --- a/internal/bot/appservice.go +++ b/internal/bot/appservice.go @@ -7,7 +7,6 @@ import ( "log/slog" "os" "path/filepath" - "sync" "time" "github.com/rs/zerolog" @@ -107,7 +106,15 @@ func newAppserviceSession(cfg Config) (*Session, error) { return nil, fmt.Errorf("registration sender_localpart resolves to %s but BOT_USER_ID is %s", as.BotMXID(), userID) } + // Resolve room state (is it encrypted, who is in it) from the server on first + // use, since there is no /sync to backfill it. Must be installed before the + // first BotClient() call: makeClient copies as.StateStore into the client, and + // caches the client. + store := newLazyStateStore(as.StateStore) + as.StateStore = store + client := as.BotClient() // as_token auth + SetAppServiceUserID (?user_id=) assertion + store.client = client // Assert the device via ?org.matrix.msc3202.device_id= on E2EE requests, and // satisfy cryptohelper.Init's syncer check: with this set, Init permits a nil // Syncer (we drive the crypto machine from transactions, not /sync). The param @@ -172,29 +179,6 @@ func newAppserviceSession(cfg Config) (*Session, error) { mach.HandleMemberEvent(ctx, evt) }) - // Without /sync there is no state backfill, so the client's StateStore starts - // empty. Before we hand off a decrypted event (whose reply may need to be - // encrypted), resolve the room once: mark it encrypted and populate its member - // list. Otherwise outbound sends go plaintext (IsEncrypted=false) and, worse, - // the group session is shared to nobody (GetRoomJoinedOrInvitedMembers empty) - // so recipients can't decrypt the bot's replies. - var resolved sync.Map // roomID -> struct{}, resolved once - resolveRoom := func(ctx context.Context, roomID id.RoomID) { - if _, done := resolved.LoadOrStore(roomID, struct{}{}); done { - return - } - var enc event.EncryptionEventContent - if err := client.StateEvent(ctx, roomID, event.StateEncryption, "", &enc); err == nil && enc.Algorithm != "" { - _ = as.StateStore.SetEncryptionEvent(ctx, roomID, &enc) - } - if members, err := client.Members(ctx, roomID); err == nil { - _ = as.StateStore.ReplaceCachedMembers(ctx, roomID, members.Chunk) - } else { - slog.Warn("appservice: failed to fetch room members; will retry", "room", roomID, "err", err) - resolved.Delete(roomID) // allow a later event to retry - } - } - // recoverSession handles an event whose megolm session we don't have yet. The // keys are often merely in flight (a to-device m.room_key racing the room // event), so wait briefly; if they never land, ask the sender to re-share and @@ -236,7 +220,6 @@ func newAppserviceSession(cfg Config) (*Session, error) { // Decrypt inbound encrypted room events and re-dispatch the plaintext so the // normal message/reaction handlers fire (mirrors cryptohelper.HandleEncrypted). ep.On(event.EventEncrypted, func(ctx context.Context, evt *event.Event) { - resolveRoom(ctx, evt.RoomID) decrypted, err := ch.Decrypt(ctx, evt) if err != nil { if errors.Is(err, cryptohelper.NoSessionFound) { diff --git a/internal/bot/statestore.go b/internal/bot/statestore.go new file mode 100644 index 0000000..d5694b5 --- /dev/null +++ b/internal/bot/statestore.go @@ -0,0 +1,135 @@ +package bot + +import ( + "context" + "errors" + "fmt" + "sync" + + "maunium.net/go/mautrix" + "maunium.net/go/mautrix/appservice" + "maunium.net/go/mautrix/event" + "maunium.net/go/mautrix/id" +) + +// lazyStateStore resolves room state from the homeserver on first use instead of +// waiting for it to arrive over a sync. +// +// The appservice has no /sync, so the state store only ever learns a room is +// encrypted from an m.room.encryption event pushed in a transaction — which for +// an already-configured room never happens again. Anything the bot says on its +// own initiative (ambient DMs, expedition beats, briefings) therefore reached +// SendMessageEvent with IsEncrypted=false and went out as plaintext into an +// encrypted room. The state store being in-memory made every restart a fresh +// chance to leak. +// +// Both reads the send path depends on are resolved here on the first miss and +// cached: whether the room is encrypted, and who to share the group session with +// (an empty member list encrypts to nobody, which is its own outage). Neither +// read is allowed to fail open — a lookup that cannot be resolved returns an +// error, which aborts the send. A failed message is recoverable; a plaintext one +// that has already landed in an encrypted room is not. +type lazyStateStore struct { + appservice.StateStore + + client *mautrix.Client + + mu sync.Mutex + locks map[id.RoomID]*sync.Mutex + // Rooms whose m.room.encryption state we have asked the server about. The + // underlying store cannot distinguish "not encrypted" from "never heard of + // it", so the distinction is tracked here. + encryptionResolved map[id.RoomID]bool +} + +func newLazyStateStore(inner appservice.StateStore) *lazyStateStore { + return &lazyStateStore{ + StateStore: inner, + locks: make(map[id.RoomID]*sync.Mutex), + encryptionResolved: make(map[id.RoomID]bool), + } +} + +// lockRoom serialises resolution per room so a burst of sends into a cold room +// makes one request rather than one per message. +func (s *lazyStateStore) lockRoom(roomID id.RoomID) func() { + s.mu.Lock() + lock, ok := s.locks[roomID] + if !ok { + lock = &sync.Mutex{} + s.locks[roomID] = lock + } + s.mu.Unlock() + + lock.Lock() + return lock.Unlock +} + +func (s *lazyStateStore) isEncryptionResolved(roomID id.RoomID) bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.encryptionResolved[roomID] +} + +func (s *lazyStateStore) markEncryptionResolved(roomID id.RoomID) { + s.mu.Lock() + defer s.mu.Unlock() + s.encryptionResolved[roomID] = true +} + +// IsEncrypted answers from the underlying store once the room's encryption state +// has been resolved, fetching it from the server the first time. +func (s *lazyStateStore) IsEncrypted(ctx context.Context, roomID id.RoomID) (bool, error) { + if !s.isEncryptionResolved(roomID) { + unlock := s.lockRoom(roomID) + defer unlock() + + if !s.isEncryptionResolved(roomID) { + var enc event.EncryptionEventContent + err := s.client.StateEvent(ctx, roomID, event.StateEncryption, "", &enc) + switch { + case err == nil: + if enc.Algorithm != "" { + if err := s.StateStore.SetEncryptionEvent(ctx, roomID, &enc); err != nil { + return false, fmt.Errorf("cache encryption state for %s: %w", roomID, err) + } + } + case errors.Is(err, mautrix.MNotFound): + // No m.room.encryption at all: the room really is unencrypted. + default: + // Unresolved. Refuse rather than guess "unencrypted" and leak. + return false, fmt.Errorf("resolve encryption state for %s: %w", roomID, err) + } + s.markEncryptionResolved(roomID) + } + } + return s.StateStore.IsEncrypted(ctx, roomID) +} + +// GetRoomJoinedOrInvitedMembers backs megolm session sharing. Without a sync +// there is no member backfill, so an unfetched room would report zero members +// and the bot would encrypt to an audience of nobody. +func (s *lazyStateStore) GetRoomJoinedOrInvitedMembers(ctx context.Context, roomID id.RoomID) ([]id.UserID, error) { + fetched, err := s.StateStore.HasFetchedMembers(ctx, roomID) + if err != nil { + return nil, fmt.Errorf("check member cache for %s: %w", roomID, err) + } + if !fetched { + unlock := s.lockRoom(roomID) + defer unlock() + + if fetched, err := s.StateStore.HasFetchedMembers(ctx, roomID); err != nil { + return nil, fmt.Errorf("check member cache for %s: %w", roomID, err) + } else if !fetched { + members, err := s.client.Members(ctx, roomID) + if err != nil { + return nil, fmt.Errorf("fetch members of %s: %w", roomID, err) + } + // No membership filter, so this also marks the room as fetched. + if err := s.StateStore.ReplaceCachedMembers(ctx, roomID, members.Chunk); err != nil { + return nil, fmt.Errorf("cache members of %s: %w", roomID, err) + } + } + } + return s.StateStore.GetRoomJoinedOrInvitedMembers(ctx, roomID) +} diff --git a/internal/bot/statestore_test.go b/internal/bot/statestore_test.go new file mode 100644 index 0000000..144c9d0 --- /dev/null +++ b/internal/bot/statestore_test.go @@ -0,0 +1,179 @@ +package bot + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + + "maunium.net/go/mautrix" + "maunium.net/go/mautrix/appservice" + "maunium.net/go/mautrix/id" +) + +// fakeHS serves the two endpoints the lazy store resolves against, counting hits +// so we can assert it caches instead of re-asking on every send. +type fakeHS struct { + encryptionStatus int + encryptionBody string + membersBody string + + encryptionHits atomic.Int32 + memberHits atomic.Int32 +} + +func (f *fakeHS) start(t *testing.T) *mautrix.Client { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.Contains(r.URL.Path, "/state/m.room.encryption"): + f.encryptionHits.Add(1) + w.WriteHeader(f.encryptionStatus) + _, _ = w.Write([]byte(f.encryptionBody)) + case strings.HasSuffix(r.URL.Path, "/members"): + f.memberHits.Add(1) + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(f.membersBody)) + default: + t.Errorf("unexpected request to %s", r.URL.Path) + w.WriteHeader(http.StatusNotFound) + } + })) + t.Cleanup(srv.Close) + + client, err := mautrix.NewClient(srv.URL, "@bot:example.org", "token") + if err != nil { + t.Fatalf("new client: %v", err) + } + return client +} + +func newTestStore(t *testing.T, hs *fakeHS) *lazyStateStore { + t.Helper() + store := newLazyStateStore(mautrix.NewMemoryStateStore().(appservice.StateStore)) + store.client = hs.start(t) + return store +} + +const roomID = id.RoomID("!room:example.org") + +// The regression: a room configured as encrypted before the process started. No +// transaction ever re-announces it, so the store must go ask. +func TestIsEncryptedResolvesFromServer(t *testing.T) { + hs := &fakeHS{ + encryptionStatus: http.StatusOK, + encryptionBody: `{"algorithm":"m.megolm.v1.aes-sha2"}`, + } + store := newTestStore(t, hs) + + enc, err := store.IsEncrypted(context.Background(), roomID) + if err != nil { + t.Fatalf("IsEncrypted: %v", err) + } + if !enc { + t.Fatal("room is encrypted on the server but IsEncrypted said false — this is the plaintext leak") + } +} + +// A room with no m.room.encryption really is plaintext, and must not re-fetch. +func TestIsEncryptedCachesUnencrypted(t *testing.T) { + hs := &fakeHS{ + encryptionStatus: http.StatusNotFound, + encryptionBody: `{"errcode":"M_NOT_FOUND","error":"Event not found."}`, + } + store := newTestStore(t, hs) + + for i := 0; i < 3; i++ { + enc, err := store.IsEncrypted(context.Background(), roomID) + if err != nil { + t.Fatalf("IsEncrypted: %v", err) + } + if enc { + t.Fatal("unencrypted room reported as encrypted") + } + } + if got := hs.encryptionHits.Load(); got != 1 { + t.Errorf("expected the M_NOT_FOUND to be cached, got %d fetches", got) + } +} + +// The important half: an unresolvable room must fail the send, not quietly +// answer "not encrypted" and let the message out in the clear. +func TestIsEncryptedFailsClosed(t *testing.T) { + hs := &fakeHS{ + encryptionStatus: http.StatusInternalServerError, + encryptionBody: `{"errcode":"M_UNKNOWN","error":"oops"}`, + } + store := newTestStore(t, hs) + + if _, err := store.IsEncrypted(context.Background(), roomID); err == nil { + t.Fatal("a failed lookup returned no error; the send would proceed in plaintext") + } + // The failure must not poison the cache: a later send has to retry. + hs.encryptionStatus = http.StatusOK + hs.encryptionBody = `{"algorithm":"m.megolm.v1.aes-sha2"}` + enc, err := store.IsEncrypted(context.Background(), roomID) + if err != nil { + t.Fatalf("IsEncrypted after recovery: %v", err) + } + if !enc { + t.Fatal("store did not retry after a transient failure") + } +} + +// A burst of proactive sends into a cold room should resolve it once. +func TestIsEncryptedResolvesOncePerRoom(t *testing.T) { + hs := &fakeHS{ + encryptionStatus: http.StatusOK, + encryptionBody: `{"algorithm":"m.megolm.v1.aes-sha2"}`, + } + store := newTestStore(t, hs) + + var wg sync.WaitGroup + for i := 0; i < 20; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if _, err := store.IsEncrypted(context.Background(), roomID); err != nil { + t.Errorf("IsEncrypted: %v", err) + } + }() + } + wg.Wait() + + if got := hs.encryptionHits.Load(); got != 1 { + t.Errorf("expected 1 state fetch for 20 concurrent sends, got %d", got) + } +} + +// Encrypting to an empty member list would share the group session with nobody, +// so members must be resolved too. +func TestMembersResolveFromServer(t *testing.T) { + hs := &fakeHS{ + encryptionStatus: http.StatusOK, + encryptionBody: `{"algorithm":"m.megolm.v1.aes-sha2"}`, + membersBody: `{"chunk":[ + {"type":"m.room.member","room_id":"!room:example.org","state_key":"@alice:example.org","sender":"@alice:example.org","content":{"membership":"join"}}, + {"type":"m.room.member","room_id":"!room:example.org","state_key":"@bot:example.org","sender":"@bot:example.org","content":{"membership":"join"}} + ]}`, + } + store := newTestStore(t, hs) + + members, err := store.GetRoomJoinedOrInvitedMembers(context.Background(), roomID) + if err != nil { + t.Fatalf("GetRoomJoinedOrInvitedMembers: %v", err) + } + if len(members) != 2 { + t.Fatalf("expected 2 members, got %d (%v) — the group session would be shared to nobody", len(members), members) + } + + if _, err := store.GetRoomJoinedOrInvitedMembers(context.Background(), roomID); err != nil { + t.Fatalf("second call: %v", err) + } + if got := hs.memberHits.Load(); got != 1 { + t.Errorf("expected members to be cached after one fetch, got %d", got) + } +}