mirror of
https://github.com/OpenListTeam/OpenList.git
synced 2026-10-10 04:53:09 +08:00
fix(s3): enforce conditional writes with valid object ETags
- Pin gofakes3 with conditional PUT, multipart completion, and copy support - Derive object ETags from content and preserve matching multipart validators - Keep conditional requests in the S3 handler instead of redirecting them - Format modification times in UTC and propagate storage errors - Close content readers and discard metadata after successful deletion - Add regression tests for object hashes, metadata, and conditional redirects Co-authored-by: Codex <267193182+codex@users.noreply.github.com> Signed-off-by: jyxjjj <16695261+jyxjjj@users.noreply.github.com>
This commit is contained in:
@@ -10,7 +10,7 @@ require (
|
||||
github.com/KarpelesLab/reflink v1.0.2
|
||||
github.com/KirCute/zip v1.0.1
|
||||
github.com/OpenListTeam/go-cache v0.1.0
|
||||
github.com/OpenListTeam/gofakes3 v0.8.1
|
||||
github.com/OpenListTeam/gofakes3 v0.8.2-0.20260910152701-89a46c65a7cf
|
||||
github.com/OpenListTeam/sftpd-openlist v1.0.1
|
||||
github.com/OpenListTeam/tache v0.2.2
|
||||
github.com/OpenListTeam/times v0.1.0
|
||||
|
||||
@@ -51,8 +51,8 @@ github.com/OpenListTeam/115-sdk-go v0.2.6 h1:ehXyStvncvn4qRBuknor3kyGZtUmHc0+stj
|
||||
github.com/OpenListTeam/115-sdk-go v0.2.6/go.mod h1:cfvitk2lwe6036iNi2h+iNxwxWDifKZsSvNtrur5BqU=
|
||||
github.com/OpenListTeam/go-cache v0.1.0 h1:eV2+FCP+rt+E4OCJqLUW7wGccWZNJMV0NNkh+uChbAI=
|
||||
github.com/OpenListTeam/go-cache v0.1.0/go.mod h1:AHWjKhNK3LE4rorVdKyEALDHoeMnP8SjiNyfVlB+Pz4=
|
||||
github.com/OpenListTeam/gofakes3 v0.8.1 h1:uihJ7Zgb4qIafFcXhcm71BzxCyGRIqBVJYg4YOUa6uY=
|
||||
github.com/OpenListTeam/gofakes3 v0.8.1/go.mod h1:mS9Ywbo6aId6BrRzeYjOIOpK0QDVnMoKOIb0hpaQZ3U=
|
||||
github.com/OpenListTeam/gofakes3 v0.8.2-0.20260910152701-89a46c65a7cf h1:p3Wby2o00mEAZPGLZx+X6QGKoInCDBNjKpPWn0Lpprc=
|
||||
github.com/OpenListTeam/gofakes3 v0.8.2-0.20260910152701-89a46c65a7cf/go.mod h1:mS9Ywbo6aId6BrRzeYjOIOpK0QDVnMoKOIb0hpaQZ3U=
|
||||
github.com/OpenListTeam/gsync v0.1.0 h1:ywzGybOvA3lW8K1BUjKZ2IUlT2FSlzPO4DOazfYXjcs=
|
||||
github.com/OpenListTeam/gsync v0.1.0/go.mod h1:h/Rvv9aX/6CdW/7B8di3xK3xNV8dUg45Fehrd/ksZ9s=
|
||||
github.com/OpenListTeam/reflink v0.0.0-20260701021214-78760eaeafef h1:67uGHancMF/abMrnkc8abVUWQiG73Wk5d8CKt3RzkFo=
|
||||
|
||||
+59
-24
@@ -4,9 +4,11 @@ package s3
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/md5"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"path"
|
||||
"strings"
|
||||
"sync"
|
||||
@@ -30,7 +32,7 @@ import (
|
||||
|
||||
var (
|
||||
emptyPrefix = &gofakes3.Prefix{}
|
||||
timeFormat = "Mon, 2 Jan 2006 15:04:05 GMT"
|
||||
timeFormat = http.TimeFormat
|
||||
)
|
||||
|
||||
// s3Backend implements the gofakes3.Backend interface to make an S3
|
||||
@@ -117,9 +119,13 @@ func (b *s3Backend) HeadObject(ctx context.Context, bucketName, objectName strin
|
||||
|
||||
fp := path.Join(bucketPath, objectName)
|
||||
fmeta, _ := op.GetNearestMeta(fp)
|
||||
node, err := fs.Get(context.WithValue(ctx, conf.MetaKey, fmeta), fp, &fs.GetArgs{})
|
||||
ctx = context.WithValue(ctx, conf.MetaKey, fmeta)
|
||||
node, err := fs.Get(ctx, fp, &fs.GetArgs{})
|
||||
if err != nil {
|
||||
return nil, gofakes3.KeyNotFound(objectName)
|
||||
if errs.IsObjectNotFound(err) {
|
||||
return nil, gofakes3.KeyNotFound(objectName)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if node.IsDir() {
|
||||
@@ -127,23 +133,27 @@ func (b *s3Backend) HeadObject(ctx context.Context, bucketName, objectName strin
|
||||
}
|
||||
|
||||
size := node.GetSize()
|
||||
// hash := getFileHashByte(fobj)
|
||||
hash, err := getObjectHash(ctx, fp, node)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
meta := map[string]string{
|
||||
"Last-Modified": node.ModTime().Format(timeFormat),
|
||||
"Last-Modified": node.ModTime().UTC().Format(timeFormat),
|
||||
"Content-Type": utils.GetMimeType(fp),
|
||||
}
|
||||
|
||||
if val, ok := b.meta.Load(fp); ok {
|
||||
metaMap := val.(map[string]string)
|
||||
for k, v := range metaMap {
|
||||
stored, etag := b.loadMetadata(fp, hash)
|
||||
for k, v := range stored {
|
||||
if k != "Last-Modified" {
|
||||
meta[k] = v
|
||||
}
|
||||
}
|
||||
|
||||
return &gofakes3.Object{
|
||||
Name: objectName,
|
||||
// Hash: hash,
|
||||
Name: objectName,
|
||||
Hash: hash,
|
||||
ETag: etag,
|
||||
Metadata: meta,
|
||||
Size: size,
|
||||
Contents: noOpReadCloser{},
|
||||
@@ -160,15 +170,24 @@ func (b *s3Backend) GetObject(ctx context.Context, bucketName, objectName string
|
||||
|
||||
fp := path.Join(bucketPath, objectName)
|
||||
fmeta, _ := op.GetNearestMeta(fp)
|
||||
node, err := fs.Get(context.WithValue(ctx, conf.MetaKey, fmeta), fp, &fs.GetArgs{})
|
||||
ctx = context.WithValue(ctx, conf.MetaKey, fmeta)
|
||||
node, err := fs.Get(ctx, fp, &fs.GetArgs{})
|
||||
if err != nil {
|
||||
return nil, gofakes3.KeyNotFound(objectName)
|
||||
if errs.IsObjectNotFound(err) {
|
||||
return nil, gofakes3.KeyNotFound(objectName)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if node.IsDir() {
|
||||
return nil, gofakes3.KeyNotFound(objectName)
|
||||
}
|
||||
|
||||
hash, err := getObjectHash(ctx, fp, node)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
link, file, err := fs.Link(ctx, fp, model.LinkArgs{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -193,7 +212,7 @@ func (b *s3Backend) GetObject(ctx context.Context, bucketName, objectName string
|
||||
return nil, fmt.Errorf("the remote storage driver need to be enhanced to support s3")
|
||||
}
|
||||
|
||||
var rd io.Reader
|
||||
var rd io.ReadCloser
|
||||
if rnge != nil {
|
||||
rd, err = rrf.RangeRead(ctx, http_range.Range(*rnge))
|
||||
} else {
|
||||
@@ -204,26 +223,34 @@ func (b *s3Backend) GetObject(ctx context.Context, bucketName, objectName string
|
||||
}
|
||||
|
||||
meta := map[string]string{
|
||||
"Last-Modified": node.ModTime().Format(timeFormat),
|
||||
"Last-Modified": node.ModTime().UTC().Format(timeFormat),
|
||||
"Content-Disposition": utils.GenerateContentDisposition(file.GetName()),
|
||||
"Content-Type": utils.GetMimeType(fp),
|
||||
}
|
||||
|
||||
if val, ok := b.meta.Load(fp); ok {
|
||||
metaMap := val.(map[string]string)
|
||||
for k, v := range metaMap {
|
||||
stored, etag := b.loadMetadata(fp, hash)
|
||||
for k, v := range stored {
|
||||
if k != "Last-Modified" {
|
||||
meta[k] = v
|
||||
}
|
||||
}
|
||||
|
||||
return &gofakes3.Object{
|
||||
// Name: gofakes3.URLEncode(objectName),
|
||||
Name: objectName,
|
||||
// Hash: "",
|
||||
Name: objectName,
|
||||
Hash: hash,
|
||||
ETag: etag,
|
||||
Metadata: meta,
|
||||
Size: size,
|
||||
Range: rnge,
|
||||
Contents: utils.ReadCloser{Reader: rd, Closer: link},
|
||||
Contents: utils.NewReadCloser(rd, func() error {
|
||||
readErr := rd.Close()
|
||||
linkErr := link.Close()
|
||||
if readErr != nil {
|
||||
return readErr
|
||||
}
|
||||
return linkErr
|
||||
}),
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -239,7 +266,7 @@ func (b *s3Backend) PutObject(
|
||||
meta map[string]string,
|
||||
input io.Reader, size int64,
|
||||
) (result gofakes3.PutObjectResult, err error) {
|
||||
return result, b.putStream(ctx, bucketName, objectName, meta, input, size)
|
||||
return result, b.putStream(ctx, bucketName, objectName, meta, input, size, "")
|
||||
}
|
||||
|
||||
// putStream stores the given object into the underlying storage. It is shared
|
||||
@@ -249,6 +276,7 @@ func (b *s3Backend) putStream(
|
||||
ctx context.Context, bucketName, objectName string,
|
||||
meta map[string]string,
|
||||
input io.Reader, size int64,
|
||||
etag string,
|
||||
) error {
|
||||
bucket, err := getBucketByName(bucketName)
|
||||
if err != nil {
|
||||
@@ -314,9 +342,10 @@ func (b *s3Backend) putStream(
|
||||
if setting.GetBool(conf.IgnoreSystemFiles) && utils.IsSystemFile(obj.Name) {
|
||||
return errs.IgnoredSystemFile
|
||||
}
|
||||
hash := md5.New()
|
||||
stream := &stream.FileStream{
|
||||
Obj: &obj,
|
||||
Reader: input,
|
||||
Reader: io.TeeReader(input, hash),
|
||||
Mimetype: meta["Content-Type"],
|
||||
}
|
||||
if stream.Mimetype == "" {
|
||||
@@ -328,7 +357,7 @@ func (b *s3Backend) putStream(
|
||||
return err
|
||||
}
|
||||
|
||||
b.meta.Store(fp, meta)
|
||||
b.meta.Store(fp, objectMetadata{headers: meta, hash: hash.Sum(nil), etag: etag})
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -374,7 +403,10 @@ func (b *s3Backend) deleteObject(ctx context.Context, bucketName, objectName str
|
||||
return err
|
||||
}
|
||||
|
||||
fs.Remove(ctx, fp)
|
||||
if err := fs.Remove(ctx, fp); err != nil {
|
||||
return err
|
||||
}
|
||||
b.meta.Delete(fp)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -418,6 +450,9 @@ func (b *s3Backend) CopyObject(ctx context.Context, srcBucket, srcKey, dstBucket
|
||||
srcFp := path.Join(srcBucketPath, srcKey)
|
||||
fmeta, _ := op.GetNearestMeta(srcFp)
|
||||
srcNode, err := fs.Get(context.WithValue(ctx, conf.MetaKey, fmeta), srcFp, &fs.GetArgs{})
|
||||
if err != nil {
|
||||
return result, err
|
||||
}
|
||||
|
||||
c, err := b.GetObject(ctx, srcBucket, srcKey, nil)
|
||||
if err != nil {
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
package s3
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/md5"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"io"
|
||||
|
||||
"github.com/OpenListTeam/OpenList/v4/internal/fs"
|
||||
"github.com/OpenListTeam/OpenList/v4/internal/model"
|
||||
"github.com/OpenListTeam/OpenList/v4/internal/stream"
|
||||
"github.com/OpenListTeam/OpenList/v4/pkg/http_range"
|
||||
"github.com/OpenListTeam/OpenList/v4/pkg/utils"
|
||||
)
|
||||
|
||||
type objectMetadata struct {
|
||||
headers map[string]string
|
||||
hash []byte
|
||||
etag string
|
||||
}
|
||||
|
||||
// Only reuse upload metadata when it still describes the current content.
|
||||
func (b *s3Backend) loadMetadata(name string, hash []byte) (map[string]string, string) {
|
||||
if value, ok := b.meta.Load(name); ok {
|
||||
metadata := value.(objectMetadata)
|
||||
if bytes.Equal(metadata.hash, hash) {
|
||||
return metadata.headers, metadata.etag
|
||||
}
|
||||
}
|
||||
return nil, ""
|
||||
}
|
||||
|
||||
// Metadata alone cannot identify content: clients can preserve both file size
|
||||
// and modification time when overwriting an object. Use the driver's MD5 when
|
||||
// available, otherwise hash the complete content, including for HEAD and ranges.
|
||||
func getObjectHash(ctx context.Context, name string, obj model.Obj) ([]byte, error) {
|
||||
if value := obj.GetHash().GetHash(utils.MD5); value != "" {
|
||||
hash, err := hex.DecodeString(value)
|
||||
if err != nil || len(hash) != md5.Size {
|
||||
return nil, fmt.Errorf("invalid object MD5: %q", value)
|
||||
}
|
||||
return hash, nil
|
||||
}
|
||||
link, file, err := fs.Link(ctx, name, model.LinkArgs{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer link.Close()
|
||||
size := link.ContentLength
|
||||
if size <= 0 {
|
||||
size = file.GetSize()
|
||||
}
|
||||
ranges, err := stream.GetRangeReaderFromLink(size, link)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
reader, err := ranges.RangeRead(ctx, http_range.Range{Length: -1})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer reader.Close()
|
||||
hash := md5.New()
|
||||
n, err := utils.CopyWithBuffer(hash, reader)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if n != size {
|
||||
return nil, io.ErrUnexpectedEOF
|
||||
}
|
||||
return hash.Sum(nil), nil
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
package s3
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/md5"
|
||||
"encoding/hex"
|
||||
"net/http/httptest"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/OpenListTeam/OpenList/v4/internal/model"
|
||||
"github.com/OpenListTeam/OpenList/v4/pkg/utils"
|
||||
)
|
||||
|
||||
func TestObjectHashFromDriver(t *testing.T) {
|
||||
sum := md5.Sum([]byte("content"))
|
||||
obj := &model.Object{HashInfo: utils.NewHashInfo(utils.MD5, hex.EncodeToString(sum[:]))}
|
||||
hash, err := getObjectHash(context.Background(), "object", obj)
|
||||
if err != nil || hex.EncodeToString(hash) != hex.EncodeToString(sum[:]) {
|
||||
t.Fatalf("hash = %x, err = %v", hash, err)
|
||||
}
|
||||
for _, value := range []string{"invalid", "ab"} {
|
||||
obj.HashInfo = utils.NewHashInfo(utils.MD5, value)
|
||||
if _, err := getObjectHash(context.Background(), "object", obj); err == nil {
|
||||
t.Fatalf("accepted invalid MD5 %q", value)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestMultipartMetadataMatchesCurrentContent(t *testing.T) {
|
||||
b := &s3Backend{meta: new(sync.Map)}
|
||||
oldHash := md5.Sum([]byte("old"))
|
||||
newHash := md5.Sum([]byte("new"))
|
||||
b.meta.Store("object", objectMetadata{
|
||||
headers: map[string]string{"Content-Type": "text/plain"},
|
||||
hash: oldHash[:],
|
||||
etag: `"multipart-2"`,
|
||||
})
|
||||
meta, etag := b.loadMetadata("object", oldHash[:])
|
||||
if etag != `"multipart-2"` || meta["Content-Type"] != "text/plain" {
|
||||
t.Fatal("lost metadata for unchanged multipart content")
|
||||
}
|
||||
meta, etag = b.loadMetadata("object", newHash[:])
|
||||
if meta != nil || etag != "" {
|
||||
t.Fatal("reused a multipart validator after a same-size content change")
|
||||
}
|
||||
}
|
||||
|
||||
func TestConditionalRequestsDoNotRedirect(t *testing.T) {
|
||||
for _, method := range []string{"GET", "PUT"} {
|
||||
for _, name := range []string{"If-Match", "If-None-Match", "If-Modified-Since", "If-Unmodified-Since", "If-Range"} {
|
||||
for _, value := range []string{"*", ""} {
|
||||
r := httptest.NewRequest(method, "/bucket/object", nil)
|
||||
r.Header.Set(name, value)
|
||||
if url, ok := directObjectURL(r, nil); ok || url != "" {
|
||||
t.Fatalf("redirected %s with %s", method, name)
|
||||
}
|
||||
if url, ok := directUploadURL(r, nil); ok || url != "" {
|
||||
t.Fatalf("redirected %s with %s", method, name)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -234,7 +234,9 @@ func (b *s3Backend) CompleteMultipartUpload(ctx context.Context, bucket, object
|
||||
|
||||
defer combined.Close()
|
||||
|
||||
err := b.putStream(ctx, bucket, object, state.meta, combined, total)
|
||||
sum := md5.Sum(concat)
|
||||
etag := fmt.Sprintf("%q", fmt.Sprintf("%s-%d", hex.EncodeToString(sum[:]), len(ordered)))
|
||||
err := b.putStream(ctx, bucket, object, state.meta, combined, total, etag)
|
||||
if err != nil {
|
||||
// Leave the upload in place so the client may retry completion, per
|
||||
// the gofakes3 MultipartBackend contract.
|
||||
@@ -244,8 +246,6 @@ func (b *s3Backend) CompleteMultipartUpload(ctx context.Context, bucket, object
|
||||
// Success: drop bookkeeping and clean up part files.
|
||||
b.removeUpload(uploadID)
|
||||
|
||||
sum := md5.Sum(concat)
|
||||
etag := fmt.Sprintf("%q", fmt.Sprintf("%s-%d", hex.EncodeToString(sum[:]), len(ordered)))
|
||||
log.Debugf("s3 multipart: completed upload %s -> %s/%s (%d bytes)", uploadID, bucket, object, total)
|
||||
return "", etag, nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
package s3
|
||||
|
||||
import "net/http"
|
||||
|
||||
func hasPreconditions(r *http.Request) bool {
|
||||
for _, name := range []string{"If-Match", "If-None-Match", "If-Modified-Since", "If-Unmodified-Since", "If-Range"} {
|
||||
if _, ok := r.Header[name]; ok {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -38,7 +38,7 @@ func redirectHandler(next http.Handler, authPairs map[string]string) http.Handle
|
||||
}
|
||||
|
||||
func directObjectURL(r *http.Request, authPairs map[string]string) (string, bool) {
|
||||
if r.Method != http.MethodGet {
|
||||
if r.Method != http.MethodGet || hasPreconditions(r) {
|
||||
return "", false
|
||||
}
|
||||
if hasNonObjectQuery(r) || !s3RequestAuthorized(r, authPairs) {
|
||||
@@ -75,7 +75,7 @@ func directObjectURL(r *http.Request, authPairs map[string]string) (string, bool
|
||||
}
|
||||
|
||||
func directUploadURL(r *http.Request, authPairs map[string]string) (string, bool) {
|
||||
if r.Method != http.MethodPut || r.ContentLength < 0 {
|
||||
if r.Method != http.MethodPut || r.ContentLength < 0 || hasPreconditions(r) {
|
||||
return "", false
|
||||
}
|
||||
if hasNonObjectQuery(r) || !s3RequestAuthorized(r, authPairs) {
|
||||
|
||||
Reference in New Issue
Block a user