fix: allow minimal permissions for consumer destination use (#2233)

This change allows Pub/Sub Subscriber role to be sufficient to receive messages from a subscription, when auto-create-resources is disabled. Fixes: #2231.
GoogleCloudPlatform · Oct 11, 2023 · d5a42c4 · d5a42c4
1 parent ada7051
commit d5a42c4
Show file tree

Hide file tree

Showing 2 changed files with 76 additions and 70 deletions.
diff --git a/...a/com/google/cloud/spring/stream/binder/pubsub/provisioning/PubSubChannelProvisioner.java b/...a/com/google/cloud/spring/stream/binder/pubsub/provisioning/PubSubChannelProvisioner.java
@@ -68,18 +68,16 @@ public ConsumerDestination provisionConsumerDestination(
       String group,
       ExtendedConsumerProperties<PubSubConsumerProperties> properties) {
 
-    String subscriptionName;
-    Subscription subscription;
-
+    String subscriptionName = null;
     String customName = properties.getExtension().getSubscriptionName();
+
     boolean autoCreate = properties.getExtension().isAutoCreateResources();
     PubSubConsumerProperties.DeadLetterPolicy deadLetterPolicy =
         properties.getExtension().getDeadLetterPolicy();
 
     // topicName may be either the short or fully-qualified version.
     String topicShortName =
         TopicName.isParsableFrom(topicName) ? TopicName.parse(topicName).getTopic() : topicName;
-    Topic topic = ensureTopicExists(topicName, autoCreate);
 
     if (StringUtils.hasText(customName)) {
       if (StringUtils.hasText(group)) {
@@ -89,33 +87,21 @@ public ConsumerDestination provisionConsumerDestination(
                 + customName
                 + "'.");
       }
-      subscription = this.pubSubAdmin.getSubscription(customName);
       subscriptionName = customName;
     } else if (StringUtils.hasText(group)) {
       subscriptionName = topicShortName + "." + group;
-      subscription = this.pubSubAdmin.getSubscription(subscriptionName);
-    } else {
-      // Generate anonymous random group since one wasn't provided
-      subscriptionName = "anonymous." + topicShortName + "." + UUID.randomUUID().toString();
-      subscription =
-          this.createSubscription(subscriptionName, topicName, deadLetterPolicy, autoCreate);
-      this.anonymousGroupSubscriptionNames.add(subscriptionName);
     }
 
-    if (subscription == null) {
-      if (autoCreate) {
-        this.createSubscription(subscriptionName, topicName, deadLetterPolicy, autoCreate);
-      } else {
-        throw new ProvisioningException("Non-existing '" + subscriptionName + "' subscription.");
+    if (autoCreate) {
+      if (!StringUtils.hasText(subscriptionName)) {
+        // Generate anonymous random group since one wasn't provided
+        subscriptionName = "anonymous." + topicShortName + "." + UUID.randomUUID();
+        this.anonymousGroupSubscriptionNames.add(subscriptionName);
       }
-    } else if (!subscription.getTopic().equals(topic.getName())) {
-      throw new ProvisioningException(
-          "Existing '"
-              + subscriptionName
-              + "' subscription is for a different topic '"
-              + subscription.getTopic()
-              + "'.");
+      ensureSubscriptionExists(subscriptionName, topicName, deadLetterPolicy, autoCreate);
     }
+
+    Assert.hasText(subscriptionName, "Subscription Name cannot be null or empty");
     return new PubSubConsumerDestination(subscriptionName);
   }
 
@@ -151,6 +137,27 @@ Topic ensureTopicExists(String topicName, boolean autoCreate) {
     throw new ProvisioningException("Non-existing '" + topicName + "' topic.");
   }
 
+  Subscription ensureSubscriptionExists(
+      String subscriptionName,
+      String topicName,
+      PubSubConsumerProperties.DeadLetterPolicy deadLetterPolicy,
+      boolean autoCreate) {
+    Subscription subscription;
+    subscription = this.pubSubAdmin.getSubscription(subscriptionName);
+    if (subscription != null) {
+      if (!subscription.getTopic().equals(topicName)) {
+        throw new ProvisioningException(
+            "Existing '"
+                + subscriptionName
+                + "' subscription is for a different topic '"
+                + subscription.getTopic()
+                + "'.");
+      }
+      return subscription;
+    }
+    return createSubscription(subscriptionName, topicName, deadLetterPolicy, autoCreate);
+  }
+
   private Subscription createSubscription(
       String subscriptionName,
       String topicName,

diff --git a/.../google/cloud/spring/stream/binder/pubsub/provisioning/PubSubChannelProvisionerTests.java b/.../google/cloud/spring/stream/binder/pubsub/provisioning/PubSubChannelProvisionerTests.java
@@ -141,49 +141,9 @@ void testProvisionConsumerDestination_customSubscription() {
     assertThat(result.getName()).isEqualTo("my-custom-subscription");
   }
 
-  @Test
-  void testProvisionConsumerDestination_noTopicException() {
-    when(this.pubSubConsumerProperties.isAutoCreateResources()).thenReturn(false);
-    when(this.pubSubAdminMock.getTopic("topic_A")).thenReturn(null);
-
-    assertThatExceptionOfType(ProvisioningException.class)
-        .isThrownBy(
-            () ->
-                this.pubSubChannelProvisioner.provisionConsumerDestination(
-                    "topic_A", "group_A", this.properties))
-        .withMessage("Non-existing 'topic_A' topic.");
-  }
-
-  @Test
-  void testProvisionConsumerDestination_noSubscriptionException() {
-    when(this.pubSubConsumerProperties.isAutoCreateResources()).thenReturn(false);
-
-    assertThatExceptionOfType(ProvisioningException.class)
-        .isThrownBy(
-            () ->
-                this.pubSubChannelProvisioner.provisionConsumerDestination(
-                    "topic_A", "group_A", this.properties))
-        .withMessage("Non-existing 'topic_A.group_A' subscription.");
-  }
-
-  @Test
-  void testProvisionConsumerDestination_wrongTopicException() {
-    when(this.pubSubConsumerProperties.isAutoCreateResources()).thenReturn(false);
-    when(this.pubSubAdminMock.getSubscription("topic_A.group_A"))
-        .thenReturn(Subscription.newBuilder().setTopic("topic_B").build());
-
-    assertThatExceptionOfType(ProvisioningException.class)
-        .isThrownBy(
-            () ->
-                this.pubSubChannelProvisioner.provisionConsumerDestination(
-                    "topic_A", "group_A", this.properties))
-        .withMessage("Existing 'topic_A.group_A' subscription is for a different topic 'topic_B'.");
-  }
-
   @Test
   void testProvisionConsumerDestination_anonymousGroup() {
-    // should work with auto-create = false
-    when(this.pubSubConsumerProperties.isAutoCreateResources()).thenReturn(false);
+    when(this.pubSubConsumerProperties.isAutoCreateResources()).thenReturn(true);
 
     String subscriptionNameRegex = "anonymous\\.topic_A\\.[a-f0-9\\-]{36}";
 
@@ -270,14 +230,14 @@ void testAfterUnbindConsumer_nonAnonymous() {
   void testProvisionConsumerDestination_concurrentTopicCreation() {
     when(this.pubSubAdminMock.createTopic(any())).thenThrow(AlreadyExistsException.class);
     when(this.pubSubAdminMock.getTopic("already_existing_topic"))
-        .thenReturn(null)
-        .thenReturn(Topic.newBuilder().setName("already_existing_topic").build());
+            .thenReturn(null)
+            .thenReturn(Topic.newBuilder().setName("already_existing_topic").build());
 
     // Ensure no exceptions occur if topic already exists on create call
     assertThat(
-            this.pubSubChannelProvisioner.provisionConsumerDestination(
-                "already_existing_topic", "group1", this.properties))
-        .isNotNull();
+            this.pubSubChannelProvisioner.ensureTopicExists(
+                    "already_existing_topic", true))
+            .isNotNull();
   }
 
   @Test
@@ -289,4 +249,43 @@ void testProvisionConsumerDestination_recursiveExistCalls() {
     assertThatExceptionOfType(ProvisioningException.class)
         .isThrownBy(() -> this.pubSubChannelProvisioner.ensureTopicExists("new_topic", true));
   }
+
+  @Test
+  void testProvisionConsumerDestination_subscriptionNameCannotBeNull() {
+    when(this.pubSubConsumerProperties.isAutoCreateResources()).thenReturn(false);
+    when(this.pubSubConsumerProperties.getSubscriptionName()).thenReturn(null);
+    assertThatExceptionOfType(IllegalArgumentException.class)
+        .isThrownBy(
+            () ->
+                this.pubSubChannelProvisioner.provisionConsumerDestination(
+                    "topic_A", null, this.properties))
+        .withMessage("Subscription Name cannot be null or empty");
+  }
+
+  @Test
+  void testProvisionConsumerDestination_createSubscription() {
+    when(this.pubSubAdminMock.getSubscription("subscription_A"))
+        .thenReturn(
+            Subscription.newBuilder().setTopic("topic_A").setName("subscription_A").build());
+
+    Subscription subscription =
+        this.pubSubChannelProvisioner.ensureSubscriptionExists(
+            "subscription_A", "topic_A", null, true);
+
+    assertThat(subscription.getName()).isEqualTo("subscription_A");
+    assertThat(subscription.getTopic()).isEqualTo("topic_A");
+  }
+
+  @Test
+  void testProvisionConsumerDestination_subscriptionHasDifferentTopic() {
+    when(this.pubSubAdminMock.getSubscription("subscription_A"))
+        .thenReturn(
+            Subscription.newBuilder().setTopic("topic_A").setName("subscription_A").build());
+
+    assertThatExceptionOfType(ProvisioningException.class)
+        .isThrownBy(
+            () ->
+                this.pubSubChannelProvisioner.ensureSubscriptionExists(
+                    "subscription_A", "topic_B", null, true));
+  }
 }