Google Secret Manager as Source

CHANGELOG.md

@@ -4,6 +4,9 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
+## [0.10 - SNAPSHOT]
+- Added Google Secret Manager as an ACL source
+
 ## [0.9 - SNAPSHOT]
 - TODO: Upgrade to Kafka 2.4.x (PR welcome)
 - Added Bitbucket Cloud as an ACL source

README.md

@@ -30,6 +30,7 @@ Current sources shipping with KSM include:
 - GitLab (using Personal Auth Tokens)
 - BitBucket
 - Amazon S3
+- Google Secret Manager
 - Build your own (and contribute back!)
 
 # Building
@@ -145,6 +146,10 @@ The [default configurations](src/main/resources/application.conf) can be overwri
       - `SOURCE_S3_OBJECTKEY` The Object containing the ACL CSV in S3
     - `com.github.simplesteph.ksm.source.BitbucketServerSourceAcl`: get the ACL from Bitbucket Server using the v1 REST API. Great if you have private repos in Bitbucket.
     - `com.github.simplesteph.ksm.source.BitbucketCloudSourceAcl`: get the ACL from Bitbucket Cloud using the Bitbucket Cloud REST API v2.
+    - `com.github.simplesteph.ksm.source.GoogleSecretManagerAcl`: get the ACL from Google Secret Manager using Google Cloud SDK. You can store ACL contents in many secrets as you want (secrets contents are limited in 64kb) and filter it using labels. Use the same ACL CSV layout and **do not include** the Header. This requires `projectid` and `label_filter`:
+      - `SOURCE_GCP_PROJECTID`: GCP Project ID
+      - `SOURCE_GCP_LABEL_FILTER`: You can filter Secrets using the format: `label=value,label2=value`
+      - `GOOGLE_APPLICATION_CREDENTIALS`: One option to configure your GCP credentials. More information: [GCP Authentication](https://cloud.google.com/docs/authentication/getting-started)
 - `NOTIFICATION_CLASS`: Class for notification in case of ACL changes in Kafka.
     - `com.github.simplesteph.ksm.notification.ConsoleNotification` (default): Print changes to the console. Useful for logging
     - `com.github.simplesteph.ksm.notification.SlackNotification`: Send notifications to a Slack channel (useful for devops / admin team)

build.sbt

@@ -2,7 +2,7 @@ name := "kafka-security-manager"
 
 organization := "com.github.simplesteph.ksm"
 
-version := "0.9-SNAPSHOT"
+version := "0.10-SNAPSHOT"
 
 scalaVersion := "2.12.8"
 
@@ -52,7 +52,12 @@ libraryDependencies ++= Seq(
   "beyondthelines" %% "grpcgatewayruntime" % "0.0.9" % "compile,protobuf",
 
   // AWS SDK to access S3
-  "com.amazonaws" % "aws-java-sdk-s3" % "1.11.385"
+  "com.amazonaws" % "aws-java-sdk-s3" % "1.11.385",
+
+  //Google Cloud Secret Manager
+  "com.google.cloud" % "google-cloud-secretmanager" % "1.0.1",
+
+  "com.google.api" % "gax-grpc" % "1.56.0"
 
 )
 

src/main/resources/application.conf

@@ -92,6 +92,10 @@ source {
     bucketname = ${?SOURCE_S3_BUCKETNAME}
     objectkey = ${?SOURCE_S3_OBJECTKEY}
   }
+  gcp-secret-manager {
+    projectid = ${?SOURCE_GCP_PROJECTID}
+    label_filter = ${?SOURCE_GCP_LABEL_FILTER}
+  }
   bitbucket-server {
     hostname = "localhost"
     hostname = ${?SOURCE_BITBUCKET_SERVER_HOSTNAME}

src/main/scala/com/github/simplesteph/ksm/source/GoogleSecretManagerAcl.scala

@@ -0,0 +1,127 @@
+package com.github.simplesteph.ksm.source
+
+import java.io._
+import java.util.Date
+
+import com.google.cloud.secretmanager.v1._
+import com.typesafe.config.Config
+import org.slf4j.LoggerFactory
+
+import scala.collection.mutable.ListBuffer
+
+class GoogleSecretManagerAcl extends SourceAcl {
+
+  private val log = LoggerFactory.getLogger(classOf[GoogleSecretManagerAcl])
+
+  /**
+    * Config Prefix for configuring this module
+    */
+  override val CONFIG_PREFIX: String = "gcp-secret-manager"
+
+  final val PROJECT = "projectid"
+  final val LABEL_FILTER = "label_filter"
+
+  var lastModified: Date = new Date(0)
+  var projectId: String = _
+  var labelFilter: String = _
+
+  /**
+    * internal config definition for the module
+    */
+  override def configure(config: Config): Unit = {
+    projectId = config.getString(PROJECT)
+    log.info("Google Project Id: " + projectId)
+
+    labelFilter = config.getString(LABEL_FILTER)
+    log.info("Label filter: " + labelFilter)
+  }
+
+  private final val HEADER =
+    "KafkaPrincipal,ResourceType,PatternType,ResourceName,Operation,PermissionType,Host\n"
+
+  /**
+    * Refresh the current view on the external source of truth for Acl
+    * Ideally this function is smart and does not pull the entire external Acl at every iteration
+    * Return `None` if the Source Acls have not changed (usually using metadata).
+    * Return `Some(x)` if the Acls have changed. `x` represents the parsing and parsing errors if any
+    * Note: the first call to this function should never return `None`.
+    *
+    * Kafka Security Manager will not update Acls in Kafka until there are no errors in the result
+    *
+    * @return
+    */
+  override def refresh(): Option[Reader] = {
+
+    val secretManagerServiceClient = SecretManagerServiceClient.create
+
+    val parent = ProjectName.of(projectId)
+    val request = ListSecretsRequest.newBuilder.setParent(parent.toString).build
+
+    var response = new StringBuilder(HEADER)
+
+    val filter: Map[String, String] = labelFilter
+      .split(",")
+      .filter(_ != "")
+      .map({ pair =>
+        {
+          val Array(k, v) = pair.split("=")
+          k.trim -> v.trim
+        }
+      })
+      .toMap
+
+    var secrets = new ListBuffer[Secret]()
+
+    // Filter secrets by lastModified and Labels
+    // For the moment (2020-June) Google SDK doesn't have filter on API level,
+    // so we need to iterate over all secrets
+    secretManagerServiceClient
+      .listSecrets(request)
+      .iterateAll
+      .forEach(secret => {
+        // check there are any label filter in config
+        if (filter.nonEmpty) {
+          var counter = 0
+          secret.getLabelsMap.forEach((key, value) => {
+            if (filter.exists(x => x._1 == key && x._2 == value)) {
+              counter += 1
+            }
+          })
+          if (counter == filter.size) {
+            secrets += secret
+          }
+        } else {
+          secrets += secret
+        }
+      })
+
+    log.info("Found " + secrets.size + " Secrets to add.")
+
+    // Get latest version of secret
+    secrets.toList.foreach(secret =>
+      response ++= secretManagerServiceClient
+        .accessSecretVersion(
+          SecretVersionName.parse(secret.getName + "/versions/latest").toString
+        )
+        .getPayload
+        .getData
+        .toStringUtf8
+    )
+
+    log.debug("ACL Contents:")
+    log.debug(response.toString())
+
+    secretManagerServiceClient.close()
+    secrets.clear()
+
+    Some(new StringReader(response.toString()))
+
+  }
+
+  /**
+    * Close all the necessary underlying objects or connections belonging to this instance
+    */
+  override def close(): Unit = {
+    // GCP SDK closed at refresh
+  }
+}