hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/SecureBulkLoadManager.java

   1 /**
   2  * Licensed to the Apache Software Foundation (ASF) under one
   3  * or more contributor license agreements.  See the NOTICE file
   4  * distributed with this work for additional information
   5  * regarding copyright ownership.  The ASF licenses this file
   6  * to you under the Apache License, Version 2.0 (the
   7  * "License"); you may not use this file except in compliance
   8  * with the License.  You may obtain a copy of the License at
   9  *
  10  *     http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  */
  18 package org.apache.hadoop.hbase.regionserver;
  19
  20 import java.io.IOException;
  21 import java.math.BigInteger;
  22 import java.security.PrivilegedAction;
  23 import java.security.SecureRandom;
  24 import java.util.ArrayList;
  25 import java.util.HashMap;
  26 import java.util.List;
  27 import java.util.Map;
  28 import java.util.concurrent.ConcurrentHashMap;
  29 import java.util.function.Consumer;
  30 import org.apache.commons.lang3.mutable.MutableInt;
  31 import org.apache.hadoop.conf.Configuration;
  32 import org.apache.hadoop.fs.FileStatus;
  33 import org.apache.hadoop.fs.FileSystem;
  34 import org.apache.hadoop.fs.FileUtil;
  35 import org.apache.hadoop.fs.Path;
  36 import org.apache.hadoop.fs.permission.FsPermission;
  37 import org.apache.hadoop.hbase.DoNotRetryIOException;
  38 import org.apache.hadoop.hbase.HConstants;
  39 import org.apache.hadoop.hbase.TableName;
  40 import org.apache.hadoop.hbase.client.AsyncConnection;
  41 import org.apache.hadoop.hbase.ipc.RpcServer;
  42 import org.apache.hadoop.hbase.regionserver.HRegion.BulkLoadListener;
  43 import org.apache.hadoop.hbase.security.User;
  44 import org.apache.hadoop.hbase.security.UserProvider;
  45 import org.apache.hadoop.hbase.security.token.AuthenticationTokenIdentifier;
  46 import org.apache.hadoop.hbase.security.token.ClientTokenUtil;
  47 import org.apache.hadoop.hbase.security.token.FsDelegationToken;
  48 import org.apache.hadoop.hbase.util.Bytes;
  49 import org.apache.hadoop.hbase.util.CommonFSUtils;
  50 import org.apache.hadoop.hbase.util.FSUtils;
  51 import org.apache.hadoop.hbase.util.Methods;
  52 import org.apache.hadoop.hbase.util.Pair;
  53 import org.apache.hadoop.io.Text;
  54 import org.apache.hadoop.security.UserGroupInformation;
  55 import org.apache.hadoop.security.token.Token;
  56 import org.apache.yetus.audience.InterfaceAudience;
  57 import org.slf4j.Logger;
  58 import org.slf4j.LoggerFactory;
  59
  60 import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos;
  61 import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.BulkLoadHFileRequest;
  62 import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.CleanupBulkLoadRequest;
  63 import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.PrepareBulkLoadRequest;
  64
  65 /**
  66  * Bulk loads in secure mode.
  67  *
  68  * This service addresses two issues:
  69  * <ol>
  70  * <li>Moving files in a secure filesystem wherein the HBase Client
  71  * and HBase Server are different filesystem users.</li>
  72  * <li>Does moving in a secure manner. Assuming that the filesystem
  73  * is POSIX compliant.</li>
  74  * </ol>
  75  *
  76  * The algorithm is as follows:
  77  * <ol>
  78  * <li>Create an hbase owned staging directory which is
  79  * world traversable (711): {@code /hbase/staging}</li>
  80  * <li>A user writes out data to his secure output directory: {@code /user/foo/data}</li>
  81  * <li>A call is made to hbase to create a secret staging directory
  82  * which globally rwx (777): {@code /user/staging/averylongandrandomdirectoryname}</li>
  83  * <li>The user moves the data into the random staging directory,
  84  * then calls bulkLoadHFiles()</li>
  85  * </ol>
  86  *
  87  * Like delegation tokens the strength of the security lies in the length
  88  * and randomness of the secret directory.
  89  *
  90  */
  91 @InterfaceAudience.Private
  92 public class SecureBulkLoadManager {
  93
  94   public static final long VERSION = 0L;
  95
  96   //320/5 = 64 characters
  97   private static final int RANDOM_WIDTH = 320;
  98   private static final int RANDOM_RADIX = 32;
  99
 100   private static final Logger LOG = LoggerFactory.getLogger(SecureBulkLoadManager.class);
 101
 102   private final static FsPermission PERM_ALL_ACCESS = FsPermission.valueOf("-rwxrwxrwx");
 103   private final static FsPermission PERM_HIDDEN = FsPermission.valueOf("-rwx--x--x");
 104   private SecureRandom random;
 105   private FileSystem fs;
 106   private Configuration conf;
 107
 108   //two levels so it doesn't get deleted accidentally
 109   //no sticky bit in Hadoop 1.0
 110   private Path baseStagingDir;
 111
 112   private UserProvider userProvider;
 113   private ConcurrentHashMap<UserGroupInformation, MutableInt> ugiReferenceCounter;
 114   private AsyncConnection conn;
 115
 116   SecureBulkLoadManager(Configuration conf, AsyncConnection conn) {
 117     this.conf = conf;
 118     this.conn = conn;
 119   }
 120
 121   public void start() throws IOException {
 122     random = new SecureRandom();
 123     userProvider = UserProvider.instantiate(conf);
 124     ugiReferenceCounter = new ConcurrentHashMap<>();
 125     fs = FileSystem.get(conf);
 126     baseStagingDir = new Path(CommonFSUtils.getRootDir(conf), HConstants.BULKLOAD_STAGING_DIR_NAME);
 127
 128     if (conf.get("hbase.bulkload.staging.dir") != null) {
 129       LOG.warn("hbase.bulkload.staging.dir " + " is deprecated. Bulkload staging directory is "
 130           + baseStagingDir);
 131     }
 132     if (!fs.exists(baseStagingDir)) {
 133       fs.mkdirs(baseStagingDir, PERM_HIDDEN);
 134     }
 135   }
 136
 137   public void stop() throws IOException {
 138   }
 139
 140   public String prepareBulkLoad(final HRegion region, final PrepareBulkLoadRequest request)
 141       throws IOException {
 142     User user = getActiveUser();
 143     region.getCoprocessorHost().prePrepareBulkLoad(user);
 144
 145     String bulkToken =
 146         createStagingDir(baseStagingDir, user, region.getTableDescriptor().getTableName())
 147             .toString();
 148
 149     return bulkToken;
 150   }
 151
 152   public void cleanupBulkLoad(final HRegion region, final CleanupBulkLoadRequest request)
 153       throws IOException {
 154     region.getCoprocessorHost().preCleanupBulkLoad(getActiveUser());
 155
 156     Path path = new Path(request.getBulkToken());
 157     if (!fs.delete(path, true)) {
 158       if (fs.exists(path)) {
 159         throw new IOException("Failed to clean up " + path);
 160       }
 161     }
 162     LOG.trace("Cleaned up {} successfully.", path);
 163   }
 164
 165   private Consumer<HRegion> fsCreatedListener;
 166
 167   void setFsCreatedListener(Consumer<HRegion> fsCreatedListener) {
 168     this.fsCreatedListener = fsCreatedListener;
 169   }
 170
 171   private void incrementUgiReference(UserGroupInformation ugi) {
 172     // if we haven't seen this ugi before, make a new counter
 173     ugiReferenceCounter.compute(ugi, (key, value) -> {
 174       if (value == null) {
 175         value = new MutableInt(1);
 176       } else {
 177         value.increment();
 178       }
 179       return value;
 180     });
 181   }
 182
 183   private void decrementUgiReference(UserGroupInformation ugi) {
 184     // if the count drops below 1 we remove the entry by returning null
 185     ugiReferenceCounter.computeIfPresent(ugi, (key, value) -> {
 186       if (value.intValue() > 1) {
 187         value.decrement();
 188       } else {
 189         value = null;
 190       }
 191       return value;
 192     });
 193   }
 194
 195   private boolean isUserReferenced(UserGroupInformation ugi) {
 196     // if the ugi is in the map, based on invariants above
 197     // the count must be above zero
 198     return ugiReferenceCounter.containsKey(ugi);
 199   }
 200
 201   public Map<byte[], List<Path>> secureBulkLoadHFiles(final HRegion region,
 202     final BulkLoadHFileRequest request) throws IOException {
 203     return secureBulkLoadHFiles(region, request, null);
 204   }
 205
 206   public Map<byte[], List<Path>> secureBulkLoadHFiles(final HRegion region,
 207       final BulkLoadHFileRequest request, List<String> clusterIds) throws IOException {
 208     final List<Pair<byte[], String>> familyPaths = new ArrayList<>(request.getFamilyPathCount());
 209     for(ClientProtos.BulkLoadHFileRequest.FamilyPath el : request.getFamilyPathList()) {
 210       familyPaths.add(new Pair<>(el.getFamily().toByteArray(), el.getPath()));
 211     }
 212
 213     Token<AuthenticationTokenIdentifier> userToken = null;
 214     if (userProvider.isHadoopSecurityEnabled()) {
 215       userToken = new Token<>(request.getFsToken().getIdentifier().toByteArray(),
 216         request.getFsToken().getPassword().toByteArray(), new Text(request.getFsToken().getKind()),
 217         new Text(request.getFsToken().getService()));
 218     }
 219     final String bulkToken = request.getBulkToken();
 220     User user = getActiveUser();
 221     final UserGroupInformation ugi = user.getUGI();
 222     if (userProvider.isHadoopSecurityEnabled()) {
 223       try {
 224         Token<AuthenticationTokenIdentifier> tok = ClientTokenUtil.obtainToken(conn).get();
 225         if (tok != null) {
 226           boolean b = ugi.addToken(tok);
 227           LOG.debug("token added " + tok + " for user " + ugi + " return=" + b);
 228         }
 229       } catch (Exception ioe) {
 230         LOG.warn("unable to add token", ioe);
 231       }
 232     }
 233     if (userToken != null) {
 234       ugi.addToken(userToken);
 235     } else if (userProvider.isHadoopSecurityEnabled()) {
 236       //we allow this to pass through in "simple" security mode
 237       //for mini cluster testing
 238       throw new DoNotRetryIOException("User token cannot be null");
 239     }
 240
 241     if (region.getCoprocessorHost() != null) {
 242       region.getCoprocessorHost().preBulkLoadHFile(familyPaths);
 243     }
 244     Map<byte[], List<Path>> map = null;
 245
 246     try {
 247       incrementUgiReference(ugi);
 248       // Get the target fs (HBase region server fs) delegation token
 249       // Since we have checked the permission via 'preBulkLoadHFile', now let's give
 250       // the 'request user' necessary token to operate on the target fs.
 251       // After this point the 'doAs' user will hold two tokens, one for the source fs
 252       // ('request user'), another for the target fs (HBase region server principal).
 253       if (userProvider.isHadoopSecurityEnabled()) {
 254         FsDelegationToken targetfsDelegationToken = new FsDelegationToken(userProvider,"renewer");
 255         targetfsDelegationToken.acquireDelegationToken(fs);
 256
 257         Token<?> targetFsToken = targetfsDelegationToken.getUserToken();
 258         if (targetFsToken != null
 259             && (userToken == null || !targetFsToken.getService().equals(userToken.getService()))){
 260           ugi.addToken(targetFsToken);
 261         }
 262       }
 263
 264       map = ugi.doAs(new PrivilegedAction<Map<byte[], List<Path>>>() {
 265         @Override
 266         public Map<byte[], List<Path>> run() {
 267           FileSystem fs = null;
 268           try {
 269             /*
 270              * This is creating and caching a new FileSystem instance. Other code called
 271              * "beneath" this method will rely on this FileSystem instance being in the
 272              * cache. This is important as those methods make _no_ attempt to close this
 273              * FileSystem instance. It is critical that here, in SecureBulkLoadManager,
 274              * we are tracking the lifecycle and closing the FS when safe to do so.
 275              */
 276             fs = FileSystem.get(conf);
 277             for(Pair<byte[], String> el: familyPaths) {
 278               Path stageFamily = new Path(bulkToken, Bytes.toString(el.getFirst()));
 279               if(!fs.exists(stageFamily)) {
 280                 fs.mkdirs(stageFamily);
 281                 fs.setPermission(stageFamily, PERM_ALL_ACCESS);
 282               }
 283             }
 284             if (fsCreatedListener != null) {
 285               fsCreatedListener.accept(region);
 286             }
 287             //We call bulkLoadHFiles as requesting user
 288             //To enable access prior to staging
 289             return region.bulkLoadHFiles(familyPaths, true,
 290                 new SecureBulkLoadListener(fs, bulkToken, conf), request.getCopyFile(),
 291               clusterIds, request.getReplicate());
 292           } catch (Exception e) {
 293             LOG.error("Failed to complete bulk load", e);
 294           }
 295           return null;
 296         }
 297       });
 298     } finally {
 299       decrementUgiReference(ugi);
 300       try {
 301         if (!UserGroupInformation.getLoginUser().equals(ugi) && !isUserReferenced(ugi)) {
 302           FileSystem.closeAllForUGI(ugi);
 303         }
 304       } catch (IOException e) {
 305         LOG.error("Failed to close FileSystem for: {}", ugi, e);
 306       }
 307       if (region.getCoprocessorHost() != null) {
 308         region.getCoprocessorHost().postBulkLoadHFile(familyPaths, map);
 309       }
 310     }
 311     return map;
 312   }
 313
 314   private Path createStagingDir(Path baseDir,
 315                                 User user,
 316                                 TableName tableName) throws IOException {
 317     String tblName = tableName.getNameAsString().replace(":", "_");
 318     String randomDir = user.getShortName()+"__"+ tblName +"__"+
 319         (new BigInteger(RANDOM_WIDTH, random).toString(RANDOM_RADIX));
 320     return createStagingDir(baseDir, user, randomDir);
 321   }
 322
 323   private Path createStagingDir(Path baseDir,
 324                                 User user,
 325                                 String randomDir) throws IOException {
 326     Path p = new Path(baseDir, randomDir);
 327     fs.mkdirs(p, PERM_ALL_ACCESS);
 328     fs.setPermission(p, PERM_ALL_ACCESS);
 329     return p;
 330   }
 331
 332   private User getActiveUser() throws IOException {
 333     // for non-rpc handling, fallback to system user
 334     User user = RpcServer.getRequestUser().orElse(userProvider.getCurrent());
 335     // this is for testing
 336     if (userProvider.isHadoopSecurityEnabled() &&
 337         "simple".equalsIgnoreCase(conf.get(User.HBASE_SECURITY_CONF_KEY))) {
 338       return User.createUserForTesting(conf, user.getShortName(), new String[] {});
 339     }
 340
 341     return user;
 342   }
 343
 344   private static class SecureBulkLoadListener implements BulkLoadListener {
 345     // Target filesystem
 346     private final FileSystem fs;
 347     private final String stagingDir;
 348     private final Configuration conf;
 349     // Source filesystem
 350     private FileSystem srcFs = null;
 351     private Map<String, FsPermission> origPermissions = null;
 352
 353     public SecureBulkLoadListener(FileSystem fs, String stagingDir, Configuration conf) {
 354       this.fs = fs;
 355       this.stagingDir = stagingDir;
 356       this.conf = conf;
 357       this.origPermissions = new HashMap<>();
 358     }
 359
 360     @Override
 361     public String prepareBulkLoad(final byte[] family, final String srcPath, boolean copyFile)
 362         throws IOException {
 363       Path p = new Path(srcPath);
 364       Path stageP = new Path(stagingDir, new Path(Bytes.toString(family), p.getName()));
 365
 366       // In case of Replication for bulk load files, hfiles are already copied in staging directory
 367       if (p.equals(stageP)) {
 368         LOG.debug(p.getName()
 369             + " is already available in staging directory. Skipping copy or rename.");
 370         return stageP.toString();
 371       }
 372
 373       if (srcFs == null) {
 374         srcFs = FileSystem.newInstance(p.toUri(), conf);
 375       }
 376
 377       if(!isFile(p)) {
 378         throw new IOException("Path does not reference a file: " + p);
 379       }
 380
 381       // Check to see if the source and target filesystems are the same
 382       if (!FSUtils.isSameHdfs(conf, srcFs, fs)) {
 383         LOG.debug("Bulk-load file " + srcPath + " is on different filesystem than " +
 384             "the destination filesystem. Copying file over to destination staging dir.");
 385         FileUtil.copy(srcFs, p, fs, stageP, false, conf);
 386       } else if (copyFile) {
 387         LOG.debug("Bulk-load file " + srcPath + " is copied to destination staging dir.");
 388         FileUtil.copy(srcFs, p, fs, stageP, false, conf);
 389       } else {
 390         LOG.debug("Moving " + p + " to " + stageP);
 391         FileStatus origFileStatus = fs.getFileStatus(p);
 392         origPermissions.put(srcPath, origFileStatus.getPermission());
 393         if(!fs.rename(p, stageP)) {
 394           throw new IOException("Failed to move HFile: " + p + " to " + stageP);
 395         }
 396       }
 397       fs.setPermission(stageP, PERM_ALL_ACCESS);
 398       return stageP.toString();
 399     }
 400
 401     @Override
 402     public void doneBulkLoad(byte[] family, String srcPath) throws IOException {
 403       LOG.debug("Bulk Load done for: " + srcPath);
 404       closeSrcFs();
 405     }
 406
 407     private void closeSrcFs() throws IOException {
 408       if (srcFs != null) {
 409         srcFs.close();
 410         srcFs = null;
 411       }
 412     }
 413
 414     @Override
 415     public void failedBulkLoad(final byte[] family, final String srcPath) throws IOException {
 416       try {
 417         Path p = new Path(srcPath);
 418         if (srcFs == null) {
 419           srcFs = FileSystem.newInstance(p.toUri(), conf);
 420         }
 421         if (!FSUtils.isSameHdfs(conf, srcFs, fs)) {
 422           // files are copied so no need to move them back
 423           return;
 424         }
 425         Path stageP = new Path(stagingDir, new Path(Bytes.toString(family), p.getName()));
 426
 427         // In case of Replication for bulk load files, hfiles are not renamed by end point during
 428         // prepare stage, so no need of rename here again
 429         if (p.equals(stageP)) {
 430           LOG.debug(p.getName() + " is already available in source directory. Skipping rename.");
 431           return;
 432         }
 433
 434         LOG.debug("Moving " + stageP + " back to " + p);
 435         if (!fs.rename(stageP, p)) {
 436           throw new IOException("Failed to move HFile: " + stageP + " to " + p);
 437         }
 438
 439         // restore original permission
 440         if (origPermissions.containsKey(srcPath)) {
 441           fs.setPermission(p, origPermissions.get(srcPath));
 442         } else {
 443           LOG.warn("Can't find previous permission for path=" + srcPath);
 444         }
 445       } finally {
 446         closeSrcFs();
 447       }
 448     }
 449
 450     /**
 451      * Check if the path is referencing a file.
 452      * This is mainly needed to avoid symlinks.
 453      * @param p
 454      * @return true if the p is a file
 455      * @throws IOException
 456      */
 457     private boolean isFile(Path p) throws IOException {
 458       FileStatus status = srcFs.getFileStatus(p);
 459       boolean isFile = !status.isDirectory();
 460       try {
 461         isFile = isFile && !(Boolean)Methods.call(FileStatus.class, status, "isSymlink", null, null);
 462       } catch (Exception e) {
 463       }
 464       return isFile;
 465     }
 466   }
 467 }