NIFI-2757: Site-to-Site with DN mapping

Added DN identity mapping pattern support to Site-to-Site client
authorization.

This closes #1010.

Signed-off-by: Bryan Bende <bbende@apache.org>

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java

@@ -1166,7 +1166,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
         name = requireNonNull(name).intern();
         verifyPortIdDoesNotExist(id);
         return new StandardRootGroupPort(id, name, null, TransferDirection.RECEIVE, ConnectableType.INPUT_PORT,
-                authorizer, getBulletinRepository(), processScheduler, Boolean.TRUE.equals(isSiteToSiteSecure));
+                authorizer, getBulletinRepository(), processScheduler, Boolean.TRUE.equals(isSiteToSiteSecure), nifiProperties);
     }
 
     /**
@@ -1186,7 +1186,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
         name = requireNonNull(name).intern();
         verifyPortIdDoesNotExist(id);
         return new StandardRootGroupPort(id, name, null, TransferDirection.SEND, ConnectableType.OUTPUT_PORT,
-                authorizer, getBulletinRepository(), processScheduler, Boolean.TRUE.equals(isSiteToSiteSecure));
+                authorizer, getBulletinRepository(), processScheduler, Boolean.TRUE.equals(isSiteToSiteSecure), nifiProperties);
     }
 
     /**

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/StandardRootGroupPort.java

@@ -24,6 +24,8 @@ import org.apache.nifi.authorization.resource.Authorizable;
 import org.apache.nifi.authorization.resource.DataTransferAuthorizable;
 import org.apache.nifi.authorization.user.NiFiUser;
 import org.apache.nifi.authorization.user.StandardNiFiUser;
+import org.apache.nifi.authorization.util.IdentityMapping;
+import org.apache.nifi.authorization.util.IdentityMappingUtil;
 import org.apache.nifi.components.ValidationResult;
 import org.apache.nifi.connectable.ConnectableType;
 import org.apache.nifi.controller.AbstractPort;
@@ -48,6 +50,7 @@ import org.apache.nifi.reporting.BulletinRepository;
 import org.apache.nifi.reporting.ComponentType;
 import org.apache.nifi.reporting.Severity;
 import org.apache.nifi.scheduling.SchedulingStrategy;
+import org.apache.nifi.util.NiFiProperties;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -57,6 +60,7 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
@@ -79,6 +83,8 @@ public class StandardRootGroupPort extends AbstractPort implements RootGroupPort
     private final ProcessScheduler processScheduler;
     private final boolean secure;
     private final Authorizer authorizer;
+    private final NiFiProperties nifiProperties;
+    private final List<IdentityMapping> identityMappings;
 
     @SuppressWarnings("unused")
     private final BulletinRepository bulletinRepository;
@@ -94,13 +100,16 @@ public class StandardRootGroupPort extends AbstractPort implements RootGroupPort
 
     public StandardRootGroupPort(final String id, final String name, final ProcessGroup processGroup,
             final TransferDirection direction, final ConnectableType type, final Authorizer authorizer,
-            final BulletinRepository bulletinRepository, final ProcessScheduler scheduler, final boolean secure) {
+            final BulletinRepository bulletinRepository, final ProcessScheduler scheduler, final boolean secure,
+            final NiFiProperties nifiProperties) {
         super(id, name, processGroup, type, scheduler);
 
         this.processScheduler = scheduler;
         setScheduldingPeriod(MINIMUM_SCHEDULING_NANOS + " nanos");
         this.authorizer = authorizer;
         this.secure = secure;
+        this.nifiProperties = nifiProperties;
+        this.identityMappings = IdentityMappingUtil.getIdentityMappings(nifiProperties);
         this.bulletinRepository = bulletinRepository;
         this.scheduler = scheduler;
         setYieldPeriod("100 millis");
@@ -357,7 +366,9 @@ public class StandardRootGroupPort extends AbstractPort implements RootGroupPort
             return new StandardPortAuthorizationResult(false, "User DN is not known");
         }
 
-        return checkUserAuthorization(new StandardNiFiUser(dn));
+        final String identity = IdentityMappingUtil.mapIdentity(dn, identityMappings);
+
+        return checkUserAuthorization(new StandardNiFiUser(identity));
     }
 
     @Override

nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestStandardRootGroupPort.java

@@ -0,0 +1,100 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.remote;
+
+import org.apache.nifi.authorization.AuthorizationRequest;
+import org.apache.nifi.authorization.AuthorizationResult;
+import org.apache.nifi.authorization.Authorizer;
+import org.apache.nifi.connectable.ConnectableType;
+import org.apache.nifi.controller.ProcessScheduler;
+import org.apache.nifi.groups.ProcessGroup;
+import org.apache.nifi.reporting.BulletinRepository;
+import org.apache.nifi.util.NiFiProperties;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+
+public class TestStandardRootGroupPort {
+
+    private RootGroupPort createRootGroupPort(NiFiProperties nifiProperties) {
+        final BulletinRepository bulletinRepository = mock(BulletinRepository.class);
+        final ProcessScheduler processScheduler = null;
+
+        final Authorizer authorizer = mock(Authorizer.class);
+        doAnswer(invocation -> {
+            final AuthorizationRequest request = invocation.getArgumentAt(0, AuthorizationRequest.class);
+            if ("node1@nifi.test".equals(request.getIdentity())) {
+                return AuthorizationResult.approved();
+            }
+            return AuthorizationResult.denied();
+        }).when(authorizer).authorize(any(AuthorizationRequest.class));
+
+        final ProcessGroup processGroup = mock(ProcessGroup.class);
+        doReturn("process-group-id").when(processGroup).getIdentifier();
+
+        return new StandardRootGroupPort("id", "name", processGroup,
+                TransferDirection.SEND, ConnectableType.INPUT_PORT, authorizer, bulletinRepository,
+                processScheduler, true, nifiProperties);
+    }
+
+    @Test
+    public void testCheckUserAuthorizationByDn() {
+
+        final NiFiProperties nifiProperties = mock(NiFiProperties.class);
+
+        final RootGroupPort port = createRootGroupPort(nifiProperties);
+
+        PortAuthorizationResult authResult = port.checkUserAuthorization("CN=node1, OU=nifi.test");
+        Assert.assertFalse(authResult.isAuthorized());
+
+        authResult = port.checkUserAuthorization("node1@nifi.test");
+        Assert.assertTrue(authResult.isAuthorized());
+    }
+
+    @Test
+    public void testCheckUserAuthorizationByMappedDn() {
+
+        final NiFiProperties nifiProperties = mock(NiFiProperties.class);
+        final String mapKey = ".dn";
+        Set<String> propertyKeys = new LinkedHashSet<>();
+        propertyKeys.add(NiFiProperties.SECURITY_IDENTITY_MAPPING_PATTERN_PREFIX + mapKey);
+        propertyKeys.add(NiFiProperties.SECURITY_IDENTITY_MAPPING_VALUE_PREFIX + mapKey);
+        doReturn(propertyKeys).when(nifiProperties).getPropertyKeys();
+
+        final String mapPattern = "^CN=(.*?), OU=(.*?)$";
+        final String mapValue = "$1@$2";
+        doReturn(mapPattern).when(nifiProperties).getProperty(eq(NiFiProperties.SECURITY_IDENTITY_MAPPING_PATTERN_PREFIX + mapKey));
+        doReturn(mapValue).when(nifiProperties).getProperty(eq(NiFiProperties.SECURITY_IDENTITY_MAPPING_VALUE_PREFIX + mapKey));
+
+        final RootGroupPort port = createRootGroupPort(nifiProperties);
+
+        PortAuthorizationResult authResult = port.checkUserAuthorization("CN=node2, OU=nifi.test");
+        Assert.assertFalse(authResult.isAuthorized());
+
+        authResult = port.checkUserAuthorization("CN=node1, OU=nifi.test");
+        Assert.assertTrue(authResult.isAuthorized());
+    }
+
+}