plugin.go 22 KB

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