Skip to content

KAFKA-8847; Deprecate and remove usage of supporting classes in kafka.security.auth #7966

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rajinisivaram merged 2 commits into from
Jan 17, 2020
Merged

KAFKA-8847; Deprecate and remove usage of supporting classes in kafka.security.auth #7966

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
53d8342
KAFKA-8847; Deprecate and remove usage of supporting classes in kafka…
rajinisivaram Jan 14, 2020
7980812
Address review comment
rajinisivaram Jan 17, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 40 additions & 4 deletions clients/src/main/java/org/apache/kafka/common/utils/SecurityUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
package org.apache.kafka.common.utils;

import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.acl.AclPermissionType;
import org.apache.kafka.common.config.SecurityConfig;
import org.apache.kafka.common.resource.ResourceType;
import org.apache.kafka.common.security.auth.SecurityProviderCreator;
Expand All @@ -26,6 +27,7 @@

import java.security.Security;
import java.util.HashMap;
import java.util.Locale;
import java.util.Map;

public class SecurityUtils {
Expand All @@ -34,18 +36,27 @@ public class SecurityUtils {

private static final Map<String, ResourceType> NAME_TO_RESOURCE_TYPES;
private static final Map<String, AclOperation> NAME_TO_OPERATIONS;
private static final Map<String, AclPermissionType> NAME_TO_PERMISSION_TYPES;

static {
NAME_TO_RESOURCE_TYPES = new HashMap<>(ResourceType.values().length);
NAME_TO_OPERATIONS = new HashMap<>(AclOperation.values().length);
NAME_TO_PERMISSION_TYPES = new HashMap<>(AclPermissionType.values().length);

for (ResourceType resourceType : ResourceType.values()) {
String resourceTypeName = toPascalCase(resourceType.name());
NAME_TO_RESOURCE_TYPES.put(resourceTypeName, resourceType);
NAME_TO_RESOURCE_TYPES.put(resourceTypeName.toUpperCase(Locale.ROOT), resourceType);
}
for (AclOperation operation : AclOperation.values()) {
String operationName = toPascalCase(operation.name());
NAME_TO_OPERATIONS.put(operationName, operation);
NAME_TO_OPERATIONS.put(operationName.toUpperCase(Locale.ROOT), operation);
}
for (AclPermissionType permissionType : AclPermissionType.values()) {
String permissionName = toPascalCase(permissionType.name());
NAME_TO_PERMISSION_TYPES.put(permissionName, permissionType);
NAME_TO_PERMISSION_TYPES.put(permissionName.toUpperCase(Locale.ROOT), permissionType);
}
}

Expand Down Expand Up @@ -86,13 +97,38 @@ public static void addConfiguredSecurityProviders(Map<String, ?> configs) {
}

public static ResourceType resourceType(String name) {
ResourceType resourceType = NAME_TO_RESOURCE_TYPES.get(name);
return resourceType == null ? ResourceType.UNKNOWN : resourceType;
return valueFromMap(NAME_TO_RESOURCE_TYPES, name, ResourceType.UNKNOWN);
}

public static AclOperation operation(String name) {
AclOperation operation = NAME_TO_OPERATIONS.get(name);
return operation == null ? AclOperation.UNKNOWN : operation;
return valueFromMap(NAME_TO_OPERATIONS, name, AclOperation.UNKNOWN);
}

public static AclPermissionType permissionType(String name) {
return valueFromMap(NAME_TO_PERMISSION_TYPES, name, AclPermissionType.UNKNOWN);
}

// We use Pascal-case to store these values, so lookup using provided key first to avoid
// case conversion for the common case. For backward compatibility, also perform
// case-insensitive look up (without underscores) by converting the key to upper-case.
private static <T> T valueFromMap(Map<String, T> map, String key, T unknown) {
T value = map.get(key);
if (value == null) {
value = map.get(key.toUpperCase(Locale.ROOT));
}
return value == null ? unknown : value;
}

public static String resourceTypeName(ResourceType resourceType) {
return toPascalCase(resourceType.name());
}

public static String operationName(AclOperation operation) {
return toPascalCase(operation.name());
}

public static String permissionTypeName(AclPermissionType permissionType) {
return toPascalCase(permissionType.name());
}

private static String toPascalCase(String name) {
Expand Down
141 changes: 18 additions & 123 deletions core/src/main/scala/kafka/admin/AclCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,7 @@ import java.util.Properties

import joptsimple._
import joptsimple.util.EnumConverter
import kafka.security.auth._
import kafka.security.authorizer.AuthorizerUtils
import kafka.security.authorizer.{AclAuthorizer, AclEntry, AuthorizerUtils}
import kafka.server.KafkaConfig
import kafka.utils._
import org.apache.kafka.clients.admin.{Admin, AdminClientConfig}
Expand All @@ -33,7 +32,7 @@ import org.apache.kafka.common.resource.{PatternType, ResourcePattern, ResourceP
import org.apache.kafka.common.security.JaasUtils
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.utils.{Utils, SecurityUtils => JSecurityUtils}
import org.apache.kafka.server.authorizer.{Authorizer => JAuthorizer}
import org.apache.kafka.server.authorizer.Authorizer

import scala.collection.JavaConverters._
import scala.compat.java8.OptionConverters._
Expand All @@ -58,18 +57,12 @@ object AclCommand extends Logging {
if (opts.options.has(opts.bootstrapServerOpt)) {
new AdminClientService(opts)
} else {
val authorizerClass = if (opts.options.has(opts.authorizerOpt)) {
val className = opts.options.valueOf(opts.authorizerOpt)
Class.forName(className, true, Utils.getContextOrKafkaClassLoader)
} else
classOf[SimpleAclAuthorizer]

if (classOf[JAuthorizer].isAssignableFrom(authorizerClass))
new JAuthorizerService(authorizerClass.asSubclass(classOf[JAuthorizer]), opts)
else if (classOf[Authorizer].isAssignableFrom(authorizerClass))
new AuthorizerService(authorizerClass.asSubclass(classOf[Authorizer]), opts)
val authorizerClassName = if (opts.options.has(opts.authorizerOpt))
opts.options.valueOf(opts.authorizerOpt)
else
throw new IllegalArgumentException(s"Authorizer $authorizerClass does not implement ${classOf[Authorizer]} or ${classOf[JAuthorizer]}.")
classOf[AclAuthorizer].getName

new AuthorizerService(authorizerClassName, opts)
}
}

Expand Down Expand Up @@ -190,8 +183,7 @@ object AclCommand extends Logging {
}
}

@deprecated("Use JAuthorizerService", "Since 2.4")
class AuthorizerService(val authorizerClass: Class[_ <: Authorizer], val opts: AclCommandOptions) extends AclCommandService with Logging {
class AuthorizerService(val authorizerClassName: String, val opts: AclCommandOptions) extends AclCommandService with Logging {

private def withAuthorizer()(f: Authorizer => Unit): Unit = {
val defaultProps = Map(KafkaConfig.ZkEnableSecureAclsProp -> JaasUtils.isZkSecurityEnabled)
Expand All @@ -203,105 +195,7 @@ object AclCommand extends Logging {
defaultProps
}

val authZ = Utils.newInstance(authorizerClass)
try {
authZ.configure(authorizerProperties.asJava)
f(authZ)
}
finally CoreUtils.swallow(authZ.close(), this)
}

def addAcls(): Unit = {
val resourceToAcl = getResourceToAcls(opts)
withAuthorizer() { authorizer =>
for ((resource, acls) <- resourceToAcl) {
println(s"Adding ACLs for resource `$resource`: $Newline ${acls.map("\t" + _).mkString(Newline)} $Newline")
authorizer.addAcls(acls.map(AuthorizerUtils.convertToAcl), AuthorizerUtils.convertToResource(resource))
}

listAcls()
}
}

def removeAcls(): Unit = {
withAuthorizer() { authorizer =>
val filterToAcl = getResourceFilterToAcls(opts)

for ((filter, acls) <- filterToAcl) {
if (acls.isEmpty) {
if (confirmAction(opts, s"Are you sure you want to delete all ACLs for resource filter `$filter`? (y/n)"))
removeAcls(authorizer, acls, filter)
} else {
if (confirmAction(opts, s"Are you sure you want to remove ACLs: $Newline ${acls.map("\t" + _).mkString(Newline)} $Newline from resource filter `$filter`? (y/n)"))
removeAcls(authorizer, acls, filter)
}
}

listAcls()
}
}

def listAcls(): Unit = {
withAuthorizer() { authorizer =>
val filters = getResourceFilter(opts, dieIfNoResourceFound = false)
val listPrincipals = getPrincipals(opts, opts.listPrincipalsOpt)

if (listPrincipals.isEmpty) {
val resourceToAcls = getFilteredResourceToAcls(authorizer, filters)
for ((resource, acls) <- resourceToAcls)
println(s"Current ACLs for resource `$resource`: $Newline ${acls.map("\t" + _).mkString(Newline)} $Newline")
} else {
listPrincipals.foreach(principal => {
println(s"ACLs for principal `$principal`")
val resourceToAcls = getFilteredResourceToAcls(authorizer, filters, Some(principal))
for ((resource, acls) <- resourceToAcls)
println(s"Current ACLs for resource `$resource`: $Newline ${acls.map("\t" + _).mkString(Newline)} $Newline")
})
}
}
}

private def removeAcls(authorizer: Authorizer, acls: Set[AccessControlEntry], filter: ResourcePatternFilter): Unit = {
getAcls(authorizer, filter)
.keys
.foreach(resource =>
if (acls.isEmpty) authorizer.removeAcls(resource)
else authorizer.removeAcls(acls.map(AuthorizerUtils.convertToAcl), resource)
)
}

private def getFilteredResourceToAcls(authorizer: Authorizer, filters: Set[ResourcePatternFilter],
listPrincipal: Option[KafkaPrincipal] = None): Iterable[(Resource, Set[Acl])] = {
if (filters.isEmpty)
if (listPrincipal.isEmpty)
authorizer.getAcls()
else
authorizer.getAcls(listPrincipal.get)
else filters.flatMap(filter => getAcls(authorizer, filter, listPrincipal))
}

private def getAcls(authorizer: Authorizer, filter: ResourcePatternFilter,
listPrincipal: Option[KafkaPrincipal] = None): Map[Resource, Set[Acl]] =
if (listPrincipal.isEmpty)
authorizer.getAcls().filter { case (resource, _) => filter.matches(resource.toPattern) }
else
authorizer.getAcls(listPrincipal.get).filter { case (resource, _) => filter.matches(resource.toPattern) }

}

class JAuthorizerService(val authorizerClass: Class[_ <: JAuthorizer], val opts: AclCommandOptions) extends AclCommandService with Logging {

private def withAuthorizer()(f: JAuthorizer => Unit): Unit = {
val defaultProps = Map(KafkaConfig.ZkEnableSecureAclsProp -> JaasUtils.isZkSecurityEnabled)
val authorizerProperties =
if (opts.options.has(opts.authorizerPropertiesOpt)) {
val authorizerProperties = opts.options.valuesOf(opts.authorizerPropertiesOpt).asScala
defaultProps ++ CommandLineUtils.parseKeyValueArgs(authorizerProperties, acceptMissingValue = false).asScala
} else {
defaultProps
}

val authZ = Utils.newInstance(authorizerClass)
val authZ = AuthorizerUtils.createAuthorizer(authorizerClassName)
try {
authZ.configure(authorizerProperties.asJava)
f(authZ)
Expand Down Expand Up @@ -367,7 +261,7 @@ object AclCommand extends Logging {
}
}

private def removeAcls(authorizer: JAuthorizer, acls: Set[AccessControlEntry], filter: ResourcePatternFilter): Unit = {
private def removeAcls(authorizer: Authorizer, acls: Set[AccessControlEntry], filter: ResourcePatternFilter): Unit = {
val result = if (acls.isEmpty)
authorizer.deleteAcls(null, List(new AclBindingFilter(filter, AccessControlEntryFilter.ANY)).asJava)
else {
Expand All @@ -388,7 +282,7 @@ object AclCommand extends Logging {
}
}

private def getAcls(authorizer: JAuthorizer, filters: Set[ResourcePatternFilter]): Map[ResourcePattern, Set[AccessControlEntry]] = {
private def getAcls(authorizer: Authorizer, filters: Set[ResourcePatternFilter]): Map[ResourcePattern, Set[AccessControlEntry]] = {
val aclBindings =
if (filters.isEmpty) authorizer.acls(AclBindingFilter.ANY).asScala
else {
Expand Down Expand Up @@ -500,7 +394,8 @@ object AclCommand extends Logging {
}

private def getAcl(opts: AclCommandOptions): Set[AccessControlEntry] = {
val operations = opts.options.valuesOf(opts.operationsOpt).asScala.map(operation => Operation.fromString(operation.trim)).map(_.toJava).toSet
val operations = opts.options.valuesOf(opts.operationsOpt).asScala
.map(operation => JSecurityUtils.operation(operation.trim)).toSet
getAcl(opts, operations)
}

Expand All @@ -518,7 +413,7 @@ object AclCommand extends Logging {
if (opts.options.has(hostOptionSpec))
opts.options.valuesOf(hostOptionSpec).asScala.map(_.trim).toSet
else if (opts.options.has(principalOptionSpec))
Set[String](Acl.WildCardHost)
Set[String](AclEntry.WildcardHost)
else
Set.empty[String]
}
Expand Down Expand Up @@ -565,8 +460,8 @@ object AclCommand extends Logging {

private def validateOperation(opts: AclCommandOptions, resourceToAcls: Map[ResourcePatternFilter, Set[AccessControlEntry]]): Unit = {
for ((resource, acls) <- resourceToAcls) {
val validOps = ResourceType.fromJava(resource.resourceType).supportedOperations + All
if ((acls.map(_.operation) -- validOps.map(_.toJava)).nonEmpty)
val validOps = AclEntry.supportedOperations(resource.resourceType) + AclOperation.ALL
if ((acls.map(_.operation) -- validOps).nonEmpty)
CommandLineUtils.printUsageAndDie(opts.parser, s"ResourceType ${resource.resourceType} only supports operations ${validOps.mkString(",")}")
}
}
Expand Down Expand Up @@ -641,10 +536,10 @@ object AclCommand extends Logging {
val listOpt = parser.accepts("list", "List ACLs for the specified resource, use --topic <topic> or --group <group> or --cluster to specify a resource.")

val operationsOpt = parser.accepts("operation", "Operation that is being allowed or denied. Valid operation names are: " + Newline +
Operation.values.map("\t" + _).mkString(Newline) + Newline)
AclEntry.AclOperations.map("\t" + JSecurityUtils.operationName(_)).mkString(Newline) + Newline)
.withRequiredArg
.ofType(classOf[String])
.defaultsTo(All.name)
.defaultsTo(JSecurityUtils.operationName(AclOperation.ALL))

val allowPrincipalsOpt = parser.accepts("allow-principal", "principal is in principalType:name format." +
" Note that principalType must be supported by the Authorizer being used." +
Expand Down
Loading