mirror of https://github.com/minio/minio.git
1058 lines
34 KiB
Go
1058 lines
34 KiB
Go
// Copyright (c) 2015-2021 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"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"encoding/xml"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"io/ioutil"
|
|
"net/http"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/gorilla/mux"
|
|
jsoniter "github.com/json-iterator/go"
|
|
"github.com/klauspost/compress/zip"
|
|
"github.com/minio/kes"
|
|
"github.com/minio/madmin-go"
|
|
"github.com/minio/minio-go/v7/pkg/tags"
|
|
"github.com/minio/minio/internal/bucket/lifecycle"
|
|
objectlock "github.com/minio/minio/internal/bucket/object/lock"
|
|
"github.com/minio/minio/internal/bucket/versioning"
|
|
"github.com/minio/minio/internal/event"
|
|
"github.com/minio/minio/internal/kms"
|
|
"github.com/minio/minio/internal/logger"
|
|
"github.com/minio/pkg/bucket/policy"
|
|
iampolicy "github.com/minio/pkg/iam/policy"
|
|
)
|
|
|
|
const (
|
|
bucketQuotaConfigFile = "quota.json"
|
|
bucketTargetsFile = "bucket-targets.json"
|
|
)
|
|
|
|
// PutBucketQuotaConfigHandler - PUT Bucket quota configuration.
|
|
// ----------
|
|
// Places a quota configuration on the specified bucket. The quota
|
|
// specified in the quota configuration will be applied by default
|
|
// to enforce total quota for the specified bucket.
|
|
func (a adminAPIHandlers) PutBucketQuotaConfigHandler(w http.ResponseWriter, r *http.Request) {
|
|
ctx := newContext(r, w, "PutBucketQuotaConfig")
|
|
|
|
defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
|
|
|
|
objectAPI, _ := validateAdminReq(ctx, w, r, iampolicy.SetBucketQuotaAdminAction)
|
|
if objectAPI == nil {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
|
|
return
|
|
}
|
|
|
|
vars := mux.Vars(r)
|
|
bucket := pathClean(vars["bucket"])
|
|
|
|
if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
|
|
writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
|
|
return
|
|
}
|
|
|
|
data, err := ioutil.ReadAll(r.Body)
|
|
if err != nil {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrInvalidRequest), r.URL)
|
|
return
|
|
}
|
|
|
|
quotaConfig, err := parseBucketQuota(bucket, data)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
|
|
return
|
|
}
|
|
|
|
if quotaConfig.Type == "fifo" {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrInvalidRequest), r.URL)
|
|
return
|
|
}
|
|
|
|
if err = globalBucketMetadataSys.Update(ctx, bucket, bucketQuotaConfigFile, data); err != nil {
|
|
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
|
|
return
|
|
}
|
|
|
|
bucketMeta := madmin.SRBucketMeta{
|
|
Type: madmin.SRBucketMetaTypeQuotaConfig,
|
|
Bucket: bucket,
|
|
Quota: data,
|
|
}
|
|
if quotaConfig.Quota == 0 {
|
|
bucketMeta.Quota = nil
|
|
}
|
|
|
|
// Call site replication hook.
|
|
if err = globalSiteReplicationSys.BucketMetaHook(ctx, bucketMeta); err != nil {
|
|
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
|
|
return
|
|
}
|
|
|
|
// Write success response.
|
|
writeSuccessResponseHeadersOnly(w)
|
|
}
|
|
|
|
// GetBucketQuotaConfigHandler - gets bucket quota configuration
|
|
func (a adminAPIHandlers) GetBucketQuotaConfigHandler(w http.ResponseWriter, r *http.Request) {
|
|
ctx := newContext(r, w, "GetBucketQuotaConfig")
|
|
|
|
defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
|
|
|
|
objectAPI, _ := validateAdminReq(ctx, w, r, iampolicy.GetBucketQuotaAdminAction)
|
|
if objectAPI == nil {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
|
|
return
|
|
}
|
|
|
|
vars := mux.Vars(r)
|
|
bucket := pathClean(vars["bucket"])
|
|
|
|
if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
|
|
writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
|
|
return
|
|
}
|
|
|
|
config, _, err := globalBucketMetadataSys.GetQuotaConfig(ctx, bucket)
|
|
if err != nil {
|
|
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
|
|
return
|
|
}
|
|
|
|
configData, err := json.Marshal(config)
|
|
if err != nil {
|
|
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
|
|
return
|
|
}
|
|
|
|
// Write success response.
|
|
writeSuccessResponseJSON(w, configData)
|
|
}
|
|
|
|
// SetRemoteTargetHandler - sets a remote target for bucket
|
|
func (a adminAPIHandlers) SetRemoteTargetHandler(w http.ResponseWriter, r *http.Request) {
|
|
ctx := newContext(r, w, "SetBucketTarget")
|
|
|
|
defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
|
|
vars := mux.Vars(r)
|
|
bucket := pathClean(vars["bucket"])
|
|
update := r.Form.Get("update") == "true"
|
|
|
|
if globalIsGateway {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL)
|
|
return
|
|
}
|
|
|
|
// Get current object layer instance.
|
|
objectAPI, _ := validateAdminReq(ctx, w, r, iampolicy.SetBucketTargetAction)
|
|
if objectAPI == nil {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
|
|
return
|
|
}
|
|
|
|
// Check if bucket exists.
|
|
if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
|
|
writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
|
|
return
|
|
}
|
|
|
|
cred, _, _, s3Err := validateAdminSignature(ctx, r, "")
|
|
if s3Err != ErrNone {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL)
|
|
return
|
|
}
|
|
password := cred.SecretKey
|
|
|
|
reqBytes, err := madmin.DecryptData(password, io.LimitReader(r.Body, r.ContentLength))
|
|
if err != nil {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErrWithErr(ErrAdminConfigBadJSON, err), r.URL)
|
|
return
|
|
}
|
|
var target madmin.BucketTarget
|
|
json := jsoniter.ConfigCompatibleWithStandardLibrary
|
|
if err = json.Unmarshal(reqBytes, &target); err != nil {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErrWithErr(ErrAdminConfigBadJSON, err), r.URL)
|
|
return
|
|
}
|
|
sameTarget, _ := isLocalHost(target.URL().Hostname(), target.URL().Port(), globalMinioPort)
|
|
if sameTarget && bucket == target.TargetBucket {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrBucketRemoteIdenticalToSource), r.URL)
|
|
return
|
|
}
|
|
|
|
target.SourceBucket = bucket
|
|
var ops []madmin.TargetUpdateType
|
|
if update {
|
|
ops = madmin.GetTargetUpdateOps(r.Form)
|
|
} else {
|
|
target.Arn = globalBucketTargetSys.getRemoteARN(bucket, &target)
|
|
}
|
|
if target.Arn == "" {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErrWithErr(ErrAdminConfigBadJSON, err), r.URL)
|
|
return
|
|
}
|
|
if update {
|
|
// overlay the updates on existing target
|
|
tgt := globalBucketTargetSys.GetRemoteBucketTargetByArn(ctx, bucket, target.Arn)
|
|
if tgt.Empty() {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErrWithErr(ErrRemoteTargetNotFoundError, err), r.URL)
|
|
return
|
|
}
|
|
for _, op := range ops {
|
|
switch op {
|
|
case madmin.CredentialsUpdateType:
|
|
tgt.Credentials = target.Credentials
|
|
tgt.TargetBucket = target.TargetBucket
|
|
tgt.Secure = target.Secure
|
|
tgt.Endpoint = target.Endpoint
|
|
case madmin.SyncUpdateType:
|
|
tgt.ReplicationSync = target.ReplicationSync
|
|
case madmin.ProxyUpdateType:
|
|
tgt.DisableProxy = target.DisableProxy
|
|
case madmin.PathUpdateType:
|
|
tgt.Path = target.Path
|
|
case madmin.BandwidthLimitUpdateType:
|
|
tgt.BandwidthLimit = target.BandwidthLimit
|
|
case madmin.HealthCheckDurationUpdateType:
|
|
tgt.HealthCheckDuration = target.HealthCheckDuration
|
|
}
|
|
}
|
|
target = tgt
|
|
}
|
|
|
|
// enforce minimum bandwidth limit as 100MBps
|
|
if target.BandwidthLimit > 0 && target.BandwidthLimit < 100*1000*1000 {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErrWithErr(ErrReplicationBandwidthLimitError, err), r.URL)
|
|
return
|
|
}
|
|
if err = globalBucketTargetSys.SetTarget(ctx, bucket, &target, update); err != nil {
|
|
switch err.(type) {
|
|
case BucketRemoteConnectionErr:
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErrWithErr(ErrReplicationRemoteConnectionError, err), r.URL)
|
|
default:
|
|
writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
|
|
}
|
|
return
|
|
}
|
|
targets, err := globalBucketTargetSys.ListBucketTargets(ctx, bucket)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
|
|
return
|
|
}
|
|
tgtBytes, err := json.Marshal(&targets)
|
|
if err != nil {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErrWithErr(ErrAdminConfigBadJSON, err), r.URL)
|
|
return
|
|
}
|
|
if err = globalBucketMetadataSys.Update(ctx, bucket, bucketTargetsFile, tgtBytes); err != nil {
|
|
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
|
|
return
|
|
}
|
|
|
|
data, err := json.Marshal(target.Arn)
|
|
if err != nil {
|
|
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
|
|
return
|
|
}
|
|
// Write success response.
|
|
writeSuccessResponseJSON(w, data)
|
|
}
|
|
|
|
// ListRemoteTargetsHandler - lists remote target(s) for a bucket or gets a target
|
|
// for a particular ARN type
|
|
func (a adminAPIHandlers) ListRemoteTargetsHandler(w http.ResponseWriter, r *http.Request) {
|
|
ctx := newContext(r, w, "ListBucketTargets")
|
|
|
|
defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
|
|
vars := mux.Vars(r)
|
|
bucket := pathClean(vars["bucket"])
|
|
arnType := vars["type"]
|
|
|
|
if globalIsGateway {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL)
|
|
return
|
|
}
|
|
// Get current object layer instance.
|
|
objectAPI, _ := validateAdminReq(ctx, w, r, iampolicy.GetBucketTargetAction)
|
|
if objectAPI == nil {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
|
|
return
|
|
}
|
|
if bucket != "" {
|
|
// Check if bucket exists.
|
|
if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
|
|
writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
|
|
return
|
|
}
|
|
if _, err := globalBucketMetadataSys.GetBucketTargetsConfig(bucket); err != nil {
|
|
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
|
|
return
|
|
}
|
|
}
|
|
targets := globalBucketTargetSys.ListTargets(ctx, bucket, arnType)
|
|
data, err := json.Marshal(targets)
|
|
if err != nil {
|
|
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
|
|
return
|
|
}
|
|
// Write success response.
|
|
writeSuccessResponseJSON(w, data)
|
|
}
|
|
|
|
// RemoveRemoteTargetHandler - removes a remote target for bucket with specified ARN
|
|
func (a adminAPIHandlers) RemoveRemoteTargetHandler(w http.ResponseWriter, r *http.Request) {
|
|
ctx := newContext(r, w, "RemoveBucketTarget")
|
|
|
|
defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
|
|
vars := mux.Vars(r)
|
|
bucket := pathClean(vars["bucket"])
|
|
arn := vars["arn"]
|
|
|
|
if globalIsGateway {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL)
|
|
return
|
|
}
|
|
// Get current object layer instance.
|
|
objectAPI, _ := validateAdminReq(ctx, w, r, iampolicy.SetBucketTargetAction)
|
|
if objectAPI == nil {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
|
|
return
|
|
}
|
|
|
|
// Check if bucket exists.
|
|
if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
|
|
writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
|
|
return
|
|
}
|
|
|
|
if err := globalBucketTargetSys.RemoveTarget(ctx, bucket, arn); err != nil {
|
|
writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
|
|
return
|
|
}
|
|
targets, err := globalBucketTargetSys.ListBucketTargets(ctx, bucket)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
|
|
return
|
|
}
|
|
tgtBytes, err := json.Marshal(&targets)
|
|
if err != nil {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErrWithErr(ErrAdminConfigBadJSON, err), r.URL)
|
|
return
|
|
}
|
|
if err = globalBucketMetadataSys.Update(ctx, bucket, bucketTargetsFile, tgtBytes); err != nil {
|
|
writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL)
|
|
return
|
|
}
|
|
|
|
// Write success response.
|
|
writeSuccessNoContent(w)
|
|
}
|
|
|
|
// ExportBucketMetadataHandler - exports all bucket metadata as a zipped file
|
|
func (a adminAPIHandlers) ExportBucketMetadataHandler(w http.ResponseWriter, r *http.Request) {
|
|
ctx := newContext(r, w, "ExportBucketMetadata")
|
|
defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
|
|
|
|
bucket := pathClean(r.Form.Get("bucket"))
|
|
if !globalIsErasure {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL)
|
|
return
|
|
}
|
|
// Get current object layer instance.
|
|
objectAPI, _ := validateAdminReq(ctx, w, r, iampolicy.ExportBucketMetadataAction)
|
|
if objectAPI == nil {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
|
|
return
|
|
}
|
|
var (
|
|
buckets []BucketInfo
|
|
err error
|
|
)
|
|
if bucket != "" {
|
|
// Check if bucket exists.
|
|
if _, err := objectAPI.GetBucketInfo(ctx, bucket); err != nil {
|
|
writeErrorResponseJSON(ctx, w, toAPIError(ctx, err), r.URL)
|
|
return
|
|
}
|
|
buckets = append(buckets, BucketInfo{Name: bucket})
|
|
} else {
|
|
buckets, err = objectAPI.ListBuckets(ctx)
|
|
if err != nil {
|
|
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
|
|
return
|
|
}
|
|
}
|
|
|
|
// Initialize a zip writer which will provide a zipped content
|
|
// of bucket metadata
|
|
zipWriter := zip.NewWriter(w)
|
|
defer zipWriter.Close()
|
|
rawDataFn := func(r io.Reader, filename string, sz int) error {
|
|
header, zerr := zip.FileInfoHeader(dummyFileInfo{
|
|
name: filename,
|
|
size: int64(sz),
|
|
mode: 0o600,
|
|
modTime: time.Now(),
|
|
isDir: false,
|
|
sys: nil,
|
|
})
|
|
if zerr != nil {
|
|
logger.LogIf(ctx, zerr)
|
|
return nil
|
|
}
|
|
header.Method = zip.Deflate
|
|
zwriter, zerr := zipWriter.CreateHeader(header)
|
|
if zerr != nil {
|
|
logger.LogIf(ctx, zerr)
|
|
return nil
|
|
}
|
|
if _, err := io.Copy(zwriter, r); err != nil {
|
|
logger.LogIf(ctx, err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
cfgFiles := []string{
|
|
bucketPolicyConfig,
|
|
bucketNotificationConfig,
|
|
bucketLifecycleConfig,
|
|
bucketSSEConfig,
|
|
bucketTaggingConfig,
|
|
bucketQuotaConfigFile,
|
|
objectLockConfig,
|
|
bucketVersioningConfig,
|
|
bucketReplicationConfig,
|
|
bucketTargetsFile,
|
|
}
|
|
for _, bi := range buckets {
|
|
for _, cfgFile := range cfgFiles {
|
|
cfgPath := pathJoin(bi.Name, cfgFile)
|
|
bucket := bi.Name
|
|
switch cfgFile {
|
|
case bucketNotificationConfig:
|
|
config, err := globalBucketMetadataSys.GetNotificationConfig(bucket)
|
|
if err != nil {
|
|
logger.LogIf(ctx, err)
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
configData, err := xml.Marshal(config)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
if err = rawDataFn(bytes.NewReader(configData), cfgPath, len(configData)); err != nil {
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
case bucketLifecycleConfig:
|
|
config, err := globalBucketMetadataSys.GetLifecycleConfig(bucket)
|
|
if err != nil {
|
|
if errors.Is(err, BucketLifecycleNotFound{Bucket: bucket}) {
|
|
continue
|
|
}
|
|
logger.LogIf(ctx, err)
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
configData, err := xml.Marshal(config)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
if err = rawDataFn(bytes.NewReader(configData), cfgPath, len(configData)); err != nil {
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
case bucketQuotaConfigFile:
|
|
config, _, err := globalBucketMetadataSys.GetQuotaConfig(ctx, bucket)
|
|
if err != nil {
|
|
if errors.Is(err, BucketQuotaConfigNotFound{Bucket: bucket}) {
|
|
continue
|
|
}
|
|
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
|
|
return
|
|
}
|
|
configData, err := json.Marshal(config)
|
|
if err != nil {
|
|
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
|
|
return
|
|
}
|
|
if err = rawDataFn(bytes.NewReader(configData), cfgPath, len(configData)); err != nil {
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
case bucketSSEConfig:
|
|
config, _, err := globalBucketMetadataSys.GetSSEConfig(bucket)
|
|
if err != nil {
|
|
if errors.Is(err, BucketSSEConfigNotFound{Bucket: bucket}) {
|
|
continue
|
|
}
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
configData, err := xml.Marshal(config)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
if err = rawDataFn(bytes.NewReader(configData), cfgPath, len(configData)); err != nil {
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
case bucketTaggingConfig:
|
|
config, _, err := globalBucketMetadataSys.GetTaggingConfig(bucket)
|
|
if err != nil {
|
|
if errors.Is(err, BucketTaggingNotFound{Bucket: bucket}) {
|
|
continue
|
|
}
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
configData, err := xml.Marshal(config)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
if err = rawDataFn(bytes.NewReader(configData), cfgPath, len(configData)); err != nil {
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
case objectLockConfig:
|
|
config, _, err := globalBucketMetadataSys.GetObjectLockConfig(bucket)
|
|
if err != nil {
|
|
if errors.Is(err, BucketObjectLockConfigNotFound{Bucket: bucket}) {
|
|
continue
|
|
}
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
|
|
configData, err := xml.Marshal(config)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
if err = rawDataFn(bytes.NewReader(configData), cfgPath, len(configData)); err != nil {
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
case bucketVersioningConfig:
|
|
config, _, err := globalBucketMetadataSys.GetVersioningConfig(bucket)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
// ignore empty versioning configs
|
|
if config.Status != versioning.Enabled && config.Status != versioning.Suspended {
|
|
continue
|
|
}
|
|
configData, err := xml.Marshal(config)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
if err = rawDataFn(bytes.NewReader(configData), cfgPath, len(configData)); err != nil {
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
case bucketReplicationConfig:
|
|
config, _, err := globalBucketMetadataSys.GetReplicationConfig(ctx, bucket)
|
|
if err != nil {
|
|
if errors.Is(err, BucketReplicationConfigNotFound{Bucket: bucket}) {
|
|
continue
|
|
}
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
configData, err := xml.Marshal(config)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
|
|
if err = rawDataFn(bytes.NewReader(configData), cfgPath, len(configData)); err != nil {
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
case bucketTargetsFile:
|
|
config, err := globalBucketMetadataSys.GetBucketTargetsConfig(bucket)
|
|
if err != nil {
|
|
if errors.Is(err, BucketRemoteTargetNotFound{Bucket: bucket}) {
|
|
continue
|
|
}
|
|
|
|
writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL)
|
|
return
|
|
}
|
|
configData, err := xml.Marshal(config)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
if err = rawDataFn(bytes.NewReader(configData), cfgPath, len(configData)); err != nil {
|
|
writeErrorResponse(ctx, w, exportError(ctx, err, cfgFile, bucket), r.URL)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// ImportBucketMetadataHandler - imports all bucket metadata from a zipped file and overwrite bucket metadata config
|
|
// There are some caveats regarding the following:
|
|
// 1. object lock config - object lock should have been specified at time of bucket creation. Only default retention settings are imported here.
|
|
// 2. Replication config - is omitted from import as remote target credentials are not available from exported data for security reasons.
|
|
// 3. lifecycle config - if transition rules are present, tier name needs to have been defined.
|
|
func (a adminAPIHandlers) ImportBucketMetadataHandler(w http.ResponseWriter, r *http.Request) {
|
|
ctx := newContext(r, w, "ImportBucketMetadata")
|
|
|
|
defer logger.AuditLog(ctx, w, r, mustGetClaimsFromToken(r))
|
|
bucket := pathClean(r.Form.Get("bucket"))
|
|
|
|
if !globalIsErasure {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL)
|
|
return
|
|
}
|
|
// Get current object layer instance.
|
|
objectAPI, _ := validateAdminReq(ctx, w, r, iampolicy.ImportBucketMetadataAction)
|
|
if objectAPI == nil {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
|
|
return
|
|
}
|
|
data, err := ioutil.ReadAll(r.Body)
|
|
if err != nil {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrInvalidRequest), r.URL)
|
|
return
|
|
}
|
|
reader := bytes.NewReader(data)
|
|
zr, err := zip.NewReader(reader, int64(len(data)))
|
|
if err != nil {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrInvalidRequest), r.URL)
|
|
return
|
|
}
|
|
bucketMap := make(map[string]struct{}, 1)
|
|
|
|
// import object lock config if any - order of import matters here.
|
|
for _, file := range zr.File {
|
|
slc := strings.Split(file.Name, slashSeparator)
|
|
if len(slc) != 2 { // expecting bucket/configfile in the zipfile
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrInvalidRequest), r.URL)
|
|
return
|
|
}
|
|
b, fileName := slc[0], slc[1]
|
|
if bucket == "" { // use bucket requested in query parameters if specified. Otherwise default bucket name to directory name within zip
|
|
bucket = b
|
|
}
|
|
switch fileName {
|
|
case objectLockConfig:
|
|
reader, err := file.Open()
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
config, err := objectlock.ParseObjectLockConfig(reader)
|
|
if err != nil {
|
|
apiErr := errorCodes.ToAPIErr(ErrMalformedXML)
|
|
apiErr.Description = err.Error()
|
|
writeErrorResponse(ctx, w, apiErr, r.URL)
|
|
return
|
|
}
|
|
|
|
configData, err := xml.Marshal(config)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
if _, ok := bucketMap[bucket]; !ok {
|
|
opts := BucketOptions{
|
|
LockEnabled: config.ObjectLockEnabled == "Enabled",
|
|
}
|
|
err = objectAPI.MakeBucketWithLocation(ctx, bucket, opts)
|
|
if err != nil {
|
|
if _, ok := err.(BucketExists); !ok {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
}
|
|
bucketMap[bucket] = struct{}{}
|
|
}
|
|
|
|
// Deny object locking configuration settings on existing buckets without object lock enabled.
|
|
if _, _, err = globalBucketMetadataSys.GetObjectLockConfig(bucket); err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
|
|
if err = globalBucketMetadataSys.Update(ctx, bucket, objectLockConfig, configData); err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
|
|
// Call site replication hook.
|
|
//
|
|
// We encode the xml bytes as base64 to ensure there are no encoding
|
|
// errors.
|
|
cfgStr := base64.StdEncoding.EncodeToString(configData)
|
|
if err = globalSiteReplicationSys.BucketMetaHook(ctx, madmin.SRBucketMeta{
|
|
Type: madmin.SRBucketMetaTypeObjectLockConfig,
|
|
Bucket: bucket,
|
|
ObjectLockConfig: &cfgStr,
|
|
}); err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// import versioning metadata
|
|
for _, file := range zr.File {
|
|
slc := strings.Split(file.Name, slashSeparator)
|
|
if len(slc) != 2 { // expecting bucket/configfile in the zipfile
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrInvalidRequest), r.URL)
|
|
return
|
|
}
|
|
b, fileName := slc[0], slc[1]
|
|
if bucket == "" { // use bucket requested in query parameters if specified. Otherwise default bucket name to directory name within zip
|
|
bucket = b
|
|
}
|
|
switch fileName {
|
|
case bucketVersioningConfig:
|
|
reader, err := file.Open()
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
v, err := versioning.ParseConfig(io.LimitReader(reader, maxBucketVersioningConfigSize))
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
if _, ok := bucketMap[bucket]; !ok {
|
|
err = objectAPI.MakeBucketWithLocation(ctx, bucket, BucketOptions{})
|
|
if err != nil {
|
|
if _, ok := err.(BucketExists); !ok {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
}
|
|
bucketMap[bucket] = struct{}{}
|
|
}
|
|
|
|
if globalSiteReplicationSys.isEnabled() && v.Suspended() {
|
|
writeErrorResponse(ctx, w, APIError{
|
|
Code: "InvalidBucketState",
|
|
Description: "Cluster replication is enabled for this site, so the versioning state cannot be changed.",
|
|
HTTPStatusCode: http.StatusConflict,
|
|
}, r.URL)
|
|
return
|
|
}
|
|
|
|
if rcfg, _ := globalBucketObjectLockSys.Get(bucket); rcfg.LockEnabled && v.Suspended() {
|
|
writeErrorResponse(ctx, w, APIError{
|
|
Code: "InvalidBucketState",
|
|
Description: "An Object Lock configuration is present on this bucket, so the versioning state cannot be changed.",
|
|
HTTPStatusCode: http.StatusConflict,
|
|
}, r.URL)
|
|
return
|
|
}
|
|
if _, err := getReplicationConfig(ctx, bucket); err == nil && v.Suspended() {
|
|
writeErrorResponse(ctx, w, APIError{
|
|
Code: "InvalidBucketState",
|
|
Description: "A replication configuration is present on this bucket, so the versioning state cannot be changed.",
|
|
HTTPStatusCode: http.StatusConflict,
|
|
}, r.URL)
|
|
return
|
|
}
|
|
|
|
configData, err := xml.Marshal(v)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
|
|
if err = globalBucketMetadataSys.Update(ctx, bucket, bucketVersioningConfig, configData); err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
for _, file := range zr.File {
|
|
reader, err := file.Open()
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
sz := file.FileInfo().Size()
|
|
slc := strings.Split(file.Name, slashSeparator)
|
|
if len(slc) != 2 { // expecting bucket/configfile in the zipfile
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrInvalidRequest), r.URL)
|
|
return
|
|
}
|
|
b, fileName := slc[0], slc[1]
|
|
if bucket == "" { // use bucket requested in query parameters if specified. Otherwise default bucket name to directory name within zip
|
|
bucket = b
|
|
}
|
|
// create bucket if it does not exist yet.
|
|
if _, ok := bucketMap[bucket]; !ok {
|
|
err = objectAPI.MakeBucketWithLocation(ctx, bucket, BucketOptions{})
|
|
if err != nil {
|
|
if _, ok := err.(BucketExists); !ok {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
}
|
|
bucketMap[bucket] = struct{}{}
|
|
}
|
|
switch fileName {
|
|
case bucketNotificationConfig:
|
|
config, err := event.ParseConfig(io.LimitReader(reader, sz), globalSite.Region, globalNotificationSys.targetList)
|
|
if err != nil {
|
|
apiErr := errorCodes.ToAPIErr(ErrMalformedXML)
|
|
if event.IsEventError(err) {
|
|
apiErr = importError(ctx, err, file.Name, bucket)
|
|
}
|
|
writeErrorResponse(ctx, w, apiErr, r.URL)
|
|
return
|
|
}
|
|
|
|
configData, err := xml.Marshal(config)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
|
|
if err = globalBucketMetadataSys.Update(ctx, bucket, bucketNotificationConfig, configData); err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
rulesMap := config.ToRulesMap()
|
|
globalNotificationSys.AddRulesMap(bucket, rulesMap)
|
|
case bucketPolicyConfig:
|
|
// Error out if Content-Length is beyond allowed size.
|
|
if sz > maxBucketPolicySize {
|
|
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrPolicyTooLarge), r.URL)
|
|
return
|
|
}
|
|
|
|
bucketPolicyBytes, err := ioutil.ReadAll(io.LimitReader(reader, sz))
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
|
|
bucketPolicy, err := policy.ParseConfig(bytes.NewReader(bucketPolicyBytes), bucket)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
|
|
// Version in policy must not be empty
|
|
if bucketPolicy.Version == "" {
|
|
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMalformedPolicy), r.URL)
|
|
return
|
|
}
|
|
|
|
configData, err := json.Marshal(bucketPolicy)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
|
|
if err = globalBucketMetadataSys.Update(ctx, bucket, bucketPolicyConfig, configData); err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
// Call site replication hook.
|
|
if err = globalSiteReplicationSys.BucketMetaHook(ctx, madmin.SRBucketMeta{
|
|
Type: madmin.SRBucketMetaTypePolicy,
|
|
Bucket: bucket,
|
|
Policy: bucketPolicyBytes,
|
|
}); err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
|
|
case bucketLifecycleConfig:
|
|
bucketLifecycle, err := lifecycle.ParseLifecycleConfig(io.LimitReader(reader, sz))
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
|
|
// Validate the received bucket policy document
|
|
if err = bucketLifecycle.Validate(); err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
|
|
// Validate the transition storage ARNs
|
|
if err = validateTransitionTier(bucketLifecycle); err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
|
|
configData, err := xml.Marshal(bucketLifecycle)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
|
|
if err = globalBucketMetadataSys.Update(ctx, bucket, bucketLifecycleConfig, configData); err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
case bucketSSEConfig:
|
|
// Parse bucket encryption xml
|
|
encConfig, err := validateBucketSSEConfig(io.LimitReader(reader, maxBucketSSEConfigSize))
|
|
if err != nil {
|
|
apiErr := APIError{
|
|
Code: "MalformedXML",
|
|
Description: fmt.Sprintf("%s (%s)", errorCodes[ErrMalformedXML].Description, err),
|
|
HTTPStatusCode: errorCodes[ErrMalformedXML].HTTPStatusCode,
|
|
}
|
|
writeErrorResponse(ctx, w, apiErr, r.URL)
|
|
return
|
|
}
|
|
|
|
// Return error if KMS is not initialized
|
|
if GlobalKMS == nil {
|
|
writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrKMSNotConfigured), r.URL)
|
|
return
|
|
}
|
|
kmsKey := encConfig.KeyID()
|
|
if kmsKey != "" {
|
|
kmsContext := kms.Context{"MinIO admin API": "ServerInfoHandler"} // Context for a test key operation
|
|
_, err := GlobalKMS.GenerateKey(kmsKey, kmsContext)
|
|
if err != nil {
|
|
if errors.Is(err, kes.ErrKeyNotFound) {
|
|
writeErrorResponse(ctx, w, importError(ctx, errKMSKeyNotFound, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
}
|
|
|
|
configData, err := xml.Marshal(encConfig)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
|
|
// Store the bucket encryption configuration in the object layer
|
|
if err = globalBucketMetadataSys.Update(ctx, bucket, bucketSSEConfig, configData); err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
|
|
// Call site replication hook.
|
|
//
|
|
// We encode the xml bytes as base64 to ensure there are no encoding
|
|
// errors.
|
|
cfgStr := base64.StdEncoding.EncodeToString(configData)
|
|
if err = globalSiteReplicationSys.BucketMetaHook(ctx, madmin.SRBucketMeta{
|
|
Type: madmin.SRBucketMetaTypeSSEConfig,
|
|
Bucket: bucket,
|
|
SSEConfig: &cfgStr,
|
|
}); err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
|
|
case bucketTaggingConfig:
|
|
tags, err := tags.ParseBucketXML(io.LimitReader(reader, sz))
|
|
if err != nil {
|
|
apiErr := errorCodes.ToAPIErrWithErr(ErrMalformedXML, fmt.Errorf("error importing %s with %w", file.Name, err))
|
|
writeErrorResponse(ctx, w, apiErr, r.URL)
|
|
return
|
|
}
|
|
|
|
configData, err := xml.Marshal(tags)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
|
|
if err = globalBucketMetadataSys.Update(ctx, bucket, bucketTaggingConfig, configData); err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
// Call site replication hook.
|
|
//
|
|
// We encode the xml bytes as base64 to ensure there are no encoding
|
|
// errors.
|
|
cfgStr := base64.StdEncoding.EncodeToString(configData)
|
|
if err = globalSiteReplicationSys.BucketMetaHook(ctx, madmin.SRBucketMeta{
|
|
Type: madmin.SRBucketMetaTypeTags,
|
|
Bucket: bucket,
|
|
Tags: &cfgStr,
|
|
}); err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
case bucketQuotaConfigFile:
|
|
data, err := ioutil.ReadAll(reader)
|
|
if err != nil {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrInvalidRequest), r.URL)
|
|
return
|
|
}
|
|
|
|
quotaConfig, err := parseBucketQuota(bucket, data)
|
|
if err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
|
|
if quotaConfig.Type == "fifo" {
|
|
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrInvalidRequest), r.URL)
|
|
return
|
|
}
|
|
|
|
if err = globalBucketMetadataSys.Update(ctx, bucket, bucketQuotaConfigFile, data); err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
|
|
bucketMeta := madmin.SRBucketMeta{
|
|
Type: madmin.SRBucketMetaTypeQuotaConfig,
|
|
Bucket: bucket,
|
|
Quota: data,
|
|
}
|
|
if quotaConfig.Quota == 0 {
|
|
bucketMeta.Quota = nil
|
|
}
|
|
|
|
// Call site replication hook.
|
|
if err = globalSiteReplicationSys.BucketMetaHook(ctx, bucketMeta); err != nil {
|
|
writeErrorResponse(ctx, w, importError(ctx, err, file.Name, bucket), r.URL)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|