diff --git a/internal/auth/oauth_helpers.go b/internal/auth/oauth_helpers.go index 9da470a0..61e91ceb 100644 --- a/internal/auth/oauth_helpers.go +++ b/internal/auth/oauth_helpers.go @@ -237,16 +237,22 @@ func (p *OAuthProvider) postJSON(ctx context.Context, endpoint string, body any) } defer resp.Body.Close() - data, err := io.ReadAll(io.LimitReader(resp.Body, config.MaxResponseBodySize)) - if err != nil { - return nil, fmt.Errorf("reading response: %w", err) - } + data, readErr := io.ReadAll(io.LimitReader(resp.Body, config.MaxResponseBodySize)) if resp.StatusCode != http.StatusOK { + // Preserve structured HTTP status semantics even when the response + // body is truncated. The body is diagnostic-only here, so read it + // best-effort and classify retryability from the status code. + if readErr != nil { + data = nil + } return nil, &HTTPStatusError{ StatusCode: resp.StatusCode, responseBody: truncateBody(data, 200), } } + if readErr != nil { + return nil, fmt.Errorf("reading response: %w", readErr) + } return data, nil } diff --git a/internal/auth/oauth_postjson_test.go b/internal/auth/oauth_postjson_test.go new file mode 100644 index 00000000..78446fc2 --- /dev/null +++ b/internal/auth/oauth_postjson_test.go @@ -0,0 +1,69 @@ +// Copyright 2026 Alibaba Group +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package auth + +import ( + "context" + "errors" + "io" + "net/http" + "testing" +) + +type postJSONRoundTripFunc func(*http.Request) (*http.Response, error) + +func (f postJSONRoundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) { + return f(req) +} + +type oauthBrokenBody struct{} + +func (oauthBrokenBody) Read([]byte) (int, error) { return 0, io.ErrUnexpectedEOF } +func (oauthBrokenBody) Close() error { return nil } + +func TestCrossPlatformCoveragePostJSONTruncatedErrorBodyKeepsHTTPStatus(t *testing.T) { + for _, status := range []int{http.StatusTooManyRequests, http.StatusServiceUnavailable} { + t.Run(http.StatusText(status), func(t *testing.T) { + provider := &OAuthProvider{httpClient: &http.Client{Transport: postJSONRoundTripFunc(func(*http.Request) (*http.Response, error) { + return &http.Response{StatusCode: status, Body: oauthBrokenBody{}, Header: make(http.Header)}, nil + })}} + + _, err := provider.postJSON(context.Background(), "https://oauth.test/token", map[string]string{"grantType": "refresh_token"}) + var statusErr *HTTPStatusError + if !errors.As(err, &statusErr) || statusErr.StatusCode != status { + t.Fatalf("postJSON() error = %v, want HTTPStatusError %d", err, status) + } + if errors.Is(err, io.ErrUnexpectedEOF) { + t.Fatalf("HTTP status error should not expose diagnostic body read failure: %v", err) + } + if got := ClassifyRefreshFailure(err); got != RefreshFailureTransient { + t.Fatalf("ClassifyRefreshFailure() = %s, want transient", got) + } + }) + } +} + +func TestCrossPlatformCoveragePostJSONOKTruncatedBodyIsTransient(t *testing.T) { + provider := &OAuthProvider{httpClient: &http.Client{Transport: postJSONRoundTripFunc(func(*http.Request) (*http.Response, error) { + return &http.Response{StatusCode: http.StatusOK, Body: oauthBrokenBody{}, Header: make(http.Header)}, nil + })}} + + _, err := provider.postJSON(context.Background(), "https://oauth.test/token", map[string]string{"grantType": "refresh_token"}) + if !errors.Is(err, io.ErrUnexpectedEOF) { + t.Fatalf("postJSON() error = %v, want io.ErrUnexpectedEOF", err) + } + if got := ClassifyRefreshFailure(err); got != RefreshFailureTransient { + t.Fatalf("ClassifyRefreshFailure() = %s, want transient", got) + } +} diff --git a/internal/auth/refresh_failure.go b/internal/auth/refresh_failure.go index 0c553607..1e841894 100644 --- a/internal/auth/refresh_failure.go +++ b/internal/auth/refresh_failure.go @@ -17,6 +17,7 @@ import ( "context" "errors" "fmt" + "io" "net" "net/http" ) @@ -63,6 +64,9 @@ func ClassifyRefreshFailure(err error) RefreshFailureClass { if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { return RefreshFailureTransient } + if errors.Is(err, io.ErrUnexpectedEOF) { + return RefreshFailureTransient + } var netErr net.Error if errors.As(err, &netErr) { return RefreshFailureTransient diff --git a/internal/event/source/portal_ticket.go b/internal/event/source/portal_ticket.go index 75f5ca3f..148bb76d 100644 --- a/internal/event/source/portal_ticket.go +++ b/internal/event/source/portal_ticket.go @@ -308,8 +308,11 @@ func requestPortalTicketAttempt(ctx context.Context, cfg *PortalTicketConfig, ht return portalStreamTicket{}, 0, &portalStageError{stage: "ticket_request", retryable: true, cause: fmt.Errorf("source: portal ticket request: %w", err)} } defer resp.Body.Close() - raw, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) - if resp.StatusCode >= 400 { + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + // Preserve HTTP status semantics before touching the body. A truncated + // 401 body must not become a retryable read error that bypasses the + // single token-refresh guard; the body is only best-effort diagnostics. + raw, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) httpErr := fmt.Errorf("source: portal ticket HTTP %d: %s", resp.StatusCode, truncatePortalTicketLog(string(raw), 300)) if retryableTicketStatus(resp.StatusCode) { @@ -317,6 +320,15 @@ func requestPortalTicketAttempt(ctx context.Context, cfg *PortalTicketConfig, ht } return portalStreamTicket{}, resp.StatusCode, httpErr } + raw, readErr := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + if readErr != nil { + return portalStreamTicket{}, resp.StatusCode, &portalStageError{ + stage: "ticket_request", + status: resp.StatusCode, + retryable: true, + cause: fmt.Errorf("source: portal ticket read: %w", readErr), + } + } var direct portalStreamTicket if err := json.Unmarshal(raw, &direct); err == nil && direct.Endpoint != "" && direct.Ticket != "" { diff --git a/internal/event/source/portal_ticket_refresh_test.go b/internal/event/source/portal_ticket_refresh_test.go index 319c7504..374755cf 100644 --- a/internal/event/source/portal_ticket_refresh_test.go +++ b/internal/event/source/portal_ticket_refresh_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "io" "net/http" "net/http/httptest" @@ -322,6 +323,60 @@ type brokenBody struct{} func (brokenBody) Read([]byte) (int, error) { return 0, io.ErrUnexpectedEOF } func (brokenBody) Close() error { return nil } +func TestCrossPlatformCoveragePortalTicketNon2xxTruncatedBodyKeepsStatus(t *testing.T) { + for _, tc := range []struct { + status int + retryable bool + }{ + {status: http.StatusUnauthorized}, + {status: http.StatusTooManyRequests, retryable: true}, + {status: http.StatusServiceUnavailable, retryable: true}, + } { + t.Run(http.StatusText(tc.status), func(t *testing.T) { + client := &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) { + return &http.Response{StatusCode: tc.status, Body: brokenBody{}, Header: make(http.Header)}, nil + })} + + _, status, err := requestPortalTicketAttempt(context.Background(), &PortalTicketConfig{ + TicketURL: "https://ticket.test", + SourceID: "source", + }, client, "token") + if status != tc.status || err == nil || !strings.Contains(err.Error(), fmt.Sprintf("HTTP %d", tc.status)) { + t.Fatalf("requestPortalTicketAttempt() status=%d err=%v, want HTTP %d", status, err, tc.status) + } + if errors.Is(err, io.ErrUnexpectedEOF) { + t.Fatalf("non-2xx status error should not expose diagnostic body read failure: %v", err) + } + var stageErr *portalStageError + if tc.retryable { + if !errors.As(err, &stageErr) || stageErr.status != tc.status || !stageErr.retryable { + t.Fatalf("retryable status should return retryable portalStageError, got %T %v", err, err) + } + } else if errors.As(err, &stageErr) && stageErr.retryable { + t.Fatalf("fatal status should not become retryable stage error: %v", err) + } + }) + } +} + +func TestCrossPlatformCoveragePortalTicket200TruncatedBodyIsRetryable(t *testing.T) { + client := &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) { + return &http.Response{StatusCode: http.StatusOK, Body: brokenBody{}, Header: make(http.Header)}, nil + })} + + _, status, err := requestPortalTicketAttempt(context.Background(), &PortalTicketConfig{ + TicketURL: "https://ticket.test", + SourceID: "source", + }, client, "token") + if status != http.StatusOK || !errors.Is(err, io.ErrUnexpectedEOF) { + t.Fatalf("requestPortalTicketAttempt() status=%d err=%v, want 200 + io.ErrUnexpectedEOF", status, err) + } + var stageErr *portalStageError + if !errors.As(err, &stageErr) || !stageErr.retryable || stageErr.stage != "ticket_request" { + t.Fatalf("2xx truncated body should be retryable ticket_request stage, got %T %v", err, err) + } +} + // TestCrossPlatformCoveragePersonalFetchTicket401TruncatedBodyStaysFatal guards the single // refresh-retry protection: a 401 whose body fails with unexpected EOF must // be classified by status (fatal) and never wrapped as retryable, otherwise