From 141a340eaecf6fad79eed375d9b2d917ca8f47f5 Mon Sep 17 00:00:00 2001 From: NIKOLYA PRODIGY <_prodigy_@mail.ru> Date: Sat, 2 May 2026 19:22:38 +0300 Subject: [PATCH] connector: add optional Telegram portal approval workflow --- pkg/connector/approval.go | 217 +++++++++++ pkg/connector/capabilities.go | 2 +- pkg/connector/chatsync.go | 58 +-- pkg/connector/cleanupusers.go | 216 +++++++++++ pkg/connector/commands.go | 365 ++++++++++++++++++ pkg/connector/config.go | 48 +++ pkg/connector/connector.go | 6 +- pkg/connector/example-config.yaml | 31 ++ pkg/connector/handletelegram.go | 75 +++- pkg/connector/startchat.go | 6 +- pkg/connector/store/container.go | 2 + pkg/connector/store/portalapproval.go | 183 +++++++++ pkg/connector/store/upgrades/00-latest.sql | 20 + .../store/upgrades/10-portal-approval.sql | 21 + pkg/connector/tgapicall.go | 4 +- pkg/connector/tomatrix.go | 5 +- 16 files changed, 1215 insertions(+), 44 deletions(-) create mode 100644 pkg/connector/approval.go create mode 100644 pkg/connector/cleanupusers.go create mode 100644 pkg/connector/store/portalapproval.go create mode 100644 pkg/connector/store/upgrades/10-portal-approval.sql diff --git a/pkg/connector/approval.go b/pkg/connector/approval.go new file mode 100644 index 000000000..4f398f18b --- /dev/null +++ b/pkg/connector/approval.go @@ -0,0 +1,217 @@ +// mautrix-telegram - A Matrix-Telegram puppeting bridge. +// Copyright (C) 2026 Tulir Asokan +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +package connector + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/rs/zerolog" + "maunium.net/go/mautrix/bridgev2/networkid" + + "go.mau.fi/mautrix-telegram/pkg/connector/ids" + "go.mau.fi/mautrix-telegram/pkg/connector/store" + "go.mau.fi/mautrix-telegram/pkg/gotd/tg" +) + +type portalApprovalInfo struct { + PeerType string + EntityID int64 + TopicID int + Title string + Username string + IsBot bool +} + +func (tc *TelegramClient) portalApprovalInfoFromPeer(peer tg.PeerClass, topicID int, entities tg.Entities) portalApprovalInfo { + info := portalApprovalInfo{TopicID: topicID} + switch typed := peer.(type) { + case *tg.PeerUser: + info.PeerType = string(ids.PeerTypeUser) + info.EntityID = typed.UserID + if user, ok := entities.Users[typed.UserID]; ok { + info.Title = strings.TrimSpace(user.FirstName + " " + user.LastName) + info.Username = user.Username + info.IsBot = user.Bot + } + case *tg.PeerChat: + info.PeerType = string(ids.PeerTypeChat) + info.EntityID = typed.ChatID + if chat, ok := entities.Chats[typed.ChatID]; ok { + info.Title = chat.Title + } + case *tg.PeerChannel: + info.PeerType = string(ids.PeerTypeChannel) + info.EntityID = typed.ChannelID + if channel, ok := entities.Channels[typed.ChannelID]; ok { + if channel.Megagroup { + info.PeerType = "supergroup" + } + info.Title = channel.Title + info.Username = channel.Username + } + } + if info.Title == "" { + info.Title = fmt.Sprintf("%s:%d", info.PeerType, info.EntityID) + } + return info +} + +func (tc *TelegramClient) portalApprovalInfoFromObject(portalKey networkid.PortalKey, rawChat any) portalApprovalInfo { + peerType, entityID, topicID, _ := ids.ParsePortalID(portalKey.ID) + info := portalApprovalInfo{ + PeerType: string(peerType), + EntityID: entityID, + TopicID: topicID, + Title: fmt.Sprintf("%s:%d", peerType, entityID), + } + switch chat := rawChat.(type) { + case *tg.User: + info.Title = strings.TrimSpace(chat.FirstName + " " + chat.LastName) + info.Username = chat.Username + info.IsBot = chat.Bot + case *tg.Chat: + info.Title = chat.Title + case *tg.Channel: + if chat.Megagroup { + info.PeerType = "supergroup" + } + info.Title = chat.Title + info.Username = chat.Username + } + return info +} + +func (tc *TelegramClient) portalApprovalInfoFromDialog(portalKey networkid.PortalKey, dialog *tg.Dialog, users map[int64]tg.UserClass, chats map[int64]tg.ChatClass) portalApprovalInfo { + switch peer := dialog.Peer.(type) { + case *tg.PeerUser: + return tc.portalApprovalInfoFromObject(portalKey, users[peer.UserID]) + case *tg.PeerChat: + return tc.portalApprovalInfoFromObject(portalKey, chats[peer.ChatID]) + case *tg.PeerChannel: + return tc.portalApprovalInfoFromObject(portalKey, chats[peer.ChannelID]) + default: + return tc.portalApprovalInfoFromObject(portalKey, nil) + } +} + +func (tc *TelegramClient) portalApprovalAutoAllowed(info portalApprovalInfo) bool { + cfg := tc.main.Config.PortalApproval.AutoCreate + switch info.PeerType { + case string(ids.PeerTypeUser): + if info.IsBot { + if cfg.Bots != nil { + return *cfg.Bots + } + return cfg.PrivateChats + } + return cfg.PrivateChats + case string(ids.PeerTypeChat): + return cfg.Groups + case "supergroup": + return cfg.Supergroups + case string(ids.PeerTypeChannel): + return cfg.Channels + default: + return false + } +} + +func (tc *TelegramClient) portalApprovalStorageKey(portalKey networkid.PortalKey) networkid.PortalKey { + peerType, entityID, _, err := ids.ParsePortalID(portalKey.ID) + if err != nil { + return portalKey + } + return tc.makePortalKeyFromID(peerType, entityID, 0) +} + +func (tc *TelegramClient) ensurePortalApproved(ctx context.Context, portalKey networkid.PortalKey, info portalApprovalInfo, lastEvent string) (bool, error) { + if !tc.main.Config.PortalApproval.Enabled { + return true, nil + } + if err := tc.cleanupOldPendingPortalApprovals(ctx); err != nil { + return false, err + } + approvalKey := tc.portalApprovalStorageKey(portalKey) + userID := tc.telegramUserID + item, err := tc.main.Store.Approval.GetByPortal(ctx, userID, approvalKey.ID, approvalKey.Receiver) + if err != nil { + return false, err + } + if item != nil && item.Status == store.PortalApprovalAllowed { + return true, nil + } else if item != nil { + return false, nil + } + portal, err := tc.main.Bridge.GetExistingPortalByKey(ctx, portalKey) + if err != nil { + return false, err + } else if portal != nil && portal.MXID != "" { + return true, nil + } + + status := store.PortalApprovalPending + overwriteStatus := false + if tc.portalApprovalAutoAllowed(info) { + status = store.PortalApprovalAllowed + overwriteStatus = true + } else if !tc.main.Config.PortalApproval.Pending.Enabled { + return false, nil + } + + _, err = tc.main.Store.Approval.Upsert(ctx, store.PortalApproval{ + UserID: userID, + PortalID: approvalKey.ID, + PortalReceiver: approvalKey.Receiver, + PeerType: info.PeerType, + EntityID: info.EntityID, + TopicID: 0, + Title: info.Title, + Username: info.Username, + Status: status, + LastEvent: lastEvent, + }, overwriteStatus) + if err != nil { + return false, err + } + zerolog.Ctx(ctx).Info(). + Stringer("portal_key", portalKey). + Str("title", info.Title). + Str("approval_status", string(status)). + Msg("Stored Telegram portal approval state") + return status == store.PortalApprovalAllowed, nil +} + +func (tc *TelegramClient) ensurePortalApprovedForPeer(ctx context.Context, portalKey networkid.PortalKey, peer tg.PeerClass, topicID int, entities tg.Entities, lastEvent string) (bool, error) { + return tc.ensurePortalApproved(ctx, portalKey, tc.portalApprovalInfoFromPeer(peer, topicID, entities), lastEvent) +} + +func (tc *TelegramClient) ensurePortalApprovedForObject(ctx context.Context, portalKey networkid.PortalKey, chat any, lastEvent string) (bool, error) { + return tc.ensurePortalApproved(ctx, portalKey, tc.portalApprovalInfoFromObject(portalKey, chat), lastEvent) +} + +func (tc *TelegramClient) cleanupOldPendingPortalApprovals(ctx context.Context) error { + maxAgeHours := tc.main.Config.PortalApproval.Pending.MaxAgeHours + if maxAgeHours <= 0 { + return nil + } + cutoff := time.Now().Add(-time.Duration(maxAgeHours) * time.Hour).Unix() + deleted, err := tc.main.Store.Approval.DeletePendingOlderThan(ctx, tc.telegramUserID, cutoff) + if err != nil { + return err + } else if deleted > 0 { + zerolog.Ctx(ctx).Info(). + Int("max_age_hours", maxAgeHours). + Int64("deleted", deleted). + Msg("Cleaned old pending Telegram portal approval entries") + } + return nil +} diff --git a/pkg/connector/capabilities.go b/pkg/connector/capabilities.go index c024a51ef..366618a00 100644 --- a/pkg/connector/capabilities.go +++ b/pkg/connector/capabilities.go @@ -41,7 +41,7 @@ func (tc *TelegramConnector) GetCapabilities() *bridgev2.NetworkGeneralCapabilit CreateDM: true, LookupPhone: true, LookupUsername: true, - ContactList: true, + ContactList: tc.Config.ContactListEnabled(), Search: true, }, GroupCreation: map[string]bridgev2.GroupTypeCapabilities{ diff --git a/pkg/connector/chatsync.go b/pkg/connector/chatsync.go index b0a68bf13..01f4a6468 100644 --- a/pkg/connector/chatsync.go +++ b/pkg/connector/chatsync.go @@ -221,10 +221,44 @@ func (tc *TelegramClient) handleDialogs(ctx context.Context, dialogList []tg.Dia log.Debug().Msg("Syncing dialog") portalKey := tc.makePortalKeyFromPeer(dialog.GetPeer(), 0) - portal, err := tc.main.Bridge.GetPortalByKey(ctx, portalKey) + portal, err := tc.main.Bridge.GetExistingPortalByKey(ctx, portalKey) if err != nil { return err } + if portal == nil || portal.MXID == "" { + // Check what the latest message is before exposing the source in pending. + topMessage := messages[ids.MakeMessageID(dialog.Peer, dialog.TopMessage)] + if topMessage == nil { + if dialog.TopMessage == 0 { + log.Debug().Msg("Not syncing portal because there are no messages") + continue + } + log.Warn().Msg("TopMessage of dialog not in messages map") + } else if topMessage.TypeID() == tg.MessageServiceTypeID { + action := topMessage.(*tg.MessageService).Action + if action.TypeID() == tg.MessageActionContactSignUpTypeID || action.TypeID() == tg.MessageActionHistoryClearTypeID { + log.Debug().Str("action_type", action.TypeName()).Msg("Not syncing portal because it's a contact sign up or history clear") + continue + } + } + + if createLimit >= 0 && i >= createLimit { + continue + } + + ok, err := tc.ensurePortalApproved(ctx, portalKey, tc.portalApprovalInfoFromDialog(portalKey, dialog, users, chats), "dialog sync") + if err != nil { + return err + } else if !ok { + continue + } + + portal, err = tc.main.Bridge.GetPortalByKey(ctx, portalKey) + if err != nil { + return err + } + } + if dialog.UnreadCount == 0 && !dialog.UnreadMark { portal.Metadata.(*PortalMetadata).ReadUpTo = dialog.TopMessage } @@ -295,28 +329,6 @@ func (tc *TelegramClient) handleDialogs(ctx context.Context, dialogList []tg.Dia } } - if portal.MXID == "" { - // Check what the latest message is - topMessage := messages[ids.MakeMessageID(dialog.Peer, dialog.TopMessage)] - if topMessage == nil { - if dialog.TopMessage == 0 { - log.Debug().Msg("Not syncing portal because there are no messages") - continue - } - log.Warn().Msg("TopMessage of dialog not in messages map") - } else if topMessage.TypeID() == tg.MessageServiceTypeID { - action := topMessage.(*tg.MessageService).Action - if action.TypeID() == tg.MessageActionContactSignUpTypeID || action.TypeID() == tg.MessageActionHistoryClearTypeID { - log.Debug().Str("action_type", action.TypeName()).Msg("Not syncing portal because it's a contact sign up or history clear") - continue - } - } - - if createLimit >= 0 && i >= createLimit { - continue - } - } - tc.fillUserLocalMeta(chatInfo, dialog) res := tc.main.Bridge.QueueRemoteEvent(tc.userLogin, &simplevent.ChatResync{ diff --git a/pkg/connector/cleanupusers.go b/pkg/connector/cleanupusers.go new file mode 100644 index 000000000..39c2afaf4 --- /dev/null +++ b/pkg/connector/cleanupusers.go @@ -0,0 +1,216 @@ +// mautrix-telegram - A Matrix-Telegram puppeting bridge. +// Copyright (C) 2026 Tulir Asokan +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +package connector + +import ( + "context" + "fmt" + "strconv" + "strings" + + "maunium.net/go/mautrix/bridgev2/commands" + "maunium.net/go/mautrix/bridgev2/networkid" + "maunium.net/go/mautrix/format" + "maunium.net/go/mautrix/id" + + "go.mau.fi/mautrix-telegram/pkg/connector/ids" +) + +type cleanupUserCandidate struct { + ID networkid.UserID + MXID id.UserID + Name string + IsBot bool +} + +const ( + countUnusedGhostUsersQuery = ` + SELECT COUNT(*) + FROM ghost AS g + WHERE g.bridge_id=$1 + AND g.id<>'' + AND g.id NOT LIKE 'channel-%' + AND NOT EXISTS (SELECT 1 FROM user_login AS ul WHERE ul.bridge_id=g.bridge_id AND ul.id=g.id) + AND NOT EXISTS (SELECT 1 FROM portal AS p WHERE p.bridge_id=g.bridge_id AND p.other_user_id=g.id) + AND NOT EXISTS (SELECT 1 FROM message AS m WHERE m.bridge_id=g.bridge_id AND m.sender_id=g.id) + AND NOT EXISTS (SELECT 1 FROM reaction AS r WHERE r.bridge_id=g.bridge_id AND r.sender_id=g.id) + ` + listUnusedGhostUsersQuery = ` + SELECT g.id, g.name, g.is_bot + FROM ghost AS g + WHERE g.bridge_id=$1 + AND g.id<>'' + AND g.id NOT LIKE 'channel-%' + AND NOT EXISTS (SELECT 1 FROM user_login AS ul WHERE ul.bridge_id=g.bridge_id AND ul.id=g.id) + AND NOT EXISTS (SELECT 1 FROM portal AS p WHERE p.bridge_id=g.bridge_id AND p.other_user_id=g.id) + AND NOT EXISTS (SELECT 1 FROM message AS m WHERE m.bridge_id=g.bridge_id AND m.sender_id=g.id) + AND NOT EXISTS (SELECT 1 FROM reaction AS r WHERE r.bridge_id=g.bridge_id AND r.sender_id=g.id) + ORDER BY g.id + LIMIT $2 + ` + deleteGhostUserQuery = "DELETE FROM ghost WHERE bridge_id=$1 AND id=$2" +) + +var cmdCleanupUsers = &commands.FullHandler{ + Func: fnCleanupUsers, + Name: "cleanup-users", + Help: commands.HelpMeta{ + Section: commands.HelpSectionAdmin, + Description: "Find or delete unused Telegram ghost user records", + Args: "[--delete] [--limit N]", + }, + RequiresAdmin: true, +} + +func fnCleanupUsers(ce *commands.Event) { + deleteMode, limit, ok := parseCleanupUsersArgs(ce) + if !ok { + return + } + candidates, total, err := getCleanupUserCandidates(ce, limit) + if err != nil { + ce.Log.Err(err).Msg("Failed to list unused Telegram ghost users") + ce.Reply("Failed to list unused Telegram ghost users: %v", err) + return + } else if total == 0 { + ce.Reply("No unused Telegram ghost users found.") + return + } + + if deleteMode { + deleted, err := deleteCleanupUserCandidates(ce, candidates) + if err != nil { + ce.Log.Err(err).Msg("Failed to delete unused Telegram ghost users") + ce.Reply("Deleted %d unused Telegram ghost records, then failed: %v", deleted, err) + return + } + ce.Reply( + "Deleted %d unused Telegram ghost records from the bridge database.\n\n"+ + "Note: this command does not purge Synapse user rows or media files; it only removes bridge-side records that have no portals, messages, reactions or login.", + deleted, + ) + return + } + + var builder strings.Builder + fmt.Fprintf( + &builder, + "Found %d unused Telegram ghost users. Showing %d.\n\n"+ + "These records have no Telegram login, no DM portal, no bridged messages and no reactions.\n"+ + "Run `$cmdprefix cleanup-users --delete` to delete the listed bridge-side records.\n\n", + total, len(candidates), + ) + for i, candidate := range candidates { + fmt.Fprintf(&builder, "%d\\. **%s**\n", i+1, format.EscapeMarkdown(cleanupUserDisplayName(candidate))) + fmt.Fprintf(&builder, " mxid: `%s`\n", candidate.MXID) + fmt.Fprintf(&builder, " tg id: `%s`\n", candidate.ID) + if candidate.IsBot { + builder.WriteString(" bot: yes\n\n") + } else { + builder.WriteString(" bot: no\n\n") + } + } + if total > len(candidates) { + fmt.Fprintf(&builder, "Use `$cmdprefix cleanup-users --limit %d` to show more.\n", min(total, 200)) + } + ce.Reply(builder.String()) +} + +func parseCleanupUsersArgs(ce *commands.Event) (deleteMode bool, limit int, ok bool) { + limit = 20 + ok = true + for i := 0; i < len(ce.Args); i++ { + switch ce.Args[i] { + case "--delete": + deleteMode = true + case "--limit": + i++ + if i >= len(ce.Args) { + ce.Reply("Usage: `$cmdprefix cleanup-users [--delete] [--limit N]`") + return false, 0, false + } + parsed, err := strconv.Atoi(ce.Args[i]) + if err != nil || parsed <= 0 { + ce.Reply("Invalid limit: %s", format.SafeMarkdownCode(ce.Args[i])) + return false, 0, false + } + limit = parsed + default: + ce.Reply("Usage: `$cmdprefix cleanup-users [--delete] [--limit N]`") + return false, 0, false + } + } + if limit > 200 { + limit = 200 + } + return deleteMode, limit, true +} + +func getCleanupUserCandidates(ce *commands.Event, limit int) ([]cleanupUserCandidate, int, error) { + bridgeID := ce.Bridge.DB.BridgeID + var total int + err := ce.Bridge.DB.QueryRow(ce.Ctx, countUnusedGhostUsersQuery, bridgeID).Scan(&total) + if err != nil { + return nil, 0, err + } + rows, err := ce.Bridge.DB.Query(ce.Ctx, listUnusedGhostUsersQuery, bridgeID, limit) + if err != nil { + return nil, 0, err + } + defer rows.Close() + + candidates := make([]cleanupUserCandidate, 0, min(total, limit)) + for rows.Next() { + var candidate cleanupUserCandidate + if err = rows.Scan(&candidate.ID, &candidate.Name, &candidate.IsBot); err != nil { + return nil, 0, err + } + candidate.MXID = ce.Bridge.Matrix.GhostIntent(candidate.ID).GetMXID() + candidates = append(candidates, candidate) + } + return candidates, total, rows.Err() +} + +func deleteCleanupUserCandidates(ce *commands.Event, candidates []cleanupUserCandidate) (int, error) { + deleted := 0 + for _, candidate := range candidates { + if err := cleanupTelegramGhostIndexes(ce.Ctx, ce.Bridge.Network.(*TelegramConnector), candidate.ID); err != nil { + return deleted, err + } + if _, err := ce.Bridge.DB.Exec(ce.Ctx, deleteGhostUserQuery, ce.Bridge.DB.BridgeID, candidate.ID); err != nil { + return deleted, err + } + deleted++ + } + return deleted, nil +} + +func cleanupTelegramGhostIndexes(ctx context.Context, tc *TelegramConnector, userID networkid.UserID) error { + peerType, entityID, err := ids.ParseUserID(userID) + if err != nil { + // Corrupt unused ghost rows should not block cleanup of the bridge row itself. + return nil + } + if err = tc.Store.Username.Set(ctx, peerType, entityID, ""); err != nil { + return fmt.Errorf("failed to clear username index for %s: %w", userID, err) + } + if peerType == ids.PeerTypeUser { + if err = tc.Store.PhoneNumber.Set(ctx, entityID, ""); err != nil { + return fmt.Errorf("failed to clear phone index for %s: %w", userID, err) + } + } + return nil +} + +func cleanupUserDisplayName(candidate cleanupUserCandidate) string { + if candidate.Name != "" { + return candidate.Name + } + return string(candidate.MXID) +} diff --git a/pkg/connector/commands.go b/pkg/connector/commands.go index 509aea5fc..4adc243cd 100644 --- a/pkg/connector/commands.go +++ b/pkg/connector/commands.go @@ -18,21 +18,28 @@ package connector import ( "errors" + "fmt" "regexp" "slices" + "strconv" "strings" + "github.com/rs/zerolog" "golang.org/x/net/html" "maunium.net/go/mautrix/bridgev2" "maunium.net/go/mautrix/bridgev2/commands" "maunium.net/go/mautrix/bridgev2/networkid" + "maunium.net/go/mautrix/bridgev2/simplevent" "maunium.net/go/mautrix/format" "go.mau.fi/mautrix-telegram/pkg/connector/ids" + "go.mau.fi/mautrix-telegram/pkg/connector/store" "go.mau.fi/mautrix-telegram/pkg/gotd/tg" "go.mau.fi/mautrix-telegram/pkg/gotd/tgerr" ) +var helpSectionPortalApproval = commands.HelpSection{Name: "Telegram portal approval", Order: 21} + var cmdSyncChats = &commands.FullHandler{ Func: fnSyncChats, Name: "sync-chats", @@ -217,6 +224,364 @@ func fnJoin(ce *commands.Event) { ce.Reply("Successfully joined %s", html.EscapeString(chatName)) } +var cmdPending = &commands.FullHandler{ + Func: fnApprovalList(store.PortalApprovalPending), + Name: "pending", + Help: commands.HelpMeta{ + Section: helpSectionPortalApproval, + Description: "List Telegram chats waiting for approval", + }, + RequiresLogin: true, +} + +var cmdAllowed = &commands.FullHandler{ + Func: fnApprovalList(store.PortalApprovalAllowed), + Name: "allowed", + Help: commands.HelpMeta{ + Section: helpSectionPortalApproval, + Description: "List Telegram chats approved for portal creation", + }, + RequiresLogin: true, +} + +var cmdDenied = &commands.FullHandler{ + Func: fnApprovalList(store.PortalApprovalDenied), + Name: "denied", + Help: commands.HelpMeta{ + Section: helpSectionPortalApproval, + Description: "List Telegram chats denied for portal creation", + }, + RequiresLogin: true, +} + +var cmdAllow = &commands.FullHandler{ + Func: fnApprovalSetStatus(store.PortalApprovalAllowed, true, false, store.PortalApprovalPending), + Name: "allow", + Help: commands.HelpMeta{ + Section: helpSectionPortalApproval, + Description: "Approve a Telegram chat from the pending list and create its portal", + Args: "", + }, + RequiresLogin: true, +} + +var cmdDeny = &commands.FullHandler{ + Func: fnApprovalSetStatus(store.PortalApprovalDenied, false, true, store.PortalApprovalPending, store.PortalApprovalAllowed, store.PortalApprovalDenied), + Name: "deny", + Help: commands.HelpMeta{ + Section: helpSectionPortalApproval, + Description: "Deny a Telegram chat and optionally clean up its portal", + Args: "", + }, + RequiresLogin: true, +} + +var cmdUnallow = &commands.FullHandler{ + Func: fnApprovalSetStatus(store.PortalApprovalPending, false, true, store.PortalApprovalAllowed, store.PortalApprovalPending), + Name: "unallow", + Aliases: []string{"disallow"}, + Help: commands.HelpMeta{ + Section: helpSectionPortalApproval, + Description: "Move a Telegram chat back to pending and optionally clean up its portal", + Args: "", + }, + RequiresLogin: true, +} + +var cmdUndeny = &commands.FullHandler{ + Func: fnApprovalSetStatus(store.PortalApprovalPending, false, false, store.PortalApprovalDenied, store.PortalApprovalPending), + Name: "undeny", + Aliases: []string{"pardon"}, + Help: commands.HelpMeta{ + Section: helpSectionPortalApproval, + Description: "Move a denied Telegram chat back to pending", + Args: "", + }, + RequiresLogin: true, +} + +func approvalCommandClient(ce *commands.Event) *TelegramClient { + login := ce.User.GetDefaultLogin() + if login == nil { + ce.Reply("You're not logged in") + return nil + } + client, ok := login.Client.(*TelegramClient) + if !ok { + ce.Reply("Your default login is not a Telegram login") + return nil + } + return client +} + +func fnApprovalList(status store.PortalApprovalStatus) func(*commands.Event) { + return func(ce *commands.Event) { + client := approvalCommandClient(ce) + if client == nil { + return + } + if err := client.cleanupOldPendingPortalApprovals(ce.Ctx); err != nil { + ce.Log.Err(err).Msg("Failed to clean old pending Telegram portal approvals") + ce.Reply("Failed to clean old pending Telegram chats: %v", err) + return + } + items, err := client.main.Store.Approval.GetByStatus(ce.Ctx, client.telegramUserID, status) + if err != nil { + ce.Log.Err(err).Str("approval_status", string(status)).Msg("Failed to list Telegram portal approvals") + ce.Reply("Failed to list %s Telegram chats: %v", status, err) + return + } else if len(items) == 0 { + ce.Reply("No %s Telegram chats.", status) + return + } + + var builder strings.Builder + fmt.Fprintf(&builder, "%s Telegram chats:\n", approvalStatusTitle(status)) + firstSection := true + printed := map[string]struct{}{} + for _, peerType := range []string{string(ids.PeerTypeUser), string(ids.PeerTypeChat), "supergroup", string(ids.PeerTypeChannel)} { + printedAny := false + for _, item := range items { + if item.PeerType != peerType { + continue + } + if !printedAny { + printedAny = true + writeApprovalSectionHeader(&builder, approvalPeerTypeTitle(peerType), firstSection) + firstSection = false + } + builder.WriteString(formatApprovalItem(item)) + printed[item.PeerType] = struct{}{} + } + } + for _, item := range items { + if _, ok := printed[item.PeerType]; ok { + continue + } + if _, ok := printed[""]; !ok { + printed[""] = struct{}{} + writeApprovalSectionHeader(&builder, approvalPeerTypeTitle(item.PeerType), firstSection) + firstSection = false + } + builder.WriteString(formatApprovalItem(item)) + } + ce.Reply(builder.String()) + } +} + +func fnApprovalSetStatus(status store.PortalApprovalStatus, createPortal, cleanupPortal bool, allowedSourceStatuses ...store.PortalApprovalStatus) func(*commands.Event) { + return func(ce *commands.Event) { + client := approvalCommandClient(ce) + if client == nil { + return + } + if len(ce.Args) != 1 { + ce.Reply("Usage: `$cmdprefix allow|deny|unallow|undeny `") + return + } + approvalID, err := strconv.ParseInt(ce.Args[0], 10, 64) + if err != nil { + ce.Reply("Invalid approval number: %s", format.SafeMarkdownCode(ce.Args[0])) + return + } + if err = client.cleanupOldPendingPortalApprovals(ce.Ctx); err != nil { + ce.Log.Err(err).Msg("Failed to clean old pending Telegram portal approvals") + ce.Reply("Failed to clean old pending Telegram chats: %v", err) + return + } + item, err := client.main.Store.Approval.GetByID(ce.Ctx, client.telegramUserID, approvalID) + if err != nil { + ce.Log.Err(err).Int64("approval_id", approvalID).Msg("Failed to fetch Telegram portal approval") + ce.Reply("Failed to fetch approval item: %v", err) + return + } else if item == nil { + ce.Reply("No Telegram approval item found with number %d.", approvalID) + return + } + if !approvalStatusAllowed(item.Status, allowedSourceStatuses) { + ce.Reply( + "Approval item %d is currently %s, not %s. Run `$cmdprefix %s` to refresh the list.", + approvalID, item.Status, approvalAllowedSourceStatusText(allowedSourceStatuses), item.Status, + ) + return + } + if err = client.main.Store.Approval.SetStatus(ce.Ctx, client.telegramUserID, approvalID, status); err != nil { + ce.Log.Err(err).Int64("approval_id", approvalID).Str("approval_status", string(status)).Msg("Failed to update Telegram portal approval") + ce.Reply("Failed to update approval item: %v", err) + return + } + if createPortal { + portalKey := networkid.PortalKey{ID: item.PortalID, Receiver: item.PortalReceiver} + res := client.main.Bridge.QueueRemoteEvent(client.userLogin, &simplevent.ChatResync{ + EventMeta: simplevent.EventMeta{ + Type: bridgev2.RemoteEventChatResync, + PortalKey: portalKey, + CreatePortal: true, + LogContext: func(c zerolog.Context) zerolog.Context { + return c. + Int64("approval_id", approvalID). + Str("approval_command", "allow") + }, + }, + GetChatInfoFunc: client.GetChatInfo, + }) + if err = resultToError(res); err != nil { + ce.Log.Err(err).Int64("approval_id", approvalID).Msg("Failed to create approved Telegram portal") + ce.Reply("Approved %s, but failed to create the Matrix room: %v", approvalDisplayName(*item), err) + return + } + ce.Reply("Approved %s and requested Matrix room creation.", approvalDisplayName(*item)) + } else { + if cleanupPortal && client.shouldDeleteApprovalPortal(status) { + deleted, err := client.deleteApprovalPortal(ce, *item) + if err != nil { + ce.Log.Err(err).Int64("approval_id", approvalID).Msg("Failed to delete Telegram approval portal") + ce.Reply("Moved %s to %s, but failed to delete the Matrix portal: %v", approvalDisplayName(*item), status, err) + return + } else if deleted { + ce.Reply("Moved %s to %s and deleted the Matrix portal room.", approvalDisplayName(*item), status) + return + } + } + ce.Reply("Moved %s to %s.", approvalDisplayName(*item), status) + } + } +} + +func (tc *TelegramClient) shouldDeleteApprovalPortal(status store.PortalApprovalStatus) bool { + switch status { + case store.PortalApprovalDenied: + return tc.main.Config.PortalApproval.Cleanup.OnDeny.DeletePortal + case store.PortalApprovalPending: + return tc.main.Config.PortalApproval.Cleanup.OnUnallow.DeletePortal + default: + return false + } +} + +func (tc *TelegramClient) deleteApprovalPortal(ce *commands.Event, item store.PortalApproval) (bool, error) { + portalKey := networkid.PortalKey{ID: item.PortalID, Receiver: item.PortalReceiver} + portal, err := tc.main.Bridge.GetExistingPortalByKey(ce.Ctx, portalKey) + if err != nil { + return false, fmt.Errorf("failed to get Matrix portal: %w", err) + } else if portal == nil { + return false, nil + } + if err = tc.ensureApprovalPortalNotShared(ce, portalKey); err != nil { + return false, err + } + + roomID := portal.MXID + if err = portal.Delete(ce.Ctx); err != nil { + return false, fmt.Errorf("failed to delete bridge portal row: %w", err) + } + if roomID != "" { + if err = ce.Bot.DeleteRoom(ce.Ctx, roomID, false); err != nil { + return true, fmt.Errorf("failed to clean up Matrix room %s: %w", roomID, err) + } + } + return true, nil +} + +func (tc *TelegramClient) ensureApprovalPortalNotShared(ce *commands.Event, portalKey networkid.PortalKey) error { + userPortals, err := tc.main.Bridge.DB.UserPortal.GetAllInPortal(ce.Ctx, portalKey) + if err != nil { + return fmt.Errorf("failed to check portal users: %w", err) + } + for _, userPortal := range userPortals { + if userPortal.LoginID != tc.userLogin.ID { + return fmt.Errorf("portal is also used by another login (%s), not deleting shared Matrix room", userPortal.LoginID) + } + } + return nil +} + +func approvalStatusAllowed(status store.PortalApprovalStatus, allowed []store.PortalApprovalStatus) bool { + for _, allowedStatus := range allowed { + if status == allowedStatus { + return true + } + } + return false +} + +func approvalAllowedSourceStatusText(statuses []store.PortalApprovalStatus) string { + if len(statuses) == 0 { + return "any status" + } + parts := make([]string, len(statuses)) + for i, status := range statuses { + parts[i] = string(status) + } + return strings.Join(parts, " or ") +} + +func approvalStatusTitle(status store.PortalApprovalStatus) string { + switch status { + case store.PortalApprovalPending: + return "Pending" + case store.PortalApprovalAllowed: + return "Allowed" + case store.PortalApprovalDenied: + return "Denied" + default: + return string(status) + } +} + +func approvalPeerTypeTitle(peerType string) string { + switch peerType { + case string(ids.PeerTypeUser): + return "Private chats" + case string(ids.PeerTypeChat): + return "Groups" + case "supergroup": + return "Supergroups" + case string(ids.PeerTypeChannel): + return "Channels" + default: + return "Other" + } +} + +func writeApprovalSectionHeader(builder *strings.Builder, title string, first bool) { + if first { + fmt.Fprintf(builder, "\n**%s:**\n\n", format.EscapeMarkdown(title)) + } else { + fmt.Fprintf(builder, "\n\n**%s:**\n\n", format.EscapeMarkdown(title)) + } +} + +func approvalDisplayName(item store.PortalApproval) string { + if item.Username != "" { + return fmt.Sprintf("%s (@%s)", item.Title, item.Username) + } + return item.Title +} + +func formatApprovalItem(item store.PortalApproval) string { + var builder strings.Builder + fmt.Fprintf(&builder, "%d\\. **%s**\n", item.ApprovalID, format.EscapeMarkdown(item.Title)) + if item.Username != "" { + fmt.Fprintf(&builder, " username: @%s\n", format.EscapeMarkdown(item.Username)) + } else { + builder.WriteString(" username: -\n") + } + fmt.Fprintf(&builder, " id: %s\n\n", format.EscapeMarkdown(approvalTelegramID(item))) + return builder.String() +} + +func approvalTelegramID(item store.PortalApproval) string { + switch item.PeerType { + case "supergroup", string(ids.PeerTypeChannel): + return fmt.Sprintf("-100%d", item.EntityID) + case string(ids.PeerTypeChat): + return fmt.Sprintf("-%d", item.EntityID) + default: + return fmt.Sprintf("%d", item.EntityID) + } +} + var cmdEmojiPack = &commands.FullHandler{ Func: fnEmojiPack, Name: "emoji-pack", diff --git a/pkg/connector/config.go b/pkg/connector/config.go index 7308846bf..547212659 100644 --- a/pkg/connector/config.go +++ b/pkg/connector/config.go @@ -62,6 +62,32 @@ type ProxyConfig struct { Password string `yaml:"password"` } +type PortalApprovalConfig struct { + Enabled bool `yaml:"enabled"` + + AutoCreate struct { + PrivateChats bool `yaml:"private_chats"` + Bots *bool `yaml:"bots"` + Groups bool `yaml:"groups"` + Supergroups bool `yaml:"supergroups"` + Channels bool `yaml:"channels"` + } `yaml:"auto_create"` + + Pending struct { + Enabled bool `yaml:"enabled"` + MaxAgeHours int `yaml:"max_age_hours"` + } `yaml:"pending"` + + Cleanup struct { + OnDeny struct { + DeletePortal bool `yaml:"delete_portal"` + } `yaml:"on_deny"` + OnUnallow struct { + DeletePortal bool `yaml:"delete_portal"` + } `yaml:"on_unallow"` + } `yaml:"cleanup"` +} + type TelegramConfig struct { APIID int `yaml:"api_id"` APIHash string `yaml:"api_hash"` @@ -77,6 +103,12 @@ type TelegramConfig struct { ProxyConfig ProxyConfig `yaml:"proxy"` + PortalApproval PortalApprovalConfig `yaml:"portal_approval"` + + ContactList struct { + Enabled *bool `yaml:"enabled"` + } `yaml:"contact_list"` + Sync struct { UpdateLimit int `yaml:"update_limit"` CreateLimit int `yaml:"create_limit"` @@ -106,6 +138,10 @@ func (c TelegramConfig) ShouldBridge(participantCount int) bool { return c.MaxMemberCount < 0 || participantCount <= c.MaxMemberCount } +func (c TelegramConfig) ContactListEnabled() bool { + return c.ContactList.Enabled == nil || *c.ContactList.Enabled +} + type DisplaynameParams struct { FullName string FirstName string @@ -174,6 +210,17 @@ func upgradeConfig(helper up.Helper) { helper.Copy(up.Str|up.Null, "proxy", "address") helper.Copy(up.Str|up.Null, "proxy", "username") helper.Copy(up.Str|up.Null, "proxy", "password") + helper.Copy(up.Bool, "portal_approval", "enabled") + helper.Copy(up.Bool, "portal_approval", "auto_create", "private_chats") + helper.Copy(up.Bool, "portal_approval", "auto_create", "bots") + helper.Copy(up.Bool, "portal_approval", "auto_create", "groups") + helper.Copy(up.Bool, "portal_approval", "auto_create", "supergroups") + helper.Copy(up.Bool, "portal_approval", "auto_create", "channels") + helper.Copy(up.Bool, "portal_approval", "pending", "enabled") + helper.Copy(up.Int, "portal_approval", "pending", "max_age_hours") + helper.Copy(up.Bool, "portal_approval", "cleanup", "on_deny", "delete_portal") + helper.Copy(up.Bool, "portal_approval", "cleanup", "on_unallow", "delete_portal") + helper.Copy(up.Bool, "contact_list", "enabled") helper.Copy(up.Int, "sync", "update_limit") helper.Copy(up.Int, "sync", "create_limit") helper.Copy(up.Int, "sync", "login_sync_limit") @@ -201,6 +248,7 @@ func (tc *TelegramConnector) GetConfig() (example string, data any, upgrader up. {"member_list"}, {"ping"}, {"proxy"}, + {"portal_approval"}, {"sync"}, {"takeout"}, {"max_member_count"}, diff --git a/pkg/connector/connector.go b/pkg/connector/connector.go index 81367a404..f37e8379c 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -41,7 +41,11 @@ var _ bridgev2.MaxFileSizeingNetwork = (*TelegramConnector)(nil) func (tc *TelegramConnector) Init(bridge *bridgev2.Bridge) { tc.Store = store.NewStore(bridge.DB.Database, dbutil.ZeroLogger(bridge.Log.With().Str("db_section", "telegram").Logger())) tc.Bridge = bridge - tc.Bridge.Commands.(*commands.Processor).AddHandlers(cmdSyncChats, cmdEmojiPack, cmdUpgrade, cmdJoin) + tc.Bridge.Commands.(*commands.Processor).AddHandlers( + cmdSyncChats, cmdEmojiPack, cmdUpgrade, cmdJoin, + cmdPending, cmdAllowed, cmdDenied, cmdAllow, cmdDeny, cmdUnallow, cmdUndeny, + cmdCleanupUsers, + ) } func (tc *TelegramConnector) Start(ctx context.Context) error { diff --git a/pkg/connector/example-config.yaml b/pkg/connector/example-config.yaml index 98e46ef33..4f2337c51 100644 --- a/pkg/connector/example-config.yaml +++ b/pkg/connector/example-config.yaml @@ -65,6 +65,37 @@ proxy: username: password: +portal_approval: + # If enabled, portals are only created automatically for chat types allowed + # below. Other sources can be approved with Matrix commands. + enabled: false + auto_create: + private_chats: true + # Bot direct chats are controlled separately from normal private chats. + bots: true + groups: false + supergroups: false + channels: false + pending: + enabled: true + # Automatically remove pending approval entries that haven't been seen + # for this many hours. Set to 0 to keep them forever. + max_age_hours: 72 + cleanup: + on_deny: + # Delete the existing Matrix portal room and bridge portal database row + # when a pending/allowed Telegram chat is denied. + delete_portal: false + on_unallow: + # Delete the existing Matrix portal room and bridge portal database row + # when an allowed Telegram chat is moved back to pending. + delete_portal: false + +contact_list: + # Allow clients/provisioning API to fetch Telegram contacts. + # Fetching contacts may create Matrix ghost users for those contacts. + enabled: true + sync: # Number of most recently active dialogs to check when syncing chats. # Set to -1 to remove limit. diff --git a/pkg/connector/handletelegram.go b/pkg/connector/handletelegram.go index f893a33d5..9f07f50df 100644 --- a/pkg/connector/handletelegram.go +++ b/pkg/connector/handletelegram.go @@ -160,6 +160,12 @@ func (tc *TelegramClient) onUpdateChannel(ctx context.Context, e tg.Entities, up log.Debug().Msg("Update was for a left channel. Leaving the channel.") return tc.selfLeaveChat(ctx, portalKey, fmt.Errorf("channel has left=true after UpdateChannel")) } + ok, err := tc.ensurePortalApprovedForObject(ctx, portalKey, channel, "update channel") + if err != nil { + return err + } else if !ok { + return nil + } res := tc.main.Bridge.QueueRemoteEvent(tc.userLogin, &simplevent.ChatResync{ EventMeta: simplevent.EventMeta{ Type: bridgev2.RemoteEventChatResync, @@ -208,8 +214,6 @@ func (tc *TelegramClient) onUpdateNewMessage(ctx context.Context, entities tg.En } } - sender := tc.getEventSender(msg, isBroadcastChannel) - if media, ok := msg.GetMedia(); ok && media.TypeID() == tg.MessageMediaContactTypeID { contact := media.(*tg.MessageMediaContact) // TODO update the corresponding puppet @@ -217,6 +221,14 @@ func (tc *TelegramClient) onUpdateNewMessage(ctx context.Context, entities tg.En } topicID := tc.getTopicID(ctx, msg.PeerID, msg.ReplyTo) + portalKey := tc.makePortalKeyFromPeer(msg.PeerID, topicID) + ok, err := tc.ensurePortalApprovedForPeer(ctx, portalKey, msg.PeerID, topicID, entities, "new message") + if err != nil { + return err + } else if !ok { + return nil + } + sender := tc.getEventSender(msg, isBroadcastChannel) res := tc.main.Bridge.QueueRemoteEvent(tc.userLogin, &simplevent.Message[*tg.Message]{ EventMeta: simplevent.EventMeta{ Type: bridgev2.RemoteEventMessage, @@ -229,7 +241,7 @@ func (tc *TelegramClient) onUpdateNewMessage(ctx context.Context, entities tg.En Stringer("peer_id", msg.PeerID) }, Sender: sender, - PortalKey: tc.makePortalKeyFromPeer(msg.PeerID, topicID), + PortalKey: portalKey, CreatePortal: true, Timestamp: time.Unix(int64(msg.Date), 0), StreamOrder: int64(msg.GetID()), @@ -248,7 +260,7 @@ func (tc *TelegramClient) onUpdateNewMessage(ctx context.Context, entities tg.En } return nil case *tg.MessageService: - return tc.handleServiceMessage(ctx, msg) + return tc.handleServiceMessage(ctx, entities, msg) default: log.Warn(). @@ -283,12 +295,20 @@ func rawGetTopicID(rawReplyTo tg.MessageReplyHeaderClass) int { return 0 } -func (tc *TelegramClient) handleServiceMessage(ctx context.Context, msg *tg.MessageService) error { +func (tc *TelegramClient) handleServiceMessage(ctx context.Context, entities tg.Entities, msg *tg.MessageService) error { log := zerolog.Ctx(ctx) + topicID := tc.getTopicID(ctx, msg.PeerID, msg.ReplyTo) + portalKey := tc.makePortalKeyFromPeer(msg.PeerID, topicID) + ok, err := tc.ensurePortalApprovedForPeer(ctx, portalKey, msg.PeerID, topicID, entities, "service message") + if err != nil { + return err + } else if !ok { + return nil + } sender := tc.getEventSender(msg, false) eventMeta := simplevent.EventMeta{ - PortalKey: tc.makePortalKeyFromPeer(msg.PeerID, tc.getTopicID(ctx, msg.PeerID, msg.ReplyTo)), + PortalKey: portalKey, Sender: sender, Timestamp: time.Unix(int64(msg.Date), 0), LogContext: func(c zerolog.Context) zerolog.Context { @@ -518,8 +538,11 @@ func (tc *TelegramClient) handleServiceMessage(ctx context.Context, msg *tg.Mess body.WriteString(", ") } - if ghost, err := tc.main.Bridge.GetGhostByID(ctx, ids.MakeUserID(userID)); err != nil { + if ghost, err := tc.main.Bridge.GetExistingGhostByID(ctx, ids.MakeUserID(userID)); err != nil { return err + } else if ghost == nil { + body.WriteString(fmt.Sprintf("user %d", userID)) + html.WriteString(fmt.Sprintf("user %d", userID)) } else { var name string if username, err := tc.main.Store.Username.Get(ctx, ids.PeerTypeUser, userID); err != nil { @@ -722,9 +745,12 @@ func (tc *TelegramClient) getPeerSender(peer tg.PeerClass) bridgev2.EventSender } func (tc *TelegramClient) onUserName(ctx context.Context, e tg.Entities, update *tg.UpdateUserName) error { - ghost, err := tc.main.Bridge.GetGhostByID(ctx, ids.MakeUserID(update.UserID)) + ghost, err := tc.main.Bridge.GetExistingGhostByID(ctx, ids.MakeUserID(update.UserID)) if err != nil { return err + } else if ghost == nil { + // Don't auto-create ghosts from background profile updates. + return nil } meta := ghost.Metadata.(*GhostMetadata) @@ -809,8 +835,17 @@ func (tc *TelegramClient) onDeleteMessages(ctx context.Context, channelID int64, return nil } -func (tc *TelegramClient) updateGhost(ctx context.Context, userID int64, user *tg.User) (*bridgev2.UserInfo, error) { - ghost, err := tc.main.Bridge.GetGhostByID(ctx, ids.MakeUserID(userID)) +func (tc *TelegramClient) updateGhost(ctx context.Context, userID int64, user *tg.User, createIfMissing bool) (*bridgev2.UserInfo, error) { + var ghost *bridgev2.Ghost + var err error + if createIfMissing { + ghost, err = tc.main.Bridge.GetGhostByID(ctx, ids.MakeUserID(userID)) + } else { + ghost, err = tc.main.Bridge.GetExistingGhostByID(ctx, ids.MakeUserID(userID)) + if err == nil && ghost == nil { + return nil, nil + } + } if err != nil { return nil, err } @@ -827,13 +862,21 @@ func (tc *TelegramClient) updateGhost(ctx context.Context, userID int64, user *t return userInfo, nil } -func (tc *TelegramClient) updateChannel(ctx context.Context, channel *tg.Channel) (*bridgev2.UserInfo, error) { +func (tc *TelegramClient) updateChannel(ctx context.Context, channel *tg.Channel, createIfMissing bool) (*bridgev2.UserInfo, error) { // TODO resync portal metadata? userInfo, err := tc.wrapChannelGhostInfo(ctx, channel) if err != nil { return nil, err } - ghost, err := tc.main.Bridge.GetGhostByID(ctx, ids.MakeChannelUserID(channel.ID)) + var ghost *bridgev2.Ghost + if createIfMissing { + ghost, err = tc.main.Bridge.GetGhostByID(ctx, ids.MakeChannelUserID(channel.ID)) + } else { + ghost, err = tc.main.Bridge.GetExistingGhostByID(ctx, ids.MakeChannelUserID(channel.ID)) + if err == nil && ghost == nil { + return nil, nil + } + } if err != nil { return nil, err } @@ -884,7 +927,7 @@ func (tc *TelegramClient) onUpdate(ctx context.Context, e tg.Entities, upd tg.Up zerolog.Ctx(ctx).Trace().Stringer("update", upd).Msg("Raw update") for userID, user := range e.Users { zerolog.Ctx(ctx).Trace().Stringer("user", user).Msg("Raw user info in update") - if _, err := tc.updateGhost(ctx, userID, user); err != nil { + if _, err := tc.updateGhost(ctx, userID, user, false); err != nil { return err } } @@ -900,7 +943,7 @@ func (tc *TelegramClient) onUpdate(ctx context.Context, e tg.Entities, upd tg.Up if channel.GetLeft() { tc.selfLeaveChat(ctx, tc.makePortalKeyFromID(ids.PeerTypeChannel, channel.ID, 0), fmt.Errorf("left flag in entity update")) } - if _, err := tc.updateChannel(ctx, channel); err != nil { + if _, err := tc.updateChannel(ctx, channel, false); err != nil { return err } } @@ -1059,9 +1102,11 @@ func (tc *TelegramClient) onMessageEdit(ctx context.Context, update IGetMessage) } portalKey := tc.makePortalKeyFromPeer(msg.PeerID, topicID) - portal, err := tc.main.Bridge.GetPortalByKey(ctx, portalKey) + portal, err := tc.main.Bridge.GetExistingPortalByKey(ctx, portalKey) if err != nil { return err + } else if portal == nil || portal.MXID == "" { + return nil } sender := tc.getEventSender(msg, !portal.Metadata.(*PortalMetadata).IsSuperGroup) diff --git a/pkg/connector/startchat.go b/pkg/connector/startchat.go index 58e322c3d..eb7206fd8 100644 --- a/pkg/connector/startchat.go +++ b/pkg/connector/startchat.go @@ -27,6 +27,7 @@ import ( "github.com/rs/zerolog" "go.mau.fi/util/ptr" + "maunium.net/go/mautrix" "maunium.net/go/mautrix/bridgev2" "go.mau.fi/mautrix-telegram/pkg/connector/ids" @@ -97,7 +98,7 @@ func (tc *TelegramClient) resolveUserID(ctx context.Context, userID int64) (resp return nil, fmt.Errorf("failed to get user with ID %d: %w", userID, err) } else if user.TypeID() != tg.UserTypeID { return nil, fmt.Errorf("unexpected user type: %T", user) - } else if userInfo, err := tc.updateGhost(ctx, userID, user.(*tg.User)); err != nil { + } else if userInfo, err := tc.updateGhost(ctx, userID, user.(*tg.User), true); err != nil { return nil, fmt.Errorf("failed to update ghost: %w", err) } else { if resp.Ghost == nil { @@ -231,6 +232,9 @@ func (tc *TelegramClient) SearchUsers(ctx context.Context, query string) (resp [ } func (tc *TelegramClient) GetContactList(ctx context.Context) (resp []*bridgev2.ResolveIdentifierResponse, err error) { + if !tc.main.Config.ContactListEnabled() { + return nil, bridgev2.RespError(mautrix.MForbidden.WithMessage("Telegram contact list fetching is disabled")) + } tc.contactsLock.Lock() defer tc.contactsLock.Unlock() var contacts *tg.ContactsContacts diff --git a/pkg/connector/store/container.go b/pkg/connector/store/container.go index b18a376a0..60fd8ec4b 100644 --- a/pkg/connector/store/container.go +++ b/pkg/connector/store/container.go @@ -32,6 +32,7 @@ type Container struct { Username *UsernameQuery PhoneNumber *PhoneNumberQuery Topic *TopicQuery + Approval *PortalApprovalQuery } func NewStore(db *dbutil.Database, log dbutil.DatabaseLogger) *Container { @@ -42,6 +43,7 @@ func NewStore(db *dbutil.Database, log dbutil.DatabaseLogger) *Container { Username: &UsernameQuery{db}, PhoneNumber: &PhoneNumberQuery{db}, Topic: &TopicQuery{db: db, existingTopics: exsync.NewSet[topicKey]()}, + Approval: &PortalApprovalQuery{db: db}, } } diff --git a/pkg/connector/store/portalapproval.go b/pkg/connector/store/portalapproval.go new file mode 100644 index 000000000..527e377b4 --- /dev/null +++ b/pkg/connector/store/portalapproval.go @@ -0,0 +1,183 @@ +// mautrix-telegram - A Matrix-Telegram puppeting bridge. +// Copyright (C) 2026 Tulir Asokan +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +package store + +import ( + "context" + "database/sql" + "errors" + "time" + + "go.mau.fi/util/dbutil" + "maunium.net/go/mautrix/bridgev2/networkid" +) + +type PortalApprovalStatus string + +const ( + PortalApprovalPending PortalApprovalStatus = "pending" + PortalApprovalAllowed PortalApprovalStatus = "allowed" + PortalApprovalDenied PortalApprovalStatus = "denied" +) + +type PortalApproval struct { + ApprovalID int64 + UserID int64 + PortalID networkid.PortalID + PortalReceiver networkid.UserLoginID + PeerType string + EntityID int64 + TopicID int + Title string + Username string + Status PortalApprovalStatus + LastEvent string + CreatedTS int64 + LastSeenTS int64 +} + +type PortalApprovalQuery struct { + db *dbutil.Database +} + +const ( + getNextPortalApprovalIDQuery = "SELECT COALESCE(MAX(approval_id), 0) + 1 FROM telegram_portal_approval" + upsertPortalApprovalQuery = ` + INSERT INTO telegram_portal_approval ( + approval_id, user_id, portal_id, portal_receiver, peer_type, entity_id, topic_id, + title, username, status, last_event, created_ts, last_seen_ts + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) + ON CONFLICT (user_id, portal_id, portal_receiver) DO UPDATE SET + peer_type=excluded.peer_type, + entity_id=excluded.entity_id, + topic_id=excluded.topic_id, + title=excluded.title, + username=excluded.username, + last_event=excluded.last_event, + last_seen_ts=excluded.last_seen_ts + ` + upsertPortalApprovalWithStatusQuery = ` + INSERT INTO telegram_portal_approval ( + approval_id, user_id, portal_id, portal_receiver, peer_type, entity_id, topic_id, + title, username, status, last_event, created_ts, last_seen_ts + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) + ON CONFLICT (user_id, portal_id, portal_receiver) DO UPDATE SET + peer_type=excluded.peer_type, + entity_id=excluded.entity_id, + topic_id=excluded.topic_id, + title=excluded.title, + username=excluded.username, + status=excluded.status, + last_event=excluded.last_event, + last_seen_ts=excluded.last_seen_ts + ` + getPortalApprovalByPortalQuery = ` + SELECT approval_id, user_id, portal_id, portal_receiver, peer_type, entity_id, topic_id, + title, username, status, last_event, created_ts, last_seen_ts + FROM telegram_portal_approval + WHERE user_id=$1 AND portal_id=$2 AND portal_receiver=$3 + ` + getPortalApprovalByIDQuery = ` + SELECT approval_id, user_id, portal_id, portal_receiver, peer_type, entity_id, topic_id, + title, username, status, last_event, created_ts, last_seen_ts + FROM telegram_portal_approval + WHERE user_id=$1 AND approval_id=$2 + ` + getPortalApprovalByStatusQuery = ` + SELECT approval_id, user_id, portal_id, portal_receiver, peer_type, entity_id, topic_id, + title, username, status, last_event, created_ts, last_seen_ts + FROM telegram_portal_approval + WHERE user_id=$1 AND status=$2 + ORDER BY approval_id + ` + setPortalApprovalStatusQuery = "UPDATE telegram_portal_approval SET status=$3, last_seen_ts=$4 WHERE user_id=$1 AND approval_id=$2" + deleteOldPendingPortalApprovalQuery = "DELETE FROM telegram_portal_approval WHERE user_id=$1 AND status=$2 AND last_seen_ts<$3" +) + +var portalApprovalScanner = dbutil.ConvertRowFn[PortalApproval](func(row dbutil.Scannable) (item PortalApproval, err error) { + var portalID, portalReceiver, status string + err = row.Scan( + &item.ApprovalID, &item.UserID, &portalID, &portalReceiver, &item.PeerType, + &item.EntityID, &item.TopicID, &item.Title, &item.Username, &status, + &item.LastEvent, &item.CreatedTS, &item.LastSeenTS, + ) + item.PortalID = networkid.PortalID(portalID) + item.PortalReceiver = networkid.UserLoginID(portalReceiver) + item.Status = PortalApprovalStatus(status) + return +}) + +func (q *PortalApprovalQuery) Upsert(ctx context.Context, item PortalApproval, overwriteStatus bool) (*PortalApproval, error) { + now := time.Now().Unix() + if item.CreatedTS == 0 { + item.CreatedTS = now + } + if item.LastSeenTS == 0 { + item.LastSeenTS = now + } + if item.Status == "" { + item.Status = PortalApprovalPending + } + if item.ApprovalID == 0 { + if err := q.db.QueryRow(ctx, getNextPortalApprovalIDQuery).Scan(&item.ApprovalID); err != nil { + return nil, err + } + } + query := upsertPortalApprovalQuery + if overwriteStatus { + query = upsertPortalApprovalWithStatusQuery + } + _, err := q.db.Exec( + ctx, query, + item.ApprovalID, item.UserID, item.PortalID, item.PortalReceiver, item.PeerType, + item.EntityID, item.TopicID, item.Title, item.Username, item.Status, + item.LastEvent, item.CreatedTS, item.LastSeenTS, + ) + if err != nil { + return nil, err + } + return q.GetByPortal(ctx, item.UserID, item.PortalID, item.PortalReceiver) +} + +func (q *PortalApprovalQuery) GetByPortal(ctx context.Context, userID int64, portalID networkid.PortalID, receiver networkid.UserLoginID) (*PortalApproval, error) { + item, err := portalApprovalScanner(q.db.QueryRow(ctx, getPortalApprovalByPortalQuery, userID, portalID, receiver)) + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + return &item, err +} + +func (q *PortalApprovalQuery) GetByID(ctx context.Context, userID, approvalID int64) (*PortalApproval, error) { + item, err := portalApprovalScanner(q.db.QueryRow(ctx, getPortalApprovalByIDQuery, userID, approvalID)) + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + return &item, err +} + +func (q *PortalApprovalQuery) GetByStatus(ctx context.Context, userID int64, status PortalApprovalStatus) ([]PortalApproval, error) { + return portalApprovalScanner.NewRowIter(q.db.Query(ctx, getPortalApprovalByStatusQuery, userID, status)).AsList() +} + +func (q *PortalApprovalQuery) SetStatus(ctx context.Context, userID, approvalID int64, status PortalApprovalStatus) error { + _, err := q.db.Exec(ctx, setPortalApprovalStatusQuery, userID, approvalID, status, time.Now().Unix()) + return err +} + +func (q *PortalApprovalQuery) DeletePendingOlderThan(ctx context.Context, userID, cutoffTS int64) (int64, error) { + res, err := q.db.Exec(ctx, deleteOldPendingPortalApprovalQuery, userID, PortalApprovalPending, cutoffTS) + if err != nil { + return 0, err + } + count, err := res.RowsAffected() + if err != nil { + return 0, nil + } + return count, nil +} diff --git a/pkg/connector/store/upgrades/00-latest.sql b/pkg/connector/store/upgrades/00-latest.sql index aafb56cc4..391812166 100644 --- a/pkg/connector/store/upgrades/00-latest.sql +++ b/pkg/connector/store/upgrades/00-latest.sql @@ -63,3 +63,23 @@ CREATE TABLE telegram_topic ( PRIMARY KEY (channel_id, topic_id) ); + +CREATE TABLE telegram_portal_approval ( + approval_id BIGINT NOT NULL PRIMARY KEY, + user_id BIGINT NOT NULL, + portal_id TEXT NOT NULL, + portal_receiver TEXT NOT NULL, + peer_type TEXT NOT NULL, + entity_id BIGINT NOT NULL, + topic_id BIGINT NOT NULL, + title TEXT NOT NULL, + username TEXT NOT NULL, + status TEXT NOT NULL, + last_event TEXT NOT NULL, + created_ts BIGINT NOT NULL, + last_seen_ts BIGINT NOT NULL, + + UNIQUE (user_id, portal_id, portal_receiver) +); + +CREATE INDEX telegram_portal_approval_user_status_idx ON telegram_portal_approval (user_id, status, approval_id); diff --git a/pkg/connector/store/upgrades/10-portal-approval.sql b/pkg/connector/store/upgrades/10-portal-approval.sql new file mode 100644 index 000000000..13f1328b2 --- /dev/null +++ b/pkg/connector/store/upgrades/10-portal-approval.sql @@ -0,0 +1,21 @@ +-- v10 (compatible with v2+): Store Telegram portal approval state + +CREATE TABLE telegram_portal_approval ( + approval_id BIGINT NOT NULL PRIMARY KEY, + user_id BIGINT NOT NULL, + portal_id TEXT NOT NULL, + portal_receiver TEXT NOT NULL, + peer_type TEXT NOT NULL, + entity_id BIGINT NOT NULL, + topic_id BIGINT NOT NULL, + title TEXT NOT NULL, + username TEXT NOT NULL, + status TEXT NOT NULL, + last_event TEXT NOT NULL, + created_ts BIGINT NOT NULL, + last_seen_ts BIGINT NOT NULL, + + UNIQUE (user_id, portal_id, portal_receiver) +); + +CREATE INDEX telegram_portal_approval_user_status_idx ON telegram_portal_approval (user_id, status, approval_id); diff --git a/pkg/connector/tgapicall.go b/pkg/connector/tgapicall.go index 72bc7afc4..c0c97347f 100644 --- a/pkg/connector/tgapicall.go +++ b/pkg/connector/tgapicall.go @@ -42,7 +42,7 @@ func handleUserUpdates[U hasUserUpdates](ctx context.Context, t *TelegramClient, if !ok { return fmt.Errorf("user is %T not *tg.User", user) } - _, err := t.updateGhost(ctx, user.ID, user) + _, err := t.updateGhost(ctx, user.ID, user, false) if err != nil { return err } @@ -53,7 +53,7 @@ func handleUserUpdates[U hasUserUpdates](ctx context.Context, t *TelegramClient, func handleChatUpdates[U hasChatUpdates](ctx context.Context, t *TelegramClient, resp hasChatUpdates) error { for _, c := range resp.GetChats() { if channel, ok := c.(*tg.Channel); ok { - if _, err := t.updateChannel(ctx, channel); err != nil { + if _, err := t.updateChannel(ctx, channel, false); err != nil { return err } } diff --git a/pkg/connector/tomatrix.go b/pkg/connector/tomatrix.go index 326f3374d..c16e8aa01 100644 --- a/pkg/connector/tomatrix.go +++ b/pkg/connector/tomatrix.go @@ -306,8 +306,11 @@ func (tc *TelegramClient) addForwardHeader(ctx context.Context, part *bridgev2.C if user != nil { mxid = user.UserMXID fwdFromText = cmp.Or(user.RemoteName, user.UserMXID.String()) - } else if ghost, err := tc.main.Bridge.GetGhostByID(ctx, ids.MakeUserID(from.UserID)); err != nil { + } else if ghost, err := tc.main.Bridge.GetExistingGhostByID(ctx, ids.MakeUserID(from.UserID)); err != nil { return err + } else if ghost == nil { + fwdFromText = cmp.Or(fwd.FromName, "unknown user") + fwdFromHTML = fmt.Sprintf("%s", html.EscapeString(fwdFromText)) } else { if ghost.Name == "" { info, err := tc.GetUserInfo(ctx, ghost)