diff --git a/include/fluent-bit/flb_config.h b/include/fluent-bit/flb_config.h index 9cba1017ff2..bea5ab85633 100644 --- a/include/fluent-bit/flb_config.h +++ b/include/fluent-bit/flb_config.h @@ -334,6 +334,8 @@ struct flb_config { #ifdef FLB_HAVE_CHUNK_TRACE int enable_chunk_trace; + /* Comma separated host:port[/uri] destinations allowed for the trace http output */ + char *chunk_trace_http_allowed_endpoints; #endif /* FLB_HAVE_CHUNK_TRACE */ int fips_mode; @@ -454,6 +456,7 @@ enum conf_type { #ifdef FLB_HAVE_CHUNK_TRACE #define FLB_CONF_STR_ENABLE_CHUNK_TRACE "Enable_Chunk_Trace" +#define FLB_CONF_STR_CHUNK_TRACE_HTTP_ALLOWED_ENDPOINTS "Chunk_Trace_HTTP_Allowed_Endpoints" #endif /* FLB_HAVE_CHUNK_TRACE */ #define FLB_CONF_STR_HOT_RELOAD "Hot_Reload" diff --git a/src/flb_config.c b/src/flb_config.c index 16fdf11a737..d20af383569 100644 --- a/src/flb_config.c +++ b/src/flb_config.c @@ -255,6 +255,9 @@ struct flb_service_config service_configs[] = { {FLB_CONF_STR_ENABLE_CHUNK_TRACE, FLB_CONF_TYPE_BOOL, offsetof(struct flb_config, enable_chunk_trace)}, + {FLB_CONF_STR_CHUNK_TRACE_HTTP_ALLOWED_ENDPOINTS, + FLB_CONF_TYPE_STR, + offsetof(struct flb_config, chunk_trace_http_allowed_endpoints)}, #endif #ifdef FLB_SYSTEM_WINDOWS @@ -599,6 +602,12 @@ void flb_config_exit(struct flb_config *config) flb_free(config->plugins_file); } +#ifdef FLB_HAVE_CHUNK_TRACE + if (config->chunk_trace_http_allowed_endpoints) { + flb_free(config->chunk_trace_http_allowed_endpoints); + } +#endif + if (config->kernel) { flb_kernel_destroy(config->kernel); } diff --git a/src/http_server/api/v1/trace.c b/src/http_server/api/v1/trace.c index e6d33e26032..af6466f7af7 100644 --- a/src/http_server/api/v1/trace.c +++ b/src/http_server/api/v1/trace.c @@ -28,9 +28,11 @@ #include #include #include +#include #include #include #include +#include #include #define STR_INPUTS "inputs" @@ -81,8 +83,10 @@ static struct flb_input_instance *find_input(struct flb_hs *hs, const char *name * not let a remote client instantiate arbitrary output plugins: outputs such * as file, http or s3 can write files, open connections or run the AWS * credential_process. Only stdout, and the calyptia output which reuses the - * properties of the configured instance, can be requested. Any other output - * can still be used locally through the --trace command line option. + * properties of the configured instance, can be requested. The http output is + * only accepted when the operator allowlists its destinations through the + * chunk_trace_http_allowed_endpoints service option. Any other output can + * still be used locally through the --trace command line option. */ static const char *trace_allowed_outputs[] = { "stdout", @@ -97,6 +101,24 @@ static const char *trace_allowed_params[] = { NULL }; +/* + * Only parameters that select the endpoint and encode the payload: proxy, + * file based TLS settings, AWS and OAuth2 options could redirect the records + * or read local files and credentials, so they are rejected. + */ +static const char *trace_http_allowed_params[] = { + "host", + "port", + "uri", + "format", + "json_date_key", + "json_date_format", + "header", + "compress", + "tls", + NULL +}; + /* * Output names are matched exactly: the chunk trace pipeline only reuses the * configured calyptia instance properties for the exact "calyptia" name, while @@ -123,10 +145,197 @@ static int trace_name_allowed(const char **list, const char *name, int ignore_ca return FLB_FALSE; } -static int trace_output_allowed(const char *output_name, struct mk_list *props) +static const char *trace_prop_get(struct mk_list *props, const char *key) +{ + struct mk_list *head; + struct flb_kv *kv; + + if (props == NULL) { + return NULL; + } + + mk_list_foreach(head, props) { + kv = mk_list_entry(head, struct flb_kv, _head); + if (strcasecmp(kv->key, key) == 0) { + return kv->val; + } + } + + return NULL; +} + +static int trace_http_port_parse(const char *str, size_t len, int *port) +{ + size_t i; + int val = 0; + + if (len == 0 || len > 5) { + return -1; + } + + for (i = 0; i < len; i++) { + if (!isdigit((unsigned char) str[i])) { + return -1; + } + val = (val * 10) + (str[i] - '0'); + } + + if (val < 1 || val > 65535) { + return -1; + } + + *port = val; + return 0; +} + +/* + * Match an allowlist entry against the requested destination. Entries use + * the form host:port or [ipv6]:port, optionally followed by a path, in which + * case the requested uri must be exactly that path. Host names are compared + * case-insensitively and are never resolved here. + */ +static int trace_http_endpoint_match(const char *entry, const char *host, + int port, const char *uri) +{ + int entry_port; + const char *p; + const char *host_start; + const char *host_end; + const char *path; + const char *port_end; + + if (entry[0] == '[') { + host_start = entry + 1; + host_end = strchr(host_start, ']'); + if (host_end == NULL || host_end[1] != ':') { + return -1; + } + p = host_end + 2; + } + else { + host_start = entry; + host_end = strchr(entry, ':'); + if (host_end == NULL) { + return -1; + } + p = host_end + 1; + } + + if (host_end == host_start) { + return -1; + } + + path = strchr(p, '/'); + port_end = path != NULL ? path : p + strlen(p); + if (trace_http_port_parse(p, port_end - p, &entry_port) != 0) { + return -1; + } + + if (strlen(host) != (size_t) (host_end - host_start) || + strncasecmp(host, host_start, host_end - host_start) != 0 || + port != entry_port) { + return FLB_FALSE; + } + + if (path != NULL && strcmp(path, uri) != 0) { + return FLB_FALSE; + } + + return FLB_TRUE; +} + +static int trace_http_endpoint_allowed(struct flb_config *config, struct mk_list *props) { + int ret; + int port = 80; + int allowed = FLB_FALSE; + const char *host; + const char *port_str; + const char *uri; + struct mk_list entries; struct mk_list *head; struct flb_kv *kv; + struct flb_slist_entry *entry; + + if (config->chunk_trace_http_allowed_endpoints == NULL) { + flb_error("trace output 'http' is disabled, allow destinations with " + "the '%s' service option", + FLB_CONF_STR_CHUNK_TRACE_HTTP_ALLOWED_ENDPOINTS); + return FLB_FALSE; + } + + if (props != NULL) { + mk_list_foreach(head, props) { + kv = mk_list_entry(head, struct flb_kv, _head); + if (trace_name_allowed(trace_http_allowed_params, kv->key, FLB_TRUE) == FLB_FALSE) { + flb_error("trace output parameter '%s' is not allowed", kv->key); + return FLB_FALSE; + } + /* Output properties expand ${VAR}, which could leak the environment. */ + if (kv->val != NULL && strstr(kv->val, "${") != NULL) { + flb_error("trace output parameter '%s' must not reference " + "environment variables", kv->key); + return FLB_FALSE; + } + } + } + + host = trace_prop_get(props, "host"); + if (host == NULL || host[0] == '\0') { + flb_error("trace output 'http' requires the 'host' parameter"); + return FLB_FALSE; + } + + port_str = trace_prop_get(props, "port"); + if (port_str != NULL && + trace_http_port_parse(port_str, strlen(port_str), &port) != 0) { + flb_error("trace output 'http' has an invalid port '%s'", port_str); + return FLB_FALSE; + } + + uri = trace_prop_get(props, "uri"); + if (uri == NULL) { + uri = "/"; + } + + flb_slist_create(&entries); + if (flb_slist_split_string(&entries, config->chunk_trace_http_allowed_endpoints, + ',', -1) == -1) { + flb_slist_destroy(&entries); + return FLB_FALSE; + } + + mk_list_foreach(head, &entries) { + entry = mk_list_entry(head, struct flb_slist_entry, _head); + ret = trace_http_endpoint_match(entry->str, host, port, uri); + if (ret == -1) { + flb_warn("invalid entry '%s' in '%s', expected host:port[/uri]", + entry->str, FLB_CONF_STR_CHUNK_TRACE_HTTP_ALLOWED_ENDPOINTS); + } + else if (ret == FLB_TRUE) { + allowed = FLB_TRUE; + break; + } + } + flb_slist_destroy(&entries); + + if (allowed == FLB_FALSE) { + flb_error("trace output 'http' endpoint '%s:%i%s' is not allowed", + host, port, uri); + } + + return allowed; +} + +static int trace_output_allowed(struct flb_config *config, const char *output_name, + struct mk_list *props) +{ + struct mk_list *head; + struct flb_kv *kv; + + if (strcmp(output_name, "http") == 0) { + return trace_http_endpoint_allowed(config, props); + } if (trace_name_allowed(trace_allowed_outputs, output_name, FLB_FALSE) == FLB_FALSE) { flb_error("trace output '%s' is not allowed", output_name); @@ -153,7 +362,7 @@ static int enable_trace_input(struct flb_hs *hs, const char *name, ssize_t nlen, { struct flb_input_instance *in; - if (trace_output_allowed(output_name, props) == FLB_FALSE) { + if (trace_output_allowed(hs->config, output_name, props) == FLB_FALSE) { return 403; } diff --git a/tests/integration/scenarios/internal_http_server/config/internal_http_server_trace_http.yaml b/tests/integration/scenarios/internal_http_server/config/internal_http_server_trace_http.yaml new file mode 100644 index 00000000000..7c96d6f5217 --- /dev/null +++ b/tests/integration/scenarios/internal_http_server/config/internal_http_server_trace_http.yaml @@ -0,0 +1,20 @@ +service: + flush: 1 + grace: 1 + log_level: info + http_server: on + http_listen: 127.0.0.1 + http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT} + enable_chunk_trace: on + chunk_trace_http_allowed_endpoints: invalid-entry, 127.0.0.1:${TEST_SUITE_HTTP_PORT}/data + +pipeline: + inputs: + - name: dummy + alias: traced + tag: test + dummy: '{"message":"hello"}' + + outputs: + - name: null + match: "*" diff --git a/tests/integration/scenarios/internal_http_server/tests/test_internal_http_server_001.py b/tests/integration/scenarios/internal_http_server/tests/test_internal_http_server_001.py index aa25fb6c9de..5679d2d028a 100644 --- a/tests/integration/scenarios/internal_http_server/tests/test_internal_http_server_001.py +++ b/tests/integration/scenarios/internal_http_server/tests/test_internal_http_server_001.py @@ -5,7 +5,9 @@ import subprocess import pytest +import requests import yaml +from server.http_server import data_storage, http_server_run from utils.fluent_bit_manager import FluentBitManager, FluentBitStartupError from utils.http_matrix import curl_supports_http2, run_curl_request from utils.test_service import FluentBitTestService @@ -157,6 +159,97 @@ def test_internal_http_server_trace_rejects_unsafe_outputs(tmp_path): service.stop() +class TraceHttpService(Service): + def __init__(self): + super().__init__(os.path.abspath( + os.path.join(os.path.dirname(__file__), "../config/internal_http_server_trace_http.yaml") + )) + self.service = FluentBitTestService( + self.config_file, + data_storage=data_storage, + data_keys=["payloads", "requests"], + pre_start=self._start_receiver, + post_stop=self._stop_receiver, + ) + + def _start_receiver(self, service): + http_server_run(service.test_suite_http_port) + service.wait_for_http_endpoint( + f"http://127.0.0.1:{service.test_suite_http_port}/ping", timeout=10, interval=0.5, + ) + + def _stop_receiver(self, service): + try: + requests.post(f"http://127.0.0.1:{service.test_suite_http_port}/shutdown", timeout=2) + except requests.RequestException: + pass + + +def test_internal_http_server_trace_http_output_allowlist(): + service = TraceHttpService() + service.start() + port = str(service.service.test_suite_http_port) + allowed = {"host": "127.0.0.1", "port": port, "uri": "/data", "format": "json"} + rejected = [ + {**allowed, "uri": "/other"}, + {**allowed, "port": "1"}, + {**allowed, "host": "localhost"}, + {**allowed, "port": port + "x"}, + {key: value for key, value in allowed.items() if key != "host"}, + {**allowed, "proxy": "http://127.0.0.1:9"}, + {**allowed, "tls.ca_file": "/etc/passwd"}, + {**allowed, "http_user": "user"}, + {**allowed, "header": "X-Leak ${HOME}"}, + {**allowed, "uri": "/data${HOME}"}, + ] + + try: + for params in rejected: + for path, inputs in [("/api/v1/trace/traced", {}), + ("/api/v1/traces/", {"inputs": ["traced"]})]: + result = service.request(path, method="POST", payload=json.dumps({ + **inputs, "output": "http", "params": params, + })) + if inputs: + assert result["status_code"] == 200 + assert '"returncode":403' in result["body"], params + else: + assert result["status_code"] == 403, params + + assert data_storage["requests"] == [] + + result = service.request("/api/v1/trace/traced", method="POST", payload=json.dumps({ + "output": "http", "prefix": "http-trace.", "params": allowed, + })) + assert result["status_code"] == 200, result + + requests_seen = service.service.wait_for_condition( + lambda: data_storage["requests"] or None, + timeout=10, interval=0.5, description="trace records sent to the allowed endpoint", + ) + assert all(request["path"] == "/data" for request in requests_seen) + assert any("http-trace." in request["decoded_data"] for request in requests_seen) + + result = service.request("/api/v1/trace/traced", method="DELETE") + assert result["status_code"] == 201 + finally: + service.stop() + + +def test_internal_http_server_trace_http_output_disabled_by_default(): + service = Service() + service.start() + try: + result = service.request("/api/v1/trace/dummy.0", method="POST", payload=json.dumps({ + "output": "http", "params": {"host": "127.0.0.1", "port": "9", "uri": "/"}, + })) + assert result["status_code"] == 403 + with open(service.flb.log_file, encoding="utf-8", errors="replace") as stream: + assert "chunk_trace_http_allowed_endpoints" in stream.read().lower() + finally: + service.stop() + + def test_internal_http_server_trace_rejects_duplicate_and_invalid_options(): service = Service() service.start()