diff --git a/internal/mtprotoedge/conn.go b/internal/mtprotoedge/conn.go index 87c7aecc..56087b39 100644 --- a/internal/mtprotoedge/conn.go +++ b/internal/mtprotoedge/conn.go @@ -142,6 +142,11 @@ type Conn struct { // Rewrap aliasing never delays execution. initialized stops collecting // candidates after the first valid init wrapper on this physical generation. rpcRewrapInitialized atomic.Bool + // layerRPCAdmissionTraceLogged bounds production INFO diagnostics to the + // first non-unknown exact-admission rejection on this physical generation. + // Repeated malformed requests remain visible at Debug without turning an + // authenticated reconnect/session into an unbounded INFO log source. + layerRPCAdmissionTraceLogged atomic.Bool // rpcResultAcked is invoked by the sole outbound actor after it resolves an // acknowledged server frame back to the rpc_result request msg_id. rpcResultAcked func(*Conn, int64) diff --git a/internal/mtprotoedge/inbound_layer_rpc.go b/internal/mtprotoedge/inbound_layer_rpc.go index 8ab2e468..1e9a6dcd 100644 --- a/internal/mtprotoedge/inbound_layer_rpc.go +++ b/internal/mtprotoedge/inbound_layer_rpc.go @@ -401,16 +401,11 @@ func (s *Server) prepareInboundLayerRPCBatch(ctx context.Context, c *Conn, plan if errors.Is(err, ErrLayerProfileConflict) { return err } - s.log.Debug("RPC exact admission rejected", - zap.String("method", method), - zap.String("auth_key_id", c.authKeyHex), - zap.Int64("session_id", c.sessionID), - zap.Int64("msg_id", item.msgID), - zap.Error(err), - ) + rpcError := layerRPCAdmissionError(err) + s.logLayerRPCAdmissionRejection(c, item, itemState, admissionCursor, method, rpcError, err) item.kind = inboundItemRPCAdmissionError item.method = method - item.payload = layerRPCAdmissionError(err) + item.payload = rpcError c.metrics.InboundRPCDropped(method, "layer_admission") continue } @@ -1224,6 +1219,63 @@ func layerRPCAdmissionError(err error) *mt.RPCError { return &mt.RPCError{ErrorCode: 400, ErrorMessage: "INPUT_REQUEST_INVALID"} } +// logLayerRPCAdmissionRejection preserves one production-visible diagnostic for +// an otherwise generic RPC 400 without exposing the request body. The INFO path +// is one-shot per physical Conn; compatibility-traced unknown RPCs already have +// their own warning and therefore do not consume this diagnostic slot. +func (s *Server) logLayerRPCAdmissionRejection( + c *Conn, + item *inboundItem, + state LayerProfileSnapshot, + cursor layerRPCAdmissionCursor, + method string, + rpcError *mt.RPCError, + admissionErr error, +) { + if s == nil || s.log == nil || c == nil || item == nil || rpcError == nil { + return + } + wireID := item.typeID + if wireID == 0 { + if id, err := (&bin.Buffer{Buf: item.body}).PeekID(); err == nil { + wireID = id + } + } + fields := []zap.Field{ + zap.String("method", method), + zap.String("auth_key_id", c.authKeyHex), + zap.Int64("session_id", c.sessionID), + zap.Int64("msg_id", item.msgID), + zap.Uint32("top_level_wire_id", wireID), + zap.Int("wire_bytes", len(item.body)), + zap.Int("selected_profile", int(state.Profile)), + zap.String("profile_origin", layerProfileOriginLogName(state.Origin)), + zap.Uint32("profile_epoch", state.Epoch), + zap.Int("raw_layer_evidence", cursor.rawLayer), + zap.Int64("layer_evidence_msg_id", cursor.evidenceMsgID), + zap.Bool("explicit_layer_selector", layerRPCAdmissionHasExplicitSelector(item.body, admissionErr)), + zap.Int("rpc_error_code", rpcError.ErrorCode), + zap.String("rpc_error_message", rpcError.ErrorMessage), + zap.Error(admissionErr), + } + if !errors.Is(admissionErr, tlprofile.ErrUnknownRPCMethod) && c.layerRPCAdmissionTraceLogged.CompareAndSwap(false, true) { + s.log.Info("RPC exact admission rejected", fields...) + return + } + s.log.Debug("RPC exact admission rejected", fields...) +} + +func layerProfileOriginLogName(origin LayerProfileOrigin) string { + switch origin { + case LayerProfileInherited: + return "inherited" + case LayerProfileExplicit: + return "explicit" + default: + return "unknown" + } +} + func (s *Server) layerRPCDependencies(c *Conn, msgID int64, request tlprofile.Admission) layerRPCDependencySet { result := layerRPCDependencySet{} seen := make(map[int64]struct{}) diff --git a/internal/mtprotoedge/layer_admission_budget_test.go b/internal/mtprotoedge/layer_admission_budget_test.go index 20489819..8f14b267 100644 --- a/internal/mtprotoedge/layer_admission_budget_test.go +++ b/internal/mtprotoedge/layer_admission_budget_test.go @@ -13,7 +13,9 @@ import ( "github.com/iamxvbaba/td/proto" "github.com/iamxvbaba/td/tg" "github.com/iamxvbaba/td/tgerr" + "go.uber.org/zap" "go.uber.org/zap/zaptest" + "go.uber.org/zap/zaptest/observer" "github.com/iamxvbaba/td/tlprofile" appfiles "telesrv/internal/app/files" @@ -218,10 +220,10 @@ func TestLayerRPCAdmissionAdmitsTDLibUploadParts(t *testing.T) { bare bool }{ { - name: "small_file_part", + name: "pixel_9a_small_file_part_negative_file_id", method: "upload.saveFilePart", body: &tg.UploadSaveFilePartRequest{ - FileID: 91, + FileID: -3596058967254453060, FilePart: 0, Bytes: make([]byte, 1071), }, @@ -324,6 +326,86 @@ func TestLayerRPCAdmissionAdmitsTDLibUploadParts(t *testing.T) { } } +func TestLayerRPCAdmissionRejectionLogsBoundedMetadataWithoutUploadBody(t *testing.T) { + const marker = "DO_NOT_LOG_UPLOAD_BODY_MARKER" + payload := make([]byte, 1071) + copy(payload, marker) + body := tdlibWrappedBody(t, tlprofile.Profile228, &tg.UploadSaveFilePartRequest{ + FileID: -3596058967254453060, + FilePart: 0, + Bytes: payload, + }) + // Remove the final TL padding byte. Exact admission must reject the malformed + // wire while the edge still records its explicit wrapper and bounded cause. + body = body[:len(body)-1] + + core, logs := observer.New(zap.DebugLevel) + router := rpc.New(rpc.Config{DC: 2}, rpc.Deps{}, zap.New(core), clock.System) + s := New(Options{DC: 2, LayerRPC: router, Logger: zap.New(core)}) + c := &Conn{ + authKeyID: [8]byte{8, 34}, + authKeyHex: "0822000000000000", + sessionID: 834, + metrics: NopMetrics{}, + } + c.startInboundRPCScheduler(s.rpcScheduler, 1, 2, time.Second) + defer func() { + c.closeInboundRPCScheduler() + s.rpcScheduler.stop(time.Second) + }() + + for attempt := 0; attempt < 2; attempt++ { + plan := &inboundPlan{items: []inboundItem{{ + kind: inboundItemRPC, + msgID: int64(100 + attempt*4), + typeID: tg.InvokeWithLayerRequestTypeID, + body: body, + }}} + if err := s.prepareInboundLayerRPCBatch(context.Background(), c, plan); err != nil { + plan.close() + t.Fatal(err) + } + if item := plan.items[0]; item.kind != inboundItemRPCAdmissionError { + plan.close() + t.Fatalf("malformed upload attempt %d kind = %d, want admission error", attempt, item.kind) + } + plan.close() + } + + entries := logs.FilterMessage("RPC exact admission rejected").All() + if len(entries) != 2 { + t.Fatalf("admission rejection log count = %d, want 2", len(entries)) + } + if entries[0].Level != zap.InfoLevel || entries[1].Level != zap.DebugLevel { + t.Fatalf("admission rejection levels = %s/%s, want info/debug", entries[0].Level, entries[1].Level) + } + fields := entries[0].ContextMap() + for key, want := range map[string]any{ + "method": "invokeWithLayer#da9b0d0d", + "auth_key_id": "0822000000000000", + "session_id": int64(834), + "msg_id": int64(100), + "top_level_wire_id": uint32(tg.InvokeWithLayerRequestTypeID), + "wire_bytes": int64(len(body)), + "profile_origin": "unknown", + "explicit_layer_selector": true, + "rpc_error_code": int64(400), + "rpc_error_message": "INPUT_REQUEST_INVALID", + } { + if got := fields[key]; got != want { + t.Fatalf("admission rejection field %q = %#v (%T), want %#v (%T); all=%#v", key, got, got, want, want, fields) + } + } + for _, entry := range entries { + if strings.Contains(entry.Message, marker) || strings.Contains(entry.ContextMap()["error"].(string), marker) { + t.Fatalf("admission rejection leaked upload body marker: %#v", entry.ContextMap()) + } + if _, ok := entry.ContextMap()["body"]; ok { + t.Fatalf("admission rejection exposed body field: %#v", entry.ContextMap()) + } + } +} + func TestLayerRPCAdmissionAdmitsTDLibFirstUploadContainer(t *testing.T) { router := rpc.New(rpc.Config{DC: 2}, rpc.Deps{}, zaptest.NewLogger(t), clock.System) s := New(Options{DC: 2, LayerRPC: router, Logger: zaptest.NewLogger(t)})