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;
030import org.apache.commons.lang3.mutable.MutableInt;
031import org.apache.hadoop.conf.Configuration;
032import org.apache.hadoop.fs.FileStatus;
033import org.apache.hadoop.fs.FileSystem;
034import org.apache.hadoop.fs.FileUtil;
035import org.apache.hadoop.fs.Path;
036import org.apache.hadoop.fs.permission.FsPermission;
037import org.apache.hadoop.hbase.DoNotRetryIOException;
038import org.apache.hadoop.hbase.HConstants;
039import org.apache.hadoop.hbase.TableName;
040import org.apache.hadoop.hbase.client.AsyncConnection;
041import org.apache.hadoop.hbase.ipc.RpcServer;
042import org.apache.hadoop.hbase.regionserver.HRegion.BulkLoadListener;
043import org.apache.hadoop.hbase.security.User;
044import org.apache.hadoop.hbase.security.UserProvider;
045import org.apache.hadoop.hbase.security.token.AuthenticationTokenIdentifier;
046import org.apache.hadoop.hbase.security.token.ClientTokenUtil;
047import org.apache.hadoop.hbase.security.token.FsDelegationToken;
048import org.apache.hadoop.hbase.util.Bytes;
049import org.apache.hadoop.hbase.util.FSHDFSUtils;
050import org.apache.hadoop.hbase.util.FSUtils;
051import org.apache.hadoop.hbase.util.Methods;
052import org.apache.hadoop.hbase.util.Pair;
053import org.apache.hadoop.io.Text;
054import org.apache.hadoop.security.UserGroupInformation;
055import org.apache.hadoop.security.token.Token;
056import org.apache.yetus.audience.InterfaceAudience;
057import org.slf4j.Logger;
058import org.slf4j.LoggerFactory;
059
060import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
061
062import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos;
063import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.BulkLoadHFileRequest;
064import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.CleanupBulkLoadRequest;
065import org.apache.hadoop.hbase.shaded.protobuf.generated.ClientProtos.PrepareBulkLoadRequest;
066
067/**
068 * Bulk loads in secure mode.
069 *
070 * This service addresses two issues:
071 * <ol>
072 * <li>Moving files in a secure filesystem wherein the HBase Client
073 * and HBase Server are different filesystem users.</li>
074 * <li>Does moving in a secure manner. Assuming that the filesystem
075 * is POSIX compliant.</li>
076 * </ol>
077 *
078 * The algorithm is as follows:
079 * <ol>
080 * <li>Create an hbase owned staging directory which is
081 * world traversable (711): {@code /hbase/staging}</li>
082 * <li>A user writes out data to his secure output directory: {@code /user/foo/data}</li>
083 * <li>A call is made to hbase to create a secret staging directory
084 * which globally rwx (777): {@code /user/staging/averylongandrandomdirectoryname}</li>
085 * <li>The user moves the data into the random staging directory,
086 * then calls bulkLoadHFiles()</li>
087 * </ol>
088 *
089 * Like delegation tokens the strength of the security lies in the length
090 * and randomness of the secret directory.
091 *
092 */
093@InterfaceAudience.Private
094public class SecureBulkLoadManager {
095
096  public static final long VERSION = 0L;
097
098  //320/5 = 64 characters
099  private static final int RANDOM_WIDTH = 320;
100  private static final int RANDOM_RADIX = 32;
101
102  private static final Logger LOG = LoggerFactory.getLogger(SecureBulkLoadManager.class);
103
104  private final static FsPermission PERM_ALL_ACCESS = FsPermission.valueOf("-rwxrwxrwx");
105  private final static FsPermission PERM_HIDDEN = FsPermission.valueOf("-rwx--x--x");
106  private SecureRandom random;
107  private FileSystem fs;
108  private Configuration conf;
109
110  //two levels so it doesn't get deleted accidentally
111  //no sticky bit in Hadoop 1.0
112  private Path baseStagingDir;
113
114  private UserProvider userProvider;
115  private ConcurrentHashMap<UserGroupInformation, MutableInt> ugiReferenceCounter;
116  private AsyncConnection conn;
117
118  SecureBulkLoadManager(Configuration conf, AsyncConnection conn) {
119    this.conf = conf;
120    this.conn = conn;
121  }
122
123  public void start() throws IOException {
124    random = new SecureRandom();
125    userProvider = UserProvider.instantiate(conf);
126    ugiReferenceCounter = new ConcurrentHashMap<>();
127    fs = FileSystem.get(conf);
128    baseStagingDir = new Path(FSUtils.getRootDir(conf), HConstants.BULKLOAD_STAGING_DIR_NAME);
129
130    if (conf.get("hbase.bulkload.staging.dir") != null) {
131      LOG.warn("hbase.bulkload.staging.dir " + " is deprecated. Bulkload staging directory is "
132          + baseStagingDir);
133    }
134    if (!fs.exists(baseStagingDir)) {
135      fs.mkdirs(baseStagingDir, PERM_HIDDEN);
136    }
137  }
138
139  public void stop() throws IOException {
140  }
141
142  public String prepareBulkLoad(final HRegion region, final PrepareBulkLoadRequest request)
143      throws IOException {
144    User user = getActiveUser();
145    region.getCoprocessorHost().prePrepareBulkLoad(user);
146
147    String bulkToken =
148        createStagingDir(baseStagingDir, user, region.getTableDescriptor().getTableName())
149            .toString();
150
151    return bulkToken;
152  }
153
154  public void cleanupBulkLoad(final HRegion region, final CleanupBulkLoadRequest request)
155      throws IOException {
156    try {
157      region.getCoprocessorHost().preCleanupBulkLoad(getActiveUser());
158
159      Path path = new Path(request.getBulkToken());
160      if (!fs.delete(path, true)) {
161        if (fs.exists(path)) {
162          throw new IOException("Failed to clean up " + path);
163        }
164      }
165      LOG.info("Cleaned up " + path + " successfully.");
166    } finally {
167      UserGroupInformation ugi = getActiveUser().getUGI();
168      try {
169        if (!UserGroupInformation.getLoginUser().equals(ugi) && !isUserReferenced(ugi)) {
170          FileSystem.closeAllForUGI(ugi);
171        }
172      } catch (IOException e) {
173        LOG.error("Failed to close FileSystem for: " + ugi, e);
174      }
175    }
176  }
177
178  private Consumer<HRegion> fsCreatedListener;
179
180  @VisibleForTesting
181  void setFsCreatedListener(Consumer<HRegion> fsCreatedListener) {
182    this.fsCreatedListener = fsCreatedListener;
183  }
184
185
186  private void incrementUgiReference(UserGroupInformation ugi) {
187    // if we haven't seen this ugi before, make a new counter
188    ugiReferenceCounter.compute(ugi, (key, value) -> {
189      if (value == null) {
190        value = new MutableInt(1);
191      } else {
192        value.increment();
193      }
194      return value;
195    });
196  }
197
198  private void decrementUgiReference(UserGroupInformation ugi) {
199    // if the count drops below 1 we remove the entry by returning null
200    ugiReferenceCounter.computeIfPresent(ugi, (key, value) -> {
201      if (value.intValue() > 1) {
202        value.decrement();
203      } else {
204        value = null;
205      }
206      return value;
207    });
208  }
209
210  private boolean isUserReferenced(UserGroupInformation ugi) {
211    // if the ugi is in the map, based on invariants above
212    // the count must be above zero
213    return ugiReferenceCounter.containsKey(ugi);
214  }
215
216  public Map<byte[], List<Path>> secureBulkLoadHFiles(final HRegion region,
217    final BulkLoadHFileRequest request) throws IOException {
218    return secureBulkLoadHFiles(region, request, null);
219  }
220
221  public Map<byte[], List<Path>> secureBulkLoadHFiles(final HRegion region,
222      final BulkLoadHFileRequest request, List<String> clusterIds) throws IOException {
223    final List<Pair<byte[], String>> familyPaths = new ArrayList<>(request.getFamilyPathCount());
224    for(ClientProtos.BulkLoadHFileRequest.FamilyPath el : request.getFamilyPathList()) {
225      familyPaths.add(new Pair<>(el.getFamily().toByteArray(), el.getPath()));
226    }
227
228    Token<AuthenticationTokenIdentifier> userToken = null;
229    if (userProvider.isHadoopSecurityEnabled()) {
230      userToken = new Token<>(request.getFsToken().getIdentifier().toByteArray(),
231        request.getFsToken().getPassword().toByteArray(), new Text(request.getFsToken().getKind()),
232        new Text(request.getFsToken().getService()));
233    }
234    final String bulkToken = request.getBulkToken();
235    User user = getActiveUser();
236    final UserGroupInformation ugi = user.getUGI();
237    if (userProvider.isHadoopSecurityEnabled()) {
238      try {
239        Token<AuthenticationTokenIdentifier> tok = ClientTokenUtil.obtainToken(conn).get();
240        if (tok != null) {
241          boolean b = ugi.addToken(tok);
242          LOG.debug("token added " + tok + " for user " + ugi + " return=" + b);
243        }
244      } catch (Exception ioe) {
245        LOG.warn("unable to add token", ioe);
246      }
247    }
248    if (userToken != null) {
249      ugi.addToken(userToken);
250    } else if (userProvider.isHadoopSecurityEnabled()) {
251      //we allow this to pass through in "simple" security mode
252      //for mini cluster testing
253      throw new DoNotRetryIOException("User token cannot be null");
254    }
255
256    if (region.getCoprocessorHost() != null) {
257      region.getCoprocessorHost().preBulkLoadHFile(familyPaths);
258    }
259    Map<byte[], List<Path>> map = null;
260
261    try {
262      incrementUgiReference(ugi);
263      // Get the target fs (HBase region server fs) delegation token
264      // Since we have checked the permission via 'preBulkLoadHFile', now let's give
265      // the 'request user' necessary token to operate on the target fs.
266      // After this point the 'doAs' user will hold two tokens, one for the source fs
267      // ('request user'), another for the target fs (HBase region server principal).
268      if (userProvider.isHadoopSecurityEnabled()) {
269        FsDelegationToken targetfsDelegationToken = new FsDelegationToken(userProvider,"renewer");
270        targetfsDelegationToken.acquireDelegationToken(fs);
271
272        Token<?> targetFsToken = targetfsDelegationToken.getUserToken();
273        if (targetFsToken != null
274            && (userToken == null || !targetFsToken.getService().equals(userToken.getService()))){
275          ugi.addToken(targetFsToken);
276        }
277      }
278
279      map = ugi.doAs(new PrivilegedAction<Map<byte[], List<Path>>>() {
280        @Override
281        public Map<byte[], List<Path>> run() {
282          FileSystem fs = null;
283          try {
284            fs = FileSystem.get(conf);
285            for(Pair<byte[], String> el: familyPaths) {
286              Path stageFamily = new Path(bulkToken, Bytes.toString(el.getFirst()));
287              if(!fs.exists(stageFamily)) {
288                fs.mkdirs(stageFamily);
289                fs.setPermission(stageFamily, PERM_ALL_ACCESS);
290              }
291            }
292            if (fsCreatedListener != null) {
293              fsCreatedListener.accept(region);
294            }
295            //We call bulkLoadHFiles as requesting user
296            //To enable access prior to staging
297            return region.bulkLoadHFiles(familyPaths, true,
298                new SecureBulkLoadListener(fs, bulkToken, conf), request.getCopyFile(),
299              clusterIds, request.getReplicate());
300          } catch (Exception e) {
301            LOG.error("Failed to complete bulk load", e);
302          }
303          return null;
304        }
305      });
306    } finally {
307      decrementUgiReference(ugi);
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 (!FSHDFSUtils.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 (!FSHDFSUtils.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}