Merge pull request #10 from scholzj/add-support-for-super-users

Add support for super.users

Author: anderseknert

README.md

@@ -31,6 +31,7 @@ The plugin supports the following properties:
 | `opa.authorizer.cache.initial.capacity` | `5000` | `5000` | Initial decision cache size. |
 | `opa.authorizer.cache.maximum.size` | `50000` | `50000` | Max decision cache size. |
 | `opa.authorizer.cache.expire.after.seconds` | `3600` | `3600` | Decision cache expiry in seconds. |
+| `super.users` | `User:alice;User:bob` | `` | Super users which are always allowed. |
 
 ## Usage
 

src/main/scala/com/bisnode/kafka/authorization/OpaAuthorizer.scala

@@ -11,7 +11,6 @@ import java.util.concurrent.{Callable, ExecutionException, TimeUnit}
 import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.module.scala.{DefaultScalaModule, ScalaObjectMapper}
 import com.google.common.annotations.VisibleForTesting
-import com.google.common.base.Throwables
 import com.google.common.cache.{Cache, CacheBuilder}
 import com.typesafe.scalalogging.LazyLogging
 import kafka.network.RequestChannel
@@ -26,6 +25,7 @@ class OpaAuthorizer extends Authorizer with LazyLogging {
   private var config: Map[String, String] = Map.empty
   private lazy val opaUrl = new URL(config("opa.authorizer.url")).toURI
   private lazy val allowOnError = config.getOrElse("opa.authorizer.allow.on.error", "false").toBoolean
+  private lazy val superUsers = config.getOrElse("super.users", "").split(";").toList
 
   private lazy val cache = CacheBuilder.newBuilder
     .initialCapacity(config.getOrElse("opa.authorizer.cache.initial.capacity", "5000").toInt)
@@ -35,6 +35,11 @@ class OpaAuthorizer extends Authorizer with LazyLogging {
     .asInstanceOf[Cache[Request, Boolean]]
 
   override def authorize(session: RequestChannel.Session, operation: Operation, resource: Resource): Boolean = {
+    if (superUsers.contains(session.principal.toString)) {
+      logger.trace(s"User ${session.principal} is super user")
+      return true
+    }
+
     val request = Request(Input(session, operation, resource))
     try cache.get(request, new AllowCallable(request, opaUrl, allowOnError))
     catch {

src/test/scala/com/bisnode/kafka/authorization/OpaAuthorizerSpec.scala

@@ -108,11 +108,28 @@ class OpaAuthorizerSpec extends FlatSpec with Matchers with PrivateMethodTester
     opaAuthorizer.getCache.size should be (0)
   }
 
+  "OpaAuthorizer" should "authorize super users without checking with OPA" in {
+    val opaAuthorizer = setupAuthorizer(opaUrl)
+
+    val resource = Resource(Topic, "alice-topic", PatternType.LITERAL)
+    val operation = Write
+
+    val session1 = Session(new KafkaPrincipal("User", "CN=my-user"), InetAddress.getLoopbackAddress)
+    opaAuthorizer.authorize(session1, operation, resource) should be (true)
+
+    val session2 = Session(new KafkaPrincipal("User", "CN=my-user2,O=my-org"), InetAddress.getLoopbackAddress)
+    opaAuthorizer.authorize(session2, operation, resource) should be (true)
+
+    val session3 = Session(new KafkaPrincipal("User", "CN=my-user3"), InetAddress.getLoopbackAddress)
+    opaAuthorizer.authorize(session3, operation, resource) should be (false)
+  }
+
   def setupAuthorizer(url: String = opaUrl): OpaAuthorizer = {
     val opaAuthorizer = new OpaAuthorizer()
     val config = new java.util.HashMap[String, String]
     config.put("opa.authorizer.url", url)
     config.put("opa.authorizer.allow.on.error", "false")
+    config.put("super.users", "User:CN=my-user;User:CN=my-user2,O=my-org")
     opaAuthorizer.configure(config)
     opaAuthorizer
   }