storm-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kabh...@apache.org
Subject [1/3] storm git commit: STORM-2555 handle impersonation properly for HBase delegation token
Date Sat, 24 Jun 2017 04:32:58 GMT
Repository: storm
Updated Branches:
  refs/heads/1.x-branch 0c2e229ad -> 1a6933eeb


STORM-2555 handle impersonation properly for HBase delegation token


Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/2544b6d9
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/2544b6d9
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/2544b6d9

Branch: refs/heads/1.x-branch
Commit: 2544b6d99f163cd2552485a629651f8e97fff8ee
Parents: 9588af4
Author: Jungtaek Lim <kabhwan@gmail.com>
Authored: Thu Jun 15 11:12:02 2017 +0900
Committer: Jungtaek Lim <kabhwan@gmail.com>
Committed: Thu Jun 22 15:12:27 2017 +0900

----------------------------------------------------------------------
 .../apache/storm/common/AbstractAutoCreds.java  |  4 +++
 .../apache/storm/hbase/security/AutoHBase.java  | 17 ++++++---
 .../storm/hbase/security/HBaseSecurityUtil.java |  2 +-
 .../apache/storm/hbase/common/HBaseClient.java  |  8 +----
 .../org/apache/storm/hbase/common/Utils.java    | 36 ++++++++++++++++++++
 .../hbase/trident/state/HBaseMapState.java      |  8 ++---
 6 files changed, 57 insertions(+), 18 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/storm/blob/2544b6d9/external/storm-autocreds/src/main/java/org/apache/storm/common/AbstractAutoCreds.java
----------------------------------------------------------------------
diff --git a/external/storm-autocreds/src/main/java/org/apache/storm/common/AbstractAutoCreds.java
b/external/storm-autocreds/src/main/java/org/apache/storm/common/AbstractAutoCreds.java
index 1506638..9421b13 100644
--- a/external/storm-autocreds/src/main/java/org/apache/storm/common/AbstractAutoCreds.java
+++ b/external/storm-autocreds/src/main/java/org/apache/storm/common/AbstractAutoCreds.java
@@ -214,6 +214,10 @@ public abstract class AbstractAutoCreds implements IAutoCredentials,
ICredential
                     if (allTokens != null) {
                         for (Token<? extends TokenIdentifier> token : allTokens) {
                             try {
+                                LOG.debug("Current user: {}", UserGroupInformation.getCurrentUser());
+                                LOG.debug("Token from credential: {} / {}", token.toString(),
+                                        token.decodeIdentifier().getUser());
+
                                 UserGroupInformation.getCurrentUser().addToken(token);
                                 LOG.info("Added delegation tokens to UGI.");
                             } catch (IOException e) {

http://git-wip-us.apache.org/repos/asf/storm/blob/2544b6d9/external/storm-autocreds/src/main/java/org/apache/storm/hbase/security/AutoHBase.java
----------------------------------------------------------------------
diff --git a/external/storm-autocreds/src/main/java/org/apache/storm/hbase/security/AutoHBase.java
b/external/storm-autocreds/src/main/java/org/apache/storm/hbase/security/AutoHBase.java
index fcbb463..59ee977 100644
--- a/external/storm-autocreds/src/main/java/org/apache/storm/hbase/security/AutoHBase.java
+++ b/external/storm-autocreds/src/main/java/org/apache/storm/hbase/security/AutoHBase.java
@@ -18,6 +18,8 @@
 
 package org.apache.storm.hbase.security;
 
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
 import org.apache.storm.Config;
 import org.apache.storm.common.AbstractAutoCreds;
 import org.apache.storm.hdfs.security.HdfsSecurityUtil;
@@ -111,20 +113,27 @@ public class AutoHBase extends AbstractAutoCreds {
                 }
                 provider.login(HBASE_KEYTAB_FILE_KEY, HBASE_PRINCIPAL_KEY, InetAddress.getLocalHost().getCanonicalHostName());
 
-                LOG.info("Logged into Hbase as principal = " + conf.get(HBASE_PRINCIPAL_KEY));
+                LOG.info("Logged into Hbase as principal = " + hbaseConf.get(HBASE_PRINCIPAL_KEY));
 
                 UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
 
                 final UserGroupInformation proxyUser = UserGroupInformation.createProxyUser(topologySubmitterUser,
ugi);
 
-                User user = User.create(ugi);
+                User user = User.create(proxyUser);
 
                 if(user.isHBaseSecurityEnabled(hbaseConf)) {
-                    TokenUtil.obtainAndCacheToken(hbaseConf, proxyUser);
+                    final Connection connection = ConnectionFactory.createConnection(hbaseConf,
user);
+                    TokenUtil.obtainAndCacheToken(connection, user);
 
                     LOG.info("Obtained HBase tokens, adding to user credentials.");
 
-                    Credentials credential= proxyUser.getCredentials();
+                    Credentials credential = proxyUser.getCredentials();
+
+                    for (Token<? extends TokenIdentifier> tokenForLog : credential.getAllTokens())
{
+                        LOG.debug("Obtained token info in credential: {} / {}",
+                                tokenForLog.toString(), tokenForLog.decodeIdentifier().getUser());
+                    }
+
                     ByteArrayOutputStream bao = new ByteArrayOutputStream();
                     ObjectOutputStream out = new ObjectOutputStream(bao);
                     credential.write(out);

http://git-wip-us.apache.org/repos/asf/storm/blob/2544b6d9/external/storm-autocreds/src/main/java/org/apache/storm/hbase/security/HBaseSecurityUtil.java
----------------------------------------------------------------------
diff --git a/external/storm-autocreds/src/main/java/org/apache/storm/hbase/security/HBaseSecurityUtil.java
b/external/storm-autocreds/src/main/java/org/apache/storm/hbase/security/HBaseSecurityUtil.java
index 4e0dcab..7c01653 100644
--- a/external/storm-autocreds/src/main/java/org/apache/storm/hbase/security/HBaseSecurityUtil.java
+++ b/external/storm-autocreds/src/main/java/org/apache/storm/hbase/security/HBaseSecurityUtil.java
@@ -67,7 +67,7 @@ public class HBaseSecurityUtil {
             }
             return legacyProvider;
         } else {
-            return UserProvider.instantiate(hbaseConfig);
+            return null;
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/storm/blob/2544b6d9/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/HBaseClient.java
----------------------------------------------------------------------
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/HBaseClient.java
b/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/HBaseClient.java
index f6f97b1..c73bc41 100644
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/HBaseClient.java
+++ b/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/HBaseClient.java
@@ -28,7 +28,6 @@ import org.slf4j.LoggerFactory;
 
 import java.io.Closeable;
 import java.io.IOException;
-import java.security.PrivilegedExceptionAction;
 import java.util.List;
 import java.util.Map;
 
@@ -40,12 +39,7 @@ public class HBaseClient implements Closeable{
     public HBaseClient(Map<String, Object> map , final Configuration configuration,
final String tableName) {
         try {
             UserProvider provider = HBaseSecurityUtil.login(map, configuration);
-            this.table = provider.getCurrent().getUGI().doAs(new PrivilegedExceptionAction<HTable>()
{
-                @Override
-                public HTable run() throws IOException {
-                    return new HTable(configuration, tableName);
-                }
-            });
+            this.table = Utils.getTable(provider, configuration, tableName);
         } catch(Exception e) {
             throw new RuntimeException("HBase bolt preparation failed: " + e.getMessage(),
e);
         }

http://git-wip-us.apache.org/repos/asf/storm/blob/2544b6d9/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/Utils.java
----------------------------------------------------------------------
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/Utils.java b/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/Utils.java
index 8efe098..b4851f8 100644
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/Utils.java
+++ b/external/storm-hbase/src/main/java/org/apache/storm/hbase/common/Utils.java
@@ -17,17 +17,53 @@
  */
 package org.apache.storm.hbase.common;
 
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.security.UserProvider;
 import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.security.token.TokenIdentifier;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
 import java.math.BigDecimal;
+import java.security.PrivilegedExceptionAction;
 
 public class Utils {
     private static final Logger LOG = LoggerFactory.getLogger(Utils.class);
 
     private Utils(){}
 
+    public static HTable getTable(UserProvider provider, final Configuration config, final
String tableName)
+            throws IOException, InterruptedException {
+        UserGroupInformation ugi;
+        if (provider != null) {
+            ugi = provider.getCurrent().getUGI();
+            LOG.debug("Current USER for provider: {}", ugi.getUserName());
+        } else {
+            // autocreds puts delegation token into current user UGI
+            ugi = UserGroupInformation.getCurrentUser();
+
+            LOG.debug("UGI for current USER : {}", ugi.getUserName());
+            for (Token<? extends TokenIdentifier> token : ugi.getTokens()) {
+                LOG.debug("Token in UGI (delegation token): {} / {}", token.toString(),
+                        token.decodeIdentifier().getUser());
+
+                // use UGI from token
+                ugi = token.decodeIdentifier().getUser();
+                ugi.addToken(token);
+            }
+        }
+
+        return ugi.doAs(new PrivilegedExceptionAction<HTable>() {
+            @Override public HTable run() throws IOException {
+                return new HTable(config, tableName);
+            }
+        });
+    }
+
     public static long toLong(Object obj){
         long l = 0;
         if(obj != null){

http://git-wip-us.apache.org/repos/asf/storm/blob/2544b6d9/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/state/HBaseMapState.java
----------------------------------------------------------------------
diff --git a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/state/HBaseMapState.java
b/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/state/HBaseMapState.java
index 541fa86..678753c 100644
--- a/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/state/HBaseMapState.java
+++ b/external/storm-hbase/src/main/java/org/apache/storm/hbase/trident/state/HBaseMapState.java
@@ -17,6 +17,7 @@
  */
 package org.apache.storm.hbase.trident.state;
 
+import org.apache.storm.hbase.common.Utils;
 import org.apache.storm.task.IMetricsContext;
 import org.apache.storm.topology.FailedException;
 import org.apache.storm.tuple.Values;
@@ -81,12 +82,7 @@ public class HBaseMapState<T> implements IBackingMap<T> {
 
         try{
             UserProvider provider = HBaseSecurityUtil.login(map, hbConfig);
-            this.table = provider.getCurrent().getUGI().doAs(new PrivilegedExceptionAction<HTable>()
{
-                @Override
-                public HTable run() throws IOException {
-                    return new HTable(hbConfig, options.tableName);
-                }
-            });
+            this.table = Utils.getTable(provider, hbConfig, options.tableName);
         } catch(Exception e){
             throw new RuntimeException("HBase bolt preparation failed: " + e.getMessage(),
e);
         }


Mime
View raw message