diff --git a/plugins/out_es/es.c b/plugins/out_es/es.c index 459da80cbe5..bd724c587d4 100644 --- a/plugins/out_es/es.c +++ b/plugins/out_es/es.c @@ -235,6 +235,21 @@ static flb_sds_t es_get_id_value(struct flb_elasticsearch *ctx, return tmp_str; } +static int es_action_line_value_is_safe(const char *value, size_t len) +{ + size_t i; + unsigned char c; + + for (i = 0; i < len; i++) { + c = (unsigned char) value[i]; + if (c == '\n' || c == '\r' || c == '"' || c == '\\' || c < 0x20) { + return FLB_FALSE; + } + } + + return FLB_TRUE; +} + static int compose_index_header(struct flb_elasticsearch *ctx, int es_index_custom_len, char *logstash_index, size_t logstash_index_size, @@ -293,6 +308,10 @@ static int elasticsearch_format(struct flb_config *config, int len; int map_size; int index_len = 0; + int write_op_update = FLB_FALSE; + int write_op_upsert = FLB_FALSE; + int id_key_required = FLB_FALSE; + int id_key_safe; size_t s = 0; size_t off = 0; size_t off_prev = 0; @@ -345,10 +364,16 @@ static int elasticsearch_format(struct flb_config *config, return -1; } - /* Copy logstash prefix if logstash format is enabled */ - if (ctx->logstash_format == FLB_TRUE) { - strncpy(logstash_index, ctx->logstash_prefix, sizeof(logstash_index)); - logstash_index[sizeof(logstash_index) - 1] = '\0'; + if (strcasecmp(ctx->write_operation, FLB_ES_WRITE_OP_UPDATE) == 0) { + write_op_update = FLB_TRUE; + } + else if (strcasecmp(ctx->write_operation, FLB_ES_WRITE_OP_UPSERT) == 0) { + write_op_upsert = FLB_TRUE; + } + + if (ctx->ra_id_key && ctx->generate_id == FLB_FALSE && + (write_op_update == FLB_TRUE || write_op_upsert == FLB_TRUE)) { + id_key_required = FLB_TRUE; } /* @@ -403,6 +428,12 @@ static int elasticsearch_format(struct flb_config *config, map = *log_event.body; map_size = map.via.map.size; + /* Copy logstash prefix for the per-record fallback path. */ + if (ctx->logstash_format == FLB_TRUE) { + strncpy(logstash_index, ctx->logstash_prefix, sizeof(logstash_index)); + logstash_index[sizeof(logstash_index) - 1] = '\0'; + } + es_index_custom_len = 0; if (ctx->logstash_prefix_key) { flb_sds_t v = flb_ra_translate(ctx->ra_prefix_key, @@ -410,14 +441,16 @@ static int elasticsearch_format(struct flb_config *config, map, NULL); if (v) { len = flb_sds_len(v); - if (len > 128) { - len = 128; - memcpy(logstash_index, v, 128); - } - else { - memcpy(logstash_index, v, len); + if (es_action_line_value_is_safe(v, len) == FLB_TRUE) { + if (len > 128) { + len = 128; + memcpy(logstash_index, v, 128); + } + else { + memcpy(logstash_index, v, len); + } + es_index_custom_len = len; } - es_index_custom_len = len; flb_sds_destroy(v); } } @@ -538,7 +571,15 @@ static int elasticsearch_format(struct flb_config *config, } if (ctx->ra_id_key) { id_key_str = es_get_id_value(ctx ,&map); - if (id_key_str) { + id_key_safe = FLB_FALSE; + + if (id_key_str && + es_action_line_value_is_safe(id_key_str, + flb_sds_len(id_key_str)) == FLB_TRUE) { + id_key_safe = FLB_TRUE; + } + + if (id_key_safe == FLB_TRUE) { if (ctx->suppress_type_name) { index_len = flb_sds_snprintf(&j_index, flb_sds_alloc(j_index), @@ -553,6 +594,35 @@ static int elasticsearch_format(struct flb_config *config, ctx->es_action, es_index, ctx->type, id_key_str); } + } + else if (id_key_required == FLB_TRUE) { + flb_plg_warn(ctx->ins, + "skipping record with missing or unsafe Id_Key value"); + if (id_key_str) { + flb_sds_destroy(id_key_str); + id_key_str = NULL; + } + msgpack_sbuffer_destroy(&tmp_sbuf); + continue; + } + else if (ctx->generate_id == FLB_FALSE && id_key_str) { + if (ctx->suppress_type_name) { + index_len = flb_sds_snprintf(&j_index, + flb_sds_alloc(j_index), + ES_BULK_INDEX_FMT_WITHOUT_TYPE, + ctx->es_action, + es_index); + } + else { + index_len = flb_sds_snprintf(&j_index, + flb_sds_alloc(j_index), + ES_BULK_INDEX_FMT, + ctx->es_action, + es_index, ctx->type); + } + } + + if (id_key_str) { flb_sds_destroy(id_key_str); id_key_str = NULL; } @@ -570,13 +640,13 @@ static int elasticsearch_format(struct flb_config *config, } out_buf_len = flb_sds_len(out_buf); - if (strcasecmp(ctx->write_operation, FLB_ES_WRITE_OP_UPDATE) == 0) { + if (write_op_update == FLB_TRUE) { tmp_buf = out_buf; out_buf = flb_sds_create_len(NULL, out_buf_len = out_buf_len + sizeof(ES_BULK_UPDATE_OP_BODY) - 2); out_buf_len = snprintf(out_buf, out_buf_len, ES_BULK_UPDATE_OP_BODY, tmp_buf); flb_sds_destroy(tmp_buf); } - else if (strcasecmp(ctx->write_operation, FLB_ES_WRITE_OP_UPSERT) == 0) { + else if (write_op_upsert == FLB_TRUE) { tmp_buf = out_buf; out_buf = flb_sds_create_len(NULL, out_buf_len = out_buf_len + sizeof(ES_BULK_UPSERT_OP_BODY) - 2); out_buf_len = snprintf(out_buf, out_buf_len, ES_BULK_UPSERT_OP_BODY, tmp_buf); @@ -845,6 +915,12 @@ static void cb_es_flush(struct flb_event_chunk *event_chunk, FLB_OUTPUT_RETURN(FLB_ERROR); } + if (out_size == 0) { + flb_free(out_buf); + flb_upstream_conn_release(u_conn); + FLB_OUTPUT_RETURN(FLB_OK); + } + pack = (char *) out_buf; pack_size = out_size; diff --git a/plugins/out_opensearch/opensearch.c b/plugins/out_opensearch/opensearch.c index db654be6d73..5541deea0da 100644 --- a/plugins/out_opensearch/opensearch.c +++ b/plugins/out_opensearch/opensearch.c @@ -225,6 +225,21 @@ static flb_sds_t os_get_id_value(struct flb_opensearch *ctx, return tmp_str; } +static int os_action_line_value_is_safe(const char *value, size_t len) +{ + size_t i; + unsigned char c; + + for (i = 0; i < len; i++) { + c = (unsigned char) value[i]; + if (c == '\n' || c == '\r' || c == '"' || c == '\\' || c < 0x20) { + return FLB_FALSE; + } + } + + return FLB_TRUE; +} + static int compose_index_header(struct flb_opensearch *ctx, int index_custom_len, char *logstash_index, size_t logstash_index_size, @@ -283,6 +298,8 @@ static int opensearch_format(struct flb_config *config, int index_len = 0; int write_op_update = FLB_FALSE; int write_op_upsert = FLB_FALSE; + int id_key_required = FLB_FALSE; + int id_key_safe; flb_sds_t ra_index = NULL; size_t s = 0; char *index = NULL; @@ -330,10 +347,16 @@ static int opensearch_format(struct flb_config *config, return -1; } - /* Copy logstash prefix if logstash format is enabled */ - if (ctx->logstash_format == FLB_TRUE) { - strncpy(logstash_index, ctx->logstash_prefix, sizeof(logstash_index)); - logstash_index[sizeof(logstash_index) - 1] = '\0'; + if (strcasecmp(ctx->write_operation, FLB_OS_WRITE_OP_UPDATE) == 0) { + write_op_update = FLB_TRUE; + } + else if (strcasecmp(ctx->write_operation, FLB_OS_WRITE_OP_UPSERT) == 0) { + write_op_upsert = FLB_TRUE; + } + + if (ctx->ra_id_key && ctx->generate_id == FLB_FALSE && + (write_op_update == FLB_TRUE || write_op_upsert == FLB_TRUE)) { + id_key_required = FLB_TRUE; } /* @@ -393,6 +416,12 @@ static int opensearch_format(struct flb_config *config, map = *log_event.body; map_size = map.via.map.size; + /* Copy logstash prefix for the per-record fallback path. */ + if (ctx->logstash_format == FLB_TRUE) { + strncpy(logstash_index, ctx->logstash_prefix, sizeof(logstash_index)); + logstash_index[sizeof(logstash_index) - 1] = '\0'; + } + index_custom_len = 0; if (ctx->logstash_prefix_key) { flb_sds_t v = flb_ra_translate(ctx->ra_prefix_key, @@ -400,15 +429,17 @@ static int opensearch_format(struct flb_config *config, map, NULL); if (v) { len = flb_sds_len(v); - if (len > 128) { - len = 128; - memcpy(logstash_index, v, 128); - } - else { - memcpy(logstash_index, v, len); - } + if (os_action_line_value_is_safe(v, len) == FLB_TRUE) { + if (len > 128) { + len = 128; + memcpy(logstash_index, v, 128); + } + else { + memcpy(logstash_index, v, len); + } - index_custom_len = len; + index_custom_len = len; + } flb_sds_destroy(v); } } @@ -492,6 +523,11 @@ static int opensearch_format(struct flb_config *config, if (!ra_index) { flb_plg_warn(ctx->ins, "invalid index translation from record accessor pattern, default to static index"); } + else if (os_action_line_value_is_safe(ra_index, + flb_sds_len(ra_index)) == FLB_FALSE) { + flb_sds_destroy(ra_index); + ra_index = NULL; + } else { index = ra_index; } @@ -566,7 +602,15 @@ static int opensearch_format(struct flb_config *config, } if (ctx->ra_id_key) { id_key_str = os_get_id_value(ctx ,&map); - if (id_key_str) { + id_key_safe = FLB_FALSE; + + if (id_key_str && + os_action_line_value_is_safe(id_key_str, + flb_sds_len(id_key_str)) == FLB_TRUE) { + id_key_safe = FLB_TRUE; + } + + if (id_key_safe == FLB_TRUE) { if (ctx->suppress_type_name) { index_len = flb_sds_snprintf(&j_index, flb_sds_alloc(j_index), @@ -581,6 +625,35 @@ static int opensearch_format(struct flb_config *config, ctx->action, index, ctx->type, id_key_str); } + } + else if (id_key_required == FLB_TRUE) { + flb_plg_warn(ctx->ins, + "skipping record with missing or unsafe Id_Key value"); + if (id_key_str) { + flb_sds_destroy(id_key_str); + id_key_str = NULL; + } + msgpack_sbuffer_destroy(&tmp_sbuf); + continue; + } + else if (ctx->generate_id == FLB_FALSE && id_key_str) { + if (ctx->suppress_type_name) { + index_len = flb_sds_snprintf(&j_index, + flb_sds_alloc(j_index), + OS_BULK_INDEX_FMT_NO_TYPE, + ctx->action, + index); + } + else { + index_len = flb_sds_snprintf(&j_index, + flb_sds_alloc(j_index), + OS_BULK_INDEX_FMT, + ctx->action, + index, ctx->type); + } + } + + if (id_key_str) { flb_sds_destroy(id_key_str); id_key_str = NULL; } @@ -613,13 +686,6 @@ static int opensearch_format(struct flb_config *config, return -1; } - if (strcasecmp(ctx->write_operation, FLB_OS_WRITE_OP_UPDATE) == 0) { - write_op_update = FLB_TRUE; - } - else if (strcasecmp(ctx->write_operation, FLB_OS_WRITE_OP_UPSERT) == 0) { - write_op_upsert = FLB_TRUE; - } - /* UPDATE | UPSERT */ if (write_op_update) { flb_sds_cat_safe(&bulk, @@ -913,6 +979,12 @@ static void cb_opensearch_flush(struct flb_event_chunk *event_chunk, FLB_OUTPUT_RETURN(FLB_ERROR); } + if (out_size == 0) { + flb_sds_destroy(out_buf); + flb_upstream_conn_release(u_conn); + FLB_OUTPUT_RETURN(FLB_OK); + } + pack = (char *) out_buf; pack_size = out_size; diff --git a/tests/integration/scenarios/out_es/config/out_es_id_key_ndjson.yaml b/tests/integration/scenarios/out_es/config/out_es_id_key_ndjson.yaml new file mode 100644 index 00000000000..0bfb546ef0f --- /dev/null +++ b/tests/integration/scenarios/out_es/config/out_es_id_key_ndjson.yaml @@ -0,0 +1,23 @@ +service: + flush: 1 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: dummy + tag: out_es_id_key_ndjson + dummy: '{"doc_id":"legit\"\n{\"delete\":{\"_index\":\"audit-logs\",\"_id\":\"critical-audit-record-12345\"}}\n{\"create\":{\"_index\":\"pad\",\"_id\":\"x","message":"id-key injection"}' + samples: 1 + + outputs: + - name: es + match: out_es_id_key_ndjson + host: 127.0.0.1 + port: ${TEST_SUITE_HTTP_PORT} + index: fluent-bit + suppress_type_name: on + id_key: doc_id + retry_limit: 0 diff --git a/tests/integration/scenarios/out_es/config/out_es_id_key_update_ndjson.yaml b/tests/integration/scenarios/out_es/config/out_es_id_key_update_ndjson.yaml new file mode 100644 index 00000000000..4b36befeed6 --- /dev/null +++ b/tests/integration/scenarios/out_es/config/out_es_id_key_update_ndjson.yaml @@ -0,0 +1,33 @@ +service: + flush: 1 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: dummy + tag: out_es_id_key_update_ndjson + dummy: >- + {"doc_id":"legit\"\n + {\"delete\":{\"_index\":\"audit-logs\",\"_id\":\"critical-audit-record-12345\"}}\n + {\"update\":{\"_index\":\"pad\",\"_id\":\"x", + "message":"unsafe update id"} + samples: 1 + + - name: dummy + tag: out_es_id_key_update_ndjson + dummy: '{"doc_id":"safe-update-id","message":"safe update id"}' + samples: 1 + + outputs: + - name: es + match: out_es_id_key_update_ndjson + host: 127.0.0.1 + port: ${TEST_SUITE_HTTP_PORT} + index: fluent-bit + suppress_type_name: on + id_key: doc_id + write_operation: update + retry_limit: 0 diff --git a/tests/integration/scenarios/out_es/config/out_es_logstash_prefix_key_ndjson.yaml b/tests/integration/scenarios/out_es/config/out_es_logstash_prefix_key_ndjson.yaml new file mode 100644 index 00000000000..93b4f4b39b3 --- /dev/null +++ b/tests/integration/scenarios/out_es/config/out_es_logstash_prefix_key_ndjson.yaml @@ -0,0 +1,23 @@ +service: + flush: 1 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: dummy + tag: out_es_logstash_prefix_key_ndjson + dummy: '{"idx":"victim-index\"\n{\"delete\":{\"_index\":\"audit-logs\",\"_id\":\"critical-audit-record-12345\"}}\n{\"create\":{\"_index\":\"pad","message":"prefix injection"}' + samples: 1 + + outputs: + - name: es + match: out_es_logstash_prefix_key_ndjson + host: 127.0.0.1 + port: ${TEST_SUITE_HTTP_PORT} + logstash_format: on + logstash_prefix_key: idx + suppress_type_name: on + retry_limit: 0 diff --git a/tests/integration/scenarios/out_es/config/out_opensearch_id_key_ndjson.yaml b/tests/integration/scenarios/out_es/config/out_opensearch_id_key_ndjson.yaml new file mode 100644 index 00000000000..0762e84591c --- /dev/null +++ b/tests/integration/scenarios/out_es/config/out_opensearch_id_key_ndjson.yaml @@ -0,0 +1,23 @@ +service: + flush: 1 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: dummy + tag: out_opensearch_id_key_ndjson + dummy: '{"doc_id":"legit\"\n{\"delete\":{\"_index\":\"audit-logs\",\"_id\":\"critical-audit-record-12345\"}}\n{\"create\":{\"_index\":\"pad\",\"_id\":\"x","message":"opensearch id-key injection"}' + samples: 1 + + outputs: + - name: opensearch + match: out_opensearch_id_key_ndjson + host: 127.0.0.1 + port: ${TEST_SUITE_HTTP_PORT} + index: fluent-bit + suppress_type_name: on + id_key: doc_id + retry_limit: 0 diff --git a/tests/integration/scenarios/out_es/config/out_opensearch_id_key_update_ndjson.yaml b/tests/integration/scenarios/out_es/config/out_opensearch_id_key_update_ndjson.yaml new file mode 100644 index 00000000000..4a1da4e8be4 --- /dev/null +++ b/tests/integration/scenarios/out_es/config/out_opensearch_id_key_update_ndjson.yaml @@ -0,0 +1,33 @@ +service: + flush: 1 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: dummy + tag: out_opensearch_id_key_update_ndjson + dummy: >- + {"doc_id":"legit\"\n + {\"delete\":{\"_index\":\"audit-logs\",\"_id\":\"critical-audit-record-12345\"}}\n + {\"update\":{\"_index\":\"pad\",\"_id\":\"x", + "message":"unsafe opensearch update id"} + samples: 1 + + - name: dummy + tag: out_opensearch_id_key_update_ndjson + dummy: '{"doc_id":"safe-update-id","message":"safe opensearch update id"}' + samples: 1 + + outputs: + - name: opensearch + match: out_opensearch_id_key_update_ndjson + host: 127.0.0.1 + port: ${TEST_SUITE_HTTP_PORT} + index: fluent-bit + suppress_type_name: on + id_key: doc_id + write_operation: update + retry_limit: 0 diff --git a/tests/integration/scenarios/out_es/config/out_opensearch_index_record_accessor_ndjson.yaml b/tests/integration/scenarios/out_es/config/out_opensearch_index_record_accessor_ndjson.yaml new file mode 100644 index 00000000000..46e1623e0d6 --- /dev/null +++ b/tests/integration/scenarios/out_es/config/out_opensearch_index_record_accessor_ndjson.yaml @@ -0,0 +1,22 @@ +service: + flush: 1 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: dummy + tag: out_opensearch_index_record_accessor_ndjson + dummy: '{"idx":"victim-index\"\n{\"delete\":{\"_index\":\"audit-logs\",\"_id\":\"critical-audit-record-12345\"}}\n{\"create\":{\"_index\":\"pad","message":"opensearch index injection"}' + samples: 1 + + outputs: + - name: opensearch + match: out_opensearch_index_record_accessor_ndjson + host: 127.0.0.1 + port: ${TEST_SUITE_HTTP_PORT} + index: $idx + suppress_type_name: on + retry_limit: 0 diff --git a/tests/integration/scenarios/out_es/config/out_opensearch_logstash_prefix_key_ndjson.yaml b/tests/integration/scenarios/out_es/config/out_opensearch_logstash_prefix_key_ndjson.yaml new file mode 100644 index 00000000000..090802e6ba1 --- /dev/null +++ b/tests/integration/scenarios/out_es/config/out_opensearch_logstash_prefix_key_ndjson.yaml @@ -0,0 +1,23 @@ +service: + flush: 1 + grace: 1 + log_level: info + http_server: on + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + +pipeline: + inputs: + - name: dummy + tag: out_opensearch_logstash_prefix_key_ndjson + dummy: '{"idx":"victim-index\"\n{\"delete\":{\"_index\":\"audit-logs\",\"_id\":\"critical-audit-record-12345\"}}\n{\"create\":{\"_index\":\"pad","message":"opensearch prefix injection"}' + samples: 1 + + outputs: + - name: opensearch + match: out_opensearch_logstash_prefix_key_ndjson + host: 127.0.0.1 + port: ${TEST_SUITE_HTTP_PORT} + logstash_format: on + logstash_prefix_key: idx + suppress_type_name: on + retry_limit: 0 diff --git a/tests/integration/scenarios/out_es/tests/test_out_es_ndjson_action_line_001.py b/tests/integration/scenarios/out_es/tests/test_out_es_ndjson_action_line_001.py new file mode 100644 index 00000000000..4ca0c0a78e8 --- /dev/null +++ b/tests/integration/scenarios/out_es/tests/test_out_es_ndjson_action_line_001.py @@ -0,0 +1,199 @@ +import json +import os +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import threading + +import pytest + +from utils.test_service import FluentBitTestService + + +FORGED_DELETE_ID = "critical-audit-record-12345" +SAFE_UPDATE_ID = "safe-update-id" + + +class _BulkCaptureHandler(BaseHTTPRequestHandler): + def log_message(self, fmt, *args): + return + + def do_POST(self): + content_length = int(self.headers.get("Content-Length", "0")) + body = self.rfile.read(content_length) + + self.server.requests.append( + { + "path": self.path, + "headers": dict(self.headers), + "body": body.decode("utf-8", errors="replace"), + } + ) + + response = b'{"errors":false,"items":[{"create":{"status":201}}]}' + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(response))) + self.end_headers() + self.wfile.write(response) + + +class _BulkCaptureServer(ThreadingHTTPServer): + daemon_threads = True + allow_reuse_address = True + + def __init__(self, address): + super().__init__(address, _BulkCaptureHandler) + self.requests = [] + + +class Service: + def __init__(self, config_file): + self.config_file = os.path.abspath( + os.path.join(os.path.dirname(__file__), "../config", config_file) + ) + self.bulk_server = None + self.bulk_server_thread = None + self.service = FluentBitTestService( + self.config_file, + pre_start=self._start_receiver, + post_stop=self._stop_receiver, + ) + + def _start_receiver(self, service): + self.bulk_server = _BulkCaptureServer(("127.0.0.1", service.test_suite_http_port)) + self.bulk_server_thread = threading.Thread( + target=self.bulk_server.serve_forever, + daemon=True, + ) + self.bulk_server_thread.start() + + def _stop_receiver(self, service): + if self.bulk_server is None: + return + + self.bulk_server.shutdown() + self.bulk_server.server_close() + + if self.bulk_server_thread is not None: + self.bulk_server_thread.join(timeout=5) + + def start(self): + self.service.start() + + def stop(self): + self.service.stop() + + def wait_for_requests(self, minimum_count, timeout=10): + if os.environ.get("VALGRIND"): + timeout = max(timeout * 3, 30) + + return self.service.wait_for_condition( + lambda: self.bulk_server.requests + if len(self.bulk_server.requests) >= minimum_count + else None, + timeout=timeout, + interval=0.5, + description=f"{minimum_count} Elasticsearch bulk requests", + ) + + def wait_for_action_lines(self, minimum_count, timeout=10): + if os.environ.get("VALGRIND"): + timeout = max(timeout * 3, 30) + + return self.service.wait_for_condition( + lambda: self.bulk_server.requests + if sum(len(_bulk_action_lines(request["body"])) + for request in self.bulk_server.requests) >= minimum_count + else None, + timeout=timeout, + interval=0.5, + description=f"{minimum_count} Elasticsearch bulk action lines", + ) + + +def _bulk_action_lines(body): + actions = [] + + for line in body.splitlines(): + if not line: + continue + + try: + value = json.loads(line) + except json.JSONDecodeError: + continue + + if isinstance(value, dict) and any( + key in value for key in ("create", "index", "update", "delete") + ): + actions.append(value) + + return actions + + +def _assert_no_forged_delete(body): + actions = _bulk_action_lines(body) + deletes = [ + action["delete"] + for action in actions + if "delete" in action and action["delete"].get("_id") == FORGED_DELETE_ID + ] + + assert len(actions) == 1 + assert deletes == [] + + +def _bulk_actions(requests): + actions = [] + + for request in requests: + actions.extend(_bulk_action_lines(request["body"])) + + return actions + + +@pytest.mark.parametrize( + "config_file", + [ + "out_es_logstash_prefix_key_ndjson.yaml", + "out_es_id_key_ndjson.yaml", + "out_opensearch_logstash_prefix_key_ndjson.yaml", + "out_opensearch_index_record_accessor_ndjson.yaml", + "out_opensearch_id_key_ndjson.yaml", + ], +) +def test_record_accessor_values_do_not_forge_bulk_action_lines(config_file): + service = Service(config_file) + + try: + service.start() + requests_seen = service.wait_for_requests(1) + finally: + service.stop() + + bulk_body = requests_seen[0]["body"] + assert requests_seen[0]["path"].startswith("/_bulk") + _assert_no_forged_delete(bulk_body) + + +@pytest.mark.parametrize( + "config_file", + [ + "out_es_id_key_update_ndjson.yaml", + "out_opensearch_id_key_update_ndjson.yaml", + ], +) +def test_unsafe_required_id_key_does_not_emit_idless_update(config_file): + service = Service(config_file) + + try: + service.start() + requests_seen = service.wait_for_action_lines(1) + finally: + service.stop() + + actions = _bulk_actions(requests_seen) + updates = [action["update"] for action in actions if "update" in action] + + assert all(request["path"].startswith("/_bulk") for request in requests_seen) + assert len(actions) == 1 + assert updates == [{"_index": "fluent-bit", "_id": SAFE_UPDATE_ID}]