ranger-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mad...@apache.org
Subject [15/16] incubator-ranger git commit: RANGER-246 - Kafka authorization plugin (Bosco Durai via Selvamohan Neethiraj)
Date Mon, 18 May 2015 04:53:13 GMT
RANGER-246 - Kafka authorization plugin (Bosco Durai via Selvamohan Neethiraj)


Project: http://git-wip-us.apache.org/repos/asf/incubator-ranger/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ranger/commit/f0c9216d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ranger/tree/f0c9216d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ranger/diff/f0c9216d

Branch: refs/heads/tag-policy
Commit: f0c9216dafbe81cbbe7ae9696c9f6de456ebfac6
Parents: 1031041
Author: sneethiraj <sneethir@apache.org>
Authored: Sun May 17 18:40:00 2015 -0400
Committer: sneethiraj <sneethir@apache.org>
Committed: Sun May 17 18:40:00 2015 -0400

----------------------------------------------------------------------
 .gitignore                                      |   1 +
 .../service-defs/ranger-servicedef-kafka.json   |  12 +-
 plugin-kafka/pom.xml                            | 103 +++++-----
 .../kafka/authorizer/RangerKafkaAuthorizer.java | 201 ++++++++++++++++++-
 .../kafka/client/ServiceKafkaClient.java        |   2 -
 pom.xml                                         |  13 +-
 6 files changed, 258 insertions(+), 74 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/f0c9216d/.gitignore
----------------------------------------------------------------------
diff --git a/.gitignore b/.gitignore
index dd4e2c2..7f41f0c 100644
--- a/.gitignore
+++ b/.gitignore
@@ -7,3 +7,4 @@
 .project
 /target/
 winpkg/target
+.DS_Store

http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/f0c9216d/agents-common/src/main/resources/service-defs/ranger-servicedef-kafka.json
----------------------------------------------------------------------
diff --git a/agents-common/src/main/resources/service-defs/ranger-servicedef-kafka.json b/agents-common/src/main/resources/service-defs/ranger-servicedef-kafka.json
index 9928c5d..d19b10c 100644
--- a/agents-common/src/main/resources/service-defs/ranger-servicedef-kafka.json
+++ b/agents-common/src/main/resources/service-defs/ranger-servicedef-kafka.json
@@ -59,15 +59,9 @@
 		},
 		{
 			"itemId": 7,
-			"name":"replicate",
-			"label":"Replicate"
-		},
-		{
-			"itemId": 8,
-			"name":"connect",
-			"label":"Connect"
+			"name":"kafka_admin",
+			"label":"Kafka Admin"
 		}
-		
 	],
 	"configs":[
 		{
@@ -97,7 +91,7 @@
 			"name":"commonNameForCertificate",
 			"type":"string",
 			"mandatory":false,
-			"label":"Common Name for Certificate"
+			"label":"Ranger Plugin SSL CName"
 		}
 		
 	],

http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/f0c9216d/plugin-kafka/pom.xml
----------------------------------------------------------------------
diff --git a/plugin-kafka/pom.xml b/plugin-kafka/pom.xml
index e9ea265..afee47d 100644
--- a/plugin-kafka/pom.xml
+++ b/plugin-kafka/pom.xml
@@ -1,56 +1,51 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<!--
-  Licensed to the Apache Software Foundation (ASF) under one or more
-  contributor license agreements.  See the NOTICE file distributed with
-  this work for additional information regarding copyright ownership.
-  The ASF licenses this file to You 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.
--->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-  <modelVersion>4.0.0</modelVersion>
-  <groupId>security_plugins.ranger-kafka-plugin</groupId>
-  <artifactId>ranger-kafka-plugin</artifactId>
-  <name>KAFKA Security Plugin</name>
-  <description>KAFKA Security Plugin</description>
-  <packaging>jar</packaging>
-  <properties>
-    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-  </properties>
-  <parent>
-     <groupId>org.apache.ranger</groupId>
-     <artifactId>ranger</artifactId>
-     <version>0.5.0</version>
-     <relativePath>..</relativePath>
-  </parent>
-  <dependencies>
-    <dependency>
-      <groupId>security_plugins.ranger-plugins-common</groupId>
-      <artifactId>ranger-plugins-common</artifactId>
-      <version>${project.version}</version>
-    </dependency>
-    <dependency>
-      <groupId>security_plugins.ranger-plugins-audit</groupId>
-      <artifactId>ranger-plugins-audit</artifactId>
-      <version>${project.version}</version>
-    </dependency>
-    <dependency>
-      <groupId>org.apache.ranger</groupId>
-      <artifactId>credentialbuilder</artifactId>
-      <version>${project.version}</version>
-    </dependency>
-    <dependency>
-      <groupId>org.apache.kafka</groupId>
-      <artifactId>kafka_2.10</artifactId>
-      <version>${kafka.version}</version>
-    </dependency>
-  </dependencies>
+<!-- Licensed to the Apache Software Foundation (ASF) under one or more contributor 
+	license agreements. See the NOTICE file distributed with this work for additional 
+	information regarding copyright ownership. The ASF licenses this file to 
+	You 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. -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<groupId>security_plugins.ranger-kafka-plugin</groupId>
+	<artifactId>ranger-kafka-plugin</artifactId>
+	<name>KAFKA Security Plugin</name>
+	<description>KAFKA Security Plugin</description>
+	<packaging>jar</packaging>
+	<properties>
+		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+	</properties>
+	<parent>
+		<groupId>org.apache.ranger</groupId>
+		<artifactId>ranger</artifactId>
+		<version>0.5.0</version>
+		<relativePath>..</relativePath>
+	</parent>
+	<dependencies>
+		<dependency>
+			<groupId>security_plugins.ranger-plugins-common</groupId>
+			<artifactId>ranger-plugins-common</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>security_plugins.ranger-plugins-audit</groupId>
+			<artifactId>ranger-plugins-audit</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.ranger</groupId>
+			<artifactId>credentialbuilder</artifactId>
+			<version>${project.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>org.apache.kafka</groupId>
+			<artifactId>kafka_2.10</artifactId>
+			<version>${kafka.version}</version>
+		</dependency>
+	</dependencies>
 </project>

http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/f0c9216d/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/RangerKafkaAuthorizer.java
----------------------------------------------------------------------
diff --git a/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/RangerKafkaAuthorizer.java
b/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/RangerKafkaAuthorizer.java
index 40c2204..4689957 100644
--- a/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/RangerKafkaAuthorizer.java
+++ b/plugin-kafka/src/main/java/org/apache/ranger/authorization/kafka/authorizer/RangerKafkaAuthorizer.java
@@ -1,4 +1,3 @@
-
 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
@@ -20,14 +19,208 @@
 
 package org.apache.ranger.authorization.kafka.authorizer;
 
+import java.util.Date;
+
+import kafka.security.auth.Acl;
+import kafka.security.auth.Authorizer;
+import kafka.security.auth.KafkaPrincipal;
+import kafka.security.auth.Operation;
+import kafka.security.auth.Resource;
+import kafka.security.auth.ResourceType;
+import kafka.server.KafkaConfig;
+import kafka.network.RequestChannel.Session;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.ranger.authorization.utils.StringUtil;
+import org.apache.ranger.plugin.audit.RangerDefaultAuditHandler;
+import org.apache.ranger.plugin.policyengine.RangerAccessRequestImpl;
+import org.apache.ranger.plugin.policyengine.RangerAccessResourceImpl;
+import org.apache.ranger.plugin.policyengine.RangerAccessResult;
+import org.apache.ranger.plugin.service.RangerBasePlugin;
+
+import scala.collection.immutable.HashSet;
+import scala.collection.immutable.Set;
+
+public class RangerKafkaAuthorizer implements Authorizer {
+	private static final Log logger = LogFactory
+			.getLog(RangerKafkaAuthorizer.class);
+
+	public static final String KEY_TOPIC = "topic";
+	public static final String KEY_CLUSTER = "cluster";
+	public static final String KEY_CONSUMER_GROUP = "consumer_group";
+
+	public static final String ACCESS_TYPE_READ = "read";
+	public static final String ACCESS_TYPE_WRITE = "write";
+	public static final String ACCESS_TYPE_CREATE = "create";
+	public static final String ACCESS_TYPE_DELETE = "delete";
+	public static final String ACCESS_TYPE_ALTER = "alter";
+	public static final String ACCESS_TYPE_DESCRIBE = "describe";
+	public static final String ACCESS_TYPE_KAFKA_ADMIN = "kafka_admin";
+
+	private static volatile RangerBasePlugin rangerPlugin = null;
+
+	public RangerKafkaAuthorizer() {
+		if (rangerPlugin == null) {
+			rangerPlugin = new RangerBasePlugin("kafka", "kafka");
+		}
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see kafka.security.auth.Authorizer#initialize(kafka.server.KafkaConfig)
+	 */
+	@Override
+	public void initialize(KafkaConfig kafkaConfig) {
+		rangerPlugin.init();
+		RangerDefaultAuditHandler auditHandler = new RangerDefaultAuditHandler();
+
+		rangerPlugin.setResultProcessor(auditHandler);
+	}
+
+	// TODO: Fix this after Session is fixed
+	// @Override
+	public boolean authorize(Session session, Operation operation,
+			Resource resource) {
+
+		String userName = null;
+		java.util.Set<String> userGroups = getGroupsForUser(userName);
+		String ip = null;
+		Date eventTime = StringUtil.getUTCDate();
+		String accessType = mapToRangerAccessType(operation);
+		if (accessType == null) {
+			logger.fatal("Unsupported access type. session=" + session
+					+ ", operation=" + operation + ", resource=" + resource);
+			return false;
+		}
+		String action = accessType;
+
+		RangerAccessRequestImpl rangerRequest = new RangerAccessRequestImpl();
+		rangerRequest.setUser(userName);
+		rangerRequest.setUserGroups(userGroups);
+		rangerRequest.setClientIPAddress(ip);
+		rangerRequest.setAccessTime(eventTime);
+
+		RangerAccessResourceImpl rangerResource = new RangerAccessResourceImpl();
+
+		if (resource.resourceType().equals(ResourceType.TOPIC)) {
+			rangerResource.setValue(KEY_TOPIC, resource.name());
+		} else if (resource.resourceType().equals(ResourceType.CLUSTER)) {
+			rangerResource.setValue(KEY_CLUSTER, resource.name());
+		} else if (resource.resourceType().equals(ResourceType.GROUP)) {
+			rangerResource.setValue(KEY_CONSUMER_GROUP, resource.name());
+		} else {
+			logger.fatal("Unsupported resourceType=" + resource.resourceType());
+			return false;
+		}
+
+		rangerRequest.setResource(rangerResource);
+		rangerRequest.setAccessType(accessType);
+		rangerRequest.setAction(action);
+		rangerRequest.setRequestData(resource.name());
+
+		RangerAccessResult result = rangerPlugin.isAccessAllowed(rangerRequest);
+		return result.getIsAllowed();
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see
+	 * kafka.security.auth.Authorizer#addAcls(scala.collection.immutable.Set,
+	 * kafka.security.auth.Resource)
+	 */
+	@Override
+	public void addAcls(Set<Acl> acls, Resource resource) {
+		logger.error("addAcls() is not supported by Ranger for Kafka");
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see
+	 * kafka.security.auth.Authorizer#removeAcls(scala.collection.immutable.Set,
+	 * kafka.security.auth.Resource)
+	 */
+	@Override
+	public boolean removeAcls(Set<Acl> acls, Resource resource) {
+		logger.error("removeAcls() is not supported by Ranger for Kafka");
+		return false;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see
+	 * kafka.security.auth.Authorizer#removeAcls(kafka.security.auth.Resource)
+	 */
+	@Override
+	public boolean removeAcls(Resource resource) {
+		logger.error("removeAcls() is not supported by Ranger for Kafka");
+		return false;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see kafka.security.auth.Authorizer#getAcls(kafka.security.auth.Resource)
+	 */
+	@Override
+	public Set<Acl> getAcls(Resource resource) {
+		Set<Acl> aclList = new HashSet<Acl>();
+		logger.error("getAcls() is not supported by Ranger for Kafka");
 
-public class RangerKafkaAuthorizer /*KafkaAuthorizationPlugin*/ {
+		return aclList;
+	}
 
-	private static final Log LOG = LogFactory.getLog(RangerKafkaAuthorizer.class);
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see
+	 * kafka.security.auth.Authorizer#getAcls(kafka.security.auth.KafkaPrincipal
+	 * )
+	 */
+	@Override
+	public Set<Acl> getAcls(KafkaPrincipal principal) {
+		Set<Acl> aclList = new HashSet<Acl>();
+		logger.error("getAcls() is not supported by Ranger for Kafka");
+		return aclList;
+	}
 
-    //private static volatile RangerKafkaPlugin kafkaPlugin = null;
+	/**
+	 * @param userName
+	 * @return
+	 */
+	private java.util.Set<String> getGroupsForUser(String userName) {
+		if (userName == null) {
+			return null;
+		}
 
+		// TODO: Need to implement this method
+		return null;
+	}
 
+	/**
+	 * @param operation
+	 * @return
+	 */
+	private String mapToRangerAccessType(Operation operation) {
+		if (operation.equals(Operation.READ)) {
+			return ACCESS_TYPE_READ;
+		} else if (operation.equals(Operation.WRITE)) {
+			return ACCESS_TYPE_WRITE;
+		} else if (operation.equals(Operation.CREATE)) {
+			return ACCESS_TYPE_CREATE;
+		} else if (operation.equals(Operation.DELETE)) {
+			return ACCESS_TYPE_DELETE;
+		} else if (operation.equals(Operation.ALTER)) {
+			return ACCESS_TYPE_ALTER;
+		} else if (operation.equals(Operation.DESCRIBE)) {
+			return ACCESS_TYPE_DESCRIBE;
+		} else if (operation.equals(Operation.CLUSTER_ACTION)) {
+			return ACCESS_TYPE_KAFKA_ADMIN;
+		}
+		return null;
+	}
 }

http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/f0c9216d/plugin-kafka/src/main/java/org/apache/ranger/services/kafka/client/ServiceKafkaClient.java
----------------------------------------------------------------------
diff --git a/plugin-kafka/src/main/java/org/apache/ranger/services/kafka/client/ServiceKafkaClient.java
b/plugin-kafka/src/main/java/org/apache/ranger/services/kafka/client/ServiceKafkaClient.java
index a62bd95..5cca619 100644
--- a/plugin-kafka/src/main/java/org/apache/ranger/services/kafka/client/ServiceKafkaClient.java
+++ b/plugin-kafka/src/main/java/org/apache/ranger/services/kafka/client/ServiceKafkaClient.java
@@ -22,10 +22,8 @@ package org.apache.ranger.services.kafka.client;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
-import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
 

http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/f0c9216d/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 0b5608a..cbb3a08 100644
--- a/pom.xml
+++ b/pom.xml
@@ -87,8 +87,6 @@
   <module>hive-agent</module>
   <module>knox-agent</module>
   <module>storm-agent</module>
-  <module>plugin-kafka</module>
-  <!-- <module>plugin-solr</module> -->
   <module>plugin-yarn</module>
   <module>ranger_solrj</module>
   <module>security-admin</module>
@@ -224,6 +222,12 @@
         	 <module>plugin-solr</module>         
          </modules>
       </profile>
+      <profile>
+          <id>kafka-security</id>
+         <modules>
+        	 <module>plugin-kafka</module>         
+         </modules>
+      </profile>
   </profiles>
   <distributionManagement>
         <repository>
@@ -375,15 +379,14 @@
              <descriptor>src/main/assembly/hbase-agent.xml</descriptor>
              <descriptor>src/main/assembly/knox-agent.xml</descriptor>
              <descriptor>src/main/assembly/storm-agent.xml</descriptor>
-	     <descriptor>src/main/assembly/plugin-kafka.xml</descriptor>
+	         <descriptor>src/main/assembly/plugin-kafka.xml</descriptor>
              <descriptor>src/main/assembly/plugin-yarn.xml</descriptor>
-	     <descriptor>src/main/assembly/plugin-solr.xml</descriptor>
+	         <descriptor>src/main/assembly/plugin-solr.xml</descriptor>
              <descriptor>src/main/assembly/admin-web.xml</descriptor>
              <descriptor>src/main/assembly/usersync.xml</descriptor>
              <descriptor>src/main/assembly/migration-util.xml</descriptor>
              <descriptor>src/main/assembly/kms.xml</descriptor>
              <descriptor>src/main/assembly/ranger-src.xml</descriptor>
-	     <!--<descriptor>src/main/assembly/plugin-kms.xml</descriptor>-->
            </descriptors>
          </configuration>
       </plugin>


Mime
View raw message