Infrastructure/Cloudreve
0
Fork 0
Code Issues Pull requests Activity

Feat: creating upload session and credential from master server

Browse source
This commit is contained in:
HFO4 2022-02-28 17:52:59 +08:00
parent 118d738797
commit 7214e59c25
12 changed files with 153 additions and 136 deletions
middleware
cluster.go
pkg
aria2/rpc
resp.go
auth
auth.go
filesystem
archive.go
driver/remote
client.go
fsctx
stream.go
hooks.go
request
request.go
routers/controllers
file.goslave.go
service/explorer
file.goupload.go

9
middleware/cluster.go
View file

@ -1,6 +1,7 @@
package middleware
import (
"github.com/cloudreve/Cloudreve/v3/pkg/auth"
"github.com/cloudreve/Cloudreve/v3/pkg/cluster"
"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
"github.com/gin-gonic/gin"
@ -10,9 +11,9 @@ import (
// MasterMetadata 解析主机节点发来请求的包含主机节点信息的元数据
func MasterMetadata() gin.HandlerFunc {
return func(c *gin.Context) {
c.Set("MasterSiteID", c.GetHeader("X-Cr-Site-Id"))
c.Set("MasterSiteURL", c.GetHeader("X-Cr-Site-Url"))
c.Set("MasterVersion", c.GetHeader("X-Cr-Cloudreve-Version"))
c.Set("MasterSiteID", c.GetHeader(auth.CrHeaderPrefix+"Site-Id"))
c.Set("MasterSiteURL", c.GetHeader(auth.CrHeaderPrefix+"Site-Url"))
c.Set("MasterVersion", c.GetHeader(auth.CrHeaderPrefix+"Cloudreve-Version"))
c.Next()
}
}
@ -41,7 +42,7 @@ func UseSlaveAria2Instance(clusterController cluster.Controller) gin.HandlerFunc
func SlaveRPCSignRequired(nodePool cluster.Pool) gin.HandlerFunc {
return func(c *gin.Context) {
nodeID, err := strconv.ParseUint(c.GetHeader("X-Cr-Node-Id"), 10, 64)
nodeID, err := strconv.ParseUint(c.GetHeader(auth.CrHeaderPrefix+"Node-Id"), 10, 64)
if err != nil {
c.JSON(200, serializer.ParamErr("未知的主机节点ID", err))
c.Abort()

4
pkg/aria2/rpc/resp.go
View file

@ -11,7 +11,7 @@ type StatusInfo struct {
UploadLength string `json:"uploadLength"` // Uploaded length of the download in bytes.
BitField string `json:"bitfield"` // Hexadecimal representation of the download progress. The highest bit corresponds to the piece at index 0. Any set bits indicate loaded pieces, while unset bits indicate not yet loaded and/or missing pieces. Any overflow bits at the end are set to zero. When the download was not started yet, this key will not be included in the response.
DownloadSpeed string `json:"downloadSpeed"` // Download speed of this download measured in bytes/sec.
UploadSpeed string `json:"uploadSpeed"` // Upload speed of this download measured in bytes/sec.
UploadSpeed string `json:"uploadSpeed"` // LocalUpload speed of this download measured in bytes/sec.
InfoHash string `json:"infoHash"` // InfoHash. BitTorrent only.
NumSeeders string `json:"numSeeders"` // The number of seeders aria2 has connected to. BitTorrent only.
Seeder string `json:"seeder"` // true if the local endpoint is a seeder. Otherwise false. BitTorrent only.
@ -60,7 +60,7 @@ type PeerInfo struct {
AmChoking string `json:"amChoking"` // true if aria2 is choking the peer. Otherwise false.
PeerChoking string `json:"peerChoking"` // true if the peer is choking aria2. Otherwise false.
DownloadSpeed string `json:"downloadSpeed"` // Download speed (byte/sec) that this client obtains from the peer.
UploadSpeed string `json:"uploadSpeed"` // Upload speed(byte/sec) that this client uploads to the peer.
UploadSpeed string `json:"uploadSpeed"` // LocalUpload speed(byte/sec) that this client uploads to the peer.
Seeder string `json:"seeder"` // true if this peer is a seeder. Otherwise false.
}

6
pkg/auth/auth.go
View file

@ -23,6 +23,8 @@ var (
ErrExpired = serializer.NewError(serializer.CodeSignExpired, "签名已过期", nil)
)
const CrHeaderPrefix = "X-Cr-"
// General 通用的认证接口
var General Auth
@ -69,7 +71,7 @@ func CheckRequest(instance Auth, r *http.Request) error {
func getSignContent(r *http.Request) (rawSignString string) {
// 读取所有body正文
var body = []byte{}
if strings.Contains(r.URL.Path, "/api/v3/slave/upload/") {
if !strings.Contains(r.URL.Path, "/api/v3/slave/upload/") {
if r.Body != nil {
body, _ = ioutil.ReadAll(r.Body)
_ = r.Body.Close()
@ -80,7 +82,7 @@ func getSignContent(r *http.Request) (rawSignString string) {
// 决定要签名的header
var signedHeader []string
for k, _ := range r.Header {
if strings.HasPrefix(k, "X-Cr-") && k != "X-Cr-Filename" {
if strings.HasPrefix(k, CrHeaderPrefix) && k != CrHeaderPrefix+"Filename" {
signedHeader = append(signedHeader, fmt.Sprintf("%s=%s", k, r.Header.Get(k)))
}
}

1
pkg/filesystem/archive.go
View file

@ -308,7 +308,6 @@ func (fs *FileSystem) Decompress(ctx context.Context, src, dst string) error {
Size: uint64(size),
Name: path.Base(dst),
VirtualPath: path.Dir(dst),
Mode: 0,
})
fileStream.Close()
if err != nil {

10
pkg/filesystem/driver/remote/client.go
View file

@ -14,7 +14,8 @@ import (
)
const (
basePath = "/api/v3/slave"
basePath = "/api/v3/slave"
OverwriteHeader = auth.CrHeaderPrefix + "Overwrite"
)
// Client to operate remote slave server
@ -79,20 +80,17 @@ func (c *remoteClient) CreateUploadSession(ctx context.Context, session *seriali
}
func (c *remoteClient) GetUploadURL(ttl int64, sessionID string) (string, string, error) {
base, err := url.Parse(c.policy.BaseURL)
base, err := url.Parse(c.policy.Server)
if err != nil {
return "", "", err
}
base.Path = path.Join(base.Path, "upload", sessionID)
base.Path = path.Join(base.Path, basePath, "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
}

4
pkg/filesystem/fsctx/stream.go
View file

@ -10,8 +10,8 @@ type WriteMode int
const (
Overwrite WriteMode = 0x00001
// Append 只适用于本地策略
Append = 0x00002
Nop = 0x00004
Append WriteMode = 0x00002
Nop WriteMode = 0x00004
)
type UploadTaskInfo struct {

35
pkg/filesystem/hooks.go
View file

@ -2,6 +2,7 @@ package filesystem
import (
"context"
"errors"
model "github.com/cloudreve/Cloudreve/v3/models"
"github.com/cloudreve/Cloudreve/v3/pkg/cache"
"github.com/cloudreve/Cloudreve/v3/pkg/conf"
@ -53,31 +54,6 @@ func (fs *FileSystem) Trigger(ctx context.Context, name string, file fsctx.FileH
return nil
}
// HookSlaveUploadValidate Slave模式下对文件上传的一系列验证
func HookSlaveUploadValidate(ctx context.Context, fs *FileSystem, file fsctx.FileHeader) error {
policy := ctx.Value(fsctx.UploadPolicyCtx).(serializer.UploadPolicy)
fileInfo := file.Info()
// 验证单文件尺寸
if policy.MaxSize > 0 {
if fileInfo.Size > policy.MaxSize {
return ErrFileSizeTooBig
}
}
// 验证文件名
if !fs.ValidateLegalName(ctx, fileInfo.FileName) {
return ErrIllegalObjectName
}
// 验证扩展名
if len(policy.AllowedExtension) > 0 && !IsInExtensionList(policy.AllowedExtension, fileInfo.FileName) {
return ErrFileExtensionNotAllowed
}
return nil
}
// HookValidateFile 一系列对文件检验的集合
func HookValidateFile(ctx context.Context, fs *FileSystem, file fsctx.FileHeader) error {
fileInfo := file.Info()
@ -151,7 +127,7 @@ func HookCleanFileContent(ctx context.Context, fs *FileSystem, file fsctx.FileHe
File: ioutil.NopCloser(strings.NewReader("")),
SavePath: file.Info().SavePath,
Size: 0,
Model: fsctx.Overwrite,
Mode: fsctx.Overwrite,
})
}
@ -203,6 +179,7 @@ func GenericAfterUpdate(ctx context.Context, fs *FileSystem, newFile fsctx.FileH
// SlaveAfterUpload Slave模式下上传完成钩子
func SlaveAfterUpload(ctx context.Context, fs *FileSystem, fileHeader fsctx.FileHeader) error {
return errors.New("")
policy := ctx.Value(fsctx.UploadPolicyCtx).(serializer.UploadPolicy)
fileInfo := fileHeader.Info()
@ -287,10 +264,8 @@ func HookClearFileHeaderSize(ctx context.Context, fs *FileSystem, fileHeader fsc
// HookTruncateFileTo 将物理文件截断至 size
func HookTruncateFileTo(size uint64) Hook {
return func(ctx context.Context, fs *FileSystem, fileHeader fsctx.FileHeader) error {
if fs.Policy.Type == "local" {
if driver, ok := fs.Handler.(local.Driver); ok {
return driver.Truncate(ctx, fileHeader.Info().SavePath, size)
}
if handler, ok := fs.Handler.(local.Driver); ok {
return handler.Truncate(ctx, fileHeader.Info().SavePath, size)
}
return nil

8
pkg/request/request.go
View file

@ -97,13 +97,13 @@ func (c HTTPClient) Request(method, target string, body io.Reader, opts ...Optio
}
if options.masterMeta && conf.SystemConfig.Mode == "master" {
req.Header.Add("X-Cr-Site-Url", model.GetSiteURL().String())
req.Header.Add("X-Cr-Site-Id", model.GetSettingByName("siteID"))
req.Header.Add("X-Cr-Cloudreve-Version", conf.BackendVersion)
req.Header.Add(auth.CrHeaderPrefix+"Site-Url", model.GetSiteURL().String())
req.Header.Add(auth.CrHeaderPrefix+"Site-Id", model.GetSettingByName("siteID"))
req.Header.Add(auth.CrHeaderPrefix+"Cloudreve-Version", conf.BackendVersion)
}
if options.slaveNodeID != "" && conf.SystemConfig.Mode == "slave" {
req.Header.Add("X-Cr-Node-Id", options.slaveNodeID)
req.Header.Add(auth.CrHeaderPrefix+"Node-Id", options.slaveNodeID)
}
if options.contentLength != -1 {

2
routers/controllers/file.go
View file

@ -289,7 +289,7 @@ func FileUpload(c *gin.Context) {
var service explorer.UploadService
if err := c.ShouldBindUri(&service); err == nil {
res := service.Upload(ctx, c)
res := service.LocalUpload(ctx, c)
c.JSON(200, res)
request.BlackHole(c.Request.Body)
} else {

141
routers/controllers/slave.go
View file

@ -5,9 +5,7 @@ import (
"net/url"
"strconv"
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem"
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/driver/local"
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/fsctx"
"github.com/cloudreve/Cloudreve/v3/pkg/request"
"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
"github.com/cloudreve/Cloudreve/v3/service/admin"
"github.com/cloudreve/Cloudreve/v3/service/aria2"
@ -20,74 +18,87 @@ import (
func SlaveUpload(c *gin.Context) {
// 创建上下文
ctx, cancel := context.WithCancel(context.Background())
ctx = context.WithValue(ctx, fsctx.GinCtx, c)
defer cancel()
// 创建匿名文件系统
fs, err := filesystem.NewAnonymousFileSystem()
if err != nil {
c.JSON(200, serializer.Err(serializer.CodePolicyNotAllowed, err.Error(), err))
return
}
fs.Handler = local.Driver{}
// 从请求中取得上传策略
uploadPolicyRaw := c.GetHeader("X-Cr-Policy")
if uploadPolicyRaw == "" {
c.JSON(200, serializer.ParamErr("未指定上传策略", nil))
return
}
// 解析上传策略
uploadPolicy, err := serializer.DecodeUploadPolicy(uploadPolicyRaw)
if err != nil {
c.JSON(200, serializer.ParamErr("上传策略格式有误", err))
return
}
ctx = context.WithValue(ctx, fsctx.UploadPolicyCtx, *uploadPolicy)
// 取得文件大小
fileSize, err := strconv.ParseUint(c.Request.Header.Get("Content-Length"), 10, 64)
if err != nil {
var service explorer.UploadService
if err := c.ShouldBindUri(&service); err == nil {
res := service.SlaveUpload(ctx, c)
c.JSON(200, res)
request.BlackHole(c.Request.Body)
} else {
c.JSON(200, ErrorResponse(err))
return
}
// 解码文件名和路径
fileName, err := url.QueryUnescape(c.Request.Header.Get("X-Cr-FileName"))
if err != nil {
c.JSON(200, ErrorResponse(err))
return
}
fileData := fsctx.FileStream{
MIMEType: c.Request.Header.Get("Content-Type"),
File: c.Request.Body,
Name: fileName,
Size: fileSize,
}
// 给文件系统分配钩子
fs.Use("BeforeUpload", filesystem.HookSlaveUploadValidate)
fs.Use("AfterUploadCanceled", filesystem.HookDeleteTempFile)
fs.Use("AfterUpload", filesystem.SlaveAfterUpload)
fs.Use("AfterValidateFailed", filesystem.HookDeleteTempFile)
//// 是否允许覆盖
//if c.Request.Header.Get("X-Cr-Overwrite") == "false" {
// fileData.Mode = fsctx.Create
//// 创建上下文
//ctx, cancel := context.WithCancel(context.Background())
//ctx = context.WithValue(ctx, fsctx.GinCtx, c)
//defer cancel()
//
//// 创建匿名文件系统
//fs, err := filesystem.NewAnonymousFileSystem()
//if err != nil {
// c.JSON(200, serializer.Err(serializer.CodePolicyNotAllowed, err.Error(), err))
// return
//}
// 执行上传
err = fs.Upload(ctx, &fileData)
if err != nil {
c.JSON(200, serializer.Err(serializer.CodeUploadFailed, err.Error(), err))
return
}
c.JSON(200, serializer.Response{
Code: 0,
})
//fs.Handler = local.Driver{}
//
//// 从请求中取得上传策略
//uploadPolicyRaw := c.GetHeader("X-Cr-Policy")
//if uploadPolicyRaw == "" {
// c.JSON(200, serializer.ParamErr("未指定上传策略", nil))
// return
//}
//
//// 解析上传策略
//uploadPolicy, err := serializer.DecodeUploadPolicy(uploadPolicyRaw)
//if err != nil {
// c.JSON(200, serializer.ParamErr("上传策略格式有误", err))
// return
//}
//ctx = context.WithValue(ctx, fsctx.UploadPolicyCtx, *uploadPolicy)
//
//// 取得文件大小
//fileSize, err := strconv.ParseUint(c.Request.Header.Get("Content-Length"), 10, 64)
//if err != nil {
// c.JSON(200, ErrorResponse(err))
// return
//}
//
//// 解码文件名和路径
//fileName, err := url.QueryUnescape(c.Request.Header.Get("X-Cr-FileName"))
//if err != nil {
// c.JSON(200, ErrorResponse(err))
// return
//}
//
//fileData := fsctx.FileStream{
// MIMEType: c.Request.Header.Get("Content-Type"),
// File: c.Request.Body,
// Name: fileName,
// Size: fileSize,
//}
//
//// 给文件系统分配钩子
//fs.Use("BeforeUpload", filesystem.HookSlaveUploadValidate)
//fs.Use("AfterUploadCanceled", filesystem.HookDeleteTempFile)
//fs.Use("AfterUpload", filesystem.SlaveAfterUpload)
//fs.Use("AfterValidateFailed", filesystem.HookDeleteTempFile)
//
////// 是否允许覆盖
////if c.Request.Header.Get("X-Cr-Overwrite") == "false" {
//// fileData.Mode = fsctx.Create
////}
//
//// 执行上传
//err = fs.LocalUpload(ctx, &fileData)
//if err != nil {
// c.JSON(200, serializer.Err(serializer.CodeUploadFailed, err.Error(), err))
// return
//}
//
//c.JSON(200, serializer.Response{
// Code: 0,
//})
}
// SlaveGetUploadSession 从机创建上传会话

2
service/explorer/file.go
View file

@ -374,7 +374,7 @@ func (service *FileIDService) PutContent(ctx context.Context, c *gin.Context) se
MIMEType: c.Request.Header.Get("Content-Type"),
File: c.Request.Body,
Size: fileSize,
Model: fsctx.Overwrite,
Mode: fsctx.Overwrite,
}
// 创建文件系统

67
service/explorer/upload.go
View file

@ -4,8 +4,8 @@ import (
"context"
"fmt"
model "github.com/cloudreve/Cloudreve/v3/models"
"github.com/cloudreve/Cloudreve/v3/pkg/auth"
"github.com/cloudreve/Cloudreve/v3/pkg/cache"
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem"
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/fsctx"
"github.com/cloudreve/Cloudreve/v3/pkg/hashid"
@ -72,11 +72,11 @@ type UploadService struct {
Index int `uri:"index" form:"index" binding:"min=0"`
}
// Upload 处理本机文件分片上传
func (service *UploadService) Upload(ctx context.Context, c *gin.Context) serializer.Response {
// LocalUpload 处理本机文件分片上传
func (service *UploadService) LocalUpload(ctx context.Context, c *gin.Context) serializer.Response {
uploadSessionRaw, ok := cache.Get(filesystem.UploadSessionCachePrefix + service.ID)
if !ok {
return serializer.Err(serializer.CodeUploadSessionExpired, "Upload session expired or not exist", nil)
return serializer.Err(serializer.CodeUploadSessionExpired, "LocalUpload session expired or not exist", nil)
}
uploadSession := uploadSessionRaw.(serializer.UploadSession)
@ -87,13 +87,13 @@ func (service *UploadService) Upload(ctx context.Context, c *gin.Context) serial
}
if uploadSession.UID != fs.User.ID {
return serializer.Err(serializer.CodeUploadSessionExpired, "Upload session expired or not exist", nil)
return serializer.Err(serializer.CodeUploadSessionExpired, "LocalUpload session expired or not exist", nil)
}
// 查找上传会话创建的占位文件
file, err := model.GetFilesByUploadSession(service.ID, fs.User.ID)
if err != nil {
return serializer.Err(serializer.CodeUploadSessionExpired, "Upload session file placeholder not exist", err)
return serializer.Err(serializer.CodeUploadSessionExpired, "LocalUpload session file placeholder not exist", err)
}
// 重设 fs 存储策略
@ -120,10 +120,34 @@ func (service *UploadService) Upload(ctx context.Context, c *gin.Context) serial
util.Log().Info("尝试上传覆盖分片[%d] Start=%d", service.Index, actualSizeStart)
}
return processChunkUpload(ctx, c, fs, &uploadSession, service.Index, file)
return processChunkUpload(ctx, c, fs, &uploadSession, service.Index, file, fsctx.Append|fsctx.Overwrite)
}
func processChunkUpload(ctx context.Context, c *gin.Context, fs *filesystem.FileSystem, session *serializer.UploadSession, index int, file *model.File) serializer.Response {
// SlaveUpload 处理从机文件分片上传
func (service *UploadService) SlaveUpload(ctx context.Context, c *gin.Context) serializer.Response {
uploadSessionRaw, ok := cache.Get(filesystem.UploadSessionCachePrefix + service.ID)
if !ok {
return serializer.Err(serializer.CodeUploadSessionExpired, "LocalUpload session expired or not exist", nil)
}
uploadSession := uploadSessionRaw.(serializer.UploadSession)
fs, err := filesystem.NewAnonymousFileSystem()
if err != nil {
return serializer.Err(serializer.CodePolicyNotAllowed, err.Error(), err)
}
// 解析需要的参数
service.Index, _ = strconv.Atoi(c.Query("chunk"))
mode := fsctx.Append
if c.GetHeader(auth.CrHeaderPrefix+"Overwrite") == "true" {
mode |= fsctx.Overwrite
}
return processChunkUpload(ctx, c, fs, &uploadSession, service.Index, nil, mode)
}
func processChunkUpload(ctx context.Context, c *gin.Context, fs *filesystem.FileSystem, session *serializer.UploadSession, index int, file *model.File, mode fsctx.WriteMode) serializer.Response {
// 取得并校验文件大小是否符合分片要求
chunkSize := session.Policy.OptionsSerialized.ChunkSize
isLastChunk := session.Policy.OptionsSerialized.ChunkSize == 0 || uint64(index+1)*chunkSize >= session.Size
@ -148,23 +172,30 @@ func processChunkUpload(ctx context.Context, c *gin.Context, fs *filesystem.File
Name: session.Name,
VirtualPath: session.VirtualPath,
SavePath: session.SavePath,
Mode: fsctx.Append | fsctx.Overwrite,
Mode: mode,
AppendStart: chunkSize * uint64(index),
Model: file,
LastModified: session.LastModified,
}
// 给文件系统分配钩子
fs.Use("BeforeUpload", filesystem.HookValidateCapacity)
fs.Use("AfterUploadCanceled", filesystem.HookTruncateFileTo(fileData.AppendStart))
fs.Use("AfterUpload", filesystem.HookChunkUploaded)
if isLastChunk {
fs.Use("AfterUpload", filesystem.HookChunkUploadFinished)
fs.Use("AfterUpload", filesystem.HookGenerateThumb)
fs.Use("AfterUpload", filesystem.HookDeleteUploadSession(session.Key))
}
fs.Use("AfterValidateFailed", filesystem.HookTruncateFileTo(fileData.AppendStart))
fs.Use("AfterValidateFailed", filesystem.HookChunkUploadFailed)
if file != nil {
fs.Use("BeforeUpload", filesystem.HookValidateCapacity)
fs.Use("AfterUpload", filesystem.HookChunkUploaded)
fs.Use("AfterValidateFailed", filesystem.HookChunkUploadFailed)
if isLastChunk {
fs.Use("AfterUpload", filesystem.HookChunkUploadFinished)
fs.Use("AfterUpload", filesystem.HookGenerateThumb)
fs.Use("AfterUpload", filesystem.HookDeleteUploadSession(session.Key))
}
} else {
if isLastChunk {
fs.Use("AfterUpload", filesystem.SlaveAfterUpload)
}
}
// 执行上传
uploadCtx := context.WithValue(ctx, fsctx.GinCtx, c)
@ -193,7 +224,7 @@ func (service *UploadSessionService) Delete(ctx context.Context, c *gin.Context)
// 查找需要删除的上传会话的占位文件
file, err := model.GetFilesByUploadSession(service.ID, fs.User.ID)
if err != nil {
return serializer.Err(serializer.CodeUploadSessionExpired, "Upload session file placeholder not exist", err)
return serializer.Err(serializer.CodeUploadSessionExpired, "LocalUpload session file placeholder not exist", err)
}
// 删除文件