python tests: compatibility and stability improvements

This commit improves the compatibility and stability of Python tests
for Filebeat. It addresses two main probmes:
1. Some deprecated TLS functions are now raising deprecation
exceptions (since Python 3.10.0)
2. Some tests, on failure, leave Filebeat running and bound to port
8080, this creates a cascading failure of the subsequent tests that
require the same port to be available.

The main changes are:
 - Ensuring Filebeat process is always terminated, even if the test
 fails
 - Using the replacement methods/objects for the deprecated TLS
 functions.

CHANGELOG-developer.next.asciidoc

@@ -138,6 +138,7 @@ The list below covers the major changes between 7.0.0-rc2 and main only.
 - Add gcp dataproc metricset. {pull}30008[30008]
 - Add Github action for linting
 - Add regex support for drop_fields processor.
+- Improve compatibility and reduce flakyness of Python tests {pull}31588[31588]
 
 ==== Deprecated
 

filebeat/tests/system/test_syslog.py

@@ -30,6 +30,7 @@ def test_syslog_with_tcp(self):
         )
 
         filebeat = self.start_beat()
+        self.addCleanup(filebeat.kill_and_wait)
 
         self.wait_until(lambda: self.log_contains("Start accepting connections"))
 
@@ -72,6 +73,7 @@ def test_syslog_with_tcp_invalid_message(self):
         )
 
         filebeat = self.start_beat()
+        self.addCleanup(filebeat.kill_and_wait)
 
         self.wait_until(lambda: self.log_contains("Start accepting connections"))
 
@@ -112,6 +114,7 @@ def test_syslog_with_udp(self):
         )
 
         filebeat = self.start_beat()
+        self.addCleanup(filebeat.kill_and_wait)
 
         self.wait_until(lambda: self.log_contains("Started listening for UDP connection"))
 
@@ -172,6 +175,7 @@ def run_filebeat_and_send_using_socket(self, socket_type, send_over_socket):
             )
 
             filebeat = self.start_beat()
+            self.addCleanup(filebeat.kill_and_wait)
 
             if socket_type == "stream":
                 self.wait_until(lambda: self.log_contains("Start accepting connections"))
@@ -235,6 +239,7 @@ def run_filebeat_and_send_invalid_message_using_socket(self, socket_type, send_o
             )
 
             filebeat = self.start_beat()
+            self.addCleanup(filebeat.kill_and_wait)
 
             if socket_type == "stream":
                 self.wait_until(lambda: self.log_contains("Start accepting connections"))

filebeat/tests/system/test_tcp_tls.py

@@ -64,13 +64,18 @@ def test_tcp_over_tls_and_verify_valid_server_without_mutual_auth(self):
         )
 
         filebeat = self.start_beat()
+        self.addCleanup(filebeat.kill_and_wait)
 
         self.wait_until(lambda: self.log_contains(
             "Start accepting connections"))
 
         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  # TCP
-        tls = ssl.wrap_socket(sock, cert_reqs=ssl.CERT_REQUIRED,
-                              ca_certs=CERTIFICATE1, do_handshake_on_connect=True)
+        context = ssl.create_default_context()
+        context.load_verify_locations(cafile=CERTIFICATE1)
+        context.options = ssl.CERT_REQUIRED
+        context.check_hostname = False
+
+        tls = context.wrap_socket(sock, do_handshake_on_connect=True)
         tls.connect((config.get('host'), config.get('port')))
 
         for n in range(0, NUMBER_OF_EVENTS):
@@ -116,14 +121,18 @@ def test_tcp_over_tls_and_verify_invalid_server_without_mutual_auth(self):
         )
 
         filebeat = self.start_beat()
+        self.addCleanup(filebeat.kill_and_wait)
 
         self.wait_until(lambda: self.log_contains(
             "Start accepting connections"))
 
         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)  # TCP
-        tls = ssl.wrap_socket(sock, cert_reqs=ssl.CERT_REQUIRED,
-                              ca_certs=CERTIFICATE2, do_handshake_on_connect=True)
+        context = ssl.create_default_context()
+        context.load_verify_locations(capath=CERTIFICATE2)
+        context.options = ssl.CERT_REQUIRED
+        context.check_hostname = False
 
+        tls = context.wrap_socket(sock, do_handshake_on_connect=True, server_hostname=None)
         with pytest.raises(ssl.SSLError):
             tls.connect((config.get('host'), config.get('port')))
 
@@ -157,13 +166,19 @@ def test_tcp_over_tls_mutual_auth_fails(self):
         )
 
         filebeat = self.start_beat()
+        self.addCleanup(filebeat.kill_and_wait)
 
         self.wait_until(lambda: self.log_contains(
             "Start accepting connections"))
 
         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        tls = ssl.wrap_socket(sock, cert_reqs=ssl.CERT_REQUIRED,
-                              ca_certs=CERTIFICATE1, do_handshake_on_connect=True)
+
+        context = ssl.create_default_context()
+        context.load_verify_locations(capath=CERTIFICATE1)
+        context.options = ssl.CERT_REQUIRED
+        context.check_hostname = False
+
+        tls = context.wrap_socket(sock, do_handshake_on_connect=True, server_hostname=None)
 
         with pytest.raises(ssl.SSLError):
             tls.connect((config.get('host'), config.get('port')))
@@ -204,16 +219,18 @@ def test_tcp_over_tls_mutual_auth_succeed(self):
         )
 
         filebeat = self.start_beat()
+        self.addCleanup(filebeat.kill_and_wait)
 
         self.wait_until(lambda: self.log_contains(
             "Start accepting connections"))
 
         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
-        context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+        context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
         context.verify_mode = ssl.CERT_REQUIRED
         context.load_verify_locations(CACERT)
         context.load_cert_chain(certfile=CLIENT2, keyfile=CLIENTKEY2)
+        context.check_hostname = False
 
         tls = context.wrap_socket(sock, server_side=False)
 
@@ -262,6 +279,7 @@ def test_tcp_tls_with_a_plain_text_socket(self):
         )
 
         filebeat = self.start_beat()
+        self.addCleanup(filebeat.kill_and_wait)
 
         self.wait_until(lambda: self.log_contains(
             "Start accepting connections"))
@@ -315,18 +333,20 @@ def test_tcp_over_tls_mutual_auth_rfc6587_framing(self):
         )
 
         filebeat = self.start_beat()
+        self.addCleanup(filebeat.kill_and_wait)
 
         self.wait_until(lambda: self.log_contains(
             "Start accepting connections"))
 
         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
-        context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+        context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
         context.verify_mode = ssl.CERT_REQUIRED
         context.load_verify_locations(CACERT)
         context.load_cert_chain(certfile=CLIENT2, keyfile=CLIENTKEY2)
+        context.check_hostname = False
 
-        tls = context.wrap_socket(sock, server_side=False)
+        tls = context.wrap_socket(sock)
 
         tls.connect((config.get('host'), config.get('port')))
 

libbeat/tests/system/beat/beat.py

@@ -137,8 +137,11 @@ def kill_and_wait(self):
         """
         kill_and_wait will kill the process and wait for it to return
         """
-        self.kill()
-        os.close(self.stdin_write)
+        # If the process is running, kill it and close self.stdin_write
+        # this is done so this method can safely be called multiple times
+        if self.proc.poll() is None:
+            self.kill()
+            os.close(self.stdin_write)
         return self.wait()
 
     def check_kill_and_wait(self, exit_code=0):
@@ -151,6 +154,8 @@ def check_kill_and_wait(self, exit_code=0):
 
     def __del__(self):
         # Ensure the process is stopped.
+        # For some reason when some tests error/timeout this
+        # method is not called in a timely manner.
         try:
             self.proc.terminate()
             self.proc.kill()