diff --git a/weed/s3api/s3api_object_anonymous_read_test.go b/weed/s3api/s3api_object_anonymous_read_test.go index c40f772a94..daead9114c 100644 --- a/weed/s3api/s3api_object_anonymous_read_test.go +++ b/weed/s3api/s3api_object_anonymous_read_test.go @@ -4,6 +4,7 @@ import ( "context" "crypto/md5" "encoding/json" + "fmt" "net/http" "net/http/httptest" "path" @@ -15,6 +16,8 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/s3" "github.com/gorilla/mux" + "github.com/seaweedfs/seaweedfs/weed/cluster" + "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/s3api/policy_engine" "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" @@ -32,11 +35,15 @@ type anonymousReadFiler struct { mu sync.Mutex entries map[string]*filer_pb.Entry lookupErr error + onLookup func() } func (f *anonymousReadFiler) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { f.mu.Lock() defer f.mu.Unlock() + if f.onLookup != nil { + f.onLookup() + } if f.lookupErr != nil { return nil, f.lookupErr } @@ -268,6 +275,80 @@ func TestSignedReadOfVersionFileKey(t *testing.T) { require.Contains(t, f.entries, "/buckets/b/o.versions/v_v1") } +func TestWriteToVersionFileKey(t *testing.T) { + for _, state := range []string{"Suspended", "Enabled"} { + s3a, f := newAnonymousReadTestServer(t) + config, _ := s3a.getBucketConfig("b") + config.Versioning = state + version := anonymousReadEntry(nil) + version.Name = "v_v1" + version.Extended[s3_constants.ExtVersionIdKey] = []byte("v1") + f.entries["/buckets/b/o.versions/v_v1"] = version + f.entries["/buckets/b/u.versions/v_literal"] = anonymousReadEntry(nil) + nullVersion := anonymousReadEntry(nil) + nullVersion.Name = "v_null" + nullVersion.Extended[s3_constants.ExtVersionIdKey] = []byte("null") + f.entries["/buckets/b/o.versions/v_null"] = nullVersion + f.entries["/buckets/b/o.versions/v_null.versions"] = &filer_pb.Entry{Name: "v_null.versions", IsDirectory: true, Extended: map[string][]byte{ + s3_constants.ExtLatestVersionIdKey: []byte("v9"), + s3_constants.ExtLatestVersionFileNameKey: []byte("v_v9"), + }} + nested := anonymousReadEntry(nil) + nested.Name = "v_v9" + nested.Extended[s3_constants.ExtVersionIdKey] = []byte("v9") + f.entries["/buckets/b/o.versions/v_null.versions/v_v9"] = nested + require.Equal(t, s3err.ErrAccessDenied, s3a.checkVersionFileWrite("b", "o.versions/v_v1", "t"), state) + require.Equal(t, s3err.ErrAccessDenied, s3a.checkVersionFileWrite("b", "o.versions/v_null", "t"), state) + require.Equal(t, s3err.ErrAccessDenied, s3a.checkVersionFileWrite("b", "/o.versions/v_v1/", "t"), state) + require.Equal(t, s3err.ErrNone, s3a.checkVersionFileWrite("b", "o.versions/v_v2", "t"), state) + require.Equal(t, s3err.ErrNone, s3a.checkVersionFileWrite("b", "u.versions/v_literal", "t"), state) + require.Equal(t, s3err.ErrNone, s3a.checkVersionFileWrite("b", "o", "t"), state) + } +} + +func TestWriteToVersionFileKeyReadsLogicalOwner(t *testing.T) { + gateway := &anonymousReadFiler{entries: make(map[string]*filer_pb.Entry)} + owner := &anonymousReadFiler{entries: make(map[string]*filer_pb.Entry)} + gatewayAddr, ownerAddr := startFakeFiler(t, gateway), startFakeFiler(t, owner) + s3a := newPutTestServer(t, gatewayAddr) + s3a.bucketConfigCache = NewBucketConfigCache(time.Minute) + s3a.bucketConfigCache.Set("b", &BucketConfig{Name: "b", Versioning: s3_constants.VersioningSuspended}) + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, gatewayAddr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{gatewayAddr, ownerAddr}, 1) + + object := "" + for i := 0; i < 1000 && object == ""; i++ { + candidate := fmt.Sprintf("o%d", i) + if s3a.routableWriteOwner("b", candidate) == ownerAddr && s3a.routableWriteOwner("b", candidate+".versions/v_v1") == gatewayAddr { + object = candidate + } + } + require.NotEmpty(t, object) + version := anonymousReadEntry(nil) + version.Name = "v_v1" + version.Extended[s3_constants.ExtVersionIdKey] = []byte("v1") + owner.entries["/buckets/b/"+object+".versions/v_v1"] = version + require.Equal(t, s3err.ErrAccessDenied, s3a.checkVersionFileWrite("b", object+".versions/v_v1", "t")) + + s3a.markOwnerUnreachable(ownerAddr) + require.Equal(t, s3err.ErrAccessDenied, s3a.checkVersionFileWrite("b", object+".versions/v_v1", "t")) + gateway.mu.Lock() + gateway.onLookup = func() { s3a.unreachableOwners.Delete(ownerAddr) } + gateway.mu.Unlock() + s3a.markOwnerUnreachable(ownerAddr) + require.Equal(t, s3err.ErrAccessDenied, s3a.checkVersionFileWrite("b", object+".versions/v_v1", "t")) + gateway.mu.Lock() + gateway.onLookup = nil + gateway.mu.Unlock() + s3a.markOwnerUnreachable(ownerAddr) + require.Equal(t, s3err.ErrNone, s3a.checkVersionFileWrite("b", object+".versions/v_v2", "t")) + owner.mu.Lock() + owner.lookupErr = status.Error(codes.Unavailable, "owner down") + owner.mu.Unlock() + s3a.markOwnerUnreachable(ownerAddr) + require.Equal(t, s3err.ErrInternalError, s3a.checkVersionFileWrite("b", object+".versions/v_v2", "t")) +} + func TestAnonymousObjectACLDoesNotReplaceAuthentication(t *testing.T) { s3a, f := newAnonymousReadTestServer(t) f.entries["/buckets/b/o"] = anonymousReadEntry(anonymousReadGrant("READ")) diff --git a/weed/s3api/s3api_object_handlers_copy.go b/weed/s3api/s3api_object_handlers_copy.go index 7b94772c4a..5540fa8de4 100644 --- a/weed/s3api/s3api_object_handlers_copy.go +++ b/weed/s3api/s3api_object_handlers_copy.go @@ -156,6 +156,10 @@ func (s3a *S3ApiServer) CopyObjectHandler(w http.ResponseWriter, r *http.Request return } } + if errCode := s3a.checkVersionFileWrite(dstBucket, dstObject, "CopyObjectHandler"); errCode != s3err.ErrNone { + s3err.WriteErrorResponse(w, r, errCode) + return + } // Validate copy source and destination if err := validateCopySource(cpSrcPath, srcBucket, srcObject, srcVersionId); err != nil { diff --git a/weed/s3api/s3api_object_handlers_extract.go b/weed/s3api/s3api_object_handlers_extract.go index d57884d361..04d1fe0c00 100644 --- a/weed/s3api/s3api_object_handlers_extract.go +++ b/weed/s3api/s3api_object_handlers_extract.go @@ -178,6 +178,9 @@ func (s3a *S3ApiServer) extractZipObject(w http.ResponseWriter, r *http.Request, if err := s3a.validateTableBucketObjectPath(bucket, memberKey); err != nil { return s3err.ErrAccessDenied } + if code := s3a.checkVersionFileWrite(bucket, memberKey, "extractZipObject"); code != s3err.ErrNone { + return code + } if code, _ := s3a.checkPolicyWithEntry(r, bucket, memberKey, string(s3_constants.ACTION_WRITE), "", nil); code != s3err.ErrNone { return code } diff --git a/weed/s3api/s3api_object_handlers_multipart.go b/weed/s3api/s3api_object_handlers_multipart.go index ace520829e..d2a0455081 100644 --- a/weed/s3api/s3api_object_handlers_multipart.go +++ b/weed/s3api/s3api_object_handlers_multipart.go @@ -61,6 +61,10 @@ func (s3a *S3ApiServer) NewMultipartUploadHandler(w http.ResponseWriter, r *http s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied) return } + if errCode := s3a.checkVersionFileWrite(bucket, object, "NewMultipartUploadHandler"); errCode != s3err.ErrNone { + s3err.WriteErrorResponse(w, r, errCode) + return + } // Check if versioning is enabled for the bucket (needed for object lock) versioningEnabled, err := s3a.isVersioningEnabled(bucket) @@ -161,6 +165,10 @@ func (s3a *S3ApiServer) CompleteMultipartUploadHandler(w http.ResponseWriter, r s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied) return } + if errCode := s3a.checkVersionFileWrite(bucket, object, "CompleteMultipartUploadHandler"); errCode != s3err.ErrNone { + s3err.WriteErrorResponse(w, r, errCode) + return + } parts := &CompleteMultipartUpload{} if err := xmlDecoder(r.Body, parts, r.ContentLength); err != nil { diff --git a/weed/s3api/s3api_object_handlers_postpolicy.go b/weed/s3api/s3api_object_handlers_postpolicy.go index a316a9b7da..a739ad95be 100644 --- a/weed/s3api/s3api_object_handlers_postpolicy.go +++ b/weed/s3api/s3api_object_handlers_postpolicy.go @@ -70,6 +70,10 @@ func (s3a *S3ApiServer) PostPolicyBucketHandler(w http.ResponseWriter, r *http.R s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied) return } + if errCode := s3a.checkVersionFileWrite(bucket, object, "PostPolicyBucketHandler"); errCode != s3err.ErrNone { + s3err.WriteErrorResponse(w, r, errCode) + return + } successRedirect := formValues.Get("success_action_redirect") successStatus := formValues.Get("success_action_status") diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index e2954e468a..425a26101b 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -123,6 +123,11 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request) return } + if errCode := s3a.checkVersionFileWrite(bucket, object, "PutObjectHandler"); errCode != s3err.ErrNone { + s3err.WriteErrorResponse(w, r, errCode) + return + } + // Check bucket policy if errCode, _ := s3a.checkPolicyWithEntry(r, bucket, object, string(s3_constants.ACTION_WRITE), "", nil); errCode != s3err.ErrNone { s3err.WriteErrorResponse(w, r, errCode) diff --git a/weed/s3api/s3api_object_handlers_rename.go b/weed/s3api/s3api_object_handlers_rename.go index 4ad6286691..de2447acb8 100644 --- a/weed/s3api/s3api_object_handlers_rename.go +++ b/weed/s3api/s3api_object_handlers_rename.go @@ -58,6 +58,10 @@ func (s3a *S3ApiServer) RenameObjectHandler(w http.ResponseWriter, r *http.Reque s3err.WriteErrorResponse(w, r, s3err.ErrAccessDenied) return } + if errCode := s3a.checkVersionFileWrite(bucket, dstObject, "RenameObjectHandler"); errCode != s3err.ErrNone { + s3err.WriteErrorResponse(w, r, errCode) + return + } // A trailing slash names a directory, and renaming one would move a whole // subtree rather than an object. if strings.HasSuffix(dstObject, "/") { diff --git a/weed/s3api/s3api_server.go b/weed/s3api/s3api_server.go index 8d120ed4d2..2cea8868c8 100644 --- a/weed/s3api/s3api_server.go +++ b/weed/s3api/s3api_server.go @@ -756,6 +756,36 @@ func (s3a *S3ApiServer) recheckTagConditionedPolicyWithLookup(r *http.Request, b return s3a.recheckPolicyWithObjectEntry(r, bucket, object, action, entry.Extended, handlerName) } +// checkVersionFileWrite refuses a write whose key is a stored version's +// physical path, which would replace that version's content in place. The +// physical entry is read on the logical object's owner, where versions commit. +func (s3a *S3ApiServer) checkVersionFileWrite(bucket, object, handlerName string) s3err.ErrorCode { + if !s3a.hasVersionFileShape(object) { + return s3err.ErrNone + } + key := strings.TrimSuffix(NormalizeObjectKey(object), "/") + dir, name := util.NewFullPath(s3a.bucketDir(bucket), key).DirAndName() + logicalObject := strings.TrimSuffix(strings.TrimSuffix(key, "/"+name), VersionsFolder) + entry, err := s3a.getObjectScopedEntryRoutedByKey(bucket, logicalObject, dir, name) + if errors.Is(err, filer_pb.ErrNotFound) || status.Code(err) == codes.NotFound { + // The routed lookup may have answered from a replica, whose miss proves nothing. + if owner := s3a.routableWriteOwner(bucket, logicalObject); owner != "" { + entry, err = s3a.lookupEntryOnFiler(context.Background(), owner, dir, name) + } + } + if err != nil { + if errors.Is(err, filer_pb.ErrNotFound) || status.Code(err) == codes.NotFound { + return s3err.ErrNone + } + glog.Errorf("%s: version file lookup for %s/%s failed: %v", handlerName, bucket, object, err) + return s3err.ErrInternalError + } + if entry != nil && s3a.isVersionFileKey(object, entry.Extended) { + return s3err.ErrAccessDenied + } + return s3err.ErrNone +} + // recheckPolicyForKeyWithLookup re-decides a first-phase deny on one key for a // synthesized method, resolving the entry the way the read path does (versions // included) so s3:ExistingObjectTag conditions see the object's real tags.