// Copyright (C) 2019 Nicola Murino // // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU Affero General Public License as published // by the Free Software Foundation, version 3. // // This program is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU Affero General Public License for more details. // // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . package common import ( "errors" "fmt" "io/fs" "path" "sync" "sync/atomic" "time" "github.com/drakkan/sftpgo/v2/internal/dataprovider" "github.com/drakkan/sftpgo/v2/internal/logger" "github.com/drakkan/sftpgo/v2/internal/metric" "github.com/drakkan/sftpgo/v2/internal/vfs" ) var ( // ErrTransferClosed defines the error returned for a closed transfer ErrTransferClosed = errors.New("transfer already closed") ) // BaseTransfer contains protocols common transfer details for an upload or a download. type BaseTransfer struct { //nolint:maligned ID int64 BytesSent atomic.Int64 BytesReceived atomic.Int64 Fs vfs.Fs File vfs.File Connection *BaseConnection cancelFn func() fsPath string effectiveFsPath string requestPath string ftpMode string start time.Time MaxWriteSize int64 MinWriteOffset int64 InitialSize int64 truncatedSize int64 isNewFile bool transferType int AbortTransfer atomic.Bool aTime time.Time mTime time.Time transferQuota dataprovider.TransferQuota metadata map[string]string sync.Mutex errAbort error ErrTransfer error } // NewBaseTransfer returns a new BaseTransfer and adds it to the given connection func NewBaseTransfer(file vfs.File, conn *BaseConnection, cancelFn func(), fsPath, effectiveFsPath, requestPath string, transferType int, minWriteOffset, initialSize, maxWriteSize, truncatedSize int64, isNewFile bool, fs vfs.Fs, transferQuota dataprovider.TransferQuota, ) *BaseTransfer { t := &BaseTransfer{ ID: conn.GetTransferID(), File: file, Connection: conn, cancelFn: cancelFn, fsPath: fsPath, effectiveFsPath: effectiveFsPath, start: time.Now(), transferType: transferType, MinWriteOffset: minWriteOffset, InitialSize: initialSize, isNewFile: isNewFile, requestPath: requestPath, MaxWriteSize: maxWriteSize, truncatedSize: truncatedSize, transferQuota: transferQuota, Fs: fs, } t.AbortTransfer.Store(false) t.BytesSent.Store(0) t.BytesReceived.Store(0) conn.AddTransfer(t) return t } // GetTransferQuota returns data transfer quota limits func (t *BaseTransfer) GetTransferQuota() dataprovider.TransferQuota { return t.transferQuota } // SetFtpMode sets the FTP mode for the current transfer func (t *BaseTransfer) SetFtpMode(mode string) { t.ftpMode = mode } // GetID returns the transfer ID func (t *BaseTransfer) GetID() int64 { return t.ID } // GetType returns the transfer type func (t *BaseTransfer) GetType() int { return t.transferType } // GetSize returns the transferred size func (t *BaseTransfer) GetSize() int64 { if t.transferType == TransferDownload { return t.BytesSent.Load() } return t.BytesReceived.Load() } // GetDownloadedSize returns the transferred size func (t *BaseTransfer) GetDownloadedSize() int64 { return t.BytesSent.Load() } // GetUploadedSize returns the transferred size func (t *BaseTransfer) GetUploadedSize() int64 { return t.BytesReceived.Load() } // GetStartTime returns the start time func (t *BaseTransfer) GetStartTime() time.Time { return t.start } // GetAbortError returns the error to send to the client if the transfer was aborted func (t *BaseTransfer) GetAbortError() error { t.Lock() defer t.Unlock() if t.errAbort != nil { return t.errAbort } return getQuotaExceededError(t.Connection.protocol) } // SignalClose signals that the transfer should be closed after the next read/write. // The optional error argument allow to send a specific error, otherwise a generic // transfer aborted error is sent func (t *BaseTransfer) SignalClose(err error) { t.Lock() t.errAbort = err t.Unlock() t.AbortTransfer.Store(true) } // GetTruncatedSize returns the truncated sized if this is an upload overwriting // an existing file func (t *BaseTransfer) GetTruncatedSize() int64 { return t.truncatedSize } // HasSizeLimit returns true if there is an upload or download size limit func (t *BaseTransfer) HasSizeLimit() bool { if t.MaxWriteSize > 0 { return true } if t.transferQuota.HasSizeLimits() { return true } return false } // GetVirtualPath returns the transfer virtual path func (t *BaseTransfer) GetVirtualPath() string { return t.requestPath } // GetFsPath returns the transfer filesystem path func (t *BaseTransfer) GetFsPath() string { return t.fsPath } // SetTimes stores access and modification times if fsPath matches the current file func (t *BaseTransfer) SetTimes(fsPath string, atime time.Time, mtime time.Time) bool { if fsPath == t.GetFsPath() { t.aTime = atime t.mTime = mtime return true } return false } // GetRealFsPath returns the real transfer filesystem path. // If atomic uploads are enabled this differ from fsPath func (t *BaseTransfer) GetRealFsPath(fsPath string) string { if fsPath == t.GetFsPath() { if t.File != nil || vfs.IsLocalOsFs(t.Fs) { return t.effectiveFsPath } return t.fsPath } return "" } // SetMetadata sets the metadata for the file func (t *BaseTransfer) SetMetadata(val map[string]string) { t.metadata = val } // SetCancelFn sets the cancel function for the transfer func (t *BaseTransfer) SetCancelFn(cancelFn func()) { t.cancelFn = cancelFn } // ConvertError accepts an error that occurs during a read or write and // converts it into a more understandable form for the client if it is a // well-known type of error func (t *BaseTransfer) ConvertError(err error) error { var pathError *fs.PathError if errors.As(err, &pathError) { return fmt.Errorf("%s %s: %s", pathError.Op, t.GetVirtualPath(), pathError.Err.Error()) } return t.Connection.GetFsError(t.Fs, err) } // CheckRead returns an error if read if not allowed func (t *BaseTransfer) CheckRead() error { if t.transferQuota.AllowedDLSize == 0 && t.transferQuota.AllowedTotalSize == 0 { return nil } if t.transferQuota.AllowedTotalSize > 0 { if t.BytesSent.Load()+t.BytesReceived.Load() > t.transferQuota.AllowedTotalSize { return t.Connection.GetReadQuotaExceededError() } } else if t.transferQuota.AllowedDLSize > 0 { if t.BytesSent.Load() > t.transferQuota.AllowedDLSize { return t.Connection.GetReadQuotaExceededError() } } return nil } // CheckWrite returns an error if write if not allowed func (t *BaseTransfer) CheckWrite() error { if t.MaxWriteSize > 0 && t.BytesReceived.Load() > t.MaxWriteSize { return t.Connection.GetQuotaExceededError() } if t.transferQuota.AllowedULSize == 0 && t.transferQuota.AllowedTotalSize == 0 { return nil } if t.transferQuota.AllowedTotalSize > 0 { if t.BytesSent.Load()+t.BytesReceived.Load() > t.transferQuota.AllowedTotalSize { return t.Connection.GetQuotaExceededError() } } else if t.transferQuota.AllowedULSize > 0 { if t.BytesReceived.Load() > t.transferQuota.AllowedULSize { return t.Connection.GetQuotaExceededError() } } return nil } // Truncate changes the size of the opened file. // Supported for local fs only func (t *BaseTransfer) Truncate(fsPath string, size int64) (int64, error) { if fsPath == t.GetFsPath() { if t.File != nil { initialSize := t.InitialSize err := t.File.Truncate(size) if err == nil { t.Lock() t.InitialSize = size if t.MaxWriteSize > 0 { sizeDiff := initialSize - size t.MaxWriteSize += sizeDiff metric.TransferCompleted(t.BytesSent.Load(), t.BytesReceived.Load(), t.transferType, t.ErrTransfer, vfs.IsSFTPFs(t.Fs)) if t.transferQuota.HasSizeLimits() { go func(ulSize, dlSize int64, user dataprovider.User) { dataprovider.UpdateUserTransferQuota(&user, ulSize, dlSize, false) //nolint:errcheck }(t.BytesReceived.Load(), t.BytesSent.Load(), t.Connection.User) } t.BytesReceived.Store(0) } t.Unlock() } t.Connection.Log(logger.LevelDebug, "file %q truncated to size %v max write size %v new initial size %v err: %v", fsPath, size, t.MaxWriteSize, t.InitialSize, err) return initialSize, err } if size == 0 && t.BytesSent.Load() == 0 { // for cloud providers the file is always truncated to zero, we don't support append/resume for uploads. // For buffered SFTP and local fs we can have buffered bytes so we returns an error if !vfs.IsBufferedLocalOrSFTPFs(t.Fs) { return 0, nil } } return 0, vfs.ErrVfsUnsupported } return 0, errTransferMismatch } // TransferError is called if there is an unexpected error. // For example network or client issues func (t *BaseTransfer) TransferError(err error) { t.Lock() defer t.Unlock() if t.ErrTransfer != nil { return } t.ErrTransfer = err if t.cancelFn != nil { t.cancelFn() } elapsed := time.Since(t.start).Nanoseconds() / 1000000 t.Connection.Log(logger.LevelError, "Unexpected error for transfer, path: %q, error: \"%v\" bytes sent: %v, "+ "bytes received: %v transfer running since %v ms", t.fsPath, t.ErrTransfer, t.BytesSent.Load(), t.BytesReceived.Load(), elapsed) } func (t *BaseTransfer) getUploadFileSize() (int64, int, error) { var fileSize int64 var deletedFiles int info, err := t.Fs.Stat(t.fsPath) if err == nil { fileSize = info.Size() } if t.ErrTransfer != nil && vfs.IsCryptOsFs(t.Fs) { errDelete := t.Fs.Remove(t.fsPath, false) if errDelete != nil { t.Connection.Log(logger.LevelWarn, "error removing partial crypto file %q: %v", t.fsPath, errDelete) } else { fileSize = 0 deletedFiles = 1 t.BytesReceived.Store(0) t.MinWriteOffset = 0 } } return fileSize, deletedFiles, err } // return 1 if the file is outside the user home dir func (t *BaseTransfer) checkUploadOutsideHomeDir(err error) int { if err == nil { return 0 } if t.ErrTransfer == nil { t.ErrTransfer = err } if Config.TempPath == "" { return 0 } err = t.Fs.Remove(t.effectiveFsPath, false) t.Connection.Log(logger.LevelWarn, "upload in temp path cannot be renamed, delete temporary file: %q, deletion error: %v", t.effectiveFsPath, err) // the file is outside the home dir so don't update the quota t.BytesReceived.Store(0) t.MinWriteOffset = 0 return 1 } // Close it is called when the transfer is completed. // It logs the transfer info, updates the user quota (for uploads) // and executes any defined action. // If there is an error no action will be executed and, in atomic mode, // we try to delete the temporary file func (t *BaseTransfer) Close() error { defer t.Connection.RemoveTransfer(t) var err error numFiles := t.getUploadedFiles() metric.TransferCompleted(t.BytesSent.Load(), t.BytesReceived.Load(), t.transferType, t.ErrTransfer, vfs.IsSFTPFs(t.Fs)) if t.transferQuota.HasSizeLimits() { dataprovider.UpdateUserTransferQuota(&t.Connection.User, t.BytesReceived.Load(), //nolint:errcheck t.BytesSent.Load(), false) } if (t.File != nil || vfs.IsLocalOsFs(t.Fs)) && t.Connection.IsQuotaExceededError(t.ErrTransfer) { // if quota is exceeded we try to remove the partial file for uploads to local filesystem err = t.Fs.Remove(t.effectiveFsPath, false) if err == nil { t.BytesReceived.Store(0) t.MinWriteOffset = 0 } t.Connection.Log(logger.LevelWarn, "upload denied due to space limit, delete temporary file: %q, deletion error: %v", t.effectiveFsPath, err) } else if t.isAtomicUpload() { if t.ErrTransfer == nil || Config.UploadMode&UploadModeAtomicWithResume != 0 { _, _, err = t.Fs.Rename(t.effectiveFsPath, t.fsPath) t.Connection.Log(logger.LevelDebug, "atomic upload completed, rename: %q -> %q, error: %v", t.effectiveFsPath, t.fsPath, err) // the file must be removed if it is uploaded to a path outside the home dir and cannot be renamed t.checkUploadOutsideHomeDir(err) } else { err = t.Fs.Remove(t.effectiveFsPath, false) t.Connection.Log(logger.LevelWarn, "atomic upload completed with error: \"%v\", delete temporary file: %q, deletion error: %v", t.ErrTransfer, t.effectiveFsPath, err) if err == nil { t.BytesReceived.Store(0) t.MinWriteOffset = 0 } } } elapsed := time.Since(t.start).Nanoseconds() / 1000000 var uploadFileSize int64 if t.transferType == TransferDownload { logger.TransferLog(downloadLogSender, t.fsPath, elapsed, t.BytesSent.Load(), t.Connection.User.Username, t.Connection.ID, t.Connection.protocol, t.Connection.localAddr, t.Connection.remoteAddr, t.ftpMode, t.ErrTransfer) ExecuteActionNotification(t.Connection, operationDownload, t.fsPath, t.requestPath, "", "", "", //nolint:errcheck t.BytesSent.Load(), t.ErrTransfer, elapsed, t.metadata) } else { statSize, deletedFiles, errStat := t.getUploadFileSize() if errStat == nil { uploadFileSize = statSize } else { uploadFileSize = t.BytesReceived.Load() + t.MinWriteOffset if t.Fs.IsNotExist(errStat) { uploadFileSize = 0 numFiles-- } } numFiles -= deletedFiles t.Connection.Log(logger.LevelDebug, "upload file size %d, num files %d, deleted files %d, fs path %q", uploadFileSize, numFiles, deletedFiles, t.fsPath) numFiles, uploadFileSize = t.executeUploadHook(numFiles, uploadFileSize, elapsed) t.updateQuota(numFiles, uploadFileSize) t.updateTimes() logger.TransferLog(uploadLogSender, t.fsPath, elapsed, t.BytesReceived.Load(), t.Connection.User.Username, t.Connection.ID, t.Connection.protocol, t.Connection.localAddr, t.Connection.remoteAddr, t.ftpMode, t.ErrTransfer) } if t.ErrTransfer != nil { t.Connection.Log(logger.LevelError, "transfer error: %v, path: %q", t.ErrTransfer, t.fsPath) if err == nil { err = t.ErrTransfer } } t.updateTransferTimestamps(uploadFileSize, elapsed) return err } func (t *BaseTransfer) isAtomicUpload() bool { return t.transferType == TransferUpload && t.effectiveFsPath != t.fsPath } func (t *BaseTransfer) updateTransferTimestamps(uploadFileSize, elapsed int64) { if t.ErrTransfer != nil { return } if t.transferType == TransferUpload { if t.Connection.User.FirstUpload == 0 && !t.Connection.uploadDone.Load() { if err := dataprovider.UpdateUserTransferTimestamps(t.Connection.User.Username, true); err == nil { t.Connection.uploadDone.Store(true) ExecuteActionNotification(t.Connection, operationFirstUpload, t.fsPath, t.requestPath, "", //nolint:errcheck "", "", uploadFileSize, t.ErrTransfer, elapsed, t.metadata) } } return } if t.Connection.User.FirstDownload == 0 && !t.Connection.downloadDone.Load() && t.BytesSent.Load() > 0 { if err := dataprovider.UpdateUserTransferTimestamps(t.Connection.User.Username, false); err == nil { t.Connection.downloadDone.Store(true) ExecuteActionNotification(t.Connection, operationFirstDownload, t.fsPath, t.requestPath, "", //nolint:errcheck "", "", t.BytesSent.Load(), t.ErrTransfer, elapsed, t.metadata) } } } func (t *BaseTransfer) executeUploadHook(numFiles int, fileSize, elapsed int64) (int, int64) { err := ExecuteActionNotification(t.Connection, operationUpload, t.fsPath, t.requestPath, "", "", "", fileSize, t.ErrTransfer, elapsed, t.metadata) if err != nil { if t.ErrTransfer == nil { t.ErrTransfer = err } // try to remove the uploaded file err = t.Fs.Remove(t.fsPath, false) if err == nil { numFiles-- fileSize = 0 t.BytesReceived.Store(0) t.MinWriteOffset = 0 } else { t.Connection.Log(logger.LevelWarn, "unable to remove path %q after upload hook failure: %v", t.fsPath, err) } } return numFiles, fileSize } func (t *BaseTransfer) getUploadedFiles() int { numFiles := 0 if t.isNewFile { numFiles = 1 } return numFiles } func (t *BaseTransfer) updateTimes() { if !t.aTime.IsZero() && !t.mTime.IsZero() { err := t.Fs.Chtimes(t.fsPath, t.aTime, t.mTime, false) t.Connection.Log(logger.LevelDebug, "set times for file %q, atime: %v, mtime: %v, err: %v", t.fsPath, t.aTime, t.mTime, err) } } func (t *BaseTransfer) updateQuota(numFiles int, fileSize int64) bool { // Uploads on some filesystem (S3 and similar) are atomic, if there is an error nothing is uploaded if t.File == nil && t.ErrTransfer != nil && vfs.HasImplicitAtomicUploads(t.Fs) { return false } sizeDiff := fileSize - t.InitialSize if t.transferType == TransferUpload && (numFiles != 0 || sizeDiff != 0) { vfolder, err := t.Connection.User.GetVirtualFolderForPath(path.Dir(t.requestPath)) if err == nil { dataprovider.UpdateUserFolderQuota(&vfolder, &t.Connection.User, numFiles, sizeDiff, false) } else { dataprovider.UpdateUserQuota(&t.Connection.User, numFiles, sizeDiff, false) //nolint:errcheck } return true } return false } // HandleThrottle manage bandwidth throttling func (t *BaseTransfer) HandleThrottle() { var wantedBandwidth int64 var trasferredBytes int64 if t.transferType == TransferDownload { wantedBandwidth = t.Connection.User.DownloadBandwidth trasferredBytes = t.BytesSent.Load() } else { wantedBandwidth = t.Connection.User.UploadBandwidth trasferredBytes = t.BytesReceived.Load() } if wantedBandwidth > 0 { // real and wanted elapsed as milliseconds, bytes as kilobytes realElapsed := time.Since(t.start).Nanoseconds() / 1000000 // trasferredBytes / 1024 = KB/s, we multiply for 1000 to get milliseconds wantedElapsed := 1000 * (trasferredBytes / 1024) / wantedBandwidth if wantedElapsed > realElapsed { toSleep := time.Duration(wantedElapsed - realElapsed) time.Sleep(toSleep * time.Millisecond) } } }