All code for 1.6 Pen Test

CMakeLists.txt

@@ -174,6 +174,8 @@ option(FLB_OUT_LOGDNA         "Enable LogDNA output plugin"         Yes)
 option(FLB_OUT_KAFKA          "Enable Kafka output plugin"           No)
 option(FLB_OUT_KAFKA_REST     "Enable Kafka Rest output plugin"     Yes)
 option(FLB_OUT_CLOUDWATCH_LOGS "Enable AWS CloudWatch output plugin" Yes)
+option(FLB_OUT_KINESIS_FIREHOSE "Enable AWS Firehose output plugin" Yes)
+option(FLB_OUT_S3             "Enable AWS S3 output plugin"         Yes)
 option(FLB_FILTER_ALTER_SIZE  "Enable alter_size filter"            Yes)
 option(FLB_FILTER_AWS         "Enable aws filter"                   Yes)
 option(FLB_FILTER_EXPECT      "Enable expect filter"                Yes)
@@ -224,6 +226,7 @@ if(FLB_ALL)
   set(FLB_OUT_RETRY    1)
   set(FLB_OUT_TD       1)
   set(FLB_OUT_STDOUT   1)
+  set(FLB_OUT_S3       1)
   set(FLB_OUT_SYSLOG   1)
   set(FLB_OUT_LIB      1)
   set(FLB_OUT_FLOWCOUNTER 1)

include/fluent-bit/flb_aws_credentials.h

@@ -229,8 +229,6 @@ struct flb_aws_provider *flb_profile_provider_create();
 
 time_t flb_aws_cred_expiration(const char* timestamp);
 
-int flb_read_file(const char *path, char **out_buf, size_t *out_size);
-
 struct flb_aws_credentials *flb_parse_sts_resp(char *response,
                                                time_t *expiration);
 flb_sds_t flb_sts_uri(char *action, char *role_arn, char *session_name,

include/fluent-bit/flb_aws_util.h

@@ -25,9 +25,6 @@
 
 #define FLB_AWS_UTIL_H
 
-#define AWS_SERVICE_ENDPOINT_FORMAT            "%s.%s.amazonaws.com"
-#define AWS_SERVICE_ENDPOINT_BASE_LEN          15
-
 #define FLB_AWS_CREDENTIAL_REFRESH_LIMIT       60
 
 /*
@@ -71,6 +68,7 @@ struct flb_aws_client {
 
     /* Sigv4 */
     int has_auth;
+    int s3_mode;
     struct flb_aws_provider *provider;
     char *region;
     char *service;
@@ -129,24 +127,39 @@ struct flb_aws_client_generator *flb_aws_client_generator();
  */
 char *flb_aws_endpoint(char* service, char* region);
 
+char *flb_s3_endpoint(char* bucket, char* region);
+
+/* Parses AWS XML API Error responses and returns the value of the <code> tag */
+flb_sds_t flb_aws_xml_error(char *response, size_t response_len);
+
 /*
- * Parses an AWS API error type returned by a request.
+ * Parses an AWS JSON API error type returned by a request.
  */
 flb_sds_t flb_aws_error(char *response, size_t response_len);
 
 /*
- * Similar to 'flb_aws_error', except it prints the error type and message
+ * Similar to 'flb_aws_error', except it prints the JSON error type and message
  * to the user in a error log.
  * 'api' is the name of the API that was called; this is used in the error log.
  */
 void flb_aws_print_error(char *response, size_t response_len,
                          char *api, struct flb_output_instance *ins);
 
+/* Similar to 'flb_aws_print_error', but for APIs that return XML */
+void flb_aws_print_xml_error(char *response, size_t response_len,
+                             char *api, struct flb_output_instance *ins);
+
 /*
  * Parses the JSON and gets the value for 'key'
  */
 flb_sds_t flb_json_get_val(char *response, size_t response_len, char *key);
 
+/*
+ * Parses an XML document and returns the value of the given tag
+ * Param `tag` should include angle brackets; ex "<code>"
+ */
+flb_sds_t flb_xml_get_val(char *response, size_t response_len, char *tag);
+
 /*
  * Request data from an IMDS path.
  */
@@ -158,6 +171,10 @@ int flb_imds_request(struct flb_aws_client *client, char *metadata_path,
  */
 int flb_aws_is_auth_error(char *payload, size_t payload_size);
 
+int flb_read_file(const char *path, char **out_buf, size_t *out_size);
+
+//* Constructs S3 object key as per the format. */
+flb_sds_t flb_get_s3_key(const char *format, time_t time, const char *tag, char *tag_delimiter);
 
 #endif
 #endif /* FLB_HAVE_AWS */

include/fluent-bit/flb_s3_local_buffer.h

@@ -0,0 +1,77 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2019-2020 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#ifdef FLB_HAVE_AWS
+
+#ifndef flb_local_buffer_H
+#define flb_local_buffer_H
+
+struct flb_local_chunk {
+    /* identifies this chunk in the buffer dir; created with simple_hash fn */
+    flb_sds_t key;
+    /* the original fluent tag for this data */
+    flb_sds_t tag;
+    flb_sds_t file_path;
+    size_t size;
+    struct timespec ts;
+    time_t create_time;
+
+    struct mk_list _head;
+};
+
+struct flb_local_buffer {
+    char *dir;
+    struct flb_output_instance *ins;
+
+    struct mk_list chunks;
+};
+
+/*
+ * "Initializes" the local buffer from the file system
+ * Reads buffer directory and finds any existing files
+ * This ensures the plugin will still send buffered data even if FB is restarted
+ */
+int flb_init_local_buffer(struct flb_local_buffer *store);
+
+/*
+ * Stores data in the local file system
+ * 'c' should be NULL if no local chunk suitable for this data has been created yet
+ */
+int flb_buffer_put(struct flb_local_buffer *store, struct flb_local_chunk *c,
+                   const char *tag, char *data, size_t bytes);
+
+/*
+ * Returns the chunk associated with the given tag
+ */
+struct flb_local_chunk *flb_chunk_get(struct flb_local_buffer *store, const char *tag);
+
+/*
+ * Recursively creates directories
+ */
+int flb_mkdir_all(const char *dir);
+
+/* Removes all files associated with a chunk once it has been removed */
+int flb_remove_chunk_files(struct flb_local_chunk *c);
+
+void flb_chunk_destroy(struct flb_local_chunk *c);
+
+void flb_local_buffer_destroy_chunks(struct flb_local_buffer *store);
+
+#endif
+#endif /* FLB_HAVE_AWS */

include/fluent-bit/flb_signv4.h

@@ -27,13 +27,20 @@
 #ifndef FLB_SIGNV4_H
 #define FLB_SIGNV4_H
 
+/* Request is not Amazon S3 PutObject */
+#define S3_MODE_NONE             0
+/* Set the x-amz-content-sha256 header with the sha value */
+#define S3_MODE_SIGNED_PAYLOAD   1
+/* Set the x-amz-content-sha256 header with the value UNSIGNED-PAYLOAD */
+#define S3_MODE_UNSIGNED_PAYLOAD 2
 
 flb_sds_t flb_signv4_uri_normalize_path(char *uri, size_t len);
 
 flb_sds_t flb_signv4_do(struct flb_http_client *c, int normalize_uri,
                         int amz_date,
                         time_t t_now,
                         char *region, char *service,
+                        int s3_mode,
                         struct flb_aws_provider *provider);
 
 #endif

plugins/CMakeLists.txt

@@ -199,6 +199,8 @@ REGISTER_OUT_PLUGIN("out_lib")
 REGISTER_OUT_PLUGIN("out_flowcounter")
 REGISTER_OUT_PLUGIN("out_gelf")
 REGISTER_OUT_PLUGIN("out_cloudwatch_logs")
+REGISTER_OUT_PLUGIN("out_s3")
+REGISTER_OUT_PLUGIN("out_kinesis_firehose")
 
 # FILTERS
 # =======

plugins/out_es/es.c

@@ -62,6 +62,7 @@ static flb_sds_t add_aws_auth(struct flb_http_client *c,
 
     signature = flb_signv4_do(c, FLB_TRUE, FLB_TRUE, time(NULL),
                               ctx->aws_region, "es",
+                              0,
                               ctx->aws_provider);
     if (!signature) {
         flb_plg_error(ctx->ins, "could not sign request with sigv4");

plugins/out_kinesis_firehose/CMakeLists.txt

@@ -0,0 +1,5 @@
+set(src
+  firehose.c
+  firehose_api.c)
+
+FLB_PLUGIN(out_kinesis_firehose "${src}" "")