313539854 Add Pub/Sub as notification backend

sources/pom.xml

@@ -69,6 +69,11 @@
       <artifactId>google-api-services-cloudasset</artifactId>
       <version>v1-rev20231103-2.0.0</version>
     </dependency>
+    <dependency>
+      <groupId>com.google.apis</groupId>
+      <artifactId>google-api-services-pubsub</artifactId>
+      <version>v1-rev20230830-2.0.0</version>
+    </dependency>
     <dependency>
       <groupId>com.google.apis</groupId>
       <artifactId>google-api-services-iamcredentials</artifactId>
@@ -79,6 +84,11 @@
       <artifactId>google-api-services-secretmanager</artifactId>
       <version>v1-rev20230804-2.0.0</version>
     </dependency>
+    <dependency>
+      <groupId>com.google.code.gson</groupId>
+      <artifactId>gson</artifactId>
+      <version>2.10.1</version>
+    </dependency>
     <dependency>
       <groupId>com.sun.mail</groupId>
       <artifactId>jakarta.mail</artifactId>

sources/src/main/java/com/google/solutions/jitaccess/core/adapters/PubSubAdapter.java

@@ -0,0 +1,111 @@
+//
+// Copyright 2023 Google LLC
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+//
+
+package com.google.solutions.jitaccess.core.adapters;
+
+import com.google.api.client.json.gson.GsonFactory;
+import com.google.api.client.googleapis.json.GoogleJsonResponseException;
+import com.google.auth.http.HttpCredentialsAdapter;
+import com.google.auth.oauth2.GoogleCredentials;
+import com.google.api.services.pubsub.Pubsub;
+import com.google.api.services.pubsub.model.PubsubMessage;
+import com.google.api.services.pubsub.model.PublishRequest;
+import com.google.common.base.Preconditions;
+import com.google.solutions.jitaccess.core.AccessDeniedException;
+import com.google.solutions.jitaccess.core.AccessException;
+import com.google.solutions.jitaccess.core.ApplicationVersion;
+import com.google.solutions.jitaccess.core.NotAuthenticatedException;
+import com.google.solutions.jitaccess.core.data.Topic;
+import jakarta.enterprise.context.ApplicationScoped;
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.util.Arrays;
+
+@ApplicationScoped
+public class PubSubAdapter {
+  private final GoogleCredentials credentials;
+  private final HttpTransport.Options httpOptions;
+
+  public PubSubAdapter(
+    GoogleCredentials credentials,
+    HttpTransport.Options httpOptions)
+  {
+    Preconditions.checkNotNull(credentials, "credentials");
+    Preconditions.checkNotNull(httpOptions, "httpOptions");
+
+    this.credentials = credentials;
+    this.httpOptions = httpOptions;
+  }
+
+  private Pubsub createClient() throws IOException {
+    try {
+      return new Pubsub.Builder(
+          HttpTransport.newTransport(),
+          new GsonFactory(),
+          HttpTransport.newAuthenticatingRequestInitializer(this.credentials, this.httpOptions))
+        .setApplicationName(ApplicationVersion.USER_AGENT)
+        .build();
+    }
+    catch (GeneralSecurityException e) {
+      throw new IOException("Creating a PubSub client failed", e);
+    }
+  }
+
+  public String publish(
+    Topic topic,
+    PubsubMessage message
+  ) throws AccessException, IOException {
+    var client = createClient();
+
+    try {
+      var request = new PublishRequest();
+      request.setMessages(Arrays.asList(message));
+
+      var result = client
+        .projects()
+        .topics()
+        .publish(topic.getFullResourceName(), request)
+        .execute();
+      if (result.getMessageIds().size() < 1){
+        throw new IOException(
+          String.format("Publishing message to topic %s returned empty response", topic));
+      }
+
+      return result.getMessageIds().get(0);
+    }
+    catch (GoogleJsonResponseException e) {
+      switch (e.getStatusCode()) {
+        case 401:
+          throw new NotAuthenticatedException("Not authenticated", e);
+        case 403:
+        case 404:
+          throw new AccessDeniedException(
+            String.format(
+              "Pub/Sub topic '%s' cannot be accessed or does not exist: %s",
+              topic,
+              e.getMessage()),
+            e);
+        default:
+          throw (GoogleJsonResponseException)e.fillInStackTrace();
+      }
+    }
+  }
+}

sources/src/main/java/com/google/solutions/jitaccess/core/data/Topic.java

@@ -0,0 +1,33 @@
+//
+// Copyright 2023 Google LLC
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+//
+
+package com.google.solutions.jitaccess.core.data;
+
+public record Topic(String projectId, String topicName) {
+  @Override
+  public String toString() {
+    return getFullResourceName();
+  }
+
+  public String getFullResourceName() {
+    return String.format("projects/%s/topics/%s", this.projectId, this.topicName);
+  }
+}

sources/src/main/java/com/google/solutions/jitaccess/core/services/MailNotificationService.java

@@ -34,7 +34,10 @@
 import java.time.ZoneOffset;
 import java.time.format.DateTimeFormatter;
 import java.time.temporal.ChronoUnit;
+import java.util.Collection;
 import java.util.EnumSet;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 /**
  * Concrete class that delivers notifications over SMTP.
@@ -176,6 +179,11 @@ public String format(NotificationService.Notification notification) {
             .truncatedTo(ChronoUnit.SECONDS)
             .format(DateTimeFormatter.RFC_1123_DATE_TIME);
         }
+        else if (property.getValue() instanceof Collection<?>) {
+          propertyValue = ((Collection<?>)property.getValue()).stream()
+            .map(i -> i.toString())
+            .collect(Collectors.joining(", "));
+        }
         else {
           //
           // Convert to a safe string.

sources/src/main/java/com/google/solutions/jitaccess/core/services/NotificationService.java

@@ -25,14 +25,13 @@
 import com.google.solutions.jitaccess.core.data.UserId;
 
 import jakarta.enterprise.context.ApplicationScoped;
-
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.stream.Collectors;
 
 /**
- * Service for notifying users about activation requests..
+ * Service for notifying users about activation requests.
  */
 @ApplicationScoped
 public abstract class NotificationService {
@@ -49,17 +48,25 @@ public abstract class NotificationService {
    * Concrete class that prints notifications to STDOUT. Useful for local development only.
    */
   public static class SilentNotificationService extends NotificationService {
+    private final boolean printToConsole;
+
+    public SilentNotificationService(boolean printToConsole) {
+      this.printToConsole = printToConsole;
+    }
+
     @Override
     public boolean canSendNotifications() {
       return false;
     }
 
     @Override
     public void sendNotification(Notification notification) throws NotificationException {
-      //
-      // Print it so that we can see the message during development.
-      //
-      System.out.println(notification);
+      if (this.printToConsole) {
+        //
+        // Print it so that we can see the message during development.
+        //
+        System.out.println(notification);
+      }
     }
   }
 

sources/src/main/java/com/google/solutions/jitaccess/core/services/PubSubNotificationService.java

@@ -0,0 +1,95 @@
+package com.google.solutions.jitaccess.core.services;
+
+import com.google.api.client.json.GenericJson;
+import com.google.api.services.pubsub.model.PubsubMessage;
+import com.google.common.base.Preconditions;
+import com.google.gson.Gson;
+import com.google.solutions.jitaccess.core.AccessException;
+import com.google.solutions.jitaccess.core.adapters.PubSubAdapter;
+import com.google.solutions.jitaccess.core.data.Topic;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.util.Collection;
+import java.util.stream.Collectors;
+
+/**
+ * Concrete class that delivers notifications over Pub/Sub.
+ */
+public class PubSubNotificationService extends NotificationService {
+  private final PubSubAdapter adapter;
+  private final Options options;
+
+  public PubSubNotificationService(
+    PubSubAdapter adapter,
+    Options options
+  ) {
+    Preconditions.checkNotNull(adapter, "adapter");
+    Preconditions.checkNotNull(options, "options");
+    Preconditions.checkNotNull(options.topic, "options");
+
+    this.adapter = adapter;
+    this.options = options;
+  }
+
+  // -------------------------------------------------------------------------
+  // NotificationService implementation.
+  // -------------------------------------------------------------------------
+
+  @Override
+  public boolean canSendNotifications() {
+    return true;
+  }
+
+  @Override
+  public void sendNotification(Notification notification) throws NotificationException {
+    var attributes = new GenericJson();
+    for (var property : notification.properties.entrySet())
+    {
+      Object propertyValue;
+      if (property.getValue() instanceof Instant) {
+        //
+        // Serialize ISO-8601 representation instead of individual
+        // object fields.
+        //
+        propertyValue = ((Instant)property.getValue()).toString();
+      }
+      else if (property.getValue() instanceof Collection<?>) {
+        propertyValue = ((Collection<?>)property.getValue()).stream()
+          .map(i -> i.toString())
+          .collect(Collectors.toList());
+      }
+      else {
+        propertyValue = property.getValue().toString();
+      }
+
+      attributes.set(property.getKey().toLowerCase(), propertyValue);
+    }
+
+    var payload = new GenericJson()
+      .set("type", notification.getType())
+      .set("attributes", attributes);
+
+    var payloadAsJson = new Gson().toJson(payload);
+
+    try {
+      var message = new PubsubMessage()
+        .encodeData(payloadAsJson.getBytes(StandardCharsets.UTF_8));
+
+      this.adapter.publish(options.topic, message);
+    } catch (AccessException | IOException e){
+      throw new NotificationException("Publishing event to Pub/Sub failed", e);
+    }
+  }
+
+  // -------------------------------------------------------------------------
+  // Inner classes.
+  // -------------------------------------------------------------------------
+
+  /**
+   * @param topicResourceName PubSub topic in format projects/{project}/topics/{topic}
+   */
+  public record Options(Topic topic) {
+  }
+}

sources/src/main/java/com/google/solutions/jitaccess/core/services/RoleActivationService.java

@@ -39,6 +39,7 @@
 import java.security.SecureRandom;
 import java.time.Duration;
 import java.time.Instant;
+import java.time.temporal.ChronoUnit;
 import java.util.Base64;
 import java.util.EnumSet;
 import java.util.List;
@@ -170,7 +171,7 @@ public Activation activateProjectRoleForSelf(
     // accumulating junk, and to prevent hitting the binding limit.
     //
 
-    var activationTime = Instant.now();
+    var activationTime = Instant.now().truncatedTo(ChronoUnit.SECONDS);;
     var expiryTime = activationTime.plus(activationTimeout);
     var bindingDescription = String.format(
       "Self-approved, justification: %s",
@@ -318,7 +319,7 @@ public ActivationRequest createActivationRequestForPeer(
     //
     // Issue an activation request.
     //
-    var startTime = Instant.now();
+    var startTime = Instant.now().truncatedTo(ChronoUnit.SECONDS);
     var endTime = startTime.plus(activationTimeout);
 
     return new ActivationRequest(