feat: added validation of dt-asymmetric-signature.

The data connector now includes a JWT in the header dt-asymmertric-signature that is signed by an internal DT key. The public key can be accessed from the oidc discovery endpoint. Added validation of the token. The signature secrets are now considered deprecated and optional.

dtintegrations/__init__.py

@@ -1,4 +1,4 @@
-__version__ = "0.6.1"
+__version__ = "0.7.0"
 
 from dtintegrations import data_connector as data_connector  # noqa
 from dtintegrations import provider as provider  # noqa

dtintegrations/data_connector/http_push.py

@@ -1,6 +1,7 @@
 import jwt
 import json
 import hashlib
+import requests
 from typing import Any, Optional
 
 import disruptive  # type: ignore
@@ -21,15 +22,27 @@ class HttpPush(outputs.OutputBase):
         Labels from the source device forwarded by the Data Connector.
 
     """
-    def __init__(self, headers: dict, body: bytes, secret: str):
+    def __init__(
+        self,
+        headers: dict,
+        body: bytes,
+        secret: str = '',
+        org_id: str = '',
+        oidc_config_uri: str = (
+            'https://identity.disruptive-technologies.com/'
+            'data-connector/.well-known/openid-configuration'
+        )
+    ):
         """
-        Constructs the HtttpPush object given request contents.
+        Constructs the HttpPush object given request contents.
 
         """
 
         self._headers = headers
         self._body = body
         self._secret = secret
+        self._org_id = org_id
+        self._oidc_config_uri = oidc_config_uri
 
         self._body_dict = self._decode(headers, body, secret)
         super().__init__(self._body_dict)
@@ -43,13 +56,17 @@ def __repr__(self) -> str:
             'headers={},'\
             'body={},'\
             'secret={},'\
+            'org_id={},'\
+            'oidc_config_uri={}'\
             ')'
         return string.format(
             self.__class__.__module__,
             self.__class__.__name__,
             self._headers,
             self._body,
             repr(self._secret),
+            repr(self._org_id),
+            repr(self._oidc_config_uri),
         )
 
     def get_device_metadata(self) -> Optional[metadata.DeviceMetadata]:
@@ -69,7 +86,7 @@ def get_device_metadata(self) -> Optional[metadata.DeviceMetadata]:
         except KeyError:
             return None
 
-    def _decode(self, headers: dict, body: bytes, secret: str) -> dict:
+    def _decode(self, headers: dict, body: bytes, secret: str = '') -> dict:
         """
         Decodes the incoming event, first validating the source- and origin
         using a signature secret and the request header- and body.
@@ -84,6 +101,11 @@ def _decode(self, headers: dict, body: bytes, secret: str) -> dict:
         secret : str
             The secret to sign the request at source.
 
+            Deprecated:
+            This is now deprecated in favor of the
+            X-DT-JWT-Assertion header.
+            The secret is no longer required.
+
         Returns
         -------
         payload : HttpPush
@@ -97,28 +119,136 @@ def _decode(self, headers: dict, body: bytes, secret: str) -> dict:
 
         """
 
-        # Do some mild secret sanitization, ensuring populated string.
-        if isinstance(secret, str):
-            if len(secret) == 0:
-                raise disruptive.errors.ConfigurationError(
-                    'Secret is empty string.'
-                )
-        else:
-            raise TypeError(
-                f'Got secret type <{type(secret).__name__}>. Expected <str>.'
-            )
-
         # Isolate the token in request headers.
+        custom_token = None
         token = None
         for key in headers:
             if key.lower() == 'x-dt-signature':
+                custom_token = headers[key]
+            if key.lower() == 'dt-asymmetric-signature':
                 token = headers[key]
-                break
-        if token is None:
+
+        # Calculate the body SHA-256 checksum.
+        m = hashlib.sha256()
+        m.update(body)
+        checksum_sha256 = m.digest().hex()
+
+        # Validate the custom token if it exists.
+        if custom_token:
+            self._validate_custom_token(custom_token, checksum_sha256, secret)
+
+        # Validate the DT Data Connector token.
+        if token:
+            self._validate_dt_token(token, checksum_sha256)
+        else:
             raise disruptive.errors.ConfigurationError(
-                'Missing header X-Dt-Signature.'
+                'No Data Connector token found.'
             )
 
+        # Convert the body bytes string into a dictionary.
+        body_dict = json.loads(body.decode('utf-8'))
+
+        return dict(body_dict)
+
+    def _validate_dt_token(self, token: str, checksum: str) -> None:
+        """
+        Validates the Data Connector token using the signature secret.
+
+        Parameters
+        ----------
+        token : str
+            The token to validate.
+        checksum : str
+            The SHA-256 checksum of the request body.
+
+        Raises
+        ------
+        ConfigurationError
+            If the token is invalid, expired, or the checksum does not match,
+            or if is not signed by DTs OIDC provider.
+
+        """
+
+        try:
+            oidc_config = requests.get(self._oidc_config_uri).json()
+        except requests.exceptions.RequestException:
+            raise disruptive.errors.ConfigurationError(
+                'Failed to fetch OIDC configuration.'
+            )
+        except json.JSONDecodeError:
+            raise disruptive.errors.ConfigurationError(
+                'Failed to parse OIDC configuration.'
+            )
+
+        # Decode the token using the JWK client.
+        try:
+            jwks_client = jwt.PyJWKClient(
+                oidc_config['jwks_uri'], cache_jwk_set=True
+            )
+            signing_key = jwks_client.get_signing_key_from_jwt(token)
+            payload = jwt.decode(
+                token,
+                signing_key.key,
+                algorithms=oidc_config[
+                    'id_token_signing_alg_values_supported'
+                ],
+                issuer=oidc_config['issuer'],
+                options={
+                    'verify_signature': True,
+                    'verify_iss': True,
+                    'verify_exp': True,
+                    'verify_iat': True,
+                }
+            )
+        except jwt.exceptions.InvalidSignatureError:
+            raise disruptive.errors.ConfigurationError(
+                'Invalid signature.'
+            )
+        except jwt.exceptions.ExpiredSignatureError:
+            raise disruptive.errors.ConfigurationError(
+                'Signature has expired.'
+            )
+        except jwt.exceptions.InvalidIssuerError:
+            raise disruptive.errors.ConfigurationError(
+                'Invalid issuer: {}'.format(payload['iss'])
+            )
+        except jwt.exceptions.InvalidAlgorithmError:
+            raise disruptive.errors.ConfigurationError(
+                'Invalid algorithm.'
+            )
+        if self._org_id and payload['sub'] != self._org_id:
+            raise disruptive.errors.ConfigurationError(
+                'Invalid subject, should match the organization ID.'
+            )
+
+        # Calculate and compare the body SHA-256 checksum.
+        if payload['checksum_sha256'] != checksum:
+            raise disruptive.errors.ConfigurationError(
+                'Checksum mismatch.'
+            )
+
+    @staticmethod
+    def _validate_custom_token(token: str, checksum: str, secret: str) -> None:
+        """
+        Validates the custom token using the signature secret.
+
+        Parameters
+        ----------
+        token : str
+            The token to validate.
+        body : bytes
+            The request body bytes.
+        secret : str
+            The secret to sign the token at source.
+
+        Raises
+        ------
+        ConfigurationError
+            If the token is invalid, expired, or the checksum does not match,
+            or if the secret does not match the token signature.
+
+        """
+
         # Decode the token using the signature secret.
         try:
             payload = jwt.decode(
@@ -136,30 +266,31 @@ def _decode(self, headers: dict, body: bytes, secret: str) -> dict:
             )
 
         # Calculate and compare the body SHA-256 checksum.
-        m = hashlib.sha256()
-        m.update(body)
-        checksum_sha256 = m.digest().hex()
-        if payload['checksum_sha256'] != checksum_sha256:
+        if payload['checksum_sha256'] != checksum:
             raise disruptive.errors.ConfigurationError(
                 'Checksum mismatch.'
             )
 
-        # Convert the body bytes string into a dictionary.
-        body_dict = json.loads(body.decode('utf-8'))
-
-        return dict(body_dict)
-
     @staticmethod
-    def from_provider(request: Any, provider: str, secret: str) -> Any:
+    def from_provider(
+            request: Any,
+            provider: str,
+            secret: str = '',
+            org_id: str = '',
+            oidc_config_uri: str = (
+            'https://identity.disruptive-technologies.com/'
+            'data-connector/.well-known/openid-configuration'
+            ),
+    ) -> Any:
         """
         Decodes the incoming event using a specified provider, first validating
-        the the source- and origin using a signature secretand
+        the the source- and origin using a signature secret
         and the provider-specific request.
 
         Parameters
         ----------
         request : Any
-            Unmodified incoming request format of the spcified provider.
+            Unmodified incoming request format of the specified provider.
         provider : {"flask", "gcloud", "lambda", "azure"}, str
             Name of the :ref:`provider <integrations_provider>`
             used to receive the request.
@@ -183,4 +314,10 @@ def from_provider(request: Any, provider: str, secret: str) -> Any:
         r = dtrequest.Request(request, provider)
 
         # Use a more generic function for the validation process.
-        return HttpPush(r.headers, r.body_bytes, secret)
+        return HttpPush(
+            r.headers,
+            r.body_bytes,
+            secret,
+            org_id=org_id,
+            oidc_config_uri=oidc_config_uri,
+        )

setup.cfg

@@ -29,6 +29,8 @@ classifiers =
 python_requires = >=3.8
 install_requires =
     disruptive >= 1.6.0
+    cryptography
+    types-requests
 packages = find:
 
 [options.packages.find]

tests/events.py

@@ -16,20 +16,22 @@ def __init__(self,
 
 touch = Event(
     headers={
-        'X-Dt-Signature': 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJjaGVja3N1bSI6ImYyYjVkYzA0OTY0MzcyMDFkM2NlZTE1NmMyMzNhMTMzNTYwM2Q0NjUiLCJjaGVja3N1bV9zaGEyNTYiOiJhZmQ4MWExNmJiOWMwNzZlNTdmOTI3MjhmMjg0ZmY1NTgzMWJmZGYxZDNjMmMxMGI4MzM1NjdmNTIxZDc4MmRlIiwiZXhwIjoxNjQ0NTAwNTMzLCJpYXQiOjE2NDQ0OTY5MzMsImp0aSI6ImM4MmdnOTloZ2EzNWpjMmFma2VnIn0.gm8m_HrhlOnAyS08CfR_RNYdZQIpGV6E8bk3UiAR3uo'  # noqa
+        'X-Dt-Signature': 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJjaGVja3N1bSI6ImYyYjVkYzA0OTY0MzcyMDFkM2NlZTE1NmMyMzNhMTMzNTYwM2Q0NjUiLCJjaGVja3N1bV9zaGEyNTYiOiJhZmQ4MWExNmJiOWMwNzZlNTdmOTI3MjhmMjg0ZmY1NTgzMWJmZGYxZDNjMmMxMGI4MzM1NjdmNTIxZDc4MmRlIiwiZXhwIjoxNjQ0NTAwNTMzLCJpYXQiOjE2NDQ0OTY5MzMsImp0aSI6ImM4MmdnOTloZ2EzNWpjMmFma2VnIn0.gm8m_HrhlOnAyS08CfR_RNYdZQIpGV6E8bk3UiAR3uo',  # noqa
+        'DT-Asymmetric-Signature': 'eyJhbGciOiJFUzI1NiIsImtpZCI6ImNwNjVnZG9uMWU0YmhiNWljYm1nIiwidHlwIjoiSldUIn0.eyJjaGVja3N1bV9zaGEyNTYiOiJhZmQ4MWExNmJiOWMwNzZlNTdmOTI3MjhmMjg0ZmY1NTgzMWJmZGYxZDNjMmMxMGI4MzM1NjdmNTIxZDc4MmRlIiwiZXhwIjoxNzMwMzgyNTg0LCJpYXQiOjE3MzAzODIyODQsImlzcyI6Imh0dHBzOi8vaWRlbnRpdHkuZGV2LmRpc3J1cHRpdmUtdGVjaG5vbG9naWVzLmNvbS9kYXRhLWNvbm5lY3RvciIsInN1YiI6InRlc3Qtb3JnLWlkIn0.RYTAdKD_CVhhu-kChnqtHzSB3wFOtidBYYhu39k5mJCuO3PLBzkHwzKe73kx4hEJ1YUR-jQGNZzqhSxpQ523Yw' # noqa
     },
     body_str='{"event":{"eventId":"c82gg9catj3vmh4248og","targetName":"projects/c14u9p094l47cdv1o3pg/devices/emuc17m6d7lq0bgk44smcqg","eventType":"touch","data":{"touch":{"updateTime":"2022-02-10T12:42:13.313949Z"}},"timestamp":"2022-02-10T12:42:13.313949Z"},"labels":{"name":"touch"},"metadata":{"deviceId":"emuc17m6d7lq0bgk44smcqg","projectId":"c14u9p094l47cdv1o3pg","deviceType":"touch","productNumber":""}}',  # noqa
     body={'event': {'eventId': 'c82gg9catj3vmh4248og', 'targetName': 'projects/c14u9p094l47cdv1o3pg/devices/emuc17m6d7lq0bgk44smcqg', 'eventType': 'touch', 'data': {'touch': {'updateTime': '2022-02-10T12:42:13.313949Z'}}, 'timestamp': '2022-02-10T12:42:13.313949Z'}, 'labels': {'name': 'touch'}, 'metadata': {'deviceId': 'emuc17m6d7lq0bgk44smcqg', 'projectId': 'c14u9p094l47cdv1o3pg', 'deviceType': 'touch', 'productNumber': ''}},  # noqa
-    payload={'checksum': 'f2b5dc0496437201d3cee156c233a1335603d465', 'checksum_sha256': 'afd81a16bb9c076e57f92728f284ff55831bfdf1d3c2c10b833567f521d782de', 'exp': 1644500533, 'iat': 1644496933, 'jti': 'c82gg99hga35jc2afkeg'},  # noqa
+    payload={'checksum': 'f2b5dc0496437201d3cee156c233a1335603d465', 'checksum_sha256': 'afd81a16bb9c076e57f92728f284ff55831bfdf1d3c2c10b833567f521d782de', 'exp': 1644500533, 'iat': 1644496933, 'jti': 'c82gg99hga35jc2afkeg', 'sub': 'test-org-id'},  # noqa
 )
 
 temperature = Event(
     headers={
-        'X-Dt-Signature': 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJjaGVja3N1bSI6IjBmNzUzOTM5YTEzNzdhM2M5NGFiZDJjZmViZGMyY2I0NzU2YzNmYzMiLCJjaGVja3N1bV9zaGEyNTYiOiI1YjVmYzJlN2M3NjM3YWIzN2RiNDJiNzQ1YmY2ZjEyOWZhZTk3ODJmOTUzZWQzZGM2NDM1YzMyMTQ4ZjQxNWQ0IiwiZXhwIjoxNzE2OTgyNDQzLCJpYXQiOjE3MTY5Nzg4NDMsImp0aSI6ImNwYmc5NnQxaXNqbHIydmRvcmMwIn0.A4r91SKDW4IqYtb1eXuTQCUxM6GnouPbt94Ywd-vCpU' # noqa
+        'X-Dt-Signature': 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJjaGVja3N1bSI6IjBmNzUzOTM5YTEzNzdhM2M5NGFiZDJjZmViZGMyY2I0NzU2YzNmYzMiLCJjaGVja3N1bV9zaGEyNTYiOiI1YjVmYzJlN2M3NjM3YWIzN2RiNDJiNzQ1YmY2ZjEyOWZhZTk3ODJmOTUzZWQzZGM2NDM1YzMyMTQ4ZjQxNWQ0IiwiZXhwIjoxNzE2OTgyNDQzLCJpYXQiOjE3MTY5Nzg4NDMsImp0aSI6ImNwYmc5NnQxaXNqbHIydmRvcmMwIn0.A4r91SKDW4IqYtb1eXuTQCUxM6GnouPbt94Ywd-vCpU', # noqa
+        'DT-Asymmetric-Signature': 'eyJhbGciOiJFUzI1NiIsImtpZCI6ImNwNjVnZG9uMWU0YmhiNWljYm1nIiwidHlwIjoiSldUIn0.eyJjaGVja3N1bV9zaGEyNTYiOiJhZmQ4MWExNmJiOWMwNzZlNTdmOTI3MjhmMjg0ZmY1NTgzMWJmZGYxZDNjMmMxMGI4MzM1NjdmNTIxZDc4MmRlIiwiZXhwIjoxNzMwMzgyNTg0LCJpYXQiOjE3MzAzODIyODQsImlzcyI6Imh0dHBzOi8vaWRlbnRpdHkuZGV2LmRpc3J1cHRpdmUtdGVjaG5vbG9naWVzLmNvbS9kYXRhLWNvbm5lY3RvciIsInN1YiI6InRlc3Qtb3JnLWlkIn0.RYTAdKD_CVhhu-kChnqtHzSB3wFOtidBYYhu39k5mJCuO3PLBzkHwzKe73kx4hEJ1YUR-jQGNZzqhSxpQ523Yw' # noqa
     },
     body_str='{"event":{"eventId":"cpbg96qngr7ai7ldarrg","targetName":"projects/ccol8iuk9smqiha4e8l0/devices/emucdd6ef2eu277bo8f52vg","eventType":"temperature","data":{"temperature":{"value":27,"updateTime":"2024-05-29T10:34:03.970415Z","samples":[{"value":27,"sampleTime":"2024-05-29T10:34:03.970415Z"}],"isBackfilled":false}},"timestamp":"2024-05-29T10:34:03.970415Z"},"labels":{"name":"test temp"},"metadata":{"deviceId":"emucdd6ef2eu277bo8f52vg","projectId":"ccol8iuk9smqiha4e8l0","deviceType":"temperature","productNumber":""}}', # noqa
     body={'event':{'eventId':'cpbg96qngr7ai7ldarrg','targetName':'projects/ccol8iuk9smqiha4e8l0/devices/emucdd6ef2eu277bo8f52vg','eventType':'temperature','data':{'temperature':{'value':27,'updateTime':'2024-05-29T10:34:03.970415Z','samples':[{'value':27,'sampleTime':'2024-05-29T10:34:03.970415Z'}],'isBackfilled':False}},'timestamp':'2024-05-29T10:34:03.970415Z'},'labels':{'name':'test temp'},'metadata':{'deviceId':'emucdd6ef2eu277bo8f52vg','projectId':'ccol8iuk9smqiha4e8l0','deviceType':'temperature','productNumber':''}}, # noqa
-    payload={'checksum': '0f753939a1377a3c94abd2cfebdc2cb4756c3fc3', 'checksum_sha256': '5b5fc2e7c7637ab37db42b745bf6f129fae9782f953ed3dc6435c32148f415d4', 'exp': 1716982443, 'iat': 1716978843, 'jti': 'cpbg96t1isjlr2vdorc0'}, # noqa
+    payload={'checksum': '0f753939a1377a3c94abd2cfebdc2cb4756c3fc3', 'checksum_sha256': '5b5fc2e7c7637ab37db42b745bf6f129fae9782f953ed3dc6435c32148f415d4', 'exp': 1716982443, 'iat': 1716978843, 'jti': 'cpbg96t1isjlr2vdorc0', 'sub': 'test-org-id'}, # noqa
 )
 
 metadata = {

tests/framework.py

@@ -91,5 +91,10 @@ def __init__(self, mocker):
             side_effect=self._patched_jwt_decode,
         )
 
-    def _patched_jwt_decode(self, token: str, secret: str, algorithms: list):
+    def _patched_jwt_decode(self,
+        token: str,
+        secret: str,
+        algorithms: list,
+        issuer: str = '',
+        options: str = '') -> dict:
         return self.event.payload