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