NIFI-5114: Added Basic auth support to WebSocket components

This closes #2652

Signed-off-by: Mike Thomsen <mikerthomsen@gmail.com>

nifi-nar-bundles/nifi-websocket-bundle/nifi-websocket-services-jetty/src/main/java/org/apache/nifi/websocket/jetty/JettyWebSocketClient.java

@@ -16,6 +16,7 @@
  */
 package org.apache.nifi.websocket.jetty;
 
+import org.apache.commons.codec.binary.Base64;
 import org.apache.nifi.annotation.documentation.CapabilityDescription;
 import org.apache.nifi.annotation.documentation.Tags;
 import org.apache.nifi.annotation.lifecycle.OnDisabled;
@@ -28,9 +29,11 @@ import org.apache.nifi.expression.ExpressionLanguageScope;
 import org.apache.nifi.logging.ComponentLog;
 import org.apache.nifi.processor.util.StandardValidators;
 import org.apache.nifi.ssl.SSLContextService;
+import org.apache.nifi.util.StringUtils;
 import org.apache.nifi.websocket.WebSocketClientService;
 import org.apache.nifi.websocket.WebSocketConfigurationException;
 import org.apache.nifi.websocket.WebSocketMessageRouter;
+import org.eclipse.jetty.http.HttpHeader;
 import org.eclipse.jetty.util.ssl.SslContextFactory;
 import org.eclipse.jetty.websocket.api.Session;
 import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
@@ -38,6 +41,7 @@ import org.eclipse.jetty.websocket.client.WebSocketClient;
 
 import java.io.IOException;
 import java.net.URI;
+import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -103,6 +107,35 @@ public class JettyWebSocketClient extends AbstractJettyWebSocketService implemen
             .defaultValue("10 sec")
             .build();
 
+    public static final PropertyDescriptor USER_NAME = new PropertyDescriptor.Builder()
+            .name("user-name")
+            .displayName("User Name")
+            .description("The user name for Basic Authentication.")
+            .required(false)
+            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+            .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR)
+            .build();
+
+    public static final PropertyDescriptor USER_PASSWORD = new PropertyDescriptor.Builder()
+            .name("user-password")
+            .displayName("User Password")
+            .description("The user password for Basic Authentication.")
+            .required(false)
+            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+            .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR)
+            .sensitive(true)
+            .build();
+
+    public static final PropertyDescriptor AUTH_CHARSET = new PropertyDescriptor.Builder()
+            .name("authentication-charset")
+            .displayName("Authentication Header Charset")
+            .description("The charset for Basic Authentication header base64 string.")
+            .required(true)
+            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+            .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR)
+            .defaultValue("US-ASCII")
+            .build();
+
     private static final List<PropertyDescriptor> properties;
 
     static {
@@ -112,12 +145,16 @@ public class JettyWebSocketClient extends AbstractJettyWebSocketService implemen
         props.add(SSL_CONTEXT);
         props.add(CONNECTION_TIMEOUT);
         props.add(SESSION_MAINTENANCE_INTERVAL);
+        props.add(USER_NAME);
+        props.add(USER_PASSWORD);
+        props.add(AUTH_CHARSET);
 
         properties = Collections.unmodifiableList(props);
     }
 
     private WebSocketClient client;
     private URI webSocketUri;
+    private String authorizationHeader;
     private long connectionTimeoutMillis;
     private volatile ScheduledExecutorService sessionMaintenanceScheduler;
     private final ReentrantLock connectionLock = new ReentrantLock();
@@ -139,6 +176,19 @@ public class JettyWebSocketClient extends AbstractJettyWebSocketService implemen
         client = new WebSocketClient(sslContextFactory);
 
         configurePolicy(context, client.getPolicy());
+        final String userName = context.getProperty(USER_NAME).evaluateAttributeExpressions().getValue();
+        final String userPassword = context.getProperty(USER_PASSWORD).evaluateAttributeExpressions().getValue();
+        if (!StringUtils.isEmpty(userName) && !StringUtils.isEmpty(userPassword)) {
+            final String charsetName = context.getProperty(AUTH_CHARSET).evaluateAttributeExpressions().getValue();
+            if (StringUtils.isEmpty(charsetName)) {
+                throw new IllegalArgumentException(AUTH_CHARSET.getDisplayName() + " was not specified.");
+            }
+            final Charset charset = Charset.forName(charsetName);
+            final String base64String = Base64.encodeBase64String((userName + ":" + userPassword).getBytes(charset));
+            authorizationHeader = "Basic " + base64String;
+        } else {
+            authorizationHeader = null;
+        }
 
         client.start();
         activeSessions.clear();
@@ -201,6 +251,9 @@ public class JettyWebSocketClient extends AbstractJettyWebSocketService implemen
             listener.setSessionId(sessionId);
 
             final ClientUpgradeRequest request = new ClientUpgradeRequest();
+            if (!StringUtils.isEmpty(authorizationHeader)) {
+                request.setHeader(HttpHeader.AUTHORIZATION.asString(), authorizationHeader);
+            }
             final Future<Session> connect = client.connect(listener, webSocketUri, request);
             getLogger().info("Connecting to : {}", new Object[]{webSocketUri});
 

nifi-nar-bundles/nifi-websocket-bundle/nifi-websocket-services-jetty/src/main/java/org/apache/nifi/websocket/jetty/JettyWebSocketServer.java

@@ -16,12 +16,6 @@
  */
 package org.apache.nifi.websocket.jetty;
 
-import java.net.URI;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
 import org.apache.nifi.annotation.documentation.CapabilityDescription;
 import org.apache.nifi.annotation.documentation.Tags;
 import org.apache.nifi.annotation.lifecycle.OnDisabled;
@@ -29,6 +23,8 @@ import org.apache.nifi.annotation.lifecycle.OnEnabled;
 import org.apache.nifi.annotation.lifecycle.OnShutdown;
 import org.apache.nifi.components.AllowableValue;
 import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
 import org.apache.nifi.controller.ConfigurationContext;
 import org.apache.nifi.expression.ExpressionLanguageScope;
 import org.apache.nifi.processor.util.StandardValidators;
@@ -36,6 +32,11 @@ import org.apache.nifi.ssl.SSLContextService;
 import org.apache.nifi.websocket.WebSocketConfigurationException;
 import org.apache.nifi.websocket.WebSocketMessageRouter;
 import org.apache.nifi.websocket.WebSocketServerService;
+import org.eclipse.jetty.security.ConstraintMapping;
+import org.eclipse.jetty.security.ConstraintSecurityHandler;
+import org.eclipse.jetty.security.DefaultAuthenticatorFactory;
+import org.eclipse.jetty.security.HashLoginService;
+import org.eclipse.jetty.security.LoginService;
 import org.eclipse.jetty.server.Connector;
 import org.eclipse.jetty.server.Handler;
 import org.eclipse.jetty.server.HttpConfiguration;
@@ -47,6 +48,7 @@ import org.eclipse.jetty.server.SslConnectionFactory;
 import org.eclipse.jetty.server.handler.ContextHandlerCollection;
 import org.eclipse.jetty.servlet.ServletContextHandler;
 import org.eclipse.jetty.servlet.ServletHandler;
+import org.eclipse.jetty.util.security.Constraint;
 import org.eclipse.jetty.util.ssl.SslContextFactory;
 import org.eclipse.jetty.websocket.api.Session;
 import org.eclipse.jetty.websocket.api.WebSocketPolicy;
@@ -56,6 +58,14 @@ import org.eclipse.jetty.websocket.servlet.WebSocketCreator;
 import org.eclipse.jetty.websocket.servlet.WebSocketServlet;
 import org.eclipse.jetty.websocket.servlet.WebSocketServletFactory;
 
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
 @Tags({"WebSocket", "Jetty", "server"})
 @CapabilityDescription("Implementation of WebSocketServerService." +
         " This service uses Jetty WebSocket server module to provide" +
@@ -73,13 +83,17 @@ public class JettyWebSocketServer extends AbstractJettyWebSocketService implemen
     public static final AllowableValue CLIENT_WANT = new AllowableValue("want", "Want Authentication",
             "Processor will try to verify the client but if unable to verify will allow the client to communicate anonymously");
     public static final AllowableValue CLIENT_NEED = new AllowableValue("need", "Need Authentication",
-            "Processor will reject communications from any client unless the client provides a certificate that is trusted by the TrustStore"
+            "Processor will reject communications from any client unless the client provides a certificate that is trusted by the TrustStore "
                     + "specified in the SSL Context Service");
 
+    public static final AllowableValue LOGIN_SERVICE_HASH = new AllowableValue("hash", "HashLoginService",
+            "See http://www.eclipse.org/jetty/javadoc/current/org/eclipse/jetty/security/HashLoginService.html for detail.");
+
     public static final PropertyDescriptor CLIENT_AUTH = new PropertyDescriptor.Builder()
             .name("client-authentication")
-            .displayName("Client Authentication")
-            .description("Specifies whether or not the Processor should authenticate clients. This value is ignored if the <SSL Context Service> "
+            .displayName("SSL Client Authentication")
+            .description("Specifies whether or not the Processor should authenticate client by its certificate. "
+                    + "This value is ignored if the <SSL Context Service> "
                     + "Property is not specified or the SSL Context provided uses only a KeyStore and not a TrustStore.")
             .required(true)
             .allowableValues(CLIENT_NONE, CLIENT_WANT, CLIENT_NEED)
@@ -95,6 +109,58 @@ public class JettyWebSocketServer extends AbstractJettyWebSocketService implemen
             .addValidator(StandardValidators.PORT_VALIDATOR)
             .build();
 
+    public static final PropertyDescriptor BASIC_AUTH = new PropertyDescriptor.Builder()
+            .name("basic-auth")
+            .displayName("Enable Basic Authentication")
+            .description("If enabled, client connection requests are authenticated with "
+                    + "Basic authentication using the specified Login Provider.")
+            .required(true)
+            .allowableValues("true", "false")
+            .defaultValue("false")
+            .build();
+
+    public static final PropertyDescriptor AUTH_PATH_SPEC = new PropertyDescriptor.Builder()
+            .name("auth-path-spec")
+            .displayName("Basic Authentication Path Spec")
+            .description("Specify a Path Spec to apply Basic Authentication.")
+            .required(false)
+            .defaultValue("/*")
+            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+            .build();
+
+    public static final PropertyDescriptor AUTH_ROLES = new PropertyDescriptor.Builder()
+            .name("auth-roles")
+            .displayName("Basic Authentication Roles")
+            .description("The authenticated user must have one of specified role. "
+                    + "Multiple roles can be set as comma separated string. "
+                    + "'*' represents any role and so does '**' any role including no role.")
+            .required(false)
+            .defaultValue("**")
+            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+            .build();
+
+    public static final PropertyDescriptor LOGIN_SERVICE = new PropertyDescriptor.Builder()
+            .name("login-service")
+            .displayName("Login Service")
+            .description("Specify which Login Service to use for Basic Authentication.")
+            .required(false)
+            .allowableValues(LOGIN_SERVICE_HASH)
+            .defaultValue(LOGIN_SERVICE_HASH.getValue())
+            .build();
+
+
+    public static final PropertyDescriptor USERS_PROPERTIES_FILE = new PropertyDescriptor.Builder()
+            .name("users-properties-file")
+            .displayName("Users Properties File")
+            .description("Specify a property file containing users for Basic Authentication using HashLoginService. "
+                    + "See http://www.eclipse.org/jetty/documentation/current/configuring-security.html for detail.")
+            .required(false)
+            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
+            .addValidator(StandardValidators.FILE_EXISTS_VALIDATOR)
+            .build();
+
     private static final List<PropertyDescriptor> properties;
 
     static {
@@ -103,6 +169,12 @@ public class JettyWebSocketServer extends AbstractJettyWebSocketService implemen
         props.add(LISTEN_PORT);
         props.add(SSL_CONTEXT);
         props.add(CLIENT_AUTH);
+        props.add(BASIC_AUTH);
+        props.add(AUTH_PATH_SPEC);
+        props.add(AUTH_ROLES);
+        props.add(LOGIN_SERVICE);
+        props.add(USERS_PROPERTIES_FILE);
+
 
         properties = Collections.unmodifiableList(props);
     }
@@ -118,6 +190,23 @@ public class JettyWebSocketServer extends AbstractJettyWebSocketService implemen
     }
 
 
+    @Override
+    protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
+
+        final List<ValidationResult> results = new ArrayList<>();
+        if (validationContext.getProperty(BASIC_AUTH).asBoolean()) {
+            final String loginServiceValue = validationContext.getProperty(LOGIN_SERVICE).getValue();
+            if (LOGIN_SERVICE_HASH.equals(loginServiceValue)) {
+                if (!validationContext.getProperty(USERS_PROPERTIES_FILE).isSet()) {
+                    results.add(new ValidationResult.Builder().subject(USERS_PROPERTIES_FILE.getDisplayName())
+                            .explanation("it is required by HashLoginService").valid(false).build());
+                }
+            }
+        }
+
+        return results;
+    }
+
     public static class JettyWebSocketServlet extends WebSocketServlet implements WebSocketCreator {
         @Override
         public void configure(WebSocketServletFactory webSocketServletFactory) {
@@ -174,6 +263,42 @@ public class JettyWebSocketServer extends AbstractJettyWebSocketService implemen
         final ContextHandlerCollection handlerCollection = new ContextHandlerCollection();
 
         final ServletContextHandler contextHandler = new ServletContextHandler();
+
+        // Add basic auth.
+        if (context.getProperty(BASIC_AUTH).asBoolean()) {
+            final ConstraintSecurityHandler securityHandler = new ConstraintSecurityHandler();
+            contextHandler.insertHandler(securityHandler);
+
+            final Constraint constraint = new Constraint();
+            constraint.setName("auth");
+            constraint.setAuthenticate(true);
+            // Accessible from any role and any auth once authentication succeeds.
+            final String roles = context.getProperty(AUTH_ROLES).evaluateAttributeExpressions().getValue();
+            constraint.setRoles(roles.split(","));
+
+            final ConstraintMapping constraintMapping = new ConstraintMapping();
+            constraintMapping.setPathSpec(context.getProperty(AUTH_PATH_SPEC).evaluateAttributeExpressions().getValue());
+            constraintMapping.setConstraint(constraint);
+
+            final DefaultAuthenticatorFactory authenticatorFactory = new DefaultAuthenticatorFactory();
+            securityHandler.setAuthenticatorFactory(authenticatorFactory);
+            securityHandler.setAuthMethod(Constraint.__BASIC_AUTH);
+            securityHandler.setRealmName(getClass().getSimpleName());
+            securityHandler.setConstraintMappings(Collections.singletonList(constraintMapping));
+
+            final LoginService loginService;
+            final String loginServiceValue = context.getProperty(LOGIN_SERVICE).getValue();
+            if (LOGIN_SERVICE_HASH.equals(loginServiceValue)) {
+                final String usersFilePath = context.getProperty(USERS_PROPERTIES_FILE).evaluateAttributeExpressions().getValue();
+                loginService = new HashLoginService("HashLoginService", usersFilePath);
+            } else {
+                throw new IllegalArgumentException("Unsupported Login Service: " + loginServiceValue);
+            }
+
+            server.addBean(loginService);
+            securityHandler.setLoginService(loginService);
+        }
+
         servletHandler = new ServletHandler();
         contextHandler.insertHandler(servletHandler);
 
@@ -181,6 +306,7 @@ public class JettyWebSocketServer extends AbstractJettyWebSocketService implemen
 
         server.setHandler(handlerCollection);
 
+
         listenPort = context.getProperty(LISTEN_PORT).evaluateAttributeExpressions().asInteger();
         final SslContextFactory sslContextFactory = createSslFactory(context);
 

nifi-nar-bundles/nifi-websocket-bundle/nifi-websocket-services-jetty/src/test/java/org/apache/nifi/websocket/jetty/ITJettyWebSocketCommunication.java

@@ -74,6 +74,10 @@ public class ITJettyWebSocketCommunication {
         serverService = new JettyWebSocketServer();
         serverServiceContext = new ControllerServiceTestContext(serverService, "JettyWebSocketServer1");
         serverServiceContext.setCustomValue(JettyWebSocketServer.LISTEN_PORT, String.valueOf(serverPort));
+        serverServiceContext.setCustomValue(JettyWebSocketServer.BASIC_AUTH, "true");
+        serverServiceContext.setCustomValue(JettyWebSocketServer.USERS_PROPERTIES_FILE,
+                getClass().getResource("/users.properties").getPath());
+        serverServiceContext.setCustomValue(JettyWebSocketServer.AUTH_ROLES, "user,test");
 
         customizeServer();
 
@@ -89,6 +93,9 @@ public class ITJettyWebSocketCommunication {
         clientServiceContext = new ControllerServiceTestContext(clientService, "JettyWebSocketClient1");
         clientServiceContext.setCustomValue(JettyWebSocketClient.WS_URI, (isSecure() ? "wss" : "ws") + "://localhost:" + serverPort + serverPath);
 
+        clientServiceContext.setCustomValue(JettyWebSocketClient.USER_NAME, "user2");
+        clientServiceContext.setCustomValue(JettyWebSocketClient.USER_PASSWORD, "password2");
+
         customizeClient();
 
         clientService.initialize(clientServiceContext.getInitializationContext());

nifi-nar-bundles/nifi-websocket-bundle/nifi-websocket-services-jetty/src/test/java/org/apache/nifi/websocket/jetty/TestJettyWebSocketServer.java

@@ -37,6 +37,20 @@ public class TestJettyWebSocketServer {
         assertEquals(JettyWebSocketServer.LISTEN_PORT.getDisplayName(), result.getSubject());
     }
 
+    @Test
+    public void testValidationHashLoginService() throws Exception {
+        final JettyWebSocketServer service = new JettyWebSocketServer();
+        final ControllerServiceTestContext context = new ControllerServiceTestContext(service, "service-id");
+        context.setCustomValue(JettyWebSocketServer.LISTEN_PORT, "9001");
+        context.setCustomValue(JettyWebSocketServer.LOGIN_SERVICE, "hash");
+        context.setCustomValue(JettyWebSocketServer.BASIC_AUTH, "true");
+        service.initialize(context.getInitializationContext());
+        final Collection<ValidationResult> results = service.validate(context.getValidationContext());
+        assertEquals(1, results.size());
+        final ValidationResult result = results.iterator().next();
+        assertEquals(JettyWebSocketServer.USERS_PROPERTIES_FILE.getDisplayName(), result.getSubject());
+    }
+
     @Test
     public void testValidationSuccess() throws Exception {
         final JettyWebSocketServer service = new JettyWebSocketServer();

nifi-nar-bundles/nifi-websocket-bundle/nifi-websocket-services-jetty/src/test/resources/users.properties

@@ -0,0 +1,20 @@
+#
+# 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.
+#
+user1=password1,admin
+user2=password2,user
+# Not associated with any role
+user3=password3