Kafka-manager add BasicAuth

add-basicauth.patch

From 68bfd9b2ef2da377c65b512451dbfe60be85bc4d Mon Sep 17 00:00:00 2001
From: Serho Liu <serholiu@gmail.com>
Date: Sun, 24 May 2015 11:21:58 +0800
Subject: [PATCH] add BasicAuth

---
 app/GlobalKafkaManager.scala | 75 ++++++++++++++++++++++++++++++++++++++++++++
 conf/application.conf        |  3 ++
 2 files changed, 78 insertions(+)

diff --git a/app/GlobalKafkaManager.scala b/app/GlobalKafkaManager.scala
index 90aec67..de6e8e8 100644
--- a/app/GlobalKafkaManager.scala
+++ b/app/GlobalKafkaManager.scala
@@ -6,6 +6,9 @@
 import controllers.KafkaManagerContext
 import kafka.manager.KafkaManager
 import play.api._
+import play.api.mvc._
+import sun.misc.BASE64Decoder
+import scala.concurrent.Future
 
 /**
  * @author hiral
@@ -27,3 +30,75 @@ object GlobalKafkaManager extends GlobalSettings {
 }
 
 
+object BasicAuthFilter extends Filter {
+
+  import play.api.libs.concurrent.Execution.Implicits.defaultContext
+
+  //need the space at the end
+  private lazy val basicSt = "basic "
+
+  private lazy val unauthResult = Results.Unauthorized.withHeaders(("WWW-Authenticate", "Basic realm=\"Need Auth\""))
+
+  private lazy val username = Play.current.configuration.getString("auth.username") match {
+    case Some(pass) => pass
+    case None => "test"
+  }
+
+  private lazy val password = Play.current.configuration.getString("auth.password") match {
+    case Some(word) => word
+    case None => "test"
+  }
+
+  //This is needed if you are behind a load balancer or a proxy
+  private def getUserIPAddress(request: RequestHeader): String = {
+    return request.headers.get("x-forwarded-for").getOrElse(request.remoteAddress.toString)
+  }
+
+  private def logFailedAttempt(requestHeader: RequestHeader) = {
+    Logger.warn(s"IP address ${getUserIPAddress(requestHeader)} failed to log in, " +
+      s"requested uri: ${requestHeader.uri}")
+  }
+
+  private def decodeBasicAuth(auth: String): Option[(String, String)] = {
+    if (auth.length() < basicSt.length()) {
+      return None
+    }
+    val basicReqSt = auth.substring(0, basicSt.length())
+    if (basicReqSt.toLowerCase() != basicSt) {
+      return None
+    }
+    val basicAuthSt = auth.replaceFirst(basicReqSt, "")
+
+    //BESE64Decoder is not thread safe, don't make it a field of this object
+    val decoder = new BASE64Decoder()
+    val decodedAuthSt = new String(decoder.decodeBuffer(basicAuthSt), "UTF-8")
+    val usernamePassword = decodedAuthSt.split(":")
+    if (usernamePassword.length >= 2) {
+      //account for ":" in passwords
+      return Some((usernamePassword(0), usernamePassword.splitAt(1)._2.mkString))
+    }
+    None
+  }
+
+  def apply(nextFilter: (RequestHeader) => Future[Result])(requestHeader: RequestHeader): Future[Result] = {
+
+    requestHeader.headers.get("authorization").map { basicAuth =>
+      decodeBasicAuth(basicAuth) match {
+        case Some((user, pass)) => {
+          if (username == user && password == pass) {
+            return nextFilter(requestHeader)
+          }
+        }
+        case _ => ;
+      }
+      logFailedAttempt(requestHeader)
+      return Future.successful(unauthResult)
+    }.getOrElse({
+      logFailedAttempt(requestHeader)
+      Future.successful(unauthResult)
+    })
+  }
+}
+
+object Global extends WithFilters(BasicAuthFilter)
+
diff --git a/conf/application.conf b/conf/application.conf
index 8ed2809..e538cb9 100644
--- a/conf/application.conf
+++ b/conf/application.conf
@@ -65,3 +65,6 @@ kafka-manager.zkhosts=${?ZK_HOSTS}
 pinned-dispatcher.type="PinnedDispatcher"
 pinned-dispatcher.executor="thread-pool-executor"
 
+# Auth Info
+auth.username = "test"
+auth.password = "test123"
\ No newline at end of file
-- 
2.3.5