security: enforce AMQPS + fail fast on multiple AEM validation bindings

cds-feature-advanced-event-mesh/src/main/java/com/sap/cds/feature/messaging/aem/jms/AemMessagingConnectionProvider.java

@@ -14,6 +14,7 @@
 import com.sap.cloud.sdk.cloudplatform.connectivity.ServiceBindingDestinationOptions;
 import jakarta.jms.Connection;
 import java.net.URI;
+import java.net.URISyntaxException;
 import java.util.Map;
 import java.util.Optional;
 import java.util.function.BiFunction;
@@ -45,6 +46,7 @@ public AemMessagingConnectionProvider(ServiceBinding binding) {
                 () ->
                     new ServiceException(
                         "AMQP URI key is missing in the service binding. Please check the service binding configuration."));
+    validateAmqpUri(amqpUri, endpointView.getUri().orElse(null), binding.getName().orElse("<unnamed>"));
     amqpUri = amqpUri + SASL_MECHANISM_URI_PARAMETER;
 
     ServiceBindingDestinationOptions options =
@@ -119,4 +121,56 @@ String getToken(String value) {
 
     return token;
   }
+
+  static void validateAmqpUri(String amqpUri, String managementUri, String bindingName) {
+    URI parsed;
+    try {
+      parsed = new URI(amqpUri);
+    } catch (URISyntaxException e) {
+      throw new ServiceException(
+          "Invalid AMQP URI in binding '" + bindingName + "': " + e.getMessage(), e);
+    }
+    if (parsed.getScheme() == null || !parsed.getScheme().equalsIgnoreCase("amqps")) {
+      throw new ServiceException(
+          "AMQP URI in binding '"
+              + bindingName
+              + "' must use scheme 'amqps' (got '"
+              + parsed.getScheme()
+              + "').");
+    }
+    String amqpHost = parsed.getHost();
+    if (amqpHost == null) {
+      throw new ServiceException(
+          "AMQP URI in binding '" + bindingName + "' has no host component.");
+    }
+    if (managementUri != null) {
+      URI mgmt;
+      try {
+        mgmt = new URI(managementUri);
+      } catch (URISyntaxException e) {
+        logger.warn(
+            "Skipping host-consistency check for binding '{}': management URI is malformed ({}).",
+            bindingName,
+            e.getMessage());
+        mgmt = null;
+      }
+      if (mgmt != null) {
+        String mgmtHost = mgmt.getHost();
+        if (mgmtHost == null) {
+          throw new ServiceException(
+              "Management URI in binding '" + bindingName + "' has no host component.");
+        }
+        if (!amqpHost.equalsIgnoreCase(mgmtHost)) {
+          throw new ServiceException(
+              "AMQP URI host '"
+                  + amqpHost
+                  + "' does not match the management URI host '"
+                  + mgmtHost
+                  + "' in binding '"
+                  + bindingName
+                  + "'.");
+        }
+      }
+    }
+  }
 }

cds-feature-advanced-event-mesh/src/main/java/com/sap/cds/feature/messaging/aem/service/AemMessagingServiceConfiguration.java

@@ -47,12 +47,13 @@ public void services(CdsRuntimeConfigurer configurer) {
                         || binding.getTags().contains(BINDING_AEM_LABEL))
             .toList();
     Optional<ServiceBinding> validationBinding =
-        configurer
-            .getCdsRuntime()
-            .getEnvironment()
-            .getServiceBindings()
-            .filter(binding -> ServiceBindingUtils.matches(binding, BINDING_AEM_VALIDATION_LABEL))
-            .findFirst();
+        selectValidationBinding(
+            configurer
+                .getCdsRuntime()
+                .getEnvironment()
+                .getServiceBindings()
+                .filter(binding -> ServiceBindingUtils.matches(binding, BINDING_AEM_VALIDATION_LABEL))
+                .toList());
 
     if (bindings.isEmpty()) {
       logger.info("No service bindings with name '{}' found", BINDING_AEM_LABEL);
@@ -176,4 +177,22 @@ private MessagingService createMessagingService(
 
     return outboxed(service, serviceConfig, runtime);
   }
+
+  static Optional<ServiceBinding> selectValidationBinding(List<ServiceBinding> validationBindings) {
+    if (validationBindings.size() > 1) {
+      String names =
+          String.join(
+              ", ",
+              validationBindings.stream().map(b -> b.getName().orElse("<unnamed>")).toList());
+      throw new ServiceException(
+          "Found "
+              + validationBindings.size()
+              + " '"
+              + BINDING_AEM_VALIDATION_LABEL
+              + "' service bindings ("
+              + names
+              + "). Exactly one validation binding is supported per application.");
+    }
+    return validationBindings.isEmpty() ? Optional.empty() : Optional.of(validationBindings.get(0));
+  }
 }

cds-feature-advanced-event-mesh/src/test/java/com/sap/cds/feature/messaging/aem/jms/AemMessagingConnectionProviderTest.java

@@ -1,8 +1,14 @@
 package com.sap.cds.feature.messaging.aem.jms;
 
-import static org.junit.jupiter.api.Assertions.*;
-import static org.mockito.Mockito.*;
-
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.when;
+
+import com.sap.cds.services.ServiceException;
 import com.sap.cloud.environment.servicebinding.api.ServiceBinding;
 import com.sap.cloud.sdk.cloudplatform.connectivity.Destination;
 import com.sap.cloud.sdk.cloudplatform.connectivity.Header;
@@ -16,72 +22,178 @@
 
 class AemMessagingConnectionProviderTest {
 
-    @Mock private ServiceBinding binding;
-    @Mock private Destination destination;
-    @Mock private HttpDestination httpDestination;
-
-    private AemMessagingConnectionProvider provider;
-
-    @BeforeEach
-    void setUp() {
-        MockitoAnnotations.openMocks(this);
-        when(binding.getName()).thenReturn(Optional.of("my-aem-binding"));
-        when(destination.asHttp()).thenReturn(httpDestination);
-
-        provider = new AemMessagingConnectionProvider(binding, destination);
-    }
-
-    // --- getToken ---
-
-    @Test
-    void getToken_extracts_token_from_bearer_header() {
-        String token = provider.getToken("Bearer my-jwt-token");
-        assertEquals("my-jwt-token", token);
-    }
-
-    @Test
-    void getToken_returns_null_for_null_input() {
-        assertNull(provider.getToken(null));
-    }
-
-    @Test
-    void getToken_returns_null_when_no_space_in_value() {
-        assertNull(provider.getToken("BearerWithoutSpace"));
-    }
-
-    @Test
-    void getToken_returns_null_for_empty_string() {
-        assertNull(provider.getToken(""));
-    }
-
-    // --- fetchToken ---
-
-    @Test
-    void fetchToken_extracts_bearer_token_from_authorization_header() {
-        when(httpDestination.getHeaders()).thenReturn(
-                List.of(new Header("Authorization", "Bearer jwt-abc-123")));
-
-        Optional<String> token = provider.fetchToken();
-
-        assertTrue(token.isPresent());
-        assertEquals("jwt-abc-123", token.get());
-    }
-
-    @Test
-    void fetchToken_returns_empty_when_no_authorization_header() {
-        when(httpDestination.getHeaders()).thenReturn(List.of(new Header("X-Other", "value")));
-
-        Optional<String> token = provider.fetchToken();
-
-        assertFalse(token.isPresent());
-    }
-
-    @Test
-    void fetchToken_returns_empty_when_no_headers_at_all() {
-        when(httpDestination.getHeaders()).thenReturn(List.of());
-
-        Optional<String> token = provider.fetchToken();
-
-        assertFalse(token.isPresent());
-    }
+  @Mock private ServiceBinding binding;
+  @Mock private Destination destination;
+  @Mock private HttpDestination httpDestination;
+
+  private AemMessagingConnectionProvider provider;
+
+  @BeforeEach
+  void setUp() {
+    MockitoAnnotations.openMocks(this);
+    when(binding.getName()).thenReturn(Optional.of("my-aem-binding"));
+    when(destination.asHttp()).thenReturn(httpDestination);
+
+    provider = new AemMessagingConnectionProvider(binding, destination);
+  }
+
+  // --- getToken ---
+
+  @Test
+  void getToken_extracts_token_from_bearer_header() {
+    String token = provider.getToken("Bearer my-jwt-token");
+    assertEquals("my-jwt-token", token);
+  }
+
+  @Test
+  void getToken_returns_null_for_null_input() {
+    assertNull(provider.getToken(null));
+  }
+
+  @Test
+  void getToken_returns_null_when_no_space_in_value() {
+    assertNull(provider.getToken("BearerWithoutSpace"));
+  }
+
+  @Test
+  void getToken_returns_null_for_empty_string() {
+    assertNull(provider.getToken(""));
+  }
+
+  // --- fetchToken ---
+
+  @Test
+  void fetchToken_extracts_bearer_token_from_authorization_header() {
+    when(httpDestination.getHeaders())
+        .thenReturn(List.of(new Header("Authorization", "Bearer jwt-abc-123")));
+
+    Optional<String> token = provider.fetchToken();
+
+    assertTrue(token.isPresent());
+    assertEquals("jwt-abc-123", token.get());
+  }
+
+  @Test
+  void fetchToken_returns_empty_when_no_authorization_header() {
+    when(httpDestination.getHeaders()).thenReturn(List.of(new Header("X-Other", "value")));
+
+    Optional<String> token = provider.fetchToken();
+
+    assertFalse(token.isPresent());
+  }
+
+  @Test
+  void fetchToken_returns_empty_when_no_headers_at_all() {
+    when(httpDestination.getHeaders()).thenReturn(List.of());
+
+    Optional<String> token = provider.fetchToken();
+
+    assertFalse(token.isPresent());
+  }
+
+  // --- validateAmqpUri ---
+
+  @Test
+  void validateAmqpUri_acceptsAmqpsWithMatchingHost() {
+    assertDoesNotThrow(
+        () ->
+            AemMessagingConnectionProvider.validateAmqpUri(
+                "amqps://broker.example.com:5671",
+                "https://broker.example.com:943",
+                "my-binding"));
+  }
+
+  @Test
+  void validateAmqpUri_acceptsAmqpsWhenManagementUriIsNull() {
+    assertDoesNotThrow(
+        () ->
+            AemMessagingConnectionProvider.validateAmqpUri(
+                "amqps://broker.example.com:5671", null, "my-binding"));
+  }
+
+  @Test
+  void validateAmqpUri_isCaseInsensitiveOnScheme() {
+    assertDoesNotThrow(
+        () ->
+            AemMessagingConnectionProvider.validateAmqpUri(
+                "AMQPS://broker.example.com:5671", null, "my-binding"));
+  }
+
+  @Test
+  void validateAmqpUri_rejectsPlainAmqp() {
+    ServiceException ex =
+        assertThrows(
+            ServiceException.class,
+            () ->
+                AemMessagingConnectionProvider.validateAmqpUri(
+                    "amqp://broker.example.com:5672", null, "my-binding"));
+    assertTrue(ex.getMessage().contains("amqps"), ex.getMessage());
+    assertTrue(ex.getMessage().contains("my-binding"), ex.getMessage());
+  }
+
+  @Test
+  void validateAmqpUri_rejectsUnexpectedScheme() {
+    assertThrows(
+        ServiceException.class,
+        () ->
+            AemMessagingConnectionProvider.validateAmqpUri(
+                "ws://broker.example.com", null, "my-binding"));
+  }
+
+  @Test
+  void validateAmqpUri_rejectsMalformedUri() {
+    assertThrows(
+        ServiceException.class,
+        () ->
+            AemMessagingConnectionProvider.validateAmqpUri(
+                "::not a uri::", null, "my-binding"));
+  }
+
+  @Test
+  void validateAmqpUri_rejectsHostMismatchAgainstManagementUri() {
+    ServiceException ex =
+        assertThrows(
+            ServiceException.class,
+            () ->
+                AemMessagingConnectionProvider.validateAmqpUri(
+                    "amqps://attacker.example.com:5671",
+                    "https://broker.example.com:943",
+                    "my-binding"));
+    assertTrue(ex.getMessage().contains("attacker.example.com"), ex.getMessage());
+    assertTrue(ex.getMessage().contains("broker.example.com"), ex.getMessage());
+  }
+
+  @Test
+  void validateAmqpUri_toleratesMalformedManagementUri() {
+    assertDoesNotThrow(
+        () ->
+            AemMessagingConnectionProvider.validateAmqpUri(
+                "amqps://broker.example.com:5671", "::not a uri::", "my-binding"));
+  }
+
+  @Test
+  void validateAmqpUri_rejectsAmqpUriWithoutHost() {
+    // Opaque URI parses successfully but has no host component.
+    ServiceException ex =
+        assertThrows(
+            ServiceException.class,
+            () ->
+                AemMessagingConnectionProvider.validateAmqpUri(
+                    "amqps:opaque-body", null, "my-binding"));
+    assertTrue(ex.getMessage().contains("my-binding"), ex.getMessage());
+    assertTrue(ex.getMessage().contains("host"), ex.getMessage());
+  }
+
+  @Test
+  void validateAmqpUri_rejectsManagementUriWithoutHost() {
+    // Management URI parses but yields a null host — must not silently pass the check.
+    ServiceException ex =
+        assertThrows(
+            ServiceException.class,
+            () ->
+                AemMessagingConnectionProvider.validateAmqpUri(
+                    "amqps://broker.example.com:5671", "https:opaque-body", "my-binding"));
+    assertTrue(ex.getMessage().contains("my-binding"), ex.getMessage());
+    assertTrue(ex.getMessage().contains("host"), ex.getMessage());
+  }
 }