Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 67 additions & 0 deletions plugins/out_http/http.c
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,54 @@ static void append_headers(struct flb_http_client *c,
}
}

/*
* Reads ctx->http_bearer_token_file fresh (no caching), trims trailing whitespace/newlines,
* and sets the Authorization: Bearer header on the given HTTP client. Returns 0 on
* success, -1 on read failure, empty-after-trim content, or header-construction failure
* (error already logged, path and reason only, never token content).
*/
static int http_bearer_token_file_auth(struct flb_out_http *ctx,
struct flb_http_client *c)
{
int ret;
char *bearer_buf = NULL;
size_t bearer_size = 0;

ret = flb_utils_read_file(ctx->http_bearer_token_file, &bearer_buf, &bearer_size);
if (ret == -1) {
flb_plg_error(ctx->ins, "could not read http_bearer_token_file '%s'",
ctx->http_bearer_token_file);
return -1;
}

while (bearer_size > 0 &&
(bearer_buf[bearer_size - 1] == '\r' ||
bearer_buf[bearer_size - 1] == '\n' ||
bearer_buf[bearer_size - 1] == ' ' ||
bearer_buf[bearer_size - 1] == '\t')) {
bearer_size--;
}
bearer_buf[bearer_size] = '\0';

if (bearer_size == 0) {
flb_plg_error(ctx->ins, "http_bearer_token_file '%s' is empty after trimming",
ctx->http_bearer_token_file);
flb_free(bearer_buf);
return -1;
}

ret = flb_http_bearer_auth(c, bearer_buf);
flb_free(bearer_buf);

if (ret == -1) {
flb_plg_error(ctx->ins, "could not set Authorization header from "
"http_bearer_token_file '%s'", ctx->http_bearer_token_file);
return -1;
}

return 0;
}

static int http_request(struct flb_out_http *ctx,
const void *body, size_t body_len,
const char *tag, int tag_len,
Expand Down Expand Up @@ -266,6 +314,17 @@ static int http_request(struct flb_out_http *ctx,
if (ctx->http_user && ctx->http_passwd) {
flb_http_basic_auth(c, ctx->http_user, ctx->http_passwd);
}
else if (ctx->http_bearer_token_file) {
ret = http_bearer_token_file_auth(ctx, c);
if (ret == -1) {
if (payload_buf != body) {
flb_free(payload_buf);
}
flb_http_client_destroy(c);
flb_upstream_conn_release(u_conn);
return FLB_RETRY;
}
}

flb_http_add_header(c, "User-Agent", 10, "Fluent-Bit", 10);

Expand Down Expand Up @@ -723,6 +782,14 @@ static struct flb_config_map config_map[] = {
0, FLB_TRUE, offsetof(struct flb_out_http, http_passwd),
"Set HTTP auth password"
},
{
FLB_CONFIG_MAP_STR, "http_bearer_token_file", NULL,
0, FLB_TRUE, offsetof(struct flb_out_http, http_bearer_token_file),
"Path to a file containing a Bearer token. The file is re-read on every "
"outgoing request (no caching), so token rotation on disk is picked up "
"without a restart. Mutually exclusive with http_user/http_passwd and "
"oauth2.enable."
},
{
FLB_CONFIG_MAP_BOOL, "oauth2.enable", "false",
0, FLB_TRUE, offsetof(struct flb_out_http, oauth2_config.enabled),
Expand Down
1 change: 1 addition & 0 deletions plugins/out_http/http.h
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ struct flb_out_http {
/* HTTP Auth */
char *http_user;
char *http_passwd;
char *http_bearer_token_file;

/* AWS Auth */
#ifdef FLB_HAVE_SIGNV4
Expand Down
16 changes: 16 additions & 0 deletions plugins/out_http/http_conf.c
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,22 @@ struct flb_out_http *flb_http_conf_create(struct flb_output_instance *ins,
}
}

if (ctx->http_bearer_token_file) {
if (ctx->http_user) {
flb_plg_error(ctx->ins, "http_bearer_token_file cannot be used together with "
"http_user/http_passwd (Basic Auth); configure exactly one "
"authentication mechanism");
flb_free(ctx);
return NULL;
}
if (ctx->oauth2_config.enabled == FLB_TRUE) {
flb_plg_error(ctx->ins, "http_bearer_token_file cannot be used together with "
"oauth2.enable; configure exactly one authentication mechanism");
flb_free(ctx);
return NULL;
}
}

if (ctx->headers_key && !ctx->body_key) {
flb_plg_error(ctx->ins, "when setting headers_key, body_key is also required");
flb_free(ctx);
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
service:
flush: 1
log_level: info
http_server: on
http_port: ${FLUENT_BIT_HTTP_MONITORING_PORT}

pipeline:
inputs:
- name: dummy
tag: out_http
dummy: '{"message":"hello from bearer_token_file","source":"dummy"}'

outputs:
- name: http
match: out_http
host: 127.0.0.1
port: ${TEST_SUITE_HTTP_PORT}
uri: /data
format: json
json_date_key: false
http_bearer_token_file: ${BEARER_TOKEN_FILE_TEST}
103 changes: 98 additions & 5 deletions tests/integration/scenarios/out_http/tests/test_out_http_001.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,22 +52,25 @@ def _wait_for_http_server_port(timeout=5):


class Service:
def __init__(self, config_file, *, response_setup=None, use_tls=False):
def __init__(self, config_file, *, response_setup=None, use_tls=False, extra_env=None):
self.config_file = os.path.abspath(os.path.join(os.path.dirname(__file__), "../config", config_file))
test_path = os.path.dirname(os.path.abspath(__file__))
cert_dir = os.path.abspath(os.path.join(test_path, "../../in_splunk/certificate"))
self.tls_crt_file = os.path.join(cert_dir, "certificate.pem")
self.tls_key_file = os.path.join(cert_dir, "private_key.pem")
self.response_setup = response_setup
self.use_tls = use_tls
env = {
"CERTIFICATE_TEST": self.tls_crt_file,
"PRIVATE_KEY_TEST": self.tls_key_file,
}
if extra_env:
env.update(extra_env)
self.service = FluentBitTestService(
self.config_file,
data_storage=data_storage,
data_keys=["payloads", "requests"],
extra_env={
"CERTIFICATE_TEST": self.tls_crt_file,
"PRIVATE_KEY_TEST": self.tls_key_file,
},
extra_env=env,
pre_start=self._start_receiver,
post_stop=self._stop_receiver,
)
Expand Down Expand Up @@ -235,6 +238,96 @@ def test_out_http_oauth2_private_key_jwt_adds_bearer_token():
assert data_request["headers"].get("Authorization") == "Bearer oauth-access-token"


def test_out_http_bearer_token_file_adds_bearer_header(tmp_path):
token_path = tmp_path / "bearer_token_file_test.token"
token_path.write_text("test-token-value\n", encoding="utf-8")

service = Service(
"out_http_bearer_token_file.yaml",
extra_env={"BEARER_TOKEN_FILE_TEST": str(token_path)},
)
service.start()
try:
configure_http_response(status_code=200, body={"status": "received"})
requests_seen = service.wait_for_requests(1)
finally:
service.stop()

data_request = requests_seen[0]
assert data_request["headers"].get("Authorization") == "Bearer test-token-value"


def test_out_http_bearer_token_file_picks_up_rotation(tmp_path):
token_path = tmp_path / "bearer_token_file_rotation.token"
token_path.write_text("token-v1", encoding="utf-8")

service = Service(
"out_http_bearer_token_file.yaml",
extra_env={"BEARER_TOKEN_FILE_TEST": str(token_path)},
)
service.start()
configure_http_response(status_code=200, body={"status": "received"})

try:
service.wait_for_requests(1)
assert any(
req["headers"].get("Authorization") == "Bearer token-v1"
for req in data_storage["requests"]
)

token_path.write_text("token-v2", encoding="utf-8")

service.service.wait_for_condition(
lambda: True if any(
req["headers"].get("Authorization") == "Bearer token-v2"
for req in data_storage["requests"]
) else None,
timeout=15,
interval=0.5,
description="a request using the rotated bearer token",
)
finally:
service.stop()


def test_out_http_bearer_token_file_missing_retries(tmp_path):
missing_token_path = tmp_path / "bearer_token_file_missing.token"

service = Service(
"out_http_bearer_token_file.yaml",
extra_env={"BEARER_TOKEN_FILE_TEST": str(missing_token_path)},
)
service.start()

try:
log_text = service.wait_for_log_message("could not read http_bearer_token_file", timeout=15)
finally:
service.stop()

assert str(missing_token_path) in log_text
assert "test-token-value" not in log_text
assert not data_storage["requests"]


def test_out_http_bearer_token_file_empty_retries(tmp_path):
empty_token_path = tmp_path / "bearer_token_file_empty.token"
empty_token_path.write_text("\n", encoding="utf-8")

service = Service(
"out_http_bearer_token_file.yaml",
extra_env={"BEARER_TOKEN_FILE_TEST": str(empty_token_path)},
)
service.start()

try:
log_text = service.wait_for_log_message("is empty after trimming", timeout=15)
finally:
service.stop()

assert str(empty_token_path) in log_text
assert not data_storage["requests"]


def test_out_http_oauth2_timeout_retries_hung_token_endpoint():
service = Service(
"out_http_oauth2_timeout.yaml",
Expand Down
64 changes: 64 additions & 0 deletions tests/runtime/out_http.c
Original file line number Diff line number Diff line change
Expand Up @@ -1144,6 +1144,66 @@ void flb_test_in_http()
test_ctx_destroy(ctx);
}

void flb_test_http_bearer_token_file_conflicts_with_http_user()
{
struct test_ctx *ctx;
int ret;

ctx = test_ctx_create();
if (!TEST_CHECK(ctx != NULL)) {
TEST_MSG("test_ctx_create failed");
exit(EXIT_FAILURE);
}

ret = flb_output_set(ctx->flb, ctx->o_ffd,
"match", "*",
"http_bearer_token_file", "/tmp/does-not-need-to-exist.token",
"http_user", "user1",
"http_passwd", "passwd1",
NULL);
TEST_CHECK(ret == 0);

ret = flb_start(ctx->flb);
if (!TEST_CHECK(ret != 0)) {
TEST_MSG("expected startup failure for http_bearer_token_file + http_user");
}

/* flb_start failed, so there is no running engine to stop. */
flb_destroy(ctx->flb);
flb_free(ctx);
}

void flb_test_http_bearer_token_file_conflicts_with_oauth2()
{
struct test_ctx *ctx;
int ret;

ctx = test_ctx_create();
if (!TEST_CHECK(ctx != NULL)) {
TEST_MSG("test_ctx_create failed");
exit(EXIT_FAILURE);
}

ret = flb_output_set(ctx->flb, ctx->o_ffd,
"match", "*",
"http_bearer_token_file", "/tmp/does-not-need-to-exist.token",
"oauth2.enable", "true",
"oauth2.token_url", "http://127.0.0.1:8888/oauth/token",
"oauth2.client_id", "client1",
"oauth2.client_secret", "secret1",
NULL);
TEST_CHECK(ret == 0);

ret = flb_start(ctx->flb);
if (!TEST_CHECK(ret != 0)) {
TEST_MSG("expected startup failure for http_bearer_token_file + oauth2.enable");
}

/* flb_start failed, so there is no running engine to stop. */
flb_destroy(ctx->flb);
flb_free(ctx);
}

/* Test list */
TEST_LIST = {
{"format_msgpack" , flb_test_format_msgpack},
Expand All @@ -1161,5 +1221,9 @@ TEST_LIST = {
{"json_date_format_iso8601" , flb_test_json_date_format_iso8601},
{"json_date_format_java_sql_timestamp" , flb_test_json_date_format_java_sql_timestamp},
{"in_http", flb_test_in_http},
{"http_bearer_token_file_conflicts_with_http_user",
flb_test_http_bearer_token_file_conflicts_with_http_user},
{"http_bearer_token_file_conflicts_with_oauth2",
flb_test_http_bearer_token_file_conflicts_with_oauth2},
{NULL, NULL}
};