S3: Auto create bucket (#7549)
* auto create buckets * only admin users can auto create buckets * Update weed/s3api/s3api_bucket_handlers.go Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com> * validate bucket name * refactor * error handling * error * refetch * ensure owner * multiple errors --------- Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
This commit is contained in:
@@ -244,18 +244,8 @@ func (s3a *S3ApiServer) PutBucketHandler(w http.ResponseWriter, r *http.Request)
|
||||
return
|
||||
}
|
||||
|
||||
fn := func(entry *filer_pb.Entry) {
|
||||
// Reuse currentIdentityId from above (already retrieved from context)
|
||||
if currentIdentityId != "" {
|
||||
if entry.Extended == nil {
|
||||
entry.Extended = make(map[string][]byte)
|
||||
}
|
||||
entry.Extended[s3_constants.AmzIdentityId] = []byte(currentIdentityId)
|
||||
}
|
||||
}
|
||||
|
||||
// create the folder for bucket, but lazily create actual collection
|
||||
if err := s3a.mkdir(s3a.option.BucketsPath, bucket, fn); err != nil {
|
||||
if err := s3a.mkdir(s3a.option.BucketsPath, bucket, setBucketOwner(r)); err != nil {
|
||||
glog.Errorf("PutBucketHandler mkdir: %v", err)
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
return
|
||||
@@ -565,6 +555,93 @@ func (s3a *S3ApiServer) checkBucket(r *http.Request, bucket string) s3err.ErrorC
|
||||
return s3err.ErrNone
|
||||
}
|
||||
|
||||
// ErrAutoCreatePermissionDenied is returned when a user lacks permission to auto-create buckets
|
||||
var ErrAutoCreatePermissionDenied = errors.New("permission denied - requires Admin permission")
|
||||
|
||||
// ErrInvalidBucketName is returned when a bucket name doesn't meet S3 naming requirements
|
||||
var ErrInvalidBucketName = errors.New("invalid bucket name")
|
||||
|
||||
// setBucketOwner creates a function that sets the bucket owner from the request context
|
||||
func setBucketOwner(r *http.Request) func(entry *filer_pb.Entry) {
|
||||
currentIdentityId := s3_constants.GetIdentityNameFromContext(r)
|
||||
return func(entry *filer_pb.Entry) {
|
||||
if currentIdentityId != "" {
|
||||
if entry.Extended == nil {
|
||||
entry.Extended = make(map[string][]byte)
|
||||
}
|
||||
entry.Extended[s3_constants.AmzIdentityId] = []byte(currentIdentityId)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// autoCreateBucket creates a bucket if it doesn't exist, setting the owner from the request context
|
||||
// Only users with admin permissions are allowed to auto-create buckets
|
||||
func (s3a *S3ApiServer) autoCreateBucket(r *http.Request, bucket string) error {
|
||||
// Validate the bucket name before auto-creating
|
||||
if err := s3bucket.VerifyS3BucketName(bucket); err != nil {
|
||||
return fmt.Errorf("auto-create bucket %s: %w", bucket, errors.Join(ErrInvalidBucketName, err))
|
||||
}
|
||||
|
||||
// Check if user has admin permissions
|
||||
if !s3a.isUserAdmin(r) {
|
||||
return fmt.Errorf("auto-create bucket %s: %w", bucket, ErrAutoCreatePermissionDenied)
|
||||
}
|
||||
|
||||
if err := s3a.mkdir(s3a.option.BucketsPath, bucket, setBucketOwner(r)); err != nil {
|
||||
// In case of a race condition where another request created the bucket
|
||||
// in the meantime, check for existence before returning an error.
|
||||
if exist, err2 := s3a.exists(s3a.option.BucketsPath, bucket, true); err2 != nil {
|
||||
glog.Warningf("autoCreateBucket: failed to check existence for bucket %s: %v", bucket, err2)
|
||||
return fmt.Errorf("failed to auto-create bucket %s: %w", bucket, errors.Join(err, err2))
|
||||
} else if exist {
|
||||
// The bucket exists, which is fine. However, we should ensure it has an owner.
|
||||
// If it was created by a concurrent request that didn't set an owner,
|
||||
// we'll set it here to ensure consistency.
|
||||
if entry, getErr := s3a.getEntry(s3a.option.BucketsPath, bucket); getErr == nil {
|
||||
if entry.Extended == nil || len(entry.Extended[s3_constants.AmzIdentityId]) == 0 {
|
||||
// No owner set, assign current admin as owner
|
||||
setBucketOwner(r)(entry)
|
||||
if updateErr := s3a.updateEntry(s3a.option.BucketsPath, entry); updateErr != nil {
|
||||
glog.Warningf("autoCreateBucket: failed to set owner for existing bucket %s: %v", bucket, updateErr)
|
||||
} else {
|
||||
glog.V(1).Infof("Set owner for existing bucket %s (created by concurrent request)", bucket)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
glog.Warningf("autoCreateBucket: failed to get entry for existing bucket %s: %v", bucket, getErr)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("failed to auto-create bucket %s: %w", bucket, err)
|
||||
}
|
||||
|
||||
// Remove bucket from negative cache after successful creation
|
||||
if s3a.bucketConfigCache != nil {
|
||||
s3a.bucketConfigCache.RemoveNegativeCache(bucket)
|
||||
}
|
||||
|
||||
glog.V(1).Infof("Auto-created bucket %s", bucket)
|
||||
return nil
|
||||
}
|
||||
|
||||
// handleAutoCreateBucket attempts to auto-create a bucket and writes appropriate error responses
|
||||
// Returns true if the bucket was created successfully or already exists, false if an error was written
|
||||
func (s3a *S3ApiServer) handleAutoCreateBucket(w http.ResponseWriter, r *http.Request, bucket, handlerName string) bool {
|
||||
if err := s3a.autoCreateBucket(r, bucket); err != nil {
|
||||
glog.Warningf("%s: %v", handlerName, err)
|
||||
// Check for specific errors to return appropriate S3 error codes
|
||||
if errors.Is(err, ErrInvalidBucketName) {
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInvalidBucketName)
|
||||
} else if errors.Is(err, ErrAutoCreatePermissionDenied) {
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied)
|
||||
} else {
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
}
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func (s3a *S3ApiServer) hasAccess(r *http.Request, entry *filer_pb.Entry) bool {
|
||||
// Check if user is properly authenticated as admin through IAM system
|
||||
if s3a.isUserAdmin(r) {
|
||||
|
||||
@@ -33,8 +33,14 @@ const (
|
||||
func (s3a *S3ApiServer) NewMultipartUploadHandler(w http.ResponseWriter, r *http.Request) {
|
||||
bucket, object := s3_constants.GetBucketAndObject(r)
|
||||
|
||||
// Check if bucket exists before creating multipart upload
|
||||
if err := s3a.checkBucket(r, bucket); err != s3err.ErrNone {
|
||||
// Check if bucket exists, and create it if it doesn't (auto-create bucket)
|
||||
if err := s3a.checkBucket(r, bucket); err == s3err.ErrNoSuchBucket {
|
||||
// Auto-create bucket if it doesn't exist (requires Admin permission)
|
||||
if !s3a.handleAutoCreateBucket(w, r, bucket, "NewMultipartUploadHandler") {
|
||||
return
|
||||
}
|
||||
} else if err != s3err.ErrNone {
|
||||
// Other errors (like access denied) should still fail
|
||||
s3err.WriteErrorResponse(w, r, err)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -135,12 +135,23 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
|
||||
versioningState, err := s3a.getVersioningState(bucket)
|
||||
if err != nil {
|
||||
if errors.Is(err, filer_pb.ErrNotFound) {
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchBucket)
|
||||
// Auto-create bucket if it doesn't exist (requires Admin permission)
|
||||
if !s3a.handleAutoCreateBucket(w, r, bucket, "PutObjectHandler") {
|
||||
return
|
||||
}
|
||||
// Re-fetch versioning state to handle race conditions where
|
||||
// another process might have created the bucket with versioning enabled.
|
||||
versioningState, err = s3a.getVersioningState(bucket)
|
||||
if err != nil {
|
||||
glog.Errorf("Error re-checking versioning status for bucket %s after auto-creation: %v", bucket, err)
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
return
|
||||
}
|
||||
} else {
|
||||
glog.Errorf("Error checking versioning status for bucket %s: %v", bucket, err)
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
return
|
||||
}
|
||||
glog.Errorf("Error checking versioning status for bucket %s: %v", bucket, err)
|
||||
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
|
||||
return
|
||||
}
|
||||
|
||||
versioningEnabled := (versioningState == s3_constants.VersioningEnabled)
|
||||
|
||||
Reference in New Issue
Block a user