adding certificate_source to crt layer (#747)

Author: rakshil14-2

awscrt/__init__.py

@@ -7,6 +7,7 @@
 __all__ = [
     'aio',
     'auth',
+    'aws_iot_metrics',
     'crypto',
     'http',
     'io',
@@ -16,7 +17,6 @@
     'mqtt_request_response',
     's3',
     'websocket',
-    'aws_iot_metrics',
 ]
 
 __version__ = '1.0.0.dev0'

awscrt/aws_iot_metrics.py

@@ -63,6 +63,23 @@ class _MetricsFeatureId(str, Enum):
 # Feature Value Constants
 
 
+def _certificate_source_metrics_value(source):
+    """Map _CertificateSource to its single-character metrics value.
+
+    Mapping: CERTIFICATE_FILES->A, PKCS11->B, WINDOWS_CERT_STORE->C, PKCS12_FILE->E.
+    "D" is reserved for Java KeyStore.
+    Returns None if source is None or unrecognized.
+    """
+    from awscrt.io import _CertificateSource
+    mapping = {
+        _CertificateSource.CERTIFICATE_FILES: "A",
+        _CertificateSource.PKCS11: "B",
+        _CertificateSource.WINDOWS_CERT_STORE: "C",
+        _CertificateSource.PKCS12_FILE: "E",
+    }
+    return mapping.get(source)
+
+
 def _protocol_version_metrics_value(protocol):
     """Map protocol version to its single-character metrics value.
 
@@ -221,11 +238,10 @@ def _get_encoded_feature_list(client_options):
         - D (outbound_topic_alias_behavior): from topic_aliasing_options.outbound_behavior
         - E (inbound_topic_alias_behavior): from topic_aliasing_options.inbound_behavior
         - H (http_proxy_type): HTTP or HTTPS based on proxy TLS settings
+        - I (certificate_source): detected from TlsContextOptions
         - J (tls_cipher_preference): mapped from TlsCipherPref on the TLS context
         - K (minimum_tls_version): mapped from TlsVersion on the TLS context
 
-    Feature I (certificate_source) is set at the IoT SDK level, not here.
-
     Args:
         client_options: MQTT5 ClientOptions dataclass.
     Returns:
@@ -277,7 +293,11 @@ def _get_encoded_feature_list(client_options):
         val = _http_proxy_type_metrics_value(client_options.http_proxy_options)
         features.append(f"{_MetricsFeatureId.HTTP_PROXY_TYPE.value}/{val}")
 
-    # I: certificate_source - Would need to be tracked from TLS context setup. This is set at a IoT SDK level
+    # I: certificate_source - detected from TlsContextOptions factory method
+    if client_options.tls_ctx is not None:
+        val = _certificate_source_metrics_value(client_options.tls_ctx._certificate_source)
+        if val:
+            features.append(f"{_MetricsFeatureId.CERTIFICATE_SOURCE.value}/{val}")
 
     # J: tls_cipher_preference - security policy
     if client_options.tls_ctx is not None:
@@ -306,6 +326,7 @@ def _get_encoded_feature_list_mqtt3(proxy_options, tls_ctx=None):
 
     Conditionally includes:
         - H (http_proxy_type): HTTP or HTTPS based on proxy TLS settings
+        - I (certificate_source): detected from TlsContextOptions
         - J (tls_cipher_preference): mapped from TlsCipherPref on the TLS context
         - K (minimum_tls_version): mapped from TlsVersion on the TLS context
 
@@ -324,6 +345,12 @@ def _get_encoded_feature_list_mqtt3(proxy_options, tls_ctx=None):
         val = _http_proxy_type_metrics_value(proxy_options)
         features.append(f"{_MetricsFeatureId.HTTP_PROXY_TYPE.value}/{val}")
 
+    # I: certificate_source - detected from TlsContextOptions factory method
+    if tls_ctx is not None:
+        val = _certificate_source_metrics_value(tls_ctx._certificate_source)
+        if val:
+            features.append(f"{_MetricsFeatureId.CERTIFICATE_SOURCE.value}/{val}")
+
     # J: tls_cipher_preference - security policy
     if tls_ctx is not None:
         val = _tls_cipher_preference_metrics_value(tls_ctx._cipher_pref)
@@ -387,10 +414,8 @@ def _create_metrics(user_metrics, crt_feature_list):
        IoTSDKMetricsVersion, IoTSDKFeature) are passed through unchanged.
 
     Args:
-        user_metrics : Metrics configuration from
-            the IoT SDK. May be None if no SDK-level metrics are provided.
-        crt_feature_list : Encoded CRT feature list string generated
-            by _get_encoded_feature_list or _get_encoded_feature_list_mqtt3.
+        user_metrics: Metrics configuration from the IoT SDK. May be None if no SDK-level metrics are provided.
+        crt_feature_list: Encoded CRT feature list string generated by _get_encoded_feature_list or _get_encoded_feature_list_mqtt3.
     Returns:
         AWSIoTMetrics: The final metrics object ready to be embedded in the
             MQTT CONNECT packet username field.
@@ -434,11 +459,12 @@ def _create_metrics(user_metrics, crt_feature_list):
 def _create_metrics_mqtt5(client_options):
     """Create the final AWSIoTMetrics object for an MQTT5 client.
 
-    Generates the CRT feature list from the full set of MQTT5 ClientOptions
+    Generates the CRT feature list from the full set of MQTT5 ClientOptions,
+    including detected certificate source from the TLS context.
 
     Args:
         client_options: MQTT5 ClientOptions dataclass containing all
-            connection configuration and optional user metrics.
+            connection configuration and optional user (AWSIoTMetrics) metrics.
     Returns:
         AWSIoTMetrics: The final metrics object with merged CRT and SDK features.
     """
@@ -447,18 +473,15 @@ def _create_metrics_mqtt5(client_options):
 
 
 def _create_metrics_mqtt3(user_metrics=None, proxy_options=None, tls_ctx=None):
-    """Creates the final AWSIoTMetrics object for an MQTT3 connection.
+    """Create the final AWSIoTMetrics object for an MQTT3 connection.
 
-    Generates the CRT feature list from the MQTT3 connection parameters
+    Generates the CRT feature list from the MQTT3 connection parameters,
+    including detected certificate source from the TLS context.
 
     Args:
-        user_metrics : Optional metrics configuration
-            provided by the IoT SDK. If None, defaults are used.
-        proxy_options : Optional HTTP proxy options
-            from the Connection, used to determine proxy type feature.
-        tls_ctx : Optional TLS context from the
-            connection, used to determine cipher preference and minimum TLS
-            version features.
+        user_metrics: Optional AWSIoTMetrics configuration provided by the IoT SDK.
+        proxy_options: Optional HTTP proxy options from the Connection.
+        tls_ctx: Optional ClientTlsContext from the connection.
     Returns:
         AWSIoTMetrics: The final metrics object with merged CRT and SDK features.
     """

awscrt/io.py

@@ -15,6 +15,20 @@
 from typing import Union
 
 
+class _CertificateSource(IntEnum):
+    """Certificate source types for mTLS authentication.
+
+    Used by TlsContextOptions to track which factory method created the
+    TLS configuration. The metrics reads this and encodes it as
+    feature ID "I" in the metrics string.
+    """
+    CERTIFICATE_FILES = 0  # PEM cert + key files
+    PKCS11 = 1  # Hardware security module via PKCS11
+    WINDOWS_CERT_STORE = 2  # Windows certificate source
+    # 3 is reserved for Java keystore (not applicable to Python)
+    PKCS12_FILE = 4  # PKCS#12 (.p12/.pfx) files
+
+
 class LogLevel(IntEnum):
     NoLogs = 0  #:
     Fatal = 1  #:
@@ -340,6 +354,7 @@ class TlsContextOptions:
         '_pkcs11_cert_file_path',
         '_pkcs11_cert_file_contents',
         '_windows_cert_store_path',
+        '_certificate_source',
     )
 
     def __init__(self):
@@ -396,6 +411,7 @@ def create_client_with_mtls(cert_buffer, key_buffer):
         opt = TlsContextOptions()
         opt.certificate_buffer = cert_buffer
         opt.private_key_buffer = key_buffer
+        opt._certificate_source = _CertificateSource.CERTIFICATE_FILES
 
         return opt
 
@@ -456,6 +472,7 @@ def create_client_with_mtls_pkcs11(*,
         opt._pkcs11_private_key_label = private_key_label
         opt._pkcs11_cert_file_path = cert_file_path
         opt._pkcs11_cert_file_contents = cert_file_contents
+        opt._certificate_source = _CertificateSource.PKCS11
         return opt
 
     @staticmethod
@@ -480,6 +497,7 @@ def create_client_with_mtls_pkcs12(pkcs12_filepath, pkcs12_password):
         opt = TlsContextOptions()
         opt.pkcs12_filepath = pkcs12_filepath
         opt.pkcs12_password = pkcs12_password
+        opt._certificate_source = _CertificateSource.PKCS12_FILE
         return opt
 
     @staticmethod
@@ -501,6 +519,7 @@ def create_client_with_mtls_windows_cert_store_path(cert_path):
         assert isinstance(cert_path, str)
         opt = TlsContextOptions()
         opt._windows_cert_store_path = cert_path
+        opt._certificate_source = _CertificateSource.WINDOWS_CERT_STORE
         return opt
 
     @staticmethod
@@ -614,7 +633,7 @@ class ClientTlsContext(NativeResource):
     Args:
         options (TlsContextOptions): Configuration options.
     """
-    __slots__ = ('_min_tls_ver', '_cipher_pref')
+    __slots__ = ('_min_tls_ver', '_cipher_pref', '_certificate_source')
 
     def __init__(self, options):
         assert isinstance(options, TlsContextOptions)
@@ -623,6 +642,7 @@ def __init__(self, options):
 
         self._min_tls_ver = options.min_tls_ver
         self._cipher_pref = options.cipher_pref
+        self._certificate_source = options._certificate_source
 
         self._binding = _awscrt.client_tls_ctx_new(
             options.min_tls_ver.value,

awscrt/mqtt.py

@@ -16,7 +16,7 @@
 from awscrt.io import ClientBootstrap, ClientTlsContext, SocketOptions
 from dataclasses import dataclass
 from awscrt.mqtt5 import Client as Mqtt5Client
-from awscrt.aws_iot_metrics import AWSIoTMetrics, IoTMetricsMetadata, _create_metrics_mqtt3
+from awscrt.aws_iot_metrics import AWSIoTMetrics, _create_metrics_mqtt3
 
 
 class QoS(IntEnum):
@@ -334,7 +334,7 @@ class Connection(NativeResource):
 
         disable_metrics (bool): Disable IoT SDK metrics in MQTT CONNECT packet username field, including SDK name, version, and platform. Default to False.
 
-        metrics (Optional[AWSIoTMetrics]) :  Optional metrics configuration for IoT SDK metrics reporting. If provided, the CRT will use the given metrics. If None, a default AWSIoTMetrics will be created.
+        metrics (Optional[AWSIoTMetrics]): Optional metrics configuration for IoT SDK metrics reporting. If provided, the CRT will use the given metrics. If None, a default AWSIoTMetrics will be created.
         """
 
     def __init__(self,

awscrt/mqtt5.py

@@ -15,7 +15,7 @@
 from dataclasses import dataclass
 from collections.abc import Sequence
 from inspect import signature
-from awscrt.aws_iot_metrics import AWSIoTMetrics, IoTMetricsMetadata, _create_metrics_mqtt5
+from awscrt.aws_iot_metrics import AWSIoTMetrics, _create_metrics_mqtt5
 
 
 class QoS(IntEnum):
@@ -1372,7 +1372,7 @@ class ClientOptions:
         on_lifecycle_event_connection_failure_fn (Callable[[LifecycleConnectFailureData],]): Callback for Lifecycle Event Connection Failure.
         on_lifecycle_event_disconnection_fn (Callable[[LifecycleDisconnectData],]): Callback for Lifecycle Event Disconnection.
         disable_metrics (bool): Disable IoT SDK metrics in MQTT CONNECT packet username field, including SDK name, version, and platform. Default to False.
-        metrics (Optional[AWSIoTMetrics]) :  Optional metrics configuration for IoT SDK metrics reporting. If provided, the CRT will use the given metrics. If None, a default AWSIoTMetrics will be created.
+        metrics (Optional[AWSIoTMetrics]): Optional metrics configuration for IoT SDK metrics reporting. If provided, the CRT will use the given metrics. If None, a default AWSIoTMetrics will be created.
 
     """
     host_name: str

docsrc/source/api/aws_iot_metrics.rst

@@ -3,4 +3,3 @@ awscrt.aws_iot_metrics
 
 .. automodule:: awscrt.aws_iot_metrics
     :members:
-

test/test_aws_iot_metrics.py

@@ -26,7 +26,7 @@
     InboundTopicAliasBehaviorType,
     TopicAliasingOptions,
 )
-from awscrt.io import ClientTlsContext, TlsContextOptions, TlsConnectionOptions, TlsVersion, TlsCipherPref
+from awscrt.io import ClientTlsContext, TlsContextOptions, TlsConnectionOptions, TlsVersion, TlsCipherPref, _CertificateSource
 from awscrt.http import HttpProxyOptions
 
 
@@ -419,5 +419,69 @@ def test_custom_library_name(self):
         self.assertEqual("MyCustomSDK/1.0", result.library_name)
 
 
+class TestCertificateSourceMetrics(NativeResourceTest):
+    """Test certificate source metrics encoding through the TLS context binding path."""
+
+    def _make_tls_ctx_with_source(self, source):
+        tls_options = TlsContextOptions()
+        tls_options._certificate_source = source
+        return ClientTlsContext(tls_options)
+
+    def test_certificate_source_certificate_files(self):
+        """CERTIFICATE_FILES source encodes as 'A' in feature list."""
+        tls_ctx = self._make_tls_ctx_with_source(_CertificateSource.CERTIFICATE_FILES)
+
+        options = ClientOptions(host_name="localhost", port=8883, tls_ctx=tls_ctx)
+        result = _get_encoded_feature_list(options)
+        self.assertIn(f"{_MetricsFeatureId.CERTIFICATE_SOURCE.value}/A", result)
+
+        result_mqtt3 = _get_encoded_feature_list_mqtt3(proxy_options=None, tls_ctx=tls_ctx)
+        self.assertIn(f"{_MetricsFeatureId.CERTIFICATE_SOURCE.value}/A", result_mqtt3)
+
+    def test_certificate_source_pkcs11(self):
+        """PKCS11 source encodes as 'B' in feature list."""
+        tls_ctx = self._make_tls_ctx_with_source(_CertificateSource.PKCS11)
+
+        options = ClientOptions(host_name="localhost", port=8883, tls_ctx=tls_ctx)
+        result = _get_encoded_feature_list(options)
+        self.assertIn(f"{_MetricsFeatureId.CERTIFICATE_SOURCE.value}/B", result)
+
+        result_mqtt3 = _get_encoded_feature_list_mqtt3(proxy_options=None, tls_ctx=tls_ctx)
+        self.assertIn(f"{_MetricsFeatureId.CERTIFICATE_SOURCE.value}/B", result_mqtt3)
+
+    def test_certificate_source_windows_cert_store(self):
+        """WINDOWS_CERT_STORE source encodes as 'C' in feature list."""
+        tls_ctx = self._make_tls_ctx_with_source(_CertificateSource.WINDOWS_CERT_STORE)
+
+        options = ClientOptions(host_name="localhost", port=8883, tls_ctx=tls_ctx)
+        result = _get_encoded_feature_list(options)
+        self.assertIn(f"{_MetricsFeatureId.CERTIFICATE_SOURCE.value}/C", result)
+
+        result_mqtt3 = _get_encoded_feature_list_mqtt3(proxy_options=None, tls_ctx=tls_ctx)
+        self.assertIn(f"{_MetricsFeatureId.CERTIFICATE_SOURCE.value}/C", result_mqtt3)
+
+    def test_certificate_source_pkcs12(self):
+        """PKCS12_FILE source encodes as 'E' in feature list."""
+        tls_ctx = self._make_tls_ctx_with_source(_CertificateSource.PKCS12_FILE)
+
+        options = ClientOptions(host_name="localhost", port=8883, tls_ctx=tls_ctx)
+        result = _get_encoded_feature_list(options)
+        self.assertIn(f"{_MetricsFeatureId.CERTIFICATE_SOURCE.value}/E", result)
+
+        result_mqtt3 = _get_encoded_feature_list_mqtt3(proxy_options=None, tls_ctx=tls_ctx)
+        self.assertIn(f"{_MetricsFeatureId.CERTIFICATE_SOURCE.value}/E", result_mqtt3)
+
+    def test_no_certificate_source_omits_feature(self):
+        """No certificate source (None) should omit feature I from metrics."""
+        tls_ctx = self._make_tls_ctx_with_source(None)
+
+        options = ClientOptions(host_name="localhost", port=8883, tls_ctx=tls_ctx)
+        result = _get_encoded_feature_list(options)
+        self.assertNotIn(f"{_MetricsFeatureId.CERTIFICATE_SOURCE.value}/", result)
+
+        result_mqtt3 = _get_encoded_feature_list_mqtt3(proxy_options=None, tls_ctx=tls_ctx)
+        self.assertNotIn(f"{_MetricsFeatureId.CERTIFICATE_SOURCE.value}/", result_mqtt3)
+
+
 if __name__ == '__main__':
     unittest.main()