diff --git a/internal/teams/server.go b/internal/teams/server.go index 5e00432..587abbb 100644 --- a/internal/teams/server.go +++ b/internal/teams/server.go @@ -30,6 +30,10 @@ const ( // inbound Activities cannot spawn unbounded work. The handler acquires a slot // before acking and sheds load with 503 (the platform retries) when full. maxConcurrentDispatch = 256 + // maxConcurrentReads bounds concurrent inbound request processing (the body + // read + JSON unmarshal before JWT verification), capping peak pre-auth read + // memory at maxConcurrentReads times maxRequestBytes. Mirrors github/gitlab. + maxConcurrentReads = 16 ) // conversation holds what a reply needs: the serviceUrl to POST to and the bot's @@ -103,17 +107,37 @@ func serve(srv *http.Server, ln net.Listener, done func(error)) { } func (a *adapter) handleMessages(dispatchCtx context.Context, w http.ResponseWriter, r *http.Request, deps core.AdapterDeps) { + // Bound concurrent inbound reads before buffering + unmarshaling the body: a + // flood of large bodies is shed with 503 (the platform retries) rather than + // buffering unbounded memory. readSem is allocated once in newAdapter, so this + // direct read races nothing. Separate from dispatchSem, which bounds post-ack + // dispatch. + select { + case a.readSem <- struct{}{}: + default: + a.log().Warn("teams: inbound concurrency limit reached; shedding with 503", "limit", maxConcurrentReads) + w.WriteHeader(http.StatusServiceUnavailable) + return + } + body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxRequestBytes)) if err != nil { + <-a.readSem w.WriteHeader(http.StatusBadRequest) return } var act inboundActivity if err := json.Unmarshal(body, &act); err != nil { + <-a.readSem w.WriteHeader(http.StatusBadRequest) return } + // Release the read slot as soon as the body is buffered + unmarshaled: unlike + // the GitHub/GitLab siblings, validateInbound below can block on a cold JWKS + // fetch (up to jwksFetchTimeout), so holding readSem across it would gate + // throughput far more aggressively than the fast HMAC path readSem is sized for. + <-a.readSem // Authenticate before trusting the body. Use the request context, not runCtx: // core cancels runCtx before srv.Shutdown drains this handler, so a JWKS diff --git a/internal/teams/server_test.go b/internal/teams/server_test.go index 35e4eeb..eecd6d4 100644 --- a/internal/teams/server_test.go +++ b/internal/teams/server_test.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "errors" + "io" "log/slog" "net" "net/http" @@ -269,6 +270,32 @@ func TestHandleMessages_SaturatedDispatchReturns503(t *testing.T) { asserts.Equal(t, w2.Code, http.StatusServiceUnavailable, "saturated dispatch sheds load with 503") } +// readSpy counts Read calls so a test can prove a shed request's body was +// never touched. +type readSpy struct{ reads int } + +func (s *readSpy) Read([]byte) (int, error) { s.reads++; return 0, io.EOF } + +// A saturated read semaphore must answer 503 before the body is buffered and +// unmarshaled, so a flood of large POSTs cannot exhaust memory ahead of the JWT +// check. Mirrors the github/gitlab sibling test. +func TestHandleMessages_ReadSaturationReturns503(t *testing.T) { + a := testAdapter(t) + a.readSem = make(chan struct{}, 1) + a.readSem <- struct{}{} // occupy the only inbound slot + + var got []*core.Message + spy := &readSpy{} + r := httptest.NewRequest(http.MethodPost, a.cfg.Path, spy) + r.Header.Set("Authorization", "Bearer "+mintToken(t, testKID, validClaims(a.cfg.AppID, allowedServiceURL))) + w := httptest.NewRecorder() + a.handleMessages(context.Background(), w, r, captureDeps(&got, nil)) + + asserts.Equal(t, w.Code, http.StatusServiceUnavailable, "saturated read gate returns 503") + asserts.Equal(t, len(got), 0, "nothing dispatched") + asserts.Equal(t, spy.reads, 0, "shed request's body is never read") +} + func TestHandleMessages_IgnoresNonMessage(t *testing.T) { a := testAdapter(t) var got []*core.Message diff --git a/internal/teams/teams.go b/internal/teams/teams.go index e55c6b2..688fb4d 100644 --- a/internal/teams/teams.go +++ b/internal/teams/teams.go @@ -88,10 +88,15 @@ type adapter struct { // captured at acquire, so a slot a context-ignoring handler never releases is // confined to that connection rather than leaking for the adapter's lifetime. dispatchSem chan struct{} - convs map[string]conversation // conversationID -> reply routing info - convOrder []string // FIFO insertion order for bounded eviction - token cachedToken - keys map[string]*jwksKey // kid -> signing key + channel endorsements + // readSem bounds concurrent inbound request processing so a flood of large + // bodies can't buffer unbounded memory before JWT verification (see + // maxConcurrentReads); separate from dispatchSem, which bounds dispatch + // goroutines. Allocated once in newAdapter and never reassigned. + readSem chan struct{} + convs map[string]conversation // conversationID -> reply routing info + convOrder []string // FIFO insertion order for bounded eviction + token cachedToken + keys map[string]*jwksKey // kid -> signing key + channel endorsements // keysAt is the last JWKS fetch attempt (rate-limits refreshes); keysFreshAt // is the last successful refresh (drives the jwksMaxAge staleness gate). keysAt time.Time @@ -157,5 +162,6 @@ func newAdapter(cfg Config) (*adapter, error) { tokenURL: "https://login.microsoftonline.com/" + tenant + "/oauth2/v2.0/token", openIDURL: openIDConfigURL, dispatchSem: make(chan struct{}, maxConcurrentDispatch), + readSem: make(chan struct{}, maxConcurrentReads), }, nil } diff --git a/internal/whatsapp/cloud/cloud.go b/internal/whatsapp/cloud/cloud.go index 58292f3..d6340ae 100644 --- a/internal/whatsapp/cloud/cloud.go +++ b/internal/whatsapp/cloud/cloud.go @@ -26,6 +26,7 @@ import ( "log/slog" "net" "net/http" + "net/url" "strconv" "strings" "sync" @@ -52,6 +53,11 @@ const ( // goroutines. maxConcurrentDispatch = 256 + // maxConcurrentReads bounds concurrent inbound request processing (the body + // read before signature verification), capping peak pre-auth read memory at + // maxConcurrentReads times maxRequestBytes. Mirrors the github/gitlab siblings. + maxConcurrentReads = 16 + // maxErrorBodyBytes caps how much of a non-2xx response body is read into errors. maxErrorBodyBytes = 4 << 10 // 4 KiB @@ -148,6 +154,11 @@ type adapter struct { // handler never releases is confined to that connection rather than leaking for // the adapter's lifetime. dispatchSem chan struct{} + // readSem bounds concurrent inbound request processing so a flood of large + // bodies can't buffer unbounded memory before signature verification (see + // maxConcurrentReads); separate from dispatchSem, which bounds dispatch + // goroutines. Allocated once in newAdapter and never reassigned. + readSem chan struct{} } // log returns the Bot's logger handed over at Connect, or slog.Default() @@ -206,7 +217,7 @@ func newAdapter(cfg Config) (*adapter, error) { if cfg.HTTPClient == nil { cfg.HTTPClient = &http.Client{Timeout: 30 * time.Second} } - return &adapter{cfg: cfg, baseURL: graphBaseURL, http: cfg.HTTPClient, dispatchSem: make(chan struct{}, maxConcurrentDispatch)}, nil + return &adapter{cfg: cfg, baseURL: graphBaseURL, http: cfg.HTTPClient, dispatchSem: make(chan struct{}, maxConcurrentDispatch), readSem: make(chan struct{}, maxConcurrentReads)}, nil } func (a *adapter) Connect(ctx context.Context, deps core.AdapterDeps) error { @@ -288,6 +299,19 @@ func (a *adapter) handleVerify(w http.ResponseWriter, r *http.Request) { } func (a *adapter) handleWebhook(dispatchCtx context.Context, w http.ResponseWriter, r *http.Request, deps core.AdapterDeps) { + // Bound concurrent inbound reads before buffering the body: a flood of large + // bodies is shed with 503 (Meta retries) rather than buffering unbounded + // memory. readSem is allocated once in newAdapter, so this direct read races + // nothing. Separate from dispatchSem, which bounds post-ack dispatch. + select { + case a.readSem <- struct{}{}: + default: + a.log().Warn("whatsapp: inbound concurrency limit reached; shedding with 503", "limit", maxConcurrentReads) + w.WriteHeader(http.StatusServiceUnavailable) + return + } + defer func() { <-a.readSem }() + body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxRequestBytes)) if err != nil { w.WriteHeader(http.StatusBadRequest) @@ -500,8 +524,10 @@ func (a *adapter) ResolveAttachmentURL(ctx context.Context, att core.Attachment) return "", nil } - url := fmt.Sprintf("%s/%s/%s", a.baseURL, a.cfg.GraphVersion, media.ID) - req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + // PathEscape the wire-derived media ID so it cannot alter the request path; + // it stays confined to the pinned graph.facebook.com host either way. + endpoint := fmt.Sprintf("%s/%s/%s", a.baseURL, a.cfg.GraphVersion, url.PathEscape(media.ID)) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) if err != nil { return "", err } diff --git a/internal/whatsapp/cloud/cloud_test.go b/internal/whatsapp/cloud/cloud_test.go index b1b5315..d86c380 100644 --- a/internal/whatsapp/cloud/cloud_test.go +++ b/internal/whatsapp/cloud/cloud_test.go @@ -38,7 +38,7 @@ func validConfig() Config { func testAdapter() *adapter { cfg := validConfig() cfg.Path = defaultPath - return &adapter{cfg: cfg, baseURL: graphBaseURL, http: http.DefaultClient, dispatchSem: make(chan struct{}, maxConcurrentDispatch)} + return &adapter{cfg: cfg, baseURL: graphBaseURL, http: http.DefaultClient, dispatchSem: make(chan struct{}, maxConcurrentDispatch), readSem: make(chan struct{}, maxConcurrentReads)} } // sign returns the X-Hub-Signature-256 header value for body under secret. @@ -484,6 +484,32 @@ func TestHandleWebhook_SaturationReturns503(t *testing.T) { a.drainDispatch(ctx) } +// readSpy counts Read calls so a test can prove a shed request's body was +// never touched. +type readSpy struct{ reads int } + +func (s *readSpy) Read([]byte) (int, error) { s.reads++; return 0, io.EOF } + +// A saturated read semaphore must answer 503 before the (up to maxRequestBytes) +// body is buffered, so a flood of large POSTs cannot exhaust memory ahead of the +// signature check. Mirrors the github/gitlab sibling test. +func TestHandleWebhook_ReadSaturationReturns503(t *testing.T) { + a := testAdapter() + a.readSem = make(chan struct{}, 1) + a.readSem <- struct{}{} // occupy the only inbound slot + + var got []*core.Message + spy := &readSpy{} + r := httptest.NewRequest(http.MethodPost, "/webhook", spy) + r.Header.Set(signatureHeader, sign(a.cfg.AppSecret, []byte(textWebhook))) + w := httptest.NewRecorder() + a.handleWebhook(context.Background(), w, r, captureDeps(&got, nil)) + + asserts.Equal(t, w.Code, http.StatusServiceUnavailable, "saturated read gate returns 503") + asserts.Equal(t, len(got), 0, "nothing dispatched") + asserts.Equal(t, spy.reads, 0, "shed request's body is never read") +} + func TestHandleWebhook_StatusOnlyIgnored(t *testing.T) { a := testAdapter() var got []*core.Message @@ -713,6 +739,29 @@ func TestResolveAttachmentURL(t *testing.T) { asserts.Equal(t, gotAuth, "Bearer tok", "resolve should set the bearer token") } +// A wire-derived media ID with URL-significant characters must be PathEscaped so +// it cannot alter the request path (defense-in-depth; it can't escape the pinned +// host regardless). +func TestResolveAttachmentURL_EscapesMediaID(t *testing.T) { + var gotPath string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotPath = r.URL.EscapedPath() + _, _ = io.WriteString(w, `{"url":"https://lookaside.fbsbx.com/media/blob"}`) + })) + defer srv.Close() + + a := testAdapter() + a.baseURL = srv.URL + a.http = srv.Client() + a.cfg.GraphVersion = "v23.0" + + _, err := a.ResolveAttachmentURL(context.Background(), + core.Attachment{ExtraData: &Media{ID: "a/b?c"}}) + + asserts.NoError(t, err, "resolve should succeed") + asserts.Equal(t, gotPath, "/v23.0/a%2Fb%3Fc", "media id is PathEscaped in the request path") +} + func TestResolveAttachmentURL_NoMediaID(t *testing.T) { a := testAdapter() a.http = nil // a no-op resolve must short-circuit before any HTTP call