in_splunk: out_splunk: Propagate ingested splunk_token authentication header

plugins/in_splunk/splunk.h

@@ -42,6 +42,7 @@ struct flb_splunk {
 
     /* Token Auth */
     flb_sds_t auth_header;
+    flb_sds_t ingested_auth_header;
 
     struct flb_log_event_encoder log_encoder;
 

plugins/in_splunk/splunk_config.c

@@ -51,6 +51,7 @@ struct flb_splunk *splunk_config_create(struct flb_input_instance *ins)
     }
 
     ctx->auth_header = NULL;
+    ctx->ingested_auth_header = NULL;
     tmp = flb_input_get_property("splunk_token", ins);
     if (tmp) {
         ctx->auth_header = flb_sds_create("Splunk ");

plugins/in_splunk/splunk_prot.c

@@ -233,6 +233,15 @@ static int process_raw_payload_pack(struct flb_splunk *ctx, flb_sds_t tag, char
                 FLB_LOG_EVENT_STRING_VALUE(buf, size));
     }
 
+    if (ctx->ingested_auth_header != NULL) {
+        if (ret == FLB_EVENT_ENCODER_SUCCESS) {
+            ret = flb_log_event_encoder_append_metadata_values(
+                &ctx->log_encoder,
+                FLB_LOG_EVENT_CSTRING_VALUE("hec_token"),
+                FLB_LOG_EVENT_CSTRING_VALUE(ctx->ingested_auth_header));
+        }
+    }
+
     if (ret == FLB_EVENT_ENCODER_SUCCESS) {
         ret = flb_log_event_encoder_commit_record(&ctx->log_encoder);
     }
@@ -281,6 +290,15 @@ static void process_flb_log_append(struct flb_splunk *ctx, msgpack_object *recor
                 record);
     }
 
+    if (ctx->ingested_auth_header != NULL) {
+        if (ret == FLB_EVENT_ENCODER_SUCCESS) {
+            ret = flb_log_event_encoder_append_metadata_values(
+                &ctx->log_encoder,
+                FLB_LOG_EVENT_CSTRING_VALUE("hec_token"),
+                FLB_LOG_EVENT_CSTRING_VALUE(ctx->ingested_auth_header));
+        }
+    }
+
     if (ret == FLB_EVENT_ENCODER_SUCCESS) {
         ret = flb_log_event_encoder_commit_record(&ctx->log_encoder);
     }
@@ -477,6 +495,7 @@ static int process_hec_payload(struct flb_splunk *ctx, struct splunk_conn *conn,
     int ret = 0;
     int type = -1;
     struct mk_http_header *header;
+    struct mk_http_header *header_auth;
     int extra_size = -1;
     struct mk_http_header *headers_extra;
     int gzip_compressed = FLB_FALSE;
@@ -508,6 +527,13 @@ static int process_hec_payload(struct flb_splunk *ctx, struct splunk_conn *conn,
         return -1;
     }
 
+    header_auth = &session->parser.headers[MK_HEADER_AUTHORIZATION];
+    if (header_auth->key.data != NULL) {
+        if (strncasecmp(header_auth->val.data, "Splunk ", 7) == 0) {
+            ctx->ingested_auth_header = header_auth->val.data;
+        }
+    }
+
     extra_size = session->parser.headers_extra_count;
     if (extra_size > 0) {
         for (i = 0; i < extra_size; i++) {
@@ -548,6 +574,7 @@ static int process_hec_raw_payload(struct flb_splunk *ctx, struct splunk_conn *c
 {
     int ret = -1;
     struct mk_http_header *header;
+    struct mk_http_header *header_auth;
 
     header = &session->parser.headers[MK_HEADER_CONTENT_TYPE];
     if (header->key.data == NULL) {
@@ -565,6 +592,13 @@ static int process_hec_raw_payload(struct flb_splunk *ctx, struct splunk_conn *c
         return -1;
     }
 
+    header_auth = &session->parser.headers[MK_HEADER_AUTHORIZATION];
+    if (header_auth->key.data != NULL) {
+        if (strncasecmp(header_auth->val.data, "Splunk ", 7) == 0) {
+            ctx->ingested_auth_header = header_auth->val.data;
+        }
+    }
+
     /* Always handle as raw type of payloads here */
     ret = process_raw_payload_pack(ctx, tag, request->data.data, request->data.len);
 
@@ -889,6 +923,9 @@ static int process_hec_payload_ng(struct flb_http_request *request,
                                   struct flb_splunk *ctx)
 {
     int type = -1;
+    int ret = 0;
+    size_t size = 0;
+    char *auth_header;
 
     type = HTTP_CONTENT_UNKNOWN;
 
@@ -905,6 +942,11 @@ static int process_hec_payload_ng(struct flb_http_request *request,
         }
     }
 
+    ret = flb_hash_table_get(request->headers, "authorization", 13, (void **)&auth_header, &size);
+    if (ret != 0) {
+        ctx->ingested_auth_header = auth_header;
+    }
+
     if (request->body == NULL || cfl_sds_len(request->body) <= 0) {
         send_response_ng(response, 400, "error: no payload found\n");
 

plugins/out_splunk/splunk.c

@@ -345,19 +345,44 @@ static inline int splunk_metrics_format(struct flb_output_instance *ins,
 }
 #endif
 
+
+/* implements functionality to get auth_header from msgpack map (metadata) */
+static flb_sds_t extract_hec_token(struct flb_splunk *ctx, msgpack_object map,
+                                   char *tag, int tag_len)
+{
+    flb_sds_t hec_token;
+
+    /* Extract HEC token (map which is from metadata lookup) */
+    if (ctx->event_sourcetype_key) {
+        hec_token = flb_ra_translate(ctx->ra_metadata_auth_key, tag, tag_len,
+                                     map, NULL);
+        if (hec_token) {
+            return hec_token;
+        }
+
+        flb_plg_debug(ctx->ins, "Could not find hec_token in metadata");
+        return NULL;
+    }
+
+    flb_plg_debug(ctx->ins, "Could not find a record accessor definition of hec_token");
+    return NULL;
+}
+
 static inline int splunk_format(const void *in_buf, size_t in_bytes,
                                 char *tag, int tag_len,
                                 char **out_buf, size_t *out_size,
                                 struct flb_splunk *ctx)
 {
     int ret;
     msgpack_object map;
+    msgpack_object metadata;
     msgpack_sbuffer mp_sbuf;
     msgpack_packer mp_pck;
     char *err;
     flb_sds_t tmp;
     flb_sds_t record;
     flb_sds_t json_out;
+    flb_sds_t metadata_hec_token = NULL;
     struct flb_log_event_decoder log_decoder;
     struct flb_log_event log_event;
 
@@ -378,6 +403,8 @@ static inline int splunk_format(const void *in_buf, size_t in_bytes,
         return -1;
     }
 
+    ctx->metadata_auth_header = NULL;
+
     while ((ret = flb_log_event_decoder_next(
                     &log_decoder,
                     &log_event)) == FLB_EVENT_DECODER_SUCCESS) {
@@ -387,6 +414,19 @@ static inline int splunk_format(const void *in_buf, size_t in_bytes,
         msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write);
 
         map = *log_event.body;
+        metadata = *log_event.metadata;
+        metadata_hec_token = extract_hec_token(ctx, metadata, tag, tag_len);
+
+        if (metadata_hec_token != NULL) {
+            /* Currently, in_splunk implementation permits to
+             * specify only one splunk token per one instance.
+             * So, it should be valid if storing only last value of
+             * splunk token per one chunk. */
+            if (ctx->metadata_auth_header != NULL) {
+                cfl_sds_destroy(ctx->metadata_auth_header);
+            }
+            ctx->metadata_auth_header = metadata_hec_token;
+        }
 
         if (ctx->event_key) {
             /* Pack the value of a event key */
@@ -644,6 +684,10 @@ static void cb_splunk_flush(struct flb_event_chunk *event_chunk,
     if (ctx->http_user && ctx->http_passwd) {
         flb_http_basic_auth(c, ctx->http_user, ctx->http_passwd);
     }
+    else if (ctx->metadata_auth_header) {
+        flb_http_add_header(c, "Authorization", 13,
+                            ctx->metadata_auth_header, flb_sds_len(ctx->metadata_auth_header));
+    }
     else if (ctx->auth_header) {
         flb_http_add_header(c, "Authorization", 13,
                             ctx->auth_header, flb_sds_len(ctx->auth_header));
@@ -711,6 +755,9 @@ static void cb_splunk_flush(struct flb_event_chunk *event_chunk,
     }
 
     /* Cleanup */
+    if (ctx->metadata_auth_header != NULL) {
+        cfl_sds_destroy(ctx->metadata_auth_header);
+    }
     flb_http_client_destroy(c);
     flb_upstream_conn_release(u_conn);
     FLB_OUTPUT_RETURN(ret);
@@ -817,7 +864,8 @@ static struct flb_config_map config_map[] = {
     {
      FLB_CONFIG_MAP_STR, "splunk_token", NULL,
      0, FLB_FALSE, 0,
-     "Specify the Authentication Token for the HTTP Event Collector interface."
+     "Specify the Authentication Token for the HTTP Event Collector interface. "
+     "If event metadata contains a splunk_token, it will be prioritized to use instead of this token."
     },
 
     {

plugins/out_splunk/splunk.h

@@ -95,6 +95,12 @@ struct flb_splunk {
 
     /* Token Auth */
     flb_sds_t auth_header;
+    /* Token Auth (via metadata) */
+    flb_sds_t metadata_auth_header;
+
+    /* Metadata of Splunk Authentication */
+    flb_sds_t metadata_auth_key;
+    struct flb_record_accessor *ra_metadata_auth_key;
 
     /* Channel identifier */
     flb_sds_t channel;

plugins/out_splunk/splunk_conf.c

@@ -240,6 +240,7 @@ struct flb_splunk *flb_splunk_conf_create(struct flb_output_instance *ins,
         return NULL;
     }
 
+    ctx->metadata_auth_header = NULL;
     /* No http_user is set, fallback to splunk_token, if splunk_token is unset, fail. */
     if (!ctx->http_user) {
         /* Splunk Auth Token */
@@ -261,6 +262,21 @@ struct flb_splunk *flb_splunk_conf_create(struct flb_output_instance *ins,
         }
     }
 
+    /* Currently, Splunk HEC token is stored in a fixed key, hec_token. */
+    ctx->metadata_auth_key = "hec_token";
+    if (ctx->metadata_auth_key) {
+        ctx->ra_metadata_auth_key = flb_ra_create(ctx->metadata_auth_key, FLB_TRUE);
+        if (!ctx->ra_metadata_auth_key) {
+            flb_plg_error(ctx->ins,
+                          "cannot create record accessor for "
+                          "metadata_auth_key pattern: '%s'",
+                          ctx->event_host);
+            flb_splunk_conf_destroy(ctx);
+            return NULL;
+        }
+    }
+
+
     /* channel */
     if (ctx->channel != NULL) {
         ctx->channel_len = flb_sds_len(ctx->channel);
@@ -305,6 +321,10 @@ int flb_splunk_conf_destroy(struct flb_splunk *ctx)
         flb_ra_destroy(ctx->ra_event_index_key);
     }
 
+    if (ctx->ra_metadata_auth_key) {
+        flb_ra_destroy(ctx->ra_metadata_auth_key);
+    }
+
     event_fields_destroy(ctx);
 
     flb_free(ctx);