Merge branch 'master' into actor-handle

akka-http/src/main/resources/META-INF/services/javax.ws.rs.ext.MessageBodyWriter_FIXME

@@ -0,0 +1 @@
+se.scalablesolutions.akka.rest.ListWriter

akka-http/src/main/scala/ActorComponentProvider.scala

@@ -0,0 +1,29 @@
+/**
+ * Copyright (C) 2009-2010 Scalable Solutions AB <http://scalablesolutions.se>
+ */
+
+package se.scalablesolutions.akka.rest
+
+import com.sun.jersey.core.spi.component.ComponentScope
+import com.sun.jersey.core.spi.component.ioc.IoCFullyManagedComponentProvider
+
+import se.scalablesolutions.akka.config.Configurator
+import se.scalablesolutions.akka.util.Logging
+import se.scalablesolutions.akka.actor.Actor
+
+class ActorComponentProvider(val clazz: Class[_], val configurators: List[Configurator])
+    extends IoCFullyManagedComponentProvider with Logging {
+   
+  override def getScope = ComponentScope.Singleton
+
+  override def getInstance: AnyRef = {
+    val instances = for {
+      conf <- configurators
+      if conf.isDefined(clazz)
+      instance <- conf.getInstance(clazz)
+    } yield instance
+    if (instances.isEmpty) throw new IllegalArgumentException(
+      "No Actor or Active Object for class [" +  clazz + "] could be found.\nMake sure you have defined and configured the class as an Active Object or Actor in a supervisor hierarchy.")
+    else instances.head.asInstanceOf[AnyRef]
+  }
+}

akka-http/src/main/scala/ActorComponentProviderFactory.scala

@@ -0,0 +1,20 @@
+/**
+ * Copyright (C) 2009-2010 Scalable Solutions AB <http://scalablesolutions.se>
+ */
+
+package se.scalablesolutions.akka.rest
+
+import com.sun.jersey.core.spi.component.ioc.{IoCComponentProvider,IoCComponentProviderFactory}
+import com.sun.jersey.core.spi.component.{ComponentContext}
+
+import se.scalablesolutions.akka.config.Configurator
+import se.scalablesolutions.akka.util.Logging
+
+class ActorComponentProviderFactory(val configurators: List[Configurator])
+extends IoCComponentProviderFactory with Logging {
+  override def getComponentProvider(clazz: Class[_]): IoCComponentProvider = getComponentProvider(null, clazz)
+
+  override def getComponentProvider(context: ComponentContext, clazz: Class[_]): IoCComponentProvider = {
+    configurators.find(_.isDefined(clazz)).map(_ => new ActorComponentProvider(clazz, configurators)).getOrElse(null)
+  }
+}

akka-http/src/main/scala/AkkaBroadcaster.scala

@@ -0,0 +1,23 @@
+/**
+ * Copyright (C) 2009-2010 Scalable Solutions AB <http://scalablesolutions.se>
+ */
+
+package se.scalablesolutions.akka.comet
+
+import org.atmosphere.cpr.{AtmosphereResourceEvent, AtmosphereResource}
+import se.scalablesolutions.akka.actor.Actor._
+
+class AkkaBroadcaster extends org.atmosphere.jersey.JerseyBroadcaster {
+  name = classOf[AkkaBroadcaster].getName
+  
+  val caster = actor { case f : Function0[_] => f() }
+
+  override def destroy {
+    super.destroy
+    caster.stop
+  }
+  
+  protected override def broadcast(r :  AtmosphereResource[_,_], e : AtmosphereResourceEvent[_,_]) = {
+    caster ! (() => super.broadcast(r,e))
+  }
+}

akka-http/src/main/scala/AkkaClusterBroadcastFilter.scala

@@ -0,0 +1,54 @@
+/**
+ * Copyright (C) 2009-2010 Scalable Solutions AB <http://scalablesolutions.se>
+ */
+ 
+package se.scalablesolutions.akka.comet
+
+import se.scalablesolutions.akka.actor.Actor
+import se.scalablesolutions.akka.remote.Cluster
+import scala.reflect.BeanProperty
+import org.atmosphere.cpr.{BroadcastFilter, ClusterBroadcastFilter, Broadcaster}
+
+sealed trait ClusterCometMessageType
+case class ClusterCometBroadcast(name: String, msg: AnyRef) extends ClusterCometMessageType
+
+/**
+ * Enables explicit clustering of Atmosphere (Comet) resources
+ * Annotate the endpoint which has the @Broadcast annotation with
+ * @org.atmosphere.annotation.Cluster(Array(classOf[AkkClusterBroadcastFilter])){ name = "someUniqueName" }
+ * that's all folks!
+ * Note: In the future, clustering comet will be transparent
+ */
+
+class AkkaClusterBroadcastFilter extends Actor with ClusterBroadcastFilter[AnyRef] {
+  @BeanProperty var clusterName = ""
+  @BeanProperty var broadcaster : Broadcaster = null
+
+  override def init : Unit = ()
+
+  /** Stops the actor */
+  def destroy : Unit = stop
+
+  /**
+   * Relays all non ClusterCometBroadcast messages to the other AkkaClusterBroadcastFilters in the cluster
+   * ClusterCometBroadcasts are not broadcasted because they originate from the cluster,
+   * otherwise we'd start a chain reaction.
+   */
+  def filter(o : AnyRef) = new BroadcastFilter.BroadcastAction(o match {
+    case ClusterCometBroadcast(_,m) => m   //Do not re-broadcast, just unbox and pass along
+ 
+    case m : AnyRef => {                   //Relay message to the cluster and pass along
+      Cluster.relayMessage(classOf[AkkaClusterBroadcastFilter],ClusterCometBroadcast(clusterName,m))
+      m
+    }
+  })
+
+  def receive = { 
+    //Only handle messages intended for this particular instance
+    case b@ClusterCometBroadcast(c,_) if (c == clusterName) && (broadcaster ne null) => broadcaster broadcast b
+    case _ =>
+  }
+
+  //Since this class is instantiated by Atmosphere, we need to make sure it's started
+  start
+}

akka-http/src/main/scala/AkkaCometServlet.scala

@@ -0,0 +1,81 @@
+/**
+ * Copyright (C) 2009-2010 Scalable Solutions AB <http://scalablesolutions.se>
+ */
+
+package se.scalablesolutions.akka.comet
+
+import se.scalablesolutions.akka.util.Logging
+import se.scalablesolutions.akka.rest.{AkkaServlet => RestServlet}
+
+import java.util.{List => JList}
+import javax.servlet.ServletConfig
+import javax.servlet.http.{HttpServletRequest, HttpServletResponse}
+
+import org.atmosphere.container.GrizzlyCometSupport
+import org.atmosphere.cpr.{AtmosphereServlet, AtmosphereServletProcessor, AtmosphereResource, AtmosphereResourceEvent,CometSupport,CometSupportResolver,DefaultCometSupportResolver}
+import org.atmosphere.handler.{ReflectorServletProcessor, AbstractReflectorAtmosphereHandler}
+
+/**
+ * Akka's Comet servlet to be used when deploying actors exposed as Comet (and REST) services in a
+ * standard servlet container, e.g. not using the Akka Kernel.
+ * <p/>
+ * Used by the Akka Kernel to bootstrap REST and Comet.
+ */
+class AkkaServlet extends org.atmosphere.cpr.AtmosphereServlet with Logging {
+  val servlet = new RestServlet with AtmosphereServletProcessor {
+
+    //Delegate to implement the behavior for AtmosphereHandler
+    private val handler = new AbstractReflectorAtmosphereHandler {
+      override def onRequest(event: AtmosphereResource[HttpServletRequest, HttpServletResponse]) {
+        if (event ne null) {
+          event.getRequest.setAttribute(ReflectorServletProcessor.ATMOSPHERE_RESOURCE, event)
+          event.getRequest.setAttribute(ReflectorServletProcessor.ATMOSPHERE_HANDLER, this)
+          service(event.getRequest, event.getResponse)
+        }
+      }
+    }
+
+    override def onStateChange(event: AtmosphereResourceEvent[HttpServletRequest, HttpServletResponse]) {
+      if (event ne null) handler onStateChange event
+    }
+
+    override def onRequest(resource: AtmosphereResource[HttpServletRequest, HttpServletResponse]) {
+      handler onRequest resource
+    }
+  }
+
+  /**
+   * We override this to avoid Atmosphere looking for it's atmosphere.xml file
+   * Instead we specify what semantics we want in code.
+   */
+  override def loadConfiguration(sc: ServletConfig) {
+    config = new AtmosphereConfig { supportSession = false }
+    setDefaultBroadcasterClassName(classOf[AkkaBroadcaster].getName)
+    atmosphereHandlers.put("/*", new AtmosphereServlet.AtmosphereHandlerWrapper(servlet, new AkkaBroadcaster))
+  }
+
+   /**
+    * This method is overridden because Akka Kernel is bundles with Grizzly, so if we deploy the Kernel in another container,
+    * we need to handle that.
+    */
+   override def createCometSupportResolver() : CometSupportResolver = {
+      import scala.collection.JavaConversions._
+ 
+      new DefaultCometSupportResolver(config) {
+         type CS = CometSupport[_ <: AtmosphereResource[_,_]]
+         override def resolveMultipleNativeSupportConflict(available : JList[Class[_ <: CS]]) : CS = {
+             available.filter(_ != classOf[GrizzlyCometSupport]).toList match {
+                 case Nil => new GrizzlyCometSupport(config)
+                 case x :: Nil => newCometSupport(x.asInstanceOf[Class[_ <: CS]])
+                 case _ => super.resolveMultipleNativeSupportConflict(available)
+             }
+        }
+ 
+        override def resolve(useNativeIfPossible : Boolean, useBlockingAsDefault : Boolean) : CS = {
+           val predef = config.getInitParameter("cometSupport")
+           if (testClassExists(predef)) newCometSupport(predef)
+           else super.resolve(useNativeIfPossible, useBlockingAsDefault)
+        }
+      }
+  }
+}

akka-http/src/main/scala/AkkaLoader.scala

@@ -0,0 +1,61 @@
+/**
+ * Copyright (C) 2009-2010 Scalable Solutions AB <http://scalablesolutions.se>
+ */
+
+package se.scalablesolutions.akka.servlet
+
+import se.scalablesolutions.akka.config.Config
+import se.scalablesolutions.akka.util.{Logging, Bootable}
+
+/*
+ * This class is responsible for booting up a stack of bundles and then shutting them down
+ */
+class AkkaLoader extends Logging {
+  @volatile private var hasBooted = false
+
+  @volatile private var _bundles: Option[Bootable] = None
+  
+  def bundles = _bundles;
+
+  /*
+   * Boot initializes the specified bundles
+   */
+  def boot(withBanner: Boolean, b : Bootable): Unit = synchronized {
+    if (!hasBooted) {
+      if (withBanner) printBanner
+      log.info("Starting Akka...")
+      b.onLoad
+      Thread.currentThread.setContextClassLoader(getClass.getClassLoader)
+      log.info("Akka started successfully")
+      hasBooted = true
+      _bundles = Some(b)
+    }
+  }
+
+  /*
+   * Shutdown, well, shuts down the bundles used in boot
+   */
+  def shutdown = synchronized {
+    if (hasBooted) {
+      log.info("Shutting down Akka...")
+      _bundles.foreach(_.onUnload)
+      _bundles = None
+      log.info("Akka succesfully shut down")
+    }
+  }
+
+  private def printBanner = {
+    log.info(
+"""
+==============================
+          __    __
+ _____  |  | _|  | _______
+ \__  \ |  |/ /  |/ /\__  \
+  / __ \|    <|    <  / __ \_
+ (____  /__|_ \__|_ \(____  /
+      \/     \/    \/     \/
+""")
+    log.info(" Running version %s", Config.VERSION)
+    log.info("==============================")
+  }
+}

akka-http/src/main/scala/AkkaServlet.scala

@@ -0,0 +1,33 @@
+/**
+ * Copyright (C) 2009-2010 Scalable Solutions AB <http://scalablesolutions.se>
+ */
+
+package se.scalablesolutions.akka.rest
+
+import se.scalablesolutions.akka.config.ConfiguratorRepository
+import se.scalablesolutions.akka.config.Config.config
+
+import com.sun.jersey.api.core.ResourceConfig
+import com.sun.jersey.spi.container.servlet.ServletContainer
+import com.sun.jersey.spi.container.WebApplication
+
+/**
+ * Akka's servlet to be used when deploying actors exposed as REST services in a standard servlet container,
+ * e.g. not using the Akka Kernel.
+ *
+ * @author <a href="http://jonasboner.com">Jonas Bon&#233;r</a>
+ */
+class AkkaServlet extends ServletContainer {
+  import scala.collection.JavaConversions._
+
+  override def initiate(resourceConfig: ResourceConfig, webApplication: WebApplication) = {
+    val configurators = ConfiguratorRepository.getConfigurators
+
+    resourceConfig.getClasses.addAll(configurators.flatMap(_.getComponentInterfaces))
+    resourceConfig.getProperties.put(
+      "com.sun.jersey.spi.container.ResourceFilters",
+      config.getList("akka.rest.filters").mkString(","))
+
+    webApplication.initiate(resourceConfig, new ActorComponentProviderFactory(configurators))
+  }
+}

akka-http/src/main/scala/Initializer.scala

@@ -0,0 +1,34 @@
+/**
+ * Copyright (C) 2009-2010 Scalable Solutions AB <http://scalablesolutions.se>
+ */
+
+package se.scalablesolutions.akka.servlet
+
+import se.scalablesolutions.akka.remote.BootableRemoteActorService
+import se.scalablesolutions.akka.actor.BootableActorLoaderService
+import se.scalablesolutions.akka.camel.service.CamelService
+import se.scalablesolutions.akka.config.Config
+import se.scalablesolutions.akka.util.{Logging, Bootable}
+
+import javax.servlet.{ServletContextListener, ServletContextEvent}
+ 
+ /**
+  * This class can be added to web.xml mappings as a listener to start and shutdown Akka.
+  *
+  *<web-app>
+  * ...
+  *  <listener>
+  *    <listener-class>se.scalablesolutions.akka.servlet.Initializer</listener-class>
+  *  </listener>
+  * ...
+  *</web-app>
+  */ 
+class Initializer extends ServletContextListener {
+   lazy val loader = new AkkaLoader
+     
+   def contextDestroyed(e: ServletContextEvent): Unit = 
+     loader.shutdown
+     
+   def contextInitialized(e: ServletContextEvent): Unit = 
+     loader.boot(true, new BootableActorLoaderService with BootableRemoteActorService with CamelService)
+ }

akka-http/src/main/scala/ListWriter.scala

@@ -0,0 +1,38 @@
+/**
+ * Copyright (C) 2009-2010 Scalable Solutions AB <http://scalablesolutions.se>
+ */
+package se.scalablesolutions.akka.rest
+
+import java.io.OutputStream
+import se.scalablesolutions.akka.serialization.Serializer
+import javax.ws.rs.core.{MultivaluedMap, MediaType}
+import javax.ws.rs.ext.{MessageBodyWriter, Provider}
+import javax.ws.rs.Produces
+
+/**
+ * writes Lists of JSON serializable objects
+ */
+@Provider
+@Produces(Array("application/json"))
+class ListWriter extends MessageBodyWriter[List[_]] {
+
+  def isWriteable(aClass: Class[_], aType: java.lang.reflect.Type, annotations: Array[java.lang.annotation.Annotation], mediaType: MediaType) = {
+    classOf[List[_]].isAssignableFrom(aClass) || aClass == ::.getClass
+  }
+
+  def getSize(list: List[_], aClass: Class[_], aType: java.lang.reflect.Type, annotations: Array[java.lang.annotation.Annotation], mediaType: MediaType) = -1L
+
+  def writeTo(list: List[_],
+              aClass: Class[_],
+              aType: java.lang.reflect.Type,
+              annotations: Array[java.lang.annotation.Annotation],
+              mediaType: MediaType,
+              stringObjectMultivaluedMap: MultivaluedMap[String, Object],
+              outputStream: OutputStream) : Unit = {
+      if (list.isEmpty)
+        outputStream.write(" ".getBytes)
+      else
+        outputStream.write(Serializer.ScalaJSON.out(list))
+  }
+
+}

akka-http/src/main/scala/Security.scala

@@ -0,0 +1,470 @@
+/*
+ * Copyright 2007-2008 WorldWide Conferencing, LLC
+ *
+ * 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.
+ */
+
+/*
+ * AKKA AAS (Authentication and Authorization Service)
+ * Rework of lift's (www.liftweb.com) HTTP Authentication module
+ * All cred to the Lift team (www.liftweb.com), especially David Pollak and Tim Perrett
+ */
+
+package se.scalablesolutions.akka.security
+
+import se.scalablesolutions.akka.actor.{Scheduler, Actor, ActorID, ActorRegistry}
+import se.scalablesolutions.akka.util.Logging
+import se.scalablesolutions.akka.config.Config
+
+import com.sun.jersey.api.model.AbstractMethod
+import com.sun.jersey.spi.container.{ResourceFilterFactory, ContainerRequest, ContainerRequestFilter, ContainerResponse, ContainerResponseFilter, ResourceFilter}
+import com.sun.jersey.core.util.Base64
+
+import javax.ws.rs.core.{SecurityContext, Context, Response}
+import javax.ws.rs.WebApplicationException
+import javax.annotation.security.{DenyAll, PermitAll, RolesAllowed}
+import java.security.Principal
+import java.util.concurrent.TimeUnit
+
+import net.liftweb.util.{SecurityHelpers, StringHelpers, IoHelpers}
+
+object Enc extends SecurityHelpers with StringHelpers with IoHelpers
+
+case object OK
+
+/**
+ * Authenticate represents a message to authenticate a request
+ */
+case class Authenticate(val req: ContainerRequest, val rolesAllowed: List[String])
+
+/**
+ * User info represents a sign-on with associated credentials/roles
+ */
+case class UserInfo(val username: String, val password: String, val roles: List[String])
+
+trait Credentials
+
+case class BasicCredentials(username: String, password: String) extends Credentials
+
+case class DigestCredentials(method: String,
+                             userName: String,
+                             realm: String,
+                             nonce: String,
+                             uri: String,
+                             qop: String,
+                             nc: String,
+                             cnonce: String,
+                             response: String,
+                             opaque: String) extends Credentials
+
+case class SpnegoCredentials(token: Array[Byte]) extends Credentials
+
+/**
+ * Jersey Filter for invocation intercept and authorization/authentication
+ */
+class AkkaSecurityFilterFactory extends ResourceFilterFactory with Logging {
+  class Filter(actor: ActorID, rolesAllowed: Option[List[String]])
+      extends ResourceFilter with ContainerRequestFilter with Logging {
+
+    override def getRequestFilter: ContainerRequestFilter = this
+
+    override def getResponseFilter: ContainerResponseFilter = null
+
+    /**
+     * Here's where the magic happens. The request is authenticated by
+     * sending a request for authentication to the configured authenticator actor
+     */
+    override def filter(request: ContainerRequest): ContainerRequest =
+      rolesAllowed match {
+        case Some(roles) => {
+          val result : Option[AnyRef] = authenticator !! Authenticate(request, roles)
+          result match {
+            case Some(OK) => request
+            case Some(r) if r.isInstanceOf[Response] =>
+              throw new WebApplicationException(r.asInstanceOf[Response])
+            case None => throw new WebApplicationException(408)
+            case unknown => {
+              log.warning("Authenticator replied with unexpected result [%s]", unknown);
+              throw new WebApplicationException(Response.Status.INTERNAL_SERVER_ERROR)
+            }
+          }
+        }
+        case None => throw new WebApplicationException(Response.Status.FORBIDDEN)
+      }
+  }
+
+  lazy val authenticatorFQN =
+    Config.config.getString("akka.rest.authenticator")
+    .getOrElse(throw new IllegalStateException("akka.rest.authenticator"))
+
+  /**
+   * Currently we always take the first, since there usually should be at most one authentication actor, but a round-robin
+   * strategy could be implemented in the future
+   */
+  def authenticator: ActorID = ActorRegistry.actorsFor(authenticatorFQN).head
+
+  def mkFilter(roles: Option[List[String]]): java.util.List[ResourceFilter] =
+    java.util.Collections.singletonList(new Filter(authenticator, roles))
+
+  /**
+   * The create method is invoked for each resource, and we look for javax.annotation.security annotations
+   * and create the appropriate Filter configurations for each.
+   */
+  override def create(am: AbstractMethod): java.util.List[ResourceFilter] = {
+
+    //DenyAll takes precedence
+    if (am.isAnnotationPresent(classOf[DenyAll]))
+      return mkFilter(None)
+
+    //Method-level RolesAllowed takes precedence
+    val ra = am.getAnnotation(classOf[RolesAllowed])
+
+    if (ra ne null)
+      return mkFilter(Some(ra.value.toList))
+
+    //PermitAll takes precedence over resource-level RolesAllowed annotation
+    if (am.isAnnotationPresent(classOf[PermitAll]))
+      return null;
+
+    //Last but not least, the resource-level RolesAllowed
+    val cra = am.getResource.getAnnotation(classOf[RolesAllowed])
+    if (cra ne null)
+      return mkFilter(Some(cra.value.toList))
+
+    return null;
+  }
+}
+
+/**
+ * AuthenticationActor is the super-trait for actors doing Http authentication
+ * It defines the common ground and the flow of execution
+ */
+trait AuthenticationActor[C <: Credentials] extends Actor {
+  type Req = ContainerRequest
+
+  //What realm does the authentication use?
+  def realm: String
+
+  //Creates a response to signal unauthorized
+  def unauthorized: Response
+
+  //Used to extract information from the request, returns None if no credentials found
+  def extractCredentials(r: Req): Option[C]
+
+  //returns None is unverified
+  def verify(c: Option[C]): Option[UserInfo]
+
+  //Contruct a new SecurityContext from the supplied parameters
+  def mkSecurityContext(r: Req, user: UserInfo): SecurityContext
+
+  //This is the default security context factory
+  def mkDefaultSecurityContext(r: Req, u: UserInfo, scheme: String): SecurityContext = {
+    val n = u.username
+    val p = new Principal {def getName = n}
+
+    new SecurityContext {
+      def getAuthenticationScheme = scheme
+      def getUserPrincipal = p
+      def isSecure = r.isSecure
+      def isUserInRole(role: String) = u.roles.exists(_ == role)
+    }
+  }
+
+  /**
+   * Responsible for the execution flow of authentication
+   *
+   * Credentials are extracted and verified from the request,
+   * and a se3curity context is created for the ContainerRequest
+   * this should ensure good integration with current Jersey security
+   */
+  protected val authenticate: PartialFunction[Any, Unit] = {
+    case Authenticate(req, roles) => {
+      verify(extractCredentials(req)) match {
+        case Some(u: UserInfo) => {
+          req.setSecurityContext(mkSecurityContext(req, u))
+          if (roles.exists(req.isUserInRole(_))) reply(OK)
+          else reply(Response.status(Response.Status.FORBIDDEN).build)
+        }
+        case _ => reply(unauthorized)
+      }
+    }
+  }
+
+  def receive = authenticate
+
+  //returns the string value of the "Authorization"-header of the request
+  def auth(r: Req) = r.getHeaderValue("Authorization")
+
+  //Turns the aforementioned header value into an option
+  def authOption(r: Req): Option[String] = {
+    val a = auth(r)
+    if (a != null && a.length > 0) Some(a) else None
+  }
+}
+
+/**
+ * This trait implements the logic for Http Basic authentication
+ * mix this trait into a class to create an authenticator
+ * Don't forget to set the authenticator FQN in the rest-part of the akka config
+ */
+trait BasicAuthenticationActor extends AuthenticationActor[BasicCredentials] {
+  override def unauthorized =
+    Response.status(401).header("WWW-Authenticate", "Basic realm=\"" + realm + "\"").build
+
+  override def extractCredentials(r: Req): Option[BasicCredentials] = {
+    val Authorization = """(.*):(.*)""".r
+
+    authOption(r) match {
+      case Some(token) => {
+        val authResponse = new String(Base64.decode(token.substring(6).getBytes))
+        authResponse match {
+          case Authorization(username, password) => Some(BasicCredentials(username, password))
+          case _ => None
+        }
+      }
+      case _ => None
+    }
+  }
+
+  override def mkSecurityContext(r: Req, u: UserInfo): SecurityContext =
+    mkDefaultSecurityContext(r, u, SecurityContext.BASIC_AUTH)
+}
+
+/**
+ * This trait implements the logic for Http Digest authentication mix this trait into a
+ * class to create an authenticator. Don't forget to set the authenticator FQN in the
+ * rest-part of the akka config
+ */
+trait DigestAuthenticationActor extends AuthenticationActor[DigestCredentials] with Logging {
+  import Enc._
+
+  private object InvalidateNonces
+
+  //Holds the generated nonces for the specified validity period
+  val nonceMap = mkNonceMap
+
+  //Discards old nonces
+  protected val invalidateNonces: PartialFunction[Any, Unit] = {
+    case InvalidateNonces =>
+      val ts = System.currentTimeMillis
+      nonceMap.filter(tuple => (ts - tuple._2) < nonceValidityPeriod)
+    case unknown =>
+      log.error("Don't know what to do with: ", unknown)
+  }
+
+  //Schedule the invalidation of nonces
+  Scheduler.schedule(self, InvalidateNonces, noncePurgeInterval, noncePurgeInterval, TimeUnit.MILLISECONDS)
+
+  //authenticate or invalidate nonces
+  override def receive = authenticate orElse invalidateNonces
+
+  override def unauthorized: Response = {
+    val nonce = randomString(64)
+    nonceMap.put(nonce, System.currentTimeMillis)
+    unauthorized(nonce, "auth", randomString(64))
+  }
+
+  def unauthorized(nonce: String, qop: String, opaque: String): Response = {
+    Response.status(401).header(
+      "WWW-Authenticate",
+      "Digest realm=\"" + realm + "\", " +
+      "qop=\"" + qop + "\", " +
+      "nonce=\"" + nonce + "\", " +
+      "opaque=\"" + opaque + "\"").build
+  }
+
+  //Tests wether the specified credentials are valid
+  def validate(auth: DigestCredentials, user: UserInfo): Boolean = {
+    def h(s: String) = hexEncode(md5(s.getBytes("UTF-8")))
+
+    val ha1 = h(auth.userName + ":" + auth.realm + ":" + user.password)
+    val ha2 = h(auth.method + ":" + auth.uri)
+
+    val response = h(
+      ha1 + ":" + auth.nonce + ":" +
+      auth.nc + ":" + auth.cnonce + ":" +
+      auth.qop + ":" + ha2)
+
+    (response == auth.response) && (nonceMap.getOrElse(auth.nonce, -1) != -1)
+  }
+
+  override def verify(odc: Option[DigestCredentials]): Option[UserInfo] = odc match {
+    case Some(dc) => {
+      userInfo(dc.userName) match {
+        case Some(u) if validate(dc, u) =>
+          nonceMap.get(dc.nonce).map(t => (System.currentTimeMillis - t) < nonceValidityPeriod).map(_ => u)
+        case _ => None
+      }
+    }
+    case _ => None
+  }
+
+  override def extractCredentials(r: Req): Option[DigestCredentials] = {
+    authOption(r).map(s => {
+      val ? = splitNameValuePairs(s.substring(7, s.length))
+      DigestCredentials(r.getMethod.toUpperCase,
+        ?("username"), ?("realm"), ?("nonce"),
+        ?("uri"), ?("qop"), ?("nc"),
+        ?("cnonce"), ?("response"), ?("opaque"))
+    })
+  }
+
+  override def mkSecurityContext(r: Req, u: UserInfo): SecurityContext =
+    mkDefaultSecurityContext(r, u, SecurityContext.DIGEST_AUTH)
+
+  //Mandatory overrides
+  def userInfo(username: String): Option[UserInfo]
+
+  def mkNonceMap: scala.collection.mutable.Map[String, Long]
+
+  //Optional overrides
+  def nonceValidityPeriod = 60 * 1000 //ms
+  def noncePurgeInterval = 2 * 60 * 1000 //ms
+}
+
+import java.security.Principal
+import java.security.PrivilegedActionException
+import java.security.PrivilegedExceptionAction
+
+import javax.security.auth.login.AppConfigurationEntry
+import javax.security.auth.login.Configuration
+import javax.security.auth.login.LoginContext
+import javax.security.auth.Subject
+import javax.security.auth.kerberos.KerberosPrincipal
+
+import org.ietf.jgss.GSSContext
+import org.ietf.jgss.GSSCredential
+import org.ietf.jgss.GSSManager
+
+trait SpnegoAuthenticationActor extends AuthenticationActor[SpnegoCredentials] with Logging {
+  override def unauthorized =
+    Response.status(401).header("WWW-Authenticate", "Negotiate").build
+
+  // for some reason the jersey Base64 class does not work with kerberos
+  // but the commons Base64 does
+  import org.apache.commons.codec.binary.Base64
+  override def extractCredentials(r: Req): Option[SpnegoCredentials] = {
+    val AuthHeader = """Negotiate\s(.*)""".r
+
+    authOption(r) match {
+      case Some(AuthHeader(token)) =>
+        Some(SpnegoCredentials(Base64.decodeBase64(token.trim.getBytes)))
+      case _ => None
+    }
+  }
+
+
+  override def verify(odc: Option[SpnegoCredentials]): Option[UserInfo] = odc match {
+    case Some(dc) => {
+      try {
+        val principal = Subject.doAs(this.serviceSubject, new KerberosValidateAction(dc.token));
+        val user = stripRealmFrom(principal)
+        Some(UserInfo(user, null, rolesFor(user)))
+      } catch {
+        case e: PrivilegedActionException => {
+          log.error(e, "Action not allowed")
+          return None
+        }
+      }
+    }
+    case _ => None
+  }
+
+  override def mkSecurityContext(r: Req, u: UserInfo): SecurityContext =
+    mkDefaultSecurityContext(r, u, SecurityContext.CLIENT_CERT_AUTH) // the security context does not know about spnego/kerberos
+  // not sure whether to use a constant from the security context or something like "SPNEGO/Kerberos"
+
+  /**
+   * returns the roles for the given user
+   */
+  def rolesFor(user: String): List[String]
+
+  // Kerberos
+
+  /**
+   * strips the realm from a kerberos principal name, returning only the user part
+   */
+  private def stripRealmFrom(principal: String): String = principal.split("@")(0)
+
+  /**
+   * principal name for the HTTP kerberos service, i.e HTTP/  { server } @  { realm }
+   */
+  lazy val servicePrincipal = Config.config.getString("akka.rest.kerberos.servicePrincipal").getOrElse(throw new IllegalStateException("akka.rest.kerberos.servicePrincipal"))
+
+  /**
+   * keytab location with credentials for the service principal
+   */
+  lazy val keyTabLocation = Config.config.getString("akka.rest.kerberos.keyTabLocation").getOrElse(throw new IllegalStateException("akka.rest.kerberos.keyTabLocation"))
+
+  lazy val kerberosDebug = Config.config.getString("akka.rest.kerberos.kerberosDebug").getOrElse("false")
+
+  /**
+   * is not used by this authenticator, so accept an empty value
+   */
+  lazy val realm = Config.config.getString("akka.rest.kerberos.realm").getOrElse("")
+
+  /**
+   * verify the kerberos token from a client with the server
+   */
+  class KerberosValidateAction(kerberosTicket: Array[Byte]) extends PrivilegedExceptionAction[String] {
+    def run = {
+      val context = GSSManager.getInstance().createContext(null.asInstanceOf[GSSCredential])
+      context.acceptSecContext(kerberosTicket, 0, kerberosTicket.length)
+      val user = context.getSrcName().toString()
+      context.dispose()
+      user
+    }
+  }
+
+  // service principal login to kerberos on startup
+
+  val serviceSubject = servicePrincipalLogin
+
+  /**
+   * acquire an initial ticket from the kerberos server for the HTTP service
+   */
+  def servicePrincipalLogin = {
+    val loginConfig = new LoginConfig(
+      new java.net.URL(this.keyTabLocation).toExternalForm(),
+      this.servicePrincipal,
+      this.kerberosDebug)
+    val princ = new java.util.HashSet[Principal](1)
+    princ.add(new KerberosPrincipal(this.servicePrincipal))
+    val sub = new Subject(false, princ, new java.util.HashSet[Object], new java.util.HashSet[Object])
+    val lc = new LoginContext("", sub, null, loginConfig)
+    lc.login()
+    lc.getSubject()
+  }
+
+  /**
+   * this class simulates a login-config.xml
+   */
+  class LoginConfig(keyTabLocation: String, servicePrincipal: String, debug: String) extends Configuration {
+    override def getAppConfigurationEntry(name: String): Array[AppConfigurationEntry] = {
+      val options = new java.util.HashMap[String, String]
+      options.put("useKeyTab", "true")
+      options.put("keyTab", this.keyTabLocation)
+      options.put("principal", this.servicePrincipal)
+      options.put("storeKey", "true")
+      options.put("doNotPrompt", "true")
+      options.put("isInitiator", "true")
+      options.put("debug", debug)
+
+      Array(new AppConfigurationEntry(
+        "com.sun.security.auth.module.Krb5LoginModule",
+        AppConfigurationEntry.LoginModuleControlFlag.REQUIRED,
+        options))
+    }
+  }
+
+}

akka-http/src/test/scala/AllTest.scala

@@ -0,0 +1,15 @@
+package se.scalablesolutions.akka.security
+
+import junit.framework.Test
+import junit.framework.TestCase
+import junit.framework.TestSuite
+
+object AllTest extends TestCase {
+  def suite(): Test = {
+    val suite = new TestSuite("All Scala tests")
+    suite.addTestSuite(classOf[BasicAuthenticatorSpec])
+    suite
+  }
+
+  def main(args: Array[String]) = junit.textui.TestRunner.run(suite)
+}

akka-http/src/test/scala/SecuritySpec.scala

@@ -0,0 +1,79 @@
+/**
+ * Copyright (C) 2009-2010 Scalable Solutions AB <http://scalablesolutions.se>
+ */
+
+package se.scalablesolutions.akka.security
+
+import se.scalablesolutions.akka.config.ScalaConfig._
+import se.scalablesolutions.akka.actor.Actor._
+
+import org.scalatest.Suite
+import org.scalatest.junit.JUnitSuite
+import org.scalatest.matchers.MustMatchers
+import org.scalatest.mock.MockitoSugar
+import org.mockito.Mockito._
+import org.mockito.Matchers._
+import org.junit.{Before, After, Test}
+
+import javax.ws.rs.core.{SecurityContext, Context, Response}
+import com.sun.jersey.spi.container.{ResourceFilterFactory, ContainerRequest, ContainerRequestFilter, ContainerResponse, ContainerResponseFilter, ResourceFilter}
+import com.sun.jersey.core.util.Base64
+
+object BasicAuthenticatorSpec {
+  class BasicAuthenticator extends BasicAuthenticationActor {
+    def verify(odc: Option[BasicCredentials]): Option[UserInfo] = odc match {
+      case Some(dc) => Some(UserInfo("foo", "bar", "ninja" :: "chef" :: Nil))
+      case _ => None
+    }
+    override def realm = "test"
+  }
+}
+
+class BasicAuthenticatorSpec extends junit.framework.TestCase
+    with Suite with MockitoSugar with MustMatchers {
+  import BasicAuthenticatorSpec._
+  
+  val authenticator = newActor[BasicAuthenticator]
+  authenticator.start
+
+  @Test def testChallenge = {
+    val req = mock[ContainerRequest]
+
+    val result: Response = (authenticator !! (Authenticate(req, List("foo")), 10000)).get
+
+    // the actor replies with a challenge for the browser
+    result.getStatus must equal(Response.Status.UNAUTHORIZED.getStatusCode)
+    result.getMetadata.get("WWW-Authenticate").get(0).toString must startWith("Basic")
+  }
+
+  @Test def testAuthenticationSuccess = {
+    val req = mock[ContainerRequest]
+    // fake a basic auth header -> this will authenticate the user
+    when(req.getHeaderValue("Authorization")).thenReturn("Basic " + new String(Base64.encode("foo:bar")))
+
+    // fake a request authorization -> this will authorize the user
+    when(req.isUserInRole("chef")).thenReturn(true)
+
+    val result: AnyRef = (authenticator !! (Authenticate(req, List("chef")), 10000)).get
+
+    result must be(OK)
+    // the authenticator must have set a security context 
+    verify(req).setSecurityContext(any[SecurityContext])
+  }
+
+  @Test def testUnauthorized = {
+    val req = mock[ContainerRequest]
+    
+    // fake a basic auth header -> this will authenticate the user
+    when(req.getHeaderValue("Authorization")).thenReturn("Basic " + new String(Base64.encode("foo:bar")))
+    when(req.isUserInRole("chef")).thenReturn(false) // this will deny access
+
+    val result: Response = (authenticator !! (Authenticate(req, List("chef")), 10000)).get
+
+    result.getStatus must equal(Response.Status.FORBIDDEN.getStatusCode)
+
+    // the authenticator must have set a security context
+    verify(req).setSecurityContext(any[SecurityContext])
+  }
+}
+