[RHCLOUD-32506] Send relevant data from incoming payloads to Kessel

connector-drawer/src/main/java/com/redhat/cloud/notifications/connector/drawer/DrawerCloudEventDataExtractor.java

@@ -16,5 +16,7 @@ public void extract(Exchange exchange, JsonObject cloudEventData) {
         exchange.setProperty(ExchangeProperty.DRAWER_ENTRY_PAYLOAD, notification.drawerEntryPayload());
         exchange.setProperty(ExchangeProperty.RECIPIENT_SETTINGS, notification.recipientSettings());
         exchange.setProperty(ExchangeProperty.UNSUBSCRIBERS, notification.unsubscribers());
+        exchange.setProperty(ExchangeProperty.UNSUBSCRIBERS, notification.unsubscribers());
+        exchange.setProperty(ExchangeProperty.AUTHORIZATION_CRITERIA, notification.authorizationCriteria());
     }
 }

connector-drawer/src/main/java/com/redhat/cloud/notifications/connector/drawer/constant/ExchangeProperty.java

@@ -9,4 +9,5 @@ public class ExchangeProperty {
     public static final String UNSUBSCRIBERS = "unsubscribers";
 
     public static final String DRAWER_ENTRY_PAYLOAD = "drawer_entry_payload";
+    public static final String AUTHORIZATION_CRITERIA = "authorization_criteria";
 }

connector-drawer/src/main/java/com/redhat/cloud/notifications/connector/drawer/model/DrawerNotificationToConnector.java

@@ -1,6 +1,7 @@
 package com.redhat.cloud.notifications.connector.drawer.model;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
+import io.vertx.core.json.JsonObject;
 import java.util.Collection;
 
 public record DrawerNotificationToConnector(
@@ -15,5 +16,8 @@ public record DrawerNotificationToConnector(
     Collection<RecipientSettings> recipientSettings,
 
     @JsonProperty("unsubscribers")
-    Collection<String> unsubscribers
+    Collection<String> unsubscribers,
+
+    @JsonProperty("authorization_criteria")
+    JsonObject authorizationCriteria
 )  { }

connector-drawer/src/main/java/com/redhat/cloud/notifications/connector/drawer/recipients/RecipientsResolverPreparer.java

@@ -5,6 +5,7 @@
 import com.redhat.cloud.notifications.connector.drawer.constant.ExchangeProperty;
 import com.redhat.cloud.notifications.connector.drawer.model.RecipientSettings;
 import com.redhat.cloud.notifications.connector.drawer.recipients.pojo.RecipientsQuery;
+import io.vertx.core.json.JsonObject;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
 import org.apache.camel.Exchange;
@@ -26,12 +27,14 @@ public void process(final Exchange exchange) throws JsonProcessingException {
         List<RecipientSettings> recipientSettings = exchange.getProperty(ExchangeProperty.RECIPIENT_SETTINGS, List.class);
         Set<String> unsubscribers = exchange.getProperty(ExchangeProperty.UNSUBSCRIBERS, Set.class);
         final String orgId = exchange.getProperty(ORG_ID, String.class);
+        JsonObject authorizationCriteria = exchange.getProperty(ExchangeProperty.AUTHORIZATION_CRITERIA, JsonObject.class);
 
         RecipientsQuery recipientsQuery = new RecipientsQuery();
         recipientsQuery.unsubscribers = unsubscribers;
         recipientsQuery.orgId = orgId;
         recipientsQuery.recipientSettings = Set.copyOf(recipientSettings);
         recipientsQuery.subscribedByDefault = true;
+        recipientsQuery.authorizationCriteria = authorizationCriteria;
 
         // Serialize the payload.
         exchange.getMessage().setBody(objectMapper.writeValueAsString(recipientsQuery));

connector-drawer/src/main/java/com/redhat/cloud/notifications/connector/drawer/recipients/pojo/RecipientsQuery.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.databind.annotation.JsonNaming;
 import com.redhat.cloud.notifications.connector.drawer.model.RecipientSettings;
+import io.vertx.core.json.JsonObject;
 import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
 import java.util.Set;
@@ -22,4 +23,6 @@ public class RecipientsQuery {
     public Set<String> unsubscribers;
 
     public boolean subscribedByDefault;
+
+    public JsonObject authorizationCriteria;
 }

connector-drawer/src/test/java/com/redhat/cloud/notifications/connector/drawer/DrawerConnectorRoutesTest.java

@@ -103,7 +103,7 @@ private static DrawerNotificationToConnector buildTestDrawerNotificationToConnec
         drawerEntryPayload.setBundle("My Bundle");
 
         RecipientSettings recipientSettings = new RecipientSettings();
-        return new DrawerNotificationToConnector(orgId, drawerEntryPayload, Set.of(recipientSettings), List.of("user-1", "user-2"));
+        return new DrawerNotificationToConnector(orgId, drawerEntryPayload, Set.of(recipientSettings), List.of("user-1", "user-2"), new JsonObject());
     }
 
     private HttpRequest getMockHttpRequest(String path, ExpectationResponseCallback expectationResponseCallback) {

connector-email/src/main/java/com/redhat/cloud/notifications/connector/email/EmailCloudEventDataExtractor.java

@@ -47,6 +47,8 @@ public void extract(final Exchange exchange, final JsonObject cloudEventData) {
                 .map(String.class::cast)
                 .collect(toSet());
 
+        final JsonObject authorizationCriteria = cloudEventData.getJsonObject("authorization_criteria");
+
         final Set<String> emails = recipientSettings.stream()
                 .filter(settings -> settings.getEmails() != null)
                 .flatMap(settings -> settings.getEmails().stream())
@@ -58,6 +60,7 @@ public void extract(final Exchange exchange, final JsonObject cloudEventData) {
         exchange.setProperty(ExchangeProperty.SUBSCRIBED_BY_DEFAULT, cloudEventData.getBoolean("subscribed_by_default"));
         exchange.setProperty(ExchangeProperty.SUBSCRIBERS, subscribers);
         exchange.setProperty(ExchangeProperty.UNSUBSCRIBERS, unsubscribers);
+        exchange.setProperty(ExchangeProperty.AUTHORIZATION_CRITERIA, authorizationCriteria);
         exchange.setProperty(ExchangeProperty.EMAIL_RECIPIENTS, emails);
         exchange.setProperty(ExchangeProperty.EMAIL_SENDER, cloudEventData.getString("email_sender"));
 

connector-email/src/main/java/com/redhat/cloud/notifications/connector/email/constants/ExchangeProperty.java

@@ -48,4 +48,6 @@ public class ExchangeProperty {
     public static final String RECIPIENTS_SIZE = "recipientsSize";
 
     public static final String USE_EMAIL_BOP_V1_SSL = "use_email_bop_V1_ssl";
+
+    public static final String AUTHORIZATION_CRITERIA = "authorization_criteria";
 }

connector-email/src/main/java/com/redhat/cloud/notifications/connector/email/processors/recipients/RecipientsQuery.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.databind.annotation.JsonNaming;
 import com.redhat.cloud.notifications.connector.email.model.settings.RecipientSettings;
+import io.vertx.core.json.JsonObject;
 import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
 import java.util.Set;
@@ -22,4 +23,6 @@ public class RecipientsQuery {
     public Set<String> unsubscribers;
 
     public boolean subscribedByDefault;
+
+    public JsonObject authorizationCriteria;
 }

connector-email/src/main/java/com/redhat/cloud/notifications/connector/email/processors/recipients/RecipientsResolverRequestPreparer.java

@@ -4,6 +4,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.redhat.cloud.notifications.connector.email.constants.ExchangeProperty;
 import com.redhat.cloud.notifications.connector.email.model.settings.RecipientSettings;
+import io.vertx.core.json.JsonObject;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
 import org.apache.camel.Exchange;
@@ -25,6 +26,8 @@ public void process(final Exchange exchange) throws JsonProcessingException {
         List<RecipientSettings> recipientSettings = exchange.getProperty(ExchangeProperty.RECIPIENT_SETTINGS, List.class);
         Set<String> subscribers = exchange.getProperty(ExchangeProperty.SUBSCRIBERS, Set.class);
         Set<String> unsubscribers = exchange.getProperty(ExchangeProperty.UNSUBSCRIBERS, Set.class);
+        JsonObject authorizationCriteria = exchange.getProperty(ExchangeProperty.AUTHORIZATION_CRITERIA, JsonObject.class);
+
         boolean subscribedByDefault = exchange.getProperty(ExchangeProperty.SUBSCRIBED_BY_DEFAULT, boolean.class);
         final String orgId = exchange.getProperty(ORG_ID, String.class);
 
@@ -34,6 +37,7 @@ public void process(final Exchange exchange) throws JsonProcessingException {
         recipientsQuery.orgId = orgId;
         recipientsQuery.recipientSettings = Set.copyOf(recipientSettings);
         recipientsQuery.subscribedByDefault = subscribedByDefault;
+        recipientsQuery.authorizationCriteria = authorizationCriteria;
 
         // Serialize the payload.
         exchange.getMessage().setBody(objectMapper.writeValueAsString(recipientsQuery));

engine/src/main/java/com/redhat/cloud/notifications/processors/ExternalAuthorizationCriteria.java

@@ -0,0 +1,56 @@
+package com.redhat.cloud.notifications.processors;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.databind.PropertyNamingStrategies;
+import com.fasterxml.jackson.databind.annotation.JsonNaming;
+import jakarta.validation.constraints.NotNull;
+import java.util.Objects;
+
+@JsonNaming(PropertyNamingStrategies.SnakeCaseStrategy.class)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class ExternalAuthorizationCriteria {
+
+    @NotNull
+    private String assetType;
+
+    @NotNull
+    private String assetId;
+
+    @NotNull
+    private String relation;
+
+    public ExternalAuthorizationCriteria(String assetType, String assetId, String relation) {
+        this.assetType = assetType;
+        this.assetId = assetId;
+        this.relation = relation;
+    }
+
+    public @NotNull String getAssetType() {
+        return assetType;
+    }
+
+    public @NotNull String getAssetId() {
+        return assetId;
+    }
+
+    public @NotNull String getRelation() {
+        return relation;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        ExternalAuthorizationCriteria that = (ExternalAuthorizationCriteria) o;
+        return Objects.equals(assetType, that.assetType) && Objects.equals(assetId, that.assetId) && Objects.equals(relation, that.relation);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(assetType, assetId, relation);
+    }
+}

engine/src/main/java/com/redhat/cloud/notifications/processors/ExternalAuthorizationCriteriaExtractor.java

@@ -0,0 +1,41 @@
+package com.redhat.cloud.notifications.processors;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.redhat.cloud.notifications.models.EmailAggregation;
+import com.redhat.cloud.notifications.models.Event;
+import com.redhat.cloud.notifications.transformers.BaseTransformer;
+import io.quarkus.logging.Log;
+import io.vertx.core.json.JsonObject;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+
+@ApplicationScoped
+public class ExternalAuthorizationCriteriaExtractor {
+
+    public static final String EXTERNAL_AUTHORIZATION_CRITERIA = "authorization";
+
+    @Inject
+    BaseTransformer baseTransformer;
+
+    @Inject
+    ObjectMapper objectMapper;
+
+    public ExternalAuthorizationCriteria extract(Event event) {
+        return extract(baseTransformer.toJsonObject(event));
+    }
+
+    public ExternalAuthorizationCriteria extract(EmailAggregation emailAggregation) {
+        return extract(emailAggregation.getPayload());
+    }
+
+    private ExternalAuthorizationCriteria extract(JsonObject data) {
+        if (null != data.getJsonObject(BaseTransformer.CONTEXT) && null != data.getJsonObject(BaseTransformer.CONTEXT).getJsonObject(EXTERNAL_AUTHORIZATION_CRITERIA)) {
+            try {
+                return objectMapper.convertValue(data.getJsonObject(BaseTransformer.CONTEXT).getJsonObject(EXTERNAL_AUTHORIZATION_CRITERIA), ExternalAuthorizationCriteria.class);
+            } catch (IllegalArgumentException e) {
+                Log.error("Error parsing authorization criteria", e);
+            }
+        }
+        return null;
+    }
+}

engine/src/main/java/com/redhat/cloud/notifications/processors/drawer/DrawerNotificationToConnector.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.redhat.cloud.notifications.models.DrawerEntryPayload;
+import com.redhat.cloud.notifications.processors.ExternalAuthorizationCriteria;
 import com.redhat.cloud.notifications.processors.email.connector.dto.RecipientSettings;
 import java.util.Collection;
 
@@ -17,5 +18,8 @@ public record DrawerNotificationToConnector(
     Collection<RecipientSettings> recipientSettings,
 
     @JsonProperty("unsubscribers")
-    Collection<String> unsubscribers
+    Collection<String> unsubscribers,
+
+    @JsonProperty("authorization_criteria")
+    ExternalAuthorizationCriteria externalAuthorizationCriteria
 )  { }

engine/src/main/java/com/redhat/cloud/notifications/processors/drawer/DrawerProcessor.java

@@ -16,6 +16,7 @@
 import com.redhat.cloud.notifications.models.Event;
 import com.redhat.cloud.notifications.models.IntegrationTemplate;
 import com.redhat.cloud.notifications.processors.ConnectorSender;
+import com.redhat.cloud.notifications.processors.ExternalAuthorizationCriteriaExtractor;
 import com.redhat.cloud.notifications.processors.SystemEndpointTypeProcessor;
 import com.redhat.cloud.notifications.processors.email.connector.dto.RecipientSettings;
 import com.redhat.cloud.notifications.templates.TemplateService;
@@ -73,6 +74,9 @@ public class DrawerProcessor extends SystemEndpointTypeProcessor {
     @Inject
     BundleRepository bundleRepository;
 
+    @Inject
+    ExternalAuthorizationCriteriaExtractor externalAuthorizationCriteriaExtractor;
+
     @Override
     public void process(Event event, List<Endpoint> endpoints) {
         if (!engineConfig.isDrawerEnabled()) {
@@ -103,7 +107,8 @@ public void process(Event event, List<Endpoint> endpoints) {
             event.getOrgId(),
             drawerEntryPayload,
             recipientSettings,
-            unsubscribers
+            unsubscribers,
+            externalAuthorizationCriteriaExtractor.extract(event)
         );
 
         connectorSender.send(event, endpoint, JsonObject.mapFrom(drawerNotificationToConnector));

engine/src/main/java/com/redhat/cloud/notifications/processors/email/EmailAggregationProcessor.java

@@ -342,7 +342,9 @@ private void processBundleAggregation(List<AggregationCommand> aggregationComman
                      */
                     recipientsUsernames,
                     Collections.emptySet(),
-                    false
+                    false,
+                    // because recipient list has already been computed using authz constraints, we don't need it for the aggregated email
+                    null
                 );
 
                 connectorSender.send(aggregatorEvent, endpoint, JsonObject.mapFrom(emailNotification));

engine/src/main/java/com/redhat/cloud/notifications/processors/email/EmailAggregator.java

@@ -11,6 +11,8 @@
 import com.redhat.cloud.notifications.models.Endpoint;
 import com.redhat.cloud.notifications.models.EventType;
 import com.redhat.cloud.notifications.models.SubscriptionType;
+import com.redhat.cloud.notifications.processors.ExternalAuthorizationCriteria;
+import com.redhat.cloud.notifications.processors.ExternalAuthorizationCriteriaExtractor;
 import com.redhat.cloud.notifications.processors.email.aggregators.AbstractEmailPayloadAggregator;
 import com.redhat.cloud.notifications.processors.email.aggregators.EmailPayloadAggregatorFactory;
 import com.redhat.cloud.notifications.recipients.RecipientResolver;
@@ -67,6 +69,9 @@ public class EmailAggregator {
     private static final String EVENT_TYPE_KEY = "event_type";
     private static final String RECIPIENTS_KEY = "recipients";
 
+    @Inject
+    ExternalAuthorizationCriteriaExtractor externalAuthorizationCriteriaExtractor;
+
     @ConfigProperty(name = "notifications.aggregation.max-page-size", defaultValue = "100")
     int maxPageSize;
 
@@ -89,6 +94,8 @@ public Map<User, Map<String, Object>> getAggregated(UUID appId, EmailAggregation
 
             // For each aggregation...
             for (EmailAggregation aggregation : aggregations) {
+
+                ExternalAuthorizationCriteria externalAuthorizationCriteria = externalAuthorizationCriteriaExtractor.extract(aggregation);
                 // We need its event type to determine the target endpoints.
                 String eventTypeName = getEventType(aggregation);
                 EventType eventType;
@@ -128,7 +135,8 @@ public Map<User, Map<String, Object>> getAggregated(UUID appId, EmailAggregation
                             ).collect(toSet()),
                             subscribers,
                             unsubscribers,
-                            eventType.isSubscribedByDefault()
+                            eventType.isSubscribedByDefault(),
+                            externalAuthorizationCriteria
                         );
                     } catch (Exception ex) {
                         Log.error("Error calling external recipients resolver service", ex);

engine/src/main/java/com/redhat/cloud/notifications/processors/email/EmailProcessor.java

@@ -8,6 +8,8 @@
 import com.redhat.cloud.notifications.models.Event;
 import com.redhat.cloud.notifications.models.InstantEmailTemplate;
 import com.redhat.cloud.notifications.processors.ConnectorSender;
+import com.redhat.cloud.notifications.processors.ExternalAuthorizationCriteria;
+import com.redhat.cloud.notifications.processors.ExternalAuthorizationCriteriaExtractor;
 import com.redhat.cloud.notifications.processors.SystemEndpointTypeProcessor;
 import com.redhat.cloud.notifications.processors.email.connector.dto.EmailNotification;
 import com.redhat.cloud.notifications.processors.email.connector.dto.RecipientSettings;
@@ -51,6 +53,9 @@ public class EmailProcessor extends SystemEndpointTypeProcessor {
     @Inject
     SubscriptionRepository subscriptionRepository;
 
+    @Inject
+    ExternalAuthorizationCriteriaExtractor externalAuthorizationCriteriaExtractor;
+
     @Override
     public void process(final Event event, final List<Endpoint> endpoints) {
         // Generate an aggregation if the event supports it.
@@ -63,6 +68,8 @@ public void process(final Event event, final List<Endpoint> endpoints) {
             return;
         }
 
+        ExternalAuthorizationCriteria externalAuthorizationCriteria = externalAuthorizationCriteriaExtractor.extract(event);
+
         Set<String> subscribers;
         Set<String> unsubscribers;
         if (event.getEventType().isSubscribedByDefault()) {
@@ -111,7 +118,8 @@ public void process(final Event event, final List<Endpoint> endpoints) {
             recipientSettings,
             subscribers,
             unsubscribers,
-            event.getEventType().isSubscribedByDefault()
+            event.getEventType().isSubscribedByDefault(),
+            externalAuthorizationCriteria
         );
 
         final JsonObject payload = JsonObject.mapFrom(emailNotification);