common.go 47 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457
  1. // Copyright (C) 2019-2023 Nicola Murino
  2. //
  3. // This program is free software: you can redistribute it and/or modify
  4. // it under the terms of the GNU Affero General Public License as published
  5. // by the Free Software Foundation, version 3.
  6. //
  7. // This program is distributed in the hope that it will be useful,
  8. // but WITHOUT ANY WARRANTY; without even the implied warranty of
  9. // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  10. // GNU Affero General Public License for more details.
  11. //
  12. // You should have received a copy of the GNU Affero General Public License
  13. // along with this program. If not, see <https://www.gnu.org/licenses/>.
  14. // Package common defines code shared among file transfer packages and protocols
  15. package common
  16. import (
  17. "context"
  18. "errors"
  19. "fmt"
  20. "net"
  21. "net/http"
  22. "net/url"
  23. "os"
  24. "os/exec"
  25. "path/filepath"
  26. "strconv"
  27. "strings"
  28. "sync"
  29. "sync/atomic"
  30. "time"
  31. "github.com/pires/go-proxyproto"
  32. "github.com/drakkan/sftpgo/v2/internal/command"
  33. "github.com/drakkan/sftpgo/v2/internal/dataprovider"
  34. "github.com/drakkan/sftpgo/v2/internal/httpclient"
  35. "github.com/drakkan/sftpgo/v2/internal/logger"
  36. "github.com/drakkan/sftpgo/v2/internal/metric"
  37. "github.com/drakkan/sftpgo/v2/internal/plugin"
  38. "github.com/drakkan/sftpgo/v2/internal/util"
  39. "github.com/drakkan/sftpgo/v2/internal/vfs"
  40. )
  41. // constants
  42. const (
  43. logSender = "common"
  44. uploadLogSender = "Upload"
  45. downloadLogSender = "Download"
  46. renameLogSender = "Rename"
  47. rmdirLogSender = "Rmdir"
  48. mkdirLogSender = "Mkdir"
  49. symlinkLogSender = "Symlink"
  50. removeLogSender = "Remove"
  51. chownLogSender = "Chown"
  52. chmodLogSender = "Chmod"
  53. chtimesLogSender = "Chtimes"
  54. copyLogSender = "Copy"
  55. truncateLogSender = "Truncate"
  56. operationDownload = "download"
  57. operationUpload = "upload"
  58. operationFirstDownload = "first-download"
  59. operationFirstUpload = "first-upload"
  60. operationDelete = "delete"
  61. operationCopy = "copy"
  62. // Pre-download action name
  63. OperationPreDownload = "pre-download"
  64. // Pre-upload action name
  65. OperationPreUpload = "pre-upload"
  66. operationPreDelete = "pre-delete"
  67. operationRename = "rename"
  68. operationMkdir = "mkdir"
  69. operationRmdir = "rmdir"
  70. // SSH command action name
  71. OperationSSHCmd = "ssh_cmd"
  72. chtimesFormat = "2006-01-02T15:04:05" // YYYY-MM-DDTHH:MM:SS
  73. idleTimeoutCheckInterval = 3 * time.Minute
  74. periodicTimeoutCheckInterval = 1 * time.Minute
  75. )
  76. // Stat flags
  77. const (
  78. StatAttrUIDGID = 1
  79. StatAttrPerms = 2
  80. StatAttrTimes = 4
  81. StatAttrSize = 8
  82. )
  83. // Transfer types
  84. const (
  85. TransferUpload = iota
  86. TransferDownload
  87. )
  88. // Supported protocols
  89. const (
  90. ProtocolSFTP = "SFTP"
  91. ProtocolSCP = "SCP"
  92. ProtocolSSH = "SSH"
  93. ProtocolFTP = "FTP"
  94. ProtocolWebDAV = "DAV"
  95. ProtocolHTTP = "HTTP"
  96. ProtocolHTTPShare = "HTTPShare"
  97. ProtocolDataRetention = "DataRetention"
  98. ProtocolOIDC = "OIDC"
  99. protocolEventAction = "EventAction"
  100. )
  101. // Upload modes
  102. const (
  103. UploadModeStandard = iota
  104. UploadModeAtomic
  105. UploadModeAtomicWithResume
  106. )
  107. func init() {
  108. Connections.clients = clientsMap{
  109. clients: make(map[string]int),
  110. }
  111. Connections.perUserConns = make(map[string]int)
  112. Connections.mapping = make(map[string]int)
  113. Connections.sshMapping = make(map[string]int)
  114. }
  115. // errors definitions
  116. var (
  117. ErrPermissionDenied = errors.New("permission denied")
  118. ErrNotExist = errors.New("no such file or directory")
  119. ErrOpUnsupported = errors.New("operation unsupported")
  120. ErrGenericFailure = errors.New("failure")
  121. ErrQuotaExceeded = errors.New("denying write due to space limit")
  122. ErrReadQuotaExceeded = errors.New("denying read due to quota limit")
  123. ErrConnectionDenied = errors.New("you are not allowed to connect")
  124. ErrNoBinding = errors.New("no binding configured")
  125. ErrCrtRevoked = errors.New("your certificate has been revoked")
  126. ErrNoCredentials = errors.New("no credential provided")
  127. ErrInternalFailure = errors.New("internal failure")
  128. ErrTransferAborted = errors.New("transfer aborted")
  129. ErrShuttingDown = errors.New("the service is shutting down")
  130. errNoTransfer = errors.New("requested transfer not found")
  131. errTransferMismatch = errors.New("transfer mismatch")
  132. )
  133. var (
  134. // Config is the configuration for the supported protocols
  135. Config Configuration
  136. // Connections is the list of active connections
  137. Connections ActiveConnections
  138. // QuotaScans is the list of active quota scans
  139. QuotaScans ActiveScans
  140. // ActiveMetadataChecks holds the active metadata checks
  141. ActiveMetadataChecks MetadataChecks
  142. transfersChecker TransfersChecker
  143. supportedProtocols = []string{ProtocolSFTP, ProtocolSCP, ProtocolSSH, ProtocolFTP, ProtocolWebDAV,
  144. ProtocolHTTP, ProtocolHTTPShare, ProtocolOIDC}
  145. disconnHookProtocols = []string{ProtocolSFTP, ProtocolSCP, ProtocolSSH, ProtocolFTP}
  146. // the map key is the protocol, for each protocol we can have multiple rate limiters
  147. rateLimiters map[string][]*rateLimiter
  148. isShuttingDown atomic.Bool
  149. )
  150. // Initialize sets the common configuration
  151. func Initialize(c Configuration, isShared int) error {
  152. isShuttingDown.Store(false)
  153. Config = c
  154. Config.Actions.ExecuteOn = util.RemoveDuplicates(Config.Actions.ExecuteOn, true)
  155. Config.Actions.ExecuteSync = util.RemoveDuplicates(Config.Actions.ExecuteSync, true)
  156. Config.ProxyAllowed = util.RemoveDuplicates(Config.ProxyAllowed, true)
  157. Config.idleLoginTimeout = 2 * time.Minute
  158. Config.idleTimeoutAsDuration = time.Duration(Config.IdleTimeout) * time.Minute
  159. startPeriodicChecks(periodicTimeoutCheckInterval)
  160. Config.defender = nil
  161. Config.whitelist = nil
  162. rateLimiters = make(map[string][]*rateLimiter)
  163. for _, rlCfg := range c.RateLimitersConfig {
  164. if rlCfg.isEnabled() {
  165. if err := rlCfg.validate(); err != nil {
  166. return fmt.Errorf("rate limiters initialization error: %w", err)
  167. }
  168. allowList, err := util.ParseAllowedIPAndRanges(rlCfg.AllowList)
  169. if err != nil {
  170. return fmt.Errorf("unable to parse rate limiter allow list %v: %v", rlCfg.AllowList, err)
  171. }
  172. rateLimiter := rlCfg.getLimiter()
  173. rateLimiter.allowList = allowList
  174. for _, protocol := range rlCfg.Protocols {
  175. rateLimiters[protocol] = append(rateLimiters[protocol], rateLimiter)
  176. }
  177. }
  178. }
  179. if c.DefenderConfig.Enabled {
  180. if !util.Contains(supportedDefenderDrivers, c.DefenderConfig.Driver) {
  181. return fmt.Errorf("unsupported defender driver %#v", c.DefenderConfig.Driver)
  182. }
  183. var defender Defender
  184. var err error
  185. switch c.DefenderConfig.Driver {
  186. case DefenderDriverProvider:
  187. defender, err = newDBDefender(&c.DefenderConfig)
  188. default:
  189. defender, err = newInMemoryDefender(&c.DefenderConfig)
  190. }
  191. if err != nil {
  192. return fmt.Errorf("defender initialization error: %v", err)
  193. }
  194. logger.Info(logSender, "", "defender initialized with config %+v", c.DefenderConfig)
  195. Config.defender = defender
  196. }
  197. if c.WhiteListFile != "" {
  198. whitelist := &whitelist{
  199. fileName: c.WhiteListFile,
  200. }
  201. if err := whitelist.reload(); err != nil {
  202. return fmt.Errorf("whitelist initialization error: %w", err)
  203. }
  204. logger.Info(logSender, "", "whitelist initialized from file: %#v", c.WhiteListFile)
  205. Config.whitelist = whitelist
  206. }
  207. vfs.SetTempPath(c.TempPath)
  208. dataprovider.SetTempPath(c.TempPath)
  209. vfs.SetAllowSelfConnections(c.AllowSelfConnections)
  210. vfs.SetRenameMode(c.RenameMode)
  211. dataprovider.SetAllowSelfConnections(c.AllowSelfConnections)
  212. transfersChecker = getTransfersChecker(isShared)
  213. return nil
  214. }
  215. // CheckClosing returns an error if the service is closing
  216. func CheckClosing() error {
  217. if isShuttingDown.Load() {
  218. return ErrShuttingDown
  219. }
  220. return nil
  221. }
  222. // WaitForTransfers waits, for the specified grace time, for currently ongoing
  223. // client-initiated transfer sessions to completes.
  224. // A zero graceTime means no wait
  225. func WaitForTransfers(graceTime int) {
  226. if graceTime == 0 {
  227. return
  228. }
  229. if isShuttingDown.Swap(true) {
  230. return
  231. }
  232. if activeHooks.Load() == 0 && getActiveConnections() == 0 {
  233. return
  234. }
  235. graceTimer := time.NewTimer(time.Duration(graceTime) * time.Second)
  236. ticker := time.NewTicker(3 * time.Second)
  237. for {
  238. select {
  239. case <-ticker.C:
  240. hooks := activeHooks.Load()
  241. logger.Info(logSender, "", "active hooks: %d", hooks)
  242. if hooks == 0 && getActiveConnections() == 0 {
  243. logger.Info(logSender, "", "no more active connections, graceful shutdown")
  244. ticker.Stop()
  245. graceTimer.Stop()
  246. return
  247. }
  248. case <-graceTimer.C:
  249. logger.Info(logSender, "", "grace time expired, hard shutdown")
  250. ticker.Stop()
  251. return
  252. }
  253. }
  254. }
  255. // getActiveConnections returns the number of connections with active transfers
  256. func getActiveConnections() int {
  257. var activeConns int
  258. Connections.RLock()
  259. for _, c := range Connections.connections {
  260. if len(c.GetTransfers()) > 0 {
  261. activeConns++
  262. }
  263. }
  264. Connections.RUnlock()
  265. logger.Info(logSender, "", "number of connections with active transfers: %d", activeConns)
  266. return activeConns
  267. }
  268. // LimitRate blocks until all the configured rate limiters
  269. // allow one event to happen.
  270. // It returns an error if the time to wait exceeds the max
  271. // allowed delay
  272. func LimitRate(protocol, ip string) (time.Duration, error) {
  273. for _, limiter := range rateLimiters[protocol] {
  274. if delay, err := limiter.Wait(ip); err != nil {
  275. logger.Debug(logSender, "", "protocol %v ip %v: %v", protocol, ip, err)
  276. return delay, err
  277. }
  278. }
  279. return 0, nil
  280. }
  281. // Reload reloads the whitelist, the IP filter plugin and the defender's block and safe lists
  282. func Reload() error {
  283. plugin.Handler.ReloadFilter()
  284. var errWithelist error
  285. if Config.whitelist != nil {
  286. errWithelist = Config.whitelist.reload()
  287. }
  288. if Config.defender == nil {
  289. return errWithelist
  290. }
  291. if err := Config.defender.Reload(); err != nil {
  292. return err
  293. }
  294. return errWithelist
  295. }
  296. // IsBanned returns true if the specified IP address is banned
  297. func IsBanned(ip string) bool {
  298. if plugin.Handler.IsIPBanned(ip) {
  299. return true
  300. }
  301. if Config.defender == nil {
  302. return false
  303. }
  304. return Config.defender.IsBanned(ip)
  305. }
  306. // GetDefenderBanTime returns the ban time for the given IP
  307. // or nil if the IP is not banned or the defender is disabled
  308. func GetDefenderBanTime(ip string) (*time.Time, error) {
  309. if Config.defender == nil {
  310. return nil, nil
  311. }
  312. return Config.defender.GetBanTime(ip)
  313. }
  314. // GetDefenderHosts returns hosts that are banned or for which some violations have been detected
  315. func GetDefenderHosts() ([]dataprovider.DefenderEntry, error) {
  316. if Config.defender == nil {
  317. return nil, nil
  318. }
  319. return Config.defender.GetHosts()
  320. }
  321. // GetDefenderHost returns a defender host by ip, if any
  322. func GetDefenderHost(ip string) (dataprovider.DefenderEntry, error) {
  323. if Config.defender == nil {
  324. return dataprovider.DefenderEntry{}, errors.New("defender is disabled")
  325. }
  326. return Config.defender.GetHost(ip)
  327. }
  328. // DeleteDefenderHost removes the specified IP address from the defender lists
  329. func DeleteDefenderHost(ip string) bool {
  330. if Config.defender == nil {
  331. return false
  332. }
  333. return Config.defender.DeleteHost(ip)
  334. }
  335. // GetDefenderScore returns the score for the given IP
  336. func GetDefenderScore(ip string) (int, error) {
  337. if Config.defender == nil {
  338. return 0, nil
  339. }
  340. return Config.defender.GetScore(ip)
  341. }
  342. // AddDefenderEvent adds the specified defender event for the given IP
  343. func AddDefenderEvent(ip string, event HostEvent) {
  344. if Config.defender == nil {
  345. return
  346. }
  347. Config.defender.AddEvent(ip, event)
  348. }
  349. func startPeriodicChecks(duration time.Duration) {
  350. startEventScheduler()
  351. spec := fmt.Sprintf("@every %s", duration)
  352. _, err := eventScheduler.AddFunc(spec, Connections.checkTransfers)
  353. util.PanicOnError(err)
  354. logger.Info(logSender, "", "scheduled overquota transfers check, schedule %q", spec)
  355. if Config.IdleTimeout > 0 {
  356. ratio := idleTimeoutCheckInterval / periodicTimeoutCheckInterval
  357. spec = fmt.Sprintf("@every %s", duration*ratio)
  358. _, err = eventScheduler.AddFunc(spec, Connections.checkIdles)
  359. util.PanicOnError(err)
  360. logger.Info(logSender, "", "scheduled idle connections check, schedule %q", spec)
  361. }
  362. }
  363. // ActiveTransfer defines the interface for the current active transfers
  364. type ActiveTransfer interface {
  365. GetID() int64
  366. GetType() int
  367. GetSize() int64
  368. GetDownloadedSize() int64
  369. GetUploadedSize() int64
  370. GetVirtualPath() string
  371. GetStartTime() time.Time
  372. SignalClose(err error)
  373. Truncate(fsPath string, size int64) (int64, error)
  374. GetRealFsPath(fsPath string) string
  375. SetTimes(fsPath string, atime time.Time, mtime time.Time) bool
  376. GetTruncatedSize() int64
  377. HasSizeLimit() bool
  378. }
  379. // ActiveConnection defines the interface for the current active connections
  380. type ActiveConnection interface {
  381. GetID() string
  382. GetUsername() string
  383. GetRole() string
  384. GetMaxSessions() int
  385. GetLocalAddress() string
  386. GetRemoteAddress() string
  387. GetClientVersion() string
  388. GetProtocol() string
  389. GetConnectionTime() time.Time
  390. GetLastActivity() time.Time
  391. GetCommand() string
  392. Disconnect() error
  393. AddTransfer(t ActiveTransfer)
  394. RemoveTransfer(t ActiveTransfer)
  395. GetTransfers() []ConnectionTransfer
  396. SignalTransferClose(transferID int64, err error)
  397. CloseFS() error
  398. }
  399. // StatAttributes defines the attributes for set stat commands
  400. type StatAttributes struct {
  401. Mode os.FileMode
  402. Atime time.Time
  403. Mtime time.Time
  404. UID int
  405. GID int
  406. Flags int
  407. Size int64
  408. }
  409. // ConnectionTransfer defines the trasfer details to expose
  410. type ConnectionTransfer struct {
  411. ID int64 `json:"-"`
  412. OperationType string `json:"operation_type"`
  413. StartTime int64 `json:"start_time"`
  414. Size int64 `json:"size"`
  415. VirtualPath string `json:"path"`
  416. HasSizeLimit bool `json:"-"`
  417. ULSize int64 `json:"-"`
  418. DLSize int64 `json:"-"`
  419. }
  420. func (t *ConnectionTransfer) getConnectionTransferAsString() string {
  421. result := ""
  422. switch t.OperationType {
  423. case operationUpload:
  424. result += "UL "
  425. case operationDownload:
  426. result += "DL "
  427. }
  428. result += fmt.Sprintf("%q ", t.VirtualPath)
  429. if t.Size > 0 {
  430. elapsed := time.Since(util.GetTimeFromMsecSinceEpoch(t.StartTime))
  431. speed := float64(t.Size) / float64(util.GetTimeAsMsSinceEpoch(time.Now())-t.StartTime)
  432. result += fmt.Sprintf("Size: %s Elapsed: %s Speed: \"%.1f KB/s\"", util.ByteCountIEC(t.Size),
  433. util.GetDurationAsString(elapsed), speed)
  434. }
  435. return result
  436. }
  437. type whitelist struct {
  438. fileName string
  439. sync.RWMutex
  440. list HostList
  441. }
  442. func (l *whitelist) reload() error {
  443. list, err := loadHostListFromFile(l.fileName)
  444. if err != nil {
  445. return err
  446. }
  447. if list == nil {
  448. return errors.New("cannot accept a nil whitelist")
  449. }
  450. l.Lock()
  451. defer l.Unlock()
  452. l.list = *list
  453. return nil
  454. }
  455. func (l *whitelist) isAllowed(ip string) bool {
  456. l.RLock()
  457. defer l.RUnlock()
  458. return l.list.isListed(ip)
  459. }
  460. // Configuration defines configuration parameters common to all supported protocols
  461. type Configuration struct {
  462. // Maximum idle timeout as minutes. If a client is idle for a time that exceeds this setting it will be disconnected.
  463. // 0 means disabled
  464. IdleTimeout int `json:"idle_timeout" mapstructure:"idle_timeout"`
  465. // UploadMode 0 means standard, the files are uploaded directly to the requested path.
  466. // 1 means atomic: the files are uploaded to a temporary path and renamed to the requested path
  467. // when the client ends the upload. Atomic mode avoid problems such as a web server that
  468. // serves partial files when the files are being uploaded.
  469. // In atomic mode if there is an upload error the temporary file is deleted and so the requested
  470. // upload path will not contain a partial file.
  471. // 2 means atomic with resume support: as atomic but if there is an upload error the temporary
  472. // file is renamed to the requested path and not deleted, this way a client can reconnect and resume
  473. // the upload.
  474. UploadMode int `json:"upload_mode" mapstructure:"upload_mode"`
  475. // Actions to execute for SFTP file operations and SSH commands
  476. Actions ProtocolActions `json:"actions" mapstructure:"actions"`
  477. // SetstatMode 0 means "normal mode": requests for changing permissions and owner/group are executed.
  478. // 1 means "ignore mode": requests for changing permissions and owner/group are silently ignored.
  479. // 2 means "ignore mode for cloud fs": requests for changing permissions and owner/group are
  480. // silently ignored for cloud based filesystem such as S3, GCS, Azure Blob. Requests for changing
  481. // modification times are ignored for cloud based filesystem if they are not supported.
  482. SetstatMode int `json:"setstat_mode" mapstructure:"setstat_mode"`
  483. // RenameMode defines how to handle directory renames. By default, renaming of non-empty directories
  484. // is not allowed for cloud storage providers (S3, GCS, Azure Blob). Set to 1 to enable recursive
  485. // renames for these providers, they may be slow, there is no atomic rename API like for local
  486. // filesystem, so SFTPGo will recursively list the directory contents and do a rename for each entry
  487. RenameMode int `json:"rename_mode" mapstructure:"rename_mode"`
  488. // TempPath defines the path for temporary files such as those used for atomic uploads or file pipes.
  489. // If you set this option you must make sure that the defined path exists, is accessible for writing
  490. // by the user running SFTPGo, and is on the same filesystem as the users home directories otherwise
  491. // the renaming for atomic uploads will become a copy and therefore may take a long time.
  492. // The temporary files are not namespaced. The default is generally fine. Leave empty for the default.
  493. TempPath string `json:"temp_path" mapstructure:"temp_path"`
  494. // Support for HAProxy PROXY protocol.
  495. // If you are running SFTPGo behind a proxy server such as HAProxy, AWS ELB or NGNIX, you can enable
  496. // the proxy protocol. It provides a convenient way to safely transport connection information
  497. // such as a client's address across multiple layers of NAT or TCP proxies to get the real
  498. // client IP address instead of the proxy IP. Both protocol versions 1 and 2 are supported.
  499. // - 0 means disabled
  500. // - 1 means proxy protocol enabled. Proxy header will be used and requests without proxy header will be accepted.
  501. // - 2 means proxy protocol required. Proxy header will be used and requests without proxy header will be rejected.
  502. // If the proxy protocol is enabled in SFTPGo then you have to enable the protocol in your proxy configuration too,
  503. // for example for HAProxy add "send-proxy" or "send-proxy-v2" to each server configuration line.
  504. ProxyProtocol int `json:"proxy_protocol" mapstructure:"proxy_protocol"`
  505. // List of IP addresses and IP ranges allowed to send the proxy header.
  506. // If proxy protocol is set to 1 and we receive a proxy header from an IP that is not in the list then the
  507. // connection will be accepted and the header will be ignored.
  508. // If proxy protocol is set to 2 and we receive a proxy header from an IP that is not in the list then the
  509. // connection will be rejected.
  510. ProxyAllowed []string `json:"proxy_allowed" mapstructure:"proxy_allowed"`
  511. // Absolute path to an external program or an HTTP URL to invoke as soon as SFTPGo starts.
  512. // If you define an HTTP URL it will be invoked using a `GET` request.
  513. // Please note that SFTPGo services may not yet be available when this hook is run.
  514. // Leave empty do disable.
  515. StartupHook string `json:"startup_hook" mapstructure:"startup_hook"`
  516. // Absolute path to an external program or an HTTP URL to invoke after a user connects
  517. // and before he tries to login. It allows you to reject the connection based on the source
  518. // ip address. Leave empty do disable.
  519. PostConnectHook string `json:"post_connect_hook" mapstructure:"post_connect_hook"`
  520. // Absolute path to an external program or an HTTP URL to invoke after an SSH/FTP connection ends.
  521. // Leave empty do disable.
  522. PostDisconnectHook string `json:"post_disconnect_hook" mapstructure:"post_disconnect_hook"`
  523. // Absolute path to an external program or an HTTP URL to invoke after a data retention check completes.
  524. // Leave empty do disable.
  525. DataRetentionHook string `json:"data_retention_hook" mapstructure:"data_retention_hook"`
  526. // Maximum number of concurrent client connections. 0 means unlimited
  527. MaxTotalConnections int `json:"max_total_connections" mapstructure:"max_total_connections"`
  528. // Maximum number of concurrent client connections from the same host (IP). 0 means unlimited
  529. MaxPerHostConnections int `json:"max_per_host_connections" mapstructure:"max_per_host_connections"`
  530. // Path to a file containing a list of IP addresses and/or networks to allow.
  531. // Only the listed IPs/networks can access the configured services, all other client connections
  532. // will be dropped before they even try to authenticate.
  533. WhiteListFile string `json:"whitelist_file" mapstructure:"whitelist_file"`
  534. // Allow users on this instance to use other users/virtual folders on this instance as storage backend.
  535. // Enable this setting if you know what you are doing.
  536. AllowSelfConnections int `json:"allow_self_connections" mapstructure:"allow_self_connections"`
  537. // Defender configuration
  538. DefenderConfig DefenderConfig `json:"defender" mapstructure:"defender"`
  539. // Rate limiter configurations
  540. RateLimitersConfig []RateLimiterConfig `json:"rate_limiters" mapstructure:"rate_limiters"`
  541. idleTimeoutAsDuration time.Duration
  542. idleLoginTimeout time.Duration
  543. defender Defender
  544. whitelist *whitelist
  545. }
  546. // IsAtomicUploadEnabled returns true if atomic upload is enabled
  547. func (c *Configuration) IsAtomicUploadEnabled() bool {
  548. return c.UploadMode == UploadModeAtomic || c.UploadMode == UploadModeAtomicWithResume
  549. }
  550. // GetProxyListener returns a wrapper for the given listener that supports the
  551. // HAProxy Proxy Protocol
  552. func (c *Configuration) GetProxyListener(listener net.Listener) (*proxyproto.Listener, error) {
  553. var err error
  554. if c.ProxyProtocol > 0 {
  555. var policyFunc func(upstream net.Addr) (proxyproto.Policy, error)
  556. if c.ProxyProtocol == 1 && len(c.ProxyAllowed) > 0 {
  557. policyFunc, err = proxyproto.LaxWhiteListPolicy(c.ProxyAllowed)
  558. if err != nil {
  559. return nil, err
  560. }
  561. }
  562. if c.ProxyProtocol == 2 {
  563. if len(c.ProxyAllowed) == 0 {
  564. policyFunc = func(upstream net.Addr) (proxyproto.Policy, error) {
  565. return proxyproto.REQUIRE, nil
  566. }
  567. } else {
  568. policyFunc, err = proxyproto.StrictWhiteListPolicy(c.ProxyAllowed)
  569. if err != nil {
  570. return nil, err
  571. }
  572. }
  573. }
  574. return &proxyproto.Listener{
  575. Listener: listener,
  576. Policy: policyFunc,
  577. ReadHeaderTimeout: 10 * time.Second,
  578. }, nil
  579. }
  580. return nil, errors.New("proxy protocol not configured")
  581. }
  582. // ExecuteStartupHook runs the startup hook if defined
  583. func (c *Configuration) ExecuteStartupHook() error {
  584. if c.StartupHook == "" {
  585. return nil
  586. }
  587. if strings.HasPrefix(c.StartupHook, "http") {
  588. var url *url.URL
  589. url, err := url.Parse(c.StartupHook)
  590. if err != nil {
  591. logger.Warn(logSender, "", "Invalid startup hook %#v: %v", c.StartupHook, err)
  592. return err
  593. }
  594. startTime := time.Now()
  595. resp, err := httpclient.RetryableGet(url.String())
  596. if err != nil {
  597. logger.Warn(logSender, "", "Error executing startup hook: %v", err)
  598. return err
  599. }
  600. defer resp.Body.Close()
  601. logger.Debug(logSender, "", "Startup hook executed, elapsed: %v, response code: %v", time.Since(startTime), resp.StatusCode)
  602. return nil
  603. }
  604. if !filepath.IsAbs(c.StartupHook) {
  605. err := fmt.Errorf("invalid startup hook %#v", c.StartupHook)
  606. logger.Warn(logSender, "", "Invalid startup hook %#v", c.StartupHook)
  607. return err
  608. }
  609. startTime := time.Now()
  610. timeout, env, args := command.GetConfig(c.StartupHook, command.HookStartup)
  611. ctx, cancel := context.WithTimeout(context.Background(), timeout)
  612. defer cancel()
  613. cmd := exec.CommandContext(ctx, c.StartupHook, args...)
  614. cmd.Env = env
  615. err := cmd.Run()
  616. logger.Debug(logSender, "", "Startup hook executed, elapsed: %v, error: %v", time.Since(startTime), err)
  617. return nil
  618. }
  619. func (c *Configuration) executePostDisconnectHook(remoteAddr, protocol, username, connID string, connectionTime time.Time) {
  620. startNewHook()
  621. defer hookEnded()
  622. ipAddr := util.GetIPFromRemoteAddress(remoteAddr)
  623. connDuration := int64(time.Since(connectionTime) / time.Millisecond)
  624. if strings.HasPrefix(c.PostDisconnectHook, "http") {
  625. var url *url.URL
  626. url, err := url.Parse(c.PostDisconnectHook)
  627. if err != nil {
  628. logger.Warn(protocol, connID, "Invalid post disconnect hook %#v: %v", c.PostDisconnectHook, err)
  629. return
  630. }
  631. q := url.Query()
  632. q.Add("ip", ipAddr)
  633. q.Add("protocol", protocol)
  634. q.Add("username", username)
  635. q.Add("connection_duration", strconv.FormatInt(connDuration, 10))
  636. url.RawQuery = q.Encode()
  637. startTime := time.Now()
  638. resp, err := httpclient.RetryableGet(url.String())
  639. respCode := 0
  640. if err == nil {
  641. respCode = resp.StatusCode
  642. resp.Body.Close()
  643. }
  644. logger.Debug(protocol, connID, "Post disconnect hook response code: %v, elapsed: %v, err: %v",
  645. respCode, time.Since(startTime), err)
  646. return
  647. }
  648. if !filepath.IsAbs(c.PostDisconnectHook) {
  649. logger.Debug(protocol, connID, "invalid post disconnect hook %#v", c.PostDisconnectHook)
  650. return
  651. }
  652. timeout, env, args := command.GetConfig(c.PostDisconnectHook, command.HookPostDisconnect)
  653. ctx, cancel := context.WithTimeout(context.Background(), timeout)
  654. defer cancel()
  655. startTime := time.Now()
  656. cmd := exec.CommandContext(ctx, c.PostDisconnectHook, args...)
  657. cmd.Env = append(env,
  658. fmt.Sprintf("SFTPGO_CONNECTION_IP=%v", ipAddr),
  659. fmt.Sprintf("SFTPGO_CONNECTION_USERNAME=%v", username),
  660. fmt.Sprintf("SFTPGO_CONNECTION_DURATION=%v", connDuration),
  661. fmt.Sprintf("SFTPGO_CONNECTION_PROTOCOL=%v", protocol))
  662. err := cmd.Run()
  663. logger.Debug(protocol, connID, "Post disconnect hook executed, elapsed: %v error: %v", time.Since(startTime), err)
  664. }
  665. func (c *Configuration) checkPostDisconnectHook(remoteAddr, protocol, username, connID string, connectionTime time.Time) {
  666. if c.PostDisconnectHook == "" {
  667. return
  668. }
  669. if !util.Contains(disconnHookProtocols, protocol) {
  670. return
  671. }
  672. go c.executePostDisconnectHook(remoteAddr, protocol, username, connID, connectionTime)
  673. }
  674. // ExecutePostConnectHook executes the post connect hook if defined
  675. func (c *Configuration) ExecutePostConnectHook(ipAddr, protocol string) error {
  676. if c.PostConnectHook == "" {
  677. return nil
  678. }
  679. if strings.HasPrefix(c.PostConnectHook, "http") {
  680. var url *url.URL
  681. url, err := url.Parse(c.PostConnectHook)
  682. if err != nil {
  683. logger.Warn(protocol, "", "Login from ip %#v denied, invalid post connect hook %#v: %v",
  684. ipAddr, c.PostConnectHook, err)
  685. return err
  686. }
  687. q := url.Query()
  688. q.Add("ip", ipAddr)
  689. q.Add("protocol", protocol)
  690. url.RawQuery = q.Encode()
  691. resp, err := httpclient.RetryableGet(url.String())
  692. if err != nil {
  693. logger.Warn(protocol, "", "Login from ip %#v denied, error executing post connect hook: %v", ipAddr, err)
  694. return err
  695. }
  696. defer resp.Body.Close()
  697. if resp.StatusCode != http.StatusOK {
  698. logger.Warn(protocol, "", "Login from ip %#v denied, post connect hook response code: %v", ipAddr, resp.StatusCode)
  699. return errUnexpectedHTTResponse
  700. }
  701. return nil
  702. }
  703. if !filepath.IsAbs(c.PostConnectHook) {
  704. err := fmt.Errorf("invalid post connect hook %#v", c.PostConnectHook)
  705. logger.Warn(protocol, "", "Login from ip %#v denied: %v", ipAddr, err)
  706. return err
  707. }
  708. timeout, env, args := command.GetConfig(c.PostConnectHook, command.HookPostConnect)
  709. ctx, cancel := context.WithTimeout(context.Background(), timeout)
  710. defer cancel()
  711. cmd := exec.CommandContext(ctx, c.PostConnectHook, args...)
  712. cmd.Env = append(env,
  713. fmt.Sprintf("SFTPGO_CONNECTION_IP=%v", ipAddr),
  714. fmt.Sprintf("SFTPGO_CONNECTION_PROTOCOL=%v", protocol))
  715. err := cmd.Run()
  716. if err != nil {
  717. logger.Warn(protocol, "", "Login from ip %#v denied, connect hook error: %v", ipAddr, err)
  718. }
  719. return err
  720. }
  721. // SSHConnection defines an ssh connection.
  722. // Each SSH connection can open several channels for SFTP or SSH commands
  723. type SSHConnection struct {
  724. id string
  725. conn net.Conn
  726. lastActivity atomic.Int64
  727. }
  728. // NewSSHConnection returns a new SSHConnection
  729. func NewSSHConnection(id string, conn net.Conn) *SSHConnection {
  730. c := &SSHConnection{
  731. id: id,
  732. conn: conn,
  733. }
  734. c.lastActivity.Store(time.Now().UnixNano())
  735. return c
  736. }
  737. // GetID returns the ID for this SSHConnection
  738. func (c *SSHConnection) GetID() string {
  739. return c.id
  740. }
  741. // UpdateLastActivity updates last activity for this connection
  742. func (c *SSHConnection) UpdateLastActivity() {
  743. c.lastActivity.Store(time.Now().UnixNano())
  744. }
  745. // GetLastActivity returns the last connection activity
  746. func (c *SSHConnection) GetLastActivity() time.Time {
  747. return time.Unix(0, c.lastActivity.Load())
  748. }
  749. // Close closes the underlying network connection
  750. func (c *SSHConnection) Close() error {
  751. return c.conn.Close()
  752. }
  753. // ActiveConnections holds the currect active connections with the associated transfers
  754. type ActiveConnections struct {
  755. // clients contains both authenticated and estabilished connections and the ones waiting
  756. // for authentication
  757. clients clientsMap
  758. transfersCheckStatus atomic.Bool
  759. sync.RWMutex
  760. connections []ActiveConnection
  761. mapping map[string]int
  762. sshConnections []*SSHConnection
  763. sshMapping map[string]int
  764. perUserConns map[string]int
  765. }
  766. // internal method, must be called within a locked block
  767. func (conns *ActiveConnections) addUserConnection(username string) {
  768. if username == "" {
  769. return
  770. }
  771. conns.perUserConns[username]++
  772. }
  773. // internal method, must be called within a locked block
  774. func (conns *ActiveConnections) removeUserConnection(username string) {
  775. if username == "" {
  776. return
  777. }
  778. if val, ok := conns.perUserConns[username]; ok {
  779. conns.perUserConns[username]--
  780. if val > 1 {
  781. return
  782. }
  783. delete(conns.perUserConns, username)
  784. }
  785. }
  786. // GetActiveSessions returns the number of active sessions for the given username.
  787. // We return the open sessions for any protocol
  788. func (conns *ActiveConnections) GetActiveSessions(username string) int {
  789. conns.RLock()
  790. defer conns.RUnlock()
  791. return conns.perUserConns[username]
  792. }
  793. // Add adds a new connection to the active ones
  794. func (conns *ActiveConnections) Add(c ActiveConnection) error {
  795. conns.Lock()
  796. defer conns.Unlock()
  797. if username := c.GetUsername(); username != "" {
  798. if maxSessions := c.GetMaxSessions(); maxSessions > 0 {
  799. if val := conns.perUserConns[username]; val >= maxSessions {
  800. return fmt.Errorf("too many open sessions: %d/%d", val, maxSessions)
  801. }
  802. }
  803. conns.addUserConnection(username)
  804. }
  805. conns.mapping[c.GetID()] = len(conns.connections)
  806. conns.connections = append(conns.connections, c)
  807. metric.UpdateActiveConnectionsSize(len(conns.connections))
  808. logger.Debug(c.GetProtocol(), c.GetID(), "connection added, local address %q, remote address %q, num open connections: %d",
  809. c.GetLocalAddress(), c.GetRemoteAddress(), len(conns.connections))
  810. return nil
  811. }
  812. // Swap replaces an existing connection with the given one.
  813. // This method is useful if you have to change some connection details
  814. // for example for FTP is used to update the connection once the user
  815. // authenticates
  816. func (conns *ActiveConnections) Swap(c ActiveConnection) error {
  817. conns.Lock()
  818. defer conns.Unlock()
  819. if idx, ok := conns.mapping[c.GetID()]; ok {
  820. conn := conns.connections[idx]
  821. conns.removeUserConnection(conn.GetUsername())
  822. if username := c.GetUsername(); username != "" {
  823. if maxSessions := c.GetMaxSessions(); maxSessions > 0 {
  824. if val, ok := conns.perUserConns[username]; ok && val >= maxSessions {
  825. conns.addUserConnection(conn.GetUsername())
  826. return fmt.Errorf("too many open sessions: %d/%d", val, maxSessions)
  827. }
  828. }
  829. conns.addUserConnection(username)
  830. }
  831. err := conn.CloseFS()
  832. conns.connections[idx] = c
  833. logger.Debug(logSender, c.GetID(), "connection swapped, close fs error: %v", err)
  834. conn = nil
  835. return nil
  836. }
  837. return errors.New("connection to swap not found")
  838. }
  839. // Remove removes a connection from the active ones
  840. func (conns *ActiveConnections) Remove(connectionID string) {
  841. conns.Lock()
  842. defer conns.Unlock()
  843. if idx, ok := conns.mapping[connectionID]; ok {
  844. conn := conns.connections[idx]
  845. err := conn.CloseFS()
  846. lastIdx := len(conns.connections) - 1
  847. conns.connections[idx] = conns.connections[lastIdx]
  848. conns.connections[lastIdx] = nil
  849. conns.connections = conns.connections[:lastIdx]
  850. delete(conns.mapping, connectionID)
  851. if idx != lastIdx {
  852. conns.mapping[conns.connections[idx].GetID()] = idx
  853. }
  854. conns.removeUserConnection(conn.GetUsername())
  855. metric.UpdateActiveConnectionsSize(lastIdx)
  856. logger.Debug(conn.GetProtocol(), conn.GetID(), "connection removed, local address %#v, remote address %#v close fs error: %v, num open connections: %v",
  857. conn.GetLocalAddress(), conn.GetRemoteAddress(), err, lastIdx)
  858. if conn.GetProtocol() == ProtocolFTP && conn.GetUsername() == "" {
  859. ip := util.GetIPFromRemoteAddress(conn.GetRemoteAddress())
  860. logger.ConnectionFailedLog("", ip, dataprovider.LoginMethodNoAuthTryed, conn.GetProtocol(),
  861. dataprovider.ErrNoAuthTryed.Error())
  862. metric.AddNoAuthTryed()
  863. AddDefenderEvent(ip, HostEventNoLoginTried)
  864. dataprovider.ExecutePostLoginHook(&dataprovider.User{}, dataprovider.LoginMethodNoAuthTryed, ip,
  865. conn.GetProtocol(), dataprovider.ErrNoAuthTryed)
  866. }
  867. Config.checkPostDisconnectHook(conn.GetRemoteAddress(), conn.GetProtocol(), conn.GetUsername(),
  868. conn.GetID(), conn.GetConnectionTime())
  869. return
  870. }
  871. logger.Warn(logSender, "", "connection id %q to remove not found!", connectionID)
  872. }
  873. // Close closes an active connection.
  874. // It returns true on success
  875. func (conns *ActiveConnections) Close(connectionID, role string) bool {
  876. conns.RLock()
  877. var result bool
  878. if idx, ok := conns.mapping[connectionID]; ok {
  879. c := conns.connections[idx]
  880. if role == "" || c.GetRole() == role {
  881. defer func(conn ActiveConnection) {
  882. err := conn.Disconnect()
  883. logger.Debug(conn.GetProtocol(), conn.GetID(), "close connection requested, close err: %v", err)
  884. }(c)
  885. result = true
  886. }
  887. }
  888. conns.RUnlock()
  889. return result
  890. }
  891. // AddSSHConnection adds a new ssh connection to the active ones
  892. func (conns *ActiveConnections) AddSSHConnection(c *SSHConnection) {
  893. conns.Lock()
  894. defer conns.Unlock()
  895. conns.sshMapping[c.GetID()] = len(conns.sshConnections)
  896. conns.sshConnections = append(conns.sshConnections, c)
  897. logger.Debug(logSender, c.GetID(), "ssh connection added, num open connections: %d", len(conns.sshConnections))
  898. }
  899. // RemoveSSHConnection removes a connection from the active ones
  900. func (conns *ActiveConnections) RemoveSSHConnection(connectionID string) {
  901. conns.Lock()
  902. defer conns.Unlock()
  903. if idx, ok := conns.sshMapping[connectionID]; ok {
  904. lastIdx := len(conns.sshConnections) - 1
  905. conns.sshConnections[idx] = conns.sshConnections[lastIdx]
  906. conns.sshConnections[lastIdx] = nil
  907. conns.sshConnections = conns.sshConnections[:lastIdx]
  908. delete(conns.sshMapping, connectionID)
  909. if idx != lastIdx {
  910. conns.sshMapping[conns.sshConnections[idx].GetID()] = idx
  911. }
  912. logger.Debug(logSender, connectionID, "ssh connection removed, num open ssh connections: %d", lastIdx)
  913. return
  914. }
  915. logger.Warn(logSender, "", "ssh connection to remove with id %q not found!", connectionID)
  916. }
  917. func (conns *ActiveConnections) checkIdles() {
  918. conns.RLock()
  919. for _, sshConn := range conns.sshConnections {
  920. idleTime := time.Since(sshConn.GetLastActivity())
  921. if idleTime > Config.idleTimeoutAsDuration {
  922. // we close an SSH connection if it has no active connections associated
  923. idToMatch := fmt.Sprintf("_%s_", sshConn.GetID())
  924. toClose := true
  925. for _, conn := range conns.connections {
  926. if strings.Contains(conn.GetID(), idToMatch) {
  927. if time.Since(conn.GetLastActivity()) <= Config.idleTimeoutAsDuration {
  928. toClose = false
  929. break
  930. }
  931. }
  932. }
  933. if toClose {
  934. defer func(c *SSHConnection) {
  935. err := c.Close()
  936. logger.Debug(logSender, c.GetID(), "close idle SSH connection, idle time: %v, close err: %v",
  937. time.Since(c.GetLastActivity()), err)
  938. }(sshConn)
  939. }
  940. }
  941. }
  942. for _, c := range conns.connections {
  943. idleTime := time.Since(c.GetLastActivity())
  944. isUnauthenticatedFTPUser := (c.GetProtocol() == ProtocolFTP && c.GetUsername() == "")
  945. if idleTime > Config.idleTimeoutAsDuration || (isUnauthenticatedFTPUser && idleTime > Config.idleLoginTimeout) {
  946. defer func(conn ActiveConnection) {
  947. err := conn.Disconnect()
  948. logger.Debug(conn.GetProtocol(), conn.GetID(), "close idle connection, idle time: %v, username: %#v close err: %v",
  949. time.Since(conn.GetLastActivity()), conn.GetUsername(), err)
  950. }(c)
  951. }
  952. }
  953. conns.RUnlock()
  954. }
  955. func (conns *ActiveConnections) checkTransfers() {
  956. if conns.transfersCheckStatus.Load() {
  957. logger.Warn(logSender, "", "the previous transfer check is still running, skipping execution")
  958. return
  959. }
  960. conns.transfersCheckStatus.Store(true)
  961. defer conns.transfersCheckStatus.Store(false)
  962. conns.RLock()
  963. if len(conns.connections) < 2 {
  964. conns.RUnlock()
  965. return
  966. }
  967. var wg sync.WaitGroup
  968. logger.Debug(logSender, "", "start concurrent transfers check")
  969. // update the current size for transfers to monitors
  970. for _, c := range conns.connections {
  971. for _, t := range c.GetTransfers() {
  972. if t.HasSizeLimit {
  973. wg.Add(1)
  974. go func(transfer ConnectionTransfer, connID string) {
  975. defer wg.Done()
  976. transfersChecker.UpdateTransferCurrentSizes(transfer.ULSize, transfer.DLSize, transfer.ID, connID)
  977. }(t, c.GetID())
  978. }
  979. }
  980. }
  981. conns.RUnlock()
  982. logger.Debug(logSender, "", "waiting for the update of the transfers current size")
  983. wg.Wait()
  984. logger.Debug(logSender, "", "getting overquota transfers")
  985. overquotaTransfers := transfersChecker.GetOverquotaTransfers()
  986. logger.Debug(logSender, "", "number of overquota transfers: %v", len(overquotaTransfers))
  987. if len(overquotaTransfers) == 0 {
  988. return
  989. }
  990. conns.RLock()
  991. defer conns.RUnlock()
  992. for _, c := range conns.connections {
  993. for _, overquotaTransfer := range overquotaTransfers {
  994. if c.GetID() == overquotaTransfer.ConnID {
  995. logger.Info(logSender, c.GetID(), "user %#v is overquota, try to close transfer id %v",
  996. c.GetUsername(), overquotaTransfer.TransferID)
  997. var err error
  998. if overquotaTransfer.TransferType == TransferDownload {
  999. err = getReadQuotaExceededError(c.GetProtocol())
  1000. } else {
  1001. err = getQuotaExceededError(c.GetProtocol())
  1002. }
  1003. c.SignalTransferClose(overquotaTransfer.TransferID, err)
  1004. }
  1005. }
  1006. }
  1007. logger.Debug(logSender, "", "transfers check completed")
  1008. }
  1009. // AddClientConnection stores a new client connection
  1010. func (conns *ActiveConnections) AddClientConnection(ipAddr string) {
  1011. conns.clients.add(ipAddr)
  1012. }
  1013. // RemoveClientConnection removes a disconnected client from the tracked ones
  1014. func (conns *ActiveConnections) RemoveClientConnection(ipAddr string) {
  1015. conns.clients.remove(ipAddr)
  1016. }
  1017. // GetClientConnections returns the total number of client connections
  1018. func (conns *ActiveConnections) GetClientConnections() int32 {
  1019. return conns.clients.getTotal()
  1020. }
  1021. // IsNewConnectionAllowed returns an error if the maximum number of concurrent allowed
  1022. // connections is exceeded or a whitelist is defined and the specified ipAddr is not listed
  1023. // or the service is shutting down
  1024. func (conns *ActiveConnections) IsNewConnectionAllowed(ipAddr string) error {
  1025. if isShuttingDown.Load() {
  1026. return ErrShuttingDown
  1027. }
  1028. if Config.whitelist != nil {
  1029. if !Config.whitelist.isAllowed(ipAddr) {
  1030. return ErrConnectionDenied
  1031. }
  1032. }
  1033. if Config.MaxTotalConnections == 0 && Config.MaxPerHostConnections == 0 {
  1034. return nil
  1035. }
  1036. if Config.MaxPerHostConnections > 0 {
  1037. if total := conns.clients.getTotalFrom(ipAddr); total > Config.MaxPerHostConnections {
  1038. logger.Info(logSender, "", "active connections from %s %d/%d", ipAddr, total, Config.MaxPerHostConnections)
  1039. AddDefenderEvent(ipAddr, HostEventLimitExceeded)
  1040. return ErrConnectionDenied
  1041. }
  1042. }
  1043. if Config.MaxTotalConnections > 0 {
  1044. if total := conns.clients.getTotal(); total > int32(Config.MaxTotalConnections) {
  1045. logger.Info(logSender, "", "active client connections %d/%d", total, Config.MaxTotalConnections)
  1046. return ErrConnectionDenied
  1047. }
  1048. // on a single SFTP connection we could have multiple SFTP channels or commands
  1049. // so we check the estabilished connections too
  1050. conns.RLock()
  1051. defer conns.RUnlock()
  1052. if sess := len(conns.connections); sess >= Config.MaxTotalConnections {
  1053. logger.Info(logSender, "", "active client sessions %d/%d", sess, Config.MaxTotalConnections)
  1054. return ErrConnectionDenied
  1055. }
  1056. }
  1057. return nil
  1058. }
  1059. // GetStats returns stats for active connections
  1060. func (conns *ActiveConnections) GetStats(role string) []ConnectionStatus {
  1061. conns.RLock()
  1062. defer conns.RUnlock()
  1063. stats := make([]ConnectionStatus, 0, len(conns.connections))
  1064. node := dataprovider.GetNodeName()
  1065. for _, c := range conns.connections {
  1066. if role == "" || c.GetRole() == role {
  1067. stat := ConnectionStatus{
  1068. Username: c.GetUsername(),
  1069. ConnectionID: c.GetID(),
  1070. ClientVersion: c.GetClientVersion(),
  1071. RemoteAddress: c.GetRemoteAddress(),
  1072. ConnectionTime: util.GetTimeAsMsSinceEpoch(c.GetConnectionTime()),
  1073. LastActivity: util.GetTimeAsMsSinceEpoch(c.GetLastActivity()),
  1074. Protocol: c.GetProtocol(),
  1075. Command: c.GetCommand(),
  1076. Transfers: c.GetTransfers(),
  1077. Node: node,
  1078. }
  1079. stats = append(stats, stat)
  1080. }
  1081. }
  1082. return stats
  1083. }
  1084. // ConnectionStatus returns the status for an active connection
  1085. type ConnectionStatus struct {
  1086. // Logged in username
  1087. Username string `json:"username"`
  1088. // Unique identifier for the connection
  1089. ConnectionID string `json:"connection_id"`
  1090. // client's version string
  1091. ClientVersion string `json:"client_version,omitempty"`
  1092. // Remote address for this connection
  1093. RemoteAddress string `json:"remote_address"`
  1094. // Connection time as unix timestamp in milliseconds
  1095. ConnectionTime int64 `json:"connection_time"`
  1096. // Last activity as unix timestamp in milliseconds
  1097. LastActivity int64 `json:"last_activity"`
  1098. // Protocol for this connection
  1099. Protocol string `json:"protocol"`
  1100. // active uploads/downloads
  1101. Transfers []ConnectionTransfer `json:"active_transfers,omitempty"`
  1102. // SSH command or WebDAV method
  1103. Command string `json:"command,omitempty"`
  1104. // Node identifier, omitted for single node installations
  1105. Node string `json:"node,omitempty"`
  1106. }
  1107. // GetConnectionDuration returns the connection duration as string
  1108. func (c *ConnectionStatus) GetConnectionDuration() string {
  1109. elapsed := time.Since(util.GetTimeFromMsecSinceEpoch(c.ConnectionTime))
  1110. return util.GetDurationAsString(elapsed)
  1111. }
  1112. // GetConnectionInfo returns connection info.
  1113. // Protocol,Client Version and RemoteAddress are returned.
  1114. func (c *ConnectionStatus) GetConnectionInfo() string {
  1115. var result strings.Builder
  1116. result.WriteString(fmt.Sprintf("%v. Client: %#v From: %#v", c.Protocol, c.ClientVersion, c.RemoteAddress))
  1117. if c.Command == "" {
  1118. return result.String()
  1119. }
  1120. switch c.Protocol {
  1121. case ProtocolSSH, ProtocolFTP:
  1122. result.WriteString(fmt.Sprintf(". Command: %#v", c.Command))
  1123. case ProtocolWebDAV:
  1124. result.WriteString(fmt.Sprintf(". Method: %#v", c.Command))
  1125. }
  1126. return result.String()
  1127. }
  1128. // GetTransfersAsString returns the active transfers as string
  1129. func (c *ConnectionStatus) GetTransfersAsString() string {
  1130. result := ""
  1131. for _, t := range c.Transfers {
  1132. if result != "" {
  1133. result += ". "
  1134. }
  1135. result += t.getConnectionTransferAsString()
  1136. }
  1137. return result
  1138. }
  1139. // ActiveQuotaScan defines an active quota scan for a user
  1140. type ActiveQuotaScan struct {
  1141. // Username to which the quota scan refers
  1142. Username string `json:"username"`
  1143. // quota scan start time as unix timestamp in milliseconds
  1144. StartTime int64 `json:"start_time"`
  1145. Role string `json:"-"`
  1146. }
  1147. // ActiveVirtualFolderQuotaScan defines an active quota scan for a virtual folder
  1148. type ActiveVirtualFolderQuotaScan struct {
  1149. // folder name to which the quota scan refers
  1150. Name string `json:"name"`
  1151. // quota scan start time as unix timestamp in milliseconds
  1152. StartTime int64 `json:"start_time"`
  1153. }
  1154. // ActiveScans holds the active quota scans
  1155. type ActiveScans struct {
  1156. sync.RWMutex
  1157. UserScans []ActiveQuotaScan
  1158. FolderScans []ActiveVirtualFolderQuotaScan
  1159. }
  1160. // GetUsersQuotaScans returns the active users quota scans
  1161. func (s *ActiveScans) GetUsersQuotaScans(role string) []ActiveQuotaScan {
  1162. s.RLock()
  1163. defer s.RUnlock()
  1164. scans := make([]ActiveQuotaScan, 0, len(s.UserScans))
  1165. for _, scan := range s.UserScans {
  1166. if role == "" || role == scan.Role {
  1167. scans = append(scans, ActiveQuotaScan{
  1168. Username: scan.Username,
  1169. StartTime: scan.StartTime,
  1170. })
  1171. }
  1172. }
  1173. return scans
  1174. }
  1175. // AddUserQuotaScan adds a user to the ones with active quota scans.
  1176. // Returns false if the user has a quota scan already running
  1177. func (s *ActiveScans) AddUserQuotaScan(username, role string) bool {
  1178. s.Lock()
  1179. defer s.Unlock()
  1180. for _, scan := range s.UserScans {
  1181. if scan.Username == username {
  1182. return false
  1183. }
  1184. }
  1185. s.UserScans = append(s.UserScans, ActiveQuotaScan{
  1186. Username: username,
  1187. StartTime: util.GetTimeAsMsSinceEpoch(time.Now()),
  1188. Role: role,
  1189. })
  1190. return true
  1191. }
  1192. // RemoveUserQuotaScan removes a user from the ones with active quota scans.
  1193. // Returns false if the user has no active quota scans
  1194. func (s *ActiveScans) RemoveUserQuotaScan(username string) bool {
  1195. s.Lock()
  1196. defer s.Unlock()
  1197. for idx, scan := range s.UserScans {
  1198. if scan.Username == username {
  1199. lastIdx := len(s.UserScans) - 1
  1200. s.UserScans[idx] = s.UserScans[lastIdx]
  1201. s.UserScans = s.UserScans[:lastIdx]
  1202. return true
  1203. }
  1204. }
  1205. return false
  1206. }
  1207. // GetVFoldersQuotaScans returns the active quota scans for virtual folders
  1208. func (s *ActiveScans) GetVFoldersQuotaScans() []ActiveVirtualFolderQuotaScan {
  1209. s.RLock()
  1210. defer s.RUnlock()
  1211. scans := make([]ActiveVirtualFolderQuotaScan, len(s.FolderScans))
  1212. copy(scans, s.FolderScans)
  1213. return scans
  1214. }
  1215. // AddVFolderQuotaScan adds a virtual folder to the ones with active quota scans.
  1216. // Returns false if the folder has a quota scan already running
  1217. func (s *ActiveScans) AddVFolderQuotaScan(folderName string) bool {
  1218. s.Lock()
  1219. defer s.Unlock()
  1220. for _, scan := range s.FolderScans {
  1221. if scan.Name == folderName {
  1222. return false
  1223. }
  1224. }
  1225. s.FolderScans = append(s.FolderScans, ActiveVirtualFolderQuotaScan{
  1226. Name: folderName,
  1227. StartTime: util.GetTimeAsMsSinceEpoch(time.Now()),
  1228. })
  1229. return true
  1230. }
  1231. // RemoveVFolderQuotaScan removes a folder from the ones with active quota scans.
  1232. // Returns false if the folder has no active quota scans
  1233. func (s *ActiveScans) RemoveVFolderQuotaScan(folderName string) bool {
  1234. s.Lock()
  1235. defer s.Unlock()
  1236. for idx, scan := range s.FolderScans {
  1237. if scan.Name == folderName {
  1238. lastIdx := len(s.FolderScans) - 1
  1239. s.FolderScans[idx] = s.FolderScans[lastIdx]
  1240. s.FolderScans = s.FolderScans[:lastIdx]
  1241. return true
  1242. }
  1243. }
  1244. return false
  1245. }
  1246. // MetadataCheck defines an active metadata check
  1247. type MetadataCheck struct {
  1248. // Username to which the metadata check refers
  1249. Username string `json:"username"`
  1250. // check start time as unix timestamp in milliseconds
  1251. StartTime int64 `json:"start_time"`
  1252. Role string `json:"-"`
  1253. }
  1254. // MetadataChecks holds the active metadata checks
  1255. type MetadataChecks struct {
  1256. sync.RWMutex
  1257. checks []MetadataCheck
  1258. }
  1259. // Get returns the active metadata checks
  1260. func (c *MetadataChecks) Get(role string) []MetadataCheck {
  1261. c.RLock()
  1262. defer c.RUnlock()
  1263. checks := make([]MetadataCheck, 0, len(c.checks))
  1264. for _, check := range c.checks {
  1265. if role == "" || role == check.Role {
  1266. checks = append(checks, MetadataCheck{
  1267. Username: check.Username,
  1268. StartTime: check.StartTime,
  1269. })
  1270. }
  1271. }
  1272. return checks
  1273. }
  1274. // Add adds a user to the ones with active metadata checks.
  1275. // Return false if a metadata check is already active for the specified user
  1276. func (c *MetadataChecks) Add(username, role string) bool {
  1277. c.Lock()
  1278. defer c.Unlock()
  1279. for idx := range c.checks {
  1280. if c.checks[idx].Username == username {
  1281. return false
  1282. }
  1283. }
  1284. c.checks = append(c.checks, MetadataCheck{
  1285. Username: username,
  1286. StartTime: util.GetTimeAsMsSinceEpoch(time.Now()),
  1287. Role: role,
  1288. })
  1289. return true
  1290. }
  1291. // Remove removes a user from the ones with active metadata checks
  1292. func (c *MetadataChecks) Remove(username string) bool {
  1293. c.Lock()
  1294. defer c.Unlock()
  1295. for idx := range c.checks {
  1296. if c.checks[idx].Username == username {
  1297. lastIdx := len(c.checks) - 1
  1298. c.checks[idx] = c.checks[lastIdx]
  1299. c.checks = c.checks[:lastIdx]
  1300. return true
  1301. }
  1302. }
  1303. return false
  1304. }