diff --git a/s3api/middlewares/body-reader.go b/s3api/middlewares/body-reader.go index 84222475f..4cfddde78 100644 --- a/s3api/middlewares/body-reader.go +++ b/s3api/middlewares/body-reader.go @@ -88,7 +88,7 @@ var _ ChecksumReader = &MockChecksumReader{} func wrapBodyReader(ctx fiber.Ctx, wr func(io.Reader) io.Reader) { rdr, ok := utils.ContextKeyBodyReader.Get(ctx).(io.Reader) if !ok { - rdr = ctx.Request().BodyStream() + rdr = requestBodyStream(ctx) // Override the body reader with an empty reader to prevent panics // in case of unexpected or malformed HTTP requests. if rdr == nil { diff --git a/s3api/middlewares/drain-request-body.go b/s3api/middlewares/drain-request-body.go new file mode 100644 index 000000000..7d8ce0d66 --- /dev/null +++ b/s3api/middlewares/drain-request-body.go @@ -0,0 +1,238 @@ +// Copyright 2026 Versity Software +// This file is 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 middlewares + +import ( + "errors" + "io" + "net" + "time" + + "github.com/fil-forge/versitygw/debuglogger" + "github.com/fil-forge/versitygw/s3api/utils" + "github.com/gofiber/fiber/v3" +) + +const ( + // maxDrainBytes caps how much unread request body is read and discarded + // after the response has been decided. It matches net/http's + // maxPostHandlerReadBytes: enough to cover request bodies that are rejected + // early (bad chunk framing, auth failures, missing buckets), small enough + // that a rejected multi-gigabyte upload is not streamed through the gateway + // just to be thrown away. A body with more than this still unread is left + // alone and the connection is closed, so its client can still see a reset. + maxDrainBytes int64 = 256 << 10 +) + +// The drain is bounded twice: idle time, so a client that stops sending is cut +// loose quickly, and total time, so a client that trickles cannot hold a worker +// for long. +var ( + drainIdleTimeout = time.Second + drainTotalTimeout = 5 * time.Second +) + +// DrainRequestBody reads and discards whatever is left of the request body once +// the rest of the handler chain is done with it. +// +// The gateway can decide a response long before the client has finished sending +// the body: an invalid chunk size is detected a few kilobytes into an aws-chunked +// upload, a signature check fails before any payload is read, and so on. fasthttp +// streams request bodies (StreamRequestBody) and does not drain what the handler +// left behind, so the connection is closed with unread bytes still queued in the +// socket. The kernel answers the client's in-flight writes with an RST, and the +// client reports "connection reset by peer" instead of the S3 error the gateway +// took the trouble to produce. +// +// Draining first lets the client finish its write and read the real error. It +// also keeps keep-alive connections in sync: leftover body bytes would otherwise +// be parsed as the start of the next request, which on a connection shared by an +// upstream proxy mixes requests across tenants. +// +// It wraps the body in a bodyStreamTracker on the way in, which is what tells a +// body the handler finished from one it abandoned. Every reader in the chain +// must therefore take the body from requestBodyStream. +// +// Register it before every route so it wraps all of them. It runs before the +// fiber ErrorHandler, which fiber invokes after the handler chain returns, so +// that handler must not reset the response header the drain may have written to. +func DrainRequestBody() fiber.Handler { + return func(ctx fiber.Ctx) error { + var body *bodyStreamTracker + if stream := ctx.Request().BodyStream(); stream != nil { + body = &bodyStreamTracker{reader: stream} + utils.ContextKeyBodyStream.Set(ctx, body) + } + + // deferred so a panic unwinding through the chain still drains + defer drainRequestBody(ctx, body) + + return ctx.Next() + } +} + +func drainRequestBody(ctx fiber.Ctx, body *bodyStreamTracker) { + if body == nil || ctx.Request().BodyStream() == nil { + // The body was either absent, or buffered in full and released by + // fasthttp behind the chain's back, the way ctx.Body() does. + return + } + + if errors.Is(body.end, io.EOF) { + // The handler read the body to its end: the socket holds nothing more + // of it and the connection is already in sync for the next request. + return + } + + conn := requestConn(ctx) + if conn != nil { + defer conn.SetReadDeadline(time.Time{}) + } + + // The leftovers are read back through the tracker, so a drain that reaches + // the end of the body records it the same way the handler's reads would. + src := io.Reader(body) + if body.end != nil { + // The framing broke before the body ended, so what is still queued + // cannot be told apart from the start of the next request and the + // connection cannot carry one. Draining is still worth attempting: the + // connection is going away either way, and absorbing the client's + // in-flight write is what lets it read the S3 error instead of an RST. + // Only the raw socket can absorb it once the framing is gone. + ctx.Response().Header.SetConnectionClose() + if conn == nil { + return + } + src = conn + } + + reader := &drainReader{ + reader: src, + conn: conn, + deadline: time.Now().Add(drainTotalTimeout), + } + + n, err := io.CopyN(io.Discard, reader, maxDrainBytes) + if err == nil { + // Filled the cap exactly. One more read tells a body that happened to + // end there from one with more still to come. + err = reader.atEOF() + } + if errors.Is(err, io.EOF) { + if n > 0 { + debuglogger.Logf("discarded %v unread request body bytes before responding", n) + } + return + } + + if err != nil { + debuglogger.Logf("failed to discard the unread request body after %v bytes: %v", n, err) + } else { + debuglogger.Logf("unread request body exceeds the %v byte drain limit: %v bytes discarded", maxDrainBytes, n) + } + + // The body was not consumed to its end, so the bytes still in flight would + // be parsed as the start of the next request on a keep-alive connection. + // Tell fasthttp to close it instead. + ctx.Response().Header.SetConnectionClose() +} + +// requestBodyStream returns the reader the request body must be read through. +// Reading ctx.Request().BodyStream() directly instead hides those reads from +// DrainRequestBody, which then cannot tell a finished body from an abandoned +// one and falls back to closing the connection. +// +// It yields the raw stream where DrainRequestBody is not registered, and nil +// when the request carries no streamed body. +func requestBodyStream(ctx fiber.Ctx) io.Reader { + if body, ok := utils.ContextKeyBodyStream.Get(ctx).(*bodyStreamTracker); ok { + return body + } + + return ctx.Request().BodyStream() +} + +// bodyStreamTracker remembers how the request body ended, so the drain can tell +// a body the handler read to its end from one it stopped partway through. +// +// fasthttp cannot be asked a second time. Its requestStream reports EOF +// idempotently for a Content-Length body, but not for a chunked one: past the +// terminating chunk it goes back to the socket for another chunk header that +// will never come, so a read of a body the handler already finished blocks +// until the deadline and holds the response back with it. The first terminal +// result is recorded here and replayed instead. +type bodyStreamTracker struct { + reader io.Reader + // end is the first error the stream ended on: nil while it still has more + // to give, io.EOF once it was read out in full, and the framing error if it + // broke before that. + end error +} + +func (t *bodyStreamTracker) Read(p []byte) (int, error) { + if t.end != nil { + return 0, t.end + } + + n, err := t.reader.Read(p) + if err != nil { + t.end = err + } + + return n, err +} + +// drainReader refreshes the connection's read deadline before every read, so a +// client that keeps sending is never cut off mid-drain while one that goes quiet +// is dropped after drainIdleTimeout. deadline caps the whole drain regardless. +type drainReader struct { + reader io.Reader + conn net.Conn + deadline time.Time +} + +func (dr *drainReader) Read(p []byte) (int, error) { + if dr.conn != nil { + next := time.Now().Add(drainIdleTimeout) + if next.After(dr.deadline) { + next = dr.deadline + } + if err := dr.conn.SetReadDeadline(next); err != nil { + return 0, err + } + } + + return dr.reader.Read(p) +} + +// atEOF reports io.EOF when the body ends exactly at the drain limit. +func (dr *drainReader) atEOF() error { + var b [1]byte + n, err := dr.Read(b[:]) + if n == 0 && err == nil { + return nil + } + + return err +} + +func requestConn(ctx fiber.Ctx) net.Conn { + rctx := ctx.RequestCtx() + if rctx == nil { + return nil + } + + return rctx.Conn() +} diff --git a/s3api/middlewares/drain-request-body_test.go b/s3api/middlewares/drain-request-body_test.go new file mode 100644 index 000000000..49d315a90 --- /dev/null +++ b/s3api/middlewares/drain-request-body_test.go @@ -0,0 +1,459 @@ +// Copyright 2026 Versity Software +// This file is 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 middlewares + +import ( + "bufio" + "bytes" + "fmt" + "io" + "net" + "net/http" + "testing" + "time" + + "github.com/gofiber/fiber/v3" +) + +func TestDrainRequestBody_slowClientFinishesWritingAndReadsTheResponse(t *testing.T) { + // keep-alive is off by default in the gateway and on with --keep-alive + for _, disableKeepalive := range []bool{true, false} { + t.Run(fmt.Sprintf("disableKeepalive=%v", disableKeepalive), func(t *testing.T) { + slowClientFinishesWriting(t, disableKeepalive) + }) + } +} + +func slowClientFinishesWriting(t *testing.T, disableKeepalive bool) { + t.Helper() + + addr := startEarlyResponder(t, disableKeepalive) + // small enough to be drained in full, big enough that it cannot sit in the + // socket buffers while the server decides to close + body := bytes.Repeat([]byte("a"), 128<<10) + + conn, err := net.Dial("tcp", addr) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + if err := conn.SetDeadline(time.Now().Add(30 * time.Second)); err != nil { + t.Fatalf("set deadline: %v", err) + } + + if _, err := conn.Write(putHeaders(addr, len(body))); err != nil { + t.Fatalf("write headers: %v", err) + } + + // dribble the body out, so the response is decided well before the last byte + for off := 0; off < len(body); off += 4 << 10 { + if _, err := conn.Write(body[off:min(off+(4<<10), len(body))]); err != nil { + t.Fatalf("the server dropped the connection with %v of %v body bytes sent: %v", off, len(body), err) + } + time.Sleep(time.Millisecond) + } + + br := bufio.NewReader(conn) + resp, err := http.ReadResponse(br, nil) + if err != nil { + t.Fatalf("read response: %v", err) + } + defer resp.Body.Close() + if _, err := io.Copy(io.Discard, resp.Body); err != nil { + t.Fatalf("read response body: %v", err) + } + + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("expected status %v, got %v", http.StatusBadRequest, resp.StatusCode) + } + if disableKeepalive { + return + } + if resp.Close { + t.Fatal("expected the connection to stay usable after the body was drained") + } + + // a fully drained body leaves the connection in sync for the next request + if resp2 := reuseConnection(t, conn, br, addr, resp); resp2.StatusCode != http.StatusBadRequest { + t.Fatalf("expected status %v on the reused connection, got %v", http.StatusBadRequest, resp2.StatusCode) + } +} + +func TestDrainRequestBody_closesConnectionWhenUnreadBodyExceedsTheLimit(t *testing.T) { + addr := startEarlyResponder(t, false) + body := bytes.Repeat([]byte("a"), int(maxDrainBytes)*4) + + conn, err := net.Dial("tcp", addr) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + if err := conn.SetDeadline(time.Now().Add(30 * time.Second)); err != nil { + t.Fatalf("set deadline: %v", err) + } + + if _, err := conn.Write(putHeaders(addr, len(body))); err != nil { + t.Fatalf("write headers: %v", err) + } + // the write is expected to fail once the server gives up draining + go conn.Write(body) + + resp, err := http.ReadResponse(bufio.NewReader(conn), nil) + if err != nil { + t.Fatalf("read response: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("expected status %v, got %v", http.StatusBadRequest, resp.StatusCode) + } + if !resp.Close { + t.Fatal("expected 'Connection: close', the undrained body bytes would desync the next request") + } +} + +func TestDrainRequestBody_givesUpOnAClientThatStopsSending(t *testing.T) { + shortenDrainTimeouts(t) + + addr := startEarlyResponder(t, true) + + conn, err := net.Dial("tcp", addr) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + if err := conn.SetDeadline(time.Now().Add(30 * time.Second)); err != nil { + t.Fatalf("set deadline: %v", err) + } + + // announce a body, send only what the handler consumes, then go quiet + if _, err := conn.Write(putHeaders(addr, 64<<10)); err != nil { + t.Fatalf("write headers: %v", err) + } + if _, err := conn.Write(bytes.Repeat([]byte("a"), handlerReadBytes)); err != nil { + t.Fatalf("write body: %v", err) + } + + start := time.Now() + resp, err := http.ReadResponse(bufio.NewReader(conn), nil) + if err != nil { + t.Fatalf("read response: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("expected status %v, got %v", http.StatusBadRequest, resp.StatusCode) + } + if elapsed := time.Since(start); elapsed > 5*time.Second { + t.Fatalf("the drain held the response for %v, the timeout should have cut it short", elapsed) + } +} + +// A chunked body the handler read to its end must not be read again: fasthttp's +// requestStream goes back to the socket for another chunk header past the +// terminating chunk, which would hold the response back until the deadline. It +// also has nothing left to desync the connection, so it keeps keep-alive. +func TestDrainRequestBody_doesNotStallAChunkedBodyTheHandlerFinished(t *testing.T) { + addr := startFullReader(t) + + conn, err := net.Dial("tcp", addr) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + if err := conn.SetDeadline(time.Now().Add(30 * time.Second)); err != nil { + t.Fatalf("set deadline: %v", err) + } + + if _, err := conn.Write(chunkedRequest(addr, []byte("hello"))); err != nil { + t.Fatalf("write request: %v", err) + } + + start := time.Now() + br := bufio.NewReader(conn) + resp, err := http.ReadResponse(br, nil) + if err != nil { + t.Fatalf("read response: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + t.Fatalf("expected status %v, got %v", http.StatusOK, resp.StatusCode) + } + if elapsed := time.Since(start); elapsed > drainIdleTimeout { + t.Fatalf("the response was held back for %v: the drain re-read a finished chunked body", elapsed) + } + if resp.Close { + t.Fatal("expected the connection to stay usable after a chunked body the handler finished") + } + + if resp2 := reuseConnection(t, conn, br, addr, resp); resp2.StatusCode != http.StatusOK { + t.Fatalf("expected status %v on the reused connection, got %v", http.StatusOK, resp2.StatusCode) + } +} + +// A chunked body the handler abandoned is drained like any other. Only the read +// past its end is unsafe, and the drain never gets there on a body it finished. +func TestDrainRequestBody_drainsAnUnreadChunkedBody(t *testing.T) { + addr := startEarlyResponder(t, false) + + conn, err := net.Dial("tcp", addr) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + if err := conn.SetDeadline(time.Now().Add(30 * time.Second)); err != nil { + t.Fatalf("set deadline: %v", err) + } + + // leave one decoded byte unread after the handler's initial read + chunk := bytes.Repeat([]byte("a"), handlerReadBytes+1) + if _, err := conn.Write(chunkedRequest(addr, chunk)); err != nil { + t.Fatalf("write request: %v", err) + } + + br := bufio.NewReader(conn) + resp, err := http.ReadResponse(br, nil) + if err != nil { + t.Fatalf("read response: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("expected status %v, got %v", http.StatusBadRequest, resp.StatusCode) + } + if resp.Close { + t.Fatal("expected the connection to stay usable after the chunked body was drained") + } + + if resp2 := reuseConnection(t, conn, br, addr, resp); resp2.StatusCode != http.StatusBadRequest { + t.Fatalf("expected status %v on the reused connection, got %v", http.StatusBadRequest, resp2.StatusCode) + } +} + +func TestDrainRequestBody_closesConnectionWhenUnreadChunkedBodyExceedsTheLimit(t *testing.T) { + addr := startEarlyResponder(t, false) + chunk := bytes.Repeat([]byte("a"), int(maxDrainBytes)*4) + + conn, err := net.Dial("tcp", addr) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + if err := conn.SetDeadline(time.Now().Add(30 * time.Second)); err != nil { + t.Fatalf("set deadline: %v", err) + } + + // the write is expected to fail once the server gives up draining + go conn.Write(chunkedRequest(addr, chunk)) + + resp, err := http.ReadResponse(bufio.NewReader(conn), nil) + if err != nil { + t.Fatalf("read response: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("expected status %v, got %v", http.StatusBadRequest, resp.StatusCode) + } + if !resp.Close { + t.Fatal("expected 'Connection: close', the undrained chunk bytes would desync the next request") + } +} + +// Broken chunk framing leaves nothing that can be decoded as body bytes, so the +// connection cannot be reused. The drain still absorbs what the client is +// writing, off the socket itself, so it reaches the S3 error instead of an RST. +func TestDrainRequestBody_drainsAndClosesOnBrokenChunkedFraming(t *testing.T) { + shortenDrainTimeouts(t) + + addr := startEarlyResponder(t, false) + // small enough to be absorbed in full, big enough that it cannot sit in the + // socket buffers while the server decides to close + garbage := bytes.Repeat([]byte("a"), 128<<10) + + conn, err := net.Dial("tcp", addr) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + if err := conn.SetDeadline(time.Now().Add(30 * time.Second)); err != nil { + t.Fatalf("set deadline: %v", err) + } + + // a chunk size that is not a hex number: the handler's first read fails + head := fmt.Appendf(nil, "PUT /object HTTP/1.1\r\nHost: %s\r\nTransfer-Encoding: chunked\r\n\r\nzz\r\n", addr) + if _, err := conn.Write(head); err != nil { + t.Fatalf("write request: %v", err) + } + + // dribble the rest out, the way a client keeps uploading after the gateway + // has already given up on the request + for off := 0; off < len(garbage); off += 4 << 10 { + if _, err := conn.Write(garbage[off:min(off+(4<<10), len(garbage))]); err != nil { + t.Fatalf("the server dropped the connection with %v of %v bytes sent: %v", off, len(garbage), err) + } + time.Sleep(time.Millisecond) + } + + resp, err := http.ReadResponse(bufio.NewReader(conn), nil) + if err != nil { + t.Fatalf("read response: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusBadRequest { + t.Fatalf("expected status %v, got %v", http.StatusBadRequest, resp.StatusCode) + } + if !resp.Close { + t.Fatal("expected 'Connection: close', broken chunk framing cannot be resynchronized") + } +} + +// The same, for a Content-Length body: fasthttp reports EOF idempotently there, +// so it is drained, but a handler that already finished it must not be delayed. +func TestDrainRequestBody_doesNotStallABodyTheHandlerFinished(t *testing.T) { + addr := startFullReader(t) + + conn, err := net.Dial("tcp", addr) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + if err := conn.SetDeadline(time.Now().Add(30 * time.Second)); err != nil { + t.Fatalf("set deadline: %v", err) + } + + if _, err := conn.Write(append(putHeaders(addr, 5), "hello"...)); err != nil { + t.Fatalf("write request: %v", err) + } + + start := time.Now() + resp, err := http.ReadResponse(bufio.NewReader(conn), nil) + if err != nil { + t.Fatalf("read response: %v", err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + t.Fatalf("expected status %v, got %v", http.StatusOK, resp.StatusCode) + } + if elapsed := time.Since(start); elapsed > drainIdleTimeout { + t.Fatalf("the response was held back for %v on a body the handler had finished", elapsed) + } +} + +// fasthttp pre-reads up to 8KB of a declared body before it hands the request to +// the handler, so a test body has to be at least that big to reach the drain. +const handlerReadBytes = 8 << 10 + +// startEarlyResponder serves a fiber app that reads only the head of the request +// body and then answers, the way the gateway rejects an upload on a bad chunk +// header or a failed authorization long before the client is done sending. +func startEarlyResponder(t *testing.T, disableKeepalive bool) string { + t.Helper() + + app := fiber.New(fiber.Config{ + StreamRequestBody: true, + DisableKeepalive: disableKeepalive, + }) + app.Use("*", DrainRequestBody()) + app.Put("/object", func(ctx fiber.Ctx) error { + if body := requestBodyStream(ctx); body != nil { + // consume a little of it, the way the chunk reader parses a chunk + // header before rejecting the upload + io.CopyN(io.Discard, body, handlerReadBytes) //nolint:errcheck + } + return ctx.Status(http.StatusBadRequest).SendString("rejected") + }) + + return listen(t, app) +} + +func listen(t *testing.T, app *fiber.App) string { + t.Helper() + + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + go app.Listener(ln, fiber.ListenConfig{DisableStartupMessage: true}) //nolint:errcheck + t.Cleanup(func() { app.Shutdown() }) //nolint:errcheck + + return ln.Addr().String() +} + +func putHeaders(addr string, contentLength int) []byte { + return fmt.Appendf(nil, "PUT /object HTTP/1.1\r\nHost: %s\r\nContent-Length: %d\r\n\r\n", addr, contentLength) +} + +// startFullReader serves a fiber app whose handler consumes the whole request +// body, so the drain has nothing left to do. +func startFullReader(t *testing.T) string { + t.Helper() + + app := fiber.New(fiber.Config{StreamRequestBody: true}) + app.Use("*", DrainRequestBody()) + app.Put("/object", func(ctx fiber.Ctx) error { + if body := requestBodyStream(ctx); body != nil { + if _, err := io.Copy(io.Discard, body); err != nil { + return err + } + } + return ctx.SendStatus(http.StatusOK) + }) + + return listen(t, app) +} + +// chunkedRequest builds a PUT that carries body as a single chunk, the framing +// a client uses when it cannot announce a Content-Length up front. +func chunkedRequest(addr string, body []byte) []byte { + req := fmt.Appendf(nil, "PUT /object HTTP/1.1\r\nHost: %s\r\nTransfer-Encoding: chunked\r\n\r\n%x\r\n", addr, len(body)) + req = append(req, body...) + + return append(req, "\r\n0\r\n\r\n"...) +} + +// reuseConnection sends a second, bodiless request on the same connection once +// prev is off the wire. It is answered only while the connection is still in +// sync with the client. +func reuseConnection(t *testing.T, conn net.Conn, br *bufio.Reader, addr string, prev *http.Response) *http.Response { + t.Helper() + + if _, err := io.Copy(io.Discard, prev.Body); err != nil { + t.Fatalf("read response body: %v", err) + } + if _, err := conn.Write(putHeaders(addr, 0)); err != nil { + t.Fatalf("write second request: %v", err) + } + resp, err := http.ReadResponse(br, nil) + if err != nil { + t.Fatalf("read second response: %v", err) + } + t.Cleanup(func() { resp.Body.Close() }) //nolint:errcheck + + return resp +} + +// shortenDrainTimeouts keeps a test that waits the drain out from taking the +// production timeouts to finish. +func shortenDrainTimeouts(t *testing.T) { + t.Helper() + + idle, total := drainIdleTimeout, drainTotalTimeout + drainIdleTimeout, drainTotalTimeout = 100*time.Millisecond, 250*time.Millisecond + t.Cleanup(func() { drainIdleTimeout, drainTotalTimeout = idle, total }) +} diff --git a/s3api/middlewares/object-post-auth.go b/s3api/middlewares/object-post-auth.go index d5e9c7187..b329a441d 100644 --- a/s3api/middlewares/object-post-auth.go +++ b/s3api/middlewares/object-post-auth.go @@ -69,7 +69,7 @@ func AuthorizePostObject(root RootUserConfig, iam auth.IAMService, region string return s3err.GetAPIError(s3err.ErrMalformedPOSTRequest) } - bodyRdr := ctx.Request().BodyStream() + bodyRdr := requestBodyStream(ctx) if bodyRdr == nil { bodyRdr = bytes.NewReader(ctx.BodyRaw()) } diff --git a/s3api/middlewares/public-bucket.go b/s3api/middlewares/public-bucket.go index e9fad0646..a890015a7 100644 --- a/s3api/middlewares/public-bucket.go +++ b/s3api/middlewares/public-bucket.go @@ -98,7 +98,7 @@ func AuthorizePublicBucketAccess(be backend.Backend, s3action string, policyPerm return err } else if utils.IsUnsignedPaylod(payloadHash) { // for UNSIGNED-PAYLOD simply store the body reader in context locals - utils.ContextKeyBodyReader.Set(ctx, ctx.Request().BodyStream()) + utils.ContextKeyBodyReader.Set(ctx, requestBodyStream(ctx)) return nil } else { // stack a hash reader to calculated the payload sha256 hash diff --git a/s3api/server.go b/s3api/server.go index 4dce95218..6ade20989 100644 --- a/s3api/server.go +++ b/s3api/server.go @@ -127,6 +127,11 @@ func New( StackTraceHandler: stackTraceHandler, })) + // initialize the request body drainer. it goes right after the panic + // recovery and before every route, so it wraps all of them and a panic in + // the drain itself is still recovered + app.Use("*", middlewares.DrainRequestBody()) + // Logging middlewares if !server.quiet { app.Use("*", logger.New(logger.Config{ diff --git a/s3api/utils/context-keys.go b/s3api/utils/context-keys.go index 96e84273f..3df344b7e 100644 --- a/s3api/utils/context-keys.go +++ b/s3api/utils/context-keys.go @@ -34,6 +34,7 @@ const ( ContextKeyParsedAcl ContextKey = "parsed-acl" ContextKeySkipResBodyLog ContextKey = "skip-res-body-log" ContextKeyBodyReader ContextKey = "body-reader" + ContextKeyBodyStream ContextKey = "body-stream" ContextKeySkip ContextKey = "__skip" ContextKeyStack ContextKey = "stack" ContextKeyBucketOwner ContextKey = "bucket-owner" diff --git a/tests/integration/ListBuckets.go b/tests/integration/ListBuckets.go index e032ebb38..ca4f9c114 100644 --- a/tests/integration/ListBuckets.go +++ b/tests/integration/ListBuckets.go @@ -227,6 +227,13 @@ func ListBuckets_with_prefix(s *S3Conf) error { return fmt.Errorf("expected prefix to be %v, instead got %v", prefix, getString(out.Prefix)) } + // ListBuckets returns buckets ordered lexically by name; compareBuckets + // is order-sensitive, so sort the expected set to match (creation order + // only coincides with lexical order within a single digit band of the + // bucket-name counter). + sort.Slice(prefixedBuckets, func(i, j int) bool { + return *prefixedBuckets[i].Name < *prefixedBuckets[j].Name + }) if !compareBuckets(out.Buckets, prefixedBuckets, ignore...) { return fmt.Errorf("expected list buckets result to be %v, instead got %v", prefixedBuckets, out.Buckets) @@ -418,9 +425,18 @@ func ListBuckets_success(s *S3Conf) error { return fmt.Errorf("expected owner to be %v, instead got %v", s.awsID, getString(out.Owner.ID)) } - if !compareBuckets(out.Buckets, buckets, ignore...) { + // ListBuckets returns buckets ordered lexically by name; compareBuckets + // is order-sensitive, so compare against a lexically-sorted copy (the + // bucket-name counter's creation order only matches lexical order within + // a single digit band). Keep `buckets` in creation order for teardown, + // which relies on buckets[0] being the action bucket. + expected := append([]types.Bucket(nil), buckets...) + sort.Slice(expected, func(i, j int) bool { + return *expected[i].Name < *expected[j].Name + }) + if !compareBuckets(out.Buckets, expected, ignore...) { return fmt.Errorf("expected list buckets result to be %v, instead got %v", - sprintBuckets(buckets), sprintBuckets(out.Buckets)) + sprintBuckets(expected), sprintBuckets(out.Buckets)) } for _, elem := range buckets[1:] {