Repository: kentik/ktranslate
Pull request #193 (2021-11-16 23:02:55)

Fixes for Security and AWS
Grab bag of small things. 


Patch:
diff --git a/README.md b/README.md
index d446cf7..58ed02b 100644
--- a/README.md
+++ b/README.md
@@ -44,8 +44,6 @@ Usage of ./bin/ktranslate:
     	City mapping file
   -compression string
     	compression algo to use (none|gzip|snappy|deflate|null) (default "none")
-  -ddog_url string
-    	URL to post to (default "https://api.datadoghq.com/api/v1/series")
   -dns string
     	Resolve IPs at this ip:port
   -file_flush_sec int
@@ -57,9 +55,9 @@ Usage of ./bin/ktranslate:
   -filters value
     	Any filters to use. Format: type dimension operator value
   -format string
-    	Format to convert kflow to: (json|flat_json|avro|netflow|influx|prometheus|new_relic|new_relic_metric|splunk|ddog) (default "json")
+	Format to convert kflow to: (json|flat_json|avro|netflow|influx|prometheus|new_relic|new_relic_metric) (default "json")
   -format_rollup string
-    	Format to convert rollups to: (json|avro|netflow|influx|prometheus|new_relic|new_relic_metric|splunk|ddog)
+	Format to convert rollups to: (json|avro|netflow|influx|prometheus|new_relic|new_relic_metric)
   -gcloud_bucket string
     	GCloud Storage Bucket to write flows to
   -gcloud_content_type string
@@ -175,7 +173,7 @@ Usage of ./bin/ktranslate:
   -service_name string
     	Service identifier (default "ktranslate")
   -sinks string
-    	List of sinks to send data to. Options: (kafka|stdout|new_relic|kentik|net|http|splunk|prometheus|file|s3|gcloud|ddog) (default "stdout")
+	List of sinks to send data to. Options: (kafka|stdout|new_relic|kentik|net|http|prometheus|file|s3|gcloud) (default "stdout")
   -snmp string
     	yaml file containing snmp config to use
   -snmp_discovery
diff --git a/cmd/ktranslate/main.go b/cmd/ktranslate/main.go
index ee26b9b..f11565e 100644
--- a/cmd/ktranslate/main.go
+++ b/cmd/ktranslate/main.go
@@ -166,6 +166,8 @@ func setMode(bs *baseserver.BaseServer, mode string, sample int, syslog string)
 		return // noop
 	case "nr1.vpc.lambda":
 		setNr() // Here, we only send the flow in as events to NR.
+		flag.Set("vpc", "aws")
+		flag.Set("aws_lambda", "true")
 	case "vpc":
 		flag.Set("rollups", "s_sum,vpc.xmt.bytes,out_bytes,custom_str.source_vpc,custom_str.application_type,custom_str.source_account,custom_str.source_region,src_addr,custom_str.src_as_name,src_geo,l4_src_port,protocol")
 		flag.Set("rollups", "s_sum,vpc.rcv.bytes,in_bytes,custom_str.dest_vpc,custom_str.application_type,custom_str.dest_account,custom_str.dest_region,dst_addr,custom_str.dst_as_name,dst_geo,l4_dst_port,protocol")
diff --git a/pkg/eggs/baseserver/baseserver.go b/pkg/eggs/baseserver/baseserver.go
index 684653d..220dabc 100644
--- a/pkg/eggs/baseserver/baseserver.go
+++ b/pkg/eggs/baseserver/baseserver.go
@@ -80,9 +80,9 @@ type BaseServerConfiguration struct {
 
 var BaseServerConfigurationDefaults = BaseServerConfiguration{
 	LogToStdout:             true,
-	LogLevel:                "debug",
-	MetricsDestination:      "syslog",
-	MetaListen:              "localhost:0",
+	LogLevel:                "info",
+	MetricsDestination:      "none",
+	MetaListen:              "",
 	ShutdownSettleTime:      1 * time.Second,
 	HealthCheckStartupDelay: 5 * time.Second,
 	HealthCheckPeriod:       30 * time.Second,
@@ -177,7 +177,7 @@ func (bs *BaseServer) ParseFlags() {
 	flag.CommandLine.StringVar(&bs.LogLevel, "log_level", bs.LogLevel, "Logging Level")
 	flag.CommandLine.BoolVar(&bs.LogToStdout, "stdout", bs.LogToStdout, "Log to stdout")
 	flag.CommandLine.StringVar(&bs.MetricsDestination, "metrics", bs.MetricsDestination, "Metrics Configuration. none|syslog|stderr|graphite:127.0.0.1:2003")
-	flag.CommandLine.StringVar(&bs.MetaListen, "metalisten", bs.MetaListen, "HTTP port to bind on")
+	flag.CommandLine.StringVar(&bs.MetaListen, "metalisten", bs.MetaListen, "HTTP interface and port to bind on")
 	flag.CommandLine.StringVar(&bs.OllyDataset, "olly_dataset", bs.OllyDataset, "Olly dataset name")
 	flag.CommandLine.StringVar(&bs.OllyWriteKey, "olly_write_key", bs.OllyWriteKey, "Olly dataset name")
 
@@ -394,6 +394,10 @@ func (bs *BaseServer) spawnLegacyHealthCheck(ctx context.Context, service Servic
 }
 
 func (bs *BaseServer) spawnMetaServer(ctx context.Context, service Service) {
+	if bs.MetaListen == "" { // This is turned off for now.
+		return
+	}
+
 	if bs.hce == nil {
 		bs.Fail("initMetaServer: hce cannot be nil")
 	}
diff --git a/pkg/eggs/baseserver/meta.go b/pkg/eggs/baseserver/meta.go
index 6983633..4f59aac 100644
--- a/pkg/eggs/baseserver/meta.go
+++ b/pkg/eggs/baseserver/meta.go
@@ -36,6 +36,7 @@ var ENV_BLACKLIST = []string{
 	"pg_write_connection",
 	"mailer",
 	"alert_connection",
+	"new_relic_api_key",
 }
 
 type MetaServer struct {
diff --git a/pkg/inputs/snmp/metadata/device_metadata.go b/pkg/inputs/snmp/metadata/device_metadata.go
index e3bdbc0..98c01f1 100644
--- a/pkg/inputs/snmp/metadata/device_metadata.go
+++ b/pkg/inputs/snmp/metadata/device_metadata.go
@@ -107,7 +107,7 @@ func GetDeviceMetadata(log logger.ContextL, server *gosnmp.GoSNMP, deviceMetadat
 		case SNMP_sysServices:
 			md.SysServices = int(snmp_util.ToInt64(value))
 		case SNMP_engineID:
-			_, md.EngineID = snmp_util.GetFromConv(pdu, snmp_util.CONV_HWADDR, log)
+			_, md.EngineID = snmp_util.GetFromConv(pdu, snmp_util.CONV_ENGINE_ID, log)
 		default:
 			if oid.Tag != "" {
 				oidName = oid.Tag
diff --git a/pkg/inputs/snmp/util/util.go b/pkg/inputs/snmp/util/util.go
index 3829ecc..1eee242 100644
--- a/pkg/inputs/snmp/util/util.go
+++ b/pkg/inputs/snmp/util/util.go
@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/base64"
 	"encoding/binary"
+	"encoding/hex"
 	"fmt"
 	"net"
 	"strconv"
@@ -18,10 +19,11 @@ import (
 )
 
 const (
-	CONV_HWADDR   = "hwaddr"
-	CONV_POWERSET = "powerset_status"
-	CONV_HEXTOINT = "hextoint"
-	CONV_HEXTOIP  = "hextoip"
+	CONV_HWADDR    = "hwaddr"
+	CONV_POWERSET  = "powerset_status"
+	CONV_HEXTOINT  = "hextoint"
+	CONV_HEXTOIP   = "hextoip"
+	CONV_ENGINE_ID = "engine_id"
 
 	NRUserTagPrefix = "tags."
 )
@@ -243,6 +245,8 @@ func GetFromConv(pdu gosnmp.SnmpPDU, conv string, log logger.ContextL) (int64, s
 		return vendor.HandlePowersetStatus(bv)
 	case CONV_HEXTOIP:
 		return hexToIP(bv)
+	case CONV_ENGINE_ID:
+		return engineID(bv)
 	default:
 		// Otherwise, try out some custom conversions.
 		split := strings.Split(conv, ":")
@@ -311,3 +315,13 @@ func hexToIP(bv []byte) (int64, string) {
 		return 0, ""
 	}
 }
+
+func engineID(bv []byte) (int64, string) {
+	buf := make([]byte, 0, 3*len(bv))
+	x := buf[1*len(bv) : 3*len(bv)]
+	hex.Encode(x, bv)
+	for i := 0; i < len(x); i += 2 {
+		buf = append(buf, x[i], x[i+1], ':')
+	}
+	return 0, string(buf[:len(buf)-1])
+}
diff --git a/pkg/sinks/nr/nr.go b/pkg/sinks/nr/nr.go
index b6bd844..4fb680f 100644
--- a/pkg/sinks/nr/nr.go
+++ b/pkg/sinks/nr/nr.go
@@ -33,6 +33,12 @@ const (
 	REGION_US_STAGING = "us_stage"
 )
 
+var (
+	NrUrl        = "https://insights-collector.newrelic.com/v1/accounts/%s/events"
+	NrMetricsUrl = "https://metric-api.newrelic.com/metric/v1"
+	NrLogUrl     = "https://log-api.newrelic.com/log/v1"
+)
+
 type NRSink struct {
 	logger.ContextL
 	NRAccount   string
@@ -70,9 +76,6 @@ type NRResponce struct {
 
 var (
 	NrAccount    = flag.String("nr_account_id", kt.LookupEnvString("NR_ACCOUNT_ID", ""), "If set, sends flow to New Relic")
-	NrUrl        = flag.String("nr_url", "https://insights-collector.newrelic.com/v1/accounts/%s/events", "URL to use to send into NR")
-	NrMetricsUrl = flag.String("nr_metrics_url", "https://metric-api.newrelic.com/metric/v1", "URL to use to send into NR Metrics API")
-	NrLogUrl     = flag.String("nr_log_url", "https://log-api.newrelic.com/log/v1", "URL to use to logs into NR")
 	EstimateSize = flag.Bool("nr_estimate_only", false, "If true, record size of inputs to NR but don't actually send anything")
 	NrRegion     = flag.String("nr_region", kt.LookupEnvString("NR_REGION", ""), "NR Region to use. US|EU")
 	NrCheckJson  = flag.Bool("nr_check_json", false, "Verify body is valid json before sending on")
@@ -127,15 +130,15 @@ func (s *NRSink) Init(ctx context.Context, format formats.Format, compression kt
 	switch rval {
 	case "": // noop
 	case REGION_US, REGION_EU, REGION_GOV, REGION_US_STAGING:
-		*NrUrl = regions[rval]["events"]
-		*NrMetricsUrl = regions[rval]["metrics"]
+		NrUrl = regions[rval]["events"]
+		NrMetricsUrl = regions[rval]["metrics"]
 		s.NRUrlLog = regions[rval]["logs"]
 	default:
 		return fmt.Errorf("You used an unsupported New Relic One region: %s. The possible values are EU, US, GOV and US_STAGE.", *NrRegion)
 	}
 
 	s.NRAccount = *NrAccount
-	s.NRUrl = *NrUrl
+	s.NRUrl = NrUrl
 	s.format = format
 	s.compression = compression
 
@@ -150,7 +153,7 @@ func (s *NRSink) Init(ctx context.Context, format formats.Format, compression kt
 		IdleConnTimeout:       90 * time.Second,
 		TLSHandshakeTimeout:   10 * time.Second,
 		ExpectContinueTimeout: 1 * time.Second,
-		TLSClientConfig:       &tls.Config{InsecureSkipVerify: true},
+		TLSClientConfig:       &tls.Config{InsecureSkipVerify: false}, // TODO, any time that we want this to be false?
 	}
 	s.client = &http.Client{Transport: s.tr}
 
@@ -175,13 +178,13 @@ func (s *NRSink) Init(ctx context.Context, format formats.Format, compression kt
 		s.NRUrl = fmt.Sprintf(s.NRUrl, s.NRAccount)
 	}
 	s.NRUrlEvent = s.NRUrl
-	s.NRUrlMetric = *NrMetricsUrl
+	s.NRUrlMetric = NrMetricsUrl
 	if s.format == formats.FORMAT_NRM {
-		s.NRUrl = *NrMetricsUrl
+		s.NRUrl = NrMetricsUrl
 	}
 
 	if s.NRUrlLog == "" {
-		s.NRUrlLog = *NrLogUrl
+		s.NRUrlLog = NrLogUrl
 	}
 
 	// Send logs on to NR if this is set.
@@ -370,6 +373,7 @@ func (s *NRSink) sendLogBatch(ctx context.Context, logs []string) {
 
 	if s.compression != kt.CompressionGzip {
 		s.sendNR(ctx, kt.NewOutput(target), s.NRUrlLog)
+		return
 	}
 
 	serBuf := []byte{}


Test patch:
diff --git a/pkg/eggs/baseserver/meta_test.go b/pkg/eggs/baseserver/meta_test.go
index e47b294..f908913 100644
--- a/pkg/eggs/baseserver/meta_test.go
+++ b/pkg/eggs/baseserver/meta_test.go
@@ -30,6 +30,7 @@ func TestMetaServer(t *testing.T) {
 	BaseServerConfigurationDefaults.HealthCheckStartupDelay = 50 * time.Millisecond
 	BaseServerConfigurationDefaults.HealthCheckPeriod = 1 * time.Second
 	BaseServerConfigurationDefaults.HealthCheckTimeout = 1 * time.Second
+	BaseServerConfigurationDefaults.MetaListen = "localhost:0"
 
 	bs := Boilerplate("dumdum", versionInfo, nil, nil)
 	assert.NotNil(t, bs, "Boilerplate should return a non-nil value")
diff --git a/pkg/inputs/snmp/util/util_test.go b/pkg/inputs/snmp/util/util_test.go
index 9fbf8a4..2c08e89 100644
--- a/pkg/inputs/snmp/util/util_test.go
+++ b/pkg/inputs/snmp/util/util_test.go
@@ -158,3 +158,20 @@ func TestHexToIP(t *testing.T) {
 		assert.Equal(expt, str, "%s", in)
 	}
 }
+
+func TestEngineID(t *testing.T) {
+	assert := assert.New(t)
+	l := lt.NewTestContextL(logger.NilContext, t)
+
+	tests := map[string][]byte{
+		"38:30:3a:30:30:3a:31:66:3a:38:38:3a:38:30:3a:37:00:00":       []byte{0x38, 0x30, 0x3a, 0x30, 0x30, 0x3a, 0x31, 0x66, 0x3a, 0x38, 0x38, 0x3a, 0x38, 0x30, 0x3a, 0x37, 0x00, 0x00},
+		"30:30:3a:31:66:3a:38:38:3a:38:30:3a:37:00:00":                []byte{0x30, 0x30, 0x3a, 0x31, 0x66, 0x3a, 0x38, 0x38, 0x3a, 0x38, 0x30, 0x3a, 0x37, 0x00, 0x00},
+		"38:30:3a:30:30:3a:31:66:3a:38:38:3a:38:30:3a:37:00:00:00:00": []byte{0x38, 0x30, 0x3a, 0x30, 0x30, 0x3a, 0x31, 0x66, 0x3a, 0x38, 0x38, 0x3a, 0x38, 0x30, 0x3a, 0x37, 0x00, 0x00, 0x00, 0x00},
+	}
+
+	for expt, in := range tests {
+		pdu := gosnmp.SnmpPDU{Value: in}
+		_, str := GetFromConv(pdu, CONV_ENGINE_ID, l)
+		assert.Equal(expt, str, "%s", in)
+	}
+}