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.CommonFSUtils;
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(CommonFSUtils.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    region.getCoprocessorHost().preCleanupBulkLoad(getActiveUser());
157
158    Path path = new Path(request.getBulkToken());
159    if (!fs.delete(path, true)) {
160      if (fs.exists(path)) {
161        throw new IOException("Failed to clean up " + path);
162      }
163    }
164    LOG.trace("Cleaned up {} successfully.", path);
165  }
166
167  private Consumer<HRegion> fsCreatedListener;
168
169  @VisibleForTesting
170  void setFsCreatedListener(Consumer<HRegion> fsCreatedListener) {
171    this.fsCreatedListener = fsCreatedListener;
172  }
173
174
175  private void incrementUgiReference(UserGroupInformation ugi) {
176    // if we haven't seen this ugi before, make a new counter
177    ugiReferenceCounter.compute(ugi, (key, value) -> {
178      if (value == null) {
179        value = new MutableInt(1);
180      } else {
181        value.increment();
182      }
183      return value;
184    });
185  }
186
187  private void decrementUgiReference(UserGroupInformation ugi) {
188    // if the count drops below 1 we remove the entry by returning null
189    ugiReferenceCounter.computeIfPresent(ugi, (key, value) -> {
190      if (value.intValue() > 1) {
191        value.decrement();
192      } else {
193        value = null;
194      }
195      return value;
196    });
197  }
198
199  private boolean isUserReferenced(UserGroupInformation ugi) {
200    // if the ugi is in the map, based on invariants above
201    // the count must be above zero
202    return ugiReferenceCounter.containsKey(ugi);
203  }
204
205  public Map<byte[], List<Path>> secureBulkLoadHFiles(final HRegion region,
206    final BulkLoadHFileRequest request) throws IOException {
207    return secureBulkLoadHFiles(region, request, null);
208  }
209
210  public Map<byte[], List<Path>> secureBulkLoadHFiles(final HRegion region,
211      final BulkLoadHFileRequest request, List<String> clusterIds) throws IOException {
212    final List<Pair<byte[], String>> familyPaths = new ArrayList<>(request.getFamilyPathCount());
213    for(ClientProtos.BulkLoadHFileRequest.FamilyPath el : request.getFamilyPathList()) {
214      familyPaths.add(new Pair<>(el.getFamily().toByteArray(), el.getPath()));
215    }
216
217    Token<AuthenticationTokenIdentifier> userToken = null;
218    if (userProvider.isHadoopSecurityEnabled()) {
219      userToken = new Token<>(request.getFsToken().getIdentifier().toByteArray(),
220        request.getFsToken().getPassword().toByteArray(), new Text(request.getFsToken().getKind()),
221        new Text(request.getFsToken().getService()));
222    }
223    final String bulkToken = request.getBulkToken();
224    User user = getActiveUser();
225    final UserGroupInformation ugi = user.getUGI();
226    if (userProvider.isHadoopSecurityEnabled()) {
227      try {
228        Token<AuthenticationTokenIdentifier> tok = ClientTokenUtil.obtainToken(conn).get();
229        if (tok != null) {
230          boolean b = ugi.addToken(tok);
231          LOG.debug("token added " + tok + " for user " + ugi + " return=" + b);
232        }
233      } catch (Exception ioe) {
234        LOG.warn("unable to add token", ioe);
235      }
236    }
237    if (userToken != null) {
238      ugi.addToken(userToken);
239    } else if (userProvider.isHadoopSecurityEnabled()) {
240      //we allow this to pass through in "simple" security mode
241      //for mini cluster testing
242      throw new DoNotRetryIOException("User token cannot be null");
243    }
244
245    if (region.getCoprocessorHost() != null) {
246      region.getCoprocessorHost().preBulkLoadHFile(familyPaths);
247    }
248    Map<byte[], List<Path>> map = null;
249
250    try {
251      incrementUgiReference(ugi);
252      // Get the target fs (HBase region server fs) delegation token
253      // Since we have checked the permission via 'preBulkLoadHFile', now let's give
254      // the 'request user' necessary token to operate on the target fs.
255      // After this point the 'doAs' user will hold two tokens, one for the source fs
256      // ('request user'), another for the target fs (HBase region server principal).
257      if (userProvider.isHadoopSecurityEnabled()) {
258        FsDelegationToken targetfsDelegationToken = new FsDelegationToken(userProvider,"renewer");
259        targetfsDelegationToken.acquireDelegationToken(fs);
260
261        Token<?> targetFsToken = targetfsDelegationToken.getUserToken();
262        if (targetFsToken != null
263            && (userToken == null || !targetFsToken.getService().equals(userToken.getService()))){
264          ugi.addToken(targetFsToken);
265        }
266      }
267
268      map = ugi.doAs(new PrivilegedAction<Map<byte[], List<Path>>>() {
269        @Override
270        public Map<byte[], List<Path>> run() {
271          FileSystem fs = null;
272          try {
273            /*
274             * This is creating and caching a new FileSystem instance. Other code called
275             * "beneath" this method will rely on this FileSystem instance being in the
276             * cache. This is important as those methods make _no_ attempt to close this
277             * FileSystem instance. It is critical that here, in SecureBulkLoadManager,
278             * we are tracking the lifecycle and closing the FS when safe to do so.
279             */
280            fs = FileSystem.get(conf);
281            for(Pair<byte[], String> el: familyPaths) {
282              Path stageFamily = new Path(bulkToken, Bytes.toString(el.getFirst()));
283              if(!fs.exists(stageFamily)) {
284                fs.mkdirs(stageFamily);
285                fs.setPermission(stageFamily, PERM_ALL_ACCESS);
286              }
287            }
288            if (fsCreatedListener != null) {
289              fsCreatedListener.accept(region);
290            }
291            //We call bulkLoadHFiles as requesting user
292            //To enable access prior to staging
293            return region.bulkLoadHFiles(familyPaths, true,
294                new SecureBulkLoadListener(fs, bulkToken, conf), request.getCopyFile(),
295              clusterIds, request.getReplicate());
296          } catch (Exception e) {
297            LOG.error("Failed to complete bulk load", e);
298          }
299          return null;
300        }
301      });
302    } finally {
303      decrementUgiReference(ugi);
304      try {
305        if (!UserGroupInformation.getLoginUser().equals(ugi) && !isUserReferenced(ugi)) {
306          FileSystem.closeAllForUGI(ugi);
307        }
308      } catch (IOException e) {
309        LOG.error("Failed to close FileSystem for: {}", ugi, e);
310      }
311      if (region.getCoprocessorHost() != null) {
312        region.getCoprocessorHost().postBulkLoadHFile(familyPaths, map);
313      }
314    }
315    return map;
316  }
317
318  private Path createStagingDir(Path baseDir,
319                                User user,
320                                TableName tableName) throws IOException {
321    String tblName = tableName.getNameAsString().replace(":", "_");
322    String randomDir = user.getShortName()+"__"+ tblName +"__"+
323        (new BigInteger(RANDOM_WIDTH, random).toString(RANDOM_RADIX));
324    return createStagingDir(baseDir, user, randomDir);
325  }
326
327  private Path createStagingDir(Path baseDir,
328                                User user,
329                                String randomDir) throws IOException {
330    Path p = new Path(baseDir, randomDir);
331    fs.mkdirs(p, PERM_ALL_ACCESS);
332    fs.setPermission(p, PERM_ALL_ACCESS);
333    return p;
334  }
335
336  private User getActiveUser() throws IOException {
337    // for non-rpc handling, fallback to system user
338    User user = RpcServer.getRequestUser().orElse(userProvider.getCurrent());
339    // this is for testing
340    if (userProvider.isHadoopSecurityEnabled() &&
341        "simple".equalsIgnoreCase(conf.get(User.HBASE_SECURITY_CONF_KEY))) {
342      return User.createUserForTesting(conf, user.getShortName(), new String[] {});
343    }
344
345    return user;
346  }
347
348  private static class SecureBulkLoadListener implements BulkLoadListener {
349    // Target filesystem
350    private final FileSystem fs;
351    private final String stagingDir;
352    private final Configuration conf;
353    // Source filesystem
354    private FileSystem srcFs = null;
355    private Map<String, FsPermission> origPermissions = null;
356
357    public SecureBulkLoadListener(FileSystem fs, String stagingDir, Configuration conf) {
358      this.fs = fs;
359      this.stagingDir = stagingDir;
360      this.conf = conf;
361      this.origPermissions = new HashMap<>();
362    }
363
364    @Override
365    public String prepareBulkLoad(final byte[] family, final String srcPath, boolean copyFile)
366        throws IOException {
367      Path p = new Path(srcPath);
368      Path stageP = new Path(stagingDir, new Path(Bytes.toString(family), p.getName()));
369
370      // In case of Replication for bulk load files, hfiles are already copied in staging directory
371      if (p.equals(stageP)) {
372        LOG.debug(p.getName()
373            + " is already available in staging directory. Skipping copy or rename.");
374        return stageP.toString();
375      }
376
377      if (srcFs == null) {
378        srcFs = FileSystem.newInstance(p.toUri(), conf);
379      }
380
381      if(!isFile(p)) {
382        throw new IOException("Path does not reference a file: " + p);
383      }
384
385      // Check to see if the source and target filesystems are the same
386      if (!FSUtils.isSameHdfs(conf, srcFs, fs)) {
387        LOG.debug("Bulk-load file " + srcPath + " is on different filesystem than " +
388            "the destination filesystem. Copying file over to destination staging dir.");
389        FileUtil.copy(srcFs, p, fs, stageP, false, conf);
390      } else if (copyFile) {
391        LOG.debug("Bulk-load file " + srcPath + " is copied to destination staging dir.");
392        FileUtil.copy(srcFs, p, fs, stageP, false, conf);
393      } else {
394        LOG.debug("Moving " + p + " to " + stageP);
395        FileStatus origFileStatus = fs.getFileStatus(p);
396        origPermissions.put(srcPath, origFileStatus.getPermission());
397        if(!fs.rename(p, stageP)) {
398          throw new IOException("Failed to move HFile: " + p + " to " + stageP);
399        }
400      }
401      fs.setPermission(stageP, PERM_ALL_ACCESS);
402      return stageP.toString();
403    }
404
405    @Override
406    public void doneBulkLoad(byte[] family, String srcPath) throws IOException {
407      LOG.debug("Bulk Load done for: " + srcPath);
408      closeSrcFs();
409    }
410
411    private void closeSrcFs() throws IOException {
412      if (srcFs != null) {
413        srcFs.close();
414        srcFs = null;
415      }
416    }
417
418    @Override
419    public void failedBulkLoad(final byte[] family, final String srcPath) throws IOException {
420      try {
421        Path p = new Path(srcPath);
422        if (srcFs == null) {
423          srcFs = FileSystem.newInstance(p.toUri(), conf);
424        }
425        if (!FSUtils.isSameHdfs(conf, srcFs, fs)) {
426          // files are copied so no need to move them back
427          return;
428        }
429        Path stageP = new Path(stagingDir, new Path(Bytes.toString(family), p.getName()));
430
431        // In case of Replication for bulk load files, hfiles are not renamed by end point during
432        // prepare stage, so no need of rename here again
433        if (p.equals(stageP)) {
434          LOG.debug(p.getName() + " is already available in source directory. Skipping rename.");
435          return;
436        }
437
438        LOG.debug("Moving " + stageP + " back to " + p);
439        if (!fs.rename(stageP, p)) {
440          throw new IOException("Failed to move HFile: " + stageP + " to " + p);
441        }
442
443        // restore original permission
444        if (origPermissions.containsKey(srcPath)) {
445          fs.setPermission(p, origPermissions.get(srcPath));
446        } else {
447          LOG.warn("Can't find previous permission for path=" + srcPath);
448        }
449      } finally {
450        closeSrcFs();
451      }
452    }
453
454    /**
455     * Check if the path is referencing a file.
456     * This is mainly needed to avoid symlinks.
457     * @param p
458     * @return true if the p is a file
459     * @throws IOException
460     */
461    private boolean isFile(Path p) throws IOException {
462      FileStatus status = srcFs.getFileStatus(p);
463      boolean isFile = !status.isDirectory();
464      try {
465        isFile = isFile && !(Boolean)Methods.call(FileStatus.class, status, "isSymlink", null, null);
466      } catch (Exception e) {
467      }
468      return isFile;
469    }
470  }
471}