mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-15 00:09:00 +01:00
s3api: refuse writes to a stored version's physical path (#12622)
* s3api: refuse a PUT to a stored version's physical path In a Suspended bucket a PUT to <key>.versions/v_<id> replaced the stored version's body in place. Look the entry up when the key has the version file shape and deny the write when it is a version file. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * s3api: refuse copy, multipart, POST and rename writes to a version file Apply the same version file check to the CopyObject, CreateMultipartUpload, CompleteMultipartUpload, POST Object and RenameObject destinations. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * s3api: read a version file write target on its logical owner The guard resolved the key as an object, which on a ring reads the physical path's owner and, in a versioned bucket, may return a nested version under <key>.versions/v_<id>.versions instead of the stored version. Look up the physical entry directly, routed by the logical key. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * s3api: confirm a version file miss on a skipped owner When the owner was recently unreachable the routed lookup answers from a replica, whose miss does not show the version is absent. Ask the owner directly in that case and refuse the write when it cannot answer. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * s3api: confirm every routed version file miss on the owner Whether the routed lookup skipped the owner can change between the lookup and a second unreachable check, letting a replica's miss through. Ask the owner after any miss instead. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
d1a97f4c3d
commit
d685709832
8 files changed
+139
No files matched your search
@@ -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"))
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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, "/") {
|
||||
|
||||
@@ -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.
|
||||
|
||||
Reference in new issue
Block a user