http_server: accept empty HTTP header values

src/http_server/flb_http_server.c

@@ -1399,7 +1399,11 @@ int flb_http_server_session_ingest(struct flb_http_server_session *session,
             }
         }
 
-        if (session->version <= HTTP_PROTOCOL_VERSION_11) {
+        if (session->version == HTTP_PROTOCOL_VERSION_AUTODETECT) {
+            /* Wait for the remainder of a split HTTP/2 connection preface. */
+            return HTTP_SERVER_SUCCESS;
+        }
+        else if (session->version <= HTTP_PROTOCOL_VERSION_11) {
             result = flb_http1_server_session_init(&session->http1, session);
 
             if (result != 0) {
@@ -1412,6 +1416,10 @@ int flb_http_server_session_ingest(struct flb_http_server_session *session,
             if (result != 0) {
                 return -1;
             }
+
+            /* Protocol detection may have accumulated the preface over multiple reads. */
+            buffer = (unsigned char *) session->incoming_data;
+            length = cfl_sds_len(session->incoming_data);
         }
     }
 

src/http_server/flb_http_server_http1.c

@@ -566,6 +566,12 @@ int flb_http1_server_session_ingest(struct flb_http1_server_session *session,
          * performance overhead in exchange for ensuring safety.
          */
     }
+    else if (result == MK_HTTP_PARSER_ERROR) {
+        /* The caller closes the session when a parser error is returned. */
+        session->stream.status = HTTP_STREAM_STATUS_ERROR;
+
+        return HTTP_SERVER_PROVIDER_ERROR;
+    }
 
     dummy_mk_http_request_init(&session->inner_session, &session->inner_request);
     mk_http_parser_init(&session->inner_parser);

tests/integration/scenarios/in_http/tests/test_in_http_001.py

@@ -2,6 +2,7 @@
 import json
 import os
 import logging
+import socket
 import time
 
 import pytest
@@ -66,6 +67,74 @@ def send_requests(conn, num_requests, headers, json_payload):
     return responses
 
 
+def assert_connection_open_without_response(connection):
+    connection.settimeout(1)
+
+    try:
+        data = connection.recv(1)
+    except socket.timeout:
+        return
+
+    if not data:
+        pytest.fail("HTTP server closed the connection while the request was incomplete")
+
+    pytest.fail("HTTP server responded before the request was complete")
+
+
+def send_raw_http1_request(port, request, split_at=None):
+    response = bytearray()
+
+    with socket.create_connection(("127.0.0.1", port), timeout=2) as connection:
+        connection.settimeout(2)
+
+        if split_at is None:
+            connection.sendall(request)
+        else:
+            connection.sendall(request[:split_at])
+            assert_connection_open_without_response(connection)
+            connection.settimeout(2)
+            connection.sendall(request[split_at:])
+
+        while len(response) < 4096 and b"\r\n" not in response:
+            try:
+                data = connection.recv(4096 - len(response))
+            except ConnectionResetError:
+                break
+            except socket.timeout:
+                pytest.fail("HTTP/1 server did not respond or close the connection")
+
+            if not data:
+                break
+            response.extend(data)
+
+    return bytes(response)
+
+
+def send_split_http2_preface(port):
+    preface = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"
+    settings_frame = b"\x00\x00\x00\x04\x00\x00\x00\x00\x00"
+    response = bytearray()
+
+    with socket.create_connection(("127.0.0.1", port), timeout=2) as connection:
+        connection.settimeout(2)
+        connection.sendall(preface[:-2])
+        assert_connection_open_without_response(connection)
+        connection.settimeout(2)
+        connection.sendall(preface[-2:] + settings_frame)
+
+        while len(response) < 9:
+            try:
+                data = connection.recv(4096)
+            except socket.timeout:
+                pytest.fail("HTTP/2 server did not respond to a split connection preface")
+
+            if not data:
+                break
+            response.extend(data)
+
+    return bytes(response)
+
+
 def test_send_data():
     try:
         service = Service("in_http_config")
@@ -118,6 +187,45 @@ def test_in_http_protocol_matrix(case):
     assert forwarded_payloads[0][0]["message"] == "Este es un mensaje de prueba"
 
 
+def test_in_http_accepts_split_http2_preface():
+    service = Service("in_http_http2_cleartext.yaml")
+
+    try:
+        service.start()
+        response = send_split_http2_preface(service.flb_listener_port)
+    finally:
+        service.stop()
+
+    assert len(response) >= 9
+    assert response[3] == 0x04
+    assert data_storage["payloads"] == []
+
+
+def test_in_http_accepts_http1_request_split_before_autodetect_boundary():
+    service = Service("in_http_http2_cleartext.yaml")
+    body = b'{"message":"split-http1"}'
+    request = (
+        b"POST / HTTP/1.1\r\n"
+        b"Host: localhost\r\n"
+        b"Content-Type: application/json\r\n"
+        + f"Content-Length: {len(body)}\r\n".encode()
+        + b"Connection: close\r\n"
+        b"\r\n"
+        + body
+    )
+
+    try:
+        service.start()
+        response = send_raw_http1_request(service.flb_listener_port, request, split_at=1)
+        forwarded_payloads = service.read_forwarded_payloads()
+    finally:
+        service.stop()
+
+    assert b"HTTP/1.1 201" in response
+    assert len(forwarded_payloads) == 1
+    assert forwarded_payloads[0][0]["message"] == "split-http1"
+
+
 def test_in_http_rejects_bad_json():
     service = Service("in_http_config")
     service.start()
@@ -150,6 +258,88 @@ def test_in_http_rejects_get_requests():
     assert result["status_code"] >= 400
 
 
+def test_in_http_accepts_post_with_empty_generic_headers():
+    service = Service("in_http_config")
+    body = b'{"message":"empty-header"}'
+    request = (
+        b"POST / HTTP/1.1\r\n"
+        b"Host: localhost\r\n"
+        b"Content-Type: application/json\r\n"
+        + f"Content-Length: {len(body)}\r\n".encode()
+        + b"X-Empty:\r\n"
+        b"X-Empty-Whitespace: \t\r\n"
+        b"Connection: close\r\n"
+        b"\r\n"
+        + body
+    )
+
+    try:
+        service.start()
+        response = send_raw_http1_request(service.flb_listener_port, request)
+        forwarded_payloads = service.read_forwarded_payloads()
+    finally:
+        service.stop()
+
+    assert b"HTTP/1.1 201" in response
+    assert len(forwarded_payloads) == 1
+    assert forwarded_payloads[0][0]["message"] == "empty-header"
+
+
+def test_in_http_accepts_empty_connection_and_transfer_encoding():
+    service = Service("in_http_config")
+    body = b'{"message":"empty-semantic-headers"}'
+    request = (
+        b"POST / HTTP/1.1\r\n"
+        b"Host: localhost\r\n"
+        b"Content-Type: application/json\r\n"
+        + f"Content-Length: {len(body)}\r\n".encode()
+        + b"Connection:\r\n"
+        b"Transfer-Encoding: \t\r\n"
+        b"Connection: close\r\n"
+        b"\r\n"
+        + body
+    )
+
+    try:
+        service.start()
+        response = send_raw_http1_request(service.flb_listener_port, request)
+        forwarded_payloads = service.read_forwarded_payloads()
+    finally:
+        service.stop()
+
+    assert b"HTTP/1.1 201" in response
+    assert len(forwarded_payloads) == 1
+    assert forwarded_payloads[0][0]["message"] == "empty-semantic-headers"
+
+
+@pytest.mark.parametrize("header_value", [b"", b" \t"], ids=["empty", "whitespace"])
+@pytest.mark.parametrize(
+    "following_data",
+    [b"1-X: value\r\nConnection: close\r\n\r\n1", b"\r\n1"],
+    ids=["numeric-header", "numeric-body"],
+)
+def test_in_http_rejects_empty_content_length(header_value, following_data):
+    service = Service("in_http_config")
+    request = (
+        b"POST / HTTP/1.1\r\n"
+        b"Host: localhost\r\n"
+        b"Content-Type: application/json\r\n"
+        b"Content-Length:" + header_value + b"\r\n"
+        + following_data
+    )
+
+    try:
+        service.start()
+        response = send_raw_http1_request(service.flb_listener_port, request)
+        service.assert_no_forwarded_payloads_for()
+    finally:
+        service.stop()
+
+    forwarded_payloads = list(data_storage["payloads"])
+    assert response == b"" or b"HTTP/1.1 400" in response
+    assert forwarded_payloads == []
+
+
 @pytest.mark.parametrize(
     "case",
     [
@@ -405,5 +595,12 @@ def read_forwarded_payloads(self, timeout=10):
             time.sleep(0.5)
         raise TimeoutError("Timed out waiting for forwarded HTTP payloads")
 
+    def assert_no_forwarded_payloads_for(self, quiet_period=1.5):
+        deadline = time.time() + quiet_period
+
+        while time.time() < deadline:
+            assert data_storage["payloads"] == []
+            time.sleep(0.1)
+
     def stop(self):
         self.service.stop()