package kafka.security.auth

import kafka.network.RequestChannel.Session
import kafka.server.KafkaConfig

/**
 * Top level interface that all plugable authorizer must implement. Kafka server will read "authorizer.class" config
 * value at startup time, create an instance of the specified class and call initialize method.
 * authorizer.class must be a class that implements this interface.
 * If authorizer.class has no value specified no authorization will be performed.
 *
 * From that point onwards, every client request will first be routed to authorize method and the request will only be
 * authorized if the method returns true.
 */
trait Authorizer {
  /**
   * Guaranteed to be called before any authorize call is made.
   */
  def initialize(kafkaConfig: KafkaConfig): Unit
  
  /**
   * @param session The session being authenticated.
   * @param operation Type of operation client is trying to perform on resource.
   * @param resource Resource the client is trying to access.
   * @return
   */
  def authorize(session: Session, operation: Operation, resource: Resource): Boolean

  /**
   * add the acls to resource, this is an additive operation so existing acls will not be overwritten, instead these new
   * acls will be added to existing acls.
   * @param acls set of acls to add to existing acls
   * @param resource the resource to which these acls should be attached.
   */
  def addAcls(acls: Set[Acl], resource: Resource): Unit

  /**
   * remove these acls from the resource.
   * @param acls set of acls to be removed.
   * @param resource resource from which the acls should be removed.
   * @return true if some acl got removed, false if no acl was removed.
   */
  def removeAcls(acls: Set[Acl], resource: Resource): Boolean

  /**
   * remove a resource along with all of its acls from acl store.
   * @param resource
   * @return
   */
  def removeAcls(resource: Resource): Boolean

  /**
   * get set of acls for this resource
   * @param resource
   * @return empty set if no acls are found, otherwise the acls for the resource.
   */
  def getAcls(resource: Resource): Set[Acl]

  /**
   * get the acls for this principal.
   * @param principal
   * @return empty set if no acls exist for this principal, otherwise the acls for the principal.
   */
  def getAcls(principal: KafkaPrincipal): Set[Acl]
}

object RequestChannel extends Logging {
	case class Session(principal: Principal, host: String)
}

/**
 * 
 * @param principalType type of principal. (i.e. for default implementation we will support user and group but custom authorizer can add more types or chose to ignore this field.)
 * @param name name of the principal
 */
case class KafkaPrincipal(principalType: String, name: String) extends Principal 

package kafka.security.auth

/**
 * Different operations a client may perform on kafka resources.
 */
public enum Operation {
   READ, 
   WRITE,
   CREATE,
   DELETE,
   EDIT,
   DESCRIBE,
   SEND_CONTROL_MESSAGE
}

public enum PermissionType {
   ALLOW,
   DENY
}

/**
 * Exception thrown when a principal is not authorized to perform an operation.
 * @param message
 */
class AuthorizationException(principal: String, operation: Operation, host: String, resource: String) extends RuntimeException {
}

package kafka.security.auth


/**
 * An instance of this class will represent an acl that can express following statement.
 * <pre>
 * Principal P has permissionType PT on Operations O1,O2 from hosts H1,H2.
 * </pre>
 * @param principal A value of *:* indicates all users.
 * @param permissionType
 * @param hosts A value of * indicates all hosts.
 * @param operations A value of ALL indicates all operations.
 */
case class Acl(val principal: KafkaPrincipal,val permissionType: PermissionType,val hosts: Set[String],val operations: Set[Operation]) 

{"version": 1, 
"acls": [
  {
    "principal": "alice”,    
    "permissionType": "ALLOW",
    "operations": [
      "READ",
      "WRITE"
    ],
    "hosts": [
      "host1",
      "host2"
    ]
  },
  {
    "principal": "bob”,
    "permissionType": "ALLOW",
    "operations": [
      "READ"
    ],
    "hosts": [
      "*"
    ]
  },
  {
    "principal": "bob”,
    "permissionType": "DENY",
    "operations": [
      "READ"
    ],
    "hosts": [
      "host1",
	  "host2"
    ]
  }
]
}

authorize(session, operation, resource) { 
principal = session.principal  
remoteAddress = session.host   
if(principal is one of the superuser) {   
	return true 
} 
 
acls = getAcl(resource) //will have a cache. 
if(acls.isEmpty || acls.contains(allowAccessToEveryoneAcl)) return true   
 
if any deny acl's are configured for this principal/host/operation combination, deny the access. 
otherwise check if any acl exist that explicitly allows operation for this principal/host/operation and return true.

no explicit allow acls were found so deny the access. 
}