Introduced JWT based authorization for registry communication

build.sbt

@@ -55,6 +55,8 @@ libraryDependencies ++= Seq(
 
 libraryDependencies += "org.json4s" %% "json4s-jackson" % "3.5.3"
 
+libraryDependencies += "com.pauldijou" %% "jwt-core" % "1.0.0"
+
 libraryDependencies += "ch.qos.logback" % "logback-classic" % "1.2.3" % Runtime
 
 val elastic4sVersion = "6.3.0"

src/main/scala/de/upb/cs/swt/delphi/crawler/Configuration.scala

@@ -96,6 +96,8 @@ class Configuration {
 
   lazy val instanceId : Option[Long] = InstanceRegistry.handleInstanceStart(this)
 
+  val jwtSecretKey: String  = sys.env.getOrElse("DELPHI_JWT_SECRET","changeme")
+
   case class Throttle(element : Int, per : FiniteDuration, maxBurst : Int, mode : ThrottleMode)
 }
 

src/main/scala/de/upb/cs/swt/delphi/crawler/authorization/AuthProvider.scala

@@ -0,0 +1,35 @@
+// Copyright (C) 2018 The Delphi Team.
+// See the LICENCE file distributed with this work for additional
+// information regarding copyright ownership.
+//
+// Licensed 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 de.upb.cs.swt.delphi.crawler.authorization
+
+import de.upb.cs.swt.delphi.crawler.Configuration
+import pdi.jwt.{Jwt, JwtAlgorithm, JwtClaim}
+
+object AuthProvider {
+
+  def generateJwt(validFor: Long = 1, useGenericName: Boolean = false) (implicit configuration: Configuration): String = {
+    val claim = JwtClaim()
+      .issuedNow
+      .expiresIn(validFor * 60)
+      .startsNow
+      . + ("user_id", if (useGenericName) configuration.instanceName else s"${configuration.instanceId.get}")
+      . + ("user_type", "Component")
+
+
+    Jwt.encode(claim, configuration.jwtSecretKey, JwtAlgorithm.HS256)
+  }
+
+}

src/main/scala/de/upb/cs/swt/delphi/crawler/instancemanagement/InstanceRegistry.scala

@@ -21,9 +21,11 @@ import java.net.InetAddress
 import akka.actor.ActorSystem
 import akka.http.scaladsl.Http
 import akka.http.scaladsl.model._
+import akka.http.scaladsl.model.headers.RawHeader
 import akka.http.scaladsl.unmarshalling.Unmarshal
 import akka.stream.ActorMaterializer
 import akka.util.ByteString
+import de.upb.cs.swt.delphi.crawler.authorization.AuthProvider
 import de.upb.cs.swt.delphi.crawler.instancemanagement.InstanceEnums.{ComponentType, InstanceState}
 import de.upb.cs.swt.delphi.crawler.{AppLogging, Configuration, Crawler}
 
@@ -99,7 +101,10 @@ object InstanceRegistry extends InstanceJsonSupport with AppLogging
           method = HttpMethods.POST,
           configuration.instanceRegistryUri + ReportOperationType.toOperationUriString(operationType, id))
 
-        Await.result(Http(system).singleRequest(request) map {response =>
+        val useGenericNameForToken = operationType == ReportOperationType.Start //Must use generic name for startup, no id known at that point
+
+        Await.result(Http(system).singleRequest(request.withHeaders(RawHeader("Authorization",
+          s"Bearer ${AuthProvider.generateJwt(useGenericName = useGenericNameForToken)(configuration)}"))) map { response =>
           if(response.status == StatusCodes.OK){
             log.info(s"Successfully reported ${operationType.toString} to Instance Registry.")
             Success()
@@ -122,7 +127,7 @@ object InstanceRegistry extends InstanceJsonSupport with AppLogging
   private def register(configuration: Configuration) : Try[Long] = {
     val instance = createInstance(None,configuration.controlServerPort, configuration.instanceName)
 
-    Await.result(postInstance(instance, configuration.instanceRegistryUri + "/register") map {response =>
+    Await.result(postInstance(instance, configuration.instanceRegistryUri + "/register")(configuration) map {response =>
       if(response.status == StatusCodes.OK){
         Await.result(Unmarshal(response.entity).to[String] map { assignedID =>
           val id = assignedID.toLong
@@ -152,7 +157,8 @@ object InstanceRegistry extends InstanceJsonSupport with AppLogging
       val request = HttpRequest(method = HttpMethods.GET, configuration.instanceRegistryUri +
         s"/matchingInstance?Id=${configuration.instanceId.getOrElse(-1)}&ComponentType=ElasticSearch")
 
-      Await.result(Http(system).singleRequest(request) map {response =>
+      Await.result(Http(system).singleRequest(request
+        .withHeaders(RawHeader("Authorization", s"Bearer ${AuthProvider.generateJwt()(configuration)}"))) map {response =>
         response.status match {
           case StatusCodes.OK =>
             val instanceString : String = Await.result(response.entity.dataBytes.runFold(ByteString(""))(_ ++ _).map(_.utf8String), 5 seconds)
@@ -192,7 +198,8 @@ object InstanceRegistry extends InstanceJsonSupport with AppLogging
           configuration.instanceRegistryUri +
             s"/matchingResult?CallerId=${configuration.instanceId.getOrElse(-1)}&MatchedInstanceId=$idToPost&MatchingSuccessful=$isElasticSearchReachable")
 
-        Await.result(Http(system).singleRequest(request) map {response =>
+        Await.result(Http(system).singleRequest(request
+          .withHeaders(RawHeader("Authorization", s"Bearer ${AuthProvider.generateJwt()(configuration)}"))) map {response =>
           if(response.status == StatusCodes.OK){
             log.info("Successfully posted matching result to Instance Registry.")
             Success()
@@ -220,7 +227,8 @@ object InstanceRegistry extends InstanceJsonSupport with AppLogging
 
       val request = HttpRequest(method = HttpMethods.POST, configuration.instanceRegistryUri + s"/deregister?Id=$id")
 
-      Await.result(Http(system).singleRequest(request) map {response =>
+      Await.result(Http(system).singleRequest(request
+        .withHeaders(RawHeader("Authorization", s"Bearer ${AuthProvider.generateJwt()(configuration)}"))) map {response =>
         if(response.status == StatusCodes.OK){
           log.info("Successfully deregistered from Instance Registry.")
           Success()
@@ -238,9 +246,11 @@ object InstanceRegistry extends InstanceJsonSupport with AppLogging
     }
   }
 
-  def postInstance(instance : Instance, uri: String) () : Future[HttpResponse] = {
+  def postInstance(instance : Instance, uri: String) (implicit configuration: Configuration) : Future[HttpResponse] = {
     Try(HttpRequest(method = HttpMethods.POST, uri = uri, entity = instance.toJson(instanceFormat).toString())) match {
-      case Success(request) => Http(system).singleRequest(request)
+      //use generic name for startup, no id present at this point
+      case Success(request) => Http(system).singleRequest(request
+        .withHeaders(RawHeader("Authorization", s"Bearer ${AuthProvider.generateJwt(useGenericName = true)(configuration)}")))
       case Failure(dx) =>
         log.warning(s"Failed to deregister to Instance Registry, exception: $dx")
         Future.failed(dx)