From 51e35e1ba0790cfcfbaa668c0eb5d522a792b71d Mon Sep 17 00:00:00 2001 From: dsbaha <10324923+dsbaha@users.noreply.github.com> Date: Wed, 27 Oct 2021 21:51:17 -0700 Subject: [PATCH] Implementation of a Kafka REST Proxy logging plugin (#2534) This PR implements the status/result logger functions necessary interface with a Kafka REST Proxy service. Specifically, this is compatible with the [Confluent KAFKA Rest Proxy Service ](https://docs.confluent.io/1.0/kafka-rest/docs/intro.html). --- docs/02-Deploying/02-Configuration.md | 66 +++++++++++++++- server/config/config.go | 26 +++++- server/fleet/app.go | 7 ++ server/logging/kafkarest.go | 110 ++++++++++++++++++++++++++ server/logging/logging.go | 18 +++++ server/service/service_appconfig.go | 16 ++++ 6 files changed, 238 insertions(+), 5 deletions(-) create mode 100644 server/logging/kafkarest.go diff --git a/docs/02-Deploying/02-Configuration.md b/docs/02-Deploying/02-Configuration.md index 2f2d11d5d2..e2757aa7bf 100644 --- a/docs/02-Deploying/02-Configuration.md +++ b/docs/02-Deploying/02-Configuration.md @@ -875,7 +875,7 @@ Valid time units are `s`, `m`, `h`. Which log output plugin should be used for osquery status logs received from clients. -Options are `filesystem`, `firehose`, `kinesis`, `lambda`, `pubsub`, and `stdout`. +Options are `filesystem`, `firehose`, `kinesis`, `lambda`, `pubsub`, `kafkarest`, and `stdout`. - Default value: `filesystem` - Environment variable: `FLEET_OSQUERY_STATUS_LOG_PLUGIN` @@ -890,7 +890,7 @@ Options are `filesystem`, `firehose`, `kinesis`, `lambda`, `pubsub`, and `stdout Which log output plugin should be used for osquery result logs received from clients. -Options are `filesystem`, `firehose`, `kinesis`, `lambda`, `pubsub`, and `stdout`. +Options are `filesystem`, `firehose`, `kinesis`, `lambda`, `pubsub`, `kafkarest`, and `stdout`. - Default value: `filesystem` - Environment variable: `FLEET_OSQUERY_RESULT_LOG_PLUGIN` @@ -1425,6 +1425,68 @@ This feature is useful when combined with [subscription filters](https://cloud.g status_topic: osquery_status ``` +#### Kafka Rest Proxy Logging + +##### kafkarest_proxyhost + +This flag only has effect if `osquery_status_log_plugin` or `osquery_result_log_plugin` is set to `kafkarest`. + +The URL of the host which to check for the topic existence and post messages to the specified topic. + +- Default value: none +- Environment variable: `FLEET_KAFKAREST_PROXYHOST` +- Config file format: + + ``` + kafkarest: + proxyhost: "https://localhost:8443" + ``` + +##### kafkarest_status_topic + +This flag only has effect if `osquery_status_log_plugin` is set to `kafkarest`. + +The identifier of the kafka topic that osquery status logs will be published to. + +- Default value: none +- Environment variable: `FLEET_KAFKAREST_STATUS_TOPIC` +- Config file format: + + ``` + kafkarest: + status_topic: osquery_status + ``` + +##### kafkarest_result_topic + +This flag only has effect if `osquery_result_log_plugin` is set to `kafkarest`. + +The identifier of the kafka topic that osquery status logs will be published to. + +- Default value: none +- Environment variable: `FLEET_KAFKAREST_RESULT_TOPIC` +- Config file format: + + ``` + kafkarest: + status_topic: osquery_result + ``` + +##### kafkarest_timeout + +This flag only has effect if `osquery_status_log_plugin` or `osquery_result_log_plugin` is set to `kafkarest`. + +The timeout value for the http post attempt. Value is in units of seconds. + +- Default value: 5 +- Environment variable: `FLEET_KAFKAREST_TIMEOUT` +- Config file format: + + ``` + kafkarest: + timeout: 5 + ``` + #### S3 file carving backend ##### s3_bucket diff --git a/server/config/config.go b/server/config/config.go index 1c23f325a8..3cc46beb78 100644 --- a/server/config/config.go +++ b/server/config/config.go @@ -33,7 +33,7 @@ type MysqlConfig struct { TLSKey string `yaml:"tls_key"` TLSCA string `yaml:"tls_ca"` TLSServerName string `yaml:"tls_server_name"` - TLSConfig string `yaml:"tls_config"` //tls=customValue in DSN + TLSConfig string `yaml:"tls_config"` // tls=customValue in DSN MaxOpenConns int `yaml:"max_open_conns"` MaxIdleConns int `yaml:"max_idle_conns"` ConnMaxLifetime int `yaml:"conn_max_lifetime"` @@ -183,6 +183,14 @@ type FilesystemConfig struct { EnableLogCompression bool `json:"enable_log_compression" yaml:"enable_log_compression"` } +// KafkaRESTConfig defines configs for the Kafka REST Proxy logging plugin. +type KafkaRESTConfig struct { + StatusTopic string `json:"status_topic" yaml:"status_topic"` + ResultTopic string `json:"result_topic" yaml:"result_topic"` + ProxyHost string `json:"proxyhost" yaml:"proxyhost"` + Timeout int `json:"timeout" yaml:"timeout"` +} + // LicenseConfig defines configs related to licensing Fleet. type LicenseConfig struct { Key string `yaml:"key"` @@ -218,6 +226,7 @@ type FleetConfig struct { S3 S3Config PubSub PubSubConfig Filesystem FilesystemConfig + KafkaREST KafkaRESTConfig License LicenseConfig Vulnerabilities VulnerabilitiesConfig } @@ -454,6 +463,12 @@ func (man Manager) addConfigs() { man.addConfigBool("filesystem.enable_log_compression", false, "Enable compression for the rotated osquery log files") + // KafkaREST + man.addConfigString("kafkarest.status_topic", "", "Kafka REST topic for status logs") + man.addConfigString("kafkarest.result_topic", "", "Kafka REST topic for result logs") + man.addConfigString("kafkarest.proxyhost", "", "Kafka REST proxy host url") + man.addConfigInt("kafkarest.timeout", 5, "Kafka REST proxy json post timeout") + // License man.addConfigString("license.key", "", "Fleet license key (to enable Fleet Premium features)") @@ -609,6 +624,12 @@ func (man Manager) LoadConfig() FleetConfig { EnableLogRotation: man.getConfigBool("filesystem.enable_log_rotation"), EnableLogCompression: man.getConfigBool("filesystem.enable_log_compression"), }, + KafkaREST: KafkaRESTConfig{ + StatusTopic: man.getConfigString("kafkarest.status_topic"), + ResultTopic: man.getConfigString("kafkarest.result_topic"), + ProxyHost: man.getConfigString("kafkarest.proxyhost"), + Timeout: man.getConfigInt("kafkarest.timeout"), + }, License: LicenseConfig{ Key: man.getConfigString("license.key"), }, @@ -807,7 +828,6 @@ func (man Manager) loadConfigFile() { man.viper.SetConfigFile(configFile) err := man.viper.ReadInConfig() - if err != nil { fmt.Println("Error loading config file:", err) os.Exit(1) @@ -819,7 +839,7 @@ func (man Manager) loadConfigFile() { // TestConfig returns a barebones configuration suitable for use in tests. // Individual tests may want to override some of the values provided. func TestConfig() FleetConfig { - var testLogFile = "/dev/null" + testLogFile := "/dev/null" if runtime.GOOS == "windows" { testLogFile = "NUL" } diff --git a/server/fleet/app.go b/server/fleet/app.go index 3feeecffa3..668b3d976c 100644 --- a/server/fleet/app.go +++ b/server/fleet/app.go @@ -383,3 +383,10 @@ type LambdaConfig struct { StatusFunction string `json:"status_function"` ResultFunction string `json:"result_function"` } + +// KafkaRESTConfig shadows config.KafkaRESTConfig +type KafkaRESTConfig struct { + StatusTopic string `json:"status_topic"` + ResultTopic string `json:"result_topic"` + ProxyHost string `json:"proxyhost"` +} diff --git a/server/logging/kafkarest.go b/server/logging/kafkarest.go new file mode 100644 index 0000000000..bff481db18 --- /dev/null +++ b/server/logging/kafkarest.go @@ -0,0 +1,110 @@ +package logging + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io/ioutil" + "net/http" + "time" + + "github.com/pkg/errors" +) + +const ( + krContentTypeValue = "application/vnd.kafka.json.v1+json" + krContentTypeHeader = "Content-Type" + krTimestampHeader = "TimeStamp" + krPublishTopicURL = "%s/topics/%s" + krCheckTopicURL = "%s/topics/?topic=%s" +) + +type KafkaRESTParams struct { + KafkaProxyHost string + KafkaTopic string + KafkaTimeout int +} + +type kafkaRESTProducer struct { + client *http.Client + URL string + CheckURL string +} + +type kafkaRecords struct { + Records []kafkaValue `json:"records"` +} + +type kafkaValue struct { + Value json.RawMessage `json:"value"` +} + +func NewKafkaRESTWriter(p *KafkaRESTParams) (*kafkaRESTProducer, error) { + producer := &kafkaRESTProducer{ + URL: fmt.Sprintf(krPublishTopicURL, p.KafkaProxyHost, p.KafkaTopic), + CheckURL: fmt.Sprintf(krCheckTopicURL, p.KafkaProxyHost, p.KafkaTopic), + client: &http.Client{ + Timeout: time.Duration(p.KafkaTimeout) * time.Second, + }, + } + + return producer, producer.checkTopic() +} + +func (l *kafkaRESTProducer) Write(ctx context.Context, logs []json.RawMessage) error { + data := kafkaRecords{ + Records: make([]kafkaValue, len(logs)), + } + + for i, log := range logs { + data.Records[i] = kafkaValue{ + Value: log, + } + } + + output, err := json.Marshal(data) + if err != nil { + return errors.Wrap(err, "kafka rest marshal") + } + + resp, err := l.post(l.URL, bytes.NewBuffer(output)) + if err != nil { + return errors.Wrap(err, "kafka rest post") + } + defer resp.Body.Close() + + return checkResponse(resp) +} + +func checkResponse(resp *http.Response) (err error) { + if resp.StatusCode != http.StatusOK { + body, _ := ioutil.ReadAll(resp.Body) + return errors.Errorf("Error: %d. %s", resp.StatusCode, string(body)) + } + + return nil +} + +func (l *kafkaRESTProducer) checkTopic() (err error) { + resp, err := l.client.Get(l.CheckURL) + if err != nil { + return errors.Wrap(err, "kafka rest topic check") + } + defer resp.Body.Close() + + return checkResponse(resp) +} + +func (l *kafkaRESTProducer) post(url string, buf *bytes.Buffer) (*http.Response, error) { + req, err := http.NewRequest(http.MethodPost, url, buf) + if err != nil { + return nil, errors.Wrap(err, "kafka rest new request") + } + + now := float64(time.Now().UnixNano()) / float64(time.Second) + req.Header.Set(krContentTypeHeader, krContentTypeValue) + req.Header.Set(krTimestampHeader, fmt.Sprintf("%f", now)) + + return l.client.Do(req) +} diff --git a/server/logging/logging.go b/server/logging/logging.go index 80cc331cdc..86138614e3 100644 --- a/server/logging/logging.go +++ b/server/logging/logging.go @@ -87,6 +87,15 @@ func New(config config.FleetConfig, logger log.Logger) (*OsqueryLogger, error) { if err != nil { return nil, errors.Wrap(err, "create stdout status logger") } + case "kafkarest": + status, err = NewKafkaRESTWriter(&KafkaRESTParams{ + KafkaProxyHost: config.KafkaREST.ProxyHost, + KafkaTopic: config.KafkaREST.StatusTopic, + KafkaTimeout: config.KafkaREST.Timeout, + }) + if err != nil { + return nil, errors.Wrap(err, "create kafka rest status logger") + } default: return nil, errors.Errorf( "unknown status log plugin: %s", config.Osquery.StatusLogPlugin, @@ -161,6 +170,15 @@ func New(config config.FleetConfig, logger log.Logger) (*OsqueryLogger, error) { if err != nil { return nil, errors.Wrap(err, "create stdout result logger") } + case "kafkarest": + result, err = NewKafkaRESTWriter(&KafkaRESTParams{ + KafkaProxyHost: config.KafkaREST.ProxyHost, + KafkaTopic: config.KafkaREST.ResultTopic, + KafkaTimeout: config.KafkaREST.Timeout, + }) + if err != nil { + return nil, errors.Wrap(err, "create kafka rest result logger") + } default: return nil, errors.Errorf( "unknown result log plugin: %s", config.Osquery.StatusLogPlugin, diff --git a/server/service/service_appconfig.go b/server/service/service_appconfig.go index 89987d429b..60a35e121c 100644 --- a/server/service/service_appconfig.go +++ b/server/service/service_appconfig.go @@ -248,6 +248,14 @@ func (svc *Service) LoggingConfig(ctx context.Context) (*fleet.Logging, error) { } case "stdout": logging.Status = fleet.LoggingPlugin{Plugin: "stdout"} + case "kafkarest": + logging.Status = fleet.LoggingPlugin{ + Plugin: "kafkarest", + Config: fleet.KafkaRESTConfig{ + StatusTopic: conf.KafkaREST.StatusTopic, + ProxyHost: conf.KafkaREST.ProxyHost, + }, + } default: return nil, errors.Errorf("unrecognized logging plugin: %s", conf.Osquery.StatusLogPlugin) } @@ -294,6 +302,14 @@ func (svc *Service) LoggingConfig(ctx context.Context) (*fleet.Logging, error) { logging.Result = fleet.LoggingPlugin{ Plugin: "stdout", } + case "kafkarest": + logging.Result = fleet.LoggingPlugin{ + Plugin: "kafkarest", + Config: fleet.KafkaRESTConfig{ + ResultTopic: conf.KafkaREST.ResultTopic, + ProxyHost: conf.KafkaREST.ProxyHost, + }, + } default: return nil, errors.Errorf("unrecognized logging plugin: %s", conf.Osquery.ResultLogPlugin)