/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ /* Fluent Bit * ========== * Copyright (C) 2015-2022 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. */ #include #include #include #include #include #include #include "http.h" #include "http_conn.h" #define HTTP_CONTENT_JSON 0 #define HTTP_CONTENT_URLENCODED 1 static inline char hex2nibble(char c) { if ((c >= 0x30) && (c <= '9')) { return c - 0x30; } // 0x30-0x39 are digits, 0x41-0x46 A-F, // so there is a gap at 0x40 if ((c >= 'A') && (c <= 'F')) { return (c - 'A') + 10; } if ((c >= 'a') && (c <= 'f')) { return (c - 'a') + 10; } return 0; } static int sds_uri_decode(flb_sds_t s) { char buf[1024]; char *optr; char *iptr; for (optr = buf, iptr = s; iptr < s + flb_sds_len(s) && optr-buf < sizeof(buf); iptr++) { if (*iptr == '%') { if (iptr+2 > (s + flb_sds_len(s))) { return -1; } *optr++ = hex2nibble(*(iptr+1)) << 4 | hex2nibble(*(iptr+2)); iptr+=2; } else if (*iptr == '+') { *optr++ = ' '; } else { *optr++ = *iptr; } } memcpy(s, buf, optr-buf); s[optr-buf] = '\0'; flb_sds_len_set(s, (optr-buf)); return 0; } static int send_response(struct http_conn *conn, int http_status, char *message) { struct flb_http *context; size_t sent; int len; flb_sds_t out; context = (struct flb_http *) conn->ctx; out = flb_sds_create_size(256); if (!out) { return -1; } if (message) { len = strlen(message); } else { len = 0; } if (http_status == 201) { flb_sds_printf(&out, "HTTP/1.1 201 Created \r\n" "Server: Fluent Bit v%s\r\n" "%s" "Content-Length: 0\r\n\r\n", FLB_VERSION_STR, context->success_headers_str); } else if (http_status == 200) { flb_sds_printf(&out, "HTTP/1.1 200 OK\r\n" "Server: Fluent Bit v%s\r\n" "%s" "Content-Length: 0\r\n\r\n", FLB_VERSION_STR, context->success_headers_str); } else if (http_status == 204) { flb_sds_printf(&out, "HTTP/1.1 204 No Content\r\n" "Server: Fluent Bit v%s\r\n" "%s" "\r\n\r\n", FLB_VERSION_STR, context->success_headers_str); } else if (http_status == 400) { flb_sds_printf(&out, "HTTP/1.1 400 Forbidden\r\n" "Server: Fluent Bit v%s\r\n" "Content-Length: %i\r\n\r\n%s", FLB_VERSION_STR, len, message); } /* We should check this operations result */ flb_io_net_write(conn->connection, (void *) out, flb_sds_len(out), &sent); flb_sds_destroy(out); return 0; } /* implements functionality to get tag from key in record */ static flb_sds_t tag_key(struct flb_http *ctx, msgpack_object *map) { size_t map_size = map->via.map.size; msgpack_object_kv *kv; msgpack_object key; msgpack_object val; char *key_str = NULL; char *val_str = NULL; size_t key_str_size = 0; size_t val_str_size = 0; int j; int check = FLB_FALSE; int found = FLB_FALSE; flb_sds_t tag; kv = map->via.map.ptr; for(j=0; j < map_size; j++) { check = FLB_FALSE; found = FLB_FALSE; key = (kv+j)->key; if (key.type == MSGPACK_OBJECT_BIN) { key_str = (char *) key.via.bin.ptr; key_str_size = key.via.bin.size; check = FLB_TRUE; } if (key.type == MSGPACK_OBJECT_STR) { key_str = (char *) key.via.str.ptr; key_str_size = key.via.str.size; check = FLB_TRUE; } if (check == FLB_TRUE) { if (strncmp(ctx->tag_key, key_str, key_str_size) == 0) { val = (kv+j)->val; if (val.type == MSGPACK_OBJECT_BIN) { val_str = (char *) val.via.bin.ptr; val_str_size = val.via.str.size; found = FLB_TRUE; break; } if (val.type == MSGPACK_OBJECT_STR) { val_str = (char *) val.via.str.ptr; val_str_size = val.via.str.size; found = FLB_TRUE; break; } } } } if (found == FLB_TRUE) { tag = flb_sds_create_len(val_str, val_str_size); if (!tag) { flb_errno(); return NULL; } return tag; } flb_plg_error(ctx->ins, "Could not find tag_key %s in record", ctx->tag_key); return NULL; } int process_pack(struct flb_http *ctx, flb_sds_t tag, char *buf, size_t size) { int ret; size_t off = 0; msgpack_unpacked result; struct flb_time tm; int i = 0; msgpack_object *obj; msgpack_object record; flb_sds_t tag_from_record = NULL; flb_time_get(&tm); msgpack_unpacked_init(&result); while (msgpack_unpack_next(&result, buf, size, &off) == MSGPACK_UNPACK_SUCCESS) { if (result.data.type == MSGPACK_OBJECT_MAP) { tag_from_record = NULL; if (ctx->tag_key) { obj = &result.data; tag_from_record = tag_key(ctx, obj); } ret = flb_log_event_encoder_begin_record(&ctx->log_encoder); if (ret == FLB_EVENT_ENCODER_SUCCESS) { ret = flb_log_event_encoder_set_timestamp( &ctx->log_encoder, &tm); } if (ret == FLB_EVENT_ENCODER_SUCCESS) { ret = flb_log_event_encoder_set_body_from_msgpack_object( &ctx->log_encoder, &result.data); } if (ret == FLB_EVENT_ENCODER_SUCCESS) { ret = flb_log_event_encoder_commit_record(&ctx->log_encoder); } if (ret == FLB_EVENT_ENCODER_SUCCESS) { if (tag_from_record) { flb_input_log_append(ctx->ins, tag_from_record, flb_sds_len(tag_from_record), ctx->log_encoder.output_buffer, ctx->log_encoder.output_length); flb_sds_destroy(tag_from_record); } else if (tag) { flb_input_log_append(ctx->ins, tag, flb_sds_len(tag), ctx->log_encoder.output_buffer, ctx->log_encoder.output_length); } else { /* use default plugin Tag (it internal name, e.g: http.0 */ flb_input_log_append(ctx->ins, NULL, 0, ctx->log_encoder.output_buffer, ctx->log_encoder.output_length); } } else { flb_plg_error(ctx->ins, "Error encoding record : %d", ret); } flb_log_event_encoder_reset(&ctx->log_encoder); } else if (result.data.type == MSGPACK_OBJECT_ARRAY) { obj = &result.data; for (i = 0; i < obj->via.array.size; i++) { record = obj->via.array.ptr[i]; tag_from_record = NULL; if (ctx->tag_key) { tag_from_record = tag_key(ctx, &record); } ret = flb_log_event_encoder_begin_record(&ctx->log_encoder); if (ret == FLB_EVENT_ENCODER_SUCCESS) { ret = flb_log_event_encoder_set_timestamp( &ctx->log_encoder, &tm); } if (ret == FLB_EVENT_ENCODER_SUCCESS) { ret = flb_log_event_encoder_set_body_from_msgpack_object( &ctx->log_encoder, &record); } if (ret == FLB_EVENT_ENCODER_SUCCESS) { ret = flb_log_event_encoder_commit_record(&ctx->log_encoder); } if (ret == FLB_EVENT_ENCODER_SUCCESS) { if (tag_from_record) { flb_input_log_append(ctx->ins, tag_from_record, flb_sds_len(tag_from_record), ctx->log_encoder.output_buffer, ctx->log_encoder.output_length); flb_sds_destroy(tag_from_record); } else if (tag) { flb_input_log_append(ctx->ins, tag, flb_sds_len(tag), ctx->log_encoder.output_buffer, ctx->log_encoder.output_length); } else { /* use default plugin Tag (it internal name, e.g: http.0 */ flb_input_log_append(ctx->ins, NULL, 0, ctx->log_encoder.output_buffer, ctx->log_encoder.output_length); } } else { flb_plg_error(ctx->ins, "Error encoding record : %d", ret); } /* TODO : Optimize this * * This is wasteful, considering that we are emitting a series * of records we should start and commit each one and then * emit them all at once after the loop. */ flb_log_event_encoder_reset(&ctx->log_encoder); } break; } else { flb_plg_error(ctx->ins, "skip record from invalid type: %i", result.data.type); msgpack_unpacked_destroy(&result); return -1; } } msgpack_unpacked_destroy(&result); return 0; } static ssize_t parse_payload_json(struct flb_http *ctx, flb_sds_t tag, char *payload, size_t size) { int ret; int out_size; char *pack; struct flb_pack_state pack_state; /* Initialize packer */ flb_pack_state_init(&pack_state); /* Pack JSON as msgpack */ ret = flb_pack_json_state(payload, size, &pack, &out_size, &pack_state); flb_pack_state_reset(&pack_state); /* Handle exceptions */ if (ret == FLB_ERR_JSON_PART) { flb_plg_warn(ctx->ins, "JSON data is incomplete, skipping"); return -1; } else if (ret == FLB_ERR_JSON_INVAL) { flb_plg_warn(ctx->ins, "invalid JSON message, skipping"); return -1; } else if (ret == -1) { return -1; } /* Process the packaged JSON and return the last byte used */ process_pack(ctx, tag, pack, out_size); flb_free(pack); return 0; } static ssize_t parse_payload_urlencoded(struct flb_http *ctx, flb_sds_t tag, char *payload, size_t size) { struct mk_list *kvs; struct mk_list *head = NULL; struct flb_split_entry *cur = NULL; char **keys = NULL; char **vals = NULL; char *sep; char *start; int idx = 0; int ret = -1; msgpack_packer pck; msgpack_sbuffer sbuf; /* initialize buffers */ msgpack_sbuffer_init(&sbuf); msgpack_packer_init(&pck, &sbuf, msgpack_sbuffer_write); kvs = flb_utils_split(payload, '&', -1 ); if (kvs == NULL) { goto split_error; } keys = flb_calloc(mk_list_size(kvs), sizeof(char *)); if (keys == NULL) { goto keys_calloc_error; } vals = flb_calloc(mk_list_size(kvs), sizeof(char *)); if (vals == NULL) { goto vals_calloc_error; } mk_list_foreach(head, kvs) { cur = mk_list_entry(head, struct flb_split_entry, _head); if (cur->value[0] == '\n') { start = &cur->value[1]; } else { start = cur->value; } sep = strchr(start, '='); if (sep == NULL) { vals[idx] = NULL; continue; } *sep++ = '\0'; keys[idx] = flb_sds_create_len(start, strlen(start)); vals[idx] = flb_sds_create_len(sep, strlen(sep)); flb_sds_trim(keys[idx]); flb_sds_trim(vals[idx]); idx++; } msgpack_pack_map(&pck, mk_list_size(kvs)); for (idx = 0; idx < mk_list_size(kvs); idx++) { msgpack_pack_str(&pck, flb_sds_len(keys[idx])); msgpack_pack_str_body(&pck, keys[idx], flb_sds_len(keys[idx])); if (sds_uri_decode(vals[idx]) != 0) { goto decode_error; } else { msgpack_pack_str(&pck, flb_sds_len(vals[idx])); msgpack_pack_str_body(&pck, vals[idx], strlen(vals[idx])); } } ret = process_pack(ctx, tag, sbuf.data, sbuf.size); decode_error: for (idx = 0; idx < mk_list_size(kvs); idx++) { if (keys[idx]) { flb_sds_destroy(keys[idx]); } if (vals[idx]) { flb_sds_destroy(vals[idx]); } } flb_free(vals); vals_calloc_error: flb_free(keys); keys_calloc_error: flb_utils_split_free(kvs); split_error: msgpack_sbuffer_destroy(&sbuf); return ret; } static int process_payload(struct flb_http *ctx, struct http_conn *conn, flb_sds_t tag, struct mk_http_session *session, struct mk_http_request *request) { int type = -1; struct mk_http_header *header; header = &session->parser.headers[MK_HEADER_CONTENT_TYPE]; if (header->key.data == NULL) { send_response(conn, 400, "error: header 'Content-Type' is not set\n"); return -1; } if (header->val.len == 16 && strncasecmp(header->val.data, "application/json", 16) == 0) { type = HTTP_CONTENT_JSON; } if (header->val.len == 33 && strncasecmp(header->val.data, "application/x-www-form-urlencoded", 33) == 0) { type = HTTP_CONTENT_URLENCODED; } if (type == -1) { send_response(conn, 400, "error: invalid 'Content-Type'\n"); return -1; } if (request->data.len <= 0) { send_response(conn, 400, "error: no payload found\n"); return -1; } if (type == HTTP_CONTENT_JSON) { parse_payload_json(ctx, tag, request->data.data, request->data.len); } else if (type == HTTP_CONTENT_URLENCODED) { parse_payload_urlencoded(ctx, tag, request->data.data, request->data.len); } return 0; } static inline int mk_http_point_header(mk_ptr_t *h, struct mk_http_parser *parser, int key) { struct mk_http_header *header; header = &parser->headers[key]; if (header->type == key) { h->data = header->val.data; h->len = header->val.len; return 0; } else { h->data = NULL; h->len = -1; } return -1; } /* * Handle an incoming request. It perform extra checks over the request, if * everything is OK, it enqueue the incoming payload. */ int http_prot_handle(struct flb_http *ctx, struct http_conn *conn, struct mk_http_session *session, struct mk_http_request *request) { int i; int ret; int len; char *uri; char *qs; off_t diff; flb_sds_t tag; struct mk_http_header *header; if (request->uri.data[0] != '/') { send_response(conn, 400, "error: invalid request\n"); return -1; } /* Decode URI */ uri = mk_utils_url_decode(request->uri); if (!uri) { uri = mk_mem_alloc_z(request->uri.len + 1); if (!uri) { return -1; } memcpy(uri, request->uri.data, request->uri.len); uri[request->uri.len] = '\0'; } /* Try to match a query string so we can remove it */ qs = strchr(uri, '?'); if (qs) { /* remove the query string part */ diff = qs - uri; uri[diff] = '\0'; } /* Compose the query string using the URI */ len = strlen(uri); if (len == 1) { tag = NULL; /* use default tag */ } else { tag = flb_sds_create_size(len); if (!tag) { mk_mem_free(uri); return -1; } /* New tag skipping the URI '/' */ flb_sds_cat(tag, uri + 1, len - 1); /* Sanitize, only allow alphanum chars */ for (i = 0; i < flb_sds_len(tag); i++) { if (!isalnum(tag[i]) && tag[i] != '_' && tag[i] != '.') { tag[i] = '_'; } } } mk_mem_free(uri); /* Check if we have a Host header: Hostname ; port */ mk_http_point_header(&request->host, &session->parser, MK_HEADER_HOST); /* Header: Connection */ mk_http_point_header(&request->connection, &session->parser, MK_HEADER_CONNECTION); /* HTTP/1.1 needs Host header */ if (!request->host.data && request->protocol == MK_HTTP_PROTOCOL_11) { flb_sds_destroy(tag); return -1; } /* Should we close the session after this request ? */ mk_http_keepalive_check(session, request, ctx->server); /* Content Length */ header = &session->parser.headers[MK_HEADER_CONTENT_LENGTH]; if (header->type == MK_HEADER_CONTENT_LENGTH) { request->_content_length.data = header->val.data; request->_content_length.len = header->val.len; } else { request->_content_length.data = NULL; } if (request->method != MK_METHOD_POST) { flb_sds_destroy(tag); send_response(conn, 400, "error: invalid HTTP method\n"); return -1; } ret = process_payload(ctx, conn, tag, session, request); flb_sds_destroy(tag); send_response(conn, ctx->successful_response_code, NULL); return ret; } /* * Handle an incoming request which has resulted in an http parser error. */ int http_prot_handle_error(struct flb_http *ctx, struct http_conn *conn, struct mk_http_session *session, struct mk_http_request *request) { send_response(conn, 400, "error: invalid request\n"); return -1; }