Merge branch 'main' into feat-add-checksumtype-completemultipartupload-response

This commit is contained in:
Feng Ruohang
2026-08-26 10:15:15 +08:00
committed by GitHub
47 changed files with 2933 additions and 260 deletions
+3 -3
View File
@@ -859,7 +859,7 @@ func (a adminAPIHandlers) UpdateServiceAccount(w http.ResponseWriter, r *http.Re
var sp *policy.Policy
if len(updateReq.NewPolicy) > 0 {
sp, err = policy.ParseConfig(bytes.NewReader(updateReq.NewPolicy))
sp, err = policy.ParseConfigStrict(bytes.NewReader(updateReq.NewPolicy))
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
@@ -1729,7 +1729,7 @@ func (a adminAPIHandlers) AddCannedPolicy(w http.ResponseWriter, r *http.Request
return
}
iamPolicy, err := policy.ParseConfig(bytes.NewReader(iamPolicyBytes))
iamPolicy, err := policy.ParseConfigStrict(bytes.NewReader(iamPolicyBytes))
if err != nil {
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
return
@@ -2981,7 +2981,7 @@ func commonAddServiceAccount(r *http.Request, ldap bool) (context.Context, auth.
var sp *policy.Policy
if len(createReq.Policy) > 0 {
sp, err = policy.ParseConfig(bytes.NewReader(createReq.Policy))
sp, err = policy.ParseConfigStrict(bytes.NewReader(createReq.Policy))
if err != nil {
return ctx, auth.Credentials{}, newServiceAccountOpts{}, madmin.AddServiceAccountReq{}, "", toAdminAPIErr(ctx, err)
}
+60
View File
@@ -204,6 +204,7 @@ func TestIAMInternalIDPServerSuite(t *testing.T) {
suite.TestUserCreate(c)
suite.TestUserPolicyEscalationBug(c)
suite.TestPolicyCreate(c)
suite.TestServiceAccountBareARNPolicyRejected(c)
suite.TestCannedPolicies(c)
suite.TestGroupAddRemove(c)
suite.TestServiceAccountOpsByAdmin(c)
@@ -600,6 +601,20 @@ func (s *TestSuiteIAM) TestPolicyCreate(c *check) {
c.Fatalf("invalid policy creation success")
}
for i, resource := range []string{"arn:aws:s3:::", "*arn:aws:s3:::"} {
barePolicyBytes := fmt.Appendf(nil, `{
"Version": "2012-10-17",
"Statement": [{
"Effect": "Deny",
"Action": ["s3:GetObject"],
"Resource": ["%s"]
}]
}`, resource)
if err = s.adm.AddCannedPolicy(ctx, fmt.Sprintf("%s-bare-%d", policy, i), barePolicyBytes); err == nil {
c.Fatalf("bare ARN policy creation succeeded for %q", resource)
}
}
// 3. Create a user, associate policy and verify access
accessKey, secretKey := mustGenerateCredentials(c)
err = s.adm.SetUser(ctx, accessKey, secretKey, madmin.AccountEnabled)
@@ -653,6 +668,51 @@ func (s *TestSuiteIAM) TestPolicyCreate(c *check) {
}
}
func (s *TestSuiteIAM) TestServiceAccountBareARNPolicyRejected(c *check) {
ctx, cancel := context.WithTimeout(context.Background(), testDefaultTimeout)
defer cancel()
barePolicy := []byte(`{
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Action": ["s3:GetObject"],
"NotResource": ["arn:aws:s3:::"]
}]
}`)
if _, err := s.adm.AddServiceAccount(ctx, madmin.AddServiceAccountReq{
TargetUser: globalActiveCred.AccessKey,
Policy: barePolicy,
}); err == nil {
c.Fatal("service account creation accepted a bare ARN policy")
}
validPolicy := []byte(`{
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Action": ["s3:GetObject"],
"Resource": ["arn:aws:s3:::*"]
}]
}`)
credentials, err := s.adm.AddServiceAccount(ctx, madmin.AddServiceAccountReq{
TargetUser: globalActiveCred.AccessKey,
Policy: validPolicy,
})
if err != nil {
c.Fatalf("service account creation rejected an explicit resource: %v", err)
}
defer func() {
_ = s.adm.DeleteServiceAccount(ctx, credentials.AccessKey)
}()
if err = s.adm.UpdateServiceAccount(ctx, credentials.AccessKey, madmin.UpdateServiceAccountReq{
NewPolicy: barePolicy,
}); err == nil {
c.Fatal("service account update accepted a bare ARN policy")
}
}
func (s *TestSuiteIAM) TestCannedPolicies(c *check) {
ctx, cancel := context.WithTimeout(context.Background(), testDefaultTimeout)
defer cancel()
+33 -9
View File
@@ -27,7 +27,6 @@ import (
"path"
"strconv"
"strings"
"time"
"github.com/minio/minio/internal/amztime"
"github.com/minio/minio/internal/crypto"
@@ -380,6 +379,13 @@ type CopyObjectResponse struct {
XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ CopyObjectResult" json:"-"`
LastModified string // time string of format "2006-01-02T15:04:05.000Z"
ETag string // md5sum of the copied object.
ChecksumCRC32 string `xml:",omitempty"`
ChecksumCRC32C string `xml:",omitempty"`
ChecksumSHA1 string `xml:",omitempty"`
ChecksumSHA256 string `xml:",omitempty"`
ChecksumCRC64NVME string `xml:",omitempty"`
ChecksumType string `xml:",omitempty"`
}
// CopyObjectPartResponse container returns ETag and LastModified of the successfully copied object
@@ -387,6 +393,12 @@ type CopyObjectPartResponse struct {
XMLName xml.Name `xml:"http://s3.amazonaws.com/doc/2006-03-01/ CopyPartResult" json:"-"`
LastModified string // time string of format "2006-01-02T15:04:05.000Z"
ETag string // md5sum of the copied object part.
ChecksumCRC32 string `xml:",omitempty"`
ChecksumCRC32C string `xml:",omitempty"`
ChecksumSHA1 string `xml:",omitempty"`
ChecksumSHA256 string `xml:",omitempty"`
ChecksumCRC64NVME string `xml:",omitempty"`
}
// Initiator inherit from Owner struct, fields are same
@@ -764,19 +776,31 @@ func generateListObjectsV2Response(ctx context.Context, bucket, prefix, token, n
type metaCheckFn = func(name string, action policy.Action) (s3Err APIErrorCode)
// generates CopyObjectResponse from etag and lastModified time.
func generateCopyObjectResponse(etag string, lastModified time.Time) CopyObjectResponse {
// generates CopyObjectResponse from the committed object information.
func generateCopyObjectResponse(oi ObjectInfo, h http.Header) CopyObjectResponse {
cs, _ := oi.decryptChecksums(0, h)
return CopyObjectResponse{
ETag: "\"" + etag + "\"",
LastModified: amztime.ISO8601Format(lastModified.UTC()),
ETag: "\"" + oi.ETag + "\"",
LastModified: amztime.ISO8601Format(oi.ModTime.UTC()),
ChecksumCRC32: cs[hash.ChecksumCRC32.String()],
ChecksumCRC32C: cs[hash.ChecksumCRC32C.String()],
ChecksumSHA1: cs[hash.ChecksumSHA1.String()],
ChecksumSHA256: cs[hash.ChecksumSHA256.String()],
ChecksumCRC64NVME: cs[hash.ChecksumCRC64NVME.String()],
ChecksumType: cs[xhttp.AmzChecksumType],
}
}
// generates CopyObjectPartResponse from etag and lastModified time.
func generateCopyObjectPartResponse(etag string, lastModified time.Time) CopyObjectPartResponse {
// generates CopyObjectPartResponse from the uploaded part information.
func generateCopyObjectPartResponse(partInfo PartInfo) CopyObjectPartResponse {
return CopyObjectPartResponse{
ETag: "\"" + etag + "\"",
LastModified: amztime.ISO8601Format(lastModified.UTC()),
ETag: "\"" + partInfo.ETag + "\"",
LastModified: amztime.ISO8601Format(partInfo.LastModified.UTC()),
ChecksumCRC32: partInfo.ChecksumCRC32,
ChecksumCRC32C: partInfo.ChecksumCRC32C,
ChecksumSHA1: partInfo.ChecksumSHA1,
ChecksumSHA256: partInfo.ChecksumSHA256,
ChecksumCRC64NVME: partInfo.ChecksumCRC64NVME,
}
}
+2 -2
View File
@@ -198,7 +198,7 @@ func enforceRetentionBypassForPut(ctx context.Context, r *http.Request, oi Objec
byPassSet, r, cred, owner)
// Governance mode retention period cannot be shortened, if x-amz-bypass-governance is not set.
if !byPassSet {
if objRetention.Mode != objectlock.RetGovernance || objRetention.RetainUntilDate.Before((ret.RetainUntilDate.Time)) {
if objRetention.Mode != objectlock.RetGovernance || objRetention.RetainUntilDate.Before(ret.RetainUntilDate.Time) {
return ObjectLocked{Bucket: oi.Bucket, Object: oi.Name, VersionID: oi.VersionID}
}
}
@@ -209,7 +209,7 @@ func enforceRetentionBypassForPut(ctx context.Context, r *http.Request, oi Objec
case objectlock.RetCompliance:
// Compliance retention mode cannot be changed or shortened.
// https://docs.aws.amazon.com/AmazonS3/latest/dev/object-lock-overview.html#object-lock-retention-modes
if objRetention.Mode != objectlock.RetCompliance || objRetention.RetainUntilDate.Before((ret.RetainUntilDate.Time)) {
if objRetention.Mode != objectlock.RetCompliance || objRetention.RetainUntilDate.Before(ret.RetainUntilDate.Time) {
return ObjectLocked{Bucket: oi.Bucket, Object: oi.Name, VersionID: oi.VersionID}
}
apiErr := isPutRetentionAllowed(oi.Bucket, oi.Name,
+48 -8
View File
@@ -36,6 +36,7 @@ import (
"strings"
"syscall"
"time"
"unicode"
"github.com/dustin/go-humanize"
fcolor "github.com/fatih/color"
@@ -540,6 +541,34 @@ func (e envKV) String() string {
return fmt.Sprintf("%s=%s", e.Key, e.Value)
}
func isValidEnvName(name string) bool {
if name == "" || !isEnvNameStart(name[0]) {
return false
}
for i := 1; i < len(name); i++ {
if !isEnvNameStart(name[i]) && (name[i] < '0' || name[i] > '9') {
return false
}
}
return true
}
func isEnvNameStart(ch byte) bool {
return ch == '_' || ch >= 'A' && ch <= 'Z' || ch >= 'a' && ch <= 'z'
}
func trimExportPrefix(envEntry string) string {
rest, ok := strings.CutPrefix(envEntry, "export")
if !ok || rest == "" {
return envEntry
}
trimmed := strings.TrimLeftFunc(rest, unicode.IsSpace)
if len(trimmed) == len(rest) {
return envEntry
}
return trimmed
}
func parsEnvEntry(envEntry string) (envKV, error) {
envEntry = strings.TrimSpace(envEntry)
if envEntry == "" {
@@ -554,13 +583,19 @@ func parsEnvEntry(envEntry string) (envKV, error) {
Skip: true,
}, nil
}
envTokens := strings.SplitN(strings.TrimSpace(strings.TrimPrefix(envEntry, "export")), config.EnvSeparator, 2)
envTokens := strings.SplitN(trimExportPrefix(envEntry), config.EnvSeparator, 2)
if len(envTokens) != 2 {
return envKV{}, fmt.Errorf("envEntry malformed; %s, expected to be of form 'KEY=value'", envEntry)
return envKV{}, errors.New("missing '='")
}
key := envTokens[0]
val := envTokens[1]
key := strings.TrimSpace(envTokens[0])
val := strings.TrimSpace(envTokens[1])
if !isValidEnvName(key) {
return envKV{}, fmt.Errorf("invalid environment variable name %q", key)
}
if strings.IndexByte(val, 0) >= 0 {
return envKV{}, errors.New("environment variable value contains NUL")
}
// Remove quotes from the value if found
if len(val) >= 2 {
@@ -587,10 +622,12 @@ func minioEnvironFromFile(envConfigFile string) ([]envKV, error) {
defer f.Close()
var ekvs []envKV
scanner := bufio.NewScanner(f)
lineNo := 0
for scanner.Scan() {
lineNo++
ekv, err := parsEnvEntry(scanner.Text())
if err != nil {
return nil, err
return nil, fmt.Errorf("%s:%d: %w", envConfigFile, lineNo, err)
}
if ekv.Skip {
// Skips empty lines
@@ -599,7 +636,7 @@ func minioEnvironFromFile(envConfigFile string) ([]envKV, error) {
ekvs = append(ekvs, ekv)
}
if err = scanner.Err(); err != nil {
return nil, err
return nil, fmt.Errorf("%s: %w", envConfigFile, err)
}
return ekvs, nil
}
@@ -666,12 +703,15 @@ func loadEnvVarsFromFiles() {
}
if env.IsSet(config.EnvConfigEnvFile) {
ekvs, err := minioEnvironFromFile(env.Get(config.EnvConfigEnvFile, ""))
envConfigFile := env.Get(config.EnvConfigEnvFile, "")
ekvs, err := minioEnvironFromFile(envConfigFile)
if err != nil && !os.IsNotExist(err) {
logger.Fatal(err, "Unable to read the config environment file")
}
for _, ekv := range ekvs {
os.Setenv(ekv.Key, ekv.Value)
if err := os.Setenv(ekv.Key, ekv.Value); err != nil {
logger.Fatal(err, "Unable to set %s from config environment file %s", ekv.Key, envConfigFile)
}
}
}
}
+161
View File
@@ -19,8 +19,10 @@ package cmd
import (
"errors"
"fmt"
"os"
"reflect"
"strings"
"testing"
)
@@ -181,3 +183,162 @@ MINIO_ROOT_PASSWORD=minio123`,
})
}
}
func Test_minioEnvironFromFileWhitespaceAndValidation(t *testing.T) {
testCases := []struct {
name string
content string
want []envKV
errLine int
errContains string
errExcludes string
}{
{
name: "spaces and tabs around separator",
content: "MINIO_ROOT_USER = minio\nMINIO_ROOT_PASSWORD\t=\tminio123",
want: []envKV{
{Key: "MINIO_ROOT_USER", Value: "minio"},
{Key: "MINIO_ROOT_PASSWORD", Value: "minio123"},
},
},
{
name: "export tab and quoted spaces",
content: "export\tMINIO_ROOT_USER = \" minio user \"\nexport MINIO_ROOT_PASSWORD = ' minio secret '",
want: []envKV{
{Key: "MINIO_ROOT_USER", Value: " minio user "},
{Key: "MINIO_ROOT_PASSWORD", Value: " minio secret "},
},
},
{
name: "export Unicode whitespace",
content: "export\u00a0MINIO_ROOT_USER=value",
want: []envKV{
{Key: "MINIO_ROOT_USER", Value: "value"},
},
},
{
name: "export is only a standalone prefix",
content: "export=value\nexportFOO=bar",
want: []envKV{
{Key: "export", Value: "value"},
{Key: "exportFOO", Value: "bar"},
},
},
{
name: "unquoted whitespace empty value and additional separators",
content: "UNQUOTED = value \nEMPTY =\nTOKEN = scheme://user:password@example.com?a=b",
want: []envKV{
{Key: "UNQUOTED", Value: "value"},
{Key: "EMPTY", Value: ""},
{Key: "TOKEN", Value: "scheme://user:password@example.com?a=b"},
},
},
{
name: "valid underscore and digits",
content: "_VALID_2=value",
want: []envKV{
{Key: "_VALID_2", Value: "value"},
},
},
{
name: "missing separator redacts the line",
content: "MINIO_ROOT_PASSWORD=valid\nsuper-secret-without-equals",
errLine: 2,
errContains: "missing '='",
errExcludes: "super-secret-without-equals",
},
{
name: "empty name",
content: "=empty-name-secret",
errLine: 1,
errContains: `invalid environment variable name ""`,
errExcludes: "empty-name-secret",
},
{
name: "digit leading name",
content: "1MINIO_ROOT_USER=digit-leading-secret",
errLine: 1,
errContains: `invalid environment variable name "1MINIO_ROOT_USER"`,
errExcludes: "digit-leading-secret",
},
{
name: "hyphenated name",
content: "MINIO-ROOT-USER=hyphen-secret",
errLine: 1,
errContains: `invalid environment variable name "MINIO-ROOT-USER"`,
errExcludes: "hyphen-secret",
},
{
name: "whitespace in name",
content: "MINIO ROOT USER=whitespace-secret",
errLine: 1,
errContains: `invalid environment variable name "MINIO ROOT USER"`,
errExcludes: "whitespace-secret",
},
{
name: "NUL in name",
content: "MINIO\x00ROOT=nul-name-secret",
errLine: 1,
errContains: "invalid environment variable name",
errExcludes: "nul-name-secret",
},
{
name: "NUL in value",
content: "MINIO_ROOT_USER=before\x00nul-value-secret",
errLine: 1,
errContains: "environment variable value contains NUL",
errExcludes: "nul-value-secret",
},
{
name: "diagnostic has file and line but no value",
content: "MINIO_ROOT_USER=valid\nBAD KEY=super-secret-value",
errLine: 2,
errContains: `invalid environment variable name "BAD KEY"`,
errExcludes: "super-secret-value",
},
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
tmpfile, err := os.CreateTemp(t.TempDir(), "testfile")
if err != nil {
t.Fatal(err)
}
if _, err = tmpfile.WriteString(testCase.content); err != nil {
t.Fatal(err)
}
if err = tmpfile.Close(); err != nil {
t.Fatal(err)
}
got, err := minioEnvironFromFile(tmpfile.Name())
if testCase.errContains == "" {
if err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(got, testCase.want) {
t.Errorf("expected %v, got %v", testCase.want, got)
}
return
}
if err == nil {
t.Fatal("expected an error")
}
errText := err.Error()
location := fmt.Sprintf("%s:%d:", tmpfile.Name(), testCase.errLine)
if !strings.Contains(errText, location) {
t.Errorf("expected error to contain %q, got %q", location, errText)
}
if !strings.Contains(errText, testCase.errContains) {
t.Errorf("expected error to contain %q, got %q", testCase.errContains, errText)
}
if testCase.errExcludes != "" && strings.Contains(errText, testCase.errExcludes) {
t.Errorf("expected error to redact %q, got %q", testCase.errExcludes, errText)
}
if got != nil {
t.Errorf("expected no entries on parse error, got %v", got)
}
})
}
}
+6 -2
View File
@@ -167,7 +167,9 @@ func readConfigWithoutMigrate(ctx context.Context, objAPI ObjectLayer) (config.C
notify.SetNotifyMQTT(newCfg, k, args)
}
for k, args := range cfg.Notify.MySQL {
notify.SetNotifyMySQL(newCfg, k, args)
if err := notify.SetNotifyMySQL(newCfg, k, args); err != nil {
return nil, err
}
}
for k, args := range cfg.Notify.NATS {
notify.SetNotifyNATS(newCfg, k, args)
@@ -176,7 +178,9 @@ func readConfigWithoutMigrate(ctx context.Context, objAPI ObjectLayer) (config.C
notify.SetNotifyNSQ(newCfg, k, args)
}
for k, args := range cfg.Notify.PostgreSQL {
notify.SetNotifyPostgres(newCfg, k, args)
if err := notify.SetNotifyPostgres(newCfg, k, args); err != nil {
return nil, err
}
}
for k, args := range cfg.Notify.Redis {
notify.SetNotifyRedis(newCfg, k, args)
+264
View File
@@ -0,0 +1,264 @@
// Copyright (c) 2015-2026 MinIO, Inc.
//
// 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 cmd
import (
"bytes"
"context"
"encoding/json"
"errors"
"os"
"reflect"
"strings"
"testing"
"github.com/minio/minio/internal/config"
"github.com/minio/minio/internal/config/notify"
"github.com/minio/minio/internal/event/target"
)
func installLegacyConfigFile(t *testing.T, configure func(*serverConfigV33)) (string, []byte) {
t.Helper()
cfg := &serverConfigV33{
Version: "33",
Notify: notify.NewConfig(),
}
configure(cfg)
data, err := json.Marshal(cfg)
if err != nil {
t.Fatal(err)
}
oldConfigDir := globalConfigDir
globalConfigDir = &ConfigDir{path: t.TempDir()}
t.Cleanup(func() { globalConfigDir = oldConfigDir })
configFile := getConfigFile()
if err = os.WriteFile(configFile, data, 0o600); err != nil {
t.Fatal(err)
}
return configFile, data
}
func assertLegacyMigrationError(t *testing.T, err error, subsystem, name, key, secret string) {
t.Helper()
var targetErr *notify.LegacyDatabaseTargetError
if !errors.As(err, &targetErr) {
t.Fatalf("error = %v, want *notify.LegacyDatabaseTargetError", err)
}
msg := err.Error()
for _, want := range []string{subsystem + config.SubSystemSeparator + name, key} {
if !strings.Contains(msg, want) {
t.Errorf("error %q does not contain %q", msg, want)
}
}
if strings.Contains(msg, secret) {
t.Errorf("migration error leaks database password %q: %s", secret, msg)
}
}
func TestReadConfigWithoutMigrateRejectsLegacyDatabaseTargets(t *testing.T) {
tests := []struct {
name string
subsystem string
key string
secret string
configure func(*serverConfigV33)
}{
{
name: "postgres",
subsystem: config.NotifyPostgresSubSys,
key: target.PostgresConnectionString,
secret: "postgres-migration-secret",
configure: func(cfg *serverConfigV33) {
cfg.Notify.PostgreSQL["archive"] = target.PostgreSQLArgs{
Enable: true,
Port: "5432",
Username: "legacy-user",
Password: "postgres-migration-secret",
Database: "events",
}
},
},
{
name: "mysql",
subsystem: config.NotifyMySQLSubSys,
key: target.MySQLDSNString,
secret: "mysql-migration-secret",
configure: func(cfg *serverConfigV33) {
cfg.Notify.MySQL["archive"] = target.MySQLArgs{
Enable: true,
Port: "3306",
User: "legacy-user",
Password: "mysql-migration-secret",
Database: "events",
}
},
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
configFile, original := installLegacyConfigFile(t, test.configure)
got, err := readConfigWithoutMigrate(t.Context(), nil)
if got != nil {
t.Fatalf("config = %v, want nil on failed migration", got)
}
assertLegacyMigrationError(t, err, test.subsystem, "archive", test.key, test.secret)
after, readErr := os.ReadFile(configFile)
if readErr != nil {
t.Fatal(readErr)
}
if !bytes.Equal(after, original) {
t.Fatal("failed migration rewrote the legacy source config")
}
if _, statErr := os.Stat(configFile + ".old"); !errors.Is(statErr, os.ErrNotExist) {
t.Fatalf("failed migration created a backup/persistence artifact: %v", statErr)
}
})
}
}
func TestReadConfigWithoutMigrateMigratesCanonicalDatabaseTargets(t *testing.T) {
const (
postgresConnection = "host=postgres.example port=5432 dbname=events user=app password=secret sslmode=disable"
mysqlDSN = "app:secret@tcp(mysql.example:3306)/events?parseTime=true"
discardedLegacyValue = "discarded-legacy-value"
)
installLegacyConfigFile(t, func(cfg *serverConfigV33) {
cfg.Notify.PostgreSQL["archive"] = target.PostgreSQLArgs{
Enable: true,
Format: "namespace",
ConnectionString: postgresConnection,
Table: "events",
Port: discardedLegacyValue,
Username: discardedLegacyValue,
Password: discardedLegacyValue,
Database: discardedLegacyValue,
}
cfg.Notify.MySQL["archive"] = target.MySQLArgs{
Enable: true,
Format: "namespace",
DSN: mysqlDSN,
Table: "events",
Port: discardedLegacyValue,
User: discardedLegacyValue,
Password: discardedLegacyValue,
Database: discardedLegacyValue,
}
})
got, err := readConfigWithoutMigrate(t.Context(), nil)
if err != nil {
t.Fatalf("readConfigWithoutMigrate: %v", err)
}
tests := []struct {
subsystem string
key string
want string
discarded string
}{
{config.NotifyPostgresSubSys, target.PostgresConnectionString, postgresConnection, discardedLegacyValue},
{config.NotifyMySQLSubSys, target.MySQLDSNString, mysqlDSN, discardedLegacyValue},
}
for _, test := range tests {
kvs := got[test.subsystem]["archive"]
if value := kvs.Get(test.key); value != test.want {
t.Errorf("%s %s = %q, want %q", test.subsystem, test.key, value, test.want)
}
if err := config.CheckValidKeys(test.subsystem+config.SubSystemSeparator+"archive", kvs, notify.DefaultNotificationKVS[test.subsystem]); err != nil {
t.Errorf("migrated %s target failed key validation: %v", test.subsystem, err)
}
for _, key := range []string{"host", "port", "username", "password", "database"} {
if _, ok := kvs.Lookup(key); ok {
t.Errorf("migrated %s target contains legacy key %q", test.subsystem, key)
}
}
for _, kv := range kvs {
if strings.Contains(kv.Value, test.discarded) {
t.Errorf("migrated %s target contains discarded legacy value in %q", test.subsystem, kv.Key)
}
}
}
postgresTargets, err := notify.GetNotifyPostgres(got[config.NotifyPostgresSubSys])
if err != nil {
t.Fatalf("GetNotifyPostgres: %v", err)
}
if value := postgresTargets["archive"].ConnectionString; value != postgresConnection {
t.Errorf("Postgres connection string = %q, want %q", value, postgresConnection)
}
mysqlTargets, err := notify.GetNotifyMySQL(got[config.NotifyMySQLSubSys])
if err != nil {
t.Fatalf("GetNotifyMySQL: %v", err)
}
if value := mysqlTargets["archive"].DSN; value != mysqlDSN {
t.Errorf("MySQL DSN = %q, want %q", value, mysqlDSN)
}
}
func TestInitConfigSubsystemReturnsLegacyDatabaseTargetError(t *testing.T) {
obj, fsDir, err := prepareFS(t.Context())
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
_ = obj.Shutdown(context.Background())
_ = os.RemoveAll(fsDir)
})
const secret = "startup-migration-secret"
installLegacyConfigFile(t, func(cfg *serverConfigV33) {
cfg.Notify.PostgreSQL["archive"] = target.PostgreSQLArgs{
Enable: true,
Port: "5432",
Username: "legacy-user",
Password: secret,
Database: "events",
}
})
globalServerConfigMu.RLock()
var before config.Config
if globalServerConfig != nil {
before = globalServerConfig.Clone()
}
globalServerConfigMu.RUnlock()
err = initConfigSubsystem(t.Context(), obj)
assertLegacyMigrationError(t, err, config.NotifyPostgresSubSys, "archive", target.PostgresConnectionString, secret)
if configRetriableErrors(err) {
t.Fatal("legacy database migration error must be startup-fatal, not retriable")
}
if !fatalServerConfigError(err) {
t.Fatal("legacy database migration error must abort server startup")
}
globalServerConfigMu.RLock()
var after config.Config
if globalServerConfig != nil {
after = globalServerConfig.Clone()
}
globalServerConfigMu.RUnlock()
if !reflect.DeepEqual(after, before) {
t.Fatal("failed migration activated a partial server configuration")
}
}
+1 -1
View File
@@ -326,7 +326,7 @@ func (h dataUsageHash) modAlt(cycle uint32, cycles uint32) bool {
if cycles <= 1 {
return cycles == 1
}
return uint32(xxhash.Sum64String(string(h))>>32)%(cycles) == cycle%cycles
return uint32(xxhash.Sum64String(string(h))>>32)%cycles == cycle%cycles
}
// addChild will add a child based on its hash.
@@ -0,0 +1,596 @@
// Copyright (c) 2015-2026 MinIO, Inc.
//
// 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 cmd
import (
"bytes"
"crypto/md5"
"encoding/base64"
"encoding/xml"
"net/http"
"net/http/httptest"
"strconv"
"testing"
"github.com/minio/minio/internal/auth"
"github.com/minio/minio/internal/hash"
xhttp "github.com/minio/minio/internal/http"
"github.com/minio/minio/internal/kms"
)
func uploadPartHTTP(t *testing.T, apiRouter http.Handler, creds auth.Credentials,
bucket, object, uploadID string, partNumber int, data []byte, headers map[string]string,
) (string, *httptest.ResponseRecorder) {
t.Helper()
req, err := newTestSignedRequestV4(http.MethodPut,
getPutObjectPartURL("", bucket, object, uploadID, strconv.Itoa(partNumber)),
int64(len(data)), bytes.NewReader(data), creds.AccessKey, creds.SecretKey, headers)
if err != nil {
t.Fatalf("failed to build UploadPart request: %v", err)
}
rec := httptest.NewRecorder()
apiRouter.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("UploadPart failed: %d %s", rec.Code, rec.Body.String())
}
return canonicalizeETag(rec.Header()[xhttp.ETag][0]), rec
}
func listPartsHTTP(t *testing.T, apiRouter http.Handler, creds auth.Credentials,
bucket, object, uploadID string, headers map[string]string,
) ListPartsResponse {
t.Helper()
req, err := newTestSignedRequestV4(http.MethodGet,
getListMultipartURLWithParams("", bucket, object, uploadID, "1000", "", ""),
0, nil, creds.AccessKey, creds.SecretKey, headers)
if err != nil {
t.Fatalf("failed to build ListParts request: %v", err)
}
rec := httptest.NewRecorder()
apiRouter.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("ListParts failed: %d %s", rec.Code, rec.Body.String())
}
var response ListPartsResponse
if err := xml.Unmarshal(rec.Body.Bytes(), &response); err != nil {
t.Fatalf("failed to decode ListParts response: %v", err)
}
return response
}
func partChecksum(typ hash.ChecksumType, part Part) string {
switch typ.Base() {
case hash.ChecksumCRC32:
return part.ChecksumCRC32
case hash.ChecksumCRC32C:
return part.ChecksumCRC32C
case hash.ChecksumSHA1:
return part.ChecksumSHA1
case hash.ChecksumSHA256:
return part.ChecksumSHA256
case hash.ChecksumCRC64NVME:
return part.ChecksumCRC64NVME
default:
return ""
}
}
func copyPartChecksum(typ hash.ChecksumType, response CopyObjectPartResponse) string {
switch typ.Base() {
case hash.ChecksumCRC32:
return response.ChecksumCRC32
case hash.ChecksumCRC32C:
return response.ChecksumCRC32C
case hash.ChecksumSHA1:
return response.ChecksumSHA1
case hash.ChecksumSHA256:
return response.ChecksumSHA256
case hash.ChecksumCRC64NVME:
return response.ChecksumCRC64NVME
default:
return ""
}
}
func completePartWithChecksum(typ hash.ChecksumType, partNumber int, etag, checksum string) CompletePart {
part := CompletePart{PartNumber: partNumber, ETag: etag}
switch typ.Base() {
case hash.ChecksumCRC32:
part.ChecksumCRC32 = checksum
case hash.ChecksumCRC32C:
part.ChecksumCRC32C = checksum
case hash.ChecksumSHA1:
part.ChecksumSHA1 = checksum
case hash.ChecksumSHA256:
part.ChecksumSHA256 = checksum
case hash.ChecksumCRC64NVME:
part.ChecksumCRC64NVME = checksum
}
return part
}
func completePartsHTTP(t *testing.T, apiRouter http.Handler, creds auth.Credentials,
bucket, object, uploadID string, parts []CompletePart, headers map[string]string,
) *httptest.ResponseRecorder {
t.Helper()
body, err := xml.Marshal(CompleteMultipartUpload{Parts: parts})
if err != nil {
t.Fatalf("failed to encode CompleteMultipartUpload request: %v", err)
}
req, err := newTestSignedRequestV4(http.MethodPost,
getCompleteMultipartUploadURL("", bucket, object, uploadID),
int64(len(body)), bytes.NewReader(body), creds.AccessKey, creds.SecretKey, headers)
if err != nil {
t.Fatalf("failed to build CompleteMultipartUpload request: %v", err)
}
rec := httptest.NewRecorder()
apiRouter.ServeHTTP(rec, req)
return rec
}
func copyPartWithoutChecksumHTTP(t *testing.T, apiRouter http.Handler, creds auth.Credentials,
bucket, source, object, uploadID, sourceRange string, headers map[string]string,
) CopyObjectPartResponse {
t.Helper()
req, err := newTestSignedRequestV4(http.MethodPut,
getCopyObjectPartURL("", bucket, object, uploadID, "1"),
0, nil, creds.AccessKey, creds.SecretKey, headers)
if err != nil {
t.Fatalf("failed to build UploadPartCopy request: %v", err)
}
req.Header.Set(xhttp.AmzCopySource, SlashSeparator+pathJoin(bucket, source))
if sourceRange != "" {
req.Header.Set(xhttp.AmzCopySourceRange, sourceRange)
}
rec := httptest.NewRecorder()
apiRouter.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("UploadPartCopy failed: %d %s", rec.Code, rec.Body.String())
}
var response CopyObjectPartResponse
if err := xml.Unmarshal(rec.Body.Bytes(), &response); err != nil {
t.Fatalf("failed to decode UploadPartCopy response: %v", err)
}
return response
}
// TestAPIUploadPartServerSideChecksum exercises the data transformations that
// made installing a checksum hasher in the object layer unsafe. The checksum
// must always cover logical plaintext, regardless of compression or encryption.
func TestAPIUploadPartServerSideChecksum(t *testing.T) {
defer DetectTestLeak(t)()
ExecExtendedObjectLayerAPITest(t, testAPIUploadPartServerSideChecksum,
[]string{"CopyObjectPart", "PutObjectPart", "NewMultipart", "ListObjectParts", "CompleteMultipart"})
}
func testAPIUploadPartServerSideChecksum(obj ObjectLayer, instanceType, bucketName string, apiRouter http.Handler,
credentials auth.Credentials, t *testing.T,
) {
typ := hash.ChecksumCRC32
data := bytes.Repeat([]byte("multipart-checksum-plaintext-"), 48*1024)
want := mustChecksum(t, typ, data)
t.Run("upload", func(t *testing.T) {
object := "checksums/upload"
uploadID := newMultipartUploadHTTP(t, apiRouter, credentials, bucketName, object,
typ.String(), xhttp.AmzChecksumTypeFullObject)
etag, rec := uploadPartHTTP(t, apiRouter, credentials,
bucketName, object, uploadID, 1, data, nil)
if got := rec.Header().Get(typ.Key()); got != "" {
t.Fatalf("%s: UploadPart returned server-computed checksum %q", instanceType, got)
}
listed := listPartsHTTP(t, apiRouter, credentials, bucketName, object, uploadID, nil)
if len(listed.Parts) != 1 || partChecksum(typ, listed.Parts[0]) != want {
t.Fatalf("%s: ListParts checksum mismatch: %+v, want %q", instanceType, listed.Parts, want)
}
rec = completePartsHTTP(t, apiRouter, credentials, bucketName, object, uploadID,
[]CompletePart{{PartNumber: 1, ETag: etag}}, nil)
if rec.Code != http.StatusOK {
t.Fatalf("%s: CompleteMultipartUpload failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
oi, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{})
if err != nil {
t.Fatalf("%s: GetObjectInfo failed: %v", instanceType, err)
}
checksums, _ := oi.decryptChecksums(0, nil)
if got := checksums[typ.String()]; got != want {
t.Fatalf("%s: stored checksum %q, want plaintext checksum %q", instanceType, got, want)
}
})
t.Run("copy", func(t *testing.T) {
source := "checksums/source"
if _, err := obj.PutObject(t.Context(), bucketName, source,
mustGetPutObjReader(t, bytes.NewReader(data), int64(len(data)), "", ""), ObjectOptions{}); err != nil {
t.Fatalf("%s: source PutObject failed: %v", instanceType, err)
}
object := "checksums/copy"
uploadID := newMultipartUploadHTTP(t, apiRouter, credentials, bucketName, object,
typ.String(), xhttp.AmzChecksumTypeFullObject)
response := copyPartWithoutChecksumHTTP(t, apiRouter, credentials,
bucketName, source, object, uploadID, "", nil)
if got := copyPartChecksum(typ, response); got != want {
t.Fatalf("%s: UploadPartCopy checksum %q, want %q", instanceType, got, want)
}
listed := listPartsHTTP(t, apiRouter, credentials, bucketName, object, uploadID, nil)
if len(listed.Parts) != 1 || partChecksum(typ, listed.Parts[0]) != want {
t.Fatalf("%s: copied ListParts checksum mismatch: %+v, want %q", instanceType, listed.Parts, want)
}
rec := completePartsHTTP(t, apiRouter, credentials, bucketName, object, uploadID,
[]CompletePart{{PartNumber: 1, ETag: canonicalizeETag(response.ETag)}}, nil)
if rec.Code != http.StatusOK {
t.Fatalf("%s: copied CompleteMultipartUpload failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
})
}
func TestAPIUploadPartServerSideChecksumAlgorithms(t *testing.T) {
defer DetectTestLeak(t)()
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
t: t,
objAPITest: testAPIUploadPartServerSideChecksumAlgorithms,
endpoints: []string{"CopyObjectPart", "PutObjectPart", "NewMultipart", "ListObjectParts", "CompleteMultipart"},
})
}
func testAPIUploadPartServerSideChecksumAlgorithms(obj ObjectLayer, instanceType, bucketName string, apiRouter http.Handler,
credentials auth.Credentials, t *testing.T,
) {
tests := []struct {
typ hash.ChecksumType
objType string
composite bool
}{
{hash.ChecksumCRC32, xhttp.AmzChecksumTypeFullObject, false},
{hash.ChecksumCRC32C, xhttp.AmzChecksumTypeFullObject, false},
{hash.ChecksumCRC64NVME, xhttp.AmzChecksumTypeFullObject, false},
{hash.ChecksumCRC32, xhttp.AmzChecksumTypeComposite, true},
{hash.ChecksumSHA1, xhttp.AmzChecksumTypeComposite, true},
{hash.ChecksumSHA256, xhttp.AmzChecksumTypeComposite, true},
}
data := bytes.Repeat([]byte("server-side-part-checksum"), 1024)
for _, test := range tests {
t.Run(test.typ.String()+"/"+test.objType, func(t *testing.T) {
object := "algorithms/" + test.typ.String() + "/" + test.objType
uploadID := newMultipartUploadHTTP(t, apiRouter, credentials, bucketName, object,
test.typ.String(), test.objType)
etag, rec := uploadPartHTTP(t, apiRouter, credentials,
bucketName, object, uploadID, 1, data, nil)
if got := rec.Header().Get(test.typ.Key()); got != "" {
t.Fatalf("%s: UploadPart returned server-computed checksum %q", instanceType, got)
}
want := mustChecksum(t, test.typ, data)
listed := listPartsHTTP(t, apiRouter, credentials, bucketName, object, uploadID, nil)
if len(listed.Parts) != 1 || partChecksum(test.typ, listed.Parts[0]) != want {
t.Fatalf("%s: ListParts checksum mismatch: %+v, want %q", instanceType, listed.Parts, want)
}
part := CompletePart{PartNumber: 1, ETag: etag}
if test.composite {
part = completePartWithChecksum(test.typ, 1, etag, want)
}
rec = completePartsHTTP(t, apiRouter, credentials, bucketName, object, uploadID,
[]CompletePart{part}, nil)
if rec.Code != http.StatusOK {
t.Fatalf("%s: CompleteMultipartUpload failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
})
}
t.Run("multi-part/FULL_OBJECT", func(t *testing.T) {
typ := hash.ChecksumCRC32
parts, full := multipartChecksumTestData()
object := "algorithms/multi-part-full-object"
uploadID := newMultipartUploadHTTP(t, apiRouter, credentials, bucketName, object,
typ.String(), xhttp.AmzChecksumTypeFullObject)
etags := make([]string, len(parts))
for i, data := range parts {
etag, rec := uploadPartHTTP(t, apiRouter, credentials,
bucketName, object, uploadID, i+1, data, nil)
if got := rec.Header().Get(typ.Key()); got != "" {
t.Fatalf("%s: UploadPart returned server-computed checksum %q", instanceType, got)
}
etags[i] = etag
}
listed := listPartsHTTP(t, apiRouter, credentials, bucketName, object, uploadID, nil)
if len(listed.Parts) != len(parts) {
t.Fatalf("%s: ListParts returned %d parts, want %d", instanceType, len(listed.Parts), len(parts))
}
for i, part := range listed.Parts {
if got, want := partChecksum(typ, part), mustChecksum(t, typ, parts[i]); got != want {
t.Fatalf("%s: part %d checksum %q, want %q", instanceType, i+1, got, want)
}
}
complete := make([]CompletePart, len(etags))
for i, etag := range etags {
complete[i] = CompletePart{PartNumber: i + 1, ETag: etag}
}
rec := completePartsHTTP(t, apiRouter, credentials, bucketName, object, uploadID, complete,
map[string]string{
typ.Key(): mustChecksum(t, typ, full),
xhttp.AmzChecksumType: xhttp.AmzChecksumTypeFullObject,
})
if rec.Code != http.StatusOK {
t.Fatalf("%s: multi-part CompleteMultipartUpload failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
})
t.Run("zero-length-part", func(t *testing.T) {
typ := hash.ChecksumCRC32
object := "algorithms/zero-length"
uploadID := newMultipartUploadHTTP(t, apiRouter, credentials, bucketName, object,
typ.String(), xhttp.AmzChecksumTypeFullObject)
etag, _ := uploadPartHTTP(t, apiRouter, credentials,
bucketName, object, uploadID, 1, nil, nil)
listed := listPartsHTTP(t, apiRouter, credentials, bucketName, object, uploadID, nil)
if len(listed.Parts) != 1 || partChecksum(typ, listed.Parts[0]) != mustChecksum(t, typ, nil) {
t.Fatalf("%s: zero-length ListParts checksum mismatch: %+v", instanceType, listed.Parts)
}
rec := completePartsHTTP(t, apiRouter, credentials, bucketName, object, uploadID,
[]CompletePart{{PartNumber: 1, ETag: etag}}, nil)
if rec.Code != http.StatusOK {
t.Fatalf("%s: zero-length CompleteMultipartUpload failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
})
t.Run("overwrite-part-checksum", func(t *testing.T) {
typ := hash.ChecksumCRC32
object := "algorithms/overwrite"
uploadID := newMultipartUploadHTTP(t, apiRouter, credentials, bucketName, object,
typ.String(), xhttp.AmzChecksumTypeFullObject)
first := []byte("first part contents")
second := []byte("replacement part contents")
uploadPartHTTP(t, apiRouter, credentials, bucketName, object, uploadID, 1, first, nil)
etag, _ := uploadPartHTTP(t, apiRouter, credentials, bucketName, object, uploadID, 1, second, nil)
listed := listPartsHTTP(t, apiRouter, credentials, bucketName, object, uploadID, nil)
if len(listed.Parts) != 1 || partChecksum(typ, listed.Parts[0]) != mustChecksum(t, typ, second) {
t.Fatalf("%s: overwritten ListParts checksum mismatch: %+v", instanceType, listed.Parts)
}
rec := completePartsHTTP(t, apiRouter, credentials, bucketName, object, uploadID,
[]CompletePart{{PartNumber: 1, ETag: etag}}, nil)
if rec.Code != http.StatusOK {
t.Fatalf("%s: overwritten CompleteMultipartUpload failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
})
t.Run("copy/SHA256/COMPOSITE", func(t *testing.T) {
typ := hash.ChecksumSHA256
source := "algorithms/copy-source"
if _, err := obj.PutObject(t.Context(), bucketName, source,
mustGetPutObjReader(t, bytes.NewReader(data), int64(len(data)), "", ""), ObjectOptions{}); err != nil {
t.Fatalf("%s: source PutObject failed: %v", instanceType, err)
}
object := "algorithms/copy-SHA256"
uploadID := newMultipartUploadHTTP(t, apiRouter, credentials, bucketName, object,
typ.String(), xhttp.AmzChecksumTypeComposite)
start, end := 7, len(data)-9
response := copyPartWithoutChecksumHTTP(t, apiRouter, credentials,
bucketName, source, object, uploadID, "bytes="+strconv.Itoa(start)+"-"+strconv.Itoa(end-1), nil)
want := mustChecksum(t, typ, data[start:end])
if got := copyPartChecksum(typ, response); got != want {
t.Fatalf("%s: UploadPartCopy checksum %q, want %q", instanceType, got, want)
}
part := completePartWithChecksum(typ, 1, canonicalizeETag(response.ETag), want)
rec := completePartsHTTP(t, apiRouter, credentials, bucketName, object, uploadID,
[]CompletePart{part}, nil)
if rec.Code != http.StatusOK {
t.Fatalf("%s: copied CompleteMultipartUpload failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
})
}
func TestAPIUploadPartServerSideChecksumDoesNotMaskClientErrors(t *testing.T) {
defer DetectTestLeak(t)()
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
t: t,
objAPITest: testAPIUploadPartServerSideChecksumDoesNotMaskClientErrors,
endpoints: []string{"PutObjectPart", "NewMultipart", "ListObjectParts"},
})
}
func testAPIUploadPartServerSideChecksumDoesNotMaskClientErrors(_ ObjectLayer, instanceType, bucketName string,
apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
) {
data := []byte("client checksum must remain authoritative")
t.Run("correct-value", func(t *testing.T) {
typ := hash.ChecksumCRC32
object := "errors/correct-value"
uploadID := newMultipartUploadHTTP(t, apiRouter, credentials, bucketName, object,
typ.String(), xhttp.AmzChecksumTypeFullObject)
want := mustChecksum(t, typ, data)
_, rec := uploadPartHTTP(t, apiRouter, credentials,
bucketName, object, uploadID, 1, data, map[string]string{typ.Key(): want})
if got := rec.Header().Get(typ.Key()); got != want {
t.Fatalf("%s: client checksum response %q, want %q", instanceType, got, want)
}
listed := listPartsHTTP(t, apiRouter, credentials, bucketName, object, uploadID, nil)
if len(listed.Parts) != 1 || partChecksum(typ, listed.Parts[0]) != want {
t.Fatalf("%s: client checksum ListParts mismatch: %+v", instanceType, listed.Parts)
}
})
t.Run("wrong-algorithm", func(t *testing.T) {
object := "errors/wrong-algorithm"
uploadID := newMultipartUploadHTTP(t, apiRouter, credentials, bucketName, object,
hash.ChecksumCRC32.String(), xhttp.AmzChecksumTypeFullObject)
req, err := newTestSignedRequestV4(http.MethodPut,
getPutObjectPartURL("", bucketName, object, uploadID, "1"),
int64(len(data)), bytes.NewReader(data), credentials.AccessKey, credentials.SecretKey,
map[string]string{hash.ChecksumSHA256.Key(): mustChecksum(t, hash.ChecksumSHA256, data)})
if err != nil {
t.Fatalf("failed to build UploadPart request: %v", err)
}
rec := httptest.NewRecorder()
apiRouter.ServeHTTP(rec, req)
if rec.Code != http.StatusBadRequest || apiErrorCode(t, rec) != "InvalidArgument" {
t.Fatalf("%s: wrong algorithm returned %d %s", instanceType, rec.Code, rec.Body.String())
}
})
t.Run("wrong-value", func(t *testing.T) {
object := "errors/wrong-value"
uploadID := newMultipartUploadHTTP(t, apiRouter, credentials, bucketName, object,
hash.ChecksumCRC32.String(), xhttp.AmzChecksumTypeFullObject)
req, err := newTestSignedRequestV4(http.MethodPut,
getPutObjectPartURL("", bucketName, object, uploadID, "1"),
int64(len(data)), bytes.NewReader(data), credentials.AccessKey, credentials.SecretKey,
map[string]string{hash.ChecksumCRC32.Key(): mustChecksum(t, hash.ChecksumCRC32, []byte("wrong"))})
if err != nil {
t.Fatalf("failed to build UploadPart request: %v", err)
}
rec := httptest.NewRecorder()
apiRouter.ServeHTTP(rec, req)
if rec.Code != http.StatusBadRequest || apiErrorCode(t, rec) != "XAmzContentChecksumMismatch" {
t.Fatalf("%s: wrong value returned %d %s", instanceType, rec.Code, rec.Body.String())
}
})
}
func TestAPIUploadPartServerSideChecksumSSEC(t *testing.T) {
defer DetectTestLeak(t)()
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
t: t,
objAPITest: testAPIUploadPartServerSideChecksumSSEC,
endpoints: []string{"PutObjectPart", "NewMultipart", "CompleteMultipart"},
})
}
func testAPIUploadPartServerSideChecksumSSEC(_ ObjectLayer, instanceType, bucketName string, apiRouter http.Handler,
credentials auth.Credentials, t *testing.T,
) {
globalIsTLS = true
defer func() { globalIsTLS = false }()
key := bytes.Repeat([]byte{0x2a}, 32)
keyMD5 := md5.Sum(key)
ssecHeaders := map[string]string{
xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES,
xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(key),
xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(keyMD5[:]),
}
initHeaders := map[string]string{
xhttp.AmzChecksumAlgo: hash.ChecksumCRC32.String(),
xhttp.AmzChecksumType: xhttp.AmzChecksumTypeFullObject,
xhttp.AmzServerSideEncryptionCustomerAlgorithm: ssecHeaders[xhttp.AmzServerSideEncryptionCustomerAlgorithm],
xhttp.AmzServerSideEncryptionCustomerKey: ssecHeaders[xhttp.AmzServerSideEncryptionCustomerKey],
xhttp.AmzServerSideEncryptionCustomerKeyMD5: ssecHeaders[xhttp.AmzServerSideEncryptionCustomerKeyMD5],
}
object := "checksums/ssec"
req, err := newTestSignedRequestV4(http.MethodPost, getNewMultipartURL("", bucketName, object),
0, nil, credentials.AccessKey, credentials.SecretKey, initHeaders)
if err != nil {
t.Fatalf("failed to build NewMultipartUpload request: %v", err)
}
rec := httptest.NewRecorder()
apiRouter.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("%s: NewMultipartUpload failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
var initiated InitiateMultipartUploadResponse
if err := xml.Unmarshal(rec.Body.Bytes(), &initiated); err != nil {
t.Fatalf("failed to decode NewMultipartUpload response: %v", err)
}
data := bytes.Repeat([]byte("ssec-checksum-plaintext"), 4096)
etag, uploadRec := uploadPartHTTP(t, apiRouter, credentials,
bucketName, object, initiated.UploadID, 1, data, ssecHeaders)
if got := uploadRec.Header().Get(hash.ChecksumCRC32.Key()); got != "" {
t.Fatalf("%s: UploadPart returned server-computed checksum %q", instanceType, got)
}
completeHeaders := map[string]string{
xhttp.AmzChecksumCRC32: mustChecksum(t, hash.ChecksumCRC32, data),
xhttp.AmzChecksumType: xhttp.AmzChecksumTypeFullObject,
xhttp.AmzServerSideEncryptionCustomerAlgorithm: ssecHeaders[xhttp.AmzServerSideEncryptionCustomerAlgorithm],
xhttp.AmzServerSideEncryptionCustomerKey: ssecHeaders[xhttp.AmzServerSideEncryptionCustomerKey],
xhttp.AmzServerSideEncryptionCustomerKeyMD5: ssecHeaders[xhttp.AmzServerSideEncryptionCustomerKeyMD5],
}
rec = completePartsHTTP(t, apiRouter, credentials, bucketName, object, initiated.UploadID,
[]CompletePart{{PartNumber: 1, ETag: etag}}, completeHeaders)
if rec.Code != http.StatusOK {
t.Fatalf("%s: CompleteMultipartUpload failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
}
func TestAPIUploadPartServerSideChecksumSSES3(t *testing.T) {
defer DetectTestLeak(t)()
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
t: t,
objAPITest: testAPIUploadPartServerSideChecksumSSES3,
endpoints: []string{"PutObjectPart", "NewMultipart", "CompleteMultipart"},
})
}
func testAPIUploadPartServerSideChecksumSSES3(_ ObjectLayer, instanceType, bucketName string, apiRouter http.Handler,
credentials auth.Credentials, t *testing.T,
) {
KMS, err := kms.ParseSecretKey("my-minio-key:5lF+0pJM0OWwlQrvK2S/I7W9mO4a6rJJI7wzj7v09cw=")
if err != nil {
t.Fatal(err)
}
GlobalKMS = KMS
defer func() { GlobalKMS = nil }()
object := "checksums/sse-s3"
initHeaders := map[string]string{
xhttp.AmzChecksumAlgo: hash.ChecksumCRC32.String(),
xhttp.AmzChecksumType: xhttp.AmzChecksumTypeFullObject,
xhttp.AmzServerSideEncryption: xhttp.AmzEncryptionAES,
}
req, err := newTestSignedRequestV4(http.MethodPost, getNewMultipartURL("", bucketName, object),
0, nil, credentials.AccessKey, credentials.SecretKey, initHeaders)
if err != nil {
t.Fatalf("failed to build NewMultipartUpload request: %v", err)
}
rec := httptest.NewRecorder()
apiRouter.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("%s: NewMultipartUpload failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
var initiated InitiateMultipartUploadResponse
if err := xml.Unmarshal(rec.Body.Bytes(), &initiated); err != nil {
t.Fatalf("failed to decode NewMultipartUpload response: %v", err)
}
data := bytes.Repeat([]byte("sse-s3-checksum-plaintext"), 4096)
etag, uploadRec := uploadPartHTTP(t, apiRouter, credentials,
bucketName, object, initiated.UploadID, 1, data, nil)
if got := uploadRec.Header().Get(hash.ChecksumCRC32.Key()); got != "" {
t.Fatalf("%s: UploadPart returned server-computed checksum %q", instanceType, got)
}
rec = completePartsHTTP(t, apiRouter, credentials, bucketName, object, initiated.UploadID,
[]CompletePart{{PartNumber: 1, ETag: etag}}, map[string]string{
xhttp.AmzChecksumCRC32: mustChecksum(t, hash.ChecksumCRC32, data),
xhttp.AmzChecksumType: xhttp.AmzChecksumTypeFullObject,
})
if rec.Code != http.StatusOK {
t.Fatalf("%s: CompleteMultipartUpload failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
}
+14 -4
View File
@@ -597,12 +597,15 @@ func (er erasureObjects) PutObjectPart(ctx context.Context, bucket, object, uplo
onlineDisks := er.getDisks()
writeQuorum := fi.WriteQuorum(er.defaultWQuorum())
if cs := fi.Metadata[hash.MinIOMultipartChecksum]; cs != "" {
if r.ContentCRCType().String() != cs {
expectedChecksumType, checksumEnabled := multipartChecksumType(fi.Metadata)
if checksumEnabled {
got := r.contentChecksumType()
if !expectedChecksumType.IsSet() || !got.IsSet() || got.Base() != expectedChecksumType {
return pi, InvalidArgument{
Bucket: bucket,
Object: fi.Name,
Err: fmt.Errorf("checksum missing, want %q, got %q", cs, r.ContentCRCType().String()),
Err: fmt.Errorf("checksum missing, want %q, got %q",
fi.Metadata[hash.MinIOMultipartChecksum], got.String()),
}
}
}
@@ -725,6 +728,13 @@ func (er erasureObjects) PutObjectPart(ctx context.Context, bucket, object, uplo
}
}
partChecksums := r.contentChecksum()
if checksumEnabled && partChecksums[expectedChecksumType.String()] == "" {
err := fmt.Errorf("internal error: checksum missing after reading part, want %q", expectedChecksumType.String())
bugLogIf(ctx, err)
return pi, toObjectErr(err, bucket, object, uploadID)
}
partInfo := ObjectPartInfo{
Number: partID,
ETag: md5hex,
@@ -732,7 +742,7 @@ func (er erasureObjects) PutObjectPart(ctx context.Context, bucket, object, uplo
ActualSize: actualSize,
ModTime: UTCNow(),
Index: index,
Checksums: r.ContentCRC(),
Checksums: partChecksums,
}
partFI, err := partInfo.MarshalMsg(nil)
+9 -5
View File
@@ -1485,11 +1485,15 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
// over opts.WantChecksum.
if opts.WantServerSideChecksumType.IsSet() {
serverSideChecksum := r.RawServerSideChecksumResult()
if serverSideChecksum != nil {
fi.Checksum = serverSideChecksum.AppendTo(nil, nil)
if opts.EncryptFn != nil {
fi.Checksum = opts.EncryptFn("object-checksum", fi.Checksum)
}
if serverSideChecksum == nil || !serverSideChecksum.Valid() ||
serverSideChecksum.Type.Base() != opts.WantServerSideChecksumType.Base() {
err := fmt.Errorf("internal error: server-side checksum missing, invalid, or mismatched after reading object, want %q", opts.WantServerSideChecksumType.String())
bugLogIf(ctx, err)
return ObjectInfo{}, toObjectErr(err, bucket, object)
}
fi.Checksum = serverSideChecksum.AppendTo(nil, nil)
if opts.EncryptFn != nil {
fi.Checksum = opts.EncryptFn("object-checksum", fi.Checksum)
}
} else if fi.Checksum == nil && opts.WantChecksum != nil {
// Trailing headers checksums should now be filled.
+2
View File
@@ -191,7 +191,9 @@ func collectLocalDisksMetrics(disks map[string]struct{}) map[string]madmin.DiskM
}
}
//nolint:staticcheck // Linux implementations can fail; BSD stubs return a constant nil error.
st, err := disk.GetDriveStats(d.Major, d.Minor)
//nolint:staticcheck // Keep the shared cross-platform error handling.
if err == nil {
dm.IOStats = madmin.DiskIOStats{
ReadIOs: st.ReadIOs,
+45 -8
View File
@@ -1038,9 +1038,10 @@ type SealMD5CurrFn func([]byte) []byte
// PutObjReader is a type that wraps sio.EncryptReader and
// underlying hash.Reader in a struct
type PutObjReader struct {
*hash.Reader // actual data stream
rawReader *hash.Reader // original data stream
sealMD5Fn SealMD5CurrFn
*hash.Reader // actual data stream
rawReader *hash.Reader // original data stream used for ETag calculation
checksumReader *hash.Reader // logical plaintext stream used for S3 checksum calculation
sealMD5Fn SealMD5CurrFn
}
// Size returns the absolute number of bytes the Reader
@@ -1093,15 +1094,51 @@ func (p *PutObjReader) WithEncryption(encReader *hash.Reader, objEncKey *crypto.
// NewPutObjReader returns a new PutObjReader. It uses given hash.Reader's
// MD5Current method to construct md5sum when requested downstream.
func NewPutObjReader(rawReader *hash.Reader) *PutObjReader {
return &PutObjReader{Reader: rawReader, rawReader: rawReader}
return &PutObjReader{Reader: rawReader, rawReader: rawReader, checksumReader: rawReader}
}
// setChecksumReader sets the logical plaintext reader used for S3 checksums.
// It can differ from rawReader when the storage stream is compressed.
func (p *PutObjReader) setChecksumReader(r *hash.Reader) {
if r != nil {
p.checksumReader = r
}
}
// contentChecksumType returns the effective client-provided or server-computed
// checksum type for the logical plaintext stream.
func (p *PutObjReader) contentChecksumType() hash.ChecksumType {
if p.checksumReader == nil {
return hash.ChecksumNone
}
if t := p.checksumReader.ContentCRCType(); t.IsSet() {
return t
}
return p.checksumReader.ServerSideChecksumType
}
// contentChecksum returns the effective checksum for part metadata. A
// client-provided checksum takes precedence; server computation is only a
// fallback when the client omitted one.
func (p *PutObjReader) contentChecksum() map[string]string {
if p.checksumReader == nil {
return nil
}
if checksum := p.checksumReader.ContentCRC(); checksum != nil {
return checksum
}
if checksum := p.checksumReader.ServerSideChecksumResult; checksum != nil && checksum.Valid() {
return map[string]string{checksum.Type.String(): checksum.Encoded}
}
return nil
}
// RawServerSideChecksumResult returns the ServerSideChecksumResult from the
// underlying rawReader, since the PutObjReader might be encrypted data and
// thus any checksum from that would be incorrect.
// logical plaintext checksum reader, since the PutObjReader might contain
// compressed or encrypted data and thus any checksum from that would be incorrect.
func (p *PutObjReader) RawServerSideChecksumResult() *hash.Checksum {
if p.rawReader != nil {
return p.rawReader.ServerSideChecksumResult
if p.checksumReader != nil {
return p.checksumReader.ServerSideChecksumResult
}
return nil
}
+556
View File
@@ -0,0 +1,556 @@
// Copyright (c) 2015-2026 MinIO, Inc.
//
// 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 cmd
import (
"bytes"
"crypto/md5"
"encoding/base64"
"encoding/hex"
"encoding/xml"
"io"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/minio/minio/internal/auth"
"github.com/minio/minio/internal/hash"
xhttp "github.com/minio/minio/internal/http"
"github.com/minio/minio/internal/kms"
)
func setCopyChecksumCompression(allowEncrypted bool) func() {
globalCompressConfigMu.Lock()
previous := globalCompressConfig
globalCompressConfig.Enabled = true
globalCompressConfig.Extensions = []string{".txt"}
globalCompressConfig.MimeTypes = nil
globalCompressConfig.AllowEncrypted = allowEncrypted
globalCompressConfigMu.Unlock()
return func() {
globalCompressConfigMu.Lock()
globalCompressConfig = previous
globalCompressConfigMu.Unlock()
}
}
func copyChecksumRequest(t *testing.T, apiRouter http.Handler, credentials auth.Credentials,
bucket, source, destination string, headers map[string]string,
) *httptest.ResponseRecorder {
t.Helper()
req, err := newTestSignedRequestV4(http.MethodPut, getCopyObjectURL("", bucket, destination),
0, nil, credentials.AccessKey, credentials.SecretKey, headers)
if err != nil {
t.Fatalf("failed to build CopyObject request: %v", err)
}
req.Header.Set(xhttp.AmzCopySource, SlashSeparator+pathJoin(bucket, source))
rec := httptest.NewRecorder()
apiRouter.ServeHTTP(rec, req)
return rec
}
func putCopyChecksumSource(t *testing.T, apiRouter http.Handler, credentials auth.Credentials,
bucket, object string, data []byte, headers map[string]string,
) {
t.Helper()
req, err := newTestSignedRequestV4(http.MethodPut, getPutObjectURL("", bucket, object),
int64(len(data)), bytes.NewReader(data), credentials.AccessKey, credentials.SecretKey, headers)
if err != nil {
t.Fatalf("failed to build PutObject request: %v", err)
}
rec := httptest.NewRecorder()
apiRouter.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("PutObject(%s) failed: %d %s", object, rec.Code, rec.Body.String())
}
}
func readCopyChecksumObject(t *testing.T, obj ObjectLayer, bucket, object string, opts ObjectOptions) []byte {
t.Helper()
gr, err := obj.GetObjectNInfo(t.Context(), bucket, object, nil, nil, opts)
if err != nil {
t.Fatalf("GetObjectNInfo(%s) failed: %v", object, err)
}
defer gr.Close()
data, err := io.ReadAll(gr)
if err != nil {
t.Fatalf("reading %s failed: %v", object, err)
}
return data
}
func assertCopyChecksum(t *testing.T, obj ObjectLayer, bucket, object string, typ hash.ChecksumType,
data []byte, compressed bool, decryptHeaders http.Header,
) ObjectInfo {
t.Helper()
oi, err := obj.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{})
if err != nil {
t.Fatalf("GetObjectInfo(%s) failed: %v", object, err)
}
if oi.IsCompressed() != compressed {
t.Fatalf("%s compressed=%v, want %v", object, oi.IsCompressed(), compressed)
}
checksums, _ := oi.decryptChecksums(0, decryptHeaders)
if got, want := checksums[typ.String()], mustChecksum(t, typ, data); got != want {
t.Fatalf("%s stored %s checksum %q, want logical object checksum %q (all: %v)",
object, typ.String(), got, want, checksums)
}
if got := checksums[xhttp.AmzChecksumType]; got != xhttp.AmzChecksumTypeFullObject {
t.Fatalf("%s checksum type %q, want %q", object, got, xhttp.AmzChecksumTypeFullObject)
}
return oi
}
func assertCopyChecksumResponse(t *testing.T, rec *httptest.ResponseRecorder, typ hash.ChecksumType, data []byte) {
t.Helper()
var response CopyObjectResponse
if err := xml.Unmarshal(rec.Body.Bytes(), &response); err != nil {
t.Fatalf("unable to decode CopyObjectResult: %v", err)
}
var got string
switch typ.Base() {
case hash.ChecksumCRC32:
got = response.ChecksumCRC32
case hash.ChecksumCRC32C:
got = response.ChecksumCRC32C
case hash.ChecksumSHA1:
got = response.ChecksumSHA1
case hash.ChecksumSHA256:
got = response.ChecksumSHA256
case hash.ChecksumCRC64NVME:
got = response.ChecksumCRC64NVME
}
if want := mustChecksum(t, typ, data); got != want {
t.Fatalf("CopyObjectResult %s checksum %q, want %q: %s", typ.String(), got, want, rec.Body.String())
}
if response.ChecksumType != xhttp.AmzChecksumTypeFullObject {
t.Fatalf("CopyObjectResult checksum type %q, want %q", response.ChecksumType, xhttp.AmzChecksumTypeFullObject)
}
}
// TestAPICopyObjectServerSideChecksum verifies that server-computed checksums
// cover the logical object, never the compressed storage stream.
func TestAPICopyObjectServerSideChecksum(t *testing.T) {
defer DetectTestLeak(t)()
for _, versioned := range []bool{false, true} {
name := "unversioned"
if versioned {
name = "versioned"
}
t.Run(name, func(t *testing.T) {
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
t: t,
objAPITest: testAPICopyObjectServerSideChecksum,
endpoints: []string{"CopyObject", "PutObject", "HeadObject", "GetObject"},
makeBucketOptions: MakeBucketOptions{VersioningEnabled: versioned},
})
})
}
}
func testAPICopyObjectServerSideChecksum(obj ObjectLayer, instanceType, bucketName string,
apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
) {
restoreCompression := setCopyChecksumCompression(true)
defer restoreCompression()
data := bytes.Repeat([]byte("copy-object-checksum-plaintext-"), 64*1024)
source := "copy-checksum/source.bin"
if _, err := obj.PutObject(t.Context(), bucketName, source,
mustGetPutObjReader(t, bytes.NewReader(data), int64(len(data)), "", ""), ObjectOptions{}); err != nil {
t.Fatalf("%s: source PutObject failed: %v", instanceType, err)
}
compressedReader, _ := newS2CompressReader(bytes.NewReader(data), int64(len(data)), false)
compressed, err := io.ReadAll(compressedReader)
if closeErr := compressedReader.Close(); err == nil {
err = closeErr
}
if err != nil {
t.Fatalf("%s: independently compressing test data failed: %v", instanceType, err)
}
cases := []struct {
name string
typ hash.ChecksumType
explicit bool
extension string
compressed bool
}{
{name: "compressed/CRC32", typ: hash.ChecksumCRC32, explicit: true, extension: ".txt", compressed: true},
{name: "compressed/CRC32C", typ: hash.ChecksumCRC32C, explicit: true, extension: ".txt", compressed: true},
{name: "compressed/SHA1", typ: hash.ChecksumSHA1, explicit: true, extension: ".txt", compressed: true},
{name: "compressed/SHA256", typ: hash.ChecksumSHA256, explicit: true, extension: ".txt", compressed: true},
{name: "compressed/CRC64NVME", typ: hash.ChecksumCRC64NVME, explicit: true, extension: ".txt", compressed: true},
{name: "compressed/default", typ: hash.ChecksumCRC64NVME, extension: ".txt", compressed: true},
{name: "plain/CRC32", typ: hash.ChecksumCRC32, explicit: true, extension: ".bin"},
{name: "plain/default", typ: hash.ChecksumCRC64NVME, extension: ".bin"},
}
for i, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
headers := map[string]string(nil)
if tc.explicit {
headers = map[string]string{xhttp.AmzChecksumAlgo: tc.typ.String()}
}
destination := "copy-checksum/destination-" + tc.typ.String() + "-" + string(rune('a'+i)) + tc.extension
rec := copyChecksumRequest(t, apiRouter, credentials, bucketName, source, destination, headers)
if rec.Code != http.StatusOK {
t.Fatalf("%s: CopyObject failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
assertCopyChecksumResponse(t, rec, tc.typ, data)
info := assertCopyChecksum(t, obj, bucketName, destination, tc.typ, data, tc.compressed, nil)
md5sum := md5.Sum(data)
if got, want := info.ETag, hex.EncodeToString(md5sum[:]); got != want {
t.Fatalf("%s: ETag %q, want logical object MD5 %q", instanceType, got, want)
}
if tc.compressed {
logical := mustChecksum(t, tc.typ, data)
if transformed := mustChecksum(t, tc.typ, compressed); logical == transformed {
t.Fatalf("%s: test payload does not distinguish logical and compressed checksum domains", instanceType)
}
}
if got := readCopyChecksumObject(t, obj, bucketName, destination, ObjectOptions{}); !bytes.Equal(got, data) {
t.Fatalf("%s: round-trip body differs for %s", instanceType, tc.name)
}
if tc.name == "compressed/CRC32" {
for _, method := range []string{http.MethodHead, http.MethodGet} {
url := getHeadObjectURL("", bucketName, destination)
if method == http.MethodGet {
url = getGetObjectURL("", bucketName, destination)
}
req, err := newTestSignedRequestV4(method, url, 0, nil,
credentials.AccessKey, credentials.SecretKey,
map[string]string{xhttp.AmzChecksumMode: "ENABLED"})
if err != nil {
t.Fatalf("failed to build %s request: %v", method, err)
}
response := httptest.NewRecorder()
apiRouter.ServeHTTP(response, req)
if response.Code != http.StatusOK {
t.Fatalf("%s returned %d: %s", method, response.Code, response.Body.String())
}
if got, want := response.Header().Get(tc.typ.Key()), mustChecksum(t, tc.typ, data); got != want {
t.Fatalf("%s returned checksum %q, want %q", method, got, want)
}
if method == http.MethodGet && !bytes.Equal(response.Body.Bytes(), data) {
t.Fatalf("GET response body differs")
}
}
}
})
}
}
func TestAPICopyObjectServerSideChecksumEncryption(t *testing.T) {
defer DetectTestLeak(t)()
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
t: t,
objAPITest: testAPICopyObjectServerSideChecksumEncryption,
endpoints: []string{"CopyObject", "PutObject", "GetObject"},
})
}
func testAPICopyObjectServerSideChecksumEncryption(obj ObjectLayer, instanceType, bucketName string,
apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
) {
restoreCompression := setCopyChecksumCompression(true)
defer restoreCompression()
data := bytes.Repeat([]byte("encrypted-copy-checksum-plaintext-"), 48*1024)
source := "copy-checksum/encrypted-source.bin"
if _, err := obj.PutObject(t.Context(), bucketName, source,
mustGetPutObjReader(t, bytes.NewReader(data), int64(len(data)), "", ""), ObjectOptions{}); err != nil {
t.Fatalf("%s: source PutObject failed: %v", instanceType, err)
}
t.Run("SSE-S3", func(t *testing.T) {
secretKey, err := kms.ParseSecretKey("my-minio-key:5lF+0pJM0OWwlQrvK2S/I7W9mO4a6rJJI7wzj7v09cw=")
if err != nil {
t.Fatal(err)
}
previousKMS := GlobalKMS
GlobalKMS = secretKey
defer func() { GlobalKMS = previousKMS }()
for _, variant := range []struct {
name string
extension string
compressed bool
}{
{name: "encrypted-only", extension: ".bin"},
{name: "compressed-encrypted", extension: ".txt", compressed: true},
} {
t.Run(variant.name, func(t *testing.T) {
destination := "copy-checksum/sse-s3-" + variant.name + variant.extension
rec := copyChecksumRequest(t, apiRouter, credentials, bucketName, source, destination, map[string]string{
xhttp.AmzChecksumAlgo: hash.ChecksumCRC32.String(),
xhttp.AmzServerSideEncryption: xhttp.AmzEncryptionAES,
})
if rec.Code != http.StatusOK {
t.Fatalf("%s: SSE-S3 CopyObject failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
assertCopyChecksumResponse(t, rec, hash.ChecksumCRC32, data)
assertCopyChecksum(t, obj, bucketName, destination, hash.ChecksumCRC32, data, variant.compressed, nil)
if got := readCopyChecksumObject(t, obj, bucketName, destination, ObjectOptions{}); !bytes.Equal(got, data) {
t.Fatalf("%s: SSE-S3 round-trip body differs", instanceType)
}
})
}
encryptedSource := "copy-checksum/sse-s3-source.bin"
putCopyChecksumSource(t, apiRouter, credentials, bucketName, encryptedSource, data,
map[string]string{xhttp.AmzServerSideEncryption: xhttp.AmzEncryptionAES})
destination := "copy-checksum/sse-s3-source-copy.txt"
rec := copyChecksumRequest(t, apiRouter, credentials, bucketName, encryptedSource, destination,
map[string]string{xhttp.AmzChecksumAlgo: hash.ChecksumCRC32.String()})
if rec.Code != http.StatusOK {
t.Fatalf("%s: SSE-S3 source CopyObject failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
assertCopyChecksumResponse(t, rec, hash.ChecksumCRC32, data)
assertCopyChecksum(t, obj, bucketName, destination, hash.ChecksumCRC32, data, true, nil)
if got := readCopyChecksumObject(t, obj, bucketName, destination, ObjectOptions{}); !bytes.Equal(got, data) {
t.Fatalf("%s: SSE-S3 source round-trip body differs", instanceType)
}
})
t.Run("SSE-C", func(t *testing.T) {
previousTLS := globalIsTLS
globalIsTLS = true
defer func() { globalIsTLS = previousTLS }()
key := bytes.Repeat([]byte{0x2a}, 32)
keyMD5 := md5.Sum(key)
headers := map[string]string{
xhttp.AmzChecksumAlgo: hash.ChecksumCRC32.String(),
xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES,
xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(key),
xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(keyMD5[:]),
}
decryptHeaders := http.Header{}
for key, value := range headers {
decryptHeaders.Set(key, value)
}
getHeaders := make(map[string]string, len(headers))
for key, value := range headers {
if key != xhttp.AmzChecksumAlgo {
getHeaders[key] = value
}
}
for _, variant := range []struct {
name string
extension string
compressed bool
}{
{name: "encrypted-only", extension: ".bin"},
{name: "compressed-encrypted", extension: ".txt", compressed: true},
} {
t.Run(variant.name, func(t *testing.T) {
destination := "copy-checksum/sse-c-" + variant.name + variant.extension
rec := copyChecksumRequest(t, apiRouter, credentials, bucketName, source, destination, headers)
if rec.Code != http.StatusOK {
t.Fatalf("%s: SSE-C CopyObject failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
assertCopyChecksumResponse(t, rec, hash.ChecksumCRC32, data)
assertCopyChecksum(t, obj, bucketName, destination, hash.ChecksumCRC32, data, variant.compressed, decryptHeaders)
req, err := newTestSignedRequestV4(http.MethodGet, getGetObjectURL("", bucketName, destination),
0, nil, credentials.AccessKey, credentials.SecretKey, getHeaders)
if err != nil {
t.Fatalf("failed to build SSE-C GetObject request: %v", err)
}
response := httptest.NewRecorder()
apiRouter.ServeHTTP(response, req)
if response.Code != http.StatusOK || !bytes.Equal(response.Body.Bytes(), data) {
t.Fatalf("%s: SSE-C GetObject returned %d with %d bytes, want 200 with %d bytes",
instanceType, response.Code, response.Body.Len(), len(data))
}
})
}
})
}
func TestAPICopyObjectServerSideChecksumSourceVariants(t *testing.T) {
defer DetectTestLeak(t)()
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
t: t,
objAPITest: testAPICopyObjectServerSideChecksumSourceVariants,
endpoints: []string{
"NewMultipart", "PutObjectPart", "CompleteMultipart", "ListObjectParts",
"CopyObject", "PutObject", "HeadObject", "GetObject",
},
})
}
func testAPICopyObjectServerSideChecksumSourceVariants(obj ObjectLayer, instanceType, bucketName string,
apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
) {
restoreCompression := setCopyChecksumCompression(true)
defer restoreCompression()
data := bytes.Repeat([]byte("source-variant-plaintext-"), 64*1024)
t.Run("compressed-source", func(t *testing.T) {
source := "copy-checksum/compressed-source.txt"
putCopyChecksumSource(t, apiRouter, credentials, bucketName, source, data, nil)
if info, err := obj.GetObjectInfo(t.Context(), bucketName, source, ObjectOptions{}); err != nil || !info.IsCompressed() {
t.Fatalf("%s: compressed source precondition failed: compressed=%v err=%v", instanceType, info.IsCompressed(), err)
}
destination := "copy-checksum/compressed-source-copy.txt"
rec := copyChecksumRequest(t, apiRouter, credentials, bucketName, source, destination,
map[string]string{xhttp.AmzChecksumAlgo: hash.ChecksumCRC32.String()})
if rec.Code != http.StatusOK {
t.Fatalf("%s: CopyObject failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
assertCopyChecksumResponse(t, rec, hash.ChecksumCRC32, data)
assertCopyChecksum(t, obj, bucketName, destination, hash.ChecksumCRC32, data, true, nil)
rec = copyChecksumRequest(t, apiRouter, credentials, bucketName, source, source, map[string]string{
xhttp.AmzChecksumAlgo: hash.ChecksumSHA256.String(),
xhttp.AmzMetadataDirective: "REPLACE",
})
if rec.Code != http.StatusOK {
t.Fatalf("%s: in-place CopyObject failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
assertCopyChecksumResponse(t, rec, hash.ChecksumSHA256, data)
assertCopyChecksum(t, obj, bucketName, source, hash.ChecksumSHA256, data, true, nil)
if got := readCopyChecksumObject(t, obj, bucketName, source, ObjectOptions{}); !bytes.Equal(got, data) {
t.Fatalf("%s: in-place CopyObject body differs", instanceType)
}
})
t.Run("full-checksum-source", func(t *testing.T) {
source := "copy-checksum/full-checksum-source.bin"
want := mustChecksum(t, hash.ChecksumCRC32, data)
putCopyChecksumSource(t, apiRouter, credentials, bucketName, source, data,
map[string]string{xhttp.AmzChecksumCRC32: want})
destination := "copy-checksum/full-checksum-copy.txt"
rec := copyChecksumRequest(t, apiRouter, credentials, bucketName, source, destination, nil)
if rec.Code != http.StatusOK {
t.Fatalf("%s: CopyObject failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
assertCopyChecksumResponse(t, rec, hash.ChecksumCRC32, data)
assertCopyChecksum(t, obj, bucketName, destination, hash.ChecksumCRC32, data, true, nil)
})
t.Run("multipart-composite-source", func(t *testing.T) {
typ := hash.ChecksumCRC32
parts, full := multipartChecksumTestData()
source := "copy-checksum/multipart-source.bin"
uploadID := newMultipartUploadHTTP(t, apiRouter, credentials, bucketName, source,
typ.String(), xhttp.AmzChecksumTypeComposite)
etags := uploadPartsHTTP(t, apiRouter, credentials, bucketName, source, uploadID, typ, parts)
partChecksums := make([]string, len(parts))
for i, part := range parts {
partChecksums[i] = mustChecksum(t, typ, part)
}
rec := completeMultipartUploadHTTP(t, apiRouter, credentials, bucketName, source, uploadID,
etags, partChecksums, nil)
if rec.Code != http.StatusOK {
t.Fatalf("%s: CompleteMultipartUpload failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
sourceInfo, err := obj.GetObjectInfo(t.Context(), bucketName, source, ObjectOptions{})
if err != nil {
t.Fatalf("%s: source GetObjectInfo failed: %v", instanceType, err)
}
if _, multipart := sourceInfo.decryptChecksums(0, nil); !multipart {
t.Fatalf("%s: source checksum is not multipart composite", instanceType)
}
destination := "copy-checksum/multipart-copy.txt"
rec = copyChecksumRequest(t, apiRouter, credentials, bucketName, source, destination, nil)
if rec.Code != http.StatusOK {
t.Fatalf("%s: CopyObject failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
assertCopyChecksumResponse(t, rec, typ, full)
assertCopyChecksum(t, obj, bucketName, destination, typ, full, true, nil)
if got := readCopyChecksumObject(t, obj, bucketName, destination, ObjectOptions{}); !bytes.Equal(got, full) {
t.Fatalf("%s: multipart source round-trip body differs", instanceType)
}
})
for _, boundary := range []struct {
name string
data []byte
compressed bool
}{
{name: "at-threshold", data: bytes.Repeat([]byte{'a'}, minCompressibleSize)},
{name: "over-threshold", data: bytes.Repeat([]byte{'a'}, minCompressibleSize+1), compressed: true},
{name: "indexed", data: bytes.Repeat([]byte{'a'}, compMinIndexSize+1), compressed: true},
{name: "empty", data: nil},
} {
t.Run(boundary.name, func(t *testing.T) {
source := "copy-checksum/" + boundary.name + "-source.bin"
if _, err := obj.PutObject(t.Context(), bucketName, source,
mustGetPutObjReader(t, bytes.NewReader(boundary.data), int64(len(boundary.data)), "", ""), ObjectOptions{}); err != nil {
t.Fatalf("%s: source PutObject failed: %v", instanceType, err)
}
destination := "copy-checksum/" + boundary.name + "-copy.txt"
rec := copyChecksumRequest(t, apiRouter, credentials, bucketName, source, destination,
map[string]string{xhttp.AmzChecksumAlgo: hash.ChecksumCRC32.String()})
if rec.Code != http.StatusOK {
t.Fatalf("%s: CopyObject failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
assertCopyChecksumResponse(t, rec, hash.ChecksumCRC32, boundary.data)
assertCopyChecksum(t, obj, bucketName, destination, hash.ChecksumCRC32, boundary.data, boundary.compressed, nil)
})
}
}
func TestPutObjectRejectsMissingServerSideChecksum(t *testing.T) {
defer DetectTestLeak(t)()
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
t: t,
objAPITest: testPutObjectRejectsMissingServerSideChecksum,
endpoints: []string{"PutObject"},
})
}
func testPutObjectRejectsMissingServerSideChecksum(obj ObjectLayer, instanceType, bucketName string,
_ http.Handler, _ auth.Credentials, t *testing.T,
) {
data := []byte("the object layer must not silently omit a requested checksum")
for _, test := range []struct {
name string
hasherType hash.ChecksumType
}{
{name: "missing"},
{name: "mismatched", hasherType: hash.ChecksumCRC32C},
} {
t.Run(test.name, func(t *testing.T) {
object := "copy-checksum/" + test.name + "-server-side-checksum"
reader := mustGetPutObjReader(t, bytes.NewReader(data), int64(len(data)), "", "")
if test.hasherType.IsSet() {
reader.AddServerSideChecksumHasher(test.hasherType)
}
_, err := obj.PutObject(t.Context(), bucketName, object, reader,
ObjectOptions{WantServerSideChecksumType: hash.ChecksumCRC32})
if err == nil || !strings.Contains(err.Error(), "server-side checksum") {
t.Fatalf("%s: PutObject error %v, want server-side checksum invariant error", instanceType, err)
}
if _, err = obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{}); !isErrObjectNotFound(err) {
t.Fatalf("%s: failed PutObject left an object behind: %v", instanceType, err)
}
})
}
}
+202
View File
@@ -0,0 +1,202 @@
// Copyright (c) 2015-2026 MinIO, Inc.
//
// 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 cmd
import (
"bytes"
"crypto/md5"
"encoding/base64"
"net/http"
"net/http/httptest"
"testing"
"github.com/minio/minio/internal/auth"
"github.com/minio/minio/internal/hash"
xhttp "github.com/minio/minio/internal/http"
)
func TestAPICopyObjectMetadataOnlyCompression(t *testing.T) {
defer DetectTestLeak(t)()
for _, versioned := range []bool{false, true} {
name := "unversioned"
if versioned {
name = "versioned"
}
t.Run(name, func(t *testing.T) {
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
t: t,
objAPITest: testAPICopyObjectMetadataOnlyCompression,
endpoints: []string{"CopyObject", "PutObject", "GetObject"},
makeBucketOptions: MakeBucketOptions{VersioningEnabled: versioned},
})
})
}
}
func testAPICopyObjectMetadataOnlyCompression(obj ObjectLayer, instanceType, bucketName string,
apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
) {
data := bytes.Repeat([]byte("metadata-only-copy-plaintext-"), 64*1024)
want := mustChecksum(t, hash.ChecksumCRC32, data)
object := "copy-metadata/existing-checksum.txt"
putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, data,
map[string]string{xhttp.AmzChecksumCRC32: want})
before, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{})
if err != nil || before.IsCompressed() {
t.Fatalf("%s: invalid metadata-copy precondition: compressed=%v size=%d err=%v",
instanceType, before.IsCompressed(), before.Size, err)
}
restoreCompression := setCopyChecksumCompression(true)
compressionRestored := false
defer func() {
if !compressionRestored {
restoreCompression()
}
}()
rec := copyChecksumRequest(t, apiRouter, credentials, bucketName, object, object,
map[string]string{xhttp.AmzMetadataDirective: "REPLACE"})
if rec.Code != http.StatusOK {
t.Fatalf("%s: metadata-only CopyObject failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
assertCopyChecksum(t, obj, bucketName, object, hash.ChecksumCRC32, data, false, nil)
if got := readCopyChecksumObject(t, obj, bucketName, object, ObjectOptions{}); !bytes.Equal(got, data) {
prefix := got
if len(prefix) > 100 {
prefix = prefix[:100]
}
t.Fatalf("%s: metadata-only CopyObject body differs: got %d bytes, want %d, prefix %q",
instanceType, len(got), len(data), prefix)
}
afterMetadataCopy, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{})
if err != nil {
t.Fatal(err)
}
if before.VersionID != "" && afterMetadataCopy.VersionID == before.VersionID {
t.Fatalf("%s: versioned metadata-only copy did not create a new version", instanceType)
}
destination := "copy-metadata/rewritten.txt"
rec = copyChecksumRequest(t, apiRouter, credentials, bucketName, object, destination, nil)
if rec.Code != http.StatusOK {
t.Fatalf("%s: data-rewriting CopyObject failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
assertCopyChecksum(t, obj, bucketName, destination, hash.ChecksumCRC32, data, true, nil)
compressedObject := "copy-metadata/preserve-compressed.txt"
putCopyChecksumSource(t, apiRouter, credentials, bucketName, compressedObject, data,
map[string]string{xhttp.AmzChecksumCRC32: want})
assertCopyChecksum(t, obj, bucketName, compressedObject, hash.ChecksumCRC32, data, true, nil)
restoreCompression()
compressionRestored = true
rec = copyChecksumRequest(t, apiRouter, credentials, bucketName, compressedObject, compressedObject,
map[string]string{xhttp.AmzMetadataDirective: "REPLACE"})
if rec.Code != http.StatusOK {
t.Fatalf("%s: compressed metadata-only CopyObject failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
assertCopyChecksum(t, obj, bucketName, compressedObject, hash.ChecksumCRC32, data, true, nil)
if got := readCopyChecksumObject(t, obj, bucketName, compressedObject, ObjectOptions{}); !bytes.Equal(got, data) {
t.Fatalf("%s: compressed metadata-only CopyObject body differs", instanceType)
}
}
func TestAPICopyObjectSSECKeyRotationKeepsCompressionState(t *testing.T) {
defer DetectTestLeak(t)()
for _, versioned := range []bool{false, true} {
name := "unversioned"
if versioned {
name = "versioned"
}
t.Run(name, func(t *testing.T) {
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
t: t,
objAPITest: testAPICopyObjectSSECKeyRotationKeepsCompressionState,
endpoints: []string{"CopyObject", "PutObject", "GetObject"},
makeBucketOptions: MakeBucketOptions{VersioningEnabled: versioned},
})
})
}
}
func testAPICopyObjectSSECKeyRotationKeepsCompressionState(obj ObjectLayer, instanceType, bucketName string,
apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
) {
previousTLS := globalIsTLS
globalIsTLS = true
defer func() { globalIsTLS = previousTLS }()
data := bytes.Repeat([]byte("key-rotation-plaintext-"), 64*1024)
object := "copy-metadata/key-rotation.txt"
oldKey := bytes.Repeat([]byte{0x11}, 32)
oldMD5 := md5.Sum(oldKey)
newKey := bytes.Repeat([]byte{0x22}, 32)
newMD5 := md5.Sum(newKey)
putCopyChecksumSource(t, apiRouter, credentials, bucketName, object, data, map[string]string{
xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES,
xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(oldKey),
xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(oldMD5[:]),
})
before, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{})
if err != nil || before.IsCompressed() {
t.Fatalf("%s: invalid key-rotation precondition: compressed=%v err=%v", instanceType, before.IsCompressed(), err)
}
restoreCompression := setCopyChecksumCompression(true)
defer restoreCompression()
rec := copyChecksumRequest(t, apiRouter, credentials, bucketName, object, object, map[string]string{
xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES,
xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(newKey),
xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(newMD5[:]),
xhttp.AmzServerSideEncryptionCopyCustomerAlgorithm: xhttp.AmzEncryptionAES,
xhttp.AmzServerSideEncryptionCopyCustomerKey: base64.StdEncoding.EncodeToString(oldKey),
xhttp.AmzServerSideEncryptionCopyCustomerKeyMD5: base64.StdEncoding.EncodeToString(oldMD5[:]),
})
if rec.Code != http.StatusOK {
t.Fatalf("%s: key rotation failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
after, err := obj.GetObjectInfo(t.Context(), bucketName, object, ObjectOptions{})
if err != nil {
t.Fatal(err)
}
if after.IsCompressed() {
t.Fatalf("%s: metadata-only key rotation stamped compression metadata", instanceType)
}
if before.VersionID != "" && after.VersionID == before.VersionID {
t.Fatalf("%s: versioned key rotation did not create a new version", instanceType)
}
getHeaders := map[string]string{
xhttp.AmzServerSideEncryptionCustomerAlgorithm: xhttp.AmzEncryptionAES,
xhttp.AmzServerSideEncryptionCustomerKey: base64.StdEncoding.EncodeToString(newKey),
xhttp.AmzServerSideEncryptionCustomerKeyMD5: base64.StdEncoding.EncodeToString(newMD5[:]),
}
req, err := newTestSignedRequestV4(http.MethodGet, getGetObjectURL("", bucketName, object),
0, nil, credentials.AccessKey, credentials.SecretKey, getHeaders)
if err != nil {
t.Fatalf("failed to build GetObject request: %v", err)
}
response := httptest.NewRecorder()
apiRouter.ServeHTTP(response, req)
if response.Code != http.StatusOK || !bytes.Equal(response.Body.Bytes(), data) {
t.Fatalf("%s: post-rotation GetObject returned %d with %d bytes, want 200 with %d bytes: %s",
instanceType, response.Code, response.Body.Len(), len(data), response.Body.String())
}
}
+51 -13
View File
@@ -1122,6 +1122,12 @@ func getRemoteInstanceTransport() http.RoundTripper {
return nil
}
// federatedInternalAppName is the minio-go application token that
// getRemoteInstanceClient attaches to every legacy federation proxy request. It
// is declared next to its only producer so that the literal keeps its historical
// file attribution in the rebrand compatibility baseline.
const federatedInternalAppName = "minio-federated"
// Returns a minio-go Client configured to access remote host described by destDNSRecord
// Applicable only in a federated deployment
var getRemoteInstanceClient = func(r *http.Request, host string) (*miniogo.Core, error) {
@@ -1136,7 +1142,7 @@ var getRemoteInstanceClient = func(r *http.Request, host string) (*miniogo.Core,
if err != nil {
return nil, err
}
core.SetAppInfo("minio-federated", ReleaseTag)
core.SetAppInfo(federatedInternalAppName, ReleaseTag)
return core, nil
}
@@ -1357,6 +1363,15 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
} // no changes in storage-class expected so its a metadataonly operation.
var reader io.Reader = gr
sourceCompressMetadata := make(map[string]string, 2)
for _, key := range []string{
ReservedMetadataPrefix + "compression",
ReservedMetadataPrefix + "actual-size",
} {
if value, ok := srcInfo.UserDefined[key]; ok {
sourceCompressMetadata[key] = value
}
}
// Set the actual size to the compressed/decrypted size if encrypted.
actualSize, err := srcInfo.GetActualSize()
@@ -1386,15 +1401,7 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
compressMetadata[ReservedMetadataPrefix+"actual-size"] = strconv.FormatInt(actualSize, 10)
reader = etag.NewReader(ctx, reader, nil, nil)
wantEncryption := crypto.Requested(r.Header)
s2c, cb := newS2CompressReader(reader, actualSize, wantEncryption)
dstOpts.IndexCB = cb
defer s2c.Close()
reader = etag.Wrap(s2c, reader)
length = -1
} else {
delete(srcInfo.UserDefined, ReservedMetadataPrefix+"compression")
delete(srcInfo.UserDefined, ReservedMetadataPrefix+"actual-size")
reader = gr
}
@@ -1541,6 +1548,23 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
}
}
if isDstCompressed {
checksumReader := srcInfo.Reader
wantEncryption := crypto.Requested(r.Header)
s2c, cb := newS2CompressReader(checksumReader, actualSize, wantEncryption)
dstOpts.IndexCB = cb
defer s2c.Close()
reader = etag.Wrap(s2c, checksumReader)
srcInfo.Reader, err = hash.NewReader(ctx, reader, -1, "", "", actualSize)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
// The storage reader consumes compressed data; checksums remain bound to plaintext.
pReader = NewPutObjReader(srcInfo.Reader)
pReader.setChecksumReader(checksumReader)
}
if isTargetEncrypted {
var encReader io.Reader
kind, _ := crypto.IsRequested(r.Header)
@@ -1676,8 +1700,17 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
srcInfo.UserDefined[ReservedMetadataPrefixLower+ReplicationStatus] = dsc.PendingStatus()
srcInfo.UserDefined[ReservedMetadataPrefixLower+ReplicationTimestamp] = UTCNow().Format(time.RFC3339Nano)
}
// Store the preserved compression metadata.
maps.Copy(srcInfo.UserDefined, compressMetadata)
// Compression metadata must describe data that is actually rewritten.
if !srcInfo.metadataOnly || srcInfo.Legacy || dstOpts.WantServerSideChecksumType.IsSet() {
if isDstCompressed {
maps.Copy(srcInfo.UserDefined, compressMetadata)
} else {
delete(srcInfo.UserDefined, ReservedMetadataPrefix+"compression")
delete(srcInfo.UserDefined, ReservedMetadataPrefix+"actual-size")
}
} else {
maps.Copy(srcInfo.UserDefined, sourceCompressMetadata)
}
// We need to preserve the encryption headers set in EncryptRequest,
// so we do not want to override them, copy them instead.
@@ -1767,9 +1800,14 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
copyObjectFn := objectAPI.CopyObject
copySrcOpts := srcOpts
if srcInfo.metadataOnly && dstOpts.Versioned && copySrcOpts.VersionID == "" {
copySrcOpts.VersionID = srcInfo.VersionID
}
// Copy source object to destination, if source and destination
// object is same then only metadata is updated.
objInfo, err = copyObjectFn(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, srcOpts, dstOpts)
objInfo, err = copyObjectFn(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, copySrcOpts, dstOpts)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
@@ -1778,7 +1816,7 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
origETag := objInfo.ETag
objInfo.ETag = getDecryptedETag(r.Header, objInfo, false)
response := generateCopyObjectResponse(objInfo.ETag, objInfo.ModTime)
response := generateCopyObjectResponse(objInfo, r.Header)
encodedSuccessResponse := encodeResponse(response)
if dsc := mustReplicate(ctx, dstBucket, dstObject, objInfo.getMustReplicateOptions(replication.ObjectReplicationType, dstOpts)); dsc.ReplicateAny() {
@@ -0,0 +1,367 @@
// Copyright (c) 2015-2025 MinIO, Inc.
// Copyright (c) 2025-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 cmd
import (
"bytes"
"context"
"crypto/md5"
"encoding/hex"
"encoding/json"
"encoding/xml"
"net/http"
"net/http/httptest"
"strings"
"sync"
"testing"
miniogo "github.com/minio/minio-go/v7"
miniocredentials "github.com/minio/minio-go/v7/pkg/credentials"
"github.com/minio/minio-go/v7/pkg/set"
"github.com/minio/minio/internal/auth"
"github.com/minio/minio/internal/config/dns"
"github.com/minio/minio/internal/hash"
xhttp "github.com/minio/minio/internal/http"
)
const federatedTestUserAgent = "MinIO (linux; amd64) minio-go/v7.0.99 minio-federated/RELEASE.TEST"
func TestAPIFederatedUploadPartChecksumResponse(t *testing.T) {
defer DetectTestLeak(t)()
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
t: t,
objAPITest: testAPIFederatedUploadPartChecksumResponse,
endpoints: []string{"PutObjectPart", "NewMultipart"},
})
}
func testAPIFederatedUploadPartChecksumResponse(_ ObjectLayer, instanceType, bucketName string,
apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
) {
algorithms := []struct {
name string
typ hash.ChecksumType
checksumType string
}{
{name: "crc32-full-object", typ: hash.ChecksumCRC32, checksumType: xhttp.AmzChecksumTypeFullObject},
{name: "sha256-composite", typ: hash.ChecksumSHA256, checksumType: xhttp.AmzChecksumTypeComposite},
}
userAgents := []struct {
name string
ua string
want bool
}{
{name: "absent"},
{name: "ordinary-sdk", ua: "aws-sdk-go/1.55.5"},
{name: "federation", ua: federatedTestUserAgent, want: true},
{name: "lookalike-prefix", ua: "evil-minio-federated/RELEASE.TEST"},
{name: "lookalike-suffix", ua: "minio-federated-extra/RELEASE.TEST"},
{name: "missing-version", ua: "minio-federated"},
{name: "empty-version", ua: "minio-federated/"},
}
data := []byte("federated upload part checksum response")
for _, algorithm := range algorithms {
for _, userAgent := range userAgents {
t.Run(algorithm.name+"/"+userAgent.name, func(t *testing.T) {
object := "federation/response/" + algorithm.name + "/" + userAgent.name
uploadID := newMultipartUploadHTTP(t, apiRouter, credentials, bucketName, object,
algorithm.typ.String(), algorithm.checksumType)
headers := map[string]string{}
if userAgent.ua != "" {
headers["User-Agent"] = userAgent.ua
}
_, rec := uploadPartHTTP(t, apiRouter, credentials,
bucketName, object, uploadID, 1, data, headers)
got := rec.Header().Get(algorithm.typ.Key())
if userAgent.want {
if want := mustChecksum(t, algorithm.typ, data); got != want {
t.Fatalf("%s: checksum %q, want %q", instanceType, got, want)
}
} else if got != "" {
t.Fatalf("%s: ordinary UploadPart exposed server checksum %q", instanceType, got)
}
if got := rec.Header().Get(xhttp.AmzChecksumType); got != "" {
t.Fatalf("%s: UploadPart returned checksum type %q", instanceType, got)
}
})
}
}
}
func TestAPIFederatedUploadPartChecksumMinIOGoWire(t *testing.T) {
defer DetectTestLeak(t)()
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
t: t,
objAPITest: testAPIFederatedUploadPartChecksumMinIOGoWire,
endpoints: []string{"PutObjectPart", "NewMultipart"},
})
}
func testAPIFederatedUploadPartChecksumMinIOGoWire(_ ObjectLayer, instanceType, bucketName string,
apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
) {
server := httptest.NewServer(apiRouter)
defer server.Close()
core, err := miniogo.NewCore(server.Listener.Addr().String(), &miniogo.Options{
Creds: miniocredentials.NewStaticV4(credentials.AccessKey, credentials.SecretKey, ""),
Secure: false,
Region: globalMinioDefaultRegion,
BucketLookup: miniogo.BucketLookupPath,
})
if err != nil {
t.Fatalf("%s: create minio-go Core: %v", instanceType, err)
}
core.SetAppInfo("minio-federated", ReleaseTag)
object := "federation/minio-go-wire"
uploadID := newMultipartUploadHTTP(t, apiRouter, credentials, bucketName, object,
hash.ChecksumCRC32.String(), xhttp.AmzChecksumTypeFullObject)
data := []byte("minio-go must parse the remote computed checksum")
part, err := core.PutObjectPart(t.Context(), bucketName, object, uploadID, 1,
bytes.NewReader(data), int64(len(data)), miniogo.PutObjectPartOptions{})
if err != nil {
t.Fatalf("%s: minio-go PutObjectPart: %v", instanceType, err)
}
if want := mustChecksum(t, hash.ChecksumCRC32, data); part.ChecksumCRC32 != want {
t.Fatalf("%s: minio-go checksum %q, want %q", instanceType, part.ChecksumCRC32, want)
}
if part.ETag == "" {
t.Fatalf("%s: minio-go returned an empty ETag", instanceType)
}
}
func TestAPIFederatedUploadPartChecksumConcurrentOverwrite(t *testing.T) {
defer DetectTestLeak(t)()
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
t: t,
objAPITest: testAPIFederatedUploadPartChecksumConcurrentOverwrite,
endpoints: []string{"PutObjectPart", "NewMultipart"},
})
}
func testAPIFederatedUploadPartChecksumConcurrentOverwrite(_ ObjectLayer, instanceType, bucketName string,
apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
) {
object := "federation/concurrent-overwrite"
uploadID := newMultipartUploadHTTP(t, apiRouter, credentials, bucketName, object,
hash.ChecksumSHA256.String(), xhttp.AmzChecksumTypeComposite)
data := [][]byte{
bytes.Repeat([]byte("first-writer-"), 4096),
bytes.Repeat([]byte("second-writer-"), 4096),
}
reqs := make([]*http.Request, len(data))
recorders := make([]*httptest.ResponseRecorder, len(data))
for i := range data {
req, err := newTestSignedRequestV4(http.MethodPut,
getPutObjectPartURL("", bucketName, object, uploadID, "1"),
int64(len(data[i])), bytes.NewReader(data[i]), credentials.AccessKey, credentials.SecretKey,
map[string]string{"User-Agent": federatedTestUserAgent})
if err != nil {
t.Fatalf("%s: build concurrent request %d: %v", instanceType, i, err)
}
reqs[i] = req
recorders[i] = httptest.NewRecorder()
}
start := make(chan struct{})
var wg sync.WaitGroup
for i := range reqs {
wg.Add(1)
go func() {
defer wg.Done()
<-start
apiRouter.ServeHTTP(recorders[i], reqs[i])
}()
}
close(start)
wg.Wait()
for i, rec := range recorders {
if rec.Code != http.StatusOK {
t.Fatalf("%s: concurrent request %d failed: %d %s", instanceType, i, rec.Code, rec.Body.String())
}
got := rec.Header().Get(hash.ChecksumSHA256.Key())
if want := mustChecksum(t, hash.ChecksumSHA256, data[i]); got != want {
t.Fatalf("%s: concurrent request %d checksum %q, want %q", instanceType, i, got, want)
}
// The ETag and the checksum must describe the same write, so a losing
// writer can never publish the winner's checksum next to its own ETag.
etags := rec.Header()[xhttp.ETag]
if len(etags) != 1 {
t.Fatalf("%s: concurrent request %d returned %d ETags", instanceType, i, len(etags))
}
md5sum := md5.Sum(data[i])
if want := hex.EncodeToString(md5sum[:]); canonicalizeETag(etags[0]) != want {
t.Fatalf("%s: concurrent request %d ETag %q, want %q", instanceType, i, etags[0], want)
}
}
}
// federationTestDNS is a minimal dns.Store so a single test process can play
// both federation roles.
type federationTestDNS struct {
records map[string][]dns.SrvRecord
}
func (f federationTestDNS) Put(string) error { return nil }
func (f federationTestDNS) Get(bucket string) ([]dns.SrvRecord, error) {
records, ok := f.records[bucket]
if !ok {
return nil, dns.ErrNoEntriesFound
}
return records, nil
}
func (f federationTestDNS) Delete(string) error { return nil }
func (f federationTestDNS) List() (map[string][]dns.SrvRecord, error) { return f.records, nil }
func (f federationTestDNS) DeleteRecord(dns.SrvRecord) error { return nil }
func (f federationTestDNS) Close() error { return nil }
func (f federationTestDNS) String() string { return "federation-test-dns" }
// remoteBucketObjectLayer reports one existing bucket as missing so that
// isRemoteCopyRequired takes the legacy federation branch while the same
// process can still serve that bucket as the remote deployment.
type remoteBucketObjectLayer struct {
ObjectLayer
remoteBucket string
}
func (l remoteBucketObjectLayer) GetBucketInfo(ctx context.Context, bucket string, opts BucketOptions) (BucketInfo, error) {
if bucket == l.remoteBucket {
return BucketInfo{}, toObjectErr(errVolumeNotFound, bucket)
}
return l.ObjectLayer.GetBucketInfo(ctx, bucket, opts)
}
// TestAPIFederatedCopyObjectPartChecksum drives the legacy etcd federation
// branch of CopyObjectPartHandler end to end: the proxy forwards the copied
// bytes through the real getRemoteInstanceClient and minio-go, a second HTTP
// endpoint serves the real PutObjectPartHandler, and CopyPartResult must carry
// the checksum computed by that exact remote write.
func TestAPIFederatedCopyObjectPartChecksum(t *testing.T) {
defer DetectTestLeak(t)()
ExecObjectLayerAPITest(ExecObjectLayerAPITestArgs{
t: t,
objAPITest: testAPIFederatedCopyObjectPartChecksum,
endpoints: []string{
"CopyObjectPart", "NewMultipart", "PutObjectPart",
"ListObjectParts", "CompleteMultipart", "PutObject",
},
})
}
func testAPIFederatedCopyObjectPartChecksum(objectAPI ObjectLayer, instanceType, bucketName string,
apiRouter http.Handler, credentials auth.Credentials, t *testing.T,
) {
algorithms := []struct {
name string
typ hash.ChecksumType
checksumType string
}{
{name: "crc32-full-object", typ: hash.ChecksumCRC32, checksumType: xhttp.AmzChecksumTypeFullObject},
{name: "sha256-composite", typ: hash.ChecksumSHA256, checksumType: xhttp.AmzChecksumTypeComposite},
}
data := bytes.Repeat([]byte("federated-upload-part-copy-"), 1024)
srcObject := "federation/copy-source.bin"
putCopyChecksumSource(t, apiRouter, credentials, bucketName, srcObject, data, nil)
// The destination bucket really exists so the remote endpoint can serve it;
// only the proxy's own bucket lookup is told that it lives elsewhere.
remoteBucket := getRandomBucketName()
if err := objectAPI.MakeBucket(t.Context(), remoteBucket, MakeBucketOptions{}); err != nil {
t.Fatalf("%s: unable to create the remote bucket: %v", instanceType, err)
}
remote := httptest.NewServer(apiRouter)
defer remote.Close()
host, port, _ := strings.Cut(remote.Listener.Addr().String(), ":")
globalObjLayerMutex.Lock()
previousLayer := globalObjectAPI
globalObjectAPI = remoteBucketObjectLayer{ObjectLayer: previousLayer, remoteBucket: remoteBucket}
globalObjLayerMutex.Unlock()
previousDNS, previousFederation, previousIPs := globalDNSConfig, globalBucketFederation, globalDomainIPs
globalDNSConfig = federationTestDNS{records: map[string][]dns.SrvRecord{
bucketName: {{Host: host, Port: json.Number(port)}},
remoteBucket: {{Host: host, Port: json.Number(port)}},
}}
// Every DNS record resolves to this process, so the bucket forwarding
// middleware always serves locally and only the handler proxies.
globalDomainIPs = set.CreateStringSet(remote.Listener.Addr().String())
globalBucketFederation = true
defer func() {
globalObjLayerMutex.Lock()
globalObjectAPI = previousLayer
globalObjLayerMutex.Unlock()
globalDNSConfig, globalBucketFederation, globalDomainIPs = previousDNS, previousFederation, previousIPs
}()
for _, algorithm := range algorithms {
t.Run(algorithm.name, func(t *testing.T) {
object := "federation/copy-destination-" + algorithm.name + ".bin"
uploadID := newMultipartUploadHTTP(t, apiRouter, credentials, remoteBucket, object,
algorithm.typ.String(), algorithm.checksumType)
req, err := newTestSignedRequestV4(http.MethodPut,
getCopyObjectPartURL("", remoteBucket, object, uploadID, "1"),
0, nil, credentials.AccessKey, credentials.SecretKey,
map[string]string{xhttp.AmzCopySource: SlashSeparator + pathJoin(bucketName, srcObject)})
if err != nil {
t.Fatalf("%s: unable to build UploadPartCopy request: %v", instanceType, err)
}
rec := httptest.NewRecorder()
apiRouter.ServeHTTP(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("%s: federated UploadPartCopy failed: %d %s", instanceType, rec.Code, rec.Body.String())
}
var response CopyObjectPartResponse
if err := xml.Unmarshal(rec.Body.Bytes(), &response); err != nil {
t.Fatalf("%s: unable to decode CopyPartResult: %v", instanceType, err)
}
want := mustChecksum(t, algorithm.typ, data)
if got := copyPartChecksum(algorithm.typ, response); got != want {
t.Fatalf("%s: CopyPartResult %s is %q, want %q: %s",
instanceType, algorithm.typ.String(), got, want, rec.Body.String())
}
// The persisted part must carry the same value, and the client must be
// able to complete the upload with what CopyPartResult returned.
parts := listPartsHTTP(t, apiRouter, credentials, remoteBucket, object, uploadID, nil)
if len(parts.Parts) != 1 {
t.Fatalf("%s: ListParts returned %d parts, want 1", instanceType, len(parts.Parts))
}
if got := partChecksum(algorithm.typ, parts.Parts[0]); got != want {
t.Fatalf("%s: persisted part %s is %q, want %q", instanceType, algorithm.typ.String(), got, want)
}
etag := canonicalizeETag(response.ETag)
completed := completePartsHTTP(t, apiRouter, credentials, remoteBucket, object, uploadID,
[]CompletePart{completePartWithChecksum(algorithm.typ, 1, etag, want)}, nil)
if completed.Code != http.StatusOK {
t.Fatalf("%s: CompleteMultipartUpload rejected the federated part: %d %s",
instanceType, completed.Code, completed.Body.String())
}
})
}
}
+127 -2
View File
@@ -55,6 +55,88 @@ import (
// Multipart objectAPIHandlers
// isFederatedInternalRequest reports whether User-Agent carries the minio-go
// application token attached by getRemoteInstanceClient.
//
// This is only a response-shape hint. User-Agent is not authenticated and must
// never gate authorization, object visibility, or request validation. It is
// safe here because the only effect is returning the checksum of the body the
// caller was already authorized to upload.
func isFederatedInternalRequest(userAgent string) bool {
for _, product := range strings.Fields(userAgent) {
name, version, ok := strings.Cut(product, "/")
if ok && name == federatedInternalAppName && version != "" {
return true
}
}
return false
}
// partChecksumMap returns the non-empty part checksums in the form expected by
// hash.AddChecksumHeader. x-amz-checksum-type is deliberately excluded because
// UploadPart does not return it and minio-go cannot carry it in ObjectPart.
func partChecksumMap(partInfo PartInfo) map[string]string {
checksums := make(map[string]string, 1)
if partInfo.ChecksumCRC32 != "" {
checksums[hash.ChecksumCRC32.String()] = partInfo.ChecksumCRC32
}
if partInfo.ChecksumCRC32C != "" {
checksums[hash.ChecksumCRC32C.String()] = partInfo.ChecksumCRC32C
}
if partInfo.ChecksumSHA1 != "" {
checksums[hash.ChecksumSHA1.String()] = partInfo.ChecksumSHA1
}
if partInfo.ChecksumSHA256 != "" {
checksums[hash.ChecksumSHA256.String()] = partInfo.ChecksumSHA256
}
if partInfo.ChecksumCRC64NVME != "" {
checksums[hash.ChecksumCRC64NVME.String()] = partInfo.ChecksumCRC64NVME
}
return checksums
}
// multipartChecksumType returns the base checksum type recorded when a
// multipart upload was created. The boolean reports whether an algorithm was
// recorded at all.
func multipartChecksumType(metadata map[string]string) (hash.ChecksumType, bool) {
algorithm := metadata[hash.MinIOMultipartChecksum]
if algorithm == "" {
return hash.ChecksumNone, false
}
t := hash.NewChecksumType(algorithm, metadata[hash.MinIOMultipartChecksumType])
if !t.IsSet() {
return t, true
}
return t.Base(), true
}
// prepareMultipartChecksumReader validates a supplied part checksum algorithm,
// or installs a server-side hasher when the client omitted the optional
// checksum. It must run before compression or encryption can consume reader.
func prepareMultipartChecksumReader(reader *hash.Reader, metadata map[string]string, bucket, object string) error {
want, ok := multipartChecksumType(metadata)
if !ok {
return nil
}
got := reader.ContentCRCType()
if !got.IsSet() && reader.ServerSideChecksumType.IsSet() {
got = reader.ServerSideChecksumType
}
if !want.IsSet() || (got.IsSet() && got.Base() != want) {
return InvalidArgument{
Bucket: bucket,
Object: object,
Err: fmt.Errorf("checksum missing, want %q, got %q",
metadata[hash.MinIOMultipartChecksum], got.String()),
}
}
if !got.IsSet() {
reader.AddServerSideChecksumHasher(want)
}
return nil
}
// NewMultipartUploadHandler - New multipart upload.
// Notice: The S3 client can send secret keys in headers for encryption related jobs,
// the handler should ensure to remove these keys before sending them to the object layer.
@@ -465,7 +547,15 @@ func (api objectAPIHandlers) CopyObjectPartHandler(w http.ResponseWriter, r *htt
return
}
response := generateCopyObjectPartResponse(partInfo.ETag, partInfo.LastModified)
response := generateCopyObjectPartResponse(PartInfo{
ETag: partInfo.ETag,
LastModified: partInfo.LastModified,
ChecksumCRC32: partInfo.ChecksumCRC32,
ChecksumCRC32C: partInfo.ChecksumCRC32C,
ChecksumSHA1: partInfo.ChecksumSHA1,
ChecksumSHA256: partInfo.ChecksumSHA256,
ChecksumCRC64NVME: partInfo.ChecksumCRC64NVME,
})
encodedSuccessResponse := encodeResponse(response)
// Write success response.
@@ -475,12 +565,25 @@ func (api objectAPIHandlers) CopyObjectPartHandler(w http.ResponseWriter, r *htt
actualPartSize = length
var reader io.Reader = etag.NewReader(ctx, gr, nil, nil)
var checksumReader *hash.Reader
mi, err := objectAPI.GetMultipartInfo(ctx, dstBucket, dstObject, uploadID, dstOpts)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
if _, ok := multipartChecksumType(mi.UserDefined); ok {
checksumReader, err = hash.NewReader(ctx, reader, length, "", "", actualPartSize)
if err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
if err = prepareMultipartChecksumReader(checksumReader, mi.UserDefined, dstBucket, dstObject); err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
reader = checksumReader
}
_, isEncrypted := crypto.IsEncrypted(mi.UserDefined)
@@ -512,6 +615,7 @@ func (api objectAPIHandlers) CopyObjectPartHandler(w http.ResponseWriter, r *htt
rawReader := srcInfo.Reader
pReader := NewPutObjReader(rawReader)
pReader.setChecksumReader(checksumReader)
var objectEncryptionKey crypto.ObjectKey
if isEncrypted {
@@ -591,7 +695,7 @@ func (api objectAPIHandlers) CopyObjectPartHandler(w http.ResponseWriter, r *htt
partInfo.ETag = tryDecryptETag(objectEncryptionKey[:], partInfo.ETag, sseS3)
}
response := generateCopyObjectPartResponse(partInfo.ETag, partInfo.LastModified)
response := generateCopyObjectPartResponse(partInfo)
encodedSuccessResponse := encodeResponse(response)
// Write success response.
@@ -745,6 +849,7 @@ func (api objectAPIHandlers) PutObjectPartHandler(w http.ResponseWriter, r *http
// Read compression metadata preserved in the init multipart for the decision.
_, isCompressed := mi.UserDefined[ReservedMetadataPrefix+"compression"]
var idxCb func() []byte
var checksumReader *hash.Reader
if isCompressed {
actualReader, err := hash.NewReader(ctx, reader, size, md5hex, sha256hex, actualSize)
if err != nil {
@@ -755,6 +860,11 @@ func (api objectAPIHandlers) PutObjectPartHandler(w http.ResponseWriter, r *http
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidChecksum), r.URL)
return
}
if err = prepareMultipartChecksumReader(actualReader, mi.UserDefined, bucket, object); err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
checksumReader = actualReader
// Set compression metrics.
wantEncryption := crypto.Requested(r.Header)
@@ -791,8 +901,16 @@ func (api objectAPIHandlers) PutObjectPartHandler(w http.ResponseWriter, r *http
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidChecksum), r.URL)
return
}
if checksumReader == nil {
if err = prepareMultipartChecksumReader(hashReader, mi.UserDefined, bucket, object); err != nil {
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
return
}
checksumReader = hashReader
}
pReader := NewPutObjReader(hashReader)
pReader.setChecksumReader(checksumReader)
_, isEncrypted := crypto.IsEncrypted(mi.UserDefined)
_, replicationStatus := mi.UserDefined[xhttp.AmzBucketReplicationStatus]
@@ -918,6 +1036,13 @@ func (api objectAPIHandlers) PutObjectPartHandler(w http.ResponseWriter, r *http
// Therefore, we have to set the ETag directly as map entry.
w.Header()[xhttp.ETag] = []string{"\"" + etag + "\""}
hash.TransferChecksumHeader(w, r)
if isFederatedInternalRequest(r.UserAgent()) {
// Legacy federation proxies UploadPartCopy through minio-go
// Core.PutObjectPart, which can only recover checksums from response
// headers. Use the PartInfo returned by this exact write so the ETag and
// checksum cannot be mixed with a concurrent overwrite.
hash.AddChecksumHeader(w, partChecksumMap(partInfo))
}
writeSuccessResponseHeadersOnly(w)
}
-6
View File
@@ -80,8 +80,6 @@ func setupTestReadDirFiles(t *testing.T) (testResults []result) {
for i := range 10 {
name := fmt.Sprintf("file-%d", i)
if err := os.WriteFile(filepath.Join(dir, name), []byte{}, os.ModePerm); err != nil {
// For cleanup, its required to add these entries into test results.
testResults = append(testResults, result{dir, entries})
t.Fatalf("Unable to create file, %s", err)
}
entries = append(entries, name)
@@ -105,8 +103,6 @@ func setupTestReadDirGeneric(t *testing.T) (testResults []result) {
for i := range 10 {
name := fmt.Sprintf("file-%d", i)
if err := os.WriteFile(filepath.Join(dir, "mydir", name), []byte{}, os.ModePerm); err != nil {
// For cleanup, its required to add these entries into test results.
testResults = append(testResults, result{dir, entries})
t.Fatalf("Unable to write file, %s", err)
}
}
@@ -130,8 +126,6 @@ func setupTestReadDirSymlink(t *testing.T) (testResults []result) {
name1 := fmt.Sprintf("file-%d", i)
name2 := fmt.Sprintf("file-%d", i+10)
if err := os.WriteFile(filepath.Join(dir, name1), []byte{}, os.ModePerm); err != nil {
// For cleanup, its required to add these entries into test results.
testResults = append(testResults, result{dir, entries})
t.Fatalf("Unable to create a file, %s", err)
}
// Symlink will not be added to entries.
+15 -3
View File
@@ -48,6 +48,7 @@ import (
"github.com/minio/minio/internal/color"
"github.com/minio/minio/internal/config"
"github.com/minio/minio/internal/config/api"
"github.com/minio/minio/internal/config/notify"
"github.com/minio/minio/internal/handlers"
"github.com/minio/minio/internal/hash/sha256"
xhttp "github.com/minio/minio/internal/http"
@@ -535,6 +536,15 @@ func configRetriableErrors(err error) bool {
notInitialized
}
func fatalServerConfigError(err error) bool {
var configErr config.Err
if errors.As(err, &configErr) {
return true
}
var migrationErr *notify.LegacyDatabaseTargetError
return errors.As(err, &migrationErr)
}
func bootstrapTraceMsg(msg string) {
info := madmin.TraceInfo{
TraceType: madmin.TraceBootstrap,
@@ -632,7 +642,10 @@ func initConfigSubsystem(ctx context.Context, newObject ObjectLayer) error {
// Initialize config system.
if err := globalConfigSys.Init(newObject); err != nil {
if configRetriableErrors(err) {
var migrationErr *notify.LegacyDatabaseTargetError
// Do not use fatalServerConfigError here: existing config.Err values
// retain the historical log-and-continue behavior at this boundary.
if configRetriableErrors(err) || errors.As(err, &migrationErr) {
return fmt.Errorf("Unable to initialize config system: %w", err)
}
@@ -965,10 +978,9 @@ func serverMain(ctx *cli.Context) {
var err error
bootstrapTrace("initServerConfig", func() {
if err = initServerConfig(GlobalContext, newObject); err != nil {
var cerr config.Err
// For any config error, we don't need to drop into safe-mode
// instead its a user error and should be fixed by user.
if errors.As(err, &cerr) {
if fatalServerConfigError(err) {
logger.FatalIf(err, "Unable to initialize the server")
}
+2
View File
@@ -46,7 +46,9 @@ func oldLinux() bool {
func setMaxResources(ctx serverCtxt) (err error) {
// Set the Go runtime max threads threshold to 90% of kernel setting.
//nolint:staticcheck // Linux implementations can fail; BSD stubs return a constant nil error.
sysMaxThreads, err := sys.GetMaxThreads()
//nolint:staticcheck // Keep the shared cross-platform error handling.
if err == nil {
minioMaxThreads := (sysMaxThreads * 90) / 100
// Only set max threads if it is greater than the default one
+1 -1
View File
@@ -3734,7 +3734,7 @@ func (c *SiteReplicationSys) SiteReplicationMetaInfo(ctx context.Context, objAPI
bms.ExpiryLCConfig = &expLclCfgStr
// if all non expiry rules only, ExpiryUpdatedAt would be nil
if meta.lifecycleConfig.ExpiryUpdatedAt != nil {
bms.ExpiryLCConfigUpdatedAt = *(meta.lifecycleConfig.ExpiryUpdatedAt)
bms.ExpiryLCConfigUpdatedAt = *meta.lifecycleConfig.ExpiryUpdatedAt
}
}
+5 -5
View File
@@ -50,7 +50,7 @@ func (x *xlMetaV2VersionHeader) unmarshalV1(bts []byte) (o []byte, err error) {
err = msgp.ArrayError{Wanted: 4, Got: zb0001}
return o, err
}
bts, err = msgp.ReadExactBytes(bts, (x.VersionID)[:])
bts, err = msgp.ReadExactBytes(bts, x.VersionID[:])
if err != nil {
err = msgp.WrapError(err, "VersionID")
return o, err
@@ -145,7 +145,7 @@ func (z *xlMetaV2VersionHeaderV2) UnmarshalMsg(bts []byte) (o []byte, err error)
err = msgp.ArrayError{Wanted: 5, Got: zb0001}
return o, err
}
bts, err = msgp.ReadExactBytes(bts, (z.VersionID)[:])
bts, err = msgp.ReadExactBytes(bts, z.VersionID[:])
if err != nil {
err = msgp.WrapError(err, "VersionID")
return o, err
@@ -155,7 +155,7 @@ func (z *xlMetaV2VersionHeaderV2) UnmarshalMsg(bts []byte) (o []byte, err error)
err = msgp.WrapError(err, "ModTime")
return o, err
}
bts, err = msgp.ReadExactBytes(bts, (z.Signature)[:])
bts, err = msgp.ReadExactBytes(bts, z.Signature[:])
if err != nil {
err = msgp.WrapError(err, "Signature")
return o, err
@@ -195,7 +195,7 @@ func (z *xlMetaV2VersionHeaderV2) DecodeMsg(dc *msgp.Reader) (err error) {
err = msgp.ArrayError{Wanted: 5, Got: zb0001}
return err
}
err = dc.ReadExactBytes((z.VersionID)[:])
err = dc.ReadExactBytes(z.VersionID[:])
if err != nil {
err = msgp.WrapError(err, "VersionID")
return err
@@ -205,7 +205,7 @@ func (z *xlMetaV2VersionHeaderV2) DecodeMsg(dc *msgp.Reader) (err error) {
err = msgp.WrapError(err, "ModTime")
return err
}
err = dc.ReadExactBytes((z.Signature)[:])
err = dc.ReadExactBytes(z.Signature[:])
if err != nil {
err = msgp.WrapError(err, "Signature")
return err
+1 -1
View File
@@ -1101,7 +1101,7 @@ func (x *xlMetaV2) loadLegacy(buf []byte) error {
return msgp.WrapError(err, "Versions")
}
if cap(x.versions) >= int(zb0002) {
x.versions = (x.versions)[:zb0002]
x.versions = x.versions[:zb0002]
} else {
x.versions = make([]xlMetaV2ShallowVersion, zb0002, zb0002+1)
}