mirror of
https://github.com/drakkan/sftpgo.git
synced 2024-11-22 07:30:25 +00:00
302 lines
8.3 KiB
Go
302 lines
8.3 KiB
Go
package httpd
|
|
|
|
import (
|
|
"io"
|
|
"net/http"
|
|
"os"
|
|
"path"
|
|
"strings"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/drakkan/sftpgo/v2/common"
|
|
"github.com/drakkan/sftpgo/v2/dataprovider"
|
|
"github.com/drakkan/sftpgo/v2/logger"
|
|
"github.com/drakkan/sftpgo/v2/util"
|
|
"github.com/drakkan/sftpgo/v2/vfs"
|
|
)
|
|
|
|
// Connection details for a HTTP connection used to inteact with an SFTPGo filesystem
|
|
type Connection struct {
|
|
*common.BaseConnection
|
|
request *http.Request
|
|
}
|
|
|
|
// GetClientVersion returns the connected client's version.
|
|
func (c *Connection) GetClientVersion() string {
|
|
if c.request != nil {
|
|
return c.request.UserAgent()
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// GetLocalAddress returns local connection address
|
|
func (c *Connection) GetLocalAddress() string {
|
|
return util.GetHTTPLocalAddress(c.request)
|
|
}
|
|
|
|
// GetRemoteAddress returns the connected client's address
|
|
func (c *Connection) GetRemoteAddress() string {
|
|
if c.request != nil {
|
|
return c.request.RemoteAddr
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// Disconnect closes the active transfer
|
|
func (c *Connection) Disconnect() (err error) {
|
|
return c.SignalTransfersAbort()
|
|
}
|
|
|
|
// GetCommand returns the request method
|
|
func (c *Connection) GetCommand() string {
|
|
if c.request != nil {
|
|
return strings.ToUpper(c.request.Method)
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// Stat returns a FileInfo describing the named file/directory, or an error,
|
|
// if any happens
|
|
func (c *Connection) Stat(name string, mode int) (os.FileInfo, error) {
|
|
c.UpdateLastActivity()
|
|
|
|
name = util.CleanPath(name)
|
|
if !c.User.HasPerm(dataprovider.PermListItems, path.Dir(name)) {
|
|
return nil, c.GetPermissionDeniedError()
|
|
}
|
|
|
|
fi, err := c.DoStat(name, mode)
|
|
if err != nil {
|
|
c.Log(logger.LevelDebug, "error running stat on path %#v: %+v", name, err)
|
|
return nil, err
|
|
}
|
|
return fi, err
|
|
}
|
|
|
|
// ReadDir returns a list of directory entries
|
|
func (c *Connection) ReadDir(name string) ([]os.FileInfo, error) {
|
|
c.UpdateLastActivity()
|
|
|
|
name = util.CleanPath(name)
|
|
return c.ListDir(name)
|
|
}
|
|
|
|
func (c *Connection) getFileReader(name string, offset int64, method string) (io.ReadCloser, error) {
|
|
c.UpdateLastActivity()
|
|
|
|
name = util.CleanPath(name)
|
|
if !c.User.HasPerm(dataprovider.PermDownload, path.Dir(name)) {
|
|
return nil, c.GetPermissionDeniedError()
|
|
}
|
|
|
|
if !c.User.IsFileAllowed(name) {
|
|
c.Log(logger.LevelWarn, "reading file %#v is not allowed", name)
|
|
return nil, c.GetPermissionDeniedError()
|
|
}
|
|
|
|
fs, p, err := c.GetFsAndResolvedPath(name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if method != http.MethodHead {
|
|
if err := common.ExecutePreAction(c.BaseConnection, common.OperationPreDownload, p, name, 0, 0); err != nil {
|
|
c.Log(logger.LevelDebug, "download for file %#v denied by pre action: %v", name, err)
|
|
return nil, c.GetPermissionDeniedError()
|
|
}
|
|
}
|
|
|
|
file, r, cancelFn, err := fs.Open(p, offset)
|
|
if err != nil {
|
|
c.Log(logger.LevelWarn, "could not open file %#v for reading: %+v", p, err)
|
|
return nil, c.GetFsError(fs, err)
|
|
}
|
|
|
|
baseTransfer := common.NewBaseTransfer(file, c.BaseConnection, cancelFn, p, p, name, common.TransferDownload,
|
|
0, 0, 0, false, fs)
|
|
return newHTTPDFile(baseTransfer, nil, r), nil
|
|
}
|
|
|
|
func (c *Connection) getFileWriter(name string) (io.WriteCloser, error) {
|
|
c.UpdateLastActivity()
|
|
|
|
if !c.User.IsFileAllowed(name) {
|
|
c.Log(logger.LevelWarn, "writing file %#v is not allowed", name)
|
|
return nil, c.GetPermissionDeniedError()
|
|
}
|
|
|
|
fs, p, err := c.GetFsAndResolvedPath(name)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
filePath := p
|
|
if common.Config.IsAtomicUploadEnabled() && fs.IsAtomicUploadSupported() {
|
|
filePath = fs.GetAtomicUploadPath(p)
|
|
}
|
|
|
|
stat, statErr := fs.Lstat(p)
|
|
if (statErr == nil && stat.Mode()&os.ModeSymlink != 0) || fs.IsNotExist(statErr) {
|
|
if !c.User.HasPerm(dataprovider.PermUpload, path.Dir(name)) {
|
|
return nil, c.GetPermissionDeniedError()
|
|
}
|
|
return c.handleUploadFile(fs, p, filePath, name, true, 0)
|
|
}
|
|
|
|
if statErr != nil {
|
|
c.Log(logger.LevelError, "error performing file stat %#v: %+v", p, statErr)
|
|
return nil, c.GetFsError(fs, statErr)
|
|
}
|
|
|
|
// This happen if we upload a file that has the same name of an existing directory
|
|
if stat.IsDir() {
|
|
c.Log(logger.LevelWarn, "attempted to open a directory for writing to: %#v", p)
|
|
return nil, c.GetOpUnsupportedError()
|
|
}
|
|
|
|
if !c.User.HasPerm(dataprovider.PermOverwrite, path.Dir(name)) {
|
|
return nil, c.GetPermissionDeniedError()
|
|
}
|
|
|
|
if common.Config.IsAtomicUploadEnabled() && fs.IsAtomicUploadSupported() {
|
|
err = fs.Rename(p, filePath)
|
|
if err != nil {
|
|
c.Log(logger.LevelWarn, "error renaming existing file for atomic upload, source: %#v, dest: %#v, err: %+v",
|
|
p, filePath, err)
|
|
return nil, c.GetFsError(fs, err)
|
|
}
|
|
}
|
|
|
|
return c.handleUploadFile(fs, p, filePath, name, false, stat.Size())
|
|
}
|
|
|
|
func (c *Connection) handleUploadFile(fs vfs.Fs, resolvedPath, filePath, requestPath string, isNewFile bool, fileSize int64) (io.WriteCloser, error) {
|
|
quotaResult := c.HasSpace(isNewFile, false, requestPath)
|
|
if !quotaResult.HasSpace {
|
|
c.Log(logger.LevelInfo, "denying file write due to quota limits")
|
|
return nil, common.ErrQuotaExceeded
|
|
}
|
|
err := common.ExecutePreAction(c.BaseConnection, common.OperationPreUpload, resolvedPath, requestPath, fileSize, os.O_TRUNC)
|
|
if err != nil {
|
|
c.Log(logger.LevelDebug, "upload for file %#v denied by pre action: %v", requestPath, err)
|
|
return nil, c.GetPermissionDeniedError()
|
|
}
|
|
|
|
maxWriteSize, _ := c.GetMaxWriteSize(quotaResult, false, fileSize, fs.IsUploadResumeSupported())
|
|
|
|
file, w, cancelFn, err := fs.Create(filePath, 0)
|
|
if err != nil {
|
|
c.Log(logger.LevelWarn, "error opening existing file, source: %#v, err: %+v", filePath, err)
|
|
return nil, c.GetFsError(fs, err)
|
|
}
|
|
|
|
initialSize := int64(0)
|
|
if !isNewFile {
|
|
if vfs.IsLocalOrSFTPFs(fs) {
|
|
vfolder, err := c.User.GetVirtualFolderForPath(path.Dir(requestPath))
|
|
if err == nil {
|
|
dataprovider.UpdateVirtualFolderQuota(&vfolder.BaseVirtualFolder, 0, -fileSize, false) //nolint:errcheck
|
|
if vfolder.IsIncludedInUserQuota() {
|
|
dataprovider.UpdateUserQuota(&c.User, 0, -fileSize, false) //nolint:errcheck
|
|
}
|
|
} else {
|
|
dataprovider.UpdateUserQuota(&c.User, 0, -fileSize, false) //nolint:errcheck
|
|
}
|
|
} else {
|
|
initialSize = fileSize
|
|
}
|
|
if maxWriteSize > 0 {
|
|
maxWriteSize += fileSize
|
|
}
|
|
}
|
|
|
|
vfs.SetPathPermissions(fs, filePath, c.User.GetUID(), c.User.GetGID())
|
|
|
|
baseTransfer := common.NewBaseTransfer(file, c.BaseConnection, cancelFn, resolvedPath, filePath, requestPath,
|
|
common.TransferUpload, 0, initialSize, maxWriteSize, isNewFile, fs)
|
|
return newHTTPDFile(baseTransfer, w, nil), nil
|
|
}
|
|
|
|
func newThrottledReader(r io.ReadCloser, limit int64, conn *Connection) *throttledReader {
|
|
t := &throttledReader{
|
|
bytesRead: 0,
|
|
id: conn.GetTransferID(),
|
|
limit: limit,
|
|
r: r,
|
|
abortTransfer: 0,
|
|
start: time.Now(),
|
|
conn: conn,
|
|
}
|
|
conn.AddTransfer(t)
|
|
return t
|
|
}
|
|
|
|
type throttledReader struct {
|
|
bytesRead int64
|
|
id uint64
|
|
limit int64
|
|
r io.ReadCloser
|
|
abortTransfer int32
|
|
start time.Time
|
|
conn *Connection
|
|
}
|
|
|
|
func (t *throttledReader) GetID() uint64 {
|
|
return t.id
|
|
}
|
|
|
|
func (t *throttledReader) GetType() int {
|
|
return common.TransferUpload
|
|
}
|
|
|
|
func (t *throttledReader) GetSize() int64 {
|
|
return atomic.LoadInt64(&t.bytesRead)
|
|
}
|
|
|
|
func (t *throttledReader) GetVirtualPath() string {
|
|
return "**reading request body**"
|
|
}
|
|
|
|
func (t *throttledReader) GetStartTime() time.Time {
|
|
return t.start
|
|
}
|
|
|
|
func (t *throttledReader) SignalClose() {
|
|
atomic.StoreInt32(&(t.abortTransfer), 1)
|
|
}
|
|
|
|
func (t *throttledReader) Truncate(fsPath string, size int64) (int64, error) {
|
|
return 0, vfs.ErrVfsUnsupported
|
|
}
|
|
|
|
func (t *throttledReader) GetRealFsPath(fsPath string) string {
|
|
return ""
|
|
}
|
|
|
|
func (t *throttledReader) SetTimes(fsPath string, atime time.Time, mtime time.Time) bool {
|
|
return false
|
|
}
|
|
|
|
func (t *throttledReader) Read(p []byte) (n int, err error) {
|
|
if atomic.LoadInt32(&t.abortTransfer) == 1 {
|
|
return 0, errTransferAborted
|
|
}
|
|
|
|
t.conn.UpdateLastActivity()
|
|
n, err = t.r.Read(p)
|
|
if t.limit > 0 {
|
|
atomic.AddInt64(&t.bytesRead, int64(n))
|
|
trasferredBytes := atomic.LoadInt64(&t.bytesRead)
|
|
elapsed := time.Since(t.start).Nanoseconds() / 1000000
|
|
wantedElapsed := 1000 * (trasferredBytes / 1024) / t.limit
|
|
if wantedElapsed > elapsed {
|
|
toSleep := time.Duration(wantedElapsed - elapsed)
|
|
time.Sleep(toSleep * time.Millisecond)
|
|
}
|
|
}
|
|
return
|
|
}
|
|
|
|
func (t *throttledReader) Close() error {
|
|
return t.r.Close()
|
|
}
|