Use individual user credentials for data source authentication

hive/pom.xml

@@ -77,6 +77,11 @@
       <artifactId>hive-jdbc</artifactId>
       <version>${hive.hive.version}</version>
     </dependency>
+    <dependency>
+      <groupId>com.vertica</groupId>
+      <artifactId>vertica-jdk5</artifactId>
+      <version>6.1.3-0</version>
+    </dependency>
     <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-common</artifactId>

hive/src/main/java/org/apache/zeppelin/hive/HiveInterpreter.java

@@ -23,11 +23,9 @@ import java.sql.ResultSet;
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
 import java.sql.Statement;
-import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
-import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
 
@@ -74,30 +72,23 @@ public class HiveInterpreter extends Interpreter {
   static final String DEFAULT_PASSWORD = DEFAULT_KEY + DOT + PASSWORD_KEY;
 
   private final HashMap<String, Properties> propertiesMap;
-  private final Map<String, Statement> paragraphIdStatementMap;
-
-  private final Map<String, ArrayList<Connection>> propertyKeyUnusedConnectionListMap;
-  private final Map<String, Connection> paragraphIdConnectionMap;
 
   static {
     Interpreter.register(
-        "hql",
-        "hive",
-        HiveInterpreter.class.getName(),
-        new InterpreterPropertyBuilder()
-            .add(COMMON_MAX_LINE, MAX_LINE_DEFAULT, "Maximum line of results")
-            .add(DEFAULT_DRIVER, "org.apache.hive.jdbc.HiveDriver", "Hive JDBC driver")
-            .add(DEFAULT_URL, "jdbc:hive2://localhost:10000", "The URL for HiveServer2.")
-            .add(DEFAULT_USER, "hive", "The hive user")
-            .add(DEFAULT_PASSWORD, "", "The password for the hive user").build());
+            "hql",
+            "hive",
+            HiveInterpreter.class.getName(),
+            new InterpreterPropertyBuilder()
+                    .add(COMMON_MAX_LINE, MAX_LINE_DEFAULT, "Maximum line of results")
+                    .add(DEFAULT_DRIVER, "org.apache.hive.jdbc.HiveDriver", "Hive JDBC driver")
+                    .add(DEFAULT_URL, "jdbc:hive2://localhost:10000", "The URL for HiveServer2.")
+                    .add(DEFAULT_USER, "hive", "The hive user")
+                    .add(DEFAULT_PASSWORD, "", "The password for the hive user").build());
   }
 
   public HiveInterpreter(Properties property) {
     super(property);
     propertiesMap = new HashMap<>();
-    propertyKeyUnusedConnectionListMap = new HashMap<>();
-    paragraphIdStatementMap = new HashMap<>();
-    paragraphIdConnectionMap = new HashMap<>();
   }
 
   public HashMap<String, Properties> getPropertiesMap() {
@@ -143,101 +134,28 @@ public class HiveInterpreter extends Interpreter {
     logger.debug("propertiesMap: {}", propertiesMap);
   }
 
-  @Override
-  public void close() {
-    try {
-      for (List<Connection> connectionList : propertyKeyUnusedConnectionListMap.values()) {
-        for (Connection c : connectionList) {
-          c.close();
-        }
-      }
-
-      for (Statement statement : paragraphIdStatementMap.values()) {
-        statement.close();
-      }
-      paragraphIdStatementMap.clear();
-
-      for (Connection connection : paragraphIdConnectionMap.values()) {
-        connection.close();
-      }
-      paragraphIdConnectionMap.clear();
-
-    } catch (SQLException e) {
-      logger.error("Error while closing...", e);
-    }
-  }
-
-  public Connection getConnection(String propertyKey) throws ClassNotFoundException, SQLException {
+  public InterpreterResult executeSql(String propertyKey, String sql,
+                                      InterpreterContext interpreterContext) {
+    String executingUser = interpreterContext.getAuthenticationInfo().getDataSourceUser();
+    String password = interpreterContext.getAuthenticationInfo().getDataSourcePassword();
     Connection connection = null;
-    if (propertyKey == null || propertiesMap.get(propertyKey) == null) {
-      return null;
-    }
-    if (propertyKeyUnusedConnectionListMap.containsKey(propertyKey)) {
-      ArrayList<Connection> connectionList = propertyKeyUnusedConnectionListMap.get(propertyKey);
-      if (0 != connectionList.size()) {
-        connection = propertyKeyUnusedConnectionListMap.get(propertyKey).remove(0);
-        if (null != connection && connection.isClosed()) {
-          connection.close();
-          connection = null;
-        }
-      }
-    }
-    if (null == connection) {
+    Statement statement = null;
+    try {
       Properties properties = propertiesMap.get(propertyKey);
       Class.forName(properties.getProperty(DRIVER_KEY));
       String url = properties.getProperty(URL_KEY);
-      String user = properties.getProperty(USER_KEY);
-      String password = properties.getProperty(PASSWORD_KEY);
-      if (null != user && null != password) {
+      String user;
+      user = executingUser;
+      if (null != user) {
         connection = DriverManager.getConnection(url, user, password);
       } else {
         connection = DriverManager.getConnection(url, properties);
       }
-    }
-    return connection;
-  }
+      if (connection == null) {
+        return null;
+      }
 
-  public Statement getStatement(String propertyKey, String paragraphId)
-      throws SQLException, ClassNotFoundException {
-    Connection connection;
-    if (paragraphIdConnectionMap.containsKey(paragraphId)) {
-      // Never enter for now.
-      connection = paragraphIdConnectionMap.get(paragraphId);
-    } else {
-      connection = getConnection(propertyKey);
-    }
-    
-    if (connection == null) {
-      return null;
-    }
-
-    Statement statement = connection.createStatement();
-    if (isStatementClosed(statement)) {
-      connection = getConnection(propertyKey);
       statement = connection.createStatement();
-    }
-    paragraphIdConnectionMap.put(paragraphId, connection);
-    paragraphIdStatementMap.put(paragraphId, statement);
-
-    return statement;
-  }
-
-  private boolean isStatementClosed(Statement statement) {
-    try {
-      return statement.isClosed();
-    } catch (Throwable t) {
-      logger.debug("{} doesn't support isClosed method", statement);
-      return false;
-    }
-  }
-
-  public InterpreterResult executeSql(String propertyKey, String sql,
-                                      InterpreterContext interpreterContext) {
-    String paragraphId = interpreterContext.getParagraphId();
-
-    try {
-
-      Statement statement = getStatement(propertyKey, paragraphId);
 
       if (statement == null) {
         return new InterpreterResult(Code.ERROR, "Prefix not found.");
@@ -289,13 +207,14 @@ public class HiveInterpreter extends Interpreter {
           msg.append(updateCount).append(NEWLINE);
         }
       } finally {
-        try {
-          if (resultSet != null) {
-            resultSet.close();
-          }
+        if (resultSet != null) {
+          resultSet.close();
+        }
+        if (statement != null) {
           statement.close();
-        } finally {
-          moveConnectionToUnused(propertyKey, paragraphId);
+        }
+        if (connection != null) {
+          connection.close();
         }
       }
 
@@ -307,21 +226,6 @@ public class HiveInterpreter extends Interpreter {
     }
   }
 
-  private void moveConnectionToUnused(String propertyKey, String paragraphId) {
-    if (paragraphIdConnectionMap.containsKey(paragraphId)) {
-      Connection connection = paragraphIdConnectionMap.remove(paragraphId);
-      if (null != connection) {
-        if (propertyKeyUnusedConnectionListMap.containsKey(propertyKey)) {
-          propertyKeyUnusedConnectionListMap.get(propertyKey).add(connection);
-        } else {
-          ArrayList<Connection> connectionList = new ArrayList<>();
-          connectionList.add(connection);
-          propertyKeyUnusedConnectionListMap.put(propertyKey, connectionList);
-        }
-      }
-    }
-  }
-
   @Override
   public InterpreterResult interpret(String cmd, InterpreterContext contextInterpreter) {
     String propertyKey = getPropertyKey(cmd);
@@ -332,7 +236,8 @@ public class HiveInterpreter extends Interpreter {
 
     cmd = cmd.trim();
 
-    logger.info("PropertyKey: {}, SQL command: '{}'", propertyKey, cmd);
+    logger.info("PropertyKey: {} User: {} SQL command: '{}'", propertyKey,
+            contextInterpreter.getAuthenticationInfo().getUser(), cmd);
 
     return executeSql(propertyKey, cmd, contextInterpreter);
   }
@@ -358,19 +263,13 @@ public class HiveInterpreter extends Interpreter {
     }
   }
 
-  @Override
-  public void cancel(InterpreterContext context) {
-    String paragraphId = context.getParagraphId();
-    try {
-      paragraphIdStatementMap.get(paragraphId).cancel();
-    } catch (SQLException e) {
-      logger.error("Error while cancelling...", e);
-    }
+  public Connection getConnection(String propertyKey) throws ClassNotFoundException, SQLException {
+    return null;
   }
 
   @Override
-  public FormType getFormType() {
-    return FormType.SIMPLE;
+  public void close() {
+
   }
 
   @Override
@@ -378,10 +277,19 @@ public class HiveInterpreter extends Interpreter {
     return 0;
   }
 
+  @Override
+  public  void cancel(InterpreterContext context) {};
+
+
+  @Override
+  public FormType getFormType() {
+    return FormType.SIMPLE;
+  }
+
   @Override
   public Scheduler getScheduler() {
     return SchedulerFactory.singleton().createOrGetParallelScheduler(
-        HiveInterpreter.class.getName() + this.hashCode(), 10);
+        HiveInterpreter.class.getName() + this.hashCode(), 50);
   }
 
   @Override

zeppelin-interpreter/src/main/java/org/apache/zeppelin/user/AuthenticationInfo.java

@@ -24,6 +24,8 @@ package org.apache.zeppelin.user;
 public class AuthenticationInfo {
   String user;
   String ticket;
+  String dataSourceUser;
+  String dataSourcePassword;
 
   public AuthenticationInfo() {}
 
@@ -52,4 +54,21 @@ public class AuthenticationInfo {
   public void setTicket(String ticket) {
     this.ticket = ticket;
   }
+
+  public String getDataSourceUser() {
+    return dataSourceUser;
+  }
+
+  public void setDataSourceUser(String dataSourceUser) {
+    this.dataSourceUser = dataSourceUser;
+  }
+
+  public String getDataSourcePassword() {
+    return dataSourcePassword;
+  }
+
+  public void setDataSourcePassword(String dataSourcePassword) {
+    this.dataSourcePassword = dataSourcePassword;
+  }
+
 }

zeppelin-server/src/main/java/org/apache/zeppelin/rest/CredentialRestApi.java

@@ -0,0 +1,74 @@
+/*
+ * 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.zeppelin.rest;
+
+import com.google.gson.Gson;
+import com.google.gson.reflect.TypeToken;
+import org.apache.zeppelin.credential.Credentials;
+import org.apache.zeppelin.credential.UserCredentials;
+import org.apache.zeppelin.credential.UsernamePassword;
+import org.apache.zeppelin.server.JsonResponse;
+import org.apache.zeppelin.utils.SecurityUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.ws.rs.*;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.Response;
+import javax.ws.rs.core.Response.Status;
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * Credential Rest API
+ *
+ */
+@Path("/credential")
+@Produces("application/json")
+public class CredentialRestApi {
+  Logger logger = LoggerFactory.getLogger(CredentialRestApi.class);
+  private Credentials credentials;
+  private Gson gson = new Gson();
+
+  @Context
+  private HttpServletRequest servReq;
+
+  public CredentialRestApi() {
+    this.credentials = Credentials.getCredentials();
+  }
+
+  /**
+   * Update credentials for current user
+   */
+  @PUT
+  public Response putCredentials(String message) throws IOException {
+    Map<String, String> messageMap = gson.fromJson(message,
+            new TypeToken<Map<String, String>>(){}.getType());
+    String datasource = messageMap.get("datasource");
+    String username = messageMap.get("username");
+    String password = messageMap.get("password");
+    String user = SecurityUtils.getPrincipal();
+    logger.info("Update credentials for user {} datasource {}", user, datasource);
+    UserCredentials uc = credentials.getUserCredentials(user);
+    uc.putUsernamePassword(datasource, new UsernamePassword(username, password));
+    credentials.putUserCredentials(user, uc);
+    return new JsonResponse(Status.OK, "", "").build();
+  }
+
+}

zeppelin-server/src/main/java/org/apache/zeppelin/server/ZeppelinServer.java

@@ -292,6 +292,9 @@ public class ZeppelinServer extends Application {
     InterpreterRestApi interpreterApi = new InterpreterRestApi(replFactory);
     singletons.add(interpreterApi);
 
+    CredentialRestApi credentialApi = new CredentialRestApi();
+    singletons.add(credentialApi);
+
     SecurityRestApi securityApi = new SecurityRestApi();
     singletons.add(securityApi);
 

zeppelin-web/src/app/app.js

@@ -66,6 +66,10 @@
                     templateUrl: 'app/interpreter/interpreter.html',
                     controller: 'InterpreterCtrl'
                 })
+                .when('/credential', {
+                    templateUrl: 'app/credential/credential.html',
+                    controller: 'CredentialCtrl'
+                })
                 .when('/configuration', {
                   templateUrl: 'app/configuration/configuration.html',
                   controller: 'ConfigurationCtrl'

zeppelin-web/src/app/credential/credential.controller.js

@@ -0,0 +1,40 @@
+/* jshint loopfunc: true */
+/*
+ * Licensed 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.
+ */
+'use strict';
+
+angular.module('zeppelinWebApp').controller('CredentialCtrl', function($scope, $route, $routeParams, $location, $rootScope,
+                                                                         $http, baseUrlSrv) {
+  $scope._ = _;
+
+  $scope.updateCredentials = function() {
+    $http.put(baseUrlSrv.getRestApiBase() + '/credential',
+      { 'datasource': $scope.dataSource,
+        'username': $scope.credentialUsername,
+        'password': $scope.credentialPassword
+      } ).
+    success(function (data, status, headers, config) {
+      alert('Successfully saved credentials');
+      $scope.dataSource = '';
+      $scope.credentialUsername = '';
+      $scope.credentialPassword = '';
+      console.log('Success %o %o', status, data.message);
+    }).
+    error(function (data, status, headers, config) {
+      alert('Error saving credentials');
+      console.log('Error %o %o', status, data.message);
+    });
+  };
+
+});

zeppelin-web/src/app/credential/credential.css

@@ -0,0 +1,86 @@
+/*
+ * Licensed 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.
+ */
+
+.interpreterHead {
+  margin: -10px -10px 20px;
+  padding: 10px 15px 15px 15px;
+  background: white;
+  box-shadow: 0 2px 4px rgba(0, 0, 0, 0.15);
+  border-bottom: 1px solid #E5E5E5;
+}
+
+.interpreterHead .header {
+  font-family: 'Roboto', sans-serif;
+}
+
+.interpreterHead textarea,
+.interpreter textarea {
+  width: 100%;
+  display: block;
+  height: 20px;
+  resize: none;
+  border: 1px solid #CCCCCC;
+  font-size: 12px;
+}
+
+.interpreter .interpreter-title {
+  font-size: 20px;
+  font-weight: bold;
+  color: #3071a9;
+  float: left;
+  margin-top: 0;
+}
+
+.interpreter ul {
+  margin: 0;
+  padding: 0;
+}
+
+.interpreter .interpreterInfo {
+  list-style-type: none;
+}
+
+.interpreter table tr .interpreterPropertyKey {
+  padding : 5px 5px 5px 5px;
+}
+
+.interpreter table tr .interpreterPropertyValue {
+  padding : 5px 5px 5px 5px;
+  display: block;
+  max-height: 100px;
+  overflow-y: auto;
+}
+
+.interpreterSettingAdd {
+  margin : 5px 5px 5px 5px;
+  padding : 10px 10px 10px 10px;
+}
+
+.editable-wrap {
+  width : 100%;
+}
+
+.interpreter h5 {
+  font-weight: bold;
+}
+
+.new_h3 {
+  margin-top: 1px;
+  padding-top: 7px;
+  float: left;
+}
+
+.empty-properties-message {
+  color: #666;
+}

zeppelin-web/src/app/credential/credential.html

@@ -0,0 +1,73 @@
+<!--
+Licensed 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.
+-->
+<div class="interpreterHead">
+  <div class="header">
+    <div class="row">
+      <div class="col-md-12">
+        <h3 class="new_h3">
+          Credentials
+        </h3>
+      </div>
+    </div>
+    <div class="row">
+      <div class="col-md-12">
+        Add credentials for a data source one at a time.<br>
+        For security reasons, the credentials have to stay in memory and cannot be saved to disk.
+        Users will have to re-enter their credentials when the Zeppelin server is restarted.
+      </div>
+    </div>
+  </div>
+</div>
+
+<div class="box width-full home"
+     >
+  <div>
+    <div class="row interpreter">
+    <div class="col-md-12">
+        <table class="table table-striped">
+          <thead>
+            <tr>
+              <th style="width:30%">Data Source</th>
+              <th>username</th>
+              <th>password</th>
+              <th ng-if="valueform.$visible">action</th>
+            </tr>
+          </thead>
+          <tr>
+            <td>
+              <textarea msd-elastic ng-model="dataSource"></textarea>
+            </td>
+            <td>
+              <textarea msd-elastic ng-model="credentialUsername"></textarea>
+            </td>
+            <td>
+              <input type="password" ng-model="credentialPassword"/>
+            </td>
+          </tr>
+        </table>
+        <form editable-form name="valueform" onaftersave="updateCredentials()">
+          <button type="submit" class="btn btn-primary"
+                  ng-disabled="valueform.$waiting">
+            Save
+          </button>
+          <button type="button" class="btn btn-default"
+                  ng-disabled="valueform.$waiting"
+                  ng-click="valueform.$cancel()">
+            Cancel
+          </button>
+        </form>
+      </div>
+    </div>
+  </div>
+</div>

zeppelin-web/src/components/navbar/navbar.html

@@ -52,6 +52,9 @@ limitations under the License.
         <li>
           <a href="#/interpreter">Interpreter</a>
         </li>
+        <li>
+          <a href="#/credential">Credential</a>
+        </li>
         <li>
           <a href="#/configuration">Configuration</a>
         </li>

zeppelin-web/src/index.html

@@ -55,6 +55,7 @@ limitations under the License.
     <link rel="stylesheet" href="app/notebook/notebook.css">
     <link rel="stylesheet" href="app/notebook/paragraph/paragraph.css">
     <link rel="stylesheet" href="app/interpreter/interpreter.css">
+    <link rel="stylesheet" href="app/credential/credential.css">
     <link rel="stylesheet" href="app/configuration/configuration.css">
     <link rel="stylesheet" href="fonts/font-awesome.min.css">
     <link rel="stylesheet" href="fonts/simple-line-icons.css">
@@ -144,6 +145,7 @@ limitations under the License.
     <script src="app/home/home.controller.js"></script>
     <script src="app/notebook/notebook.controller.js"></script>
     <script src="app/interpreter/interpreter.controller.js"></script>
+    <script src="app/credential/credential.controller.js"></script>
     <script src="app/configuration/configuration.controller.js"></script>
     <script src="app/notebook/paragraph/paragraph.controller.js"></script>
     <script src="app/search/result-list.controller.js"></script>

zeppelin-zengine/src/main/java/org/apache/zeppelin/credential/Credentials.java

@@ -0,0 +1,63 @@
+/*
+* 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.zeppelin.credential;
+
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.*;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.zeppelin.conf.ZeppelinConfiguration;
+
+/**
+ * Class defining credentials for data source authorization
+ */
+public class Credentials {
+  private static final Logger LOG = LoggerFactory.getLogger(Credentials.class);
+
+  private static Credentials credentials = null;
+  private Map<String, UserCredentials> credentialsMap;
+
+  private Credentials() {
+    credentialsMap = new HashMap<>();
+  }
+
+  public static synchronized Credentials getCredentials() {
+    if (credentials == null) {
+      credentials = new Credentials();
+    }
+    return credentials;
+  }
+
+  public UserCredentials getUserCredentials(String username) {
+    UserCredentials uc = credentialsMap.get(username);
+    if (uc == null) {
+      uc = new UserCredentials();
+    }
+    return uc;
+  }
+
+  public void putUserCredentials(String username, UserCredentials uc) throws IOException {
+    credentialsMap.put(username, uc);
+  }
+
+}

zeppelin-zengine/src/main/java/org/apache/zeppelin/credential/UserCredentials.java

@@ -0,0 +1,47 @@
+/*
+ * 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.zeppelin.credential;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * User Credentials POJO
+ */
+public class UserCredentials {
+  private Map<String, UsernamePassword> userCredentials;
+
+  public UserCredentials() {
+    this.userCredentials = new HashMap<>();
+  }
+
+  public UsernamePassword getUsernamePassword(String dataSource) {
+    return userCredentials.get(dataSource);
+  }
+
+  public void putUsernamePassword(String dataSource, UsernamePassword up) {
+    userCredentials.put(dataSource, up);
+  }
+
+  @Override
+  public String toString() {
+    return "UserCredentials{" +
+        "userCredentials=" + userCredentials +
+        '}';
+  }
+}

zeppelin-zengine/src/main/java/org/apache/zeppelin/credential/UsernamePassword.java

@@ -0,0 +1,55 @@
+/*
+* 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.zeppelin.credential;
+
+/**
+ * Username and Password POJO
+ */
+public class UsernamePassword {
+  private String username;
+  private String password;
+
+  public UsernamePassword(String username, String password) {
+    this.username = username;
+    this.password = password;
+  }
+
+  public String getUsername() {
+    return username;
+  }
+
+  public void setUsername(String username) {
+    this.username = username;
+  }
+
+  public String getPassword() {
+    return password;
+  }
+
+  public void setPassword(String password) {
+    this.password = password;
+  }
+
+  @Override
+  public String toString() {
+    return "UsernamePassword{" +
+        "username='" + username + '\'' +
+        ", password='" + password + '\'' +
+        '}';
+  }
+}

zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Paragraph.java

@@ -20,6 +20,9 @@ package org.apache.zeppelin.notebook;
 import org.apache.zeppelin.display.AngularObject;
 import org.apache.zeppelin.display.AngularObjectRegistry;
 import org.apache.zeppelin.user.AuthenticationInfo;
+import org.apache.zeppelin.credential.Credentials;
+import org.apache.zeppelin.credential.UserCredentials;
+import org.apache.zeppelin.credential.UsernamePassword;
 import org.apache.zeppelin.display.GUI;
 import org.apache.zeppelin.display.Input;
 import org.apache.zeppelin.interpreter.*;
@@ -314,6 +317,29 @@ public class Paragraph extends Job implements Serializable, Cloneable {
     return true;
   }
 
+  /**
+   * Returns the property key of the specific data source
+   * e.g. getDataSourceKey("%hive(vertica)") -> "vertica"
+   * @param cmd
+   * @return property key of data source being queried
+   */
+  public static String getDataSourceKey(String cmd) {
+    if (cmd == null) {
+      return null;
+    }
+    int firstLineIndex = cmd.indexOf("\n");
+    if (-1 == firstLineIndex) {
+      firstLineIndex = cmd.length();
+    }
+    int configStartIndex = cmd.indexOf("(");
+    int configLastIndex = cmd.indexOf(")");
+    if (configStartIndex != -1 && configLastIndex != -1
+            && configLastIndex < firstLineIndex && configLastIndex < firstLineIndex) {
+      return cmd.substring(configStartIndex + 1, configLastIndex);
+    }
+    return null;
+  }
+
   private InterpreterContext getInterpreterContext() {
     AngularObjectRegistry registry = null;
     ResourcePool resourcePool = null;
@@ -330,6 +356,25 @@ public class Paragraph extends Job implements Serializable, Cloneable {
     }
 
     final Paragraph self = this;
+
+    Credentials credentials = Credentials.getCredentials();
+    String dataSourceKey = null;
+    try {
+      dataSourceKey = getDataSourceKey(getScriptBody());
+    } catch (NullPointerException e) {
+      logger().info("NPE at dataSourceKey({}). cmd = {}", dataSourceKey, getScriptBody());
+    }
+    logger().info("{} {}", getScriptBody(), dataSourceKey);
+    UserCredentials userCredentials = credentials.getUserCredentials(authenticationInfo.getUser());
+    logger().info(userCredentials.toString());
+    if (userCredentials != null) {
+      UsernamePassword userPassword = userCredentials.getUsernamePassword(dataSourceKey);
+      if (userPassword != null) {
+        authenticationInfo.setDataSourceUser(userPassword.getUsername());
+        authenticationInfo.setDataSourcePassword(userPassword.getPassword());
+      }
+    }
+
     InterpreterContext interpreterContext = new InterpreterContext(
             note.id(),
             getId(),