1 /*******************************************************************************
2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
6 * Modification copyright (C) 2021 Nordix Foundation.
7 * ================================================================================
8 * Licensed under the Apache License, Version 2.0 (the "License");
9 * you may not use this file except in compliance with the License.
10 * You may obtain a copy of the License at
11 * http://www.apache.org/licenses/LICENSE-2.0
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 * ============LICENSE_END=========================================================
21 *******************************************************************************/
22 package org.onap.dmaap.kafkaAuthorize;
24 import java.util.EnumSet;
27 import org.apache.kafka.common.acl.AclOperation;
28 import org.apache.kafka.common.security.auth.KafkaPrincipal;
29 import org.slf4j.Logger;
30 import org.slf4j.LoggerFactory;
32 import org.onap.dmaap.commonauth.kafka.base.authorization.AuthorizationProviderFactory;
33 import org.onap.dmaap.commonauth.kafka.base.authorization.Cadi3AAFProvider;
35 import kafka.network.RequestChannel.Session;
36 import kafka.security.auth.Acl;
37 import kafka.security.auth.Authorizer;
38 import kafka.security.auth.Operation;
39 import kafka.security.auth.Resource;
40 import scala.collection.immutable.Set;
43 * A trivial Kafka Authorizer for use with SSL and AAF
44 * Authentication/Authorization.
47 public class KafkaCustomAuthorizer implements Authorizer {
49 private final String[] adminPermission = new String[3];
50 protected static final EnumSet<AclOperation> TOPIC_DESCRIBE_OPERATIONS = EnumSet.of(AclOperation.DESCRIBE_CONFIGS);
51 protected static final EnumSet<AclOperation> TOPIC_READ_WRITE_DESCRIBE_OPERATIONS = EnumSet.of(AclOperation.WRITE,
52 AclOperation.READ, AclOperation.DESCRIBE_CONFIGS);
53 protected static final EnumSet<AclOperation> TOPIC_ADMIN_OPERATIONS = EnumSet.of(AclOperation.ALTER,
54 AclOperation.ALTER_CONFIGS, AclOperation.CREATE);
55 static final String TOPIC = "Topic";
57 private static final Logger logger = LoggerFactory.getLogger(KafkaCustomAuthorizer.class);
60 public void configure(final Map<String, ?> arg0) {
61 // TODO Auto-generate method stub
65 public void addAcls(final Set<Acl> arg0, final Resource arg1) {
66 // TODO Auto-generated method stub
70 private String[] getTopicPermission(String topicName, AclOperation aclOperation) {
72 String namspace = topicName.substring(0, topicName.lastIndexOf("."));
73 String[] permission = new String[3];
74 if (TOPIC_READ_WRITE_DESCRIBE_OPERATIONS.contains(aclOperation)) {
75 permission[0] = namspace + ".topic";
76 String instancePart = (System.getenv("pubSubInstPart") != null) ? System.getenv("pubSubInstPart")
78 permission[1] = instancePart + topicName;
80 if (aclOperation.equals(AclOperation.WRITE)) {
81 permission[2] = "pub";
82 } else if (aclOperation.equals(AclOperation.READ)) {
83 permission[2] = "sub";
85 } else if (TOPIC_DESCRIBE_OPERATIONS.contains(aclOperation)) {
86 permission[2] = "view";
89 } else if (aclOperation.equals(AclOperation.DELETE)) {
90 permission = (System.getProperty("msgRtr.topicfactory.aaf") + namspace + "|destroy").split("\\|");
92 } else if (TOPIC_ADMIN_OPERATIONS.contains(aclOperation)) {
93 permission = (System.getProperty("msgRtr.topicfactory.aaf") + namspace + "|create").split("\\|");
99 private String[] getAdminPermission() {
101 if (adminPermission[0] == null) {
102 adminPermission[0] = System.getProperty("namespace") + ".kafka.access";
103 adminPermission[1] = "*";
104 adminPermission[2] = "*";
107 return adminPermission;
110 private String[] getPermission(AclOperation aclOperation, String resource, String topicName) {
111 String[] permission = new String[3];
112 switch (aclOperation) {
118 if (resource.equals(TOPIC)) {
119 permission = getTopicPermission(topicName, aclOperation);
120 } else if (resource.equals("Cluster")) {
121 permission = getAdminPermission();
124 case DESCRIBE_CONFIGS:
127 if (resource.equals(TOPIC)) {
128 permission = getTopicPermission(topicName, aclOperation);
131 case IDEMPOTENT_WRITE:
132 if (resource.equals("Cluster")) {
133 permission = getAdminPermission();
145 public boolean authorize(final Session arg0, final Operation arg1, final Resource arg2) {
146 if (arg0.principal() == null) {
150 String fullName = arg0.principal().getName();
151 fullName = fullName != null ? fullName.trim() : fullName;
152 String topicName = null;
155 String resource = arg2.resourceType().name();
157 if (resource.equals(TOPIC)) {
158 topicName = arg2.name();
161 if (fullName != null && fullName.equals(Cadi3AAFProvider.getKafkaUsername())) {
165 if ((!Cadi3AAFProvider.isCadiEnabled())||(null != topicName && !topicName.startsWith("org.onap"))) {
169 permission = getPermission(arg1.toJava(), resource, topicName);
171 if (permission[0] != null) {
172 return !checkPermissions(fullName, topicName, permission);
177 private boolean checkPermissions(String fullName, String topicName, String[] permission) {
180 if (null != topicName) {
181 boolean hasResp = AuthorizationProviderFactory.getProviderFactory().getProvider()
182 .hasPermission(fullName, permission[0], permission[1], permission[2]);
184 logger.info("Successful Authorization for {} on {} for {} | {} | {}", fullName, topicName,
185 permission[0], permission[1], permission[2]);
188 logger.info("{} is not allowed in {} | {} | {}", fullName, permission[0], permission[1],
193 } catch (final Exception e) {
200 public void close() {
201 // TODO Auto-generated method stub
206 public scala.collection.immutable.Map<Resource, Set<Acl>> getAcls() {
207 // TODO Auto-generated method stub
212 public scala.collection.immutable.Map<Resource, Set<Acl>> getAcls(final KafkaPrincipal arg0) {
213 // TODO Auto-generated method stub
218 public boolean removeAcls(final Resource arg0) {
219 // TODO Auto-generated method stub
224 public boolean removeAcls(final Set<Acl> arg0, final Resource arg1) {
225 // TODO Auto-generated method stub
229 public Set<Acl> getAcls(Resource arg0) {
230 // TODO Auto-generated method stub