* org.onap.dmaap
* ================================================================================
* Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ * Modification copyright (C) 2021 Nordix Foundation.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
*******************************************************************************/
package org.onap.dmaap.kafkaAuthorize;
+import java.util.EnumSet;
import java.util.Map;
+import org.apache.kafka.common.acl.AclOperation;
import org.apache.kafka.common.security.auth.KafkaPrincipal;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import org.onap.aaf.cadi.PropAccess;
import org.onap.dmaap.commonauth.kafka.base.authorization.AuthorizationProviderFactory;
+import org.onap.dmaap.commonauth.kafka.base.authorization.Cadi3AAFProvider;
+
import kafka.network.RequestChannel.Session;
import kafka.security.auth.Acl;
import kafka.security.auth.Authorizer;
*
*/
public class KafkaCustomAuthorizer implements Authorizer {
- private PropAccess access;
+
+ private final String[] adminPermission = new String[3];
+ protected static final EnumSet<AclOperation> TOPIC_DESCRIBE_OPERATIONS = EnumSet.of(AclOperation.DESCRIBE_CONFIGS);
+ protected static final EnumSet<AclOperation> TOPIC_READ_WRITE_DESCRIBE_OPERATIONS = EnumSet.of(AclOperation.WRITE,
+ AclOperation.READ, AclOperation.DESCRIBE_CONFIGS);
+ protected static final EnumSet<AclOperation> TOPIC_ADMIN_OPERATIONS = EnumSet.of(AclOperation.ALTER,
+ AclOperation.ALTER_CONFIGS, AclOperation.CREATE);
+ static final String TOPIC = "Topic";
+
private static final Logger logger = LoggerFactory.getLogger(KafkaCustomAuthorizer.class);
- // I'm assuming this is called BEFORE any usage...
@Override
public void configure(final Map<String, ?> arg0) {
// TODO Auto-generate method stub
}
+ private String[] getTopicPermission(String topicName, AclOperation aclOperation) {
+
+ String namspace = topicName.substring(0, topicName.lastIndexOf("."));
+ String[] permission = new String[3];
+ if (TOPIC_READ_WRITE_DESCRIBE_OPERATIONS.contains(aclOperation)) {
+ permission[0] = namspace + ".topic";
+ String instancePart = (System.getenv("pubSubInstPart") != null) ? System.getenv("pubSubInstPart")
+ : ".topic";
+ permission[1] = instancePart + topicName;
+
+ if (aclOperation.equals(AclOperation.WRITE)) {
+ permission[2] = "pub";
+ } else if (aclOperation.equals(AclOperation.READ)) {
+ permission[2] = "sub";
+
+ } else if (TOPIC_DESCRIBE_OPERATIONS.contains(aclOperation)) {
+ permission[2] = "view";
+
+ }
+ } else if (aclOperation.equals(AclOperation.DELETE)) {
+ permission = (System.getProperty("msgRtr.topicfactory.aaf") + namspace + "|destroy").split("\\|");
+
+ } else if (TOPIC_ADMIN_OPERATIONS.contains(aclOperation)) {
+ permission = (System.getProperty("msgRtr.topicfactory.aaf") + namspace + "|create").split("\\|");
+ }
+
+ return permission;
+ }
+
+ private String[] getAdminPermission() {
+
+ if (adminPermission[0] == null) {
+ adminPermission[0] = System.getProperty("namespace") + ".kafka.access";
+ adminPermission[1] = "*";
+ adminPermission[2] = "*";
+ }
+
+ return adminPermission;
+ }
+
+ private String[] getPermission(AclOperation aclOperation, String resource, String topicName) {
+ String[] permission = new String[3];
+ switch (aclOperation) {
+
+ case ALTER:
+ case ALTER_CONFIGS:
+ case CREATE:
+ case DELETE:
+ if (resource.equals(TOPIC)) {
+ permission = getTopicPermission(topicName, aclOperation);
+ } else if (resource.equals("Cluster")) {
+ permission = getAdminPermission();
+ }
+ break;
+ case DESCRIBE_CONFIGS:
+ case READ:
+ case WRITE:
+ if (resource.equals(TOPIC)) {
+ permission = getTopicPermission(topicName, aclOperation);
+ }
+ break;
+ case IDEMPOTENT_WRITE:
+ if (resource.equals("Cluster")) {
+ permission = getAdminPermission();
+ }
+ break;
+ default:
+ break;
+
+ }
+ return permission;
+
+ }
+
@Override
public boolean authorize(final Session arg0, final Operation arg1, final Resource arg2) {
if (arg0.principal() == null) {
String fullName = arg0.principal().getName();
fullName = fullName != null ? fullName.trim() : fullName;
String topicName = null;
- String namspace = null;
- String ins = null;
- String type = null;
- String action = null;
-
- String kafkaactivity = arg1.name();
-
- if (kafkaactivity.equals("Read")) {
- action = "sub";
- } else if (kafkaactivity.equals("Write")) {
- action = "pub";
- } else if (kafkaactivity.equals("Describe")) {
+ String[] permission;
+
+ String resource = arg2.resourceType().name();
+
+ if (resource.equals(TOPIC)) {
+ topicName = arg2.name();
+ }
+
+ if (fullName != null && fullName.equals(Cadi3AAFProvider.getKafkaUsername())) {
return true;
}
- if (arg2.resourceType().name().equals("Topic")) {
- topicName = arg2.name();
- } else {
+
+ if ((!Cadi3AAFProvider.isCadiEnabled())||(null != topicName && !topicName.startsWith("org.onap"))) {
return true;
}
- try {
+ permission = getPermission(arg1.toJava(), resource, topicName);
- if (null != topicName && topicName.indexOf(".") > 0) {
- namspace = topicName.substring(0, topicName.lastIndexOf("."));
- ins = namspace + ".topic";
- type = ":topic." + topicName;
- logger.info("^Event Received for topic " + topicName + " , User " + fullName + " , action = " + action);
- }
+ if (permission[0] != null) {
+ return !checkPermissions(fullName, topicName, permission);
+ }
+ return true;
+ }
- if (null != fullName && fullName.equals("admin")) {
- return true;
- }
+ private boolean checkPermissions(String fullName, String topicName, String[] permission) {
+ try {
if (null != topicName) {
boolean hasResp = AuthorizationProviderFactory.getProviderFactory().getProvider()
- .hasPermission(fullName, ins, type, action);
+ .hasPermission(fullName, permission[0], permission[1], permission[2]);
if (hasResp) {
- logger.info("Successful Authorization for " + fullName + " on " + topicName + " for " + ins + "|"
- + type + "|" + action);
+ logger.info("Successful Authorization for {} on {} for {} | {} | {}", fullName, topicName,
+ permission[0], permission[1], permission[2]);
}
if (!hasResp) {
- logger.info(fullName + " is not allowed in " + ins + "|" + type + "|" + action);
- throw new Exception(fullName + " is not allowed in " + ins + "|" + type + "|" + action);
+ logger.info("{} is not allowed in {} | {} | {}", fullName, permission[0], permission[1],
+ permission[2]);
+ return true;
}
}
} catch (final Exception e) {
- return false;
+ return true;
}
- return true;
+ return false;
}
@Override
Code Review / dmaap / kafka11aaf.git / blobdiff