Tests for #1061 and #1535 issue

README.md

@@ -282,24 +282,40 @@ other options, depending on item type.
 
 Now such backends are supported:
 1) type == nginx
-    status_uri: uri where nginx status is located
-    config: nginx config
+    - status_uri: uri where nginx status is located
+    - config: nginx config
 
 2) type == deproxy
-    port: listen this port
-    response: type of response. Now only 'static' is supported
-        response == static:
-            response_content: always response this content
+    - port: listen this port
+    - response: type of response. Now only 'static' is supported
+      - response == static:
+         - response_content: always response this content,
+         - keep_original_data: optional: if set to True,
+                the original request will be kept in Request.original_data field
+                as it has arrived by the wire,
+                otherwise (say, the parameter is not defined at all)
+                Request.original_data will be left blank
+         - segment_size: optional: TCP segment size for heavy chunked testing, bytes, 0 for disable
+         - segment_gap: optional: inter-segment gap for heavy chunked testing, ms, 0 for disable
+              - you usualy do not need it; update timeouts if you use it
 
 and such clients:
 1) type == wrk
-    addr: 'ip:port'
+    - addr: 'ip:port'
 
 2) type == deproxy
-    addr: ip addr of server to connect
-    port: port
-
-All options are mandatory
+    - addr: ip addr of server to connect
+    - port: port
+    - keep_original_data: optional: if set to True,
+           the original response will be kept in Response.original_data field
+           as it has arrived by the wire,
+           otherwise (say, the parameter is not defined at all)
+           Response.original_data will be left blank
+    - segment_size: optional: TCP segment size for heavy chunked testing, bytes, 0 for disable
+    - segment_gap: optional: inter-segment gap for heavy chunked testing, ms, 0 for disable
+       - you usualy do not need it; update timeouts if you use it
+
+All options are mandatory, unless explicitly stated otherwise.
 
 nginx config, deproxy response, addr and port can use templates
 in format `${part_variable}` where `part` is one of 'server',
@@ -313,6 +329,16 @@ Example tests can be found in `selftests/test_framework.py`
 Tests can be skipped or marked as expected to fail.
 More info at [Python documentation](https://docs.python.org/3/library/unittest.html).
 
+### Testing with chunked messages
+
+Some tests require division of request or response into small TCP segments ("chunks").
+This division is controlled by segment_size parameter of the client or the backend
+(see above). Usualy better to set this parameter programmaticaly rather than in client
+or backend configuration.
+
+An example to create tests which divide requests or responsies into chunks and
+iterate over various chunk sizes is here: `malformed/test_chunking_example.py`.
+
 ## Internal structure and motivation of user configured tests
 
 User configured tests have very flexible structure. They allow arbitrary

framework/deproxy_client.py

@@ -1,10 +1,11 @@
 import abc
 import time
+import socket
 
-from helpers import deproxy, tf_cfg, stateful
+from helpers import deproxy, tf_cfg, stateful, selfproxy
 
 __author__ = 'Tempesta Technologies, Inc.'
-__copyright__ = 'Copyright (C) 2018-2019 Tempesta Technologies, Inc.'
+__copyright__ = 'Copyright (C) 2018-2022 Tempesta Technologies, Inc.'
 __license__ = 'GPL2'
 
 class BaseDeproxyClient(deproxy.Client):
@@ -21,9 +22,29 @@ def __init__(self, *args, **kwargs):
         self.rps = 0
         self.valid_req_num = 0
         self.cur_req_num = 0
+        # This parameter controls whether to keep original data with the response
+        # (See deproxy.HttpMessage.original_data)
+        self.keep_original_data = None
+        # Following 2 parameters control heavy chunked testing
+        # You can set it programmaticaly or via client config
+        # TCP segment size, bytes, 0 for disable, usualy value of 1 is sufficient
+        self.segment_size = 0
+        # Inter-segment gap, ms, 0 for disable.
+        # You usualy do not need it; update timeouts if you use it.
+        self.segment_gap = 0
+        # This state variable contains a timestamp of the last segment sent
+        self.last_segment_time = 0
+        # The following 2 variables are used to save destination address and port
+        # when overriding to connect via ssl chunking proxy
+        self.overriden_addr = None
+        self.overriden_port = None
+        # a presense of selfproxy
+        self.selfproxy_present = False
 
     def handle_connect(self):
         deproxy.Client.handle_connect(self)
+        if self.segment_size and not self.selfproxy_present:
+            self.socket.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1)
         self.start_time = time.time()
 
     def set_events(self, polling_lock):
@@ -34,12 +55,13 @@ def set_rps(self, rps):
 
     def __stop_client(self):
         tf_cfg.dbg(4, '\tStop deproxy client')
+        self.release_selfproxy()
         if self.polling_lock != None:
             self.polling_lock.acquire()
         try:
             self.close()
         except Exception as e:
-            tf_cfg.dbg(2, "Exception while start: %s" % str(e))
+            tf_cfg.dbg(2, "Exception while stop: %s" % str(e))
             if self.polling_lock != None:
                 self.polling_lock.release()
             raise e
@@ -54,6 +76,8 @@ def run_start(self):
         self.start_time = 0
         self.valid_req_num = 0
         self.cur_req_num = 0
+        if self.ssl and self.segment_size != 0:
+            self.insert_selfproxy()
         if self.polling_lock != None:
             self.polling_lock.acquire()
         try:
@@ -76,7 +100,8 @@ def handle_read(self):
             try:
                 method = self.methods[self.nrresp]
                 response = deproxy.Response(self.response_buffer,
-                                    method=method)
+                                    method=method,
+                                keep_original_data=self.keep_original_data)
                 self.response_buffer = \
                             self.response_buffer[response.original_length:]
             except deproxy.IncompleteMessage:
@@ -99,6 +124,9 @@ def writable(self):
             return False
         if time.time() < self.next_request_time():
             return False
+        if (self.segment_gap != 0 and not self.selfproxy_present and
+            time.time() - self.last_segment_time < self.segment_gap / 1000.0):
+            return False;
         return True
 
     def next_request_time(self):
@@ -110,9 +138,13 @@ def handle_write(self):
         reqs = self.request_buffers
         tf_cfg.dbg(4, '\tDeproxy: Client: Send request to Tempesta.')
         tf_cfg.dbg(5, reqs[self.cur_req_num])
-        sent = self.send(reqs[self.cur_req_num])
+        if self.segment_size != 0 and not self.selfproxy_present:
+            sent = self.send(reqs[self.cur_req_num][:self.segment_size])
+        else:
+            sent = self.send(reqs[self.cur_req_num])
         if sent < 0:
             return
+        self.last_segment_time = time.time()
         reqs[self.cur_req_num] = reqs[self.cur_req_num][sent:]
         if len(reqs[self.cur_req_num]) == 0:
             self.cur_req_num += 1
@@ -121,6 +153,8 @@ def make_requests(self, requests):
         request_buffers = []
         methods = []
         valid_req_num = 0
+        if self.selfproxy_present:
+            self.update_selfproxy()
         while len(requests) > 0:
             try:
                 req = deproxy.Request(requests)
@@ -160,6 +194,41 @@ def make_request(self, request):
     def receive_response(self, response):
         raise NotImplementedError("Not implemented 'receive_response()'")
 
+    def insert_selfproxy(self):
+        # inserting the chunking proxy between ssl client and server
+        if not self.ssl or self.segment_size == 0:
+            return
+        selfproxy.request_client_selfproxy(
+            listen_host = "127.0.0.1",
+            listen_port = selfproxy.CLIENT_MODE_PORT_REPLACE,
+            forward_host = self.conn_addr,
+            forward_port = self.port,
+            segment_size = self.segment_size,
+            segment_gap = self.segment_gap)
+        self.overriden_addr = self.conn_addr
+        self.overriden_port = self.port
+        self.conn_addr = "127.0.0.1"
+        self.port = selfproxy.CLIENT_MODE_PORT_REPLACE
+        self.selfproxy_present = True
+
+    def release_selfproxy(self):
+        # action reverse to insert_selfproxy
+        if self.selfproxy_present:
+            selfproxy.release_client_selfproxy()
+            self.selfproxy_present = False
+        if self.overriden_addr is not None:
+            self.conn_addr = self.overriden_addr
+            self.overriden_addr = None
+        if self.overriden_port is not None:
+            self.port = self.overriden_port
+            self.overriden_port = None
+
+    def update_selfproxy(self):
+        # update chunking parameters
+        if self.selfproxy_present:
+            selfproxy.update_client_selfproxy_chunking(
+                self.segment_size, self.segment_gap)
+
 
 class DeproxyClient(BaseDeproxyClient):
     last_response = None

framework/deproxy_server.py

@@ -12,7 +12,7 @@
 import port_checks
 
 __author__ = 'Tempesta Technologies, Inc.'
-__copyright__ = 'Copyright (C) 2018 Tempesta Technologies, Inc.'
+__copyright__ = 'Copyright (C) 2018-2021 Tempesta Technologies, Inc.'
 __license__ = 'GPL2'
 
 class ServerConnection(asyncore.dispatcher_with_send):
@@ -21,23 +21,36 @@ def __init__(self, server, sock=None, keep_alive=None):
         asyncore.dispatcher_with_send.__init__(self, sock)
         self.server = server
         self.keep_alive = keep_alive
+        self.last_segment_time = 0
         self.responses_done = 0
         self.request_buffer = ''
         tf_cfg.dbg(6, '\tDeproxy: SrvConnection: New server connection.')
 
     def initiate_send(self):
         """ Override dispatcher_with_send.initiate_send() which transfers
         data with too small chunks of 512 bytes.
+        However if server.segment_size is set (!=0), use this value.
         """
         num_sent = 0
-        num_sent = asyncore.dispatcher.send(self, self.out_buffer[:4096])
+        num_sent = asyncore.dispatcher.send(self, self.out_buffer[:
+                          self.server.segment_size
+                          if self.server.segment_size > 0
+                          else 4096 ])
         self.out_buffer = self.out_buffer[num_sent:]
+        self.last_segment_time = time.time()
 
     def send_pending_and_close(self):
         while len(self.out_buffer):
             self.initiate_send()
         self.handle_close()
 
+    def writable(self):
+        if ( self.server.segment_gap != 0 and
+             time.time() - self.last_segment_time
+                  < self.server.segment_gap / 1000.0 ):
+            return False;
+        return asyncore.dispatcher_with_send.writable(self)
+
     def send_response(self, response):
         if response:
             tf_cfg.dbg(4, '\tDeproxy: SrvConnection: Send response.')
@@ -66,7 +79,9 @@ def handle_close(self):
     def handle_read(self):
         self.request_buffer += self.recv(deproxy.MAX_MESSAGE_SIZE)
         try:
-            request = deproxy.Request(self.request_buffer)
+            request = deproxy.Request(self.request_buffer,
+                          keep_original_data =
+                            self.server.keep_original_data)
         except deproxy.IncompleteMessage:
             return
         except deproxy.ParseError:
@@ -89,6 +104,18 @@ def handle_read(self):
 class BaseDeproxyServer(deproxy.Server, port_checks.FreePortsChecker):
 
     def __init__(self, *args, **kwargs):
+        # This parameter controls whether to keep original data with the request
+        # (See deproxy.HttpMessage.original_data)
+        self.keep_original_data = kwargs.pop("keep_original_data", None)
+
+        # Following 2 parameters control heavy chunked testing
+        # You can set it programmaticaly or via client config
+        # TCP segment size, bytes, 0 for disable, usualy value of 1 is sufficient
+        self.segment_size = kwargs.pop("segment_size", 0)
+        # Inter-segment gap, ms, 0 for disable.
+        # You usualy do not need it; update timeouts if you use it.
+        self.segment_gap = kwargs.pop("segment_gap", 0)
+
         deproxy.Server.__init__(self, *args, **kwargs)
         self.stop_procedures = [self.__stop_server]
         self.is_polling = threading.Event()
@@ -99,6 +126,8 @@ def handle_accept(self):
         pair = self.accept()
         if pair is not None:
             sock, _ = pair
+            if self.segment_size:
+                sock.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1)
             handler = ServerConnection(server=self, sock=sock,
                                        keep_alive=self.keep_alive)
             self.connections.append(handler)
@@ -187,10 +216,16 @@ def deproxy_srv_factory(server, name, tester):
     else:
         port = int(port)
     srv = None
+    ko = server.get("keep_original_data", None)
+    ss = server.get("segment_size", 0)
+    sg = server.get("segment_gap", 0)
     rtype = server['response']
     if rtype == 'static':
         content = fill_template(server['response_content'], server)
-        srv = StaticDeproxyServer(port=port, response=content)
+        srv = StaticDeproxyServer(port=port, response=content,
+                                  keep_original_data = ko,
+                                  segment_size = ss,
+                                  segment_gap = sg)
     else:
         raise Exception("Invalid response type: %s" % str(rtype))
 

framework/tester.py

@@ -15,7 +15,7 @@
 import struct
 
 __author__ = 'Tempesta Technologies, Inc.'
-__copyright__ = 'Copyright (C) 2018-2019 Tempesta Technologies, Inc.'
+__copyright__ = 'Copyright (C) 2018-2021 Tempesta Technologies, Inc.'
 __license__ = 'GPL2'
 
 backend_defs = {}
@@ -144,6 +144,9 @@ def __create_client_deproxy(self, client, ssl, bind_addr):
             # the client configuration.
             server_hostname = fill_template(client['ssl_hostname'], client)
             clt.set_server_hostname(server_hostname)
+        clt.segment_size = int(client.get('segment_size', 0))
+        clt.segment_gap = int(client.get('segment_gap', 0))
+        clt.keep_original_data = bool(client.get('keep_original_data', None))
         return clt
 
     def __create_client_wrk(self, client, ssl):