sftpgo/common/common.go

// Package common defines code shared among file transfer packages and protocols
package common

import (
	"context"
	"errors"
	"fmt"
	"net"
	"net/http"
	"net/url"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/pires/go-proxyproto"

	"github.com/drakkan/sftpgo/dataprovider"
	"github.com/drakkan/sftpgo/httpclient"
	"github.com/drakkan/sftpgo/logger"
	"github.com/drakkan/sftpgo/metrics"
	"github.com/drakkan/sftpgo/utils"
	"github.com/drakkan/sftpgo/vfs"
)

// constants
const (
	logSender         = "common"
	uploadLogSender   = "Upload"
	downloadLogSender = "Download"
	renameLogSender   = "Rename"
	rmdirLogSender    = "Rmdir"
	mkdirLogSender    = "Mkdir"
	symlinkLogSender  = "Symlink"
	removeLogSender   = "Remove"
	chownLogSender    = "Chown"
	chmodLogSender    = "Chmod"
	chtimesLogSender  = "Chtimes"
	truncateLogSender = "Truncate"
	operationDownload = "download"
	operationUpload   = "upload"
	operationDelete   = "delete"
	// Pre-download action name
	OperationPreDownload = "pre-download"
	// Pre-upload action name
	OperationPreUpload = "pre-upload"
	operationPreDelete = "pre-delete"
	operationRename    = "rename"
	// SSH command action name
	OperationSSHCmd          = "ssh_cmd"
	chtimesFormat            = "2006-01-02T15:04:05" // YYYY-MM-DDTHH:MM:SS
	idleTimeoutCheckInterval = 3 * time.Minute
)

// Stat flags
const (
	StatAttrUIDGID = 1
	StatAttrPerms  = 2
	StatAttrTimes  = 4
	StatAttrSize   = 8
)

// Transfer types
const (
	TransferUpload = iota
	TransferDownload
)

// Supported protocols
const (
	ProtocolSFTP   = "SFTP"
	ProtocolSCP    = "SCP"
	ProtocolSSH    = "SSH"
	ProtocolFTP    = "FTP"
	ProtocolWebDAV = "DAV"
	ProtocolHTTP   = "HTTP"
)

// Upload modes
const (
	UploadModeStandard = iota
	UploadModeAtomic
	UploadModeAtomicWithResume
)

func init() {
	Connections.clients = clientsMap{
		clients: make(map[string]int),
	}
}

// errors definitions
var (
	ErrPermissionDenied     = errors.New("permission denied")
	ErrNotExist             = errors.New("no such file or directory")
	ErrOpUnsupported        = errors.New("operation unsupported")
	ErrGenericFailure       = errors.New("failure")
	ErrQuotaExceeded        = errors.New("denying write due to space limit")
	ErrSkipPermissionsCheck = errors.New("permission check skipped")
	ErrConnectionDenied     = errors.New("you are not allowed to connect")
	ErrNoBinding            = errors.New("no binding configured")
	ErrCrtRevoked           = errors.New("your certificate has been revoked")
	ErrNoCredentials        = errors.New("no credential provided")
	ErrInternalFailure      = errors.New("internal failure")
	errNoTransfer           = errors.New("requested transfer not found")
	errTransferMismatch     = errors.New("transfer mismatch")
)

var (
	// Config is the configuration for the supported protocols
	Config Configuration
	// Connections is the list of active connections
	Connections ActiveConnections
	// QuotaScans is the list of active quota scans
	QuotaScans            ActiveScans
	idleTimeoutTicker     *time.Ticker
	idleTimeoutTickerDone chan bool
	supportedProtocols    = []string{ProtocolSFTP, ProtocolSCP, ProtocolSSH, ProtocolFTP, ProtocolWebDAV, ProtocolHTTP}
	// the map key is the protocol, for each protocol we can have multiple rate limiters
	rateLimiters map[string][]*rateLimiter
)

// Initialize sets the common configuration
func Initialize(c Configuration) error {
	Config = c
	Config.idleLoginTimeout = 2 * time.Minute
	Config.idleTimeoutAsDuration = time.Duration(Config.IdleTimeout) * time.Minute
	if Config.IdleTimeout > 0 {
		startIdleTimeoutTicker(idleTimeoutCheckInterval)
	}
	Config.defender = nil
	if c.DefenderConfig.Enabled {
		defender, err := newInMemoryDefender(&c.DefenderConfig)
		if err != nil {
			return fmt.Errorf("defender initialization error: %v", err)
		}
		logger.Info(logSender, "", "defender initialized with config %+v", c.DefenderConfig)
		Config.defender = defender
	}
	rateLimiters = make(map[string][]*rateLimiter)
	for _, rlCfg := range c.RateLimitersConfig {
		if rlCfg.isEnabled() {
			if err := rlCfg.validate(); err != nil {
				return fmt.Errorf("rate limiters initialization error: %v", err)
			}
			rateLimiter := rlCfg.getLimiter()
			for _, protocol := range rlCfg.Protocols {
				rateLimiters[protocol] = append(rateLimiters[protocol], rateLimiter)
			}
		}
	}
	vfs.SetTempPath(c.TempPath)
	dataprovider.SetTempPath(c.TempPath)
	return nil
}

// LimitRate blocks until all the configured rate limiters
// allow one event to happen.
// It returns an error if the time to wait exceeds the max
// allowed delay
func LimitRate(protocol, ip string) (time.Duration, error) {
	for _, limiter := range rateLimiters[protocol] {
		if delay, err := limiter.Wait(ip); err != nil {
			logger.Debug(logSender, "", "protocol %v ip %v: %v", protocol, ip, err)
			return delay, err
		}
	}
	return 0, nil
}

// ReloadDefender reloads the defender's block and safe lists
func ReloadDefender() error {
	if Config.defender == nil {
		return nil
	}

	return Config.defender.Reload()
}

// IsBanned returns true if the specified IP address is banned
func IsBanned(ip string) bool {
	if Config.defender == nil {
		return false
	}

	return Config.defender.IsBanned(ip)
}

// GetDefenderBanTime returns the ban time for the given IP
// or nil if the IP is not banned or the defender is disabled
func GetDefenderBanTime(ip string) *time.Time {
	if Config.defender == nil {
		return nil
	}

	return Config.defender.GetBanTime(ip)
}

// Unban removes the specified IP address from the banned ones
func Unban(ip string) bool {
	if Config.defender == nil {
		return false
	}

	return Config.defender.Unban(ip)
}

// GetDefenderScore returns the score for the given IP
func GetDefenderScore(ip string) int {
	if Config.defender == nil {
		return 0
	}

	return Config.defender.GetScore(ip)
}

// AddDefenderEvent adds the specified defender event for the given IP
func AddDefenderEvent(ip string, event HostEvent) {
	if Config.defender == nil {
		return
	}

	Config.defender.AddEvent(ip, event)
}

// the ticker cannot be started/stopped from multiple goroutines
func startIdleTimeoutTicker(duration time.Duration) {
	stopIdleTimeoutTicker()
	idleTimeoutTicker = time.NewTicker(duration)
	idleTimeoutTickerDone = make(chan bool)
	go func() {
		for {
			select {
			case <-idleTimeoutTickerDone:
				return
			case <-idleTimeoutTicker.C:
				Connections.checkIdles()
			}
		}
	}()
}

func stopIdleTimeoutTicker() {
	if idleTimeoutTicker != nil {
		idleTimeoutTicker.Stop()
		idleTimeoutTickerDone <- true
		idleTimeoutTicker = nil
	}
}

// ActiveTransfer defines the interface for the current active transfers
type ActiveTransfer interface {
	GetID() uint64
	GetType() int
	GetSize() int64
	GetVirtualPath() string
	GetStartTime() time.Time
	SignalClose()
	Truncate(fsPath string, size int64) (int64, error)
	GetRealFsPath(fsPath string) string
}

// ActiveConnection defines the interface for the current active connections
type ActiveConnection interface {
	GetID() string
	GetUsername() string
	GetRemoteAddress() string
	GetClientVersion() string
	GetProtocol() string
	GetConnectionTime() time.Time
	GetLastActivity() time.Time
	GetCommand() string
	Disconnect() error
	AddTransfer(t ActiveTransfer)
	RemoveTransfer(t ActiveTransfer)
	GetTransfers() []ConnectionTransfer
	CloseFS() error
}

// StatAttributes defines the attributes for set stat commands
type StatAttributes struct {
	Mode  os.FileMode
	Atime time.Time
	Mtime time.Time
	UID   int
	GID   int
	Flags int
	Size  int64
}

// ConnectionTransfer defines the trasfer details to expose
type ConnectionTransfer struct {
	ID            uint64 `json:"-"`
	OperationType string `json:"operation_type"`
	StartTime     int64  `json:"start_time"`
	Size          int64  `json:"size"`
	VirtualPath   string `json:"path"`
}

func (t *ConnectionTransfer) getConnectionTransferAsString() string {
	result := ""
	switch t.OperationType {
	case operationUpload:
		result += "UL "
	case operationDownload:
		result += "DL "
	}
	result += fmt.Sprintf("%#v ", t.VirtualPath)
	if t.Size > 0 {
		elapsed := time.Since(utils.GetTimeFromMsecSinceEpoch(t.StartTime))
		speed := float64(t.Size) / float64(utils.GetTimeAsMsSinceEpoch(time.Now())-t.StartTime)
		result += fmt.Sprintf("Size: %#v Elapsed: %#v Speed: \"%.1f KB/s\"", utils.ByteCountIEC(t.Size),
			utils.GetDurationAsString(elapsed), speed)
	}
	return result
}

// Configuration defines configuration parameters common to all supported protocols
type Configuration struct {
	// Maximum idle timeout as minutes. If a client is idle for a time that exceeds this setting it will be disconnected.
	// 0 means disabled
	IdleTimeout int `json:"idle_timeout" mapstructure:"idle_timeout"`
	// UploadMode 0 means standard, the files are uploaded directly to the requested path.
	// 1 means atomic: the files are uploaded to a temporary path and renamed to the requested path
	// when the client ends the upload. Atomic mode avoid problems such as a web server that
	// serves partial files when the files are being uploaded.
	// In atomic mode if there is an upload error the temporary file is deleted and so the requested
	// upload path will not contain a partial file.
	// 2 means atomic with resume support: as atomic but if there is an upload error the temporary
	// file is renamed to the requested path and not deleted, this way a client can reconnect and resume
	// the upload.
	UploadMode int `json:"upload_mode" mapstructure:"upload_mode"`
	// Actions to execute for SFTP file operations and SSH commands
	Actions ProtocolActions `json:"actions" mapstructure:"actions"`
	// SetstatMode 0 means "normal mode": requests for changing permissions and owner/group are executed.
	// 1 means "ignore mode": requests for changing permissions and owner/group are silently ignored.
	// 2 means "ignore mode for cloud fs": requests for changing permissions and owner/group/time are
	// silently ignored for cloud based filesystem such as S3, GCS, Azure Blob
	SetstatMode int `json:"setstat_mode" mapstructure:"setstat_mode"`
	// TempPath defines the path for temporary files such as those used for atomic uploads or file pipes.
	// If you set this option you must make sure that the defined path exists, is accessible for writing
	// by the user running SFTPGo, and is on the same filesystem as the users home directories otherwise
	// the renaming for atomic uploads will become a copy and therefore may take a long time.
	// The temporary files are not namespaced. The default is generally fine. Leave empty for the default.
	TempPath string `json:"temp_path" mapstructure:"temp_path"`
	// Support for HAProxy PROXY protocol.
	// If you are running SFTPGo behind a proxy server such as HAProxy, AWS ELB or NGNIX, you can enable
	// the proxy protocol. It provides a convenient way to safely transport connection information
	// such as a client's address across multiple layers of NAT or TCP proxies to get the real
	// client IP address instead of the proxy IP. Both protocol versions 1 and 2 are supported.
	// - 0 means disabled
	// - 1 means proxy protocol enabled. Proxy header will be used and requests without proxy header will be accepted.
	// - 2 means proxy protocol required. Proxy header will be used and requests without proxy header will be rejected.
	// If the proxy protocol is enabled in SFTPGo then you have to enable the protocol in your proxy configuration too,
	// for example for HAProxy add "send-proxy" or "send-proxy-v2" to each server configuration line.
	ProxyProtocol int `json:"proxy_protocol" mapstructure:"proxy_protocol"`
	// List of IP addresses and IP ranges allowed to send the proxy header.
	// If proxy protocol is set to 1 and we receive a proxy header from an IP that is not in the list then the
	// connection will be accepted and the header will be ignored.
	// If proxy protocol is set to 2 and we receive a proxy header from an IP that is not in the list then the
	// connection will be rejected.
	ProxyAllowed []string `json:"proxy_allowed" mapstructure:"proxy_allowed"`
	// Absolute path to an external program or an HTTP URL to invoke as soon as SFTPGo starts.
	// If you define an HTTP URL it will be invoked using a `GET` request.
	// Please note that SFTPGo services may not yet be available when this hook is run.
	// Leave empty do disable.
	StartupHook string `json:"startup_hook" mapstructure:"startup_hook"`
	// Absolute path to an external program or an HTTP URL to invoke after a user connects
	// and before he tries to login. It allows you to reject the connection based on the source
	// ip address. Leave empty do disable.
	PostConnectHook string `json:"post_connect_hook" mapstructure:"post_connect_hook"`
	// Maximum number of concurrent client connections. 0 means unlimited
	MaxTotalConnections int `json:"max_total_connections" mapstructure:"max_total_connections"`
	// Maximum number of concurrent client connections from the same host (IP). 0 means unlimited
	MaxPerHostConnections int `json:"max_per_host_connections" mapstructure:"max_per_host_connections"`
	// Defender configuration
	DefenderConfig DefenderConfig `json:"defender" mapstructure:"defender"`
	// Rate limiter configurations
	RateLimitersConfig    []RateLimiterConfig `json:"rate_limiters" mapstructure:"rate_limiters"`
	idleTimeoutAsDuration time.Duration
	idleLoginTimeout      time.Duration
	defender              Defender
}

// IsAtomicUploadEnabled returns true if atomic upload is enabled
func (c *Configuration) IsAtomicUploadEnabled() bool {
	return c.UploadMode == UploadModeAtomic || c.UploadMode == UploadModeAtomicWithResume
}

// GetProxyListener returns a wrapper for the given listener that supports the
// HAProxy Proxy Protocol or nil if the proxy protocol is not configured
func (c *Configuration) GetProxyListener(listener net.Listener) (*proxyproto.Listener, error) {
	var proxyListener *proxyproto.Listener
	var err error
	if c.ProxyProtocol > 0 {
		var policyFunc func(upstream net.Addr) (proxyproto.Policy, error)
		if c.ProxyProtocol == 1 && len(c.ProxyAllowed) > 0 {
			policyFunc, err = proxyproto.LaxWhiteListPolicy(c.ProxyAllowed)
			if err != nil {
				return nil, err
			}
		}
		if c.ProxyProtocol == 2 {
			if len(c.ProxyAllowed) == 0 {
				policyFunc = func(upstream net.Addr) (proxyproto.Policy, error) {
					return proxyproto.REQUIRE, nil
				}
			} else {
				policyFunc, err = proxyproto.StrictWhiteListPolicy(c.ProxyAllowed)
				if err != nil {
					return nil, err
				}
			}
		}
		proxyListener = &proxyproto.Listener{
			Listener: listener,
			Policy:   policyFunc,
		}
	}
	return proxyListener, nil
}

// ExecuteStartupHook runs the startup hook if defined
func (c *Configuration) ExecuteStartupHook() error {
	if c.StartupHook == "" {
		return nil
	}
	if strings.HasPrefix(c.StartupHook, "http") {
		var url *url.URL
		url, err := url.Parse(c.StartupHook)
		if err != nil {
			logger.Warn(logSender, "", "Invalid startup hook %#v: %v", c.StartupHook, err)
			return err
		}
		startTime := time.Now()
		resp, err := httpclient.RetryableGet(url.String())
		if err != nil {
			logger.Warn(logSender, "", "Error executing startup hook: %v", err)
			return err
		}
		defer resp.Body.Close()
		logger.Debug(logSender, "", "Startup hook executed, elapsed: %v, response code: %v", time.Since(startTime), resp.StatusCode)
		return nil
	}
	if !filepath.IsAbs(c.StartupHook) {
		err := fmt.Errorf("invalid startup hook %#v", c.StartupHook)
		logger.Warn(logSender, "", "Invalid startup hook %#v", c.StartupHook)
		return err
	}
	startTime := time.Now()
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	cmd := exec.CommandContext(ctx, c.StartupHook)
	err := cmd.Run()
	logger.Debug(logSender, "", "Startup hook executed, elapsed: %v, error: %v", time.Since(startTime), err)
	return nil
}

// ExecutePostConnectHook executes the post connect hook if defined
func (c *Configuration) ExecutePostConnectHook(ipAddr, protocol string) error {
	if c.PostConnectHook == "" {
		return nil
	}
	if strings.HasPrefix(c.PostConnectHook, "http") {
		var url *url.URL
		url, err := url.Parse(c.PostConnectHook)
		if err != nil {
			logger.Warn(protocol, "", "Login from ip %#v denied, invalid post connect hook %#v: %v",
				ipAddr, c.PostConnectHook, err)
			return err
		}
		q := url.Query()
		q.Add("ip", ipAddr)
		q.Add("protocol", protocol)
		url.RawQuery = q.Encode()

		resp, err := httpclient.RetryableGet(url.String())
		if err != nil {
			logger.Warn(protocol, "", "Login from ip %#v denied, error executing post connect hook: %v", ipAddr, err)
			return err
		}
		defer resp.Body.Close()
		if resp.StatusCode != http.StatusOK {
			logger.Warn(protocol, "", "Login from ip %#v denied, post connect hook response code: %v", ipAddr, resp.StatusCode)
			return errUnexpectedHTTResponse
		}
		return nil
	}
	if !filepath.IsAbs(c.PostConnectHook) {
		err := fmt.Errorf("invalid post connect hook %#v", c.PostConnectHook)
		logger.Warn(protocol, "", "Login from ip %#v denied: %v", ipAddr, err)
		return err
	}
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
	defer cancel()
	cmd := exec.CommandContext(ctx, c.PostConnectHook)
	cmd.Env = append(os.Environ(),
		fmt.Sprintf("SFTPGO_CONNECTION_IP=%v", ipAddr),
		fmt.Sprintf("SFTPGO_CONNECTION_PROTOCOL=%v", protocol))
	err := cmd.Run()
	if err != nil {
		logger.Warn(protocol, "", "Login from ip %#v denied, connect hook error: %v", ipAddr, err)
	}
	return err
}

// SSHConnection defines an ssh connection.
// Each SSH connection can open several channels for SFTP or SSH commands
type SSHConnection struct {
	id           string
	conn         net.Conn
	lastActivity int64
}

// NewSSHConnection returns a new SSHConnection
func NewSSHConnection(id string, conn net.Conn) *SSHConnection {
	return &SSHConnection{
		id:           id,
		conn:         conn,
		lastActivity: time.Now().UnixNano(),
	}
}

// GetID returns the ID for this SSHConnection
func (c *SSHConnection) GetID() string {
	return c.id
}

// UpdateLastActivity updates last activity for this connection
func (c *SSHConnection) UpdateLastActivity() {
	atomic.StoreInt64(&c.lastActivity, time.Now().UnixNano())
}

// GetLastActivity returns the last connection activity
func (c *SSHConnection) GetLastActivity() time.Time {
	return time.Unix(0, atomic.LoadInt64(&c.lastActivity))
}

// Close closes the underlying network connection
func (c *SSHConnection) Close() error {
	return c.conn.Close()
}

// ActiveConnections holds the currect active connections with the associated transfers
type ActiveConnections struct {
	// clients contains both authenticated and estabilished connections and the ones waiting
	// for authentication
	clients clientsMap
	sync.RWMutex
	connections    []ActiveConnection
	sshConnections []*SSHConnection
}

// GetActiveSessions returns the number of active sessions for the given username.
// We return the open sessions for any protocol
func (conns *ActiveConnections) GetActiveSessions(username string) int {
	conns.RLock()
	defer conns.RUnlock()

	numSessions := 0
	for _, c := range conns.connections {
		if c.GetUsername() == username {
			numSessions++
		}
	}
	return numSessions
}

// Add adds a new connection to the active ones
func (conns *ActiveConnections) Add(c ActiveConnection) {
	conns.Lock()
	defer conns.Unlock()

	conns.connections = append(conns.connections, c)
	metrics.UpdateActiveConnectionsSize(len(conns.connections))
	logger.Debug(c.GetProtocol(), c.GetID(), "connection added, num open connections: %v", len(conns.connections))
}

// Swap replaces an existing connection with the given one.
// This method is useful if you have to change some connection details
// for example for FTP is used to update the connection once the user
// authenticates
func (conns *ActiveConnections) Swap(c ActiveConnection) error {
	conns.Lock()
	defer conns.Unlock()

	for idx, conn := range conns.connections {
		if conn.GetID() == c.GetID() {
			conn = nil
			conns.connections[idx] = c
			return nil
		}
	}
	return errors.New("connection to swap not found")
}

// Remove removes a connection from the active ones
func (conns *ActiveConnections) Remove(connectionID string) {
	conns.Lock()
	defer conns.Unlock()

	for idx, conn := range conns.connections {
		if conn.GetID() == connectionID {
			err := conn.CloseFS()
			lastIdx := len(conns.connections) - 1
			conns.connections[idx] = conns.connections[lastIdx]
			conns.connections[lastIdx] = nil
			conns.connections = conns.connections[:lastIdx]
			metrics.UpdateActiveConnectionsSize(lastIdx)
			logger.Debug(conn.GetProtocol(), conn.GetID(), "connection removed, close fs error: %v, num open connections: %v",
				err, lastIdx)
			return
		}
	}
	logger.Warn(logSender, "", "connection id %#v to remove not found!", connectionID)
}

// Close closes an active connection.
// It returns true on success
func (conns *ActiveConnections) Close(connectionID string) bool {
	conns.RLock()
	result := false

	for _, c := range conns.connections {
		if c.GetID() == connectionID {
			defer func(conn ActiveConnection) {
				err := conn.Disconnect()
				logger.Debug(conn.GetProtocol(), conn.GetID(), "close connection requested, close err: %v", err)
			}(c)
			result = true
			break
		}
	}

	conns.RUnlock()
	return result
}

// AddSSHConnection adds a new ssh connection to the active ones
func (conns *ActiveConnections) AddSSHConnection(c *SSHConnection) {
	conns.Lock()
	defer conns.Unlock()

	conns.sshConnections = append(conns.sshConnections, c)
	logger.Debug(logSender, c.GetID(), "ssh connection added, num open connections: %v", len(conns.sshConnections))
}

// RemoveSSHConnection removes a connection from the active ones
func (conns *ActiveConnections) RemoveSSHConnection(connectionID string) {
	conns.Lock()
	defer conns.Unlock()

	for idx, conn := range conns.sshConnections {
		if conn.GetID() == connectionID {
			lastIdx := len(conns.sshConnections) - 1
			conns.sshConnections[idx] = conns.sshConnections[lastIdx]
			conns.sshConnections[lastIdx] = nil
			conns.sshConnections = conns.sshConnections[:lastIdx]
			logger.Debug(logSender, conn.GetID(), "ssh connection removed, num open ssh connections: %v", lastIdx)
			return
		}
	}
	logger.Warn(logSender, "", "ssh connection to remove with id %#v not found!", connectionID)
}

func (conns *ActiveConnections) checkIdles() {
	conns.RLock()

	for _, sshConn := range conns.sshConnections {
		idleTime := time.Since(sshConn.GetLastActivity())
		if idleTime > Config.idleTimeoutAsDuration {
			// we close the an ssh connection if it has no active connections associated
			idToMatch := fmt.Sprintf("_%v_", sshConn.GetID())
			toClose := true
			for _, conn := range conns.connections {
				if strings.Contains(conn.GetID(), idToMatch) {
					toClose = false
					break
				}
			}
			if toClose {
				defer func(c *SSHConnection) {
					err := c.Close()
					logger.Debug(logSender, c.GetID(), "close idle SSH connection, idle time: %v, close err: %v",
						time.Since(c.GetLastActivity()), err)
				}(sshConn)
			}
		}
	}

	for _, c := range conns.connections {
		idleTime := time.Since(c.GetLastActivity())
		isUnauthenticatedFTPUser := (c.GetProtocol() == ProtocolFTP && c.GetUsername() == "")

		if idleTime > Config.idleTimeoutAsDuration || (isUnauthenticatedFTPUser && idleTime > Config.idleLoginTimeout) {
			defer func(conn ActiveConnection, isFTPNoAuth bool) {
				err := conn.Disconnect()
				logger.Debug(conn.GetProtocol(), conn.GetID(), "close idle connection, idle time: %v, username: %#v close err: %v",
					time.Since(conn.GetLastActivity()), conn.GetUsername(), err)
				if isFTPNoAuth {
					ip := utils.GetIPFromRemoteAddress(c.GetRemoteAddress())
					logger.ConnectionFailedLog("", ip, dataprovider.LoginMethodNoAuthTryed, c.GetProtocol(), "client idle")
					metrics.AddNoAuthTryed()
					AddDefenderEvent(ip, HostEventNoLoginTried)
					dataprovider.ExecutePostLoginHook(&dataprovider.User{}, dataprovider.LoginMethodNoAuthTryed, ip, c.GetProtocol(),
						dataprovider.ErrNoAuthTryed)
				}
			}(c, isUnauthenticatedFTPUser)
		}
	}

	conns.RUnlock()
}

// AddClientConnection stores a new client connection
func (conns *ActiveConnections) AddClientConnection(ipAddr string) {
	conns.clients.add(ipAddr)
}

// RemoveClientConnection removes a disconnected client from the tracked ones
func (conns *ActiveConnections) RemoveClientConnection(ipAddr string) {
	conns.clients.remove(ipAddr)
}

// GetClientConnections returns the total number of client connections
func (conns *ActiveConnections) GetClientConnections() int32 {
	return conns.clients.getTotal()
}

// IsNewConnectionAllowed returns false if the maximum number of concurrent allowed connections is exceeded
func (conns *ActiveConnections) IsNewConnectionAllowed(ipAddr string) bool {
	if Config.MaxTotalConnections == 0 && Config.MaxPerHostConnections == 0 {
		return true
	}

	if Config.MaxPerHostConnections > 0 {
		if total := conns.clients.getTotalFrom(ipAddr); total > Config.MaxPerHostConnections {
			logger.Debug(logSender, "", "active connections from %v %v/%v", ipAddr, total, Config.MaxPerHostConnections)
			AddDefenderEvent(ipAddr, HostEventLimitExceeded)
			return false
		}
	}

	if Config.MaxTotalConnections > 0 {
		if total := conns.clients.getTotal(); total > int32(Config.MaxTotalConnections) {
			logger.Debug(logSender, "", "active client connections %v/%v", total, Config.MaxTotalConnections)
			return false
		}

		// on a single SFTP connection we could have multiple SFTP channels or commands
		// so we check the estabilished connections too

		conns.RLock()
		defer conns.RUnlock()

		return len(conns.connections) < Config.MaxTotalConnections
	}

	return true
}

// GetStats returns stats for active connections
func (conns *ActiveConnections) GetStats() []*ConnectionStatus {
	conns.RLock()
	defer conns.RUnlock()

	stats := make([]*ConnectionStatus, 0, len(conns.connections))
	for _, c := range conns.connections {
		stat := &ConnectionStatus{
			Username:       c.GetUsername(),
			ConnectionID:   c.GetID(),
			ClientVersion:  c.GetClientVersion(),
			RemoteAddress:  c.GetRemoteAddress(),
			ConnectionTime: utils.GetTimeAsMsSinceEpoch(c.GetConnectionTime()),
			LastActivity:   utils.GetTimeAsMsSinceEpoch(c.GetLastActivity()),
			Protocol:       c.GetProtocol(),
			Command:        c.GetCommand(),
			Transfers:      c.GetTransfers(),
		}
		stats = append(stats, stat)
	}
	return stats
}

// ConnectionStatus returns the status for an active connection
type ConnectionStatus struct {
	// Logged in username
	Username string `json:"username"`
	// Unique identifier for the connection
	ConnectionID string `json:"connection_id"`
	// client's version string
	ClientVersion string `json:"client_version,omitempty"`
	// Remote address for this connection
	RemoteAddress string `json:"remote_address"`
	// Connection time as unix timestamp in milliseconds
	ConnectionTime int64 `json:"connection_time"`
	// Last activity as unix timestamp in milliseconds
	LastActivity int64 `json:"last_activity"`
	// Protocol for this connection
	Protocol string `json:"protocol"`
	// active uploads/downloads
	Transfers []ConnectionTransfer `json:"active_transfers,omitempty"`
	// SSH command or WebDAV method
	Command string `json:"command,omitempty"`
}

// GetConnectionDuration returns the connection duration as string
func (c *ConnectionStatus) GetConnectionDuration() string {
	elapsed := time.Since(utils.GetTimeFromMsecSinceEpoch(c.ConnectionTime))
	return utils.GetDurationAsString(elapsed)
}

// GetConnectionInfo returns connection info.
// Protocol,Client Version and RemoteAddress are returned.
func (c *ConnectionStatus) GetConnectionInfo() string {
	var result strings.Builder

	result.WriteString(fmt.Sprintf("%v. Client: %#v From: %#v", c.Protocol, c.ClientVersion, c.RemoteAddress))

	if c.Command == "" {
		return result.String()
	}

	switch c.Protocol {
	case ProtocolSSH, ProtocolFTP:
		result.WriteString(fmt.Sprintf(". Command: %#v", c.Command))
	case ProtocolWebDAV:
		result.WriteString(fmt.Sprintf(". Method: %#v", c.Command))
	}

	return result.String()
}

// GetTransfersAsString returns the active transfers as string
func (c *ConnectionStatus) GetTransfersAsString() string {
	result := ""
	for _, t := range c.Transfers {
		if result != "" {
			result += ". "
		}
		result += t.getConnectionTransferAsString()
	}
	return result
}

// ActiveQuotaScan defines an active quota scan for a user home dir
type ActiveQuotaScan struct {
	// Username to which the quota scan refers
	Username string `json:"username"`
	// quota scan start time as unix timestamp in milliseconds
	StartTime int64 `json:"start_time"`
}

// ActiveVirtualFolderQuotaScan defines an active quota scan for a virtual folder
type ActiveVirtualFolderQuotaScan struct {
	// folder name to which the quota scan refers
	Name string `json:"name"`
	// quota scan start time as unix timestamp in milliseconds
	StartTime int64 `json:"start_time"`
}

// ActiveScans holds the active quota scans
type ActiveScans struct {
	sync.RWMutex
	UserHomeScans []ActiveQuotaScan
	FolderScans   []ActiveVirtualFolderQuotaScan
}

// GetUsersQuotaScans returns the active quota scans for users home directories
func (s *ActiveScans) GetUsersQuotaScans() []ActiveQuotaScan {
	s.RLock()
	defer s.RUnlock()

	scans := make([]ActiveQuotaScan, len(s.UserHomeScans))
	copy(scans, s.UserHomeScans)
	return scans
}

// AddUserQuotaScan adds a user to the ones with active quota scans.
// Returns false if the user has a quota scan already running
func (s *ActiveScans) AddUserQuotaScan(username string) bool {
	s.Lock()
	defer s.Unlock()

	for _, scan := range s.UserHomeScans {
		if scan.Username == username {
			return false
		}
	}
	s.UserHomeScans = append(s.UserHomeScans, ActiveQuotaScan{
		Username:  username,
		StartTime: utils.GetTimeAsMsSinceEpoch(time.Now()),
	})
	return true
}

// RemoveUserQuotaScan removes a user from the ones with active quota scans.
// Returns false if the user has no active quota scans
func (s *ActiveScans) RemoveUserQuotaScan(username string) bool {
	s.Lock()
	defer s.Unlock()

	indexToRemove := -1
	for i, scan := range s.UserHomeScans {
		if scan.Username == username {
			indexToRemove = i
			break
		}
	}
	if indexToRemove >= 0 {
		s.UserHomeScans[indexToRemove] = s.UserHomeScans[len(s.UserHomeScans)-1]
		s.UserHomeScans = s.UserHomeScans[:len(s.UserHomeScans)-1]
		return true
	}
	return false
}

// GetVFoldersQuotaScans returns the active quota scans for virtual folders
func (s *ActiveScans) GetVFoldersQuotaScans() []ActiveVirtualFolderQuotaScan {
	s.RLock()
	defer s.RUnlock()
	scans := make([]ActiveVirtualFolderQuotaScan, len(s.FolderScans))
	copy(scans, s.FolderScans)
	return scans
}

// AddVFolderQuotaScan adds a virtual folder to the ones with active quota scans.
// Returns false if the folder has a quota scan already running
func (s *ActiveScans) AddVFolderQuotaScan(folderName string) bool {
	s.Lock()
	defer s.Unlock()

	for _, scan := range s.FolderScans {
		if scan.Name == folderName {
			return false
		}
	}
	s.FolderScans = append(s.FolderScans, ActiveVirtualFolderQuotaScan{
		Name:      folderName,
		StartTime: utils.GetTimeAsMsSinceEpoch(time.Now()),
	})
	return true
}

// RemoveVFolderQuotaScan removes a folder from the ones with active quota scans.
// Returns false if the folder has no active quota scans
func (s *ActiveScans) RemoveVFolderQuotaScan(folderName string) bool {
	s.Lock()
	defer s.Unlock()

	indexToRemove := -1
	for i, scan := range s.FolderScans {
		if scan.Name == folderName {
			indexToRemove = i
			break
		}
	}
	if indexToRemove >= 0 {
		s.FolderScans[indexToRemove] = s.FolderScans[len(s.FolderScans)-1]
		s.FolderScans = s.FolderScans[:len(s.FolderScans)-1]
		return true
	}
	return false
}