mirror of
https://github.com/pgsty/minio.git
synced 2026-09-05 18:16:16 +03:00
Merge branch 'codex/ci-fixes': CI repairs for the 20260903 release
Flush S3 Select records queued before an error message (an inherited select race that the CVE-2026-39414 regression test caught under the race detector in CI), update golang.org/x/crypto to v0.56.0 for GO-2026-6354/6355 reachable through the SFTP server, move the toolchain to Go 1.27.1, and extend the message writer tests. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QwirCh4nsrJCQp2WaXoVLK Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
@@ -1,4 +1,4 @@
|
||||
FROM golang:1.27.0-alpine AS build
|
||||
FROM golang:1.27.1-alpine AS build
|
||||
|
||||
ARG TARGETARCH
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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. |
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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 <http://www.gnu.org/licenses/>.
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user