Discharge7443/fleet
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

Implementation of a Kafka REST Proxy logging plugin (#2534)

Browse Source 
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).
This commit is contained in:
dsbaha
2021-10-27 21:51:17 -07:00
committed by GitHub
parent 37e4452274
commit 51e35e1ba0
6 changed files with 238 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/02-Deploying/02-Configuration.md
+64 -2
View File
@@ -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
server/config/config.go
+23 -3
View File
@@ -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"
}
server/fleet/app.go
+7
View File
@@ -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"`
}
server/logging/kafkarest.go
+110
View File
@@ -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)
}
server/logging/logging.go
+18
View File
@@ -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,
server/service/service_appconfig.go
+16
View File
@@ -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)