src/main/java/org/onap/dmaap/kafkaAuthorize/KafkaCustomAuthorizer.java

   1 /*******************************************************************************
   2  *  ============LICENSE_START=======================================================
   3  *  org.onap.dmaap
   4  *  ================================================================================
   5  *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
   6  *  ================================================================================
   7  *  Licensed under the Apache License, Version 2.0 (the "License");
   8  *  you may not use this file except in compliance with the License.
   9  *  You may obtain a copy of the License at
  10  *        http://www.apache.org/licenses/LICENSE-2.0
  11 *
  12  *  Unless required by applicable law or agreed to in writing, software
  13  *  distributed under the License is distributed on an "AS IS" BASIS,
  14  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  *  See the License for the specific language governing permissions and
  16  *  limitations under the License.
  17  *  ============LICENSE_END=========================================================
  18  *
  19  *
  20  *******************************************************************************/
  21 package org.onap.dmaap.kafkaAuthorize;
  22
  23 import java.util.EnumSet;
  24 import java.util.Map;
  25
  26 import org.apache.kafka.common.acl.AclOperation;
  27 import org.apache.kafka.common.security.auth.KafkaPrincipal;
  28 import org.slf4j.Logger;
  29 import org.slf4j.LoggerFactory;
  30
  31 import org.onap.dmaap.commonauth.kafka.base.authorization.AuthorizationProviderFactory;
  32 import org.onap.dmaap.commonauth.kafka.base.authorization.Cadi3AAFProvider;
  33
  34 import kafka.network.RequestChannel.Session;
  35 import kafka.security.auth.Acl;
  36 import kafka.security.auth.Authorizer;
  37 import kafka.security.auth.Operation;
  38 import kafka.security.auth.Resource;
  39 import scala.collection.immutable.Set;
  40
  41 /**
  42  * A trivial Kafka Authorizer for use with SSL and AAF
  43  * Authentication/Authorization.
  44  *
  45  */
  46 public class KafkaCustomAuthorizer implements Authorizer {
  47
  48         private String[] adminPermission = new String[3];
  49         public static final EnumSet<AclOperation> TOPIC_DESCRIBE_OPERATIONS = EnumSet.of(AclOperation.DESCRIBE_CONFIGS);
  50         public static final EnumSet<AclOperation> TOPIC_READ_WRITE_DESCRIBE_OPERATIONS = EnumSet.of(AclOperation.WRITE,
  51                         AclOperation.READ, AclOperation.DESCRIBE_CONFIGS);
  52         public static final EnumSet<AclOperation> TOPIC_ADMIN_OPERATIONS = EnumSet.of(AclOperation.ALTER,
  53                         AclOperation.ALTER_CONFIGS, AclOperation.CREATE);
  54
  55         private static final Logger logger = LoggerFactory.getLogger(KafkaCustomAuthorizer.class);
  56
  57         @Override
  58         public void configure(final Map<String, ?> arg0) {
  59                 // TODO Auto-generate method stub
  60         }
  61
  62         @Override
  63         public void addAcls(final Set<Acl> arg0, final Resource arg1) {
  64                 // TODO Auto-generated method stub
  65
  66         }
  67
  68         private String[] getTopicPermission(String topicName, AclOperation aclOperation) {
  69
  70                 String namspace = topicName.substring(0, topicName.lastIndexOf("."));
  71                 String[] permission = new String[3];
  72                 if (TOPIC_READ_WRITE_DESCRIBE_OPERATIONS.contains(aclOperation)) {
  73                         permission[0] = namspace + ".topic";
  74                         String instancePart = (System.getenv("pubSubInstPart") != null) ? System.getenv("pubSubInstPart")
  75                                         : ".topic";
  76                         permission[1] = instancePart + topicName;
  77
  78                         if (aclOperation.equals(AclOperation.WRITE)) {
  79                                 permission[2] = "pub";
  80                         } else if (aclOperation.equals(AclOperation.READ)) {
  81                                 permission[2] = "sub";
  82
  83                         } else if (TOPIC_DESCRIBE_OPERATIONS.contains(aclOperation)) {
  84                                 permission[2] = "view";
  85
  86                         }
  87                 } else if (aclOperation.equals(AclOperation.DELETE)) {
  88                         permission = new String(System.getProperty("msgRtr.topicfactory.aaf") + namspace + "|destroy").split("\\|");
  89
  90                 } else if (TOPIC_ADMIN_OPERATIONS.contains(aclOperation)) {
  91                         permission = new String(System.getProperty("msgRtr.topicfactory.aaf") + namspace + "|create").split("\\|");
  92                 }
  93
  94                 return permission;
  95         }
  96
  97         private String[] getAdminPermission() {
  98
  99                 if (adminPermission[0] == null) {
 100                         adminPermission[0] = System.getProperty("namespace") + ".kafka.access";
 101                         adminPermission[1] = "*";
 102                         adminPermission[2] = "*";
 103                 }
 104
 105                 return adminPermission;
 106         }
 107
 108         private String[] getPermission(AclOperation aclOperation, String resource, String topicName) {
 109                 String[] permission = new String[3];
 110                 switch (aclOperation) {
 111
 112                 case ALTER:
 113                 case ALTER_CONFIGS:
 114                 case CREATE:
 115                 case DELETE:
 116                         if (resource.equals("Topic")) {
 117                                 permission = getTopicPermission(topicName, aclOperation);
 118                         } else if (resource.equals("Cluster")) {
 119                                 permission = getAdminPermission();
 120                         }
 121                         break;
 122                 case DESCRIBE_CONFIGS:
 123                 case READ:
 124                 case WRITE:
 125                         if (resource.equals("Topic")) {
 126                                 permission = getTopicPermission(topicName, aclOperation);
 127                         }
 128                         break;
 129                 case IDEMPOTENT_WRITE:
 130                         if (resource.equals("Cluster")) {
 131                                 permission = getAdminPermission();
 132                         }
 133                         break;
 134                 default:
 135                         break;
 136
 137                 }
 138
 139                 return permission;
 140
 141         }
 142
 143         @Override
 144         public boolean authorize(final Session arg0, final Operation arg1, final Resource arg2) {
 145                 if (arg0.principal() == null) {
 146                         return false;
 147                 }
 148
 149                 String fullName = arg0.principal().getName();
 150                 fullName = fullName != null ? fullName.trim() : fullName;
 151                 String topicName = null;
 152                 String[] permission = new String[3];
 153
 154                 String resource = arg2.resourceType().name();
 155
 156                 if (resource.equals("Topic")) {
 157                         topicName = arg2.name();
 158                 }
 159
 160                 if (fullName != null && fullName.equals(Cadi3AAFProvider.getKafkaUsername())) {
 161                         return true;
 162                 }
 163
 164                 if (null != topicName && !topicName.startsWith("org.onap")) {
 165                         return true;
 166                 }
 167
 168                 permission = getPermission(arg1.toJava(), resource, topicName);
 169
 170                 if (permission[0] == null) {
 171                         return true;
 172                 } else {
 173
 174                         try {
 175
 176                                 if (null != topicName) {
 177                                         boolean hasResp = AuthorizationProviderFactory.getProviderFactory().getProvider()
 178                                                         .hasPermission(fullName, permission[0], permission[1], permission[2]);
 179                                         if (hasResp) {
 180                                                 logger.info("Successful Authorization for " + fullName + " on " + topicName + " for "
 181                                                                 + permission[0] + "|" + permission[1] + "|" + permission[2]);
 182                                         }
 183                                         if (!hasResp) {
 184                                                 logger.info(fullName + " is not allowed in " + permission[0] + "|" + permission[1] + "|"
 185                                                                 + permission[2]);
 186                                                 return false;
 187                                         }
 188                                 }
 189                         } catch (final Exception e) {
 190                                 return false;
 191                         }
 192                         return true;
 193
 194                 }
 195         }
 196
 197         @Override
 198         public void close() {
 199                 // TODO Auto-generated method stub
 200
 201         }
 202
 203         @Override
 204         public scala.collection.immutable.Map<Resource, Set<Acl>> getAcls() {
 205                 // TODO Auto-generated method stub
 206                 return null;
 207         }
 208
 209         @Override
 210         public scala.collection.immutable.Map<Resource, Set<Acl>> getAcls(final KafkaPrincipal arg0) {
 211                 // TODO Auto-generated method stub
 212                 return null;
 213         }
 214
 215         @Override
 216         public boolean removeAcls(final Resource arg0) {
 217                 // TODO Auto-generated method stub
 218                 return false;
 219         }
 220
 221         @Override
 222         public boolean removeAcls(final Set<Acl> arg0, final Resource arg1) {
 223                 // TODO Auto-generated method stub
 224                 return false;
 225         }
 226
 227         public Set<Acl> getAcls(Resource arg0) {
 228                 // TODO Auto-generated method stub
 229                 return null;
 230         }
 231 }