common.go 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907
  1. // Package common defines code shared among file transfer packages and protocols
  2. package common
  3. import (
  4. "context"
  5. "errors"
  6. "fmt"
  7. "net"
  8. "net/http"
  9. "net/url"
  10. "os"
  11. "os/exec"
  12. "path/filepath"
  13. "strings"
  14. "sync"
  15. "sync/atomic"
  16. "time"
  17. "github.com/pires/go-proxyproto"
  18. "github.com/drakkan/sftpgo/dataprovider"
  19. "github.com/drakkan/sftpgo/httpclient"
  20. "github.com/drakkan/sftpgo/logger"
  21. "github.com/drakkan/sftpgo/metrics"
  22. "github.com/drakkan/sftpgo/utils"
  23. )
  24. // constants
  25. const (
  26. logSender = "common"
  27. uploadLogSender = "Upload"
  28. downloadLogSender = "Download"
  29. renameLogSender = "Rename"
  30. rmdirLogSender = "Rmdir"
  31. mkdirLogSender = "Mkdir"
  32. symlinkLogSender = "Symlink"
  33. removeLogSender = "Remove"
  34. chownLogSender = "Chown"
  35. chmodLogSender = "Chmod"
  36. chtimesLogSender = "Chtimes"
  37. truncateLogSender = "Truncate"
  38. operationDownload = "download"
  39. operationUpload = "upload"
  40. operationDelete = "delete"
  41. operationPreDelete = "pre-delete"
  42. operationRename = "rename"
  43. operationSSHCmd = "ssh_cmd"
  44. chtimesFormat = "2006-01-02T15:04:05" // YYYY-MM-DDTHH:MM:SS
  45. idleTimeoutCheckInterval = 3 * time.Minute
  46. )
  47. // Stat flags
  48. const (
  49. StatAttrUIDGID = 1
  50. StatAttrPerms = 2
  51. StatAttrTimes = 4
  52. StatAttrSize = 8
  53. )
  54. // Transfer types
  55. const (
  56. TransferUpload = iota
  57. TransferDownload
  58. )
  59. // Supported protocols
  60. const (
  61. ProtocolSFTP = "SFTP"
  62. ProtocolSCP = "SCP"
  63. ProtocolSSH = "SSH"
  64. ProtocolFTP = "FTP"
  65. ProtocolWebDAV = "DAV"
  66. ProtocolHTTP = "HTTP"
  67. )
  68. // Upload modes
  69. const (
  70. UploadModeStandard = iota
  71. UploadModeAtomic
  72. UploadModeAtomicWithResume
  73. )
  74. // errors definitions
  75. var (
  76. ErrPermissionDenied = errors.New("permission denied")
  77. ErrNotExist = errors.New("no such file or directory")
  78. ErrOpUnsupported = errors.New("operation unsupported")
  79. ErrGenericFailure = errors.New("failure")
  80. ErrQuotaExceeded = errors.New("denying write due to space limit")
  81. ErrSkipPermissionsCheck = errors.New("permission check skipped")
  82. ErrConnectionDenied = errors.New("you are not allowed to connect")
  83. ErrNoBinding = errors.New("no binding configured")
  84. ErrCrtRevoked = errors.New("your certificate has been revoked")
  85. errNoTransfer = errors.New("requested transfer not found")
  86. errTransferMismatch = errors.New("transfer mismatch")
  87. )
  88. var (
  89. // Config is the configuration for the supported protocols
  90. Config Configuration
  91. // Connections is the list of active connections
  92. Connections ActiveConnections
  93. // QuotaScans is the list of active quota scans
  94. QuotaScans ActiveScans
  95. idleTimeoutTicker *time.Ticker
  96. idleTimeoutTickerDone chan bool
  97. supportedProtocols = []string{ProtocolSFTP, ProtocolSCP, ProtocolSSH, ProtocolFTP, ProtocolWebDAV}
  98. // the map key is the protocol, for each protocol we can have multiple rate limiters
  99. rateLimiters map[string][]*rateLimiter
  100. )
  101. // Initialize sets the common configuration
  102. func Initialize(c Configuration) error {
  103. Config = c
  104. Config.idleLoginTimeout = 2 * time.Minute
  105. Config.idleTimeoutAsDuration = time.Duration(Config.IdleTimeout) * time.Minute
  106. if Config.IdleTimeout > 0 {
  107. startIdleTimeoutTicker(idleTimeoutCheckInterval)
  108. }
  109. Config.defender = nil
  110. if c.DefenderConfig.Enabled {
  111. defender, err := newInMemoryDefender(&c.DefenderConfig)
  112. if err != nil {
  113. return fmt.Errorf("defender initialization error: %v", err)
  114. }
  115. logger.Info(logSender, "", "defender initialized with config %+v", c.DefenderConfig)
  116. Config.defender = defender
  117. }
  118. rateLimiters = make(map[string][]*rateLimiter)
  119. for _, rlCfg := range c.RateLimitersConfig {
  120. if rlCfg.isEnabled() {
  121. if err := rlCfg.validate(); err != nil {
  122. return fmt.Errorf("rate limiters initialization error: %v", err)
  123. }
  124. rateLimiter := rlCfg.getLimiter()
  125. for _, protocol := range rlCfg.Protocols {
  126. rateLimiters[protocol] = append(rateLimiters[protocol], rateLimiter)
  127. }
  128. }
  129. }
  130. return nil
  131. }
  132. // LimitRate blocks until all the configured rate limiters
  133. // allow one event to happen.
  134. // It returns an error if the time to wait exceeds the max
  135. // allowed delay
  136. func LimitRate(protocol, ip string) (time.Duration, error) {
  137. for _, limiter := range rateLimiters[protocol] {
  138. if delay, err := limiter.Wait(ip); err != nil {
  139. logger.Debug(logSender, "", "protocol %v ip %v: %v", protocol, ip, err)
  140. return delay, err
  141. }
  142. }
  143. return 0, nil
  144. }
  145. // ReloadDefender reloads the defender's block and safe lists
  146. func ReloadDefender() error {
  147. if Config.defender == nil {
  148. return nil
  149. }
  150. return Config.defender.Reload()
  151. }
  152. // IsBanned returns true if the specified IP address is banned
  153. func IsBanned(ip string) bool {
  154. if Config.defender == nil {
  155. return false
  156. }
  157. return Config.defender.IsBanned(ip)
  158. }
  159. // GetDefenderBanTime returns the ban time for the given IP
  160. // or nil if the IP is not banned or the defender is disabled
  161. func GetDefenderBanTime(ip string) *time.Time {
  162. if Config.defender == nil {
  163. return nil
  164. }
  165. return Config.defender.GetBanTime(ip)
  166. }
  167. // Unban removes the specified IP address from the banned ones
  168. func Unban(ip string) bool {
  169. if Config.defender == nil {
  170. return false
  171. }
  172. return Config.defender.Unban(ip)
  173. }
  174. // GetDefenderScore returns the score for the given IP
  175. func GetDefenderScore(ip string) int {
  176. if Config.defender == nil {
  177. return 0
  178. }
  179. return Config.defender.GetScore(ip)
  180. }
  181. // AddDefenderEvent adds the specified defender event for the given IP
  182. func AddDefenderEvent(ip string, event HostEvent) {
  183. if Config.defender == nil {
  184. return
  185. }
  186. Config.defender.AddEvent(ip, event)
  187. }
  188. // the ticker cannot be started/stopped from multiple goroutines
  189. func startIdleTimeoutTicker(duration time.Duration) {
  190. stopIdleTimeoutTicker()
  191. idleTimeoutTicker = time.NewTicker(duration)
  192. idleTimeoutTickerDone = make(chan bool)
  193. go func() {
  194. for {
  195. select {
  196. case <-idleTimeoutTickerDone:
  197. return
  198. case <-idleTimeoutTicker.C:
  199. Connections.checkIdles()
  200. }
  201. }
  202. }()
  203. }
  204. func stopIdleTimeoutTicker() {
  205. if idleTimeoutTicker != nil {
  206. idleTimeoutTicker.Stop()
  207. idleTimeoutTickerDone <- true
  208. idleTimeoutTicker = nil
  209. }
  210. }
  211. // ActiveTransfer defines the interface for the current active transfers
  212. type ActiveTransfer interface {
  213. GetID() uint64
  214. GetType() int
  215. GetSize() int64
  216. GetVirtualPath() string
  217. GetStartTime() time.Time
  218. SignalClose()
  219. Truncate(fsPath string, size int64) (int64, error)
  220. GetRealFsPath(fsPath string) string
  221. }
  222. // ActiveConnection defines the interface for the current active connections
  223. type ActiveConnection interface {
  224. GetID() string
  225. GetUsername() string
  226. GetRemoteAddress() string
  227. GetClientVersion() string
  228. GetProtocol() string
  229. GetConnectionTime() time.Time
  230. GetLastActivity() time.Time
  231. GetCommand() string
  232. Disconnect() error
  233. AddTransfer(t ActiveTransfer)
  234. RemoveTransfer(t ActiveTransfer)
  235. GetTransfers() []ConnectionTransfer
  236. CloseFS() error
  237. }
  238. // StatAttributes defines the attributes for set stat commands
  239. type StatAttributes struct {
  240. Mode os.FileMode
  241. Atime time.Time
  242. Mtime time.Time
  243. UID int
  244. GID int
  245. Flags int
  246. Size int64
  247. }
  248. // ConnectionTransfer defines the trasfer details to expose
  249. type ConnectionTransfer struct {
  250. ID uint64 `json:"-"`
  251. OperationType string `json:"operation_type"`
  252. StartTime int64 `json:"start_time"`
  253. Size int64 `json:"size"`
  254. VirtualPath string `json:"path"`
  255. }
  256. func (t *ConnectionTransfer) getConnectionTransferAsString() string {
  257. result := ""
  258. switch t.OperationType {
  259. case operationUpload:
  260. result += "UL "
  261. case operationDownload:
  262. result += "DL "
  263. }
  264. result += fmt.Sprintf("%#v ", t.VirtualPath)
  265. if t.Size > 0 {
  266. elapsed := time.Since(utils.GetTimeFromMsecSinceEpoch(t.StartTime))
  267. speed := float64(t.Size) / float64(utils.GetTimeAsMsSinceEpoch(time.Now())-t.StartTime)
  268. result += fmt.Sprintf("Size: %#v Elapsed: %#v Speed: \"%.1f KB/s\"", utils.ByteCountIEC(t.Size),
  269. utils.GetDurationAsString(elapsed), speed)
  270. }
  271. return result
  272. }
  273. // Configuration defines configuration parameters common to all supported protocols
  274. type Configuration struct {
  275. // Maximum idle timeout as minutes. If a client is idle for a time that exceeds this setting it will be disconnected.
  276. // 0 means disabled
  277. IdleTimeout int `json:"idle_timeout" mapstructure:"idle_timeout"`
  278. // UploadMode 0 means standard, the files are uploaded directly to the requested path.
  279. // 1 means atomic: the files are uploaded to a temporary path and renamed to the requested path
  280. // when the client ends the upload. Atomic mode avoid problems such as a web server that
  281. // serves partial files when the files are being uploaded.
  282. // In atomic mode if there is an upload error the temporary file is deleted and so the requested
  283. // upload path will not contain a partial file.
  284. // 2 means atomic with resume support: as atomic but if there is an upload error the temporary
  285. // file is renamed to the requested path and not deleted, this way a client can reconnect and resume
  286. // the upload.
  287. UploadMode int `json:"upload_mode" mapstructure:"upload_mode"`
  288. // Actions to execute for SFTP file operations and SSH commands
  289. Actions ProtocolActions `json:"actions" mapstructure:"actions"`
  290. // SetstatMode 0 means "normal mode": requests for changing permissions and owner/group are executed.
  291. // 1 means "ignore mode": requests for changing permissions and owner/group are silently ignored.
  292. // 2 means "ignore mode for cloud fs": requests for changing permissions and owner/group/time are
  293. // silently ignored for cloud based filesystem such as S3, GCS, Azure Blob
  294. SetstatMode int `json:"setstat_mode" mapstructure:"setstat_mode"`
  295. // Support for HAProxy PROXY protocol.
  296. // If you are running SFTPGo behind a proxy server such as HAProxy, AWS ELB or NGNIX, you can enable
  297. // the proxy protocol. It provides a convenient way to safely transport connection information
  298. // such as a client's address across multiple layers of NAT or TCP proxies to get the real
  299. // client IP address instead of the proxy IP. Both protocol versions 1 and 2 are supported.
  300. // - 0 means disabled
  301. // - 1 means proxy protocol enabled. Proxy header will be used and requests without proxy header will be accepted.
  302. // - 2 means proxy protocol required. Proxy header will be used and requests without proxy header will be rejected.
  303. // If the proxy protocol is enabled in SFTPGo then you have to enable the protocol in your proxy configuration too,
  304. // for example for HAProxy add "send-proxy" or "send-proxy-v2" to each server configuration line.
  305. ProxyProtocol int `json:"proxy_protocol" mapstructure:"proxy_protocol"`
  306. // List of IP addresses and IP ranges allowed to send the proxy header.
  307. // If proxy protocol is set to 1 and we receive a proxy header from an IP that is not in the list then the
  308. // connection will be accepted and the header will be ignored.
  309. // If proxy protocol is set to 2 and we receive a proxy header from an IP that is not in the list then the
  310. // connection will be rejected.
  311. ProxyAllowed []string `json:"proxy_allowed" mapstructure:"proxy_allowed"`
  312. // Absolute path to an external program or an HTTP URL to invoke as soon as SFTPGo starts.
  313. // If you define an HTTP URL it will be invoked using a `GET` request.
  314. // Please note that SFTPGo services may not yet be available when this hook is run.
  315. // Leave empty do disable.
  316. StartupHook string `json:"startup_hook" mapstructure:"startup_hook"`
  317. // Absolute path to an external program or an HTTP URL to invoke after a user connects
  318. // and before he tries to login. It allows you to reject the connection based on the source
  319. // ip address. Leave empty do disable.
  320. PostConnectHook string `json:"post_connect_hook" mapstructure:"post_connect_hook"`
  321. // Maximum number of concurrent client connections. 0 means unlimited
  322. MaxTotalConnections int `json:"max_total_connections" mapstructure:"max_total_connections"`
  323. // Defender configuration
  324. DefenderConfig DefenderConfig `json:"defender" mapstructure:"defender"`
  325. // Rate limiter configurations
  326. RateLimitersConfig []RateLimiterConfig `json:"rate_limiters" mapstructure:"rate_limiters"`
  327. idleTimeoutAsDuration time.Duration
  328. idleLoginTimeout time.Duration
  329. defender Defender
  330. }
  331. // IsAtomicUploadEnabled returns true if atomic upload is enabled
  332. func (c *Configuration) IsAtomicUploadEnabled() bool {
  333. return c.UploadMode == UploadModeAtomic || c.UploadMode == UploadModeAtomicWithResume
  334. }
  335. // GetProxyListener returns a wrapper for the given listener that supports the
  336. // HAProxy Proxy Protocol or nil if the proxy protocol is not configured
  337. func (c *Configuration) GetProxyListener(listener net.Listener) (*proxyproto.Listener, error) {
  338. var proxyListener *proxyproto.Listener
  339. var err error
  340. if c.ProxyProtocol > 0 {
  341. var policyFunc func(upstream net.Addr) (proxyproto.Policy, error)
  342. if c.ProxyProtocol == 1 && len(c.ProxyAllowed) > 0 {
  343. policyFunc, err = proxyproto.LaxWhiteListPolicy(c.ProxyAllowed)
  344. if err != nil {
  345. return nil, err
  346. }
  347. }
  348. if c.ProxyProtocol == 2 {
  349. if len(c.ProxyAllowed) == 0 {
  350. policyFunc = func(upstream net.Addr) (proxyproto.Policy, error) {
  351. return proxyproto.REQUIRE, nil
  352. }
  353. } else {
  354. policyFunc, err = proxyproto.StrictWhiteListPolicy(c.ProxyAllowed)
  355. if err != nil {
  356. return nil, err
  357. }
  358. }
  359. }
  360. proxyListener = &proxyproto.Listener{
  361. Listener: listener,
  362. Policy: policyFunc,
  363. }
  364. }
  365. return proxyListener, nil
  366. }
  367. // ExecuteStartupHook runs the startup hook if defined
  368. func (c *Configuration) ExecuteStartupHook() error {
  369. if c.StartupHook == "" {
  370. return nil
  371. }
  372. if strings.HasPrefix(c.StartupHook, "http") {
  373. var url *url.URL
  374. url, err := url.Parse(c.StartupHook)
  375. if err != nil {
  376. logger.Warn(logSender, "", "Invalid startup hook %#v: %v", c.StartupHook, err)
  377. return err
  378. }
  379. startTime := time.Now()
  380. httpClient := httpclient.GetRetraybleHTTPClient()
  381. resp, err := httpClient.Get(url.String())
  382. if err != nil {
  383. logger.Warn(logSender, "", "Error executing startup hook: %v", err)
  384. return err
  385. }
  386. defer resp.Body.Close()
  387. logger.Debug(logSender, "", "Startup hook executed, elapsed: %v, response code: %v", time.Since(startTime), resp.StatusCode)
  388. return nil
  389. }
  390. if !filepath.IsAbs(c.StartupHook) {
  391. err := fmt.Errorf("invalid startup hook %#v", c.StartupHook)
  392. logger.Warn(logSender, "", "Invalid startup hook %#v", c.StartupHook)
  393. return err
  394. }
  395. startTime := time.Now()
  396. ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
  397. defer cancel()
  398. cmd := exec.CommandContext(ctx, c.StartupHook)
  399. err := cmd.Run()
  400. logger.Debug(logSender, "", "Startup hook executed, elapsed: %v, error: %v", time.Since(startTime), err)
  401. return nil
  402. }
  403. // ExecutePostConnectHook executes the post connect hook if defined
  404. func (c *Configuration) ExecutePostConnectHook(ipAddr, protocol string) error {
  405. if c.PostConnectHook == "" {
  406. return nil
  407. }
  408. if strings.HasPrefix(c.PostConnectHook, "http") {
  409. var url *url.URL
  410. url, err := url.Parse(c.PostConnectHook)
  411. if err != nil {
  412. logger.Warn(protocol, "", "Login from ip %#v denied, invalid post connect hook %#v: %v",
  413. ipAddr, c.PostConnectHook, err)
  414. return err
  415. }
  416. httpClient := httpclient.GetRetraybleHTTPClient()
  417. q := url.Query()
  418. q.Add("ip", ipAddr)
  419. q.Add("protocol", protocol)
  420. url.RawQuery = q.Encode()
  421. resp, err := httpClient.Get(url.String())
  422. if err != nil {
  423. logger.Warn(protocol, "", "Login from ip %#v denied, error executing post connect hook: %v", ipAddr, err)
  424. return err
  425. }
  426. defer resp.Body.Close()
  427. if resp.StatusCode != http.StatusOK {
  428. logger.Warn(protocol, "", "Login from ip %#v denied, post connect hook response code: %v", ipAddr, resp.StatusCode)
  429. return errUnexpectedHTTResponse
  430. }
  431. return nil
  432. }
  433. if !filepath.IsAbs(c.PostConnectHook) {
  434. err := fmt.Errorf("invalid post connect hook %#v", c.PostConnectHook)
  435. logger.Warn(protocol, "", "Login from ip %#v denied: %v", ipAddr, err)
  436. return err
  437. }
  438. ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
  439. defer cancel()
  440. cmd := exec.CommandContext(ctx, c.PostConnectHook)
  441. cmd.Env = append(os.Environ(),
  442. fmt.Sprintf("SFTPGO_CONNECTION_IP=%v", ipAddr),
  443. fmt.Sprintf("SFTPGO_CONNECTION_PROTOCOL=%v", protocol))
  444. err := cmd.Run()
  445. if err != nil {
  446. logger.Warn(protocol, "", "Login from ip %#v denied, connect hook error: %v", ipAddr, err)
  447. }
  448. return err
  449. }
  450. // SSHConnection defines an ssh connection.
  451. // Each SSH connection can open several channels for SFTP or SSH commands
  452. type SSHConnection struct {
  453. id string
  454. conn net.Conn
  455. lastActivity int64
  456. }
  457. // NewSSHConnection returns a new SSHConnection
  458. func NewSSHConnection(id string, conn net.Conn) *SSHConnection {
  459. return &SSHConnection{
  460. id: id,
  461. conn: conn,
  462. lastActivity: time.Now().UnixNano(),
  463. }
  464. }
  465. // GetID returns the ID for this SSHConnection
  466. func (c *SSHConnection) GetID() string {
  467. return c.id
  468. }
  469. // UpdateLastActivity updates last activity for this connection
  470. func (c *SSHConnection) UpdateLastActivity() {
  471. atomic.StoreInt64(&c.lastActivity, time.Now().UnixNano())
  472. }
  473. // GetLastActivity returns the last connection activity
  474. func (c *SSHConnection) GetLastActivity() time.Time {
  475. return time.Unix(0, atomic.LoadInt64(&c.lastActivity))
  476. }
  477. // Close closes the underlying network connection
  478. func (c *SSHConnection) Close() error {
  479. return c.conn.Close()
  480. }
  481. // ActiveConnections holds the currect active connections with the associated transfers
  482. type ActiveConnections struct {
  483. sync.RWMutex
  484. connections []ActiveConnection
  485. sshConnections []*SSHConnection
  486. }
  487. // GetActiveSessions returns the number of active sessions for the given username.
  488. // We return the open sessions for any protocol
  489. func (conns *ActiveConnections) GetActiveSessions(username string) int {
  490. conns.RLock()
  491. defer conns.RUnlock()
  492. numSessions := 0
  493. for _, c := range conns.connections {
  494. if c.GetUsername() == username {
  495. numSessions++
  496. }
  497. }
  498. return numSessions
  499. }
  500. // Add adds a new connection to the active ones
  501. func (conns *ActiveConnections) Add(c ActiveConnection) {
  502. conns.Lock()
  503. defer conns.Unlock()
  504. conns.connections = append(conns.connections, c)
  505. metrics.UpdateActiveConnectionsSize(len(conns.connections))
  506. logger.Debug(c.GetProtocol(), c.GetID(), "connection added, num open connections: %v", len(conns.connections))
  507. }
  508. // Swap replaces an existing connection with the given one.
  509. // This method is useful if you have to change some connection details
  510. // for example for FTP is used to update the connection once the user
  511. // authenticates
  512. func (conns *ActiveConnections) Swap(c ActiveConnection) error {
  513. conns.Lock()
  514. defer conns.Unlock()
  515. for idx, conn := range conns.connections {
  516. if conn.GetID() == c.GetID() {
  517. conn = nil
  518. conns.connections[idx] = c
  519. return nil
  520. }
  521. }
  522. return errors.New("connection to swap not found")
  523. }
  524. // Remove removes a connection from the active ones
  525. func (conns *ActiveConnections) Remove(connectionID string) {
  526. conns.Lock()
  527. defer conns.Unlock()
  528. for idx, conn := range conns.connections {
  529. if conn.GetID() == connectionID {
  530. err := conn.CloseFS()
  531. lastIdx := len(conns.connections) - 1
  532. conns.connections[idx] = conns.connections[lastIdx]
  533. conns.connections[lastIdx] = nil
  534. conns.connections = conns.connections[:lastIdx]
  535. metrics.UpdateActiveConnectionsSize(lastIdx)
  536. logger.Debug(conn.GetProtocol(), conn.GetID(), "connection removed, close fs error: %v, num open connections: %v",
  537. err, lastIdx)
  538. return
  539. }
  540. }
  541. logger.Warn(logSender, "", "connection id %#v to remove not found!", connectionID)
  542. }
  543. // Close closes an active connection.
  544. // It returns true on success
  545. func (conns *ActiveConnections) Close(connectionID string) bool {
  546. conns.RLock()
  547. result := false
  548. for _, c := range conns.connections {
  549. if c.GetID() == connectionID {
  550. defer func(conn ActiveConnection) {
  551. err := conn.Disconnect()
  552. logger.Debug(conn.GetProtocol(), conn.GetID(), "close connection requested, close err: %v", err)
  553. }(c)
  554. result = true
  555. break
  556. }
  557. }
  558. conns.RUnlock()
  559. return result
  560. }
  561. // AddSSHConnection adds a new ssh connection to the active ones
  562. func (conns *ActiveConnections) AddSSHConnection(c *SSHConnection) {
  563. conns.Lock()
  564. defer conns.Unlock()
  565. conns.sshConnections = append(conns.sshConnections, c)
  566. logger.Debug(logSender, c.GetID(), "ssh connection added, num open connections: %v", len(conns.sshConnections))
  567. }
  568. // RemoveSSHConnection removes a connection from the active ones
  569. func (conns *ActiveConnections) RemoveSSHConnection(connectionID string) {
  570. conns.Lock()
  571. defer conns.Unlock()
  572. for idx, conn := range conns.sshConnections {
  573. if conn.GetID() == connectionID {
  574. lastIdx := len(conns.sshConnections) - 1
  575. conns.sshConnections[idx] = conns.sshConnections[lastIdx]
  576. conns.sshConnections[lastIdx] = nil
  577. conns.sshConnections = conns.sshConnections[:lastIdx]
  578. logger.Debug(logSender, conn.GetID(), "ssh connection removed, num open ssh connections: %v", lastIdx)
  579. return
  580. }
  581. }
  582. logger.Warn(logSender, "", "ssh connection to remove with id %#v not found!", connectionID)
  583. }
  584. func (conns *ActiveConnections) checkIdles() {
  585. conns.RLock()
  586. for _, sshConn := range conns.sshConnections {
  587. idleTime := time.Since(sshConn.GetLastActivity())
  588. if idleTime > Config.idleTimeoutAsDuration {
  589. // we close the an ssh connection if it has no active connections associated
  590. idToMatch := fmt.Sprintf("_%v_", sshConn.GetID())
  591. toClose := true
  592. for _, conn := range conns.connections {
  593. if strings.Contains(conn.GetID(), idToMatch) {
  594. toClose = false
  595. break
  596. }
  597. }
  598. if toClose {
  599. defer func(c *SSHConnection) {
  600. err := c.Close()
  601. logger.Debug(logSender, c.GetID(), "close idle SSH connection, idle time: %v, close err: %v",
  602. time.Since(c.GetLastActivity()), err)
  603. }(sshConn)
  604. }
  605. }
  606. }
  607. for _, c := range conns.connections {
  608. idleTime := time.Since(c.GetLastActivity())
  609. isUnauthenticatedFTPUser := (c.GetProtocol() == ProtocolFTP && c.GetUsername() == "")
  610. if idleTime > Config.idleTimeoutAsDuration || (isUnauthenticatedFTPUser && idleTime > Config.idleLoginTimeout) {
  611. defer func(conn ActiveConnection, isFTPNoAuth bool) {
  612. err := conn.Disconnect()
  613. logger.Debug(conn.GetProtocol(), conn.GetID(), "close idle connection, idle time: %v, username: %#v close err: %v",
  614. time.Since(conn.GetLastActivity()), conn.GetUsername(), err)
  615. if isFTPNoAuth {
  616. ip := utils.GetIPFromRemoteAddress(c.GetRemoteAddress())
  617. logger.ConnectionFailedLog("", ip, dataprovider.LoginMethodNoAuthTryed, c.GetProtocol(), "client idle")
  618. metrics.AddNoAuthTryed()
  619. AddDefenderEvent(ip, HostEventNoLoginTried)
  620. dataprovider.ExecutePostLoginHook(&dataprovider.User{}, dataprovider.LoginMethodNoAuthTryed, ip, c.GetProtocol(),
  621. dataprovider.ErrNoAuthTryed)
  622. }
  623. }(c, isUnauthenticatedFTPUser)
  624. }
  625. }
  626. conns.RUnlock()
  627. }
  628. // IsNewConnectionAllowed returns false if the maximum number of concurrent allowed connections is exceeded
  629. func (conns *ActiveConnections) IsNewConnectionAllowed() bool {
  630. if Config.MaxTotalConnections == 0 {
  631. return true
  632. }
  633. conns.RLock()
  634. defer conns.RUnlock()
  635. return len(conns.connections) < Config.MaxTotalConnections
  636. }
  637. // GetStats returns stats for active connections
  638. func (conns *ActiveConnections) GetStats() []*ConnectionStatus {
  639. conns.RLock()
  640. defer conns.RUnlock()
  641. stats := make([]*ConnectionStatus, 0, len(conns.connections))
  642. for _, c := range conns.connections {
  643. stat := &ConnectionStatus{
  644. Username: c.GetUsername(),
  645. ConnectionID: c.GetID(),
  646. ClientVersion: c.GetClientVersion(),
  647. RemoteAddress: c.GetRemoteAddress(),
  648. ConnectionTime: utils.GetTimeAsMsSinceEpoch(c.GetConnectionTime()),
  649. LastActivity: utils.GetTimeAsMsSinceEpoch(c.GetLastActivity()),
  650. Protocol: c.GetProtocol(),
  651. Command: c.GetCommand(),
  652. Transfers: c.GetTransfers(),
  653. }
  654. stats = append(stats, stat)
  655. }
  656. return stats
  657. }
  658. // ConnectionStatus returns the status for an active connection
  659. type ConnectionStatus struct {
  660. // Logged in username
  661. Username string `json:"username"`
  662. // Unique identifier for the connection
  663. ConnectionID string `json:"connection_id"`
  664. // client's version string
  665. ClientVersion string `json:"client_version,omitempty"`
  666. // Remote address for this connection
  667. RemoteAddress string `json:"remote_address"`
  668. // Connection time as unix timestamp in milliseconds
  669. ConnectionTime int64 `json:"connection_time"`
  670. // Last activity as unix timestamp in milliseconds
  671. LastActivity int64 `json:"last_activity"`
  672. // Protocol for this connection
  673. Protocol string `json:"protocol"`
  674. // active uploads/downloads
  675. Transfers []ConnectionTransfer `json:"active_transfers,omitempty"`
  676. // SSH command or WebDAV method
  677. Command string `json:"command,omitempty"`
  678. }
  679. // GetConnectionDuration returns the connection duration as string
  680. func (c *ConnectionStatus) GetConnectionDuration() string {
  681. elapsed := time.Since(utils.GetTimeFromMsecSinceEpoch(c.ConnectionTime))
  682. return utils.GetDurationAsString(elapsed)
  683. }
  684. // GetConnectionInfo returns connection info.
  685. // Protocol,Client Version and RemoteAddress are returned.
  686. func (c *ConnectionStatus) GetConnectionInfo() string {
  687. var result strings.Builder
  688. result.WriteString(fmt.Sprintf("%v. Client: %#v From: %#v", c.Protocol, c.ClientVersion, c.RemoteAddress))
  689. if c.Command == "" {
  690. return result.String()
  691. }
  692. switch c.Protocol {
  693. case ProtocolSSH, ProtocolFTP:
  694. result.WriteString(fmt.Sprintf(". Command: %#v", c.Command))
  695. case ProtocolWebDAV:
  696. result.WriteString(fmt.Sprintf(". Method: %#v", c.Command))
  697. }
  698. return result.String()
  699. }
  700. // GetTransfersAsString returns the active transfers as string
  701. func (c *ConnectionStatus) GetTransfersAsString() string {
  702. result := ""
  703. for _, t := range c.Transfers {
  704. if result != "" {
  705. result += ". "
  706. }
  707. result += t.getConnectionTransferAsString()
  708. }
  709. return result
  710. }
  711. // ActiveQuotaScan defines an active quota scan for a user home dir
  712. type ActiveQuotaScan struct {
  713. // Username to which the quota scan refers
  714. Username string `json:"username"`
  715. // quota scan start time as unix timestamp in milliseconds
  716. StartTime int64 `json:"start_time"`
  717. }
  718. // ActiveVirtualFolderQuotaScan defines an active quota scan for a virtual folder
  719. type ActiveVirtualFolderQuotaScan struct {
  720. // folder name to which the quota scan refers
  721. Name string `json:"name"`
  722. // quota scan start time as unix timestamp in milliseconds
  723. StartTime int64 `json:"start_time"`
  724. }
  725. // ActiveScans holds the active quota scans
  726. type ActiveScans struct {
  727. sync.RWMutex
  728. UserHomeScans []ActiveQuotaScan
  729. FolderScans []ActiveVirtualFolderQuotaScan
  730. }
  731. // GetUsersQuotaScans returns the active quota scans for users home directories
  732. func (s *ActiveScans) GetUsersQuotaScans() []ActiveQuotaScan {
  733. s.RLock()
  734. defer s.RUnlock()
  735. scans := make([]ActiveQuotaScan, len(s.UserHomeScans))
  736. copy(scans, s.UserHomeScans)
  737. return scans
  738. }
  739. // AddUserQuotaScan adds a user to the ones with active quota scans.
  740. // Returns false if the user has a quota scan already running
  741. func (s *ActiveScans) AddUserQuotaScan(username string) bool {
  742. s.Lock()
  743. defer s.Unlock()
  744. for _, scan := range s.UserHomeScans {
  745. if scan.Username == username {
  746. return false
  747. }
  748. }
  749. s.UserHomeScans = append(s.UserHomeScans, ActiveQuotaScan{
  750. Username: username,
  751. StartTime: utils.GetTimeAsMsSinceEpoch(time.Now()),
  752. })
  753. return true
  754. }
  755. // RemoveUserQuotaScan removes a user from the ones with active quota scans.
  756. // Returns false if the user has no active quota scans
  757. func (s *ActiveScans) RemoveUserQuotaScan(username string) bool {
  758. s.Lock()
  759. defer s.Unlock()
  760. indexToRemove := -1
  761. for i, scan := range s.UserHomeScans {
  762. if scan.Username == username {
  763. indexToRemove = i
  764. break
  765. }
  766. }
  767. if indexToRemove >= 0 {
  768. s.UserHomeScans[indexToRemove] = s.UserHomeScans[len(s.UserHomeScans)-1]
  769. s.UserHomeScans = s.UserHomeScans[:len(s.UserHomeScans)-1]
  770. return true
  771. }
  772. return false
  773. }
  774. // GetVFoldersQuotaScans returns the active quota scans for virtual folders
  775. func (s *ActiveScans) GetVFoldersQuotaScans() []ActiveVirtualFolderQuotaScan {
  776. s.RLock()
  777. defer s.RUnlock()
  778. scans := make([]ActiveVirtualFolderQuotaScan, len(s.FolderScans))
  779. copy(scans, s.FolderScans)
  780. return scans
  781. }
  782. // AddVFolderQuotaScan adds a virtual folder to the ones with active quota scans.
  783. // Returns false if the folder has a quota scan already running
  784. func (s *ActiveScans) AddVFolderQuotaScan(folderName string) bool {
  785. s.Lock()
  786. defer s.Unlock()
  787. for _, scan := range s.FolderScans {
  788. if scan.Name == folderName {
  789. return false
  790. }
  791. }
  792. s.FolderScans = append(s.FolderScans, ActiveVirtualFolderQuotaScan{
  793. Name: folderName,
  794. StartTime: utils.GetTimeAsMsSinceEpoch(time.Now()),
  795. })
  796. return true
  797. }
  798. // RemoveVFolderQuotaScan removes a folder from the ones with active quota scans.
  799. // Returns false if the folder has no active quota scans
  800. func (s *ActiveScans) RemoveVFolderQuotaScan(folderName string) bool {
  801. s.Lock()
  802. defer s.Unlock()
  803. indexToRemove := -1
  804. for i, scan := range s.FolderScans {
  805. if scan.Name == folderName {
  806. indexToRemove = i
  807. break
  808. }
  809. }
  810. if indexToRemove >= 0 {
  811. s.FolderScans[indexToRemove] = s.FolderScans[len(s.FolderScans)-1]
  812. s.FolderScans = s.FolderScans[:len(s.FolderScans)-1]
  813. return true
  814. }
  815. return false
  816. }