Merge commit 'refs/pull/594/head' of github.com:apache/usergrid

apache · Dec 13, 2017 · a88a875 · a88a875
2 parents c172133 + 2b35733
commit a88a875
Show file tree

Hide file tree

Showing 5 changed files with 111 additions and 44 deletions.
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/EntityManagerFig.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/EntityManagerFig.java
@@ -46,7 +46,7 @@ public interface EntityManagerFig extends GuicyFig {
      * Comma-separated list of one or more Amazon regions to use if multiregion
      * is set to true.
      */
-    @Key( "usergrid.queue.regionList" )
+    @Key( "usergrid.cluster.region.list" )
     @Default("us-east-1")
     String getRegionList();
 

diff --git a/...persistence/queue/src/main/java/org/apache/usergrid/persistence/queue/LegacyQueueFig.java b/...persistence/queue/src/main/java/org/apache/usergrid/persistence/queue/LegacyQueueFig.java
@@ -14,14 +14,14 @@ public interface LegacyQueueFig extends GuicyFig {
      * http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/using-regions-availability-zones.html*
      */
 
-    String USERGRID_CLUSTER_REGION_LIST = "usergrid.cluster.region.list";
-    String USERGRID_CLUSTER_REGION_LOCAL = "usergrid.cluster.region.local";
+    String USERGRID_QUEUE_REGION_LIST = "usergrid.queue.regionList";
+    String USERGRID_QUEUE_REGION_LOCAL = "usergrid.queue.region";
 
 
     /**
      * Primary region to use for Amazon queues.
      */
-    @Key( USERGRID_CLUSTER_REGION_LOCAL )
+    @Key(USERGRID_QUEUE_REGION_LOCAL)
     @Default("us-east-1")
     String getPrimaryRegion();
 
@@ -37,7 +37,7 @@ public interface LegacyQueueFig extends GuicyFig {
      * Comma-separated list of one or more Amazon regions to use if multiregion
      * is set to true.
      */
-    @Key( USERGRID_CLUSTER_REGION_LIST )
+    @Key(USERGRID_QUEUE_REGION_LIST)
     @Default("us-east-1")
     String getRegionList();
 

diff --git a/...e/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/SNSQueueManagerImpl.java b/...e/queue/src/main/java/org/apache/usergrid/persistence/queue/impl/SNSQueueManagerImpl.java
@@ -249,7 +249,7 @@ private String setupTopics( final String queueName ) throws Exception {
                     region = Region.getRegion(regions);
                 }
                 catch (IllegalArgumentException e) {
-                    throw new IllegalArgumentException("INVALID REGION FROM CONFIGURATION " + LegacyQueueFig.USERGRID_CLUSTER_REGION_LIST + ": " + regionName, e);
+                    throw new IllegalArgumentException("INVALID REGION FROM CONFIGURATION " + LegacyQueueFig.USERGRID_QUEUE_REGION_LIST + ": " + regionName, e);
                 }
 
                 AmazonSQSClient sqsClient = createSQSClient( region );
@@ -825,7 +825,7 @@ private Region getRegion() {
             return Region.getRegion(regions);
         }
         catch (IllegalArgumentException e) {
-            throw new IllegalArgumentException("INVALID PRIMARY REGION FROM CONFIGURATION " + LegacyQueueFig.USERGRID_CLUSTER_REGION_LOCAL + ": " + regionName, e);
+            throw new IllegalArgumentException("INVALID PRIMARY REGION FROM CONFIGURATION " + LegacyQueueFig.USERGRID_QUEUE_REGION_LOCAL + ": " + regionName, e);
         }
     }
 

diff --git a/...eue/src/main/java/org/apache/usergrid/persistence/queue/util/AmazonNotificationUtils.java b/...eue/src/main/java/org/apache/usergrid/persistence/queue/util/AmazonNotificationUtils.java
@@ -1,35 +1,23 @@
 package org.apache.usergrid.persistence.queue.util;
 
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
+import com.amazonaws.auth.policy.*;
+import com.amazonaws.auth.policy.conditions.ArnCondition;
+import com.amazonaws.services.sqs.model.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.usergrid.persistence.queue.LegacyQueueFig;
 
-import com.amazonaws.auth.policy.Condition;
-import com.amazonaws.auth.policy.Policy;
-import com.amazonaws.auth.policy.Principal;
-import com.amazonaws.auth.policy.Resource;
-import com.amazonaws.auth.policy.Statement;
 import com.amazonaws.auth.policy.actions.SQSActions;
 import com.amazonaws.auth.policy.conditions.ConditionFactory;
 import com.amazonaws.services.sns.AmazonSNSClient;
 import com.amazonaws.services.sns.model.CreateTopicResult;
 import com.amazonaws.services.sns.model.ListTopicsResult;
 import com.amazonaws.services.sns.model.Topic;
 import com.amazonaws.services.sqs.AmazonSQSClient;
-import com.amazonaws.services.sqs.model.CreateQueueRequest;
-import com.amazonaws.services.sqs.model.CreateQueueResult;
-import com.amazonaws.services.sqs.model.GetQueueAttributesRequest;
-import com.amazonaws.services.sqs.model.GetQueueAttributesResult;
-import com.amazonaws.services.sqs.model.GetQueueUrlResult;
-import com.amazonaws.services.sqs.model.QueueDoesNotExistException;
-import com.amazonaws.services.sqs.model.SetQueueAttributesRequest;
 
 
 /**
@@ -85,34 +73,113 @@ public static String createQueue( final AmazonSQSClient sqs, final String queueN
     public static void setQueuePermissionsToReceive( final AmazonSQSClient sqs, final String queueUrl,
                                                      final List<String> topicARNs ) throws Exception {
 
-        String queueARN = getQueueArnByUrl( sqs, queueUrl );
-
-        Statement statement = new Statement( Statement.Effect.Allow ).withActions( SQSActions.SendMessage )
-                                                                     .withPrincipals( new Principal( "*" ) )
-                                                                     .withResources( new Resource( queueARN ) );
+        // retrieve queue ARN and policy
+        List<String> sqsAttrNames = Arrays.asList(QueueAttributeName.QueueArn.toString(),
+            QueueAttributeName.Policy.toString());
+        GetQueueAttributesRequest getQueueAttributesRequest =
+            new GetQueueAttributesRequest( queueUrl ).withAttributeNames( sqsAttrNames );
+        GetQueueAttributesResult queueAttributesResult = sqs.getQueueAttributes( getQueueAttributesRequest );
+        Map<String, String> sqsAttributeMap = queueAttributesResult.getAttributes();
+        String queueARN = sqsAttributeMap.get(QueueAttributeName.QueueArn.toString());
+        String policyJson = sqsAttributeMap.get(QueueAttributeName.Policy.toString());
+
+        // cannot send ARN in settings update, so remove it
+        sqsAttributeMap.remove(QueueAttributeName.QueueArn.toString());
+
+        // get existing policy from JSON
+        Policy policy = policyJson != null && policyJson.length() > 0 ? Policy.fromJson(policyJson) : new Policy();
+
+        // see if permissions already exist, and find ArnLike conditions
+        boolean matchingConditionFound = false;
+        boolean policyEdited = false;
+        for (Statement statement : policy.getStatements()) {
+            logger.info("statement id: {}, effect: {}, action: {}, resources:{}",
+                statement.getId(), statement.getEffect().name(),
+                statement.getActions().get(0).getActionName(),
+                statement.getResources().get(0).getId());
+
+            // must be Allow effect
+            if (! statement.getEffect().name().equals(Statement.Effect.Allow.name())) {
+                continue;
+            }
 
-        List<Condition> conditions = new ArrayList<>();
+            // must be SendMessage action
+            boolean actionFound = false;
+            for (Action action : statement.getActions()) {
+                // do lower case comparison, since UI adds SQS.SendMessage but SDK uses sqs.SendMessage
+                if (action.getActionName().toLowerCase().equals(SQSActions.SendMessage.getActionName().toLowerCase())) {
+                    actionFound = true;
+                    break;
+                }
+            }
+            if (!actionFound) {
+                continue;
+            }
 
-        for ( String topicARN : topicARNs ) {
+            // must be same queue resource
+            boolean queueResourceFound = false;
+            for (Resource resource : statement.getResources()) {
+                if (resource.getId().equals(queueARN)) {
+                    queueResourceFound = true;
+                    break;
+                }
+            }
+            if (!queueResourceFound) {
+                continue;
+            }
 
-            conditions.add( ConditionFactory.newSourceArnCondition( topicARN ) );
+            // found matching statement, check conditions for source ARN
+            for (Condition condition : statement.getConditions()) {
+                if (logger.isTraceEnabled()) {
+                    logger.trace("condition type: {}, conditionKey: {}", condition.getType(), condition.getConditionKey());
+                }
+                if (condition.getType().equals(ArnCondition.ArnComparisonType.ArnLike.name()) &&
+                    condition.getConditionKey().equals(ConditionFactory.SOURCE_ARN_CONDITION_KEY)) {
+                    matchingConditionFound = true;
+                    for (String topicARN : topicARNs) {
+                        if (! condition.getValues().contains(topicARN)) {
+                            // topic doesn't exist, add it
+                            policyEdited = true;
+                            condition.getValues().add(topicARN);
+                        }
+                    }
+                }
+            }
         }
-        statement.setConditions( conditions );
 
-        Policy policy = new Policy( "SubscriptionPermission" ).withStatements( statement );
+        if (!matchingConditionFound) {
+            // never found ArnLike SourceArn condition, need to add a statement
+            List<Condition> conditions = new ArrayList<>();
 
+            for (String topicARN : topicARNs) {
 
-        final Map<String, String> queueAttributes = new HashMap<>();
-        queueAttributes.put( "Policy", policy.toJson() );
+                conditions.add(ConditionFactory.newSourceArnCondition(topicARN));
+            }
 
-        SetQueueAttributesRequest queueAttributesRequest = new SetQueueAttributesRequest( queueUrl, queueAttributes );
+            Statement statement = new Statement(Statement.Effect.Allow)
+                .withPrincipals(Principal.AllUsers)
+                .withActions(SQSActions.SendMessage)
+                .withResources(new Resource(queueARN));
+            statement.setConditions(conditions);
 
-        try {
-            sqs.setQueueAttributes( queueAttributesRequest );
+            policy.getStatements().add(statement);
+            policyEdited = true;
         }
-        catch ( Exception e ) {
-            logger.error( "Failed to set permissions on QUEUE ARN=[{}] for TOPIC ARNs=[{}]", queueARN,
-                topicARNs.toString(), e );
+
+        if (policyEdited) {
+            sqsAttributeMap.put(QueueAttributeName.Policy.toString(), policy.toJson());
+
+            // log if permissions are being updated
+            logger.info("updating permissions for queueARN: {}, new policy: {}", queueARN, policy.toJson());
+
+            SetQueueAttributesRequest setQueueAttributesRequest = new SetQueueAttributesRequest(queueUrl, sqsAttributeMap);
+
+            try {
+                sqs.setQueueAttributes(setQueueAttributesRequest);
+            } catch (Exception e) {
+                logger.error("Failed to set permissions on QUEUE ARN=[{}] for TOPIC ARNs=[{}]", queueARN,
+                    topicARNs.toString(), e);
+            }
         }
     }
 

diff --git a/...src/test/java/org/apache/usergrid/rest/applications/collection/CollectionsResourceIT.java b/...src/test/java/org/apache/usergrid/rest/applications/collection/CollectionsResourceIT.java
@@ -1256,22 +1256,22 @@ public void testCollectionAuthoritativeRegion() {
 
         try {
             app().collection( collectionName ).collection( "_settings" )
-                .post( new Entity().chainPut(REGION_SETTING, "us-moon-1" ) );
+                .post( new Entity().chainPut(REGION_SETTING, "us-moon" ) );
             fail( "post should have failed");
 
         } catch ( BadRequestException expected ) {}
 
         // set collection region with good region
 
         app().collection( collectionName ).collection( "_settings" )
-            .post( new Entity().chainPut( REGION_SETTING, "us-east-1" ) );
+            .post( new Entity().chainPut( REGION_SETTING, "us-east" ) );
 
         // get collection settings see that we have a region
 
         collection = app().collection( collectionName ).collection( "_settings" ).get();
         settings = (Map<String, Object>)collection.getResponse().getData();
         assertNotNull( settings.get( REGION_SETTING ));
-        assertEquals( "us-east-1", settings.get( REGION_SETTING ));
+        assertEquals( "us-east", settings.get( REGION_SETTING ));
 
         // unset the collection region