diff --git a/pkg/server/recovery.go b/pkg/server/recovery.go new file mode 100644 index 000000000..2d3b61e45 --- /dev/null +++ b/pkg/server/recovery.go @@ -0,0 +1,95 @@ +package server + +import ( + "context" + "io" + "log/slog" + "net/http" + "runtime/debug" + + "connectrpc.com/connect" + "github.com/raystack/frontier/pkg/errors" +) + +// connectPanicRecovery converts a panic anywhere in the RPC handler chain +// into a CodeInternal error response. The panic value and stack go to the +// server log only, never to the caller. +func connectPanicRecovery(logger *slog.Logger) connect.HandlerOption { + return connect.WithRecover(func(ctx context.Context, spec connect.Spec, _ http.Header, panicValue any) error { + logger.ErrorContext(ctx, "rpc handler panic", + "procedure", spec.Procedure, + "panic", panicValue, + "stack", string(debug.Stack())) + return connect.NewError(connect.CodeInternal, errors.ErrInternalServerError) + }) +} + +// committedWriter tracks whether the response status line has gone out, so +// the recovery handler knows whether a 500 can still be sent. +type committedWriter struct { + http.ResponseWriter + committed bool +} + +func (c *committedWriter) WriteHeader(statusCode int) { + c.committed = true + c.ResponseWriter.WriteHeader(statusCode) +} + +func (c *committedWriter) Write(b []byte) (int, error) { + c.committed = true + return c.ResponseWriter.Write(b) +} + +// Flush commits the response too. The connect reverse proxy on the UI server +// needs the Flusher interface for streaming. +func (c *committedWriter) Flush() { + c.committed = true + if f, ok := c.ResponseWriter.(http.Flusher); ok { + f.Flush() + } +} + +// Unwrap lets http.ResponseController reach the underlying writer's +// Hijacker, deadline, and full-duplex methods. +func (c *committedWriter) Unwrap() http.ResponseWriter { + return c.ResponseWriter +} + +// ReadFrom keeps the underlying writer's optimized copy path (pooled +// buffers, sendfile for OS files), which io.Copy looks for on the +// destination directly without walking Unwrap. +func (c *committedWriter) ReadFrom(r io.Reader) (int64, error) { + c.committed = true + if rf, ok := c.ResponseWriter.(io.ReaderFrom); ok { + return rf.ReadFrom(r) + } + return io.Copy(c.ResponseWriter, r) +} + +// httpPanicRecovery responds with a plain 500 instead of dropping the +// connection when a handler behind the wrapped mux panics. If the handler +// already wrote part of a response, the status can no longer be changed, so +// it aborts the connection instead of appending an error to the partial body. +func httpPanicRecovery(logger *slog.Logger, next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + cw := &committedWriter{ResponseWriter: w} + defer func() { + if panicValue := recover(); panicValue != nil { + // net/http checks for ErrAbortHandler with ==, so we should too. + if panicValue == http.ErrAbortHandler { + panic(panicValue) + } + logger.ErrorContext(r.Context(), "http handler panic", + "path", r.URL.Path, + "panic", panicValue, + "stack", string(debug.Stack())) + if cw.committed { + panic(http.ErrAbortHandler) + } + http.Error(cw, errors.ErrInternalServerError.Error(), http.StatusInternalServerError) + } + }() + next.ServeHTTP(cw, r) + }) +} diff --git a/pkg/server/recovery_test.go b/pkg/server/recovery_test.go new file mode 100644 index 000000000..a7b5851d8 --- /dev/null +++ b/pkg/server/recovery_test.go @@ -0,0 +1,203 @@ +package server + +import ( + "bytes" + "context" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "connectrpc.com/connect" + "google.golang.org/protobuf/types/known/emptypb" +) + +func TestConnectPanicRecoveryReturnsInternalError(t *testing.T) { + var logBuf bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&logBuf, nil)) + + handler := connect.NewUnaryHandler( + "/test.v1.TestService/Panic", + func(ctx context.Context, req *connect.Request[emptypb.Empty]) (*connect.Response[emptypb.Empty], error) { + panic("boom") + }, + connectPanicRecovery(logger), + ) + mux := http.NewServeMux() + mux.Handle("/test.v1.TestService/Panic", handler) + srv := httptest.NewServer(mux) + defer srv.Close() + + resp, err := http.Post(srv.URL+"/test.v1.TestService/Panic", "application/json", strings.NewReader("{}")) + if err != nil { + t.Fatalf("expected an error response, got a dropped connection: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusInternalServerError { + t.Errorf("status = %d, want %d", resp.StatusCode, http.StatusInternalServerError) + } + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatalf("read body: %v", err) + } + if !strings.Contains(string(body), `"internal"`) { + t.Errorf("body %q does not carry the internal error code", body) + } + if strings.Contains(string(body), "boom") { + t.Errorf("panic detail leaked to the client: %q", body) + } + + logged := logBuf.String() + if !strings.Contains(logged, `"level":"ERROR"`) { + t.Errorf("panic was not logged at error level: %q", logged) + } + if !strings.Contains(logged, "boom") || !strings.Contains(logged, "/test.v1.TestService/Panic") { + t.Errorf("log entry is missing the panic value or procedure: %q", logged) + } + if !strings.Contains(logged, "recovery_test.go") { + t.Errorf("log entry is missing the stack trace: %q", logged) + } +} + +func TestHTTPPanicRecoveryReturnsInternalError(t *testing.T) { + var logBuf bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&logBuf, nil)) + + mux := http.NewServeMux() + mux.HandleFunc("/panic", func(w http.ResponseWriter, r *http.Request) { + panic("boom") + }) + + rec := httptest.NewRecorder() + httpPanicRecovery(logger, mux).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/panic", nil)) + + if rec.Code != http.StatusInternalServerError { + t.Errorf("status = %d, want %d", rec.Code, http.StatusInternalServerError) + } + if strings.Contains(rec.Body.String(), "boom") { + t.Errorf("panic detail leaked to the client: %q", rec.Body.String()) + } + + logged := logBuf.String() + if !strings.Contains(logged, `"level":"ERROR"`) || !strings.Contains(logged, "boom") || !strings.Contains(logged, "/panic") { + t.Errorf("log entry is missing the error level, panic value, or path: %q", logged) + } +} + +func TestHTTPPanicRecoveryPassesThroughNormalRequests(t *testing.T) { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + + mux := http.NewServeMux() + mux.HandleFunc("/ok", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusTeapot) + }) + + rec := httptest.NewRecorder() + httpPanicRecovery(logger, mux).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/ok", nil)) + + if rec.Code != http.StatusTeapot { + t.Errorf("status = %d, want %d", rec.Code, http.StatusTeapot) + } +} + +func TestHTTPPanicRecoveryAbortsCommittedResponses(t *testing.T) { + var logBuf bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&logBuf, nil)) + + mux := http.NewServeMux() + mux.HandleFunc("/partial", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + if _, err := w.Write([]byte("partial")); err != nil { + t.Fatalf("write: %v", err) + } + panic("boom") + }) + + rec := httptest.NewRecorder() + func() { + defer func() { + if recovered := recover(); recovered != http.ErrAbortHandler { + t.Errorf("recovered %v, want http.ErrAbortHandler", recovered) + } + }() + httpPanicRecovery(logger, mux).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/partial", nil)) + }() + + if rec.Code != http.StatusOK { + t.Errorf("status = %d, want the already-committed %d", rec.Code, http.StatusOK) + } + if rec.Body.String() != "partial" { + t.Errorf("body = %q, want the partial body with nothing appended", rec.Body.String()) + } + if !strings.Contains(logBuf.String(), "boom") { + t.Errorf("panic was not logged: %q", logBuf.String()) + } +} + +func TestHTTPPanicRecoveryAbortsAfterFlush(t *testing.T) { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + + mux := http.NewServeMux() + mux.HandleFunc("/flush", func(w http.ResponseWriter, r *http.Request) { + w.(http.Flusher).Flush() + panic("boom") + }) + + rec := httptest.NewRecorder() + func() { + defer func() { + if recovered := recover(); recovered != http.ErrAbortHandler { + t.Errorf("recovered %v, want http.ErrAbortHandler", recovered) + } + }() + httpPanicRecovery(logger, mux).ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/flush", nil)) + }() + + if strings.Contains(rec.Body.String(), "internal server error") { + t.Errorf("error body appended to a flushed response: %q", rec.Body.String()) + } +} + +func TestCommittedWriterUnwrapReturnsUnderlyingWriter(t *testing.T) { + rec := httptest.NewRecorder() + cw := &committedWriter{ResponseWriter: rec} + + if cw.Unwrap() != http.ResponseWriter(rec) { + t.Errorf("Unwrap() = %v, want the underlying writer", cw.Unwrap()) + } +} + +func TestCommittedWriterReadFromCommitsAndCopies(t *testing.T) { + rec := httptest.NewRecorder() + cw := &committedWriter{ResponseWriter: rec} + + n, err := cw.ReadFrom(strings.NewReader("data")) + if err != nil || n != 4 { + t.Fatalf("ReadFrom = (%d, %v), want (4, nil)", n, err) + } + if !cw.committed { + t.Error("ReadFrom did not mark the response as committed") + } + if rec.Body.String() != "data" { + t.Errorf("body = %q, want %q", rec.Body.String(), "data") + } +} + +func TestHTTPPanicRecoveryRepanicsOnErrAbortHandler(t *testing.T) { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + + mux := http.NewServeMux() + mux.HandleFunc("/abort", func(w http.ResponseWriter, r *http.Request) { + panic(http.ErrAbortHandler) + }) + + defer func() { + if recovered := recover(); recovered != http.ErrAbortHandler { + t.Errorf("recovered %v, want http.ErrAbortHandler", recovered) + } + }() + httpPanicRecovery(logger, mux).ServeHTTP(httptest.NewRecorder(), httptest.NewRequest(http.MethodGet, "/abort", nil)) +} diff --git a/pkg/server/server.go b/pkg/server/server.go index c93d02b00..144cd2166 100644 --- a/pkg/server/server.go +++ b/pkg/server/server.go @@ -100,7 +100,7 @@ func ServeUI(ctx context.Context, logger *slog.Logger, uiConfig UIConfig, apiSer server := &http.Server{ Addr: fmt.Sprintf(":%d", uiConfig.Port), - Handler: mux, + Handler: httpPanicRecovery(logger, mux), ReadHeaderTimeout: apiServerConfig.ReadHeaderTimeout, IdleTimeout: apiServerConfig.IdleTimeout, } @@ -178,8 +178,10 @@ func ServeConnect(ctx context.Context, logger *slog.Logger, cfg Config, deps api auditInterceptor, sessionInterceptor.UnaryConnectResponseInterceptor()) - frontierPath, frontierHandler := frontierv1beta1connect.NewFrontierServiceHandler(frontierService, interceptors, connect.WithCodec(connectCodec{})) - adminPath, adminHandler := frontierv1beta1connect.NewAdminServiceHandler(frontierService, interceptors, connect.WithCodec(connectCodec{})) + // Panic recovery goes first so it sits outermost and catches panics from + // the other interceptors as well, not just the handlers. + frontierPath, frontierHandler := frontierv1beta1connect.NewFrontierServiceHandler(frontierService, connectPanicRecovery(logger), interceptors, connect.WithCodec(connectCodec{})) + adminPath, adminHandler := frontierv1beta1connect.NewAdminServiceHandler(frontierService, connectPanicRecovery(logger), interceptors, connect.WithCodec(connectCodec{})) // Create mux and register handlers mux := http.NewServeMux() @@ -218,8 +220,11 @@ func ServeConnect(ctx context.Context, logger *slog.Logger, cfg Config, deps api _ = json.NewEncoder(w).Encode(map[string]string{"status": "SERVING"}) }) - // Configure and create the server - handler := connectinterceptors.WithConnectCORS(mux, cfg.ConnectCors) + // Configure and create the server. Panic recovery wraps the whole mux so + // routes outside the two service handlers (webhook bridge, ping, health, + // reflection, CORS) are covered too; RPC panics are still converted to + // connect error codes by WithRecover before they can reach this net. + handler := httpPanicRecovery(logger, connectinterceptors.WithConnectCORS(mux, cfg.ConnectCors)) // Serve HTTP/1.1 and unencrypted HTTP/2. Unlike the x/net h2c wrapper // this replaces, the server tracks these HTTP/2 connections itself, so