Make message signature verifications optional in VTN

This makes the verification of TLS signatures optional in the VTN, which
is useful if the VEN is using client-side TLS but no message signatures.

Author: stan-janssen

VERSION

@@ -1 +1 @@
-0.5.30
+0.5.32

openleadr/messaging.py

@@ -117,7 +117,9 @@ def validate_xml_signature_none(xml_tree):
     assert xml_tree.find('.//{http://www.w3.org/2000/09/xmldsig#}X509Certificate') is None
 
 
-async def authenticate_message(request, message_tree, message_payload, fingerprint_lookup=None, ven_lookup=None):
+async def authenticate_message(request, message_tree, message_payload,
+                               fingerprint_lookup=None, ven_lookup=None,
+                               verify_message_signature=True):
     if request.secure and 'ven_id' in message_payload:
         connection_fingerprint = utils.get_cert_fingerprint_from_request(request)
         if connection_fingerprint is None:
@@ -152,21 +154,22 @@ async def authenticate_message(request, message_tree, message_payload, fingerpri
                    f"does not match the expected fingerprint '{expected_fingerprint}'")
             raise errors.NotRegisteredOrAuthorizedError(msg)
 
-        message_cert = utils.extract_pem_cert(message_tree)
-        message_fingerprint = utils.certificate_fingerprint(message_cert)
-        if message_fingerprint != expected_fingerprint:
-            msg = (f"The fingerprint of the certificate used to sign the message "
-                   f"{message_fingerprint} did not match the fingerprint that this "
-                   f"VTN has for you {expected_fingerprint}. Make sure you use the correct "
-                   "certificate to sign your messages.")
-            raise errors.NotRegisteredOrAuthorizedError(msg)
-
-        try:
-            validate_xml_signature(message_tree)
-        except ValueError:
-            msg = ("The message signature did not match the message contents. Please make sure "
-                   "you are using the correct XMLDSig algorithm and C14n canonicalization.")
-            raise errors.NotRegisteredOrAuthorizedError(msg)
+        if verify_message_signature:
+            message_cert = utils.extract_pem_cert(message_tree)
+            message_fingerprint = utils.certificate_fingerprint(message_cert)
+            if message_fingerprint != expected_fingerprint:
+                msg = (f"The fingerprint of the certificate used to sign the message "
+                       f"{message_fingerprint} did not match the fingerprint that this "
+                       f"VTN has for you {expected_fingerprint}. Make sure you use the correct "
+                       "certificate to sign your messages.")
+                raise errors.NotRegisteredOrAuthorizedError(msg)
+
+            try:
+                validate_xml_signature(message_tree)
+            except ValueError:
+                msg = ("The message signature did not match the message contents. Please make sure "
+                       "you are using the correct XMLDSig algorithm and C14n canonicalization.")
+                raise errors.NotRegisteredOrAuthorizedError(msg)
 
 
 def _create_replay_protect():

openleadr/server.py

@@ -48,7 +48,8 @@ class OpenADRServer:
     def __init__(self, vtn_id, cert=None, key=None, passphrase=None, fingerprint_lookup=None,
                  show_fingerprint=True, http_port=8080, http_host='127.0.0.1', http_cert=None,
                  http_key=None, http_key_passphrase=None, http_path_prefix='/OpenADR2/Simple/2.0b',
-                 requested_poll_freq=timedelta(seconds=10), http_ca_file=None, ven_lookup=None):
+                 requested_poll_freq=timedelta(seconds=10), http_ca_file=None, ven_lookup=None,
+                 verify_message_signatures=True):
         """
         Create a new OpenADR VTN (Server).
 
@@ -73,11 +74,18 @@ def __init__(self, vtn_id, cert=None, key=None, passphrase=None, fingerprint_loo
         :param str http_key_passphrase: The passphrase for the HTTP private key.
         :param ven_lookup: A callback that takes a ven_id and returns a dict containing the
                            ven_id, ven_name, fingerprint and registration_id.
+        :param verify_message_signatures: Whether to verify message signatures.
         """
         # Set up the message queues
 
         self.app = web.Application()
         self.services = {}
+
+        # Globally enable or disable the verification of message
+        # signatures. Only used in combination with TLS.
+        VTNService.verify_message_signatures = verify_message_signatures
+
+        # Create the separate OpenADR services
         self.services['event_service'] = EventService(vtn_id)
         self.services['report_service'] = ReportService(vtn_id)
         self.services['poll_service'] = PollService(vtn_id)

openleadr/service/vtn_service.py

@@ -32,6 +32,9 @@
 
 
 class VTNService:
+
+    verify_message_signatures = True
+
     def __init__(self, vtn_id):
         self.vtn_id = vtn_id
         self.handlers = {}
@@ -79,10 +82,12 @@ async def handler(self, request):
             if request.secure and 'ven_id' in message_payload:
                 if hasattr(self, 'fingerprint_lookup'):
                     await authenticate_message(request, message_tree, message_payload,
-                                               fingerprint_lookup=self.fingerprint_lookup)
+                                               fingerprint_lookup=self.fingerprint_lookup,
+                                               verify_message_signature=self.verify_message_signatures)
                 elif hasattr(self, 'ven_lookup'):
                     await authenticate_message(request, message_tree, message_payload,
-                                               ven_lookup=self.ven_lookup)
+                                               ven_lookup=self.ven_lookup,
+                                               verify_message_signature=self.verify_message_signatures)
                 else:
                     logger.error("Could not authenticate this VEN because "
                                  "you did not provide a 'ven_lookup' function. Please see "

setup.py

@@ -20,7 +20,7 @@
     long_description = file.read()
 
 setup(name='openleadr',
-      version='0.5.31',
+      version='0.5.32',
       description='Python3 library for building OpenADR Clients (VENs) and Servers (VTNs)',
       long_description=long_description,
       long_description_content_type='text/markdown',