plugin.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809
  1. // Copyright (C) 2019-2022 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 plugin provides support for the SFTPGo plugin system
  15. package plugin
  16. import (
  17. "crypto/sha256"
  18. "crypto/x509"
  19. "encoding/hex"
  20. "errors"
  21. "fmt"
  22. "sync"
  23. "sync/atomic"
  24. "time"
  25. "github.com/hashicorp/go-hclog"
  26. "github.com/hashicorp/go-plugin"
  27. "github.com/sftpgo/sdk/plugin/auth"
  28. "github.com/sftpgo/sdk/plugin/eventsearcher"
  29. "github.com/sftpgo/sdk/plugin/ipfilter"
  30. kmsplugin "github.com/sftpgo/sdk/plugin/kms"
  31. "github.com/sftpgo/sdk/plugin/metadata"
  32. "github.com/sftpgo/sdk/plugin/notifier"
  33. "github.com/drakkan/sftpgo/v2/internal/kms"
  34. "github.com/drakkan/sftpgo/v2/internal/logger"
  35. "github.com/drakkan/sftpgo/v2/internal/util"
  36. )
  37. const (
  38. logSender = "plugins"
  39. )
  40. var (
  41. // Handler defines the plugins manager
  42. Handler Manager
  43. pluginsLogLevel = hclog.Debug
  44. // ErrNoSearcher defines the error to return for events searches if no plugin is configured
  45. ErrNoSearcher = errors.New("no events searcher plugin defined")
  46. // ErrNoMetadater returns the error to return for metadata methods if no plugin is configured
  47. ErrNoMetadater = errors.New("no metadata plugin defined")
  48. )
  49. // Renderer defines the interface for generic objects rendering
  50. type Renderer interface {
  51. RenderAsJSON(reload bool) ([]byte, error)
  52. }
  53. // Config defines a plugin configuration
  54. type Config struct {
  55. // Plugin type
  56. Type string `json:"type" mapstructure:"type"`
  57. // NotifierOptions defines options for notifiers plugins
  58. NotifierOptions NotifierConfig `json:"notifier_options" mapstructure:"notifier_options"`
  59. // KMSOptions defines options for a KMS plugin
  60. KMSOptions KMSConfig `json:"kms_options" mapstructure:"kms_options"`
  61. // AuthOptions defines options for authentication plugins
  62. AuthOptions AuthConfig `json:"auth_options" mapstructure:"auth_options"`
  63. // Path to the plugin executable
  64. Cmd string `json:"cmd" mapstructure:"cmd"`
  65. // Args to pass to the plugin executable
  66. Args []string `json:"args" mapstructure:"args"`
  67. // SHA256 checksum for the plugin executable.
  68. // If not empty it will be used to verify the integrity of the executable
  69. SHA256Sum string `json:"sha256sum" mapstructure:"sha256sum"`
  70. // If enabled the client and the server automatically negotiate mTLS for
  71. // transport authentication. This ensures that only the original client will
  72. // be allowed to connect to the server, and all other connections will be
  73. // rejected. The client will also refuse to connect to any server that isn't
  74. // the original instance started by the client.
  75. AutoMTLS bool `json:"auto_mtls" mapstructure:"auto_mtls"`
  76. // unique identifier for kms plugins
  77. kmsID int
  78. }
  79. func (c *Config) getSecureConfig() (*plugin.SecureConfig, error) {
  80. if c.SHA256Sum != "" {
  81. checksum, err := hex.DecodeString(c.SHA256Sum)
  82. if err != nil {
  83. return nil, fmt.Errorf("invalid sha256 hash %q: %w", c.SHA256Sum, err)
  84. }
  85. return &plugin.SecureConfig{
  86. Checksum: checksum,
  87. Hash: sha256.New(),
  88. }, nil
  89. }
  90. return nil, nil
  91. }
  92. func (c *Config) newKMSPluginSecretProvider(base kms.BaseSecret, url, masterKey string) kms.SecretProvider {
  93. return &kmsPluginSecretProvider{
  94. BaseSecret: base,
  95. URL: url,
  96. MasterKey: masterKey,
  97. config: c,
  98. }
  99. }
  100. // Manager handles enabled plugins
  101. type Manager struct {
  102. closed atomic.Bool
  103. done chan bool
  104. // List of configured plugins
  105. Configs []Config `json:"plugins" mapstructure:"plugins"`
  106. notifLock sync.RWMutex
  107. notifiers []*notifierPlugin
  108. kmsLock sync.RWMutex
  109. kms []*kmsPlugin
  110. authLock sync.RWMutex
  111. auths []*authPlugin
  112. searcherLock sync.RWMutex
  113. searcher *searcherPlugin
  114. metadaterLock sync.RWMutex
  115. metadater *metadataPlugin
  116. ipFilterLock sync.RWMutex
  117. filter *ipFilterPlugin
  118. authScopes int
  119. hasSearcher bool
  120. hasMetadater bool
  121. hasNotifiers bool
  122. hasAuths bool
  123. hasIPFilter bool
  124. concurrencyGuard chan struct{}
  125. }
  126. // Initialize initializes the configured plugins
  127. func Initialize(configs []Config, logLevel string) error {
  128. logger.Debug(logSender, "", "initialize")
  129. Handler = Manager{
  130. Configs: configs,
  131. done: make(chan bool),
  132. authScopes: -1,
  133. concurrencyGuard: make(chan struct{}, 250),
  134. }
  135. Handler.closed.Store(false)
  136. setLogLevel(logLevel)
  137. if len(configs) == 0 {
  138. return nil
  139. }
  140. if err := Handler.validateConfigs(); err != nil {
  141. return err
  142. }
  143. if err := initializePlugins(); err != nil {
  144. return err
  145. }
  146. startCheckTicker()
  147. return nil
  148. }
  149. func initializePlugins() error {
  150. kmsID := 0
  151. for idx, config := range Handler.Configs {
  152. switch config.Type {
  153. case notifier.PluginName:
  154. plugin, err := newNotifierPlugin(config)
  155. if err != nil {
  156. return err
  157. }
  158. Handler.notifiers = append(Handler.notifiers, plugin)
  159. case kmsplugin.PluginName:
  160. plugin, err := newKMSPlugin(config)
  161. if err != nil {
  162. return err
  163. }
  164. Handler.kms = append(Handler.kms, plugin)
  165. Handler.Configs[idx].kmsID = kmsID
  166. kmsID++
  167. kms.RegisterSecretProvider(config.KMSOptions.Scheme, config.KMSOptions.EncryptedStatus,
  168. Handler.Configs[idx].newKMSPluginSecretProvider)
  169. logger.Info(logSender, "", "registered secret provider for scheme: %v, encrypted status: %v",
  170. config.KMSOptions.Scheme, config.KMSOptions.EncryptedStatus)
  171. case auth.PluginName:
  172. plugin, err := newAuthPlugin(config)
  173. if err != nil {
  174. return err
  175. }
  176. Handler.auths = append(Handler.auths, plugin)
  177. if Handler.authScopes == -1 {
  178. Handler.authScopes = config.AuthOptions.Scope
  179. } else {
  180. Handler.authScopes |= config.AuthOptions.Scope
  181. }
  182. case eventsearcher.PluginName:
  183. plugin, err := newSearcherPlugin(config)
  184. if err != nil {
  185. return err
  186. }
  187. Handler.searcher = plugin
  188. case metadata.PluginName:
  189. plugin, err := newMetadaterPlugin(config)
  190. if err != nil {
  191. return err
  192. }
  193. Handler.metadater = plugin
  194. case ipfilter.PluginName:
  195. plugin, err := newIPFilterPlugin(config)
  196. if err != nil {
  197. return err
  198. }
  199. Handler.filter = plugin
  200. default:
  201. return fmt.Errorf("unsupported plugin type: %v", config.Type)
  202. }
  203. }
  204. return nil
  205. }
  206. func (m *Manager) validateConfigs() error {
  207. kmsSchemes := make(map[string]bool)
  208. kmsEncryptions := make(map[string]bool)
  209. m.hasSearcher = false
  210. m.hasMetadater = false
  211. m.hasNotifiers = false
  212. m.hasAuths = false
  213. m.hasIPFilter = false
  214. for _, config := range m.Configs {
  215. switch config.Type {
  216. case kmsplugin.PluginName:
  217. if _, ok := kmsSchemes[config.KMSOptions.Scheme]; ok {
  218. return fmt.Errorf("invalid KMS configuration, duplicated scheme %#v", config.KMSOptions.Scheme)
  219. }
  220. if _, ok := kmsEncryptions[config.KMSOptions.EncryptedStatus]; ok {
  221. return fmt.Errorf("invalid KMS configuration, duplicated encrypted status %#v", config.KMSOptions.EncryptedStatus)
  222. }
  223. kmsSchemes[config.KMSOptions.Scheme] = true
  224. kmsEncryptions[config.KMSOptions.EncryptedStatus] = true
  225. case eventsearcher.PluginName:
  226. if m.hasSearcher {
  227. return errors.New("only one eventsearcher plugin can be defined")
  228. }
  229. m.hasSearcher = true
  230. case metadata.PluginName:
  231. if m.hasMetadater {
  232. return errors.New("only one metadata plugin can be defined")
  233. }
  234. m.hasMetadater = true
  235. case notifier.PluginName:
  236. m.hasNotifiers = true
  237. case auth.PluginName:
  238. m.hasAuths = true
  239. case ipfilter.PluginName:
  240. m.hasIPFilter = true
  241. }
  242. }
  243. return nil
  244. }
  245. // HasAuthenticators returns true if there is at least an auth plugin
  246. func (m *Manager) HasAuthenticators() bool {
  247. return m.hasAuths
  248. }
  249. // HasNotifiers returns true if there is at least a notifier plugin
  250. func (m *Manager) HasNotifiers() bool {
  251. return m.hasNotifiers
  252. }
  253. // NotifyFsEvent sends the fs event notifications using any defined notifier plugins
  254. func (m *Manager) NotifyFsEvent(event *notifier.FsEvent) {
  255. m.notifLock.RLock()
  256. defer m.notifLock.RUnlock()
  257. for _, n := range m.notifiers {
  258. n.notifyFsAction(event)
  259. }
  260. }
  261. // NotifyProviderEvent sends the provider event notifications using any defined notifier plugins
  262. func (m *Manager) NotifyProviderEvent(event *notifier.ProviderEvent, object Renderer) {
  263. m.notifLock.RLock()
  264. defer m.notifLock.RUnlock()
  265. for _, n := range m.notifiers {
  266. n.notifyProviderAction(event, object)
  267. }
  268. }
  269. // SearchFsEvents returns the filesystem events matching the specified filters
  270. func (m *Manager) SearchFsEvents(searchFilters *eventsearcher.FsEventSearch) ([]byte, []string, []string, error) {
  271. if !m.hasSearcher {
  272. return nil, nil, nil, ErrNoSearcher
  273. }
  274. m.searcherLock.RLock()
  275. plugin := m.searcher
  276. m.searcherLock.RUnlock()
  277. return plugin.searchear.SearchFsEvents(searchFilters)
  278. }
  279. // SearchProviderEvents returns the provider events matching the specified filters
  280. func (m *Manager) SearchProviderEvents(searchFilters *eventsearcher.ProviderEventSearch) ([]byte, []string, []string, error) {
  281. if !m.hasSearcher {
  282. return nil, nil, nil, ErrNoSearcher
  283. }
  284. m.searcherLock.RLock()
  285. plugin := m.searcher
  286. m.searcherLock.RUnlock()
  287. return plugin.searchear.SearchProviderEvents(searchFilters)
  288. }
  289. // HasMetadater returns true if a metadata plugin is defined
  290. func (m *Manager) HasMetadater() bool {
  291. return m.hasMetadater
  292. }
  293. // SetModificationTime sets the modification time for the specified object
  294. func (m *Manager) SetModificationTime(storageID, objectPath string, mTime int64) error {
  295. if !m.hasMetadater {
  296. return ErrNoMetadater
  297. }
  298. m.metadaterLock.RLock()
  299. plugin := m.metadater
  300. m.metadaterLock.RUnlock()
  301. return plugin.metadater.SetModificationTime(storageID, objectPath, mTime)
  302. }
  303. // GetModificationTime returns the modification time for the specified path
  304. func (m *Manager) GetModificationTime(storageID, objectPath string, isDir bool) (int64, error) {
  305. if !m.hasMetadater {
  306. return 0, ErrNoMetadater
  307. }
  308. m.metadaterLock.RLock()
  309. plugin := m.metadater
  310. m.metadaterLock.RUnlock()
  311. return plugin.metadater.GetModificationTime(storageID, objectPath)
  312. }
  313. // GetModificationTimes returns the modification times for all the files within the specified folder
  314. func (m *Manager) GetModificationTimes(storageID, objectPath string) (map[string]int64, error) {
  315. if !m.hasMetadater {
  316. return nil, ErrNoMetadater
  317. }
  318. m.metadaterLock.RLock()
  319. plugin := m.metadater
  320. m.metadaterLock.RUnlock()
  321. return plugin.metadater.GetModificationTimes(storageID, objectPath)
  322. }
  323. // RemoveMetadata deletes the metadata stored for the specified object
  324. func (m *Manager) RemoveMetadata(storageID, objectPath string) error {
  325. if !m.hasMetadater {
  326. return ErrNoMetadater
  327. }
  328. m.metadaterLock.RLock()
  329. plugin := m.metadater
  330. m.metadaterLock.RUnlock()
  331. return plugin.metadater.RemoveMetadata(storageID, objectPath)
  332. }
  333. // GetMetadataFolders returns the folders that metadata is associated with
  334. func (m *Manager) GetMetadataFolders(storageID, from string, limit int) ([]string, error) {
  335. if !m.hasMetadater {
  336. return nil, ErrNoMetadater
  337. }
  338. m.metadaterLock.RLock()
  339. plugin := m.metadater
  340. m.metadaterLock.RUnlock()
  341. return plugin.metadater.GetFolders(storageID, limit, from)
  342. }
  343. // IsIPBanned returns true if the IP filter plugin does not allow the specified ip.
  344. // If no IP filter plugin is defined this method returns false
  345. func (m *Manager) IsIPBanned(ip string) bool {
  346. if !m.hasIPFilter {
  347. return false
  348. }
  349. m.ipFilterLock.RLock()
  350. plugin := m.filter
  351. m.ipFilterLock.RUnlock()
  352. if plugin.exited() {
  353. logger.Warn(logSender, "", "ip filter plugin is not active, cannot check ip %#v", ip)
  354. return false
  355. }
  356. return plugin.filter.CheckIP(ip) != nil
  357. }
  358. // ReloadFilter sends a reload request to the IP filter plugin
  359. func (m *Manager) ReloadFilter() {
  360. if !m.hasIPFilter {
  361. return
  362. }
  363. m.ipFilterLock.RLock()
  364. plugin := m.filter
  365. m.ipFilterLock.RUnlock()
  366. if err := plugin.filter.Reload(); err != nil {
  367. logger.Error(logSender, "", "unable to reload IP filter plugin: %v", err)
  368. }
  369. }
  370. func (m *Manager) kmsEncrypt(secret kms.BaseSecret, url string, masterKey string, kmsID int) (string, string, int32, error) {
  371. m.kmsLock.RLock()
  372. plugin := m.kms[kmsID]
  373. m.kmsLock.RUnlock()
  374. return plugin.Encrypt(secret, url, masterKey)
  375. }
  376. func (m *Manager) kmsDecrypt(secret kms.BaseSecret, url string, masterKey string, kmsID int) (string, error) {
  377. m.kmsLock.RLock()
  378. plugin := m.kms[kmsID]
  379. m.kmsLock.RUnlock()
  380. return plugin.Decrypt(secret, url, masterKey)
  381. }
  382. // HasAuthScope returns true if there is an auth plugin that support the specified scope
  383. func (m *Manager) HasAuthScope(scope int) bool {
  384. if m.authScopes == -1 {
  385. return false
  386. }
  387. return m.authScopes&scope != 0
  388. }
  389. // Authenticate tries to authenticate the specified user using an external plugin
  390. func (m *Manager) Authenticate(username, password, ip, protocol string, pkey string,
  391. tlsCert *x509.Certificate, authScope int, userAsJSON []byte,
  392. ) ([]byte, error) {
  393. switch authScope {
  394. case AuthScopePassword:
  395. return m.checkUserAndPass(username, password, ip, protocol, userAsJSON)
  396. case AuthScopePublicKey:
  397. return m.checkUserAndPublicKey(username, pkey, ip, protocol, userAsJSON)
  398. case AuthScopeKeyboardInteractive:
  399. return m.checkUserAndKeyboardInteractive(username, ip, protocol, userAsJSON)
  400. case AuthScopeTLSCertificate:
  401. cert, err := util.EncodeTLSCertToPem(tlsCert)
  402. if err != nil {
  403. logger.Error(logSender, "", "unable to encode tls certificate to pem: %v", err)
  404. return nil, fmt.Errorf("unable to encode tls cert to pem: %w", err)
  405. }
  406. return m.checkUserAndTLSCert(username, cert, ip, protocol, userAsJSON)
  407. default:
  408. return nil, fmt.Errorf("unsupported auth scope: %v", authScope)
  409. }
  410. }
  411. // ExecuteKeyboardInteractiveStep executes a keyboard interactive step
  412. func (m *Manager) ExecuteKeyboardInteractiveStep(req *KeyboardAuthRequest) (*KeyboardAuthResponse, error) {
  413. var plugin *authPlugin
  414. m.authLock.Lock()
  415. for _, p := range m.auths {
  416. if p.config.AuthOptions.Scope&AuthScopePassword != 0 {
  417. plugin = p
  418. break
  419. }
  420. }
  421. m.authLock.Unlock()
  422. if plugin == nil {
  423. return nil, errors.New("no auth plugin configured for keyaboard interactive authentication step")
  424. }
  425. return plugin.sendKeyboardIteractiveRequest(req)
  426. }
  427. func (m *Manager) checkUserAndPass(username, password, ip, protocol string, userAsJSON []byte) ([]byte, error) {
  428. var plugin *authPlugin
  429. m.authLock.Lock()
  430. for _, p := range m.auths {
  431. if p.config.AuthOptions.Scope&AuthScopePassword != 0 {
  432. plugin = p
  433. break
  434. }
  435. }
  436. m.authLock.Unlock()
  437. if plugin == nil {
  438. return nil, errors.New("no auth plugin configured for password checking")
  439. }
  440. return plugin.checkUserAndPass(username, password, ip, protocol, userAsJSON)
  441. }
  442. func (m *Manager) checkUserAndPublicKey(username, pubKey, ip, protocol string, userAsJSON []byte) ([]byte, error) {
  443. var plugin *authPlugin
  444. m.authLock.Lock()
  445. for _, p := range m.auths {
  446. if p.config.AuthOptions.Scope&AuthScopePublicKey != 0 {
  447. plugin = p
  448. break
  449. }
  450. }
  451. m.authLock.Unlock()
  452. if plugin == nil {
  453. return nil, errors.New("no auth plugin configured for public key checking")
  454. }
  455. return plugin.checkUserAndPublicKey(username, pubKey, ip, protocol, userAsJSON)
  456. }
  457. func (m *Manager) checkUserAndTLSCert(username, tlsCert, ip, protocol string, userAsJSON []byte) ([]byte, error) {
  458. var plugin *authPlugin
  459. m.authLock.Lock()
  460. for _, p := range m.auths {
  461. if p.config.AuthOptions.Scope&AuthScopeTLSCertificate != 0 {
  462. plugin = p
  463. break
  464. }
  465. }
  466. m.authLock.Unlock()
  467. if plugin == nil {
  468. return nil, errors.New("no auth plugin configured for TLS certificate checking")
  469. }
  470. return plugin.checkUserAndTLSCertificate(username, tlsCert, ip, protocol, userAsJSON)
  471. }
  472. func (m *Manager) checkUserAndKeyboardInteractive(username, ip, protocol string, userAsJSON []byte) ([]byte, error) {
  473. var plugin *authPlugin
  474. m.authLock.Lock()
  475. for _, p := range m.auths {
  476. if p.config.AuthOptions.Scope&AuthScopeKeyboardInteractive != 0 {
  477. plugin = p
  478. break
  479. }
  480. }
  481. m.authLock.Unlock()
  482. if plugin == nil {
  483. return nil, errors.New("no auth plugin configured for keyboard interactive checking")
  484. }
  485. return plugin.checkUserAndKeyboardInteractive(username, ip, protocol, userAsJSON)
  486. }
  487. func (m *Manager) checkCrashedPlugins() {
  488. m.notifLock.RLock()
  489. for idx, n := range m.notifiers {
  490. if n.exited() {
  491. defer func(cfg Config, index int) {
  492. Handler.restartNotifierPlugin(cfg, index)
  493. }(n.config, idx)
  494. } else {
  495. n.sendQueuedEvents()
  496. }
  497. }
  498. m.notifLock.RUnlock()
  499. m.kmsLock.RLock()
  500. for idx, k := range m.kms {
  501. if k.exited() {
  502. defer func(cfg Config, index int) {
  503. Handler.restartKMSPlugin(cfg, index)
  504. }(k.config, idx)
  505. }
  506. }
  507. m.kmsLock.RUnlock()
  508. m.authLock.RLock()
  509. for idx, a := range m.auths {
  510. if a.exited() {
  511. defer func(cfg Config, index int) {
  512. Handler.restartAuthPlugin(cfg, index)
  513. }(a.config, idx)
  514. }
  515. }
  516. m.authLock.RUnlock()
  517. if m.hasSearcher {
  518. m.searcherLock.RLock()
  519. if m.searcher.exited() {
  520. defer func(cfg Config) {
  521. Handler.restartSearcherPlugin(cfg)
  522. }(m.searcher.config)
  523. }
  524. m.searcherLock.RUnlock()
  525. }
  526. if m.hasMetadater {
  527. m.metadaterLock.RLock()
  528. if m.metadater.exited() {
  529. defer func(cfg Config) {
  530. Handler.restartMetadaterPlugin(cfg)
  531. }(m.metadater.config)
  532. }
  533. m.metadaterLock.RUnlock()
  534. }
  535. if m.hasIPFilter {
  536. m.ipFilterLock.RLock()
  537. if m.filter.exited() {
  538. defer func(cfg Config) {
  539. Handler.restartIPFilterPlugin(cfg)
  540. }(m.filter.config)
  541. }
  542. m.ipFilterLock.RUnlock()
  543. }
  544. }
  545. func (m *Manager) restartNotifierPlugin(config Config, idx int) {
  546. if m.closed.Load() {
  547. return
  548. }
  549. logger.Info(logSender, "", "try to restart crashed notifier plugin %#v, idx: %v", config.Cmd, idx)
  550. plugin, err := newNotifierPlugin(config)
  551. if err != nil {
  552. logger.Error(logSender, "", "unable to restart notifier plugin %#v, err: %v", config.Cmd, err)
  553. return
  554. }
  555. m.notifLock.Lock()
  556. plugin.queue = m.notifiers[idx].queue
  557. m.notifiers[idx] = plugin
  558. m.notifLock.Unlock()
  559. plugin.sendQueuedEvents()
  560. }
  561. func (m *Manager) restartKMSPlugin(config Config, idx int) {
  562. if m.closed.Load() {
  563. return
  564. }
  565. logger.Info(logSender, "", "try to restart crashed kms plugin %#v, idx: %v", config.Cmd, idx)
  566. plugin, err := newKMSPlugin(config)
  567. if err != nil {
  568. logger.Error(logSender, "", "unable to restart kms plugin %#v, err: %v", config.Cmd, err)
  569. return
  570. }
  571. m.kmsLock.Lock()
  572. m.kms[idx] = plugin
  573. m.kmsLock.Unlock()
  574. }
  575. func (m *Manager) restartAuthPlugin(config Config, idx int) {
  576. if m.closed.Load() {
  577. return
  578. }
  579. logger.Info(logSender, "", "try to restart crashed auth plugin %#v, idx: %v", config.Cmd, idx)
  580. plugin, err := newAuthPlugin(config)
  581. if err != nil {
  582. logger.Error(logSender, "", "unable to restart auth plugin %#v, err: %v", config.Cmd, err)
  583. return
  584. }
  585. m.authLock.Lock()
  586. m.auths[idx] = plugin
  587. m.authLock.Unlock()
  588. }
  589. func (m *Manager) restartSearcherPlugin(config Config) {
  590. if m.closed.Load() {
  591. return
  592. }
  593. logger.Info(logSender, "", "try to restart crashed searcher plugin %#v", config.Cmd)
  594. plugin, err := newSearcherPlugin(config)
  595. if err != nil {
  596. logger.Error(logSender, "", "unable to restart searcher plugin %#v, err: %v", config.Cmd, err)
  597. return
  598. }
  599. m.searcherLock.Lock()
  600. m.searcher = plugin
  601. m.searcherLock.Unlock()
  602. }
  603. func (m *Manager) restartMetadaterPlugin(config Config) {
  604. if m.closed.Load() {
  605. return
  606. }
  607. logger.Info(logSender, "", "try to restart crashed metadater plugin %#v", config.Cmd)
  608. plugin, err := newMetadaterPlugin(config)
  609. if err != nil {
  610. logger.Error(logSender, "", "unable to restart metadater plugin %#v, err: %v", config.Cmd, err)
  611. return
  612. }
  613. m.metadaterLock.Lock()
  614. m.metadater = plugin
  615. m.metadaterLock.Unlock()
  616. }
  617. func (m *Manager) restartIPFilterPlugin(config Config) {
  618. if m.closed.Load() {
  619. return
  620. }
  621. logger.Info(logSender, "", "try to restart crashed IP filter plugin %#v", config.Cmd)
  622. plugin, err := newIPFilterPlugin(config)
  623. if err != nil {
  624. logger.Error(logSender, "", "unable to restart IP filter plugin %#v, err: %v", config.Cmd, err)
  625. return
  626. }
  627. m.ipFilterLock.Lock()
  628. m.filter = plugin
  629. m.ipFilterLock.Unlock()
  630. }
  631. func (m *Manager) addTask() {
  632. m.concurrencyGuard <- struct{}{}
  633. }
  634. func (m *Manager) removeTask() {
  635. <-m.concurrencyGuard
  636. }
  637. // Cleanup releases all the active plugins
  638. func (m *Manager) Cleanup() {
  639. if m.closed.Swap(true) {
  640. return
  641. }
  642. logger.Debug(logSender, "", "cleanup")
  643. close(m.done)
  644. m.notifLock.Lock()
  645. for _, n := range m.notifiers {
  646. logger.Debug(logSender, "", "cleanup notifier plugin %v", n.config.Cmd)
  647. n.cleanup()
  648. }
  649. m.notifLock.Unlock()
  650. m.kmsLock.Lock()
  651. for _, k := range m.kms {
  652. logger.Debug(logSender, "", "cleanup kms plugin %v", k.config.Cmd)
  653. k.cleanup()
  654. }
  655. m.kmsLock.Unlock()
  656. m.authLock.Lock()
  657. for _, a := range m.auths {
  658. logger.Debug(logSender, "", "cleanup auth plugin %v", a.config.Cmd)
  659. a.cleanup()
  660. }
  661. m.authLock.Unlock()
  662. if m.hasSearcher {
  663. m.searcherLock.Lock()
  664. logger.Debug(logSender, "", "cleanup searcher plugin %v", m.searcher.config.Cmd)
  665. m.searcher.cleanup()
  666. m.searcherLock.Unlock()
  667. }
  668. if m.hasMetadater {
  669. m.metadaterLock.Lock()
  670. logger.Debug(logSender, "", "cleanup metadater plugin %v", m.metadater.config.Cmd)
  671. m.metadater.cleanup()
  672. m.metadaterLock.Unlock()
  673. }
  674. if m.hasIPFilter {
  675. m.ipFilterLock.Lock()
  676. logger.Debug(logSender, "", "cleanup IP filter plugin %v", m.filter.config.Cmd)
  677. m.filter.cleanup()
  678. m.ipFilterLock.Unlock()
  679. }
  680. }
  681. func setLogLevel(logLevel string) {
  682. switch logLevel {
  683. case "info":
  684. pluginsLogLevel = hclog.Info
  685. case "warn":
  686. pluginsLogLevel = hclog.Warn
  687. case "error":
  688. pluginsLogLevel = hclog.Error
  689. default:
  690. pluginsLogLevel = hclog.Debug
  691. }
  692. }
  693. func startCheckTicker() {
  694. logger.Debug(logSender, "", "start plugins checker")
  695. go func() {
  696. ticker := time.NewTicker(30 * time.Second)
  697. defer ticker.Stop()
  698. for {
  699. select {
  700. case <-Handler.done:
  701. logger.Debug(logSender, "", "handler done, stop plugins checker")
  702. return
  703. case <-ticker.C:
  704. Handler.checkCrashedPlugins()
  705. }
  706. }
  707. }()
  708. }