fix(observability-proxy): bound the token and access reviews

Both reviews ran on the request context, which net/http cancels for a
dropped client only once the body is read. A POST whose body was still
unread therefore waited on a stalled kube-apiserver past every request
limit. The reviews now share the request timeout and answer 503 when it
passes.

Assisted-by: LLM
Signed-off-by: ZverGuy <maximbel2003@gmail.com>

internal/observabilityproxy/server.go

@@ -331,7 +331,16 @@ func (s *Server) authorize(r *http.Request, route Route) (string, *HTTPError) {
 	if !strings.EqualFold(scheme, "Bearer") {
 		return "", errorf(http.StatusUnauthorized, "missing bearer token")
 	}
-	user, err := s.auth.Authenticate(r.Context(), strings.TrimSpace(token))
+	// The kube client sets no timeout of its own, and net/http cancels
+	// r.Context on a dropped client only once the body is read, so a
+	// stalled review would otherwise outlive every other limit.
+	ctx := r.Context()
+	if s.cfg.RequestTimeout > 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, s.cfg.RequestTimeout)
+		defer cancel()
+	}
+	user, err := s.auth.Authenticate(ctx, strings.TrimSpace(token))
 	if errors.Is(err, ErrUnauthenticated) {
 		return "", errorf(http.StatusUnauthorized, "invalid bearer token")
 	}
@@ -339,7 +348,7 @@ func (s *Server) authorize(r *http.Request, route Route) (string, *HTTPError) {
 		slog.Error("token review failed", "error", err)
 		return "", errorf(http.StatusServiceUnavailable, "authentication unavailable")
 	}
-	allowed, err := s.auth.Authorize(r.Context(), user, route.ResourceAttributes())
+	allowed, err := s.auth.Authorize(ctx, user, route.ResourceAttributes())
 	if err != nil {
 		slog.Error("subject access review failed", "error", err)
 		return "", errorf(http.StatusServiceUnavailable, "authorization unavailable")

internal/observabilityproxy/server_deadline_test.go

@@ -22,11 +22,16 @@ import (
 	"io"
 	"net"
 	"net/http"
+	"net/http/httptest"
+	"net/url"
 	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
+
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
 )
 
 // pipeListener hands the server one end of a net.Pipe per dial. A pipe has
@@ -386,3 +391,56 @@ func TestServerEarlyRefusalWithUnreadBody(t *testing.T) {
 		}
 	}
 }
+
+// stalledAPIServer answers the TokenReview for "good" unless stallTokens
+// is set, and never answers a SubjectAccessReview: it holds each stalled
+// call until the caller gives up or the test ends.
+func stalledAPIServer(t *testing.T, stallTokens bool) kubernetes.Interface {
+	t.Helper()
+	release := make(chan struct{})
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if strings.HasSuffix(r.URL.Path, "/tokenreviews") && !stallTokens {
+			w.Header().Set("Content-Type", "application/json")
+			_, _ = io.WriteString(w, `{"apiVersion":"authentication.k8s.io/v1","kind":"TokenReview","status":{"authenticated":true,"user":{"username":"alice"}}}`)
+			return
+		}
+		select {
+		case <-r.Context().Done():
+		case <-release:
+		}
+	}))
+	t.Cleanup(srv.Close)
+	// After srv, so it runs first and srv.Close does not wait on a stalled call.
+	t.Cleanup(func() { close(release) })
+	return kubernetes.NewForConfigOrDie(&rest.Config{Host: srv.URL})
+}
+
+// TestServerStalledReviewAnswersInTime holds TokenReview or
+// SubjectAccessReview open on a POST whose body is complete but unread:
+// net/http does not watch the connection until the body is read, so only
+// a bounded review call ends the request.
+func TestServerStalledReviewAnswersInTime(t *testing.T) {
+	for _, tc := range []struct {
+		name        string
+		stallTokens bool
+	}{{"token review", true}, {"subject access review", false}} {
+		t.Run(tc.name, func(t *testing.T) {
+			h := newHarness(t, Config{RequestTimeout: 200 * time.Millisecond}, nil)
+			h.server.auth = NewAuthorizer(stalledAPIServer(t, tc.stallTokens), 10*time.Second)
+			got := make(chan int, 1)
+			go func() {
+				resp := h.do(http.MethodPost, logsRoot+"query", "good", url.Values{"query": {"*"}}, nil)
+				_ = resp.Body.Close()
+				got <- resp.StatusCode
+			}()
+			select {
+			case code := <-got:
+				if code != http.StatusServiceUnavailable {
+					t.Fatalf("stalled %s answered %d, want 503", tc.name, code)
+				}
+			case <-time.After(2 * time.Second):
+				t.Fatalf("a stalled %s holds the request past ten request timeouts", tc.name)
+			}
+		})
+	}
+}