From 0df9529b32322307a962ab6aa522539586e17702 Mon Sep 17 00:00:00 2001 From: HFO4 <912394456@qq.com> Date: Sun, 20 Mar 2022 11:23:55 +0800 Subject: [PATCH] Feat: generating token and callback url for OSS muiltpart upload, support resume upload in sever-side uploading for OSS --- go.mod | 5 +- go.sum | 13 +-- middleware/auth.go | 24 ++-- models/migration.go | 3 +- models/policy.go | 12 +- pkg/filesystem/chunk/chunk.go | 5 + pkg/filesystem/driver/onedrive/api.go | 4 +- pkg/filesystem/driver/onedrive/api_test.go | 6 +- pkg/filesystem/driver/oss/handler.go | 126 ++++++++++++++------- pkg/filesystem/driver/remote/client.go | 2 +- pkg/serializer/upload.go | 8 +- routers/router.go | 3 +- service/callback/upload.go | 21 ++++ 13 files changed, 148 insertions(+), 84 deletions(-) diff --git a/go.mod b/go.mod index 349fedf..a7de23c 100644 --- a/go.mod +++ b/go.mod @@ -4,7 +4,7 @@ go 1.13 require ( github.com/DATA-DOG/go-sqlmock v1.3.3 - github.com/aliyun/aliyun-oss-go-sdk v2.0.5+incompatible + github.com/HFO4/aliyun-oss-go-sdk v2.2.2+incompatible github.com/aws/aws-sdk-go v1.31.5 github.com/baiyubin/aliyun-sts-go-sdk v0.0.0-20180326062324-cfa1a18b161f // indirect github.com/duo-labs/webauthn v0.0.0-20191119193225-4bf9a0f776d4 @@ -20,12 +20,11 @@ require ( github.com/gomodule/redigo v2.0.0+incompatible github.com/google/go-querystring v1.0.0 github.com/gorilla/websocket v1.4.1 - github.com/hashicorp/go-version v1.2.0 + github.com/hashicorp/go-version v1.3.0 github.com/jinzhu/gorm v1.9.11 github.com/juju/ratelimit v1.0.1 github.com/mattn/go-colorable v0.1.4 // indirect github.com/mojocn/base64Captcha v0.0.0-20190801020520-752b1cd608b2 - github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646 github.com/pkg/errors v0.9.1 github.com/pquerna/otp v1.2.0 github.com/qiniu/api.v7/v7 v7.4.0 diff --git a/go.sum b/go.sum index 402c072..b8568f5 100644 --- a/go.sum +++ b/go.sum @@ -5,6 +5,8 @@ cloud.google.com/go v0.37.4/go.mod h1:NHPJ89PdicEuT9hdPXMROBD91xc5uRDxsMtSB16k7h github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/DATA-DOG/go-sqlmock v1.3.3 h1:CWUqKXe0s8A2z6qCgkP4Kru7wC11YoAnoupUKFDnH08= github.com/DATA-DOG/go-sqlmock v1.3.3/go.mod h1:f/Ixk793poVmq4qj/V1dPUg2JEAKC73Q5eFN3EC/SaM= +github.com/HFO4/aliyun-oss-go-sdk v2.2.2+incompatible h1:cWidHgoT2ye1YwzP5sz7xkqOK6I84mqrayJSU23oOUo= +github.com/HFO4/aliyun-oss-go-sdk v2.2.2+incompatible/go.mod h1:8KDiKVrHK/UbXAhj+iQGp1m40rQa+UAvzBi7m22KywI= github.com/QcloudApi/qcloud_sign_golang v0.0.0-20141224014652-e4130a326409/go.mod h1:1pk82RBxDY/JZnPQrtqHlUFfCctgdorsd9M06fMynOM= github.com/Shopify/sarama v1.19.0/go.mod h1:FVkBWblsNy7DGZRfXLU0O9RCGt5g3g3yEuWXgklEdEo= github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI= @@ -12,9 +14,6 @@ github.com/agl/ed25519 v0.0.0-20170116200512-5312a6153412 h1:w1UutsfOrms1J05zt7I github.com/agl/ed25519 v0.0.0-20170116200512-5312a6153412/go.mod h1:WPjqKcmVOxf0XSf3YxCJs6N6AOSrOx3obionmG7T0y0= github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= -github.com/aliyun/aliyun-oss-go-sdk v2.0.0/go.mod h1:T/Aws4fEfogEE9v+HPhhw+CntffsBHJ8nXQCwKr0/g8= -github.com/aliyun/aliyun-oss-go-sdk v2.0.5+incompatible h1:A3oZlWPD/Poa19FvNbw+Zu4yKAurDBTjlRDilYGBiS4= -github.com/aliyun/aliyun-oss-go-sdk v2.0.5+incompatible/go.mod h1:T/Aws4fEfogEE9v+HPhhw+CntffsBHJ8nXQCwKr0/g8= github.com/apache/thrift v0.12.0/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ= github.com/aws/aws-sdk-go v1.31.5 h1:DFA7BzTydO4etqsTja+x7UfkOKQUv1xzEluLvNk81L0= github.com/aws/aws-sdk-go v1.31.5/go.mod h1:5zCpMtNQVjRREroY7sYe8lOMRSxkhG6MZveU8YkpAk0= @@ -119,8 +118,8 @@ github.com/gorilla/sessions v1.1.3 h1:uXoZdcdA5XdXF3QzuSlheVRUvjl+1rKY7zBXL68L9R github.com/gorilla/sessions v1.1.3/go.mod h1:8KCfur6+4Mqcc6S0FEfKuN15Vl5MgXW92AE8ovaJD0w= github.com/gorilla/websocket v1.4.1 h1:q7AeDBpnBk8AogcD4DSag/Ukw/KV+YhzLj2bP5HvKCM= github.com/gorilla/websocket v1.4.1/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= -github.com/hashicorp/go-version v1.2.0 h1:3vNe/fWF5CBgRIguda1meWhsZHy3m8gCJ5wx+dIzX/E= -github.com/hashicorp/go-version v1.2.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA= +github.com/hashicorp/go-version v1.3.0 h1:McDWVJIU/y+u1BRV06dPaLfLCaT7fUTJLp5r04x7iNw= +github.com/hashicorp/go-version v1.3.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA= github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8= github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU= github.com/jinzhu/gorm v1.9.11 h1:gaHGvE+UnWGlbWG4Y3FUwY1EcZ5n6S9WtqBA/uySMLE= @@ -179,8 +178,6 @@ github.com/mojocn/base64Captcha v0.0.0-20190801020520-752b1cd608b2/go.mod h1:wAQ github.com/mozillazg/go-httpheader v0.2.1 h1:geV7TrjbL8KXSyvghnFm+NyTux/hxwueTSrwhe88TQQ= github.com/mozillazg/go-httpheader v0.2.1/go.mod h1:jJ8xECTlalr6ValeXYdOF8fFUISeBAdw6E61aqQma60= github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= -github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646 h1:zYyBkD/k9seD2A7fsi6Oo2LfFZAehjjQMERAvZLEDnQ= -github.com/nfnt/resize v0.0.0-20180221191011-83c6a9932646/go.mod h1:jpp1/29i3P1S/RLdc7JQKbRpFeM1dOBd8T9ki5s+AY8= github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE= github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE= github.com/onsi/gomega v1.4.3/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY= @@ -247,7 +244,6 @@ golang.org/x/crypto v0.0.0-20190325154230-a5d413f7728c/go.mod h1:djNgcEr1/C05ACk golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4 h1:HuIa8hRrWRSrqYzx1qI49NNxhdi2PrY7gxVSq1JjLDc= golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= -golang.org/x/image v0.0.0-20190501045829-6d32002ffd75 h1:TbGuee8sSq15Iguxu4deQ7+Bqq/d2rsQejGcEtADAMQ= golang.org/x/image v0.0.0-20190501045829-6d32002ffd75/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js= golang.org/x/image v0.0.0-20211028202545-6944b10bf410 h1:hTftEOvwiOq2+O8k2D5/Q7COC7k5Qcrgc2TFURJYnvQ= golang.org/x/image v0.0.0-20211028202545-6944b10bf410/go.mod h1:023OzeP/+EPmXeapQh35lcL3II3LrY8Ic+EFFKVhULM= @@ -289,7 +285,6 @@ golang.org/x/sys v0.0.0-20190813064441-fde4db37ae7a h1:aYOabOQFp6Vj6W1F80affTUvO golang.org/x/sys v0.0.0-20190813064441-fde4db37ae7a/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= -golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs= golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= golang.org/x/text v0.3.6 h1:aRYxNxv6iGQlyVaZmk6ZgYEDa+Jg18DxebPSrd6bg1M= golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= diff --git a/middleware/auth.go b/middleware/auth.go index 34237d5..28c00fc 100644 --- a/middleware/auth.go +++ b/middleware/auth.go @@ -2,7 +2,9 @@ package middleware import ( "github.com/cloudreve/Cloudreve/v3/pkg/filesystem" + "github.com/cloudreve/Cloudreve/v3/pkg/filesystem/driver/oss" "github.com/cloudreve/Cloudreve/v3/pkg/mq" + "github.com/cloudreve/Cloudreve/v3/pkg/util" "net/http" model "github.com/cloudreve/Cloudreve/v3/models" @@ -209,21 +211,13 @@ func QiniuCallbackAuth() gin.HandlerFunc { // OSSCallbackAuth 阿里云OSS回调签名验证 func OSSCallbackAuth() gin.HandlerFunc { return func(c *gin.Context) { - //// 验证key并查找用户 - //resp, _ := uploadCallbackCheck(c) - //if resp.Code != 0 { - // c.JSON(401, serializer.GeneralUploadCallbackFailed{Error: resp.Msg}) - // c.Abort() - // return - //} - // - //err := oss.VerifyCallbackSignature(c.Request) - //if err != nil { - // util.Log().Debug("回调签名验证失败，%s", err) - // c.JSON(401, serializer.GeneralUploadCallbackFailed{Error: "回调签名验证失败"}) - // c.Abort() - // return - //} + err := oss.VerifyCallbackSignature(c.Request) + if err != nil { + util.Log().Debug("回调签名验证失败，%s", err) + c.JSON(401, serializer.GeneralUploadCallbackFailed{Error: "回调签名验证失败"}) + c.Abort() + return + } c.Next() } diff --git a/models/migration.go b/models/migration.go index 6b557fc..e383f7e 100644 --- a/models/migration.go +++ b/models/migration.go @@ -124,8 +124,7 @@ func addDefaultSettings() { {Name: "share_download_session_timeout", Value: `2073600`, Type: "timeout"}, {Name: "onedrive_callback_check", Value: `20`, Type: "timeout"}, {Name: "folder_props_timeout", Value: `300`, Type: "timeout"}, - {Name: "onedrive_chunk_retries", Value: `5`, Type: "retry"}, - {Name: "slave_chunk_retries", Value: `5`, Type: "retry"}, + {Name: "chunk_retries", Value: `5`, Type: "retry"}, {Name: "onedrive_source_timeout", Value: `1800`, Type: "timeout"}, {Name: "reset_after_upload_failed", Value: `0`, Type: "upload"}, {Name: "login_captcha", Value: `0`, Type: "login"}, diff --git a/models/policy.go b/models/policy.go index 6671692..0f7f299 100644 --- a/models/policy.go +++ b/models/policy.go @@ -60,6 +60,8 @@ type PolicyOption struct { ServerSideEndpoint string `json:"server_side_endpoint,omitempty"` // 分片上传的分片大小 ChunkSize uint64 `json:"chunk_size,omitempty"` + // 分片上传时是否需要预留空间 + PlaceholderWithSize bool `json:"placeholder_with_size,omitempty"` } var thumbSuffix = map[string][]string{ @@ -226,7 +228,15 @@ func (policy *Policy) IsThumbGenerateNeeded() bool { // IsUploadPlaceholderWithSize 返回此策略创建上传会话时是否需要预留空间 func (policy *Policy) IsUploadPlaceholderWithSize() bool { - return policy.Type == "remote" + if policy.Type == "remote" { + return true + } + + if policy.Type == "onedrive" || policy.Type == "oss" { + return policy.OptionsSerialized.PlaceholderWithSize + } + + return false } // CanStructureBeListed 返回存储策略是否能被前台列物理目录 diff --git a/pkg/filesystem/chunk/chunk.go b/pkg/filesystem/chunk/chunk.go index 5eb2893..82e6cac 100644 --- a/pkg/filesystem/chunk/chunk.go +++ b/pkg/filesystem/chunk/chunk.go @@ -74,6 +74,11 @@ func (c *ChunkGroup) Total() int64 { return int64(c.fileInfo.Size) } +// Num returns the total chunk number +func (c *ChunkGroup) Num() int { + return int(c.chunkNum) +} + // RangeHeader returns header value of Content-Range func (c *ChunkGroup) RangeHeader() string { return fmt.Sprintf("bytes %d-%d/%d", c.Start(), c.Start()+c.Length()-1, c.Total()) diff --git a/pkg/filesystem/driver/onedrive/api.go b/pkg/filesystem/driver/onedrive/api.go index 2aa49de..14c1278 100644 --- a/pkg/filesystem/driver/onedrive/api.go +++ b/pkg/filesystem/driver/onedrive/api.go @@ -279,7 +279,7 @@ func (client *Client) Upload(ctx context.Context, file fsctx.FileHeader) error { // Initial chunk groups chunks := chunk.NewChunkGroup(file, client.Policy.OptionsSerialized.ChunkSize, &backoff.ConstantBackoff{ - Max: model.GetIntSetting("onedrive_chunk_retries", 5), + Max: model.GetIntSetting("chunk_retries", 5), Sleep: chunkRetrySleep, }) @@ -327,7 +327,7 @@ func (client *Client) SimpleUpload(ctx context.Context, dst string, body io.Read if v, ok := ctx.Value(fsctx.RetryCtx).(int); ok { retried = v } - if retried < model.GetIntSetting("onedrive_chunk_retries", 5) { + if retried < model.GetIntSetting("chunk_retries", 5) { retried++ util.Log().Debug("文件[%s]上传失败[%s]，5秒钟后重试", dst, err) time.Sleep(time.Duration(5) * time.Second) diff --git a/pkg/filesystem/driver/onedrive/api_test.go b/pkg/filesystem/driver/onedrive/api_test.go index 02fb294..8acc6db 100644 --- a/pkg/filesystem/driver/onedrive/api_test.go +++ b/pkg/filesystem/driver/onedrive/api_test.go @@ -535,7 +535,7 @@ func TestClient_UploadChunk(t *testing.T) { // 最后分片，第一次失败，重试后成功 { - cache.Set("setting_onedrive_chunk_retries", "1", 0) + cache.Set("setting_chunk_retries", "1", 0) client.Credential.ExpiresIn = 0 go func() { time.Sleep(time.Duration(2) * time.Second) @@ -641,7 +641,7 @@ func TestClient_SimpleUpload(t *testing.T) { client, _ := NewClient(&model.Policy{}) client.Credential.AccessToken = "AccessToken" client.Credential.ExpiresIn = time.Now().Add(time.Duration(100) * time.Hour).Unix() - cache.Set("setting_onedrive_chunk_retries", "1", 0) + cache.Set("setting_chunk_retries", "1", 0) // 请求失败，并重试 { @@ -651,7 +651,7 @@ func TestClient_SimpleUpload(t *testing.T) { asserts.Nil(res) } - cache.Set("setting_onedrive_chunk_retries", "0", 0) + cache.Set("setting_chunk_retries", "0", 0) // 返回未知响应 { client.Credential.ExpiresIn = time.Now().Add(time.Duration(100) * time.Hour).Unix() diff --git a/pkg/filesystem/driver/oss/handler.go b/pkg/filesystem/driver/oss/handler.go index bbe3fc1..10e7fdd 100644 --- a/pkg/filesystem/driver/oss/handler.go +++ b/pkg/filesystem/driver/oss/handler.go @@ -2,8 +2,6 @@ package oss import ( "context" - "crypto/hmac" - "crypto/sha1" "encoding/base64" "encoding/json" "errors" @@ -15,8 +13,10 @@ import ( "strings" "time" - "github.com/aliyun/aliyun-oss-go-sdk/oss" + "github.com/HFO4/aliyun-oss-go-sdk/oss" model "github.com/cloudreve/Cloudreve/v3/models" + "github.com/cloudreve/Cloudreve/v3/pkg/filesystem/chunk" + "github.com/cloudreve/Cloudreve/v3/pkg/filesystem/chunk/backoff" "github.com/cloudreve/Cloudreve/v3/pkg/filesystem/fsctx" "github.com/cloudreve/Cloudreve/v3/pkg/filesystem/response" "github.com/cloudreve/Cloudreve/v3/pkg/request" @@ -48,6 +48,10 @@ type Driver struct { type key int const ( + chunkRetrySleep = time.Duration(5) * time.Second + + // MultiPartUploadThreshold 服务端使用分片上传的阈值 + MultiPartUploadThreshold uint64 = 5 * (1 << 30) // 5GB // VersionID 文件版本标识 VersionID key = iota ) @@ -244,12 +248,34 @@ func (handler Driver) Put(ctx context.Context, file fsctx.FileHeader) error { } // 上传文件 - err := handler.bucket.PutObject(fileInfo.SavePath, file, options...) + //err := handler.bucket.PutObject(fileInfo.SavePath, file, options...) + //if err != nil { + // return err + //} + + imur, err := handler.bucket.InitiateMultipartUpload(fileInfo.SavePath, options...) if err != nil { + return fmt.Errorf("failed to initiate multipart upload: %w", err) + } + + chunks := chunk.NewChunkGroup(file, handler.Policy.OptionsSerialized.ChunkSize, &backoff.ConstantBackoff{ + Max: model.GetIntSetting("chunk_retries", 5), + Sleep: chunkRetrySleep, + }) + + uploadFunc := func(current *chunk.ChunkGroup, content io.Reader) error { + _, err := handler.bucket.UploadPart(imur, content, current.Length(), current.Index()+1) return err } - return nil + for chunks.Next() { + if err := chunks.Process(uploadFunc); err != nil { + return fmt.Errorf("failed to upload chunk #%d: %w", chunks.Index(), err) + } + } + + _, err = handler.bucket.CompleteMultipartUpload(imur, oss.CompleteAll("yes"), oss.ForbidOverWrite(!overwrite)) + return err } // Delete 删除一个或多个文件， @@ -395,6 +421,10 @@ func (handler Driver) signSourceURL(ctx context.Context, path string, ttl int64, // Token 获取上传策略和认证Token func (handler Driver) Token(ctx context.Context, ttl int64, uploadSession *serializer.UploadSession, file fsctx.FileHeader) (*serializer.UploadCredential, error) { + if err := handler.InitOSSClient(false); err != nil { + return nil, err + } + // 生成回调地址 siteURL := model.GetSiteURL() apiBaseURI, _ := url.Parse("/api/v3/callback/oss/" + uploadSession.Key) @@ -406,61 +436,69 @@ func (handler Driver) Token(ctx context.Context, ttl int64, uploadSession *seria CallbackBody: `{"name":${x:fname},"source_name":${object},"size":${size},"pic_info":"${imageInfo.width},${imageInfo.height}"}`, CallbackBodyType: "application/json", } - - // 上传策略 - savePath := file.Info().SavePath - postPolicy := UploadPolicy{ - Expiration: time.Now().UTC().Add(time.Duration(ttl) * time.Second).Format(time.RFC3339), - Conditions: []interface{}{ - map[string]string{"bucket": handler.Policy.BucketName}, - []string{"starts-with", "$key", path.Dir(savePath)}, - }, + callbackPolicyJSON, err := json.Marshal(callbackPolicy) + if err != nil { + return nil, fmt.Errorf("failed to encode callback policy: %w", err) } + callbackPolicyEncoded := base64.StdEncoding.EncodeToString(callbackPolicyJSON) - if handler.Policy.MaxSize > 0 { - postPolicy.Conditions = append(postPolicy.Conditions, - []interface{}{"content-length-range", 0, handler.Policy.MaxSize}) + // 初始化分片上传 + fileInfo := file.Info() + options := []oss.Option{ + oss.Expires(time.Now().Add(time.Duration(ttl) * time.Second)), + oss.ForbidOverWrite(true), } - - return handler.getUploadCredential(ctx, postPolicy, callbackPolicy, ttl, savePath) -} - -func (handler Driver) getUploadCredential(ctx context.Context, policy UploadPolicy, callback CallbackPolicy, TTL int64, savePath string) (*serializer.UploadCredential, error) { - // 处理回调策略 - callbackPolicyEncoded := "" - if callback.CallbackURL != "" { - callbackPolicyJSON, err := json.Marshal(callback) + imur, err := handler.bucket.InitiateMultipartUpload(fileInfo.SavePath, options...) + if err != nil { + return nil, fmt.Errorf("failed to initialize multipart upload: %w", err) + } + uploadSession.OSSUploadID = imur.UploadID + + // 为每个分片签名上传 URL + chunks := chunk.NewChunkGroup(file, handler.Policy.OptionsSerialized.ChunkSize, &backoff.ConstantBackoff{}) + urls := make([]string, chunks.Num()) + for chunks.Next() { + err := chunks.Process(func(c *chunk.ChunkGroup, chunk io.Reader) error { + signedURL, err := handler.bucket.SignURL(fileInfo.SavePath, oss.HTTPPut, ttl, + oss.PartNumber(c.Index()+1), + oss.UploadID(imur.UploadID), + oss.ContentType("application/octet-stream")) + if err != nil { + return err + } + + urls[c.Index()] = signedURL + return nil + }) if err != nil { return nil, err } - callbackPolicyEncoded = base64.StdEncoding.EncodeToString(callbackPolicyJSON) - policy.Conditions = append(policy.Conditions, map[string]string{"callback": callbackPolicyEncoded}) } - // 编码上传策略 - policyJSON, err := json.Marshal(policy) + // 签名完成分片上传的URL + completeURL, err := handler.bucket.SignURL(fileInfo.SavePath, oss.HTTPPost, ttl, + oss.UploadID(imur.UploadID), + oss.Expires(time.Now().Add(time.Duration(ttl)*time.Second)), + oss.CompleteAll("yes"), + oss.ForbidOverWrite(true), + oss.CallbackParam(callbackPolicyEncoded)) if err != nil { return nil, err } - policyEncoded := base64.StdEncoding.EncodeToString(policyJSON) - - // 签名上传策略 - hmacSign := hmac.New(sha1.New, []byte(handler.Policy.SecretKey)) - _, err = io.WriteString(hmacSign, policyEncoded) - if err != nil { - return nil, err - } - signature := base64.StdEncoding.EncodeToString(hmacSign.Sum(nil)) return &serializer.UploadCredential{ - Policy: fmt.Sprintf("%s:%s", callbackPolicyEncoded, policyEncoded), - Path: savePath, - AccessKey: handler.Policy.AccessKey, - Token: signature, + SessionID: uploadSession.Key, + ChunkSize: handler.Policy.OptionsSerialized.ChunkSize, + UploadID: imur.UploadID, + UploadURLs: urls, + Callback: completeURL, }, nil } // 取消上传凭证 func (handler Driver) CancelToken(ctx context.Context, uploadSession *serializer.UploadSession) error { - return nil + if err := handler.InitOSSClient(false); err != nil { + return err + } + return handler.bucket.AbortMultipartUpload(oss.InitiateMultipartUploadResult{UploadID: uploadSession.OSSUploadID, Key: uploadSession.SavePath}, nil) } diff --git a/pkg/filesystem/driver/remote/client.go b/pkg/filesystem/driver/remote/client.go index 17bbb92..b6759f1 100644 --- a/pkg/filesystem/driver/remote/client.go +++ b/pkg/filesystem/driver/remote/client.go @@ -90,7 +90,7 @@ func (c *remoteClient) Upload(ctx context.Context, file fsctx.FileHeader) error // Initial chunk groups chunks := chunk.NewChunkGroup(file, c.policy.OptionsSerialized.ChunkSize, &backoff.ConstantBackoff{ - Max: model.GetIntSetting("slave_chunk_retries", 5), + Max: model.GetIntSetting("chunk_retries", 5), Sleep: chunkRetrySleep, }) diff --git a/pkg/serializer/upload.go b/pkg/serializer/upload.go index 485b843..713641b 100644 --- a/pkg/serializer/upload.go +++ b/pkg/serializer/upload.go @@ -23,8 +23,10 @@ type UploadCredential struct { SessionID string `json:"sessionID"` ChunkSize uint64 `json:"chunkSize"` // 分块大小，0 为部分快 Expires int64 `json:"expires"` // 上传凭证过期时间， Unix 时间戳 - UploadURLs []string `json:"uploadURLs"` - Credential string `json:"credential"` + UploadURLs []string `json:"uploadURLs,omitempty"` + Credential string `json:"credential,omitempty"` + UploadID string `json:"uploadID,omitempty"` + Callback string `json:"callback,omitempty"` // 回调地址 Token string `json:"token"` Policy string `json:"policy"` @@ -32,7 +34,6 @@ type UploadCredential struct { AccessKey string `json:"ak"` KeyTime string `json:"key_time,omitempty"` // COS用有效期 Key string `json:"key,omitempty"` // 文件标识符，通常为回调key - Callback string `json:"callback,omitempty"` // 回调地址 } // UploadSession 上传会话 @@ -48,6 +49,7 @@ type UploadSession struct { Callback string // 回调 URL 地址 CallbackSecret string // 回调 URL OneDriveUploadURL string + OSSUploadID string } // UploadCallback 上传回调正文 diff --git a/routers/router.go b/routers/router.go index 9f61a33..05b700a 100644 --- a/routers/router.go +++ b/routers/router.go @@ -250,7 +250,8 @@ func InitMasterRouter() *gin.Engine { ) // 阿里云OSS策略上传回调 callback.POST( - "oss/:key", + "oss/:sessionID", + middleware.UseUploadSession("oss"), middleware.OSSCallbackAuth(), controllers.OSSCallback, ) diff --git a/service/callback/upload.go b/service/callback/upload.go index 4f0495e..da0d67e 100644 --- a/service/callback/upload.go +++ b/service/callback/upload.go @@ -243,3 +243,24 @@ func (service *S3Callback) PreProcess(c *gin.Context) serializer.Response { return ProcessCallback(service, c) } + +// PreProcess 对OneDrive客户端回调进行预处理验证 +func (service *UploadCallbackService) PreProcess(c *gin.Context) serializer.Response { + // 创建文件系统 + fs, err := filesystem.NewFileSystemFromCallback(c) + if err != nil { + return serializer.Err(serializer.CodePolicyNotAllowed, err.Error(), err) + } + defer fs.Recycle() + + // 获取回调会话 + uploadSession := c.MustGet(filesystem.UploadSessionCtx).(*serializer.UploadSession) + + // 验证文件大小 + if uploadSession.Size != service.Size { + fs.Handler.Delete(context.Background(), []string{uploadSession.SavePath}) + return serializer.Err(serializer.CodeUploadFailed, "文件大小不一致", nil) + } + + return ProcessCallback(service, c) +}