Bug/tls server cert chain

modules/test/tls/bin/check_cert_chain_signature.sh

@@ -0,0 +1,25 @@
+#!/bin/bash
+
+# Copyright 2023 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+INTERMEDIATE=$1
+DEVICE_CERT=$2
+
+echo "ROOT: $ROOT_CERT"
+echo "DEVICE_CERT: $DEVICE_CERT"
+
+response=$(openssl verify -untrusted $INTERMEDIATE $DEVICE_CERT)
+
+echo "$response"

modules/test/tls/python/requirements.txt

@@ -1,4 +1,4 @@
 cryptography
 pyOpenSSL
 pyshark
-cryptography
+requests

modules/test/tls/python/src/tls_module.py

@@ -253,7 +253,7 @@ def _validate_tls_client(self, client_ip, tls_version):
         client_ip=client_ip,
         tls_version=tls_version,
         capture_files=[
-            MONITOR_CAPTURE_FILE, STARTUP_CAPTURE_FILE, GATEWAY_CAPTURE_FILE
+            MONITOR_CAPTURE_FILE, STARTUP_CAPTURE_FILE, TLS_CAPTURE_FILE
         ])
 
     # Generate results based on the state

modules/test/tls/python/src/tls_util.py

@@ -20,6 +20,10 @@
 import os
 from common import util
 import ipaddress
+import requests
+from cryptography import x509
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization
 
 LOG_NAME = 'tls_util'
 LOGGER = None
@@ -39,7 +43,8 @@ def __init__(self,
     global LOGGER
     LOGGER = logger
     self._bin_dir = bin_dir
-    self._dev_cert_file = cert_out_dir + '/device_cert.crt'
+    self._cert_out_dir = cert_out_dir
+    self._dev_cert_file = 'device_cert.crt'
     self._root_certs_dir = root_certs_dir
 
   def get_public_certificate(self,
@@ -150,6 +155,46 @@ def verify_public_key(self, public_key):
       return False, 'Key is not RSA or EC type'
 
   def validate_signature(self, host):
+    # Reconnect to the device but with validate signature option
+    # set to true which will check for proper cert chains
+    # within the valid CA root certs stored on the server
+    if self.validate_trusted_ca_signature(host):
+      LOGGER.info('Authorized Certificate Authority signature confirmed')
+      return True, 'Authorized Certificate Authority signature confirmed'
+    else:
+      LOGGER.info('Authorized Certificate Authority signature not present')
+
+      device_cert_path = os.path.join(self._cert_out_dir, self._dev_cert_file)
+      signed, ca_file = self.validate_local_ca_signature(
+          device_cert_path=device_cert_path)
+      if signed:
+        return True, 'Device signed by cert:' + ca_file
+    return False, 'Device certificate has not been signed'
+
+  def validate_local_ca_signature(self, device_cert_path):
+    bin_file = self._bin_dir + '/check_cert_signature.sh'
+    # Get a list of all root certificates
+    root_certs = os.listdir(self._root_certs_dir)
+    LOGGER.info('Root Certs Found: ' + str(len(root_certs)))
+    for root_cert in root_certs:
+      try:
+        # Create the file path
+        root_cert_path = os.path.join(self._root_certs_dir, root_cert)
+        LOGGER.info('Checking root cert: ' + str(root_cert_path))
+        args = f'{root_cert_path} {device_cert_path}'
+        command = f'{bin_file} {args}'
+        response = util.run_command(command)
+        if f'{device_cert_path}: OK' in str(response):
+          LOGGER.info('Device signed by cert:' + root_cert)
+          return True, root_cert_path
+        else:
+          LOGGER.info('Device not signed by cert: ' + root_cert)
+      except Exception as e:  # pylint: disable=W0718
+        LOGGER.error('Failed to check cert:' + root_cert)
+        LOGGER.error(str(e))
+    return False, None
+
+  def validate_trusted_ca_signature(self, host):
     # Reconnect to the device but with validate signature option
     # set to true which will check for proper cert chains
     # within the valid CA root certs stored on the server
@@ -163,47 +208,127 @@ def validate_signature(self, host):
       return True, 'Authorized Certificate Authority signature confirmed'
     else:
       LOGGER.info('Authorized Certificate Authority signature not present')
-      LOGGER.info('Resolving configured root certificates')
-      bin_file = self._bin_dir + '/check_cert_signature.sh'
-      # Get a list of all root certificates
-      root_certs = os.listdir(self._root_certs_dir)
-      LOGGER.info('Root Certs Found: ' + str(len(root_certs)))
-      for root_cert in root_certs:
-        try:
-          # Create the file path
-          root_cert_path = os.path.join(self._root_certs_dir, root_cert)
-          LOGGER.info('Checking root cert: ' + str(root_cert_path))
-          args = f'{root_cert_path} {self._dev_cert_file}'
-          command = f'{bin_file} {args}'
-          response = util.run_command(command)
-          if 'device_cert.crt: OK' in str(response):
-            LOGGER.info('Device signed by cert:' + root_cert)
-            return True, 'Device signed by cert:' + root_cert
-          else:
-            LOGGER.info('Device not signed by cert: ' + root_cert)
-        except Exception as e:  # pylint: disable=W0718
-          LOGGER.error('Failed to check cert:' + root_cert)
-          LOGGER.error(str(e))
-    return False, 'Device certificate has not been signed'
+      LOGGER.info('Checking for authorized CA certificate chain')
+      device_cert_path = os.path.join(self._cert_out_dir, self._dev_cert_file)
+      return self.validate_cert_chain(device_cert_path=device_cert_path)
+
+  def validate_cert_chain(self, device_cert_path):
+    LOGGER.info('Validating certificate chain')
+    # Load the certificate from the PEM file
+    with open(device_cert_path, 'rb') as cert_file:
+      cert_bytes = cert_file.read()
+
+    # Load pem encoding into a certifiate so we can process the contents
+    certificate = crypto.load_certificate(crypto.FILETYPE_PEM, cert_bytes)
+
+    ca_issuer_cert, cert_file_path = self.get_ca_issuer(certificate)
+    if ca_issuer_cert is not None and cert_file_path is not None:
+      LOGGER.info('CA Issuer resolved')
+      cert_text = crypto.dump_certificate(crypto.FILETYPE_TEXT,
+                                          ca_issuer_cert).decode()
+      LOGGER.info(cert_text)
+      return self.validate_trusted_ca_signature_chain(
+          device_cert_path=device_cert_path,
+          intermediate_cert_path=cert_file_path)
+    else:
+      return False
+
+  def validate_trusted_ca_signature_chain(self, device_cert_path,
+                                          intermediate_cert_path):
+    bin_file = self._bin_dir + '/check_cert_chain_signature.sh'
+    # Combine the device and intermediate certificates
+    with open(device_cert_path, 'r', encoding='utf-8') as f:
+      dev_cert = f.read()
+    with open(intermediate_cert_path, 'r', encoding='utf-8') as f:
+      inter_cert = f.read()
+
+    combined_cert_name = 'device_cert_full.crt'
+    combined_cert = dev_cert + inter_cert
+    combined_cert_path = os.path.join(self._cert_out_dir,
+                                      combined_cert_name)
+    with open(combined_cert_path, 'w', encoding='utf-8') as f:
+      f.write(combined_cert)
+
+    # Use openssl script to validate the combined certificate
+    # against the available trusted CA's
+    args = f'{intermediate_cert_path} {combined_cert_path}'
+    command = f'{bin_file} {args}'
+    response = util.run_command(command)
+    return combined_cert_name +': OK' in str(response)
+
+  def get_ca_issuer(self, certificate):
+    cert_file_path = None
+    ca_issuer_cert = None
+    ca_issuers_uri = self.resolve_ca_issuer(certificate)
+    if ca_issuers_uri is not None:
+      ca_issuer_cert = self.get_certificate(ca_issuers_uri)
+      if ca_issuer_cert is not None:
+        # Write the intermediate certificate to file
+        cert_name = ca_issuers_uri.split('/')[-1]
+        cert_file_path = self.write_cert_to_file(cert_name, ca_issuer_cert)
+    return ca_issuer_cert, cert_file_path
+
+  def resolve_ca_issuer(self, certificate):
+    LOGGER.info('Resolving CA Issuer')
+    # Print the certificate information
+    cert_text = crypto.dump_certificate(crypto.FILETYPE_TEXT,
+                                        certificate).decode()
+    # Extract 'CA Issuers - URI' field
+    ca_issuers_uri = None
+    for line in cert_text.split('\n'):
+      if 'CA Issuers - URI' in line:
+        ca_issuers_uri = line.split(':', 1)[1].strip()
+        break
+    LOGGER.info(f'CA Issuers resolved: {ca_issuers_uri}')
+    return ca_issuers_uri
+
+  def get_certificate(self, uri, timeout=10):
+    LOGGER.info(f'Resolving certificate from {uri}')
+    certificate = None
+    try:
+      # Fetch the certificate file from the URI
+      response = requests.get(uri, timeout=timeout)
+      response.raise_for_status()  # Raise an error for HTTP errors
+
+      # Load the certificate from the PEM format
+      certificate_data = response.content
+
+      try:
+        LOGGER.info('Attempting to resolve certificate in PEM format')
+        # Load the certificate from the PEM format
+        certificate = x509.load_pem_x509_certificate(certificate_data,
+                                                     default_backend())
+        LOGGER.info('PEM format certificate resolved')
+      except ValueError:
+        # Load the certificate from the DER format
+        LOGGER.info('Failed to resolve PEM format, attempting DER format')
+        certificate = x509.load_der_x509_certificate(certificate_data,
+                                                     default_backend())
+        LOGGER.info('DER format certificate resolved.')
+      except Exception:  # pylint: disable=W0718
+        LOGGER.error('Failed to load certificate in expected formats')
+    except requests.exceptions.RequestException as e:
+      LOGGER.error(f'Error fetching certificate from URI: {e}')
+    return certificate
 
   def process_tls_server_results(self, tls_1_2_results, tls_1_3_results):
     results = ''
     if tls_1_2_results[0] is None and tls_1_3_results[0] is not None:
       # Validate only TLS 1.3 results
-      description = 'TLS 1.3' + (' not' if not tls_1_3_results[
-          0] else '') + ' validated: ' + tls_1_3_results[1]
+      description = 'TLS 1.3' + (' not' if not tls_1_3_results[0] else
+                                 '') + ' validated: ' + tls_1_3_results[1]
       results = tls_1_3_results[0], description
     elif tls_1_3_results[0] is None and tls_1_2_results[0] is not None:
       # Vaidate only TLS 1.2 results
-      description = 'TLS 1.2' + (' not' if not tls_1_2_results[
-          0] else '') + ' validated: ' + tls_1_2_results[1]
+      description = 'TLS 1.2' + (' not' if not tls_1_2_results[0] else
+                                 '') + ' validated: ' + tls_1_2_results[1]
       results = tls_1_2_results[0], description
     elif tls_1_3_results[0] is not None and tls_1_2_results[0] is not None:
       # Validate both results
-      description = 'TLS 1.2' + (' not' if not tls_1_2_results[
-          0] else '') + ' validated: ' + tls_1_2_results[1]
-      description += '\nTLS 1.3' + (' not' if not tls_1_3_results[
-          0] else '') + ' validated: ' + tls_1_3_results[1]
+      description = 'TLS 1.2' + (' not' if not tls_1_2_results[0] else
+                                 '') + ' validated: ' + tls_1_2_results[1]
+      description += '\nTLS 1.3' + (' not' if not tls_1_3_results[0] else
+                                    '') + ' validated: ' + tls_1_3_results[1]
       results = tls_1_2_results[0] or tls_1_3_results[0], description
     else:
       description = f'TLS 1.2 not validated: {tls_1_2_results[1]}'
@@ -219,7 +344,7 @@ def validate_tls_server(self, host, tls_version):
     if cert_pem:
 
       # Write pem encoding to a file
-      self.write_cert_to_file(cert_pem)
+      self.write_cert_to_file(self._dev_cert_file, cert_pem)
 
       # Load pem encoding into a certifiate so we can process the contents
       public_cert = crypto.load_certificate(crypto.FILETYPE_PEM, cert_pem)
@@ -249,9 +374,30 @@ def validate_tls_server(self, host, tls_version):
       LOGGER.info('Failed to resolve public certificate')
       return None, 'Failed to resolve public certificate'
 
-  def write_cert_to_file(self, pem_cert):
-    with open(self._dev_cert_file, 'w', encoding='UTF-8') as f:
-      f.write(pem_cert)
+  def write_cert_to_file(self, cert_name, cert):
+    try:
+      cert_file = os.path.join(self._cert_out_dir, cert_name)
+      if isinstance(cert, str):
+        with open(cert_file, 'w', encoding='UTF-8') as f:
+          f.write(cert)
+        return cert_file
+      elif isinstance(cert, bytes):
+        with open(cert_file, 'wb') as f:
+          f.write(cert)
+        return cert_file
+      elif isinstance(cert, x509.Certificate):
+        with open(cert_file, 'wb') as f:
+          # Serialize the certificate to PEM format
+          certificate_bytes = cert.public_bytes(
+              encoding=serialization.Encoding.PEM)
+          f.write(certificate_bytes)
+        return cert_file
+      else:
+        LOGGER.error(f'Cannot write certificate file, '
+                     f'unsupported content type: {type(cert)}')
+    except Exception as e:  # pylint: disable=W0718
+      LOGGER.error(f'Failed to write certificate to file: {e}')
+    return None
 
   def get_ciphers(self, capture_file, dst_ip, dst_port):
     bin_file = self._bin_dir + '/get_ciphers.sh'
@@ -537,8 +683,8 @@ def validate_tls_client(self, client_ip, tls_version, capture_files):
     if len(unsupported_tls_ips) > 0:
       tls_client_valid = False
       for ip, tls_versions in unsupported_tls_ips.items():
-        for tls_version in tls_versions:
-          tls_client_details += f'''\nUnsupported TLS {tls_version}
+        for version in tls_versions:
+          tls_client_details += f'''\nUnsupported TLS {version}
           connection detected to {ip}'''
     return tls_client_valid, tls_client_details
 

testing/unit/tls/CertAuth/TestRun_CA_Root.key

@@ -0,0 +1,27 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIEpQIBAAKCAQEAwAcDvl19KuhNYsrIkiQcaPlyCwcYNtBzE3TBnOCncreYJtZn
+5FyAS9VzRkTxDS6Ff08tCthJ5nsVS5ZVPNu4X+ZG9vWEUs1m+cEHdbaXsWdgZ49x
+U2/15mnHo/DqOFhfz8SmdZpGA5kKe+4etcyaW6KFml9wHY2KZqyokMY1ehAMcpGM
+1S16rnmHnK2tyAM27ro7x0CcMSq0G24SWlEK7WGOQindwY6gFND16rpR43xj8Tmr
+vkAPs5BrALvw2bkXcZO1Tjex9wke8sGVet7pfP91VA5UGD0pDqtoQaOdCsyDVyeA
+RZdFe6tN0bt3T0cbLKqedkM4PAYIBbincabnAwIDAQABAoIBAQCoVD3S3Q8A0twr
+UslZWjBRUQDIa/Ks8jM+BeZGx6PhatEEkoRF6VRJpZXELmEEBhjeDaDVVd4KHTEA
+roqPq6fG4QyqJXRWRVoUa0JHzMTSrWUTwuk7k/SBg46Oxnv64nUyoxasFo1zT++h
+zY28Hdvdoezt8uVL8qw07vtg1W9BTFIsMFSJTwere8OZwBWMpb8Vfh94hkCZV77E
+K+vdQNRAArrMTeDGO27T7u5Qrgv9sA7a9/jUSDwchrVLLP4WVVyHeJ2uvB865NQm
+8o7+H6LGbvdWedeXoN9Yc5QUtbUyrh3EcdV0M5TlBk8Z/fUT8zHdxQI+RMgfz+E/
+pbV8yuF5AoGBAPx7RvB9XEHqRGQvsDqVnxkKkDDE/Bq0OIHLWF5CajLnnhEtM+oJ
+sNZnxmRM5VK8X+O5s/1X1Gp9wcgf9+lZxTy27TKM/Gbgaq2aDYeUzUpO6OFuCcps
+25GMzrFQRb1gB2dlqhmk7t0L362L6OMpFqSq9htpIkhu/NowqZ6e1jr3AoGBAMK0
+EYv6L+RqBKIRCg4WqesxXsJXBRg2wY/XxCyOF8rJ3U3FYnsKl+pcKMIJ3V8IL4NB
+furxv/nNSPrgyowk1Olj+B6cWDxrMQSWOdT5HbB/CO38bUGlvyFntaHMF5Zu8MuG
+i2p1ragm80D7SgU5SqAqyffWEZMfzKohDJrzdYVVAoGBAN1MpHI4PvwbfVSfJAVF
+jcziIF5O2nYBjyHc/Ripd/IkZ7zAdSdm1RQoo5DYgYySRi4RYDznley9S3PA6Ygk
+QigkYiWTw8vcWkTNqZw0Bfhiz+Z/j59Y6N9bnvN125rQp8yuJHkTwrYHKUgxQLCT
+HCC7JLoD3aPFfGU2kAXTTQ0jAoGAbh9AiWYw7kTUaAGxKTTCWEbtLIfhVsephLzp
+tLWNWWIBLDqGr8bXE3OajdzceyJ7FQbXTPT8usHUFClOBiPS1Ep5jH6rHUkXSVva
+S822cBv5pfkOpoGjb7ZjsaZodOo8gTCQ696xIJkfHlLCk9/KiHqLDwThnc/vhw34
+Pi+S+Z0CgYEAo/rlxKuUryz96By9UKHNVZxKJhQ9bAyaQLtLHru72kNr8tCZqShF
+iB3cuU+fQE1rcxsIqhOiRUdDy8vRFQdcR5XmESU0XvpxIEWlTCS96A2zl5J1OEVD
+Sz/EsFII2P1Corpfsv6BI8zV4Q5/IligHzAI7+Bg6LkuYW93e8hX4bE=
+-----END RSA PRIVATE KEY-----

testing/unit/tls/CertAuth/Testrun_CA_Root.crt

@@ -0,0 +1,22 @@
+-----BEGIN CERTIFICATE-----
+MIIDkTCCAnkCFEVPbtILjbdASrq6LFIB3o2rIIDhMA0GCSqGSIb3DQEBCwUAMIGE
+MQswCQYDVQQGEwJVUzETMBEGA1UECAwKQ2FsaWZvcm5pYTEVMBMGA1UEBwwMTW91
+bnRhaW5WaWV3MRAwDgYDVQQKDAdUZXN0cnVuMRYwFAYDVQQLDA1RdWFsaWZpY2F0
+aW9uMR8wHQYDVQQDDBZUZXN0cnVuIFJTQSBTaWduaW5nIENBMB4XDTI0MDMyMDIw
+MzUwNVoXDTM0MDMyMDIwMzUwNVowgYQxCzAJBgNVBAYTAlVTMRMwEQYDVQQIDApD
+YWxpZm9ybmlhMRUwEwYDVQQHDAxNb3VudGFpblZpZXcxEDAOBgNVBAoMB1Rlc3Ry
+dW4xFjAUBgNVBAsMDVF1YWxpZmljYXRpb24xHzAdBgNVBAMMFlRlc3RydW4gUlNB
+IFNpZ25pbmcgQ0EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDABwO+
+XX0q6E1iysiSJBxo+XILBxg20HMTdMGc4Kdyt5gm1mfkXIBL1XNGRPENLoV/Ty0K
+2EnmexVLllU827hf5kb29YRSzWb5wQd1tpexZ2Bnj3FTb/Xmacej8Oo4WF/PxKZ1
+mkYDmQp77h61zJpbooWaX3AdjYpmrKiQxjV6EAxykYzVLXqueYecra3IAzbuujvH
+QJwxKrQbbhJaUQrtYY5CKd3BjqAU0PXqulHjfGPxOau+QA+zkGsAu/DZuRdxk7VO
+N7H3CR7ywZV63ul8/3VUDlQYPSkOq2hBo50KzINXJ4BFl0V7q03Ru3dPRxssqp52
+Qzg8BggFuKdxpucDAgMBAAEwDQYJKoZIhvcNAQELBQADggEBAInXBKcbZI5kKl6m
+2IVGWWW0SPye/4uLwtLUJtHGAAxeISjwdiiebVqcwZTHoJ4qnPKd3OMdGFvEWMPi
+vI1I/j4JFraD2Q885MhitVkmx59OCLNSR6QGtVqu6E73KADL92KZTE7Gdb2DtqpX
+zf2mz8MukSeMI7NZI7hp599DAj50N4mnQZPDJfuE+jTBjuSh7oay1x1m6N1eZZRT
+lFMWH2fXjlknpCpIVAcFYo3dEZsGl04z/NiQZrODBlt6CxheGwdbOXy/wPulkZ9R
+v9A7sLrK9rDih2qZGjWpbexTrlFlvC4z2UN/gh6P3IR8QuUv75MjPqQmQ9/9SV0+
+UrVwBdw=
+-----END CERTIFICATE-----

testing/unit/tls/CertAuth/Testrun_CA_Root.csr

@@ -0,0 +1,17 @@
+-----BEGIN CERTIFICATE REQUEST-----
+MIICyjCCAbICAQAwgYQxCzAJBgNVBAYTAlVTMRMwEQYDVQQIDApDYWxpZm9ybmlh
+MRUwEwYDVQQHDAxNb3VudGFpblZpZXcxEDAOBgNVBAoMB1Rlc3RydW4xFjAUBgNV
+BAsMDVF1YWxpZmljYXRpb24xHzAdBgNVBAMMFlRlc3RydW4gUlNBIFNpZ25pbmcg
+Q0EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDABwO+XX0q6E1iysiS
+JBxo+XILBxg20HMTdMGc4Kdyt5gm1mfkXIBL1XNGRPENLoV/Ty0K2EnmexVLllU8
+27hf5kb29YRSzWb5wQd1tpexZ2Bnj3FTb/Xmacej8Oo4WF/PxKZ1mkYDmQp77h61
+zJpbooWaX3AdjYpmrKiQxjV6EAxykYzVLXqueYecra3IAzbuujvHQJwxKrQbbhJa
+UQrtYY5CKd3BjqAU0PXqulHjfGPxOau+QA+zkGsAu/DZuRdxk7VON7H3CR7ywZV6
+3ul8/3VUDlQYPSkOq2hBo50KzINXJ4BFl0V7q03Ru3dPRxssqp52Qzg8BggFuKdx
+pucDAgMBAAGgADANBgkqhkiG9w0BAQsFAAOCAQEAZBPnCRmf8mmQr26XJWOri/EU
+B+PXPQ1CZInULKAC2g+Eyz279Bx/73kQ/+G++4p2XTiHzLH6V3d0sQzTmNKm7fz9
+jGuHSLkRYlo983dlTmnJTKdE+Z9heMbzIYuusbgt8SshHQejUxLpZ7muB+Kcg5wE
+andVgm4RY8sz9/Hwk1GRaMFjEeeOcN3S6ZAs086BWfsG/2rJbntPabA9GDNvdfqJ
+pp8opOB7p0Uxfy3eYEYNWdEp4s2JwaEoAWDFI2gHeawIbJ/cldY5YOg7AHY9nl8I
+XcdcV1mreAlrrZhODqg0EatUokgAyR1KDPIt4q6ME5nqjEjLaE3+dy7kGgbYWg==
+-----END CERTIFICATE REQUEST-----

testing/unit/tls/CertAuth/Testrun_CA_Root.srl

@@ -0,0 +1 @@
+31B6EB0B91479F0EC0DFC4D82787EB2736DF6908