Use scope config for compute-engine auth

google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/ComputeEngineCredentials.scala

@@ -25,18 +25,19 @@ import scala.concurrent.Future
 @InternalApi
 private[auth] object ComputeEngineCredentials {
 
-  def apply()(implicit system: ClassicActorSystemProvider): Future[Credentials] =
+  def apply(scopes: Set[String])(implicit system: ClassicActorSystemProvider): Future[Credentials] = {
     GoogleComputeMetadata
       .getProjectId()
-      .map(new ComputeEngineCredentials(_))(system.classicSystem.dispatcher)
+      .map(projectId => new ComputeEngineCredentials(projectId, scopes))(system.classicSystem.dispatcher)
+  }
 
 }
 
 @InternalApi
-private final class ComputeEngineCredentials(projectId: String)(implicit mat: Materializer)
+private final class ComputeEngineCredentials(projectId: String, scopes: Set[String])(implicit mat: Materializer)
     extends OAuth2Credentials(projectId) {
   override protected def getAccessToken()(implicit mat: Materializer,
       settings: RequestSettings,
       clock: Clock): Future[AccessToken] =
-    GoogleComputeMetadata.getAccessToken()
+    GoogleComputeMetadata.getAccessToken(scopes)
 }

google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/Credentials.scala

@@ -20,6 +20,7 @@ import pekko.event.Logging
 import pekko.http.scaladsl.model.headers.HttpCredentials
 import pekko.stream.connectors.google.RequestSettings
 import pekko.util.JavaDurationConverters._
+import pekko.util.ccompat.JavaConverters._
 import com.google.auth.{ Credentials => GoogleCredentials }
 import com.typesafe.config.Config
 
@@ -72,8 +73,10 @@ object Credentials {
   private def parseServiceAccount(c: Config)(implicit system: ClassicActorSystemProvider) =
     ServiceAccountCredentials(c.getConfig("service-account"))
 
-  private def parseComputeEngine(c: Config)(implicit system: ClassicActorSystemProvider) =
-    Await.result(ComputeEngineCredentials(), c.getDuration("compute-engine.timeout").asScala)
+  private def parseComputeEngine(c: Config)(implicit system: ClassicActorSystemProvider) = {
+    val scopes = c.getStringList("scopes").asScala.toSet
+    Await.result(ComputeEngineCredentials(scopes), c.getDuration("compute-engine.timeout").asScala)
+  }
 
   private def parseUserAccess(c: Config)(implicit system: ClassicActorSystemProvider) =
     UserAccessCredentials(c.getConfig("user-access"))

google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleComputeMetadata.scala

@@ -19,7 +19,7 @@ import pekko.annotation.InternalApi
 import pekko.http.scaladsl.Http
 import pekko.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
 import pekko.http.scaladsl.model.HttpMethods.GET
-import pekko.http.scaladsl.model.HttpRequest
+import pekko.http.scaladsl.model.{ HttpRequest, Uri }
 import pekko.http.scaladsl.model.headers.RawHeader
 import pekko.http.scaladsl.unmarshalling.Unmarshal
 import pekko.stream.Materializer
@@ -35,17 +35,26 @@ private[auth] object GoogleComputeMetadata {
   private val projectIdUrl = s"$metadataUrl/project/project-id"
   private val `Metadata-Flavor` = RawHeader("Metadata-Flavor", "Google")
 
-  private val tokenRequest = HttpRequest(GET, tokenUrl).addHeader(`Metadata-Flavor`)
+  private def tokenRequest(scopes: Set[String]) = {
+    val finalUri =
+      if (scopes.nonEmpty)
+        Uri(tokenUrl).withQuery(Uri.Query(Map(
+          "scopes" -> scopes.mkString(","))))
+      else
+        Uri(tokenUrl)
+
+    HttpRequest(GET, finalUri).addHeader(`Metadata-Flavor`)
+  }
   private val projectIdRequest = HttpRequest(GET, projectIdUrl).addHeader(`Metadata-Flavor`)
 
-  def getAccessToken()(
+  def getAccessToken(scopes: Set[String])(
       implicit mat: Materializer,
       clock: Clock): Future[AccessToken] = {
     import SprayJsonSupport._
     import mat.executionContext
     implicit val system: ActorSystem = mat.system
     for {
-      response <- Http().singleRequest(tokenRequest)
+      response <- Http().singleRequest(tokenRequest(scopes))
       token <- Unmarshal(response.entity).to[AccessToken]
     } yield token
   }