Remove client key/cert support.

Not actively supported anyway and current handling for ws did not work after
removing default certs

kuksa-client/kuksa_client/__main__.py

@@ -47,8 +47,6 @@
 
 DEFAULT_KUKSA_ADDRESS = os.environ.get("KUKSA_ADDRESS", "grpc://127.0.0.1:55555")
 DEFAULT_TOKEN_OR_TOKENFILE = os.environ.get("TOKEN_OR_TOKENFILE", None)
-DEFAULT_CERTIFICATE = os.environ.get("CERTIFICATE", None)
-DEFAULT_KEYFILE = os.environ.get("KEYFILE", None)
 DEFAULT_CACERTIFICATE = os.environ.get("CACERTIFICATE", None)
 DEFAULT_TLS_SERVER_NAME = os.environ.get("TLS_SERVER_NAME", None)
 
@@ -317,8 +315,6 @@ def __init__(
         self,
         server=None,
         token_or_tokenfile=None,
-        certificate=None,
-        keyfile=None,
         cacertificate=None,
         tls_server_name=None,
     ):
@@ -340,8 +336,6 @@ def __init__(
         self.subscribeIds = set()
         self.commThread = None
         self.token_or_tokenfile = token_or_tokenfile
-        self.certificate = certificate
-        self.keyfile = keyfile
         self.cacertificate = cacertificate
         self.tls_server_name = tls_server_name
 
@@ -612,10 +606,6 @@ def connect(self):
         # Configs should only be added if they actually have a value
         if self.token_or_tokenfile is not None:
             config["token_or_tokenfile"] = self.token_or_tokenfile
-        if self.certificate is not None:
-            config["certificate"] = self.certificate
-        if self.keyfile is not None:
-            config["keyfile"] = self.keyfile
         if self.cacertificate is not None:
             config["cacertificate"] = self.cacertificate
         if self.tls_server_name is not None:
@@ -684,17 +674,6 @@ def main():
     )
 
     # Add TLS arguments
-    # Note: Databroker does not yet support mutual authentication, so no need to use two first arguments
-    parser.add_argument(
-        "--certificate",
-        default=DEFAULT_CERTIFICATE,
-        help="Client cert file(.pem), only needed for mutual authentication",
-    )
-    parser.add_argument(
-        "--keyfile",
-        default=DEFAULT_KEYFILE,
-        help="Client private key file (.key), only needed for mutual authentication",
-    )
     parser.add_argument(
         "--cacertificate",
         default=DEFAULT_CACERTIFICATE,
@@ -716,8 +695,6 @@ def main():
     clientApp = TestClient(
         args.server,
         token_or_tokenfile=args.token_or_tokenfile,
-        certificate=args.certificate,
-        keyfile=args.keyfile,
         cacertificate=args.cacertificate,
         tls_server_name=args.tls_server_name,
     )

kuksa-client/kuksa_client/cli_backend/__init__.py

@@ -29,8 +29,6 @@ def __init__(self, config):
         # If no CA Certificate is given we will use an insecure connection, requested or not
         if self.cacertificate is None:
             self.insecure = True
-        self.certificate = config.get('certificate', None)
-        self.keyfile = config.get('keyfile', None)
         self.tls_server_name = config.get('tls_server_name', "")
         self.token_or_tokenfile = config.get('token_or_tokenfile', None)
 

kuksa-client/kuksa_client/cli_backend/grpc.py

@@ -66,10 +66,6 @@ def __init__(self, config):
         super().__init__(config)
         if self.cacertificate is not None:
             self.cacertificate = pathlib.Path(self.cacertificate)
-        if self.keyfile is not None:
-            self.keyfile = pathlib.Path(self.keyfile)
-        if self.certificate is not None:
-            self.certificate = pathlib.Path(self.certificate)
         if self.token_or_tokenfile is not None:
             if os.path.isfile(self.token_or_tokenfile):
                 self.token_or_tokenfile = pathlib.Path(self.token_or_tokenfile)
@@ -285,8 +281,6 @@ async def mainLoop(self):
                 self.serverIP,
                 self.serverPort,
                 root_certificates=self.cacertificate,
-                private_key=self.keyfile,
-                certificate_chain=self.certificate,
                 tls_server_name=self.tls_server_name,
                 token=self.token
             ) as vss_client:

kuksa-client/kuksa_client/cli_backend/ws.py

@@ -297,8 +297,6 @@ async def connect(self, _=None):
         subprotocols = ["VISSv2"]
         if not self.insecure:
             context = ssl.create_default_context()
-            context.load_cert_chain(
-                certfile=self.certificate, keyfile=self.keyfile)
             context.load_verify_locations(cafile=self.cacertificate)
             # We want host name to match
             # For example certificates we use subjectAltName to make it match for Server, localahost and 127.0.0.1

kuksa-client/kuksa_client/grpc/__init__.py

@@ -559,8 +559,6 @@ def __init__(
         port: int,
         token: Optional[str] = None,
         root_certificates: Optional[Path] = None,
-        private_key: Optional[Path] = None,
-        certificate_chain: Optional[Path] = None,
         ensure_startup_connection: bool = True,
         connected: bool = False,
         tls_server_name: Optional[str] = None
@@ -569,8 +567,6 @@ def __init__(
         self.authorization_header = self.get_authorization_header(token)
         self.target_host = f'{host}:{port}'
         self.root_certificates = root_certificates
-        self.private_key = private_key
-        self.certificate_chain = certificate_chain
         self.tls_server_name = tls_server_name
         self.ensure_startup_connection = ensure_startup_connection
         self.connected = connected
@@ -580,15 +576,7 @@ def _load_creds(self) -> Optional[grpc.ChannelCredentials]:
         if self.root_certificates:
             logger.info(f"Using TLS with Root CA from {self.root_certificates}")
             root_certificates = self.root_certificates.read_bytes()
-            if self.private_key and self.certificate_chain:
-                private_key = self.private_key.read_bytes()
-                certificate_chain = self.certificate_chain.read_bytes()
-                # As of today there is no option in KUKSA.val Databroker to require client authentication
-                logger.info("Using client private key and certificates, mutual TLS supported if supported by server")
-                return grpc.ssl_channel_credentials(root_certificates, private_key, certificate_chain)
-            else:
-                logger.info("No client certificates provided, mutual TLS not supported!")
-                return grpc.ssl_channel_credentials(root_certificates)
+            return grpc.ssl_channel_credentials(root_certificates)
         logger.info("No Root CA present, it will not be possible to use a secure connection!")
         return None
 

kuksa-client/tests/conftest.py

@@ -65,7 +65,7 @@ async def secure_val_server_fixture(unused_tcp_port, resources_path, val_service
             (resources_path / 'test-server.pem').read_bytes(),
         )],
         root_certificates=(resources_path / 'test-ca.pem').read_bytes(),
-        require_client_auth=True,
+        require_client_auth=False,
     ))
     await server.start()
     try:

kuksa-client/tests/test_grpc.py

@@ -406,8 +406,6 @@ async def test_secure_connection(self, unused_tcp_port, resources_path, val_serv
             name='test_server', version='1.2.3')
         async with VSSClient('localhost', unused_tcp_port,
                              root_certificates=resources_path / 'test-ca.pem',
-                             private_key=resources_path / 'test-client.key',
-                             certificate_chain=resources_path / 'test-client.pem',
                              ensure_startup_connection=True
                              ):
             assert val_servicer.GetServerInfo.call_count == 1