diff --git a/pkg/auth/auth.go b/pkg/auth/auth.go
index 094a37f..84b5a5f 100644
--- a/pkg/auth/auth.go
+++ b/pkg/auth/auth.go
@@ -64,12 +64,12 @@ func CheckRequest(instance Auth, r *http.Request) error {
return instance.Check(getSignContent(r), sign[0])
}
-// getSignContent 签名请求 path、正文、以`X-`开头的 Header. 如果 Header 中包含 `X-Policy`，
+// getSignContent 签名请求 path、正文、以`X-`开头的 Header. 如果请求 path 为从机上传 API，
// 则不对正文签名。返回待签名/验证的字符串
func getSignContent(r *http.Request) (rawSignString string) {
// 读取所有body正文
var body = []byte{}
- if _, ok := r.Header["X-Cr-Policy"]; !ok {
+ if strings.Contains(r.URL.Path, "/api/v3/slave/upload/") {
if r.Body != nil {
body, _ = ioutil.ReadAll(r.Body)
_ = r.Body.Close()
diff --git a/pkg/filesystem/driver/local/handler.go b/pkg/filesystem/driver/local/handler.go
index b2b621a..e91f838 100644
--- a/pkg/filesystem/driver/local/handler.go
+++ b/pkg/filesystem/driver/local/handler.go
@@ -136,8 +136,20 @@ func (handler Driver) Put(ctx context.Context, file fsctx.FileHeader) error {
return err
}
- if uint64(stat.Size()) != fileInfo.AppendStart {
+ if uint64(stat.Size()) < fileInfo.AppendStart {
return errors.New("未上传完成的文件分片与预期大小不一致")
+ } else if uint64(stat.Size()) > fileInfo.AppendStart {
+ out.Close()
+ if err := handler.Truncate(ctx, dst, fileInfo.AppendStart); err != nil {
+ return fmt.Errorf("覆盖分片时发生错误: %w", err)
+ }
+
+ out, err = os.OpenFile(dst, os.O_APPEND|os.O_CREATE|os.O_WRONLY, Perm)
+ defer out.Close()
+ if err != nil {
+ util.Log().Warning("无法打开或创建文件，%s", err)
+ return err
+ }
}
}
diff --git a/pkg/filesystem/driver/remote/client.go b/pkg/filesystem/driver/remote/client.go
index b2f7443..feadc00 100644
--- a/pkg/filesystem/driver/remote/client.go
+++ b/pkg/filesystem/driver/remote/client.go
@@ -1,18 +1,26 @@
package remote
import (
+ "context"
"encoding/json"
model "github.com/cloudreve/Cloudreve/v3/models"
"github.com/cloudreve/Cloudreve/v3/pkg/auth"
"github.com/cloudreve/Cloudreve/v3/pkg/request"
"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
+ "net/http"
"net/url"
+ "path"
"strings"
)
+const (
+ basePath = "/api/v3/slave"
+)
+
// Client to operate remote slave server
type Client interface {
- CreateUploadSession(session *serializer.UploadSession, ttl int64) error
+ CreateUploadSession(ctx context.Context, session *serializer.UploadSession, ttl int64) error
+ GetUploadURL(ttl int64, sessionID string) (string, string, error)
}
// NewClient creates new Client from given policy
@@ -23,7 +31,7 @@ func NewClient(policy *model.Policy) (Client, error) {
return nil, err
}
- base, _ := url.Parse("/api/v3/slave")
+ base, _ := url.Parse(basePath)
signTTL := model.GetIntSetting("slave_api_timeout", 60)
return &remoteClient{
@@ -43,7 +51,7 @@ type remoteClient struct {
httpClient request.Client
}
-func (c *remoteClient) CreateUploadSession(session *serializer.UploadSession, ttl int64) error {
+func (c *remoteClient) CreateUploadSession(ctx context.Context, session *serializer.UploadSession, ttl int64) error {
reqBodyEncoded, err := json.Marshal(map[string]interface{}{
"session": session,
"ttl": ttl,
@@ -57,6 +65,7 @@ func (c *remoteClient) CreateUploadSession(session *serializer.UploadSession, tt
"PUT",
"upload",
bodyReader,
+ request.WithContext(ctx),
).CheckHTTPResponse(200).DecodeResponse()
if err != nil {
return err
@@ -68,3 +77,22 @@ func (c *remoteClient) CreateUploadSession(session *serializer.UploadSession, tt
return nil
}
+
+func (c *remoteClient) GetUploadURL(ttl int64, sessionID string) (string, string, error) {
+ base, err := url.Parse(c.policy.BaseURL)
+ if err != nil {
+ return "", "", err
+ }
+
+ base.Path = path.Join(base.Path, "upload", sessionID)
+
+ req, err := http.NewRequest("POST", base.String(), nil)
+ if err != nil {
+ return "", "", err
+ }
+
+ req.Header["X-Cr-Overwrite"] = []string{"true"}
+
+ req = auth.SignRequest(c.authInstance, req, ttl)
+ return req.URL.String(), req.Header["Authorization"][0], nil
+}
diff --git a/pkg/filesystem/driver/remote/handler.go b/pkg/filesystem/driver/remote/handler.go
index 01cf6fc..c68e8bd 100644
--- a/pkg/filesystem/driver/remote/handler.go
+++ b/pkg/filesystem/driver/remote/handler.go
@@ -323,29 +323,23 @@ func (handler Driver) Source(
// Token 获取上传策略和认证Token
func (handler Driver) Token(ctx context.Context, ttl int64, uploadSession *serializer.UploadSession, file fsctx.FileHeader) (*serializer.UploadCredential, error) {
- if err := handler.client.CreateUploadSession(uploadSession, ttl); err != nil {
+ // 在从机端创建上传会话
+ if err := handler.client.CreateUploadSession(ctx, uploadSession, ttl); err != nil {
+ return nil, err
+ }
+
+ // 获取上传地址
+ uploadURL, sign, err := handler.client.GetUploadURL(ttl, uploadSession.Key)
+ if err != nil {
return nil, err
}
return &serializer.UploadCredential{
- SessionID: uploadSession.Key,
- ChunkSize: handler.Policy.OptionsSerialized.ChunkSize,
+ SessionID: uploadSession.Key,
+ ChunkSize: handler.Policy.OptionsSerialized.ChunkSize,
+ UploadURLs: []string{uploadURL},
+ Credential: sign,
}, nil
- //// 生成回调地址
- //siteURL := model.GetSiteURL()
- //apiBaseURI, _ := url.Parse("/api/v3/callback/remote/" + uploadSession.Key)
- //apiURL := siteURL.ResolveReference(apiBaseURI)
- //
- //// 生成上传策略
- //policy := serializer.UploadPolicy{
- // SavePath: handler.Policy.DirNameRule,
- // FileName: handler.Policy.FileNameRule,
- // AutoRename: handler.Policy.AutoRename,
- // MaxSize: handler.Policy.MaxSize,
- // AllowedExtension: handler.Policy.OptionsSerialized.FileType,
- // CallbackURL: apiURL.String(),
- //}
- //return handler.getUploadCredential(ctx, policy, ttl)
}
func (handler Driver) getUploadCredential(ctx context.Context, policy serializer.UploadPolicy, TTL int64) (serializer.UploadCredential, error) {
diff --git a/pkg/filesystem/hooks.go b/pkg/filesystem/hooks.go
index 8a8951a..f07b370 100644
--- a/pkg/filesystem/hooks.go
+++ b/pkg/filesystem/hooks.go
@@ -301,7 +301,7 @@ func HookChunkUploaded(ctx context.Context, fs *FileSystem, fileHeader fsctx.Fil
fileInfo := fileHeader.Info()
// 更新文件大小
- return fileInfo.Model.(*model.File).UpdateSize(fileInfo.Model.(*model.File).GetSize() + fileInfo.Size)
+ return fileInfo.Model.(*model.File).UpdateSize(fileInfo.AppendStart + fileInfo.Size)
}
// HookChunkUploadFailed 单个分片上传失败后
@@ -309,7 +309,7 @@ func HookChunkUploadFailed(ctx context.Context, fs *FileSystem, fileHeader fsctx
fileInfo := fileHeader.Info()
// 更新文件大小
- return fileInfo.Model.(*model.File).UpdateSize(fileInfo.Model.(*model.File).GetSize() - fileInfo.Size)
+ return fileInfo.Model.(*model.File).UpdateSize(fileInfo.AppendStart)
}
// HookChunkUploadFinished 分片上传结束后处理文件
diff --git a/pkg/serializer/upload.go b/pkg/serializer/upload.go
index e13a5fe..b1f78a1 100644
--- a/pkg/serializer/upload.go
+++ b/pkg/serializer/upload.go
@@ -20,9 +20,11 @@ type UploadPolicy struct {
// UploadCredential 返回给客户端的上传凭证
type UploadCredential struct {
- SessionID string `json:"sessionID"`
- ChunkSize uint64 `json:"chunkSize"` // 分块大小，0 为部分快
- Expires int64 `json:"expires"` // 上传凭证过期时间， Unix 时间戳
+ SessionID string `json:"sessionID"`
+ ChunkSize uint64 `json:"chunkSize"` // 分块大小，0 为部分快
+ Expires int64 `json:"expires"` // 上传凭证过期时间， Unix 时间戳
+ UploadURLs []string `json:"uploadURLs"`
+ Credential string `json:"credential"`
Token string `json:"token"`
Policy string `json:"policy"`
diff --git a/routers/router.go b/routers/router.go
index 9008746..2b9052d 100644
--- a/routers/router.go
+++ b/routers/router.go
@@ -46,7 +46,7 @@ func InitSlaveRouter() *gin.Engine {
// 接收主机心跳包
v3.POST("heartbeat", controllers.SlaveHeartbeat)
// 上传
- v3.POST("upload", controllers.SlaveUpload)
+ v3.POST("upload/:sessionId", controllers.SlaveUpload)
// 创建上传会话上传
v3.PUT("upload", controllers.SlaveGetUploadSession)
// 下载
diff --git a/service/explorer/upload.go b/service/explorer/upload.go
index 2a044b0..15cf426 100644
--- a/service/explorer/upload.go
+++ b/service/explorer/upload.go
@@ -66,10 +66,10 @@ func (service *CreateUploadSessionService) Create(ctx context.Context, c *gin.Co
}
}
-// UploadService 本机策略上传服务
+// UploadService 本机及从机策略上传服务
type UploadService struct {
ID string `uri:"sessionId" binding:"required"`
- Index int `uri:"index" binding:"min=0"`
+ Index int `uri:"index" form:"index" binding:"min=0"`
}
// Upload 处理本机文件分片上传
@@ -117,8 +117,7 @@ func (service *UploadService) Upload(ctx context.Context, c *gin.Context) serial
}
if expectedSizeStart > actualSizeStart {
- util.Log().Warning("尝试上传覆盖分片[%d]，数据将被忽略", service.Index)
- return serializer.Response{}
+ util.Log().Info("尝试上传覆盖分片[%d] Start=%d", service.Index, actualSizeStart)
}
return processChunkUpload(ctx, c, fs, &uploadSession, service.Index, file)