CMR-10388: Finalizing code

Author: eereiter

subscription/src/access_control.py

@@ -21,7 +21,7 @@ class AccessControl:
     Example Use of this class
     access_control = AccessControl()
     response = access_control.get_permissions('user1', 'C1200484253-CMR_ONLY')
-    The call is the same as 'curl https://cmr.sit.earthdata.nasa.gov/access-control/permissions?user_id=eereiter&concept_id=C1200484253-CMR_ONLY'
+    The call is the same as 'curl https://cmr.sit.earthdata.nasa.gov/access-control/permissions?user_id=user1&concept_id=C1200484253-CMR_ONLY'
     Return is either None (Null or Nil) (if check on response is false) or
     {"C1200484253-CMR_ONLY":["read","update","delete","order"]}
     """
@@ -53,13 +53,11 @@ def get_url_from_parameter_store(self):
 
             # construct the access control parameter names from the environment variable
             env_name = environment_name.lower()
-            logger.info(f"Environment Name converted is: {env_name}")
             pre_fix = f"/{env_name}/ingest/"
             protocol_param_name = f"{pre_fix}CMR_ACCESS_CONTROL_PROTOCOL"
             port_param_name = f"{pre_fix}CMR_ACCESS_CONTROL_PORT"
             host_param_name = f"{pre_fix}CMR_ACCESS_CONTROL_HOST"
             context_param_name = f"{pre_fix}CMR_ACCESS_CONTROL_RELATIVE_ROOT_URL"
-            logger.info(f"protocol_param_name: {protocol_param_name}")
 
             env_vars = Env_Vars()
             protocol = env_vars.get_var(name=protocol_param_name)
@@ -113,17 +111,17 @@ def has_read_permission(self, subscriber_id, collection_concept_id):
         try:
             # Call the get_permissions function
             permissions_str = self.get_permissions(subscriber_id, collection_concept_id)
-            logger.info(f"The type of object the permissions is: {type(permissions_str)}")
-            logger.info(f"If its json then turn it into a Dictionary: {json.loads(permissions_str)}")
 
-            permissions = json.loads(permissions_str)
-
-            # Check if the permissions is a dictionary
-            if isinstance(permissions, dict):
-                # Check if the collection_concept_id is in the permissions dictionary
-                if collection_concept_id in permissions:
-                    # Check if "read" is in the list of permissions for the collection
-                    return "read" in permissions[collection_concept_id]
+            if permissions_str:
+                permissions = json.loads(permissions_str)
+
+                # Check if the permissions is a dictionary
+                if isinstance(permissions, dict):
+                    # Check if the collection_concept_id is in the permissions dictionary
+                    if collection_concept_id in permissions:
+                        # Check if "read" is in the list of permissions for the collection
+                        return "read" in permissions[collection_concept_id]
+                    else: return False
                 else: return False
             else: return False
 

subscription/src/env_vars.py

@@ -12,7 +12,11 @@ def __init__(self):
         self.ssm_client = boto3.client('ssm', region_name=os.getenv("AWS_REGION"))
 
     def get_var(self, name, decryption=False):
-        logger.debug(f"Getting the environment variable called {name}")
+        """The name parameter looks like /sit/ingest/ENVIRONMENT_VAR. To check if the environment
+           variable exists strip off everything except for the actual variable name. Otherwise
+           go to the AWS ParameterStore and get the values."""
+
+        logger.debug(f"Subscription worker: Getting the environment variable called {name}")
         parts = name.split('/')
         os_name = next(part for part in reversed(parts) if part)
 

subscription/src/subscription_worker.py

@@ -23,7 +23,7 @@ def receive_message(sqs_client, queue_url):
         WaitTimeSeconds=(int (LONG_POLL_TIME)))
 
     if len(response.get('Messages', [])) > 0:
-        logger.info(f"Number of messages received: {len(response.get('Messages', []))}")
+        logger.debug(f"Number of messages received: {len(response.get('Messages', []))}")
     return response
 
 def delete_message(sqs_client, queue_url, receipt_handle):
@@ -38,19 +38,22 @@ def process_messages(sns_client, topic, messages, access_control):
     """Proess the message by first checking if the subscriber has permission to 
        see the notification. If so send it on, otherwise send a log message."""
     for message in messages.get("Messages", []):
-        logger.info(f"In Subscription worker process messages message: {message}")
-        message_body_str = message["Body"]
-        message_body = json.loads(message_body_str)
-        message_attributes = message_body["MessageAttributes"]
+        try:
+            message_body = json.loads(message["Body"])
+            message_attributes = message_body["MessageAttributes"]
+            logger.debug(f"Subscription worker: Received message including attributes: {message_body}")
+
+            subscriber = message_attributes['subscriber']['Value']
+            collection_concept_id = message_attributes['collection-concept-id']['Value']
+
+            if(access_control.has_read_permission(subscriber, collection_concept_id)):
+                logger.debug(f"Subscription worker: {subscriber} has permission to receive granule notifications for {collection_concept_id}")
+                sns_client.publish_message(topic, message)
+            else:
+                logger.info(f"Subscription worker: {subscriber} does not have read permission to receive notifications for {collection_concept_id}.")
+        except Exception as e:
+            logger.error(f"Subscription worker: There is a problem process messages {message}. {e}")
 
-        subscriber = message_attributes['subscriber']['Value']
-        collection_concept_id = message_attributes['collection-concept-id']['Value']
-        logger.info(f"Subscriber: {subscriber}")
-        logger.info(f"collection_concept_id: {collection_concept_id}")
-        if(access_control.has_read_permission(subscriber, collection_concept_id)):
-            sns_client.publish_message(topic, message)
-        else:
-            logger.info(f"Subscription worker: {subscriber} does not have read permission to receive notifications for {collection_concept_id}.")
 
 def poll_queue(running):
     """ Poll the SQS queue and process messages. """
@@ -77,7 +80,7 @@ def poll_queue(running):
                  delete_messages(sqs_client=sqs_client, queue_url=DEAD_LETTER_QUEUE_URL, messages=dl_messages)
 
         except Exception as e:
-             logger.warning(f"An error occurred receiving or deleting messages: {e}")
+             logger.error(f"An error occurred receiving or deleting messages: {e}")
 
 app = Flask(__name__)
 @app.route('/shutdown', methods=['POST'])

subscription/test/access_control_test.py

@@ -61,32 +61,32 @@ def test_get_permissions_failure(self, mock_get):
         result = self.access_control.get_permissions("user1", "C1200484253-CMR_ONLY")
         self.assertIsNone(result)
 
-#    @patch.object(AccessControl, 'get_permissions')
-#    def test_has_read_permission(self, mock_get_permissions):
+    @patch.object(AccessControl, 'get_permissions')
+    def test_has_read_permission(self, mock_get_permissions):
         # Test when user has read permission
-#        mock_get_permissions.return_value = {"C1200484253-CMR_ONLY": ["read", "update"]}
-#        result = self.access_control.has_read_permission("user1", "C1200484253-CMR_ONLY")
-#        self.assertTrue(result)
+        mock_get_permissions.return_value = "{\"C1200484253-CMR_ONLY\": [\"read\", \"update\"]}"
+        result = self.access_control.has_read_permission("user1", "C1200484253-CMR_ONLY")
+        self.assertTrue(result)
 
         # Test when user doesn't have read permission
-#        mock_get_permissions.return_value = {"C1200484253-CMR_ONLY": ["update"]}
-#        result = self.access_control.has_read_permission("user1", "C1200484253-CMR_ONLY")
-#        self.assertFalse(result)
+        mock_get_permissions.return_value = "{\"C1200484253-CMR_ONLY\": [\"update\"]}"
+        result = self.access_control.has_read_permission("user1", "C1200484253-CMR_ONLY")
+        self.assertFalse(result)
 
         # Test when concept_id is not in permissions
-#        mock_get_permissions.return_value = {"C1200484253-OTHER": ["read"]}
-#        result = self.access_control.has_read_permission("user1", "C1200484253-CMR_ONLY")
-#        self.assertFalse(result)
+        mock_get_permissions.return_value = "{\"C1200484253-OTHER\": [\"read\"]}"
+        result = self.access_control.has_read_permission("user1", "C1200484253-CMR_ONLY")
+        self.assertFalse(result)
 
         # Test when permissions is not a dictionary
-#        mock_get_permissions.return_value = None
-#        result = self.access_control.has_read_permission("user1", "C1200484253-CMR_ONLY")
-#        self.assertFalse(result)
+        mock_get_permissions.return_value = None
+        result = self.access_control.has_read_permission("user1", "C1200484253-CMR_ONLY")
+        self.assertFalse(result)
 
         # Test when get_permissions raises an exception
-#        mock_get_permissions.side_effect = Exception("API Error")
-#        result = self.access_control.has_read_permission("user1", "C1200484253-CMR_ONLY")
-#        self.assertFalse(result)
+        mock_get_permissions.side_effect = Exception("API Error")
+        result = self.access_control.has_read_permission("user1", "C1200484253-CMR_ONLY")
+        self.assertFalse(result)
 
     @patch('access_control.logger') 
     @patch.object(AccessControl, 'get_permissions')