diff --git a/Dockerfile.goreleaser b/Dockerfile.goreleaser index a82aa8497..821658771 100644 --- a/Dockerfile.goreleaser +++ b/Dockerfile.goreleaser @@ -1,4 +1,4 @@ -FROM golang:1.27.0-alpine AS build +FROM golang:1.27.1-alpine AS build ARG TARGETARCH diff --git a/buildscripts/checkdeps.sh b/buildscripts/checkdeps.sh index 4fc385a29..58b222462 100755 --- a/buildscripts/checkdeps.sh +++ b/buildscripts/checkdeps.sh @@ -7,7 +7,7 @@ _init() { ## Minimum required versions for build dependencies GIT_VERSION="1.0" - GO_VERSION="1.27.0" + GO_VERSION="1.27.1" OSX_VERSION="10.8" KNAME=$(uname -s) ARCH=$(uname -m) diff --git a/docs/security/advisories.md b/docs/security/advisories.md index 65dd32b2f..f57451523 100644 --- a/docs/security/advisories.md +++ b/docs/security/advisories.md @@ -41,7 +41,8 @@ The first Silo community release was cut from upstream history that already cont | `CVE-2026-34986` | `68e0ba997` | Upgrades `go-jose` to `v4.1.4`. | | `CVE-2026-39883` | `1869bd30b`, `e4fa06394` | Updates OpenTelemetry dependencies. | | Upstream Go security fixes | [Go 1.26.5](https://go.dev/doc/devel/release#go1.26.5) | Bumps the required toolchain to Go 1.26.5, which includes security fixes to `crypto/tls` and `os`. | -| Toolchain and dependency refresh | [Go 1.27.0](https://go.dev/doc/devel/release#go1.27) via [`43f4bb7ed`](https://github.com/pgsty/silo/commit/43f4bb7ed), [`edc8be6ed`](https://github.com/pgsty/silo/commit/edc8be6ed), [`4d6e1ea8e`](https://github.com/pgsty/silo/commit/4d6e1ea8e) | Moves the toolchain to Go 1.27.0 and refreshes the dependency stack (etcd client v3.7.1, `jwx` v3.0.13, `klauspost/compress` v1.19.2). The pre-release cleanup then returns to upstream `minio-go` (v7.3.1 pre-release) and retires the `silo-go` fork; `govulncheck` reports no reachable vulnerability on the release candidate. | +| Toolchain and dependency refresh | [Go 1.27.1](https://go.dev/doc/devel/release#go1.27.1) via [`43f4bb7ed`](https://github.com/pgsty/silo/commit/43f4bb7ed), [`edc8be6ed`](https://github.com/pgsty/silo/commit/edc8be6ed), [`4d6e1ea8e`](https://github.com/pgsty/silo/commit/4d6e1ea8e) | Moves the toolchain to Go 1.27 (1.27.1 as of the release) and refreshes the dependency stack (etcd client v3.7.1, `jwx` v3.0.13, `klauspost/compress` v1.19.2). The pre-release cleanup then returns to upstream `minio-go` (v7.3.1 pre-release) and retires the `silo-go` fork; `govulncheck` reports no reachable vulnerability on the release candidate. | +| [GO-2026-6354](https://pkg.go.dev/vuln/GO-2026-6354) / [GO-2026-6355](https://pkg.go.dev/vuln/GO-2026-6355) | `golang.org/x/crypto` `v0.56.0` ([`edf36bcbf`](https://github.com/pgsty/silo/commit/edf36bcbf)) | Updates `x/crypto/ssh` to the first fixed version for denial of service on deadlocked undecided and established channels. Reachable through the SFTP server (`startSFTPServer` → `sftp.Server.Listen` → `ssh.NewServerConn`); every earlier release that enables SFTP is affected. | | [GO-2026-6061](https://pkg.go.dev/vuln/GO-2026-6061) / [GHSA-hrxh-6v49-42gf](https://github.com/advisories/GHSA-hrxh-6v49-42gf) | gRPC `v1.82.1` | Updates gRPC to the first fixed version for vulnerabilities in the xDS RBAC authorization engine and HTTP/2 transport server. | | [GO-2026-5970](https://pkg.go.dev/vuln/GO-2026-5970) / `CVE-2026-56852` | `x/text` `v0.39.0` | Updates `x/text` to the first fixed version for an infinite loop on invalid input. | diff --git a/go.mod b/go.mod index ee2ee4a10..612d99660 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/minio/minio -go 1.27.0 +go 1.27.1 // Use Pigsty's SILO Console while preserving upstream import paths. The // pseudo-version pins the last commit of the v2.3.0 line before Console moved @@ -118,7 +118,7 @@ require ( go.uber.org/zap v1.28.0 go.yaml.in/yaml/v3 v3.0.5 goftp.io/server/v2 v2.0.3 - golang.org/x/crypto v0.55.0 + golang.org/x/crypto v0.56.0 golang.org/x/oauth2 v0.36.0 golang.org/x/sync v0.22.0 golang.org/x/sys v0.47.0 diff --git a/go.sum b/go.sum index 1bd597f81..04ef04bc1 100644 --- a/go.sum +++ b/go.sum @@ -739,8 +739,8 @@ golang.org/x/crypto v0.0.0-20211209193657-4570a0811e8b/go.mod h1:IxCIyHEi3zRg3s0 golang.org/x/crypto v0.0.0-20220315160706-3147a52a75dd/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= golang.org/x/crypto v0.5.0/go.mod h1:NK/OQwhpMQP3MwtdjgLlYHnH9ebylxKWv3e0fK+mkQU= golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58= -golang.org/x/crypto v0.55.0 h1:+KWHjbgOaAQ66dh/YlkZKHlz9ZUlq61AFirAR9ntP8M= -golang.org/x/crypto v0.55.0/go.mod h1:uq0V9dE/fzQuJtbnL+2EhWOE63vo164FY8xqEnV9xis= +golang.org/x/crypto v0.56.0 h1:GUh5Ii4J5jtcseSMiRqr1jXCNHoxjeV9Fmekc2oLy6Y= +golang.org/x/crypto v0.56.0/go.mod h1:OMW5y6CY9l38uPLmxU6l6pwcXp1obtLo3e6gT7gQR2I= golang.org/x/exp v0.0.0-20260820142414-ca536658362e h1:01Ju2A/fZKkci4zqx0eZxw//DnRYOnBiGJG14hFBhO8= golang.org/x/exp v0.0.0-20260820142414-ca536658362e/go.mod h1:zeBbvyFKDaLwa7CH/zI8KXt7gTl14SF7sO08Pl5jBCM= golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= diff --git a/internal/s3select/message.go b/internal/s3select/message.go index e2ed15945..7aa93c23c 100644 --- a/internal/s3select/message.go +++ b/internal/s3select/message.go @@ -295,7 +295,15 @@ func (writer *messageWriter) start() { select { case data := <-writer.errCh: quitFlag = true - // Flush collected records before sending error message + // A record accepted by SendRecord may still be queued when the + // error arrives, because select picks between the two channels + // at random. Stage it first so every record produced before the + // error precedes the error message instead of being dropped. + for len(writer.payloadCh) > 0 { + if !writer.stageRecord(<-writer.payloadCh) { + break + } + } if !writer.flushRecords() { break } @@ -316,23 +324,8 @@ func (writer *messageWriter) start() { break } writer.write(endMessage) - } else { - for payload.Len() > 0 { - copiedLen := copy(writer.payloadBuffer[writer.payloadBufferIndex:], payload.Bytes()) - writer.payloadBufferIndex += copiedLen - payload.Next(copiedLen) - - // If buffer is filled, flush it now! - freeSpace := bufLength - writer.payloadBufferIndex - if freeSpace == 0 { - if !writer.flushRecords() { - quitFlag = true - break - } - } - } - - bufPool.Put(payload) + } else if !writer.stageRecord(payload) { + quitFlag = true } case <-recordStagingTicker.C: @@ -368,6 +361,26 @@ func (writer *messageWriter) start() { } } +// stageRecord copies a record into the payload buffer, flushing whenever +// the buffer fills, and returns the buffer to the pool. It reports false when +// a flush failed. +func (writer *messageWriter) stageRecord(payload *bytes.Buffer) bool { + defer bufPool.Put(payload) + for payload.Len() > 0 { + copiedLen := copy(writer.payloadBuffer[writer.payloadBufferIndex:], payload.Bytes()) + writer.payloadBufferIndex += copiedLen + payload.Next(copiedLen) + + // If buffer is filled, flush it now! + if bufLength-writer.payloadBufferIndex == 0 { + if !writer.flushRecords() { + return false + } + } + } + return true +} + // Sends a single whole record. func (writer *messageWriter) SendRecord(payload *bytes.Buffer) error { select { diff --git a/internal/s3select/message_test.go b/internal/s3select/message_test.go new file mode 100644 index 000000000..e406ee04d --- /dev/null +++ b/internal/s3select/message_test.go @@ -0,0 +1,193 @@ +// Copyright (c) 2026 PGSTY +// +// This file is part of MinIO Object Storage stack +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package s3select + +import ( + "bytes" + "io" + "net/http" + "strings" + "testing" + + "github.com/minio/minio-go/v7" +) + +// decodeEvents parses a recorded event stream with the minio-go client and +// returns the concatenated record payloads and the terminal error, if any. +func decodeEvents(t *testing.T, response []byte) ([]byte, error) { + t.Helper() + body := &testResponseBody{Reader: bytes.NewReader(response), closed: make(chan struct{})} + res, err := minio.NewSelectResults(&http.Response{StatusCode: http.StatusOK, Body: body, ContentLength: int64(len(response))}, "testbucket") + if err != nil { + t.Fatal(err) + } + records, readErr := io.ReadAll(res) + <-body.closed + return records, readErr +} + +func newTestRecord(content string) *bytes.Buffer { + payload := bufPool.Get() + payload.Reset() + payload.WriteString(content) + return payload +} + +// eventOrder returns the byte offsets of the given markers in the response, +// or -1 for a marker that is absent. +func eventOrder(response []byte, markers ...string) []int { + offsets := make([]int, len(markers)) + for i, marker := range markers { + offsets[i] = bytes.Index(response, []byte(marker)) + } + return offsets +} + +func ascending(offsets []int) bool { + for i, offset := range offsets { + if offset < 0 || (i > 0 && offset <= offsets[i-1]) { + return false + } + } + return true +} + +// TestMessageWriterFlushesQueuedRecordBeforeError sends a record and an error +// back to back, so the writer goroutine sees both channels ready and picks +// one at random. The record must appear in the response before the error +// message every time. +func TestMessageWriterFlushesQueuedRecordBeforeError(t *testing.T) { + for i := range 200 { + w := &testResponseWriter{} + writer := newMessageWriter(w, nil) + payload := bufPool.Get() + payload.Reset() + payload.WriteString(`{"id":1}` + "\n") + if err := writer.SendRecord(payload); err != nil { + t.Fatalf("run %d: SendRecord: %v", i, err) + } + if err := writer.FinishWithError("OverMaxRecordSize", "too large"); err != nil { + t.Fatalf("run %d: FinishWithError: %v", i, err) + } + record := bytes.Index(w.response, []byte(`{"id":1}`)) + errMsg := bytes.Index(w.response, []byte("OverMaxRecordSize")) + if record < 0 || errMsg < 0 || record > errMsg { + t.Fatalf("run %d: record at %d, error at %d: queued record was dropped or reordered", i, record, errMsg) + } + } +} + +// TestMessageWriterFlushesBufferedAndQueuedRecordsBeforeError: one record has +// already been staged into the buffer and a second one is still queued when +// the error arrives. Both must precede the error, in order. +func TestMessageWriterFlushesBufferedAndQueuedRecordsBeforeError(t *testing.T) { + for i := range 100 { + w := &testResponseWriter{} + writer := newMessageWriter(w, nil) + if err := writer.SendRecord(newTestRecord(`{"id":1}` + "\n")); err != nil { + t.Fatal(err) + } + // Give the writer a chance to stage the first record; whether it + // did or not, the outcome must be the same. + if i%2 == 0 { + for range 100 { + if len(writer.payloadCh) == 0 { + break + } + } + } + if err := writer.SendRecord(newTestRecord(`{"id":2}` + "\n")); err != nil { + t.Fatal(err) + } + if err := writer.FinishWithError("OverMaxRecordSize", "too large"); err != nil { + t.Fatal(err) + } + if got := eventOrder(w.response, `{"id":1}`, `{"id":2}`, "OverMaxRecordSize"); !ascending(got) { + t.Fatalf("run %d: offsets %v: records must precede the error in order", i, got) + } + } +} + +// TestMessageWriterErrorWithoutRecords: an error with nothing queued writes +// only the error message, no empty Records event. +func TestMessageWriterErrorWithoutRecords(t *testing.T) { + w := &testResponseWriter{} + writer := newMessageWriter(w, nil) + if err := writer.FinishWithError("InternalError", "boom"); err != nil { + t.Fatal(err) + } + if !bytes.Contains(w.response, []byte("InternalError")) || bytes.Contains(w.response, []byte("Records")) { + t.Fatalf("unexpected response: %q", w.response) + } +} + +// TestMessageWriterStagesRecordsLargerThanTheBuffer: a record larger than the +// staging buffer is split across several Records events and nothing is lost +// or reordered, whether the stream ends with success or with an error. +func TestMessageWriterStagesRecordsLargerThanTheBuffer(t *testing.T) { + big := strings.Repeat("x", bufLength+bufLength/2) + want := "head\n" + big + "\n" + "tail\n" + for _, withError := range []bool{false, true} { + w := &testResponseWriter{} + writer := newMessageWriter(w, nil) + for _, rec := range []string{"head\n", big + "\n", "tail\n"} { + if err := writer.SendRecord(newTestRecord(rec)); err != nil { + t.Fatal(err) + } + } + if withError { + if err := writer.FinishWithError("OverMaxRecordSize", "too large"); err != nil { + t.Fatal(err) + } + } else if err := writer.Finish(10, 10); err != nil { + t.Fatal(err) + } + records, err := decodeEvents(t, w.response) + if string(records) != want { + t.Fatalf("withError=%v: got %d record bytes, want %d; head=%q", withError, len(records), len(want), string(records[:min(len(records), 8)])) + } + if withError && (err == nil || !strings.Contains(err.Error(), "OverMaxRecordSize")) { + t.Fatalf("expected OverMaxRecordSize after the records, got %v", err) + } + if !withError && err != nil { + t.Fatalf("unexpected error on the success path: %v", err) + } + } +} + +// TestMessageWriterSuccessOrder: the success path is unchanged: every record, +// then Stats, then End, and the client sees no error. +func TestMessageWriterSuccessOrder(t *testing.T) { + w := &testResponseWriter{} + writer := newMessageWriter(w, nil) + for _, rec := range []string{`{"id":1}`, `{"id":2}`} { + if err := writer.SendRecord(newTestRecord(rec + "\n")); err != nil { + t.Fatal(err) + } + } + if err := writer.Finish(20, 20); err != nil { + t.Fatal(err) + } + if got := eventOrder(w.response, `{"id":1}`, `{"id":2}`, "Stats", "End"); !ascending(got) { + t.Fatalf("offsets %v", got) + } + records, err := decodeEvents(t, w.response) + if err != nil || string(records) != "{\"id\":1}\n{\"id\":2}\n" { + t.Fatalf("records %q err %v", records, err) + } +}