Keep TCP/TLS connection alive and close connection properly

* The client tools try to keep TCP/TLS connection now
* The HTTP header "Connection: keep-alive" is sent now and
  when server does not reply with "Connection: close", then
  RHSM tools try to keep existing connection. It helps to
  reduce traffic between client tools and candlepin server.
  It also reduce significantly time of executing all commands.
  E.g. The time of registration is reduced from 20 seconds
  to 13 seconds
* Connection using consumer authentication has to be recreated
  during registration, because new consumer certificate is
  installed
* Tested with rhsm.service and it also works as expected
* Improved little bit debugging of traffic and it is visible
  what type of authentication is used for communication with
  candlepin server. You can see the difference, when
  "subscription-manager version" is used vs e.g.
  "subscription-manager identity"
* Do TLS shutdown handshake (tear down) at the end of
  communication with candlepin server

src/rhsm/connection.py

@@ -509,6 +509,8 @@ class BaseRestLib(object):
     responses
     """
 
+    __conn = None
+
     ALPHA = 0.9
 
     def __init__(
@@ -532,6 +534,7 @@ def __init__(
         token=None,
         user_agent=None,
     ):
+        log.debug("Creating new BaseRestLib instance")
         self.host = host
         self.ssl_port = ssl_port
         self.apihandler = apihandler
@@ -572,6 +575,20 @@ def __init__(
         elif token:
             self.headers["Authorization"] = "Bearer " + token
 
+    def close_connection(self):
+        """
+        Try to close connection to server
+        :return: None
+        """
+        if self.__conn is not None:
+            log.debug(f"Closing HTTPs connection {self.__conn}")
+            # Do proper TLS shutdown handshake (TLS tear down) first
+            if self.__conn.sock is not None:
+                self.__conn.sock.unwrap()
+            # Then it is possible to close TCP connection
+            self.__conn.close()
+        self.__conn = None
+
     def _get_cert_key_list(self):
         """
         Create list of cert-key pairs to be used with the connection
@@ -643,23 +660,32 @@ def _create_connection(self, cert_file=None, key_file=None):
         if cert_file and os.path.exists(cert_file):
             context.load_cert_chain(cert_file, keyfile=key_file)
 
-        if self.proxy_hostname and self.proxy_port:
-            log.debug(
-                "Using proxy: %s:%s" % (normalized_host(self.proxy_hostname), safe_int(self.proxy_port))
-            )
-            proxy_headers = {
-                "User-Agent": self.user_agent,
-                "Host": "%s:%s" % (normalized_host(self.host), safe_int(self.ssl_port)),
-            }
-            if self.proxy_user and self.proxy_password:
-                proxy_headers["Proxy-Authorization"] = _encode_auth(self.proxy_user, self.proxy_password)
-            conn = httplib.HTTPSConnection(
-                self.proxy_hostname, self.proxy_port, context=context, timeout=self.timeout
-            )
-            conn.set_tunnel(self.host, safe_int(self.ssl_port), proxy_headers)
-            self.headers["Host"] = "%s:%s" % (normalized_host(self.host), safe_int(self.ssl_port))
+        if self.__conn is None:
+            log.debug("Creating new connection")
+            if self.proxy_hostname and self.proxy_port:
+                log.debug(
+                    "Using proxy: %s:%s" % (normalized_host(self.proxy_hostname), safe_int(self.proxy_port))
+                )
+                proxy_headers = {
+                    "User-Agent": self.user_agent,
+                    "Host": "%s:%s" % (normalized_host(self.host), safe_int(self.ssl_port)),
+                }
+                if self.proxy_user and self.proxy_password:
+                    proxy_headers["Proxy-Authorization"] = _encode_auth(self.proxy_user, self.proxy_password)
+                conn = httplib.HTTPSConnection(
+                    self.proxy_hostname, self.proxy_port, context=context, timeout=self.timeout
+                )
+                conn.set_tunnel(self.host, safe_int(self.ssl_port), proxy_headers)
+                self.headers["Host"] = "%s:%s" % (normalized_host(self.host), safe_int(self.ssl_port))
+            else:
+                conn = httplib.HTTPSConnection(
+                    self.host, self.ssl_port, context=context, timeout=self.timeout
+                )
+            log.debug(f"Created connection: {conn}")
+            self.__conn = conn
         else:
-            conn = httplib.HTTPSConnection(self.host, self.ssl_port, context=context, timeout=self.timeout)
+            log.debug("Reusing connection: %s", self.__conn)
+            conn = self.__conn
 
         return conn
 
@@ -684,10 +710,18 @@ def _print_debug_info_about_request(self, request_type, handler, final_headers,
             green_col = "\033[92m"
             red_col = "\033[91m"
             end_col = "\033[0m"
+            if self.token:
+                auth = "keycloak auth"
+            elif self.username and self.password:
+                auth = "basic auth"
+            elif self.cert_file and self.key_file:
+                auth = "consumer auth"
+            else:
+                auth = "no auth"
             if self.insecure is True:
-                msg = blue_col + "Making insecure request:" + end_col
+                msg = blue_col + f"Making insecure ({auth}) request:" + end_col
             else:
-                msg = blue_col + "Making request:" + end_col
+                msg = blue_col + f"Making ({auth}) request:" + end_col
             msg += red_col + " %s:%s %s %s" % (self.host, self.ssl_port, request_type, handler) + end_col
             if self.proxy_hostname and self.proxy_port:
                 msg += (
@@ -786,6 +820,9 @@ def _request(self, request_type, method, info=None, headers=None, cert_key_pairs
         else:
             body = None
 
+        if self.__conn is not None:
+            self.headers["Connection"] = "keep-alive"
+
         log.debug("Making request: %s %s" % (request_type, handler))
 
         if self.user_agent:
@@ -868,6 +905,17 @@ def _request(self, request_type, method, info=None, headers=None, cert_key_pairs
         response_log = '%s, request="%s %s"' % (response_log, request_type, handler)
         log.debug(response_log)
 
+        connection_http_header = response.getheader("Connection")
+        if connection_http_header == "keep-alive":
+            log.debug("Server wants to keep connection")
+        elif connection_http_header == "close":
+            log.debug("Server wants to close connection. Closing HTTP connection")
+            self.close_connection()
+        elif connection_http_header is None:
+            log.debug("HTTP header 'Connection' not included in response")
+        else:
+            log.debug(f"Unsupported value of HTTP header 'Connection': {connection_http_header}")
+
         # Look for server drift, and log a warning
         if drift_check(response.getheader("date")):
             log.warning("Clock skew detected, please check your system time")
@@ -1129,10 +1177,6 @@ def has_capability(self, capability):
             self.capabilities = self._load_manager_capabilities()
         return capability in self.capabilities
 
-    def shutDown(self):
-        self.conn.close()
-        log.debug("remote connection closed")
-
     def ping(self, username=None, password=None):
         return self.conn.request_get("/status/")
 

src/rhsmlib/services/register.py

@@ -123,6 +123,11 @@ def register(
                 usage=usage,
                 jwt_token=jwt_token,
             )
+            # When new consumer is created, then close all existing connections
+            # to be able to recreate new one
+            cp_provider = inj.require(inj.CP_PROVIDER)
+            cp_provider.close_all_connections()
+
         self.installed_mgr.write_cache()
         self.plugin_manager.run("post_register_consumer", consumer=consumer, facts=facts_dict)
         managerlib.persist_consumer_cert(consumer)

src/subscription_manager/cp_provider.py

@@ -13,6 +13,7 @@
 #
 import base64
 import json
+import logging
 
 from subscription_manager.identity import ConsumerIdentity
 from subscription_manager import utils
@@ -21,6 +22,8 @@
 
 import rhsm.connection as connection
 
+log = logging.getLogger(__name__)
+
 
 class TokenAuthUnsupportedException(Exception):
     pass
@@ -155,6 +158,24 @@ def get_dbus_sender():
         else:
             return ""
 
+    def close_all_connections(self):
+        """
+        Try to close all connections to candlepin server, CDN, etc.
+        :return: None
+        """
+        if self.consumer_auth_cp is not None:
+            log.debug("Closing consumer authenticated connection...")
+            self.consumer_auth_cp.conn.close_connection()
+        if self.no_auth_cp is not None:
+            log.debug("Closing no authenticated connection...")
+            self.no_auth_cp.conn.close_connection()
+        if self.basic_auth_cp is not None:
+            log.debug("Closing basically authenticated connection...")
+            self.basic_auth_cp.conn.close_connection()
+        if self.keycloak_auth_cp is not None:
+            log.debug("Closing keycloak authenticated connection...")
+            self.keycloak_auth_cp.conn.close_connestion()
+
     def get_consumer_auth_cp(self):
         if not self.consumer_auth_cp:
             self.consumer_auth_cp = connection.UEPConnection(

src/subscription_manager/managercli.py

@@ -89,6 +89,8 @@ def main(self):
         enabled_yum_plugins = YumPluginManager.enable_pkg_plugins()
         if len(enabled_yum_plugins) > 0:
             print("\n" + _("WARNING") + "\n\n" + YumPluginManager.warning_message(enabled_yum_plugins) + "\n")
+        # Try to close all connections
+        managerlib.close_all_connections()
         # Try to flush all outputs, see BZ: 1350402
         try:
             sys.stdout.flush()

src/subscription_manager/managerlib.py

@@ -68,6 +68,15 @@ def system_log(message, priority=syslog.LOG_NOTICE):
     utils.system_log(message, priority)
 
 
+def close_all_connections():
+    """
+    Close all connections
+    :return: None
+    """
+    cpp_provider = require(CP_PROVIDER)
+    cpp_provider.close_all_connections()
+
+
 # FIXME: move me to identity.py
 def persist_consumer_cert(consumerinfo):
     """

src/subscription_manager/utils.py

@@ -281,7 +281,7 @@ def get_server_versions(cp, exception_on_timeout=False):
 
     if cp:
         try:
-            supported_resources = get_supported_resources()
+            supported_resources = get_supported_resources(uep=cp)
             if "status" in supported_resources:
                 status = cp.getStatus()
                 cp_version = "-".join(