h2 tests and deproxy h2 update

framework/deproxy_client.py

@@ -4,13 +4,21 @@
 from io import StringIO
 
 import h2.connection
-from h2.events import DataReceived, ResponseReceived, StreamEnded, TrailersReceived
+from h2.events import (
+    ConnectionTerminated,
+    DataReceived,
+    ResponseReceived,
+    SettingsAcknowledged,
+    StreamEnded,
+    TrailersReceived,
+)
+from h2.settings import SettingCodes, Settings
 from hpack import Encoder
 
 from helpers import deproxy, selfproxy, stateful, tf_cfg
 
 __author__ = "Tempesta Technologies, Inc."
-__copyright__ = "Copyright (C) 2018-2022 Tempesta Technologies, Inc."
+__copyright__ = "Copyright (C) 2018-2023 Tempesta Technologies, Inc."
 __license__ = "GPL2"
 
 
@@ -357,13 +365,15 @@ def encode(self, headers, huffman=True):
 
 class DeproxyClientH2(DeproxyClient):
     last_response: deproxy.H2Response
+    h2_connection: h2.connection.H2Connection
 
     def __init__(self, *args, **kwargs):
         DeproxyClient.__init__(self, *args, **kwargs)
         self.encoder = HuffmanEncoder()
         self.h2_connection = None
         self.stream_id = 1
         self.active_responses = {}
+        self.ack_settings = False
 
     def make_requests(self, requests):
         for request in requests:
@@ -383,8 +393,6 @@ def make_request(self, request: tuple or list or str, end_stream=True, huffman=T
             self.h2_connection = h2.connection.H2Connection()
             self.h2_connection.encoder = self.encoder
             self.h2_connection.initiate_connection()
-            if self.selfproxy_present:
-                self.update_selfproxy()
 
         self.h2_connection.encoder.huffman = huffman
 
@@ -417,6 +425,80 @@ def make_request(self, request: tuple or list or str, end_stream=True, huffman=T
             self.stream_id += 2
             self.valid_req_num += 1
 
+    def update_initial_settings(
+        self,
+        header_table_size: int = None,
+        enable_push: int = None,
+        max_concurrent_stream: int = None,
+        initial_window_size: int = None,
+        max_frame_size: int = None,
+        max_header_list_size: int = None,
+    ) -> None:
+        """Update initial SETTINGS frame and add preamble + SETTINGS frame in `data_to_send`."""
+        if not self.h2_connection:
+            self.h2_connection = h2.connection.H2Connection()
+            self.h2_connection.encoder = self.encoder
+
+            new_settings = self.__generate_new_settings(
+                header_table_size,
+                enable_push,
+                max_concurrent_stream,
+                initial_window_size,
+                max_frame_size,
+                max_header_list_size,
+            )
+
+            # if settings is empty, we should not change them
+            if new_settings:
+                self.h2_connection.local_settings = Settings(initial_values=new_settings)
+                self.h2_connection.local_settings.update(new_settings)
+
+            self.h2_connection.initiate_connection()
+
+    def send_settings_frame(
+        self,
+        header_table_size: int = None,
+        enable_push: int = None,
+        max_concurrent_stream: int = None,
+        initial_window_size: int = None,
+        max_frame_size: int = None,
+        max_header_list_size: int = None,
+    ) -> None:
+        self.ack_settings = False
+
+        new_settings = self.__generate_new_settings(
+            header_table_size,
+            enable_push,
+            max_concurrent_stream,
+            initial_window_size,
+            max_frame_size,
+            max_header_list_size,
+        )
+
+        self.h2_connection.update_settings(new_settings)
+
+        self.send_bytes(data=self.h2_connection.data_to_send())
+        self.h2_connection.clear_outbound_data_buffer()
+
+    def wait_for_ack_settings(self, timeout=5):
+        """Wait SETTINGS frame with ack flag."""
+        if self.state != stateful.STATE_STARTED:
+            return False
+
+        t0 = time.time()
+        while not self.ack_settings:
+            t = time.time()
+            if t - t0 > timeout:
+                return False
+            time.sleep(0.01)
+        return True
+
+    def send_bytes(self, data: bytes, expect_response=False):
+        self.request_buffers.append(data)
+        self.nrreq += 1
+        if expect_response:
+            self.valid_req_num += 1
+
     def handle_read(self):
         self.response_buffer = self.recv(deproxy.MAX_MESSAGE_SIZE)
         if not self.response_buffer:
@@ -426,7 +508,6 @@ def handle_read(self):
         tf_cfg.dbg(5, self.response_buffer)
 
         try:
-            method = self.methods[self.nrresp]
             events = self.h2_connection.receive_data(self.response_buffer)
             for event in events:
                 if isinstance(event, ResponseReceived):
@@ -440,7 +521,7 @@ def handle_read(self):
                     else:
                         response = deproxy.H2Response(
                             headers + "\r\n",
-                            method=method,
+                            method=self.methods[self.nrresp],
                             body_parsing=False,
                             keep_original_data=self.keep_original_data,
                         )
@@ -464,6 +545,12 @@ def handle_read(self):
                         return
                     self.receive_response(response)
                     self.nrresp += 1
+                elif isinstance(event, ConnectionTerminated):
+                    self.error_codes.append(event.error_code)
+                elif isinstance(event, SettingsAcknowledged):
+                    self.ack_settings = True
+                    # TODO should be changed by issue #358
+                    self.handle_read()
                 # TODO should be changed by issue #358
                 else:
                     self.handle_read()
@@ -513,3 +600,27 @@ def __headers_to_string(self, headers):
 
     def __binary_headers_to_string(self, headers):
         return "".join(["%s: %s\r\n" % (h.decode(), v.decode()) for h, v in headers])
+
+    @staticmethod
+    def __generate_new_settings(
+        header_table_size: int = None,
+        enable_push: int = None,
+        max_concurrent_stream: int = None,
+        initial_window_size: int = None,
+        max_frame_size: int = None,
+        max_header_list_size: int = None,
+    ) -> dict:
+        new_settings = dict()
+        if header_table_size is not None:
+            new_settings[SettingCodes.HEADER_TABLE_SIZE] = header_table_size
+        if enable_push is not None:
+            new_settings[SettingCodes.ENABLE_PUSH] = header_table_size
+        if max_concurrent_stream is not None:
+            new_settings[SettingCodes.MAX_CONCURRENT_STREAMS] = max_concurrent_stream
+        if initial_window_size is not None:
+            new_settings[SettingCodes.INITIAL_WINDOW_SIZE] = initial_window_size
+        if max_frame_size is not None:
+            new_settings[SettingCodes.MAX_FRAME_SIZE] = max_frame_size
+        if max_header_list_size is not None:
+            new_settings[SettingCodes.MAX_HEADER_LIST_SIZE] = max_header_list_size
+        return new_settings

helpers/deproxy.py

@@ -32,7 +32,7 @@
 from . import error, stateful, tempesta, tf_cfg
 
 __author__ = "Tempesta Technologies, Inc."
-__copyright__ = "Copyright (C) 2017-2022 Tempesta Technologies, Inc."
+__copyright__ = "Copyright (C) 2017-2023 Tempesta Technologies, Inc."
 __license__ = "GPL2"
 
 # -------------------------------------------------------------------------------
@@ -741,6 +741,7 @@ def __init__(
         self.stop_procedures = [self.__stop_client]
         self.conn_is_closed = True
         self.bind_addr = bind_addr
+        self.error_codes = []
 
     def __stop_client(self):
         tf_cfg.dbg(4, "\tStop deproxy client")
@@ -821,7 +822,8 @@ def handle_write(self):
         self.request_buffer = self.request_buffer[sent:]
 
     def handle_error(self):
-        _, v, _ = sys.exc_info()
+        type_error, v, _ = sys.exc_info()
+        self.error_codes.append(type_error)
         if type(v) == ParseError or type(v) == AssertionError:
             raise v
         elif type(v) == ssl.SSLWantReadError:

http2_general/__init__.py

@@ -1,3 +1,14 @@
-__all__ = ["test_h2_specs", "test_h2_perf", "test_h2_ping", "test_h2_frame", "test_h2_hpack"]
+__all__ = [
+    "test_h2_specs",
+    "test_h2_perf",
+    "test_h2_ping",
+    "test_h2_frame",
+    "test_h2_hpack",
+    "test_flow_control_window",
+    "test_h2_streams",
+    "test_h2_headers",
+    "test_h2_sticky_cookie",
+    "test_h2_max_frame_size",
+]
 
 # vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4

http2_general/helpers.py

@@ -0,0 +1,71 @@
+__author__ = "Tempesta Technologies, Inc."
+__copyright__ = "Copyright (C) 2023 Tempesta Technologies, Inc."
+__license__ = "GPL2"
+
+from framework import deproxy_client, tester
+
+
+class H2Base(tester.TempestaTest):
+    backends = [
+        {
+            "id": "deproxy",
+            "type": "deproxy",
+            "port": "8000",
+            "response": "static",
+            "response_content": (
+                "HTTP/1.1 200 OK\r\n"
+                + "Date: test\r\n"
+                + "Server: debian\r\n"
+                + "Content-Length: 0\r\n\r\n"
+            ),
+        }
+    ]
+
+    tempesta = {
+        "config": """
+            listen 443 proto=h2;
+            server ${server_ip}:8000;
+            tls_certificate ${tempesta_workdir}/tempesta.crt;
+            tls_certificate_key ${tempesta_workdir}/tempesta.key;
+            tls_match_any_server_name;
+
+            block_action attack reply;
+            block_action error reply;
+        """
+    }
+
+    clients = [
+        {
+            "id": "deproxy",
+            "type": "deproxy_h2",
+            "addr": "${tempesta_ip}",
+            "port": "443",
+            "ssl": True,
+        },
+    ]
+
+    post_request = [
+        (":authority", "example.com"),
+        (":path", "/"),
+        (":scheme", "https"),
+        (":method", "POST"),
+    ]
+
+    get_request = [
+        (":authority", "example.com"),
+        (":path", "/"),
+        (":scheme", "https"),
+        (":method", "GET"),
+    ]
+
+    def initiate_h2_connection(self, client: deproxy_client.DeproxyClientH2):
+        # add preamble + settings frame with default variable into data_to_send
+        client.update_initial_settings()
+        # send preamble + settings frame to Tempesta
+        client.send_bytes(client.h2_connection.data_to_send())
+        client.h2_connection.clear_outbound_data_buffer()
+
+        self.assertTrue(
+            client.wait_for_ack_settings(),
+            "Tempesta foes not returns SETTINGS frame with ACK flag.",
+        )

http2_general/test_flow_control_window.py

@@ -0,0 +1,41 @@
+"""Functional tests for flow control window."""
+
+__author__ = "Tempesta Technologies, Inc."
+__copyright__ = "Copyright (C) 2023 Tempesta Technologies, Inc."
+__license__ = "GPL2"
+
+from h2.exceptions import FlowControlError
+
+from http2_general.helpers import H2Base
+
+
+class TestFlowControl(H2Base):
+    def test_flow_control_window_for_stream(self):
+        """
+        Client sets SETTINGS_INITIAL_WINDOW_SIZE = 1k bytes and backend returns response
+        with 2k bytes body.
+        Tempesta must forward DATA frame with 1k bytes and wait WindowUpdate from client.
+        """
+        self.start_all_services()
+        client = self.get_client("deproxy")
+        server = self.get_server("deproxy")
+        server.set_response(
+            "HTTP/1.1 200 OK\r\n"
+            + "Date: test\r\n"
+            + "Server: debian\r\n"
+            + "Content-Length: 2000\r\n\r\n"
+            + ("x" * 2000)
+        )
+
+        client.update_initial_settings(initial_window_size=1000)
+        client.make_request(self.post_request)
+        client.wait_for_response(3)
+
+        self.assertNotIn(
+            FlowControlError, client.error_codes, "Tempesta ignored flow control window for stream."
+        )
+        self.assertFalse(client.connection_is_closed())
+        self.assertEqual(client.last_response.status, "200", "Status code mismatch.")
+        self.assertEqual(
+            len(client.last_response.body), 2000, "Tempesta did not return full response body."
+        )