Add config flag for testing (#181)

* Add flag for testing

Add a flag that can be enabled
by the integration tests to bypass
checking that messages are coming
from google pub/sub

* Fix linting errors

lira/api/notifications.py

@@ -46,7 +46,10 @@ def post(body):
 
 def receive_messages(body):
     """Receive messages from Google pub/sub topic."""
-    if not lira_utils._is_authenticated_pubsub(connexion.request):
+    lira_config = current_app.config
+    if not lira_utils._is_authenticated_pubsub(
+        connexion.request, test_mode=lira_config.test_mode
+    ):
         raise connexion.ProblemException(
             status=401,
             title='Unauthorized',

lira/lira_config.py

@@ -186,6 +186,10 @@ def __init__(self, config_object, *args, **kwargs):
         if not config_object.get('google_pubsub_topic'):
             self.google_pubsub_topic = f'hca-notifications-{env}'
 
+        # Only enable for integration testing
+        if not config_object.get('test_mode'):
+            config_object['test_mode'] = False
+
         # Check cromwell credentials
         use_caas = config_object.get('use_caas', None)
         if not use_caas:

lira/lira_utils.py

@@ -21,25 +21,35 @@
 LIRA_SERVER_HEADER = {'Server': 'Lira Service', 'Content-type': 'application/json'}
 
 
-def _is_authenticated_pubsub(request):
+def _is_authenticated_pubsub(request, test_mode=False):
     """ Check if the message is sent by Google:
-    https://google-auth.readthedocs.io/en/latest/reference/google.oauth2.id_token.html """
-    try:
-        # Get the Cloud Pub/Sub-generated JWT in the "Authorization" header.
-        bearer_token = request.headers.get('Authorization')
-        token = bearer_token.split(' ')[1]
-
-        # Verify and decode the JWT. `verify_oauth2_token` verifies
-        # the JWT signature, the `aud` claim, and the `exp` claim.
-        claim = id_token.verify_oauth2_token(token, google_transport_requests.Request())
-        # Must also verify the `iss` claim.
-        if claim['iss'] not in ['accounts.google.com', 'https://accounts.google.com']:
-            logger.error('Wrong issuer.')
-            return False
+    https://google-auth.readthedocs.io/en/latest/reference/google.oauth2.id_token.html
+    If the test_mode flag is set to True, skip authentication (only enable for tests).
+    """
+    if test_mode:
         return True
-    except Exception as e:
-        logger.error(f'Invalid token: {e}\n')
-        return False
+    else:
+        try:
+            # Get the Cloud Pub/Sub-generated JWT in the "Authorization" header.
+            bearer_token = request.headers.get('Authorization')
+            token = bearer_token.split(' ')[1]
+
+            # Verify and decode the JWT. `verify_oauth2_token` verifies
+            # the JWT signature, the `aud` claim, and the `exp` claim.
+            claim = id_token.verify_oauth2_token(
+                token, google_transport_requests.Request()
+            )
+            # Must also verify the `iss` claim.
+            if claim['iss'] not in [
+                'accounts.google.com',
+                'https://accounts.google.com',
+            ]:
+                logger.error('Wrong issuer.')
+                return False
+            return True
+        except Exception as e:
+            logger.error(f'Invalid token: {e}\n')
+            return False
 
 
 def response_with_server_header(body, status):