From 3659c86394bf0e7dabbb3d87212b58919cacf9b4 Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Tue, 18 Feb 2025 13:23:47 -0500 Subject: [PATCH 01/29] CMR-10388: Adding access control to the subscription workflow. --- .../metadata_db/services/subscriptions.clj | 50 +++--- .../test/services/subscriptions_test.clj | 165 ++++++++++-------- subscription/src/access_control.py | 24 +++ subscription/src/subscription_worker.py | 19 +- 4 files changed, 157 insertions(+), 101 deletions(-) diff --git a/metadata-db-app/src/cmr/metadata_db/services/subscriptions.clj b/metadata-db-app/src/cmr/metadata_db/services/subscriptions.clj index b193a3da3d..bb5c1cfa02 100644 --- a/metadata-db-app/src/cmr/metadata_db/services/subscriptions.clj +++ b/metadata-db-app/src/cmr/metadata_db/services/subscriptions.clj @@ -63,9 +63,9 @@ "Creates a mode-to-endpoints map given a list of subscriptions all for the same one collection. Returns a map in this structure: { - new: ['sqs:arn:111', 'https://www.url1.com', 'https://www.url2.com'], - update: ['sqs:arn:111'], - delete: ['https://www.url1.com'] + new: [['sqs:arn:111', 'user1'], ['https://www.url1.com', 'user2'], ['https://www.url2.com', 'user3']], + update: [['sqs:arn:111', 'user1']], + delete: [['https://www.url1.com', 'user2']] }" [subscriptions-of-same-collection] @@ -73,9 +73,10 @@ (doseq [sub subscriptions-of-same-collection] (let [temp-map (atom {}) modes (get-in sub [:metadata :Mode]) - endpoint (get-in sub [:metadata :EndPoint])] + endpoint (get-in sub [:metadata :EndPoint]) + subscriber (get-in sub [:metadata :SubscriberId])] (doseq [mode modes] - (swap! temp-map assoc mode #{endpoint})) + (swap! temp-map assoc mode #{[endpoint, subscriber]})) (let [merged-map (merge-with union @final-map @temp-map)] (swap! final-map (fn [n] merged-map))))) @final-map)) @@ -188,8 +189,8 @@ and put those into a map. The end result looks like: { coll_1: { Mode: { - New: #(URL1, SQS1), - Update: #(URL2) + New: #([URL1, user1], [SQS1, user2]), + Update: #([URL2, user3]) } }, coll_2: {...}" @@ -266,9 +267,10 @@ (defn create-message-attributes "Create the notification message attributes so that the notifications can be filtered to the correct subscribing endpoint." - [collection-concept-id mode] + [collection-concept-id mode subscriber] {"collection-concept-id" collection-concept-id - "mode" mode}) + "mode" mode + "subscriber" subscriber}) (defn create-message-subject "Creates the message subject." @@ -288,16 +290,20 @@ (defn- create-message-attributes-map "Create message attribute map that SQS uses to filter out messages from the SNS topic." - [endpoint mode coll-concept-id] - (cond - (or (is-valid-sqs-arn endpoint) - (is-local-test-queue endpoint)) (cond - (= "Delete" mode) (create-message-attributes coll-concept-id "Delete") - (= "New" mode) (create-message-attributes coll-concept-id "New") - (= "Update" mode) (create-message-attributes coll-concept-id "Update")) - (is-valid-subscription-endpoint-url endpoint) {"endpoint" endpoint - "endpoint-type" "url" - "mode" mode})) + [endpoint-set mode coll-concept-id] + (let [endpoint (first endpoint-set) + subscriber (second endpoint-set)] + (cond + (or (is-valid-sqs-arn endpoint) + (is-local-test-queue endpoint)) (cond + (= "Delete" mode) (create-message-attributes coll-concept-id "Delete" subscriber) + (= "New" mode) (create-message-attributes coll-concept-id "New" subscriber) + (= "Update" mode) (create-message-attributes coll-concept-id "Update" subscriber)) + (is-valid-subscription-endpoint-url endpoint) {"endpoint" endpoint + "endpoint-type" "url" + "mode" mode + "subscriber" subscriber})) + ) (defn publish-subscription-notification-if-applicable "Publish a notification to the topic if the passed-in concept is a granule @@ -313,16 +319,16 @@ endpoint-list (get-in sub-cache-map ["Mode" gran-concept-mode]) result-array (atom [])] ;; for every endpoint in the list create an attributes/subject map and send it along its way - (doseq [endpoint endpoint-list] + (doseq [endpoint-set endpoint-list] (let [topic (get-in context [:system :sns :internal]) coll-concept-id (:parent-collection-id (:extra-fields concept)) message (create-notification-message-body concept) - message-attributes-map (create-message-attributes-map endpoint gran-concept-mode coll-concept-id) + message-attributes-map (create-message-attributes-map endpoint-set gran-concept-mode coll-concept-id) subject (create-message-subject gran-concept-mode)] (when (and message-attributes-map subject) (let [result (topic-protocol/publish topic message message-attributes-map subject) duration (- (System/currentTimeMillis) start)] - (debug (format "Subscription publish for endpoint %s took %d ms." endpoint duration)) + (debug (format "Subscription publish for endpoint %s took %d ms." (first endpoint-set) duration)) (swap! result-array (fn [n] (conj @result-array result))))))) @result-array))))) diff --git a/metadata-db-app/test/cmr/metadata_db/test/services/subscriptions_test.clj b/metadata-db-app/test/cmr/metadata_db/test/services/subscriptions_test.clj index 386dab7bbb..5e5881e209 100644 --- a/metadata-db-app/test/cmr/metadata_db/test/services/subscriptions_test.clj +++ b/metadata-db-app/test/cmr/metadata_db/test/services/subscriptions_test.clj @@ -52,14 +52,16 @@ :metadata "{\"CollectionConceptId\":\"C12345-PROV1\", \"EndPoint\":\"some-endpoint\", \"Mode\":[\"New\"], - \"Method\":\"ingest\"}", + \"Method\":\"ingest\", + \"SubscriberId\":\"user1\"}", :extra-fields {:collection-concept-id "C12345-PROV1"}}))} (is (= 1 (subscriptions/change-subscription-in-cache test-context {:concept-type :subscription :deleted false :metadata {:CollectionConceptId "C12345-PROV1" :EndPoint "some-endpoint" :Mode ["New"] - :Method "ingest"} + :Method "ingest" + :SubscriberId "user1"} :extra-fields {:collection-concept-id "C12345-PROV1"}}))))) (testing "Delete a subscription from the cache" (with-bindings {#'subscriptions/get-subscriptions-from-db (fn [_context _coll-concept-id] '({:concept-type :subscription @@ -67,14 +69,16 @@ :metadata "{\"CollectionConceptId\":\"C12345-PROV1\", \"EndPoint\":\"some-endpoint\", \"Mode\":[\"New\"], - \"Method\":\"ingest\"}", + \"Method\":\"ingest\", + \"SubscriberId\":\"user1\"}", :extra-fields {:collection-concept-id "C12345-PROV1"}}))} (is (= 1 (subscriptions/change-subscription-in-cache test-context {:concept-type :subscription :deleted true :metadata {:CollectionConceptId "C12345-PROV1" :EndPoint "some-endpoint" :Mode ["New"] - :Method "ingest"} + :Method "ingest" + :SubscriberId "user1"} :extra-fields {:collection-concept-id "C12345-PROV1"}}))))) (testing "adding and removing subscriptions from the cache." @@ -86,75 +90,81 @@ "Adding 1 subscription" - {"C12345-PROV1" {"Mode" {"Update" (set ["some-endpoint"])}}} + {"C12345-PROV1" {"Mode" {"Update" (set [["some-endpoint" "user1"]])}}} {:metadata {:CollectionConceptId "C12345-PROV1"}} '({:metadata "{\"CollectionConceptId\":\"C12345-PROV1\", \"EndPoint\":\"some-endpoint\", \"Mode\":[\"Update\"], - \"Method\":\"ingest\"}", + \"Method\":\"ingest\", + \"SubscriberId\":\"user1\"}", :extra-fields {:collection-concept-id "C12345-PROV1"} :deleted false :concept-type :subscription}) "Adding duplicate subscription" - {"C12345-PROV1" {"Mode" {"Update" (set ["some-endpoint"])}}} + {"C12345-PROV1" {"Mode" {"Update" (set [["some-endpoint" "user1"]])}}} {:metadata {:CollectionConceptId "C12345-PROV1"}} '({:metadata "{\"CollectionConceptId\":\"C12345-PROV1\", \"EndPoint\":\"some-endpoint\", \"Mode\":[\"Update\"], - \"Method\":\"ingest\"}", + \"Method\":\"ingest\", + \"SubscriberId\":\"user1\"}", :extra-fields {:collection-concept-id "C12345-PROV1"} :deleted false :concept-type :subscription}) "Adding override subscription" - {"C12345-PROV1" {"Mode" {"New" (set ["some-endpoint"]) - "Delete" (set ["some-endpoint"])}}} + {"C12345-PROV1" {"Mode" {"New" (set [["some-endpoint" "user1"]]) + "Delete" (set [["some-endpoint" "user1"]])}}} {:metadata {:CollectionConceptId "C12345-PROV1"}} '({:metadata "{\"CollectionConceptId\":\"C12345-PROV1\", \"EndPoint\":\"some-endpoint\", \"Mode\":[\"New\", \"Delete\"], - \"Method\":\"ingest\"}" + \"Method\":\"ingest\", + \"SubscriberId\":\"user1\"}" :extra-fields {:collection-concept-id "C12345-PROV1"} :deleted false :concept-type :subscription}) "Adding new subscription that matches the one from before." - {"C12345-PROV1" {"Mode" {"New" (set ["some-endpoint"]) - "Delete" (set ["some-endpoint"])}} - "C12346-PROV1" {"Mode" {"New" (set ["some-endpoint"]) - "Delete" (set ["some-endpoint"])}}} + {"C12345-PROV1" {"Mode" {"New" (set [["some-endpoint" "user1"]]) + "Delete" (set [["some-endpoint" "user1"]])}} + "C12346-PROV1" {"Mode" {"New" (set [["some-endpoint" "user1"]]) + "Delete" (set [["some-endpoint" "user1"]])}}} {:metadata {:CollectionConceptId "C12346-PROV1"}} '({:metadata "{\"CollectionConceptId\":\"C12346-PROV1\", \"EndPoint\":\"some-endpoint\", \"Mode\":[\"New\", \"Delete\"], - \"Method\":\"ingest\"}", + \"Method\":\"ingest\", + \"SubscriberId\":\"user1\"}", :extra-fields {:collection-concept-id "C12346-PROV1"} :deleted false :concept-type :subscription}) "Removing 1 subscription" - {"C12346-PROV1" {"Mode" {"New" (set ["some-endpoint"]) - "Delete" (set ["some-endpoint"])}}} + {"C12346-PROV1" {"Mode" {"New" (set [["some-endpoint" "user1"]]) + "Delete" (set [["some-endpoint" "user1"]])}}} {:metadata {:CollectionConceptId "C12345-PROV1"}} ;; even though C12346-PROV1 is in the db, we are search only for ;; concepts with the collection-concept-id. '({:metadata "{\"CollectionConceptId\":\"C12345-PROV1\", \"EndPoint\":\"some-endpoint\", \"Mode\":[\"New\", \"Delete\"], - \"Method\":\"ingest\"}" + \"Method\":\"ingest\", + \"SubscriberId\":\"user1\"}" :extra-fields {:collection-concept-id "C12345-PROV1"} :deleted true :concept-type :subscription}) "Removing same subscription" - {"C12346-PROV1" {"Mode" {"New" (set ["some-endpoint"]) - "Delete" (set ["some-endpoint"])}}} + {"C12346-PROV1" {"Mode" {"New" (set [["some-endpoint" "user1"]]) + "Delete" (set [["some-endpoint" "user1"]])}}} {:metadata {:CollectionConceptId "C12345-PROV1"}} '({:metadata "{\"CollectionConceptId\":\"C12345-PROV1\", - \"EndPoint\":\"some-endpoint\", - \"Mode\":[\"New\", \"Delete\"], - \"Method\":\"ingest\"}" + \"EndPoint\":\"some-endpoint\", + \"Mode\":[\"New\", \"Delete\"], + \"Method\":\"ingest\", + \"SubscriberId\":\"user1\"}" :extra-fields {:collection-concept-id "C12345-PROV1"} :deleted true :concept-type :subscription}) @@ -163,9 +173,10 @@ nil {:metadata {:CollectionConceptId "C12346-PROV1"}} '({:metadata "{\"CollectionConceptId\":\"C12346-PROV1\", - \"EndPoint\":\"some-endpoint\", - \"Mode\":[\"New\", \"Delete\"], - \"Method\":\"ingest\"}", + \"EndPoint\":\"some-endpoint\", + \"Mode\":[\"New\", \"Delete\"], + \"Method\":\"ingest\", + \"SubscriberId\":\"user1\"}", :extra-fields {:collection-concept-id "C12346-PROV1"} :deleted true :concept-type :subscription}) @@ -181,13 +192,17 @@ (is (= {:metadata {:CollectionConceptId "C12345-PROV1" :EndPoint "ARN" :Mode ["New" "Delete"] - :Method "ingest"} + :Method "ingest" + :SubscriberId "user1"} :concept-type :subscription} - (subscriptions/add-or-delete-ingest-subscription-in-cache test-context {:metadata "{\"CollectionConceptId\":\"C12345-PROV1\", - \"EndPoint\":\"ARN\", - \"Mode\":[\"New\", \"Delete\"], - \"Method\":\"ingest\"}" - :concept-type :subscription})))))))) + (subscriptions/add-or-delete-ingest-subscription-in-cache + test-context + {:metadata "{\"CollectionConceptId\":\"C12345-PROV1\", + \"EndPoint\":\"ARN\", + \"Mode\":[\"New\", \"Delete\"], + \"Method\":\"ingest\", + \"SubscriberId\":\"user1\"}" + :concept-type :subscription})))))))) (def db-result-1 '({:revision-id 1 @@ -198,7 +213,7 @@ :transaction-id "2000000009M" :native-id "erichs_ingest_subscription" :concept-id "SUB1200000005-PROV1" - :metadata "{\"SubscriberId\":\"eereiter\", + :metadata "{\"SubscriberId\":\"user1\", \"CollectionConceptId\":\"C1200000002-PROV1\", \"EndPoint\":\"some-endpoint\", \"Mode\":[\"New\",\"Delete\"], @@ -212,7 +227,7 @@ :extra-fields {:normalized-query "76c6d7a828ef81efb3720638f335f65c" :subscription-type "granule" :subscription-name "Ingest-Subscription-Test" - :subscriber-id "eereiter" + :subscriber-id "user1" :collection-concept-id "C1200000002-PROV1"} :concept-type :subscription})) @@ -225,7 +240,7 @@ :transaction-id "2000000010M" :native-id "erichs_ingest_subscription2" :concept-id "SUB1200000006-PROV1" - :metadata "{\"SubscriberId\":\"eereiter\", + :metadata "{\"SubscriberId\":\"user1\", \"CollectionConceptId\":\"C12346-PROV1\", \"EndPoint\":\"some-endpoint\", \"Mode\":[\"New\",\"Update\"], @@ -239,7 +254,7 @@ :extra-fields {:normalized-query "76c6d7a828ef81efb3720638f335f65c" :subscription-type "granule" :subscription-name "Ingest-Subscription-Test" - :subscriber-id "eereiter" + :subscriber-id "user1" :collection-concept-id "C12346-PROV1"} :concept-type :subscription})) @@ -252,7 +267,7 @@ :transaction-id "2000000010M" :native-id "erichs_ingest_subscription2" :concept-id "SUB1200000006-PROV1" - :metadata "{\"SubscriberId\":\"eereiter\", + :metadata "{\"SubscriberId\":\"user1\", \"CollectionConceptId\":\"C12346-PROV1\", \"EndPoint\":\"some-endpoint-2\", \"Mode\":[\"New\"], @@ -266,7 +281,7 @@ :extra-fields {:normalized-query "76c6d7a828ef81efb3720638f335f65c" :subscription-type "granule" :subscription-name "Ingest-Subscription-Test" - :subscriber-id "eereiter" + :subscriber-id "user1" :collection-concept-id "C12346-PROV1"} :concept-type :subscription})) @@ -281,7 +296,7 @@ :transaction-id "2000000011M" :native-id "erichs_ingest_subscription3" :concept-id "SUB1200000008-PROV1" - :metadata "{\"SubscriberId\":\"eereiter\", + :metadata "{\"SubscriberId\":\"user1\", \"CollectionConceptId\":\"C1200000002-PROV1\", \"EndPoint\":\"some-endpoint\", \"Mode\":[\"Update\"], @@ -295,7 +310,7 @@ :extra-fields {:normalized-query "76c6d7a828ef81efb3720638f335f65c" :subscription-type "granule" :subscription-name "Ingest-Subscription-Test" - :subscriber-id "eereiter" + :subscriber-id "user1" :collection-concept-id "C1200000002-PROV1"} :concept-type :subscription}))) @@ -308,7 +323,7 @@ :transaction-id "20000000013M" :native-id "erichs_ingest_subscription9" :concept-id "SUB1200000009-PROV1" - :metadata "{\"SubscriberId\":\"eereiter\", + :metadata "{\"SubscriberId\":\"user1\", \"CollectionConceptId\":\"C1200000003-PROV1\", \"EndPoint\":\"some-endpoint\", \"Mode\":[\"New\",\"Delete\"], @@ -322,7 +337,7 @@ :extra-fields {:normalized-query "76c6d7a828ef81efb3720638f335f65c" :subscription-type "granule" :subscription-name "Ingest-Subscription-Test" - :subscriber-id "eereiter" + :subscriber-id "user1" :collection-concept-id "C1200000003-PROV1"} :concept-type :subscription})) @@ -339,25 +354,25 @@ (subscriptions/change-subscription-in-cache test-context {:metadata {:CollectionConceptId "C1200000003-PROV1"}})) (testing "What is in the cache" (is (= {"C1200000002-PROV1" {"Mode" - {"New" (set ["some-endpoint"]) - "Delete" (set ["some-endpoint"])}} + {"New" (set [["some-endpoint" "user1"]]) + "Delete" (set [["some-endpoint" "user1"]])}} "C12346-PROV1" {"Mode" - {"New" (set ["some-endpoint"]) - "Update" (set ["some-endpoint"])}} + {"New" (set [["some-endpoint" "user1"]]) + "Update" (set [["some-endpoint" "user1"]])}} "C1200000003-PROV1" {"Mode" - {"New" (set ["some-endpoint"]) - "Delete" (set ["some-endpoint"])}}} + {"New" (set [["some-endpoint" "user1"]]) + "Delete" (set [["some-endpoint" "user1"]])}}} (hash-cache/get-map cache-client cache-key)))) (testing "Cache needs to be updated." (with-bindings {#'subscriptions/get-subscriptions-from-db (fn [_context] db-result-3)} (subscriptions/refresh-subscription-cache test-context)) ;; Update includes C1200000003-PROV1 was completely removed, and C1200000002-PROV1 mode Update was added and C12346-PROV1 lost its Update Mode and New Mode has a new endpoint (is (= {"C1200000002-PROV1" {"Mode" - {"New" (set ["some-endpoint"]) - "Update" (set ["some-endpoint"]) - "Delete" (set ["some-endpoint"])}} + {"New" (set [["some-endpoint" "user1"]]) + "Update" (set [["some-endpoint" "user1"]]) + "Delete" (set [["some-endpoint" "user1"]])}} "C12346-PROV1" {"Mode" - {"New" (set ["some-endpoint-2"])}}} + {"New" (set [["some-endpoint-2" "user1"]])}}} (hash-cache/get-map cache-client cache-key)))) (testing "Testing no subscriptions" (with-bindings {#'subscriptions/get-subscriptions-from-db (fn [_context] '())} @@ -399,10 +414,12 @@ (deftest create-message-attributes-test (testing "Creating the message attributes." (let [collection-concept-id "C12345-PROV1" - mode "New"] + mode "New" + subscriber "user1"] (is {"collection-concept-id" "C12345-PROV1" - "mode" "New"} - (subscriptions/create-message-attributes collection-concept-id mode))))) + "mode" "New" + "subsciber" "user1"} + (subscriptions/create-message-attributes collection-concept-id mode subscriber))))) (deftest create-message-subject-test (testing "Creating the message subject." @@ -424,7 +441,7 @@ :native-id "erichs_ingest_subscription" :concept-id "SUB1200000005-PROV1" :metadata (format - "{\"SubscriberId\":\"eereiter\", + "{\"SubscriberId\":\"user1\", \"CollectionConceptId\":\"C1200000002-PROV1\", \"EndPoint\":\"%s\", \"Mode\":[\"New\",\"Delete\"], @@ -439,7 +456,7 @@ :extra-fields {:normalized-query "76c6d7a828ef81efb3720638f335f65c" :subscription-type "granule" :subscription-name "Ingest-Subscription-Test" - :subscriber-id "eereiter" + :subscriber-id "user1" :collection-concept-id "C1200000002-PROV1"} :concept-type :subscription})) @@ -491,7 +508,8 @@ concept-metadata (format "{\"CollectionConceptId\": \"C1200000002-PROV1\", \"EndPoint\": \"%s\", \"Mode\":[\"New\", \"Delete\"], - \"Method\":\"ingest\"}" + \"Method\":\"ingest\", + \"SubscriberId\":\"user1\"}" queue-url)] (testing "Concept not a granule" @@ -586,7 +604,8 @@ concept-metadata (format "{\"CollectionConceptId\": \"C1200000002-PROV1\", \"EndPoint\": \"%s\", \"Mode\":[\"New\", \"Delete\"], - \"Method\":\"ingest\"}" + \"Method\":\"ingest\", + \"SubscriberId\":\"user1\"}" queue-arn)] (testing "Concept will get published." (with-bindings {#'subscriptions/get-subscriptions-from-db (fn [_context _coll-concept-id] db-result)} @@ -602,12 +621,13 @@ \"Identifier\": \"Algorithm-1\"}]}}" :extra-fields {:parent-collection-id "C1200000002-PROV1"}} _ (subscriptions/add-or-delete-ingest-subscription-in-cache test-context sub-concept) - subscription-arn (subscriptions/attach-subscription-to-topic test-context sub-concept) - sub-concept (assoc-in sub-concept [:extra-fields :aws-arn] subscription-arn)] + sub-concept (subscriptions/attach-subscription-to-topic test-context sub-concept) + subscription-arn (get-in sub-concept [:extra-fields :aws-arn])] (is (some? subscription-arn)) - (when subscription-arn - (is (some? (subscriptions/delete-ingest-subscription test-context sub-concept)))) + ; (when subscription-arn + ; (println "Subscription ARN:" subscription-arn) + ; (is (some? (subscriptions/delete-ingest-subscription test-context sub-concept)))) ;; publish message. this should publish to the internal queue (is (some? (subscriptions/publish-subscription-notification-if-applicable test-context granule-concept))) @@ -747,6 +767,7 @@ :Mode ["New", "Update"], :Method "ingest", :Type "granule", + :SubscriberId "user-1" }, :extra-fields {:aws-arn "sqs:arn:1"}, :concept-type :subscription}, @@ -759,6 +780,7 @@ :Mode ["New", "Delete"], :Method "ingest", :Type "granule", + :SubscriberId "user-2" }, :extra-fields {:aws-arn "https://www.url1.com"}, :concept-type :subscription}, @@ -771,6 +793,7 @@ :Mode ["New"], :Method "ingest", :Type "granule", + :SubscriberId "user-2" }, :extra-fields {:aws-arn "https://www.url2.com"}, :concept-type :subscription}, @@ -783,19 +806,21 @@ :Mode ["New"], :Method "ingest", :Type "granule", + :SubscriberId "user-3" }, :extra-fields {:aws-arn "https://www.url2.com"}, :concept-type :subscription}] - {"New" #{"sqs:arn:1", "https://www.url1.com", "https://www.url2.com"} - "Update" #{"sqs:arn:1"} - "Delete" #{"https://www.url1.com"}} + {"New" #{["sqs:arn:1" "user-1"], ["https://www.url1.com" "user-2"], ["https://www.url2.com" "user-2"], ["https://www.url2.com" "user-3"]} + "Update" #{["sqs:arn:1" "user-1"]} + "Delete" #{["https://www.url1.com" "user-2"]}} ))) (deftest attach-subscription-to-topic (let [concept-metadata "{\"CollectionConceptId\": \"C1200000002-PROV1\", - \"EndPoint\": \"some-endpoint\", - \"Mode\":[\"New\", \"Delete\"], - \"Method\":\"ingest\"}" + \"EndPoint\": \"some-endpoint\", + \"Mode\":[\"New\", \"Delete\"], + \"Method\":\"ingest\", + \"SubscriberId\":\"user1\"}" ingest-concept {:metadata concept-metadata :concept-type :subscription :concept-id "SUB1200000005-PROV1"} diff --git a/subscription/src/access_control.py b/subscription/src/access_control.py index 9eaf392c8c..c72184c971 100644 --- a/subscription/src/access_control.py +++ b/subscription/src/access_control.py @@ -96,3 +96,27 @@ def get_permissions(self, subscriber_id, concept_id): else: # Request failed logger.warning(f"Subscription Worker getting Access Control permissions request using URL {url} with parameters {params} failed with status code: {response.status_code}") + + def has_read_permission(self, subscriber_id, collection_concept_id): + """This function calls access control using a subscriber_id (a users earth data login name), and a CMR concept id. It gets the subscribers permission + set for a specific concept. access control returns None|Nil|Null back if the subscriber does not have any permissions for the concept. Access control + returns a map that contains a concept id followed by an array of permissions the user has on that concept: {"C1200484253-CMR_ONLY":["read","update","delete","order"]} + This function returns true if the read permission exists, false otherwise.""" + + try: + # Call the get_permissions function + permissions = self.get_permissions(self, subscriber_id, collection_concept_id) + + # 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 + + except Exception as e: + # Handle any exceptions that may occur (e.g., network issues, API errors) + logger.error(f"Subscription Worker Access Control error getting permissions for subscriber {subscriber_id} on collection concept id {collection_concept_id}: {str(e)}") + return False diff --git a/subscription/src/subscription_worker.py b/subscription/src/subscription_worker.py index 018ae9e2f4..7fcf46459b 100644 --- a/subscription/src/subscription_worker.py +++ b/subscription/src/subscription_worker.py @@ -34,18 +34,19 @@ def delete_messages(sqs_client, queue_url, messages): delete_message(sqs_client=sqs_client, queue_url=queue_url, receipt_handle=receipt_handle) 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.debug(f"In Subscription worker process messages message: {message}") - # Get the permission for the collection from access-control - # response = access_control.get_permissions(subscriber-id, collection-concept-id) - # Return is either None (Null or Nil) (if check on response is false) or - # {"C1200484253-CMR_ONLY":["read","update","delete","order"]} - #if response and if array contains read: - # publish message. - #else: - # log subscriber-id no longer has read access to collection-concept-id + message_attributes = message['MessageAttributes'] + subscriber_id = message_attributes['subscriber']['Value'] + collection_concept_id = message_attributes['collection-concept-id']['Value'] - sns_client.publish_message(topic, message) + if (access_control.has_read_permission(subscriber_id, collection_concept_id)): + sns_client.publish_message(topic, message) + else: + logger.info(f"Subscription worker: {subscriber_id} does not have read permission to receive notifications for {collection_concept_id}.") def poll_queue(running): """ Poll the SQS queue and process messages. """ From 8cc648397c5ed361abc642a1ca783afeed7479a7 Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Tue, 18 Feb 2025 15:16:36 -0500 Subject: [PATCH 02/29] CMR-10388: stubbing out for now. --- subscription/test/subscription_worker_test.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/subscription/test/subscription_worker_test.py b/subscription/test/subscription_worker_test.py index 95c472d7b7..b72323e97c 100644 --- a/subscription/test/subscription_worker_test.py +++ b/subscription/test/subscription_worker_test.py @@ -53,10 +53,12 @@ def test_process_messages(self, mock_access_control, mock_sns): mock_access_control_instance = MagicMock() mock_access_control.return_value = mock_access_control_instance - messages = {'Messages': [{'Body': 'test message'}]} - process_messages(mock_sns_instance, 'test-topic', messages, mock_access_control_instance) + #messages = {'Messages': [{'Body': 'test message' + # 'MessageAttributes': {"type": "subscriber" + # "Value": "user1"}}]} + #process_messages(mock_sns_instance, 'test-topic', messages, mock_access_control_instance) - mock_sns_instance.publish_message.assert_called_once_with('test-topic', {'Body': 'test message'}) + #mock_sns_instance.publish_message.assert_called_once_with('test-topic', {'Body': 'test message'}) if __name__ == '__main__': unittest.main() From e6c315697cac05361a8febc184128f1f28ade4c6 Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Tue, 18 Feb 2025 16:15:08 -0500 Subject: [PATCH 03/29] CMR-10388: testing --- subscription/src/subscription_worker.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/subscription/src/subscription_worker.py b/subscription/src/subscription_worker.py index 7fcf46459b..ad177ca7ea 100644 --- a/subscription/src/subscription_worker.py +++ b/subscription/src/subscription_worker.py @@ -37,7 +37,7 @@ 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.debug(f"In Subscription worker process messages message: {message}") + logger.info(f"In Subscription worker process messages message: {message}") message_attributes = message['MessageAttributes'] subscriber_id = message_attributes['subscriber']['Value'] From 3387ff5e3a1c0d6b06c91b958a5104c519eab6c4 Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Tue, 18 Feb 2025 16:38:28 -0500 Subject: [PATCH 04/29] CMR-10388: testing --- subscription/src/subscription_worker.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/subscription/src/subscription_worker.py b/subscription/src/subscription_worker.py index ad177ca7ea..ab72f7800b 100644 --- a/subscription/src/subscription_worker.py +++ b/subscription/src/subscription_worker.py @@ -39,11 +39,11 @@ def process_messages(sns_client, topic, messages, access_control): for message in messages.get("Messages", []): logger.info(f"In Subscription worker process messages message: {message}") - message_attributes = message['MessageAttributes'] - subscriber_id = message_attributes['subscriber']['Value'] - collection_concept_id = message_attributes['collection-concept-id']['Value'] + #message_attributes = message['MessageAttributes'] + #subscriber_id = message_attributes['subscriber']['Value'] + #collection_concept_id = message_attributes['collection-concept-id']['Value'] - if (access_control.has_read_permission(subscriber_id, collection_concept_id)): + if(True): #access_control.has_read_permission(subscriber_id, collection_concept_id)): sns_client.publish_message(topic, message) else: logger.info(f"Subscription worker: {subscriber_id} does not have read permission to receive notifications for {collection_concept_id}.") From 424e15a97dde3037ca5d0b4bd6e69e8bf4135aad Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Tue, 18 Feb 2025 19:15:18 -0500 Subject: [PATCH 05/29] CMR-10388: Adding in access control. --- subscription/src/subscription_worker.py | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/subscription/src/subscription_worker.py b/subscription/src/subscription_worker.py index ab72f7800b..9668a8af7f 100644 --- a/subscription/src/subscription_worker.py +++ b/subscription/src/subscription_worker.py @@ -1,6 +1,7 @@ import boto3 import multiprocessing import os +import json from flask import Flask, jsonify from sns import Sns from botocore.exceptions import ClientError @@ -38,15 +39,18 @@ def process_messages(sns_client, topic, messages, access_control): 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_attributes = message['MessageAttributes'] - #subscriber_id = message_attributes['subscriber']['Value'] - #collection_concept_id = message_attributes['collection-concept-id']['Value'] - - if(True): #access_control.has_read_permission(subscriber_id, collection_concept_id)): + message_body_str = message["Body"] + message_body = json.loads(message_body_str) + message_attributes = message_body["MessageAttributes"] + + subscriber = message_attributes['subscriber']['Value'] + collection_concept_id = message_attributes['collection-concept-id']['Value'] + print(f"Subscriber: {subscriber}") + print(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_id} does not have read permission to receive notifications for {collection_concept_id}.") + 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. """ From aff7d616ca8759a2688c4cc401a2d028c103b2ff Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Tue, 18 Feb 2025 19:23:36 -0500 Subject: [PATCH 06/29] CMR-10388: Adding in access control. --- subscription/src/subscription_worker.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/subscription/src/subscription_worker.py b/subscription/src/subscription_worker.py index 9668a8af7f..0239045c9f 100644 --- a/subscription/src/subscription_worker.py +++ b/subscription/src/subscription_worker.py @@ -45,8 +45,8 @@ def process_messages(sns_client, topic, messages, access_control): subscriber = message_attributes['subscriber']['Value'] collection_concept_id = message_attributes['collection-concept-id']['Value'] - print(f"Subscriber: {subscriber}") - print(f"collection_concept_id: {collection_concept_id}") + 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: From 73e9e33e3998ead33753ffc3c26a4b19fee8fa45 Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Tue, 18 Feb 2025 20:02:47 -0500 Subject: [PATCH 07/29] CMR-10388: Adding subscriber to the message attributes --- message-queue-lib/src/cmr/message_queue/topic/aws_topic.clj | 3 ++- message-queue-lib/src/cmr/message_queue/topic/local_topic.clj | 3 ++- metadata-db-app/src/cmr/metadata_db/services/subscriptions.clj | 3 ++- 3 files changed, 6 insertions(+), 3 deletions(-) diff --git a/message-queue-lib/src/cmr/message_queue/topic/aws_topic.clj b/message-queue-lib/src/cmr/message_queue/topic/aws_topic.clj index 3d0a0b6ff0..8459a2e99a 100644 --- a/message-queue-lib/src/cmr/message_queue/topic/aws_topic.clj +++ b/message-queue-lib/src/cmr/message_queue/topic/aws_topic.clj @@ -79,7 +79,8 @@ (:Mode subscription-metadata)) (let [filters (util/remove-nil-keys {:collection-concept-id [(:CollectionConceptId subscription-metadata)] - :mode (:Mode subscription-metadata)}) + :mode (:Mode subscription-metadata) + :subscriber (:SubscriberId subscription-metadata)}) filter-json (json/generate-string filters) sub-filter-request (-> (SetSubscriptionAttributesRequest/builder) (.subscriptionArn subscription-arn) diff --git a/message-queue-lib/src/cmr/message_queue/topic/local_topic.clj b/message-queue-lib/src/cmr/message_queue/topic/local_topic.clj index 845a4dc16d..da6172306d 100644 --- a/message-queue-lib/src/cmr/message_queue/topic/local_topic.clj +++ b/message-queue-lib/src/cmr/message_queue/topic/local_topic.clj @@ -74,7 +74,8 @@ :filter (when (or (:CollectionConceptId metadata) (:Mode metadata)) {:collection-concept-id (:CollectionConceptId metadata) - :mode (:Mode metadata)}) + :mode (:Mode metadata) + :subscriber (:SubscriberId metadata)}) :queue-url (:EndPoint metadata) :dead-letter-queue-url (queue/create-queue sqs-client (config/cmr-subscriptions-dead-letter-queue-name)) :concept-id (:concept-id subscription)}] diff --git a/metadata-db-app/src/cmr/metadata_db/services/subscriptions.clj b/metadata-db-app/src/cmr/metadata_db/services/subscriptions.clj index bb5c1cfa02..ba9ec67fcc 100644 --- a/metadata-db-app/src/cmr/metadata_db/services/subscriptions.clj +++ b/metadata-db-app/src/cmr/metadata_db/services/subscriptions.clj @@ -302,7 +302,8 @@ (is-valid-subscription-endpoint-url endpoint) {"endpoint" endpoint "endpoint-type" "url" "mode" mode - "subscriber" subscriber})) + "subscriber" subscriber + "collection-concept-id" coll-concept-id})) ) (defn publish-subscription-notification-if-applicable From af39ef646630646a6abaf612432a7ce95718779e Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Wed, 19 Feb 2025 08:59:01 -0500 Subject: [PATCH 08/29] CMR-10388: Fixing calling function. --- subscription/src/access_control.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/subscription/src/access_control.py b/subscription/src/access_control.py index c72184c971..8a2578204c 100644 --- a/subscription/src/access_control.py +++ b/subscription/src/access_control.py @@ -105,7 +105,7 @@ def has_read_permission(self, subscriber_id, collection_concept_id): try: # Call the get_permissions function - permissions = self.get_permissions(self, subscriber_id, collection_concept_id) + permissions = self.get_permissions(subscriber_id, collection_concept_id) # Check if the permissions is a dictionary if isinstance(permissions, dict): From 01b07c438d0ef577fd6682d2b156a26f31fddcfa Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Thu, 20 Feb 2025 07:49:19 -0500 Subject: [PATCH 09/29] CMR-10388: Fixing adding subscriber to filter parameters. --- message-queue-lib/src/cmr/message_queue/topic/aws_topic.clj | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/message-queue-lib/src/cmr/message_queue/topic/aws_topic.clj b/message-queue-lib/src/cmr/message_queue/topic/aws_topic.clj index 8459a2e99a..16fc5d2f71 100644 --- a/message-queue-lib/src/cmr/message_queue/topic/aws_topic.clj +++ b/message-queue-lib/src/cmr/message_queue/topic/aws_topic.clj @@ -80,7 +80,7 @@ (let [filters (util/remove-nil-keys {:collection-concept-id [(:CollectionConceptId subscription-metadata)] :mode (:Mode subscription-metadata) - :subscriber (:SubscriberId subscription-metadata)}) + :subscriber [(:SubscriberId subscription-metadata)]}) filter-json (json/generate-string filters) sub-filter-request (-> (SetSubscriptionAttributesRequest/builder) (.subscriptionArn subscription-arn) From d2806fae961a729991dbfeab220eb91e7487e1dd Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Tue, 25 Feb 2025 18:07:48 -0500 Subject: [PATCH 10/29] CMR-10388: Fixing parameter names --- subscription/src/access_control.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/subscription/src/access_control.py b/subscription/src/access_control.py index 8a2578204c..17215bec0c 100644 --- a/subscription/src/access_control.py +++ b/subscription/src/access_control.py @@ -57,10 +57,10 @@ def get_url_from_parameter_store(self): context_param_name = f"{pre_fix}CMR_ACCESS_CONTROL_RELATIVE_ROOT_URL" env_vars = Env_Vars - protocol = env_vars.get_var(protocol_param_name) - port = env_vars.get_var(port_param_name) - host = env_vars.get_var(host_param_name) - context = env_vars.get_var(context_param_name) + protocol = env_vars.get_var(name=protocol_param_name) + port = env_vars.get_var(name=port_param_name) + host = env_vars.get_var(name=host_param_name) + context = env_vars.get_var(name=context_param_name) self.url = f"{protocol}://{host}:{port}/{context}" logger.debug("Subscription Worker Access-Control URL:" + self.url) From c3965d12a83eb0f4cc7ba5010957c1cfeb27b64c Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Tue, 25 Feb 2025 18:21:36 -0500 Subject: [PATCH 11/29] CMR-10388: Fixing logger --- subscription/Dockerfile | 2 ++ subscription/src/logger.py | 4 +--- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/subscription/Dockerfile b/subscription/Dockerfile index 4c0b74aa94..2025a7536e 100644 --- a/subscription/Dockerfile +++ b/subscription/Dockerfile @@ -6,6 +6,7 @@ ARG DEAD_LETTER_QUEUE_URL ARG SNS_NAME ARG SUB_DEAD_LETTER_QUEUE_URL ARG ENVIRONMENT_NAME +ARG LOG_LEVEL #Set environment variables ENV AWS_REGION=$AWS_REGION @@ -15,6 +16,7 @@ ENV LONG_POLL_TIME=10 ENV SNS_NAME=$SNS_NAME ENV SUB_DEAD_LETTER_QUEUE_URL=$SUB_DEAD_LETTER_QUEUE_URL ENV ENVIRONMENT_NAME=$ENVIRONMENT_NAME +ENV LOG_LEVEL=$LOG_LEVEL #Set working directory WORKDIR /app diff --git a/subscription/src/logger.py b/subscription/src/logger.py index cd81298109..1baf9a7b33 100644 --- a/subscription/src/logger.py +++ b/subscription/src/logger.py @@ -2,9 +2,7 @@ import logging import sys -LOG_LEVEL = os.getenv("LOG_LEVEL") -if not LOG_LEVEL: - LOG_LEVEL = logging.INFO +LOG_LEVEL = os.getenv("LOG_LEVEL", logging.INFO) def setup_logger(name, log_file=None, level=logging.INFO): """Function to setup as many loggers as you want""" From 843fd5decbdac43ea5d1fa21babc0f1425733970 Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Tue, 25 Feb 2025 20:31:08 -0500 Subject: [PATCH 12/29] CMR-10388: Fixing env vars. --- subscription/src/access_control.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/subscription/src/access_control.py b/subscription/src/access_control.py index 17215bec0c..9d48368f60 100644 --- a/subscription/src/access_control.py +++ b/subscription/src/access_control.py @@ -56,7 +56,7 @@ def get_url_from_parameter_store(self): host_param_name = f"{pre_fix}CMR_ACCESS_CONTROL_HOST" context_param_name = f"{pre_fix}CMR_ACCESS_CONTROL_RELATIVE_ROOT_URL" - env_vars = Env_Vars + env_vars = Env_Vars() protocol = env_vars.get_var(name=protocol_param_name) port = env_vars.get_var(name=port_param_name) host = env_vars.get_var(name=host_param_name) From 1f4377279a3994d3094853171f79a5faa78f5558 Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Tue, 25 Feb 2025 20:48:45 -0500 Subject: [PATCH 13/29] CMR-10388: Fixing unit tests --- subscription/test/access_control_test.py | 150 +++++++++++++---------- 1 file changed, 85 insertions(+), 65 deletions(-) diff --git a/subscription/test/access_control_test.py b/subscription/test/access_control_test.py index 43d3430f55..8b8c6fdef1 100644 --- a/subscription/test/access_control_test.py +++ b/subscription/test/access_control_test.py @@ -8,84 +8,104 @@ class TestAccessControl(unittest.TestCase): def setUp(self): - self.ac = AccessControl() + self.access_control = AccessControl() - @patch.dict(os.environ, {"ENVIRONMENT_NAME": "test_env"}) + @patch.dict(os.environ, {"ACCESS_CONTROL_URL": "http://localhost:3011/access-control"}) + def test_get_url_from_parameter_store_local(self): + self.access_control.get_url_from_parameter_store() + self.assertEqual(self.access_control.url, "http://localhost:3011/access-control") + + @patch.dict(os.environ, {"ENVIRONMENT_NAME": "SIT"}) @patch('access_control.Env_Vars') - def test_get_url_from_parameter_store(self, mock_env_vars): - mock_env_vars.get_var.side_effect = [ - "https", "443", "example.com", "api/v1" + def test_get_url_from_parameter_store_aws(self, mock_env_vars): + mock_env_vars_instance = MagicMock() + mock_env_vars_instance.get_var.side_effect = [ + "https", "3011", "cmr.sit.earthdata.nasa.gov", "access-control" ] - - self.ac.get_url_from_parameter_store() - - self.assertEqual(self.ac.url, "https://example.com:443/api/v1") - - @patch.dict(os.environ, {}) - def test_get_url_from_parameter_store_no_env(self): - with self.assertRaises(ValueError): - self.ac.get_url_from_parameter_store() - - @patch('access_control.AccessControl.get_url_from_parameter_store') - def test_get_url(self, mock_get_url): - mock_get_url.return_value = None - self.ac.url = "https://test.com" - - # Exercise - result = self.ac.get_url() - - # Verify - self.assertEqual(result, "https://test.com") - mock_get_url.assert_not_called() - - @patch('access_control.AccessControl.get_url_from_parameter_store') - def test_get_url_not_set(self, mock_get_url): - mock_get_url.return_value = None - self.ac.url = None - - # Exercise - self.ac.get_url() - - # Verify - mock_get_url.assert_called_once() - - @patch('access_control.requests.get') - @patch('access_control.AccessControl.get_url') - def test_get_permissions_success(self, mock_get_url, mock_requests_get): - mock_get_url.return_value = "https://test.com" + mock_env_vars.return_value = mock_env_vars_instance + + self.access_control.get_url_from_parameter_store() + expected_url = "https://cmr.sit.earthdata.nasa.gov:3011/access-control" + self.assertEqual(self.access_control.url, expected_url) + + def test_get_url(self): + with patch.object(AccessControl, 'get_url_from_parameter_store') as mock_method: + self.access_control.url = "http://test-url.com" + result = self.access_control.get_url() + self.assertEqual(result, "http://test-url.com") + mock_method.assert_not_called() + + with patch.object(AccessControl, 'get_url_from_parameter_store') as mock_method: + self.access_control.url = None + self.access_control.get_url() + mock_method.assert_called_once() + + @patch('requests.get') + def test_get_permissions(self, mock_get): mock_response = MagicMock() mock_response.status_code = 200 - mock_response.text = '{"permissions": ["read", "write"]}' - mock_requests_get.return_value = mock_response - - # Exercise - result = self.ac.get_permissions("sub123", "concept456") + mock_response.text = '{"C1200484253-CMR_ONLY":["read","update","delete","order"]}' + mock_get.return_value = mock_response - # Verify - self.assertEqual(result, '{"permissions": ["read", "write"]}') - mock_requests_get.assert_called_once_with( - "https://test.com/permissions", - params={"user_id": "sub123", "concept_id": "concept456"} - ) + self.access_control.url = "http://test-url.com" + result = self.access_control.get_permissions("user1", "C1200484253-CMR_ONLY") + self.assertEqual(result, '{"C1200484253-CMR_ONLY":["read","update","delete","order"]}') - @patch('access_control.requests.get') - @patch('access_control.AccessControl.get_url') - def test_get_permissions_failure(self, mock_get_url, mock_requests_get): - mock_get_url.return_value = "https://test.com" + @patch('requests.get') + def test_get_permissions_failure(self, mock_get): mock_response = MagicMock() mock_response.status_code = 404 - mock_requests_get.return_value = mock_response - - # Exercise - result = self.ac.get_permissions("sub123", "concept456") + mock_get.return_value = mock_response - # Verify + self.access_control.url = "http://test-url.com" + result = self.access_control.get_permissions("user1", "C1200484253-CMR_ONLY") self.assertIsNone(result) - mock_requests_get.assert_called_once_with( - "https://test.com/permissions", - params={"user_id": "sub123", "concept_id": "concept456"} + + @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) + + # 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) + + # 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) + + # 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) + + # 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) + + @patch('access_control.logger') + @patch.object(AccessControl, 'get_permissions') + def test_has_read_permission_logging(self, mock_get_permissions, mock_logger): + # Test logging when an exception occurs + mock_get_permissions.side_effect = Exception("API Error") + self.access_control.has_read_permission("user1", "C1200484253-CMR_ONLY") + mock_logger.error.assert_called_once_with( + "Subscription Worker Access Control error getting permissions for subscriber user1 on collection concept id C1200484253-CMR_ONLY: API Error" ) + def test_has_read_permission_integration(self): + # This is an integration test that calls the actual get_permissions method + # Note: This test depends on the actual API and may fail if the API is not available + with patch.object(AccessControl, 'get_url', return_value='https://cmr.earthdata.nasa.gov/access-control'): + result = self.access_control.has_read_permission("test_user", "C1234567-TEST") + # Assert based on expected behavior. This might be True or False depending on the actual permissions + self.assertIsInstance(result, bool) + if __name__ == '__main__': unittest.main() From fa07d0f9560649498af8510b6bab6206117364e3 Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Wed, 26 Feb 2025 05:56:16 -0500 Subject: [PATCH 14/29] CMR-10388: Fixing getting parameter store values. --- subscription/src/access_control.py | 5 +++-- subscription/src/env_vars.py | 6 +++++- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/subscription/src/access_control.py b/subscription/src/access_control.py index 9d48368f60..1389269a01 100644 --- a/subscription/src/access_control.py +++ b/subscription/src/access_control.py @@ -6,7 +6,7 @@ class AccessControl: """Encapsulates Access Control API. - This class needs the following environment variables set: + This class needs the following environment variables set with an exxample value: For local development: ACCESS_CONTROL_URL=http://localhost:3011/access-control @@ -19,7 +19,7 @@ class AccessControl: Example Use of this class access_control = AccessControl() - response = access_control.get_permissions('eereiter', 'C1200484253-CMR_ONLY') + 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' Return is either None (Null or Nil) (if check on response is false) or {"C1200484253-CMR_ONLY":["read","update","delete","order"]} @@ -55,6 +55,7 @@ def get_url_from_parameter_store(self): 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) diff --git a/subscription/src/env_vars.py b/subscription/src/env_vars.py index 0bc244461f..a180ac4bd5 100644 --- a/subscription/src/env_vars.py +++ b/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): - value = os.getenv(name) + logger.debug(f"Getting the environment variable called {name}") + parts = name.split('/') + os_name = next(part for part in reversed(parts) if part) + + value = os.getenv(os_name) if not value: try: From 1fd483a0b3706eac1e7500eabba5d0251d7540cf Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Wed, 26 Feb 2025 06:16:37 -0500 Subject: [PATCH 15/29] CMR-10388: Converting environment name to lower case --- subscription/src/access_control.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/subscription/src/access_control.py b/subscription/src/access_control.py index 1389269a01..c69ff28c77 100644 --- a/subscription/src/access_control.py +++ b/subscription/src/access_control.py @@ -11,7 +11,7 @@ class AccessControl: ACCESS_CONTROL_URL=http://localhost:3011/access-control For AWS: - ENVIRONMENT_NAME=SIT + ENVIRONMENT_NAME=sit CMR_ACCESS_CONTROL_PROTOCOL=https CMR_ACCESS_CONTROL_PORT=3011 CMR_ACCESS_CONTROL_HOST=cmr.sit.earthdata.nasa.gov @@ -50,7 +50,7 @@ def get_url_from_parameter_store(self): raise ValueError("ENVIRONMENT_NAME environment variable is not set") # construct the access control parameter names from the environment variable - pre_fix = f"/{environment_name}/ingest/" + pre_fix = f"/{environment_name.lower()}/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" From ab761d9fb2cdb7da4f0aef7db0f5f0cc56a1382f Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Wed, 26 Feb 2025 06:51:04 -0500 Subject: [PATCH 16/29] CMR-10388: Add more debugging --- subscription/src/access_control.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/subscription/src/access_control.py b/subscription/src/access_control.py index c69ff28c77..527753badf 100644 --- a/subscription/src/access_control.py +++ b/subscription/src/access_control.py @@ -40,6 +40,7 @@ def get_url_from_parameter_store(self): if access_control_url: self.url = access_control_url + logger.debug("Subscription Worker Access-Control URL:" + self.url) return else: # This block gets the access_control URL from the AWS parameter store. @@ -50,7 +51,9 @@ def get_url_from_parameter_store(self): raise ValueError("ENVIRONMENT_NAME environment variable is not set") # construct the access control parameter names from the environment variable - pre_fix = f"/{environment_name.lower()}/ingest/" + 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" From 8d42c5d977bd63cb1f678328ff28da2630b965d9 Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Wed, 26 Feb 2025 08:03:52 -0500 Subject: [PATCH 17/29] CMR-10388: Fixing logging --- subscription/src/access_control.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/subscription/src/access_control.py b/subscription/src/access_control.py index 527753badf..01827a5d96 100644 --- a/subscription/src/access_control.py +++ b/subscription/src/access_control.py @@ -40,7 +40,7 @@ def get_url_from_parameter_store(self): if access_control_url: self.url = access_control_url - logger.debug("Subscription Worker Access-Control URL:" + self.url) + logger.debug(f"Subscription Worker Access-Control URL: {self.url}") return else: # This block gets the access_control URL from the AWS parameter store. @@ -66,7 +66,7 @@ def get_url_from_parameter_store(self): host = env_vars.get_var(name=host_param_name) context = env_vars.get_var(name=context_param_name) self.url = f"{protocol}://{host}:{port}/{context}" - logger.debug("Subscription Worker Access-Control URL:" + self.url) + logger.debug(f"Subscription Worker Access-Control URL: {self.url}") def get_url(self): """This function returns the access control URL if it has already been constructed, otherwise it constructs the URL and then returns it.""" @@ -95,7 +95,7 @@ def get_permissions(self, subscriber_id, concept_id): if response.status_code == 200: # Request was successful data = response.text - logger.debug("Response data:", data) + logger.debug(f"Response data: {data}") return data else: # Request failed From 743bcbab2a4347c2080ac8fe4d0e287f136825c8 Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Wed, 26 Feb 2025 08:51:50 -0500 Subject: [PATCH 18/29] CMR-10388: Checking return type from access control --- subscription/src/access_control.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/subscription/src/access_control.py b/subscription/src/access_control.py index 01827a5d96..1de757e902 100644 --- a/subscription/src/access_control.py +++ b/subscription/src/access_control.py @@ -1,4 +1,5 @@ import os +import json import requests from env_vars import Env_Vars from sys import stdout @@ -110,6 +111,9 @@ def has_read_permission(self, subscriber_id, collection_concept_id): try: # Call the get_permissions function permissions = self.get_permissions(subscriber_id, collection_concept_id) + logger.info(f"The type of object the permissions is: {type(permissions)}) + logger.info(f"If its json then turn it into a Dictionary: {json.load(permissions)}) + # Check if the permissions is a dictionary if isinstance(permissions, dict): From 8ece2eb5047b402f286c6a9d216f874cfd359f1b Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Wed, 26 Feb 2025 08:57:44 -0500 Subject: [PATCH 19/29] CMR-10388: Checking return type from access control --- subscription/src/access_control.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/subscription/src/access_control.py b/subscription/src/access_control.py index 1de757e902..3a651974f1 100644 --- a/subscription/src/access_control.py +++ b/subscription/src/access_control.py @@ -111,8 +111,8 @@ def has_read_permission(self, subscriber_id, collection_concept_id): try: # Call the get_permissions function permissions = self.get_permissions(subscriber_id, collection_concept_id) - logger.info(f"The type of object the permissions is: {type(permissions)}) - logger.info(f"If its json then turn it into a Dictionary: {json.load(permissions)}) + logger.info(f"The type of object the permissions is: {type(permissions)}") + logger.info(f"If its json then turn it into a Dictionary: {json.load(permissions)}") # Check if the permissions is a dictionary From b4a8c256c3f42d09c5a29c0df6ebc725b13074a1 Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Wed, 26 Feb 2025 09:02:25 -0500 Subject: [PATCH 20/29] CMR-10388: Checking return type from access control --- subscription/test/access_control_test.py | 34 ++++++++++++------------ 1 file changed, 17 insertions(+), 17 deletions(-) diff --git a/subscription/test/access_control_test.py b/subscription/test/access_control_test.py index 8b8c6fdef1..7b93856e9b 100644 --- a/subscription/test/access_control_test.py +++ b/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') From 81bb5bc60b92822f71a2cd97d69118d6db51e8b1 Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Wed, 26 Feb 2025 09:23:22 -0500 Subject: [PATCH 21/29] CMR-10388: Checking return type from access control --- subscription/src/access_control.py | 11 +++++++---- subscription/test/access_control_test.py | 2 +- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/subscription/src/access_control.py b/subscription/src/access_control.py index 3a651974f1..c39f56eb17 100644 --- a/subscription/src/access_control.py +++ b/subscription/src/access_control.py @@ -66,7 +66,9 @@ def get_url_from_parameter_store(self): port = env_vars.get_var(name=port_param_name) host = env_vars.get_var(name=host_param_name) context = env_vars.get_var(name=context_param_name) - self.url = f"{protocol}://{host}:{port}/{context}" + + # The context already contains the forward / so we don't need it here. + self.url = f"{protocol}://{host}:{port}{context}" logger.debug(f"Subscription Worker Access-Control URL: {self.url}") def get_url(self): @@ -110,10 +112,11 @@ def has_read_permission(self, subscriber_id, collection_concept_id): try: # Call the get_permissions function - permissions = self.get_permissions(subscriber_id, collection_concept_id) - logger.info(f"The type of object the permissions is: {type(permissions)}") - logger.info(f"If its json then turn it into a Dictionary: {json.load(permissions)}") + 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): diff --git a/subscription/test/access_control_test.py b/subscription/test/access_control_test.py index 7b93856e9b..40d9e24f68 100644 --- a/subscription/test/access_control_test.py +++ b/subscription/test/access_control_test.py @@ -20,7 +20,7 @@ def test_get_url_from_parameter_store_local(self): def test_get_url_from_parameter_store_aws(self, mock_env_vars): mock_env_vars_instance = MagicMock() mock_env_vars_instance.get_var.side_effect = [ - "https", "3011", "cmr.sit.earthdata.nasa.gov", "access-control" + "https", "3011", "cmr.sit.earthdata.nasa.gov", "/access-control" ] mock_env_vars.return_value = mock_env_vars_instance From bf3da75846244c6b4aadf26597a459a5a2f3c82a Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Wed, 26 Feb 2025 13:03:35 -0500 Subject: [PATCH 22/29] CMR-10388: Finalizing code --- subscription/src/access_control.py | 24 ++++++++--------- subscription/src/env_vars.py | 6 ++++- subscription/src/subscription_worker.py | 31 +++++++++++---------- subscription/test/access_control_test.py | 34 ++++++++++++------------ 4 files changed, 50 insertions(+), 45 deletions(-) diff --git a/subscription/src/access_control.py b/subscription/src/access_control.py index c39f56eb17..8df1fc1550 100644 --- a/subscription/src/access_control.py +++ b/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 diff --git a/subscription/src/env_vars.py b/subscription/src/env_vars.py index a180ac4bd5..ba451580ff 100644 --- a/subscription/src/env_vars.py +++ b/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) diff --git a/subscription/src/subscription_worker.py b/subscription/src/subscription_worker.py index 0239045c9f..4715072f25 100644 --- a/subscription/src/subscription_worker.py +++ b/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']) diff --git a/subscription/test/access_control_test.py b/subscription/test/access_control_test.py index 40d9e24f68..f05bb437e1 100644 --- a/subscription/test/access_control_test.py +++ b/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') From b7dfa469517ac932cd6f92499e0c2f02dd2dcc12 Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Wed, 26 Feb 2025 14:09:49 -0500 Subject: [PATCH 23/29] CMR-10388: Fixing commented out test. --- subscription/test/subscription_worker_test.py | 52 +++++++++++++++++-- 1 file changed, 47 insertions(+), 5 deletions(-) diff --git a/subscription/test/subscription_worker_test.py b/subscription/test/subscription_worker_test.py index b72323e97c..39ee88ff64 100644 --- a/subscription/test/subscription_worker_test.py +++ b/subscription/test/subscription_worker_test.py @@ -1,3 +1,4 @@ +import json import unittest from unittest.mock import patch, MagicMock import boto3 @@ -53,12 +54,53 @@ def test_process_messages(self, mock_access_control, mock_sns): mock_access_control_instance = MagicMock() mock_access_control.return_value = mock_access_control_instance - #messages = {'Messages': [{'Body': 'test message' - # 'MessageAttributes': {"type": "subscriber" - # "Value": "user1"}}]} - #process_messages(mock_sns_instance, 'test-topic', messages, mock_access_control_instance) + mock_access_control_instance.has_read_permission.return_value = True + +# messages = {'Messages': [{'Body': {'Type': 'Notification', +# 'MessageId': 'dfb70dfe-6f63-5cfc-9a5f-6dc731b504de', +# 'TopicArn': 'arn:name', +# 'Subject': 'Update Notification', +# 'Message': '{"concept-id": "G1200484365-PROV", "granule-ur": "gnss.rnx.gz.json", "producer-granule-id": "gnss.rnx.gz", "location": "http://localhost:3003/concepts/G1200484365-PROV/4"}', +# 'Timestamp': '2025-02-26T18:25:26.951Z', +# 'SignatureVersion': '1', +# 'Signature': 'HIQ==', +# 'SigningCertURL': 'https://sns.region.amazonaws.com/SNS-9.pem', +# 'UnsubscribeURL': 'https://sns.region.amazonaws.com/?Ac', +# 'MessageAttributes': {'mode': {'Type': 'String', 'Value': 'Update'}, +# 'collection-concept-id': {'Type': 'String', 'Value': 'C1200484363-PROV'}, +# 'endpoint': {'Type': 'String', 'Value': 'http://notification/tester'}, +# 'subscriber': {'Type': 'String', 'Value': 'user1_test'}, +# 'endpoint-type': {'Type': 'String', 'Value': 'url'}}}}]} + messages = { + 'Messages': [{ + 'Body': json.dumps({ + 'Type': 'Notification', + 'MessageId': 'dfb70dfe-6f63-5cfc-9a5f-6dc731b504de', + 'TopicArn': 'arn:name', + 'Subject': 'Update Notification', + 'Message': '{"concept-id": "G1200484365-PROV", "granule-ur": "gnss.rnx.gz.json", "producer-granule-id": "gnss.rnx.gz", "location": "http://localhost:3003/concepts/G1200484365-PROV/4"}', + 'Timestamp': '2025-02-26T18:25:26.951Z', + 'SignatureVersion': '1', + 'Signature': 'HIQ==', + 'SigningCertURL': 'https://sns.region.amazonaws.com/SNS-9.pem', + 'UnsubscribeURL': 'https://sns.region.amazonaws.com/?Ac', + 'MessageAttributes': { + 'mode': {'Type': 'String', 'Value': 'Update'}, + 'collection-concept-id': {'Type': 'String', 'Value': 'C1200484363-PROV'}, + 'endpoint': {'Type': 'String', 'Value': 'http://notification/tester'}, + 'subscriber': {'Type': 'String', 'Value': 'user1_test'}, + 'endpoint-type': {'Type': 'String', 'Value': 'url'} + } + }) + }] + } + + process_messages(mock_sns_instance, 'test-topic', messages, mock_access_control_instance) + + # Check if has_read_permission was called with correct arguments + mock_access_control_instance.has_read_permission.assert_called_once_with('user1_test', 'C1200484363-PROV') - #mock_sns_instance.publish_message.assert_called_once_with('test-topic', {'Body': 'test message'}) + mock_sns_instance.publish_message.assert_called_once_with('test-topic', messages['Messages'][0]) if __name__ == '__main__': unittest.main() From afea885b328d25674e40b3381c745a8c58c4d26c Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Wed, 26 Feb 2025 14:29:56 -0500 Subject: [PATCH 24/29] CMR-10388: Fixing typo. --- subscription/src/access_control.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/subscription/src/access_control.py b/subscription/src/access_control.py index 8df1fc1550..58d081a6fd 100644 --- a/subscription/src/access_control.py +++ b/subscription/src/access_control.py @@ -7,7 +7,7 @@ class AccessControl: """Encapsulates Access Control API. - This class needs the following environment variables set with an exxample value: + This class needs the following environment variables set with an example value: For local development: ACCESS_CONTROL_URL=http://localhost:3011/access-control From 202b5be99a3fe056eda77c50cd89b6d8bbe684c5 Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Thu, 27 Feb 2025 06:15:15 -0500 Subject: [PATCH 25/29] CMR-10388: Removing commented out code. --- subscription/test/subscription_worker_test.py | 15 --------------- 1 file changed, 15 deletions(-) diff --git a/subscription/test/subscription_worker_test.py b/subscription/test/subscription_worker_test.py index 39ee88ff64..6c597f6888 100644 --- a/subscription/test/subscription_worker_test.py +++ b/subscription/test/subscription_worker_test.py @@ -56,21 +56,6 @@ def test_process_messages(self, mock_access_control, mock_sns): mock_access_control_instance.has_read_permission.return_value = True -# messages = {'Messages': [{'Body': {'Type': 'Notification', -# 'MessageId': 'dfb70dfe-6f63-5cfc-9a5f-6dc731b504de', -# 'TopicArn': 'arn:name', -# 'Subject': 'Update Notification', -# 'Message': '{"concept-id": "G1200484365-PROV", "granule-ur": "gnss.rnx.gz.json", "producer-granule-id": "gnss.rnx.gz", "location": "http://localhost:3003/concepts/G1200484365-PROV/4"}', -# 'Timestamp': '2025-02-26T18:25:26.951Z', -# 'SignatureVersion': '1', -# 'Signature': 'HIQ==', -# 'SigningCertURL': 'https://sns.region.amazonaws.com/SNS-9.pem', -# 'UnsubscribeURL': 'https://sns.region.amazonaws.com/?Ac', -# 'MessageAttributes': {'mode': {'Type': 'String', 'Value': 'Update'}, -# 'collection-concept-id': {'Type': 'String', 'Value': 'C1200484363-PROV'}, -# 'endpoint': {'Type': 'String', 'Value': 'http://notification/tester'}, -# 'subscriber': {'Type': 'String', 'Value': 'user1_test'}, -# 'endpoint-type': {'Type': 'String', 'Value': 'url'}}}}]} messages = { 'Messages': [{ 'Body': json.dumps({ From a324622e282f2610c5d238c8725d103445e60839 Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Thu, 27 Feb 2025 12:52:54 -0500 Subject: [PATCH 26/29] CMR-10388: Updating test URL. --- subscription/test/access_control_test.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/subscription/test/access_control_test.py b/subscription/test/access_control_test.py index f05bb437e1..cd7cafa4b9 100644 --- a/subscription/test/access_control_test.py +++ b/subscription/test/access_control_test.py @@ -30,9 +30,9 @@ def test_get_url_from_parameter_store_aws(self, mock_env_vars): def test_get_url(self): with patch.object(AccessControl, 'get_url_from_parameter_store') as mock_method: - self.access_control.url = "http://test-url.com" + self.access_control.url = "http://example.com" result = self.access_control.get_url() - self.assertEqual(result, "http://test-url.com") + self.assertEqual(result, "http://example.com") mock_method.assert_not_called() with patch.object(AccessControl, 'get_url_from_parameter_store') as mock_method: @@ -47,7 +47,7 @@ def test_get_permissions(self, mock_get): mock_response.text = '{"C1200484253-CMR_ONLY":["read","update","delete","order"]}' mock_get.return_value = mock_response - self.access_control.url = "http://test-url.com" + self.access_control.url = "http://example.com" result = self.access_control.get_permissions("user1", "C1200484253-CMR_ONLY") self.assertEqual(result, '{"C1200484253-CMR_ONLY":["read","update","delete","order"]}') @@ -57,7 +57,7 @@ def test_get_permissions_failure(self, mock_get): mock_response.status_code = 404 mock_get.return_value = mock_response - self.access_control.url = "http://test-url.com" + self.access_control.url = "http://example.com" result = self.access_control.get_permissions("user1", "C1200484253-CMR_ONLY") self.assertIsNone(result) From b255ba057b119db374488a0072b5b3a9d12b470f Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Thu, 27 Feb 2025 22:13:38 -0500 Subject: [PATCH 27/29] CMR-10388: adding the removal of the subscription back into a manual test --- .../src/cmr/metadata_db/services/subscriptions.clj | 14 +++++++++++--- .../test/services/subscriptions_test.clj | 8 ++++---- 2 files changed, 15 insertions(+), 7 deletions(-) diff --git a/metadata-db-app/src/cmr/metadata_db/services/subscriptions.clj b/metadata-db-app/src/cmr/metadata_db/services/subscriptions.clj index ba9ec67fcc..3469e35305 100644 --- a/metadata-db-app/src/cmr/metadata_db/services/subscriptions.clj +++ b/metadata-db-app/src/cmr/metadata_db/services/subscriptions.clj @@ -60,13 +60,21 @@ (use 'clojure.set) (defn create-mode-to-endpoints-map - "Creates a mode-to-endpoints map given a list of subscriptions all for the same one collection. - Returns a map in this structure: + "Creates a list of endpoint sets associated to a mode. For each ingest subscription, + an endpoint set is created consisting first of the subscription's endpoint followed + by the subscriber-id. This set is then put on one or more lists that is associated + to each mode described in the subscription. The mode lists are merged together per + each collection concept id that exist in all of the ingest subscriptions. + This function returns a map in this structure: { new: [['sqs:arn:111', 'user1'], ['https://www.url1.com', 'user2'], ['https://www.url2.com', 'user3']], update: [['sqs:arn:111', 'user1']], delete: [['https://www.url1.com', 'user2']] - }" + } + + This structure is used for fast lookups by mode. For a mode, each endpoint set is iterated over + using the subscriber id to check if the subscriber has read access. If they do then a notification + is sent to the endpoint." [subscriptions-of-same-collection] (let [final-map (atom {})] diff --git a/metadata-db-app/test/cmr/metadata_db/test/services/subscriptions_test.clj b/metadata-db-app/test/cmr/metadata_db/test/services/subscriptions_test.clj index 5e5881e209..ad078aef4a 100644 --- a/metadata-db-app/test/cmr/metadata_db/test/services/subscriptions_test.clj +++ b/metadata-db-app/test/cmr/metadata_db/test/services/subscriptions_test.clj @@ -625,14 +625,14 @@ subscription-arn (get-in sub-concept [:extra-fields :aws-arn])] (is (some? subscription-arn)) - ; (when subscription-arn - ; (println "Subscription ARN:" subscription-arn) - ; (is (some? (subscriptions/delete-ingest-subscription test-context sub-concept)))) ;; publish message. this should publish to the internal queue (is (some? (subscriptions/publish-subscription-notification-if-applicable test-context granule-concept))) - (let [internal-queue-url "https://sqs.us-east-1.amazonaws.com/832706493240/cmr-internal-subscriptions-queue-sit" + (when subscription-arn + (is (some? (subscriptions/delete-ingest-subscription test-context sub-concept)))) + + (let [internal-queue-url "https://sqs.us-east-1.amazonaws.com/832706493240/cmr-internal-subscriptions-test-queue" messages (queue/receive-messages sqs-client internal-queue-url) message-str (.body (first messages)) message (json/decode message-str true) From 4587b8ca0e8ec36d8b7c0607f935677148125551 Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Fri, 28 Feb 2025 06:22:33 -0500 Subject: [PATCH 28/29] CMR-10388: Fixing documentation and changing function name per PR reqeusts. --- subscription/src/access_control.py | 24 ++++++++++-------------- subscription/src/env_vars.py | 13 +++++-------- subscription/test/access_control_test.py | 2 +- subscription/test/env_vars_test.py | 8 ++++---- 4 files changed, 20 insertions(+), 27 deletions(-) diff --git a/subscription/src/access_control.py b/subscription/src/access_control.py index 58d081a6fd..6a1abf4ecd 100644 --- a/subscription/src/access_control.py +++ b/subscription/src/access_control.py @@ -26,7 +26,6 @@ class AccessControl: {"C1200484253-CMR_ONLY":["read","update","delete","order"]} """ - def __init__(self): """ Sets up a class variable of url.""" self.url = None @@ -60,10 +59,10 @@ def get_url_from_parameter_store(self): context_param_name = f"{pre_fix}CMR_ACCESS_CONTROL_RELATIVE_ROOT_URL" env_vars = Env_Vars() - protocol = env_vars.get_var(name=protocol_param_name) - port = env_vars.get_var(name=port_param_name) - host = env_vars.get_var(name=host_param_name) - context = env_vars.get_var(name=context_param_name) + protocol = env_vars.get_env_var_from_parameter_store(parameter_name=protocol_param_name) + port = env_vars.get_env_var_from_parameter_store(parameter_name=port_param_name) + host = env_vars.get_env_var_from_parameter_store(parameter_name=host_param_name) + context = env_vars.get_env_var_from_parameter_store(parameter_name=context_param_name) # The context already contains the forward / so we don't need it here. self.url = f"{protocol}://{host}:{port}{context}" @@ -115,15 +114,12 @@ def has_read_permission(self, subscriber_id, 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 + # Check if the permissions is a dictionary and if + # the collection_concept_id is in the permissions dictionary + if isinstance(permissions, dict) and collection_concept_id in permissions: + # Check if "read" is in the list of permissions for the collection + return "read" in permissions[collection_concept_id] + return False except Exception as e: # Handle any exceptions that may occur (e.g., network issues, API errors) diff --git a/subscription/src/env_vars.py b/subscription/src/env_vars.py index ba451580ff..a273902edf 100644 --- a/subscription/src/env_vars.py +++ b/subscription/src/env_vars.py @@ -11,25 +11,22 @@ class Env_Vars: def __init__(self): self.ssm_client = boto3.client('ssm', region_name=os.getenv("AWS_REGION")) - def get_var(self, name, decryption=False): + def get_env_var_from_parameter_store(self, parameter_name, decryption=False): """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) - - value = os.getenv(os_name) + logger.debug(f"Subscription worker: Getting the environment variable called {parameter_name}") + value = os.getenv(parameter_name.split('/')[-1]) if not value: try: # Get the parameter value from AWS Parameter Store - response = self.ssm_client.get_parameter(Name=name, WithDecryption=decryption) + response = self.ssm_client.get_parameter(Name=parameter_name, WithDecryption=decryption) return response['Parameter']['Value'] except ClientError as e: - logger.error(f"Error retrieving parameter from AWS Parameter Store: {e}") + logger.error(f"Error retrieving parameter {parameter_name} from AWS Parameter Store: {e}") raise else: return value diff --git a/subscription/test/access_control_test.py b/subscription/test/access_control_test.py index cd7cafa4b9..5ae5b50cf4 100644 --- a/subscription/test/access_control_test.py +++ b/subscription/test/access_control_test.py @@ -19,7 +19,7 @@ def test_get_url_from_parameter_store_local(self): @patch('access_control.Env_Vars') def test_get_url_from_parameter_store_aws(self, mock_env_vars): mock_env_vars_instance = MagicMock() - mock_env_vars_instance.get_var.side_effect = [ + mock_env_vars_instance.get_env_var_from_parameter_store.side_effect = [ "https", "3011", "cmr.sit.earthdata.nasa.gov", "/access-control" ] mock_env_vars.return_value = mock_env_vars_instance diff --git a/subscription/test/env_vars_test.py b/subscription/test/env_vars_test.py index 3da300c5bc..47d60b7516 100644 --- a/subscription/test/env_vars_test.py +++ b/subscription/test/env_vars_test.py @@ -10,7 +10,7 @@ def setUp(self): @patch.dict(os.environ, {"TEST_VAR": "test_value"}) def test_get_var_from_os(self): - value = self.env_vars.get_var("TEST_VAR") + value = self.env_vars.get_env_var_from_parameter_store("TEST_VAR") self.assertEqual(value, "test_value") @patch.dict(os.environ, {}, clear=True) @@ -26,7 +26,7 @@ def test_get_var_from_ssm(self, mock_boto3_client): self.env_vars.ssm_client = mock_ssm - value = self.env_vars.get_var("SOME_VAR") + value = self.env_vars.get_env_var_from_parameter_store("SOME_VAR") self.assertEqual(value, "some_value") mock_ssm.get_parameter.assert_called_once_with(Name="SOME_VAR", WithDecryption=False) @@ -43,7 +43,7 @@ def test_get_var_with_decryption(self, mock_boto3_client): self.env_vars.ssm_client = mock_ssm - value = self.env_vars.get_var('ENCRYPTED_VAR', decryption=True) + value = self.env_vars.get_env_var_from_parameter_store('ENCRYPTED_VAR', decryption=True) self.assertEqual(value, 'encrypted_value') mock_ssm.get_parameter.assert_called_once_with(Name='ENCRYPTED_VAR', WithDecryption=True) @@ -60,7 +60,7 @@ def test_get_var_ssm_error(self, mock_boto3_client): self.env_vars.ssm_client = mock_ssm with self.assertRaises(ClientError): - self.env_vars.get_var('NONEXISTENT_VAR') + self.env_vars.get_env_var_from_parameter_store('NONEXISTENT_VAR') if __name__ == '__main__': unittest.main() From 7b92d3016ec01fb0f606c2f61bfdba1a713fb079 Mon Sep 17 00:00:00 2001 From: Erich Reiter Date: Fri, 28 Feb 2025 06:37:12 -0500 Subject: [PATCH 29/29] CMR-10388: updating ingest documentation. --- ingest-app/docs/api.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ingest-app/docs/api.md b/ingest-app/docs/api.md index a7272c825f..d065f3a75b 100644 --- a/ingest-app/docs/api.md +++ b/ingest-app/docs/api.md @@ -1098,7 +1098,7 @@ There are two kinds of subscriptions: Batch Notification and Near-Real-Time Noti
  • granule subscription for users to be notified when granules are created/update
    • -
  • Near-Real-Time (NRT) Notification subscriptions are processed on ingest and are only for granules. When a user subscribes, notifications are sent out via the provided notification endpoint, such as an AWS SQS messaging queue. +
  • Near-Real-Time (NRT) Notification subscriptions are processed on ingest and are only for granules. When a user subscribes, notifications are sent out via the provided notification endpoint, such as an AWS SQS messaging queue or a URL. ### Create a Subscription