astra/owpengram-server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: sync account freeze lifecycle

Browse source
This commit is contained in:
A 2026-07-15 20:24:37 +08:00
parent 76bfc5100f
commit 47fcf0ea41
40 changed files with 1363 additions and 196 deletions
Download patch file Download diff file Expand all files Collapse all files

3
internal/rpc/channels_messages.go
View file

@ -262,6 +262,9 @@ func (r *Router) onChannelsGetMessages(ctx context.Context, req *tg.ChannelsGetM
if err != nil {
return nil, err
}
if err := r.checkFrozenChannelParticipants(ctx, userID, channelID); err != nil {
return nil, err
}
ids := make([]int, 0, len(req.ID))
for _, input := range req.ID {
id, ok := inputMessageBoxID(input)

2
internal/rpc/channels_stubs.go
View file

@ -655,7 +655,7 @@ func (r *Router) channelIDFromLegacyInputPeerChecked(ctx context.Context, userID
func channelInvalidErr(err error) error {
switch {
case errors.Is(err, domain.ErrUserSendRestricted):
case errors.Is(err, domain.ErrUserFrozen):
return frozenMethodInvalidErr()
case errors.Is(err, domain.ErrChannelTitleInvalid):
return tgerr400("CHAT_TITLE_EMPTY")

14
internal/rpc/channels_updates.go
View file

@ -18,15 +18,19 @@ func (r *Router) onUpdatesGetChannelDifference(ctx context.Context, req *tg.Upda
if err != nil {
return nil, internalErr()
}
// difference 类 catch-up FLOOD_WAIT（设计 Phase 2 / §10.3）：nudge 被消费后客户端会触发
// getChannelDifference，大群 nudge 全速前需限速防风暴。未配置阈值时不限速。
if err := r.checkCatchupRateLimit(ctx, userID, channelDifferenceRateLimitKeyPrefix); err != nil {
return nil, err
}
channelID, err := r.channelIDFromInput(ctx, userID, req.Channel)
if err != nil {
return nil, err
}
if err := r.checkFrozenChannelParticipants(ctx, userID, channelID); err != nil {
return nil, err
}
// difference 类 catch-up FLOOD_WAIT（设计 Phase 2 / §10.3）：nudge 被消费后客户端会触发
// getChannelDifference，大群 nudge 全速前需限速防风暴。未配置阈值时不限速。
// participant gate 必须先于限流写入，保证冻结拒绝没有副作用。
if err := r.checkCatchupRateLimit(ctx, userID, channelDifferenceRateLimitKeyPrefix); err != nil {
return nil, err
}
r.trackChannelInterest(ctx, userID, channelID)
diff, err := r.deps.Channels.GetDifference(ctx, userID, domain.ChannelDifferenceRequest{
UserID: userID,

9
internal/rpc/deps.go
View file

@ -382,10 +382,16 @@ type PrivacyService interface {
// HelpService 抽象启动配置与国家区号目录。
type HelpService interface {
GetAppConfig(ctx context.Context, hash int) (domain.AppConfig, bool, error)
GetAppConfig(ctx context.Context, userID int64, hash int) (domain.AppConfig, bool, error)
GetCountries(ctx context.Context, langCode string, hash int) (domain.CountriesList, bool, error)
}
// AccountFreezeService exposes the account-level read-only fact used by the
// central RPC mutation gate. It is domain-only and shared with app/help.
type AccountFreezeService interface {
AccountFreeze(ctx context.Context, userID int64) (domain.AccountFreeze, bool, error)
}
// UpdatesService 抽象 update 状态查询。
type UpdatesService interface {
GetState(ctx context.Context, authKeyID [8]byte, userID int64) (domain.UpdateState, error)
@ -794,6 +800,7 @@ type Deps struct {
Account AccountService
Privacy PrivacyService
Help HelpService
AccountFreeze AccountFreezeService
AICompose AIComposeService
Users UsersService
Updates UpdatesService

3
internal/rpc/errors.go
View file

@ -91,7 +91,8 @@ func documentInvalidErr() error { return tgerr.New(400, "DOCUMENT_INVALID")
func mediaEmptyErr() error { return tgerr.New(400, "MEDIA_EMPTY") }
func frozenMethodInvalidErr() error { return tgerr.New(400, "FROZEN_METHOD_INVALID") }
func frozenMethodInvalidErr() error { return tgerr.New(420, "FROZEN_METHOD_INVALID") }
func frozenParticipantMissingErr() error { return tgerr.New(400, "FROZEN_PARTICIPANT_MISSING") }
func photoInvalidErr() error { return tgerr.New(400, "PHOTO_INVALID") }

123
internal/rpc/frozen_gate.go Normal file
View file

@ -0,0 +1,123 @@
package rpc
import (
"context"
"strings"
"telesrv/internal/domain"
)
// Frozen accounts are read-only. Classifying the finite read vocabulary and
// failing closed for every other semantic method also covers future handlers:
// an unfamiliar mutation cannot silently bypass the account-level gate.
var frozenReadOnlyOperationPrefixes = []string{
"can", "check", "find", "get", "load", "lookup", "query", "read", "resolve", "search", "translate",
}
var frozenAlwaysBlockedMethods = map[string]struct{}{
"channels.deleteMessages": {},
"channels.joinChannel": {},
"channels.searchPosts": {},
}
// These methods are security/session housekeeping or read acknowledgements
// that must remain available in read-only mode. In particular, a frozen user
// must be able to log out/delete the account and clients must not enter retry
// loops for presence, push registration, or delivery/read acknowledgements.
var frozenAllowedMutationNamedMethods = map[string]struct{}{
"account.changeAuthorizationSettings": {},
"account.deleteAccount": {},
"account.registerDevice": {},
"account.resetAuthorization": {},
"account.resetAuthorizations": {},
"account.unregisterDevice": {},
"account.updateDeviceLocked": {},
"account.updateStatus": {},
"messages.readHistory": {},
"messages.readMentions": {},
"messages.readMessageContents": {},
"messages.readReactions": {},
"messages.receivedMessages": {},
"messages.receivedQueue": {},
"messages.reportMessagesDelivery": {},
"messages.viewSponsoredMessage": {},
"channels.readHistory": {},
"channels.readMessageContents": {},
"phone.receivedCall": {},
"stories.incrementStoryViews": {},
}
func frozenMethodRequiresWriteGate(method string) bool {
if _, blocked := frozenAlwaysBlockedMethods[method]; blocked {
return true
}
if _, allowed := frozenAllowedMutationNamedMethods[method]; allowed {
return false
}
if strings.HasPrefix(method, "auth.") {
return false
}
dot := strings.IndexByte(method, '.')
if dot < 0 || dot == len(method)-1 {
return false
}
operation := method[dot+1:]
for _, prefix := range frozenReadOnlyOperationPrefixes {
if strings.HasPrefix(operation, prefix) {
return false
}
}
return true
}
func (r *Router) checkFrozenRPC(ctx context.Context, method string) error {
if r == nil || r.deps.AccountFreeze == nil || !frozenMethodRequiresWriteGate(method) {
return nil
}
userID, authorized := UserIDFrom(ctx)
if !authorized || userID == 0 {
return nil
}
freeze, found, err := r.deps.AccountFreeze.AccountFreeze(ctx, userID)
if err != nil {
return internalErr()
}
if found && freeze.Frozen {
return frozenMethodInvalidErr()
}
return nil
}
// checkFrozenChannelParticipants implements Telegram's narrower read rule for
// the methods documented with FROZEN_PARTICIPANT_MISSING: an account freeze
// does not hide joined channels, but it removes public/linked guest preview.
func (r *Router) checkFrozenChannelParticipants(ctx context.Context, userID int64, channelIDs ...int64) error {
if r == nil || r.deps.AccountFreeze == nil || r.deps.Channels == nil || userID == 0 || len(channelIDs) == 0 {
return nil
}
freeze, found, err := r.deps.AccountFreeze.AccountFreeze(ctx, userID)
if err != nil {
return internalErr()
}
if !found || !freeze.Frozen {
return nil
}
seen := make(map[int64]struct{}, len(channelIDs))
for _, channelID := range channelIDs {
if channelID <= 0 {
continue
}
if _, duplicate := seen[channelID]; duplicate {
continue
}
seen[channelID] = struct{}{}
view, err := r.deps.Channels.ResolveChannel(ctx, userID, channelID)
if err != nil {
return channelInvalidErr(err)
}
if view.Self.UserID != userID || view.Self.Status != domain.ChannelMemberActive || view.Self.Guest {
return frozenParticipantMissingErr()
}
}
return nil
}

265
internal/rpc/frozen_gate_test.go Normal file
View file

@ -0,0 +1,265 @@
package rpc
import (
"context"
"errors"
"fmt"
"testing"
"time"
"github.com/iamxvbaba/td/bin"
"github.com/iamxvbaba/td/clock"
"github.com/iamxvbaba/td/tg"
"github.com/iamxvbaba/td/tgerr"
"go.uber.org/zap/zaptest"
"telesrv/internal/domain"
)
type frozenGateFreezeProvider struct {
freeze domain.AccountFreeze
found bool
err error
calls int
items map[int64]domain.AccountFreeze
}
func (p *frozenGateFreezeProvider) AccountFreeze(_ context.Context, userID int64) (domain.AccountFreeze, bool, error) {
p.calls++
if p.items != nil {
freeze, found := p.items[userID]
return freeze, found, p.err
}
return p.freeze, p.found, p.err
}
type frozenGateChannels struct {
ChannelsService
views map[int64]domain.ChannelView
err error
}
func (s frozenGateChannels) ResolveChannel(_ context.Context, _ int64, channelID int64) (domain.ChannelView, error) {
if s.err != nil {
return domain.ChannelView{}, s.err
}
view, ok := s.views[channelID]
if !ok {
return domain.ChannelView{}, domain.ErrChannelInvalid
}
return view, nil
}
func frozenGateActiveState(userID int64) domain.AccountFreeze {
since := time.Unix(1_700_000_000, 0).UTC()
return domain.AccountFreeze{
UserID: userID,
Frozen: true,
Since: since,
Until: since.Add(7 * 24 * time.Hour),
AppealURL: "https://example.test/appeal",
}
}
func TestFrozenMethodGateIsReadOnlyAndFailsClosed(t *testing.T) {
tests := map[string]bool{
"help.getAppConfig": false,
"messages.getHistory": false,
"messages.searchGlobal": false,
"contacts.resolveUsername": false,
"payments.checkCanSendGift": false,
"messages.readHistory": false,
"messages.readDiscussion": false,
"stats.loadAsyncGraph": false,
"stories.incrementStoryViews": false,
"account.updateDeviceLocked": false,
"account.updateStatus": false,
"account.deleteAccount": false,
"auth.logOut": false,
"messages.sendMessage": true,
"messages.editMessage": true,
"messages.deleteHistory": true,
"messages.forwardMessages": true,
"messages.sendReaction": true,
"channels.joinChannel": true,
"channels.searchPosts": true,
"contacts.importContacts": true,
"account.saveAutoDownloadSettings": true,
"future.performNewMutation": true,
}
for method, want := range tests {
t.Run(method, func(t *testing.T) {
if got := frozenMethodRequiresWriteGate(method); got != want {
t.Fatalf("frozenMethodRequiresWriteGate(%q) = %v, want %v", method, got, want)
}
})
}
}
func TestFrozenMethodGateIsUserScopedAcrossSessionsAndUnfreezesImmediately(t *testing.T) {
const (
frozenUser = int64(1001)
otherUser = int64(1002)
)
provider := &frozenGateFreezeProvider{items: map[int64]domain.AccountFreeze{
frozenUser: frozenGateActiveState(frozenUser),
}}
router := &Router{deps: Deps{AccountFreeze: provider}}
// Separate MTProto sessions for one user share the same durable account fact.
for _, sessionID := range []int64{11, 22} {
ctx := WithSessionID(WithUserID(context.Background(), frozenUser), sessionID)
if err := router.checkFrozenRPC(ctx, "messages.sendMessage"); !tgerr.Is(err, "FROZEN_METHOD_INVALID") {
t.Fatalf("session %d err = %v, want FROZEN_METHOD_INVALID", sessionID, err)
}
}
if err := router.checkFrozenRPC(WithUserID(context.Background(), otherUser), "messages.sendMessage"); err != nil {
t.Fatalf("other user was gated: %v", err)
}
// Unfreeze is a durable state transition; existing sessions are admitted on
// their very next RPC without reconnecting or retaining a per-session flag.
delete(provider.items, frozenUser)
for _, sessionID := range []int64{11, 22} {
ctx := WithSessionID(WithUserID(context.Background(), frozenUser), sessionID)
if err := router.checkFrozenRPC(ctx, "messages.sendMessage"); err != nil {
t.Fatalf("session %d remained gated after unfreeze: %v", sessionID, err)
}
}
}
func TestFrozenMethodGateReturns420BeforeLayerHandler(t *testing.T) {
const userID = int64(1001)
for _, profile := range []tg.LayerProfile{
tg.LayerProfile225,
tg.LayerProfile226,
tg.LayerProfile227,
tg.LayerProfile228,
} {
t.Run(fmt.Sprintf("layer_%d", profile), func(t *testing.T) {
provider := &frozenGateFreezeProvider{freeze: frozenGateActiveState(userID), found: true}
router := New(
Config{DC: 2, IP: "127.0.0.1", Port: 2398},
Deps{AccountFreeze: provider},
zaptest.NewLogger(t),
clock.System,
)
body := encodeExactLayerRPC(t, profile, &tg.MessagesSendMessageRequest{
Peer: &tg.InputPeerSelf{},
Message: "must not reach handler",
RandomID: 1,
})
admitted, err := router.AdmitLayer(profile, &body, tg.LayerDecodeLimits{})
if err != nil {
t.Fatal(err)
}
_, method, err := router.DispatchAdmitted(WithUserID(context.Background(), userID), [8]byte{1}, 10, 0, 1, admitted)
if method != "messages.sendMessage" || !tgerr.Is(err, "FROZEN_METHOD_INVALID") {
t.Fatalf("DispatchAdmitted = method:%q err:%v", method, err)
}
rpcErr, ok := tgerr.As(err)
if !ok || rpcErr.Code != 420 {
t.Fatalf("RPC error = %#v, want code 420", rpcErr)
}
if provider.calls != 1 {
t.Fatalf("freeze provider calls = %d, want 1", provider.calls)
}
})
}
}
func TestFrozenMethodGateReturns420BeforeLegacyHandler(t *testing.T) {
const userID = int64(1001)
provider := &frozenGateFreezeProvider{freeze: frozenGateActiveState(userID), found: true}
router := New(
Config{DC: 2, IP: "127.0.0.1", Port: 2398},
Deps{AccountFreeze: provider},
zaptest.NewLogger(t),
clock.System,
)
request := &tg.MessagesSendMessageRequest{
Peer: &tg.InputPeerSelf{},
Message: "must not reach legacy handler",
RandomID: 2,
}
var body bin.Buffer
if err := request.Encode(&body); err != nil {
t.Fatal(err)
}
if _, err := router.Dispatch(WithUserID(context.Background(), userID), [8]byte{1}, 10, &body); !tgerr.Is(err, "FROZEN_METHOD_INVALID") {
t.Fatalf("legacy Dispatch err = %v, want FROZEN_METHOD_INVALID", err)
} else if rpcErr, ok := tgerr.As(err); !ok || rpcErr.Code != 420 {
t.Fatalf("legacy RPC error = %#v, want code 420", rpcErr)
}
if provider.calls != 1 {
t.Fatalf("freeze provider calls = %d, want 1", provider.calls)
}
}
func TestFrozenParticipantGateAllowsOnlyJoinedChannels(t *testing.T) {
const userID = int64(1001)
provider := &frozenGateFreezeProvider{freeze: frozenGateActiveState(userID), found: true}
router := &Router{deps: Deps{
AccountFreeze: provider,
Channels: frozenGateChannels{views: map[int64]domain.ChannelView{
10: {Self: domain.ChannelMember{UserID: userID, Status: domain.ChannelMemberActive}},
20: {Self: domain.ChannelMember{UserID: userID, Status: domain.ChannelMemberLeft}},
30: {Self: domain.ChannelMember{UserID: userID, Status: domain.ChannelMemberActive, Guest: true}},
}},
}}
ctx := WithUserID(context.Background(), userID)
if err := router.checkFrozenChannelParticipants(ctx, userID, 10, 10); err != nil {
t.Fatalf("joined channel: %v", err)
}
for _, channelID := range []int64{20, 30} {
if err := router.checkFrozenChannelParticipants(ctx, userID, channelID); !tgerr.Is(err, "FROZEN_PARTICIPANT_MISSING") {
t.Fatalf("channel %d err = %v, want FROZEN_PARTICIPANT_MISSING", channelID, err)
} else if rpcErr, ok := tgerr.As(err); !ok || rpcErr.Code != 400 {
t.Fatalf("channel %d RPC error = %#v, want code 400", channelID, rpcErr)
}
}
}
func TestFrozenParticipantGateRejectsBeforeCatchupRateLimitWrite(t *testing.T) {
const (
userID = int64(1001)
channelID = int64(20)
accessHash = int64(20020)
)
provider := &frozenGateFreezeProvider{freeze: frozenGateActiveState(userID), found: true}
limiter := &captureRateLimiter{}
router := &Router{
cfg: Config{CatchupRateLimit: 1, CatchupRateWindow: time.Minute},
deps: Deps{
AccountFreeze: provider,
Limiter: limiter,
Channels: frozenGateChannels{views: map[int64]domain.ChannelView{
channelID: {
Channel: domain.Channel{ID: channelID, AccessHash: accessHash},
Self: domain.ChannelMember{UserID: userID, Status: domain.ChannelMemberLeft},
},
}},
},
}
_, err := router.onUpdatesGetChannelDifference(
WithUserID(context.Background(), userID),
&tg.UpdatesGetChannelDifferenceRequest{
Channel: &tg.InputChannel{ChannelID: channelID, AccessHash: accessHash},
Limit: 100,
},
)
if !tgerr.Is(err, "FROZEN_PARTICIPANT_MISSING") {
t.Fatalf("getChannelDifference err = %v, want FROZEN_PARTICIPANT_MISSING", err)
}
if len(limiter.calls) != 0 {
t.Fatalf("rejected frozen participant consumed rate-limit state: %+v", limiter.calls)
}
}
func TestFrozenGatesFailClosedOnFreezeLookupError(t *testing.T) {
provider := &frozenGateFreezeProvider{err: errors.New("database unavailable")}
router := &Router{deps: Deps{AccountFreeze: provider}}
if err := router.checkFrozenRPC(WithUserID(context.Background(), 1001), "messages.sendMessage"); !tgerr.Is(err, "INTERNAL_SERVER_ERROR") {
t.Fatalf("checkFrozenRPC error = %v, want INTERNAL_SERVER_ERROR", err)
}
}

3
internal/rpc/help.go
View file

@ -30,7 +30,8 @@ func (r *Router) registerHelp(d *tg.ServerDispatcher) {
if r.deps.Help == nil {
return tdesktop.AppConfig(hash), nil
}
cfg, notModified, err := r.deps.Help.GetAppConfig(ctx, hash)
userID, _ := UserIDFrom(ctx)
cfg, notModified, err := r.deps.Help.GetAppConfig(ctx, userID, hash)
if err != nil {
return nil, internalErr()
}

3
internal/rpc/layer_dispatch.go
View file

@ -254,6 +254,9 @@ func (r *Router) DispatchAdmitted(
return nil, method, authKeyUnregisteredErr()
}
}
if err := r.checkFrozenRPC(ctx, method); err != nil {
return nil, method, err
}
if profileKnown && profileEvidenceFresh {
r.maybeMarkSessionReceivesUpdates(ctx)
}

24
internal/rpc/messages_register.go
View file

@ -201,15 +201,26 @@ func (r *Router) registerMessages(d *tg.ServerDispatcher) {
if err != nil {
return nil, internalErr()
}
// difference 类 catch-up FLOOD_WAIT（设计 Phase 2 / §10.3）：DrKLO 收 nudge 对未加载频道
// 走 loadUnknownChannel→getPeerDialogs，限速须同时覆盖它（不止 getChannelDifference）。
if err := r.checkCatchupRateLimit(ctx, userID, peerDialogsRateLimitKeyPrefix); err != nil {
return nil, err
}
domainPeers, err := r.dialogPeersFromInput(ctx, userID, peers)
if err != nil {
return nil, err
}
channelIDs := make([]int64, 0, len(domainPeers))
for _, peer := range domainPeers {
if peer.Type == domain.PeerTypeChannel {
channelIDs = append(channelIDs, peer.ID)
}
}
if err := r.checkFrozenChannelParticipants(ctx, userID, channelIDs...); err != nil {
return nil, err
}
// difference 类 catch-up FLOOD_WAIT（设计 Phase 2 / §10.3）：DrKLO 收 nudge 对未加载频道
// 走 loadUnknownChannel→getPeerDialogs，限速须同时覆盖它（不止 getChannelDifference）。
// 冻结账号的 guest/non-member 必须先返回 FROZEN_PARTICIPANT_MISSING，拒绝路径不能
// 消耗限流额度或产生其它可变状态。
if err := r.checkCatchupRateLimit(ctx, userID, peerDialogsRateLimitKeyPrefix); err != nil {
return nil, err
}
var list domain.DialogList
if len(domainPeers) > 0 && r.deps.Dialogs != nil {
var err error
@ -259,6 +270,9 @@ func (r *Router) registerMessages(d *tg.ServerDispatcher) {
if err := r.validateInputPeerChannelAccess(ctx, userID, req.Peer, filter.Peer.ID); err != nil {
return nil, err
}
if err := r.checkFrozenChannelParticipants(ctx, userID, filter.Peer.ID); err != nil {
return nil, err
}
if isLegacyInputPeerChat(req.Peer) {
return &tg.MessagesMessages{}, nil
}

2
internal/rpc/messages_send.go
View file

@ -202,7 +202,7 @@ func (r *Router) onMessagesSendMessage(ctx context.Context, req *tg.MessagesSend
func messageSendErr(err error) error {
switch {
case errors.Is(err, domain.ErrUserSendRestricted):
case errors.Is(err, domain.ErrUserFrozen):
return frozenMethodInvalidErr()
case errors.Is(err, domain.ErrReplyMessageIDInvalid):
return replyMessageIDInvalidErr()

3
internal/rpc/router.go
View file

@ -670,6 +670,9 @@ func (r *Router) dispatch(ctx context.Context, b *bin.Buffer, depth int, meta *r
if err := preflightRPCRequest(id, b); err != nil {
return nil, err
}
if err := r.checkFrozenRPC(ctx, tlTypeName(id)); err != nil {
return nil, err
}
// 任何未包 invokeWithoutUpdates 的已登录 RPC 都把当前 session 视为 updates
// 接收者。仅靠 updates.getState/getDifference 置位会漏掉 DrKLO 热恢复：
// 它重连后不重建同步基线（pts 在进程内存里），只发普通业务请求，置位