View Javadoc

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  
19  package org.apache.hadoop.hbase.security.access;
20  
21  import com.google.protobuf.RpcCallback;
22  import com.google.protobuf.RpcController;
23  import com.google.protobuf.Service;
24  
25  import org.apache.commons.logging.Log;
26  import org.apache.commons.logging.LogFactory;
27  import org.apache.hadoop.hbase.classification.InterfaceAudience;
28  import org.apache.hadoop.conf.Configuration;
29  import org.apache.hadoop.fs.FileStatus;
30  import org.apache.hadoop.fs.FileSystem;
31  import org.apache.hadoop.fs.FileUtil;
32  import org.apache.hadoop.fs.Path;
33  import org.apache.hadoop.fs.permission.FsPermission;
34  import org.apache.hadoop.hbase.Coprocessor;
35  import org.apache.hadoop.hbase.CoprocessorEnvironment;
36  import org.apache.hadoop.hbase.TableName;
37  import org.apache.hadoop.hbase.DoNotRetryIOException;
38  import org.apache.hadoop.hbase.coprocessor.BulkLoadObserver;
39  import org.apache.hadoop.hbase.coprocessor.CoprocessorService;
40  import org.apache.hadoop.hbase.coprocessor.ObserverContext;
41  import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
42  import org.apache.hadoop.hbase.ipc.RpcServer;
43  import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
44  import org.apache.hadoop.hbase.protobuf.ResponseConverter;
45  import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
46  import org.apache.hadoop.hbase.protobuf.generated.SecureBulkLoadProtos.SecureBulkLoadService;
47  import org.apache.hadoop.hbase.protobuf.generated.SecureBulkLoadProtos.PrepareBulkLoadRequest;
48  import org.apache.hadoop.hbase.protobuf.generated.SecureBulkLoadProtos.PrepareBulkLoadResponse;
49  import org.apache.hadoop.hbase.protobuf.generated.SecureBulkLoadProtos.CleanupBulkLoadRequest;
50  import org.apache.hadoop.hbase.protobuf.generated.SecureBulkLoadProtos.CleanupBulkLoadResponse;
51  import org.apache.hadoop.hbase.protobuf.generated.SecureBulkLoadProtos.SecureBulkLoadHFilesRequest;
52  import org.apache.hadoop.hbase.protobuf.generated.SecureBulkLoadProtos.SecureBulkLoadHFilesResponse;
53  import org.apache.hadoop.hbase.regionserver.Region;
54  import org.apache.hadoop.hbase.regionserver.Region.BulkLoadListener;
55  import org.apache.hadoop.hbase.security.SecureBulkLoadUtil;
56  import org.apache.hadoop.hbase.security.User;
57  import org.apache.hadoop.hbase.security.UserProvider;
58  import org.apache.hadoop.hbase.security.token.FsDelegationToken;
59  import org.apache.hadoop.hbase.util.Bytes;
60  import org.apache.hadoop.hbase.util.FSHDFSUtils;
61  import org.apache.hadoop.hbase.util.Methods;
62  import org.apache.hadoop.hbase.util.Pair;
63  import org.apache.hadoop.io.Text;
64  import org.apache.hadoop.security.UserGroupInformation;
65  import org.apache.hadoop.security.token.Token;
66  
67  import java.io.IOException;
68  import java.math.BigInteger;
69  import java.security.PrivilegedAction;
70  import java.security.SecureRandom;
71  import java.util.ArrayList;
72  import java.util.HashMap;
73  import java.util.List;
74  import java.util.Map;
75  
76  /**
77   * Coprocessor service for bulk loads in secure mode.
78   * This coprocessor has to be installed as part of enabling
79   * security in HBase.
80   *
81   * This service addresses two issues:
82   * <ol>
83   * <li>Moving files in a secure filesystem wherein the HBase Client
84   * and HBase Server are different filesystem users.</li>
85   * <li>Does moving in a secure manner. Assuming that the filesystem
86   * is POSIX compliant.</li>
87   * </ol>
88   *
89   * The algorithm is as follows:
90   * <ol>
91   * <li>Create an hbase owned staging directory which is
92   * world traversable (711): {@code /hbase/staging}</li>
93   * <li>A user writes out data to his secure output directory: {@code /user/foo/data}</li>
94   * <li>A call is made to hbase to create a secret staging directory
95   * which globally rwx (777): {@code /user/staging/averylongandrandomdirectoryname}</li>
96   * <li>The user moves the data into the random staging directory,
97   * then calls bulkLoadHFiles()</li>
98   * </ol>
99   * Like delegation tokens the strength of the security lies in the length
100  * and randomness of the secret directory.
101  *
102  */
103 @InterfaceAudience.Private
104 public class SecureBulkLoadEndpoint extends SecureBulkLoadService
105     implements CoprocessorService, Coprocessor {
106 
107   public static final long VERSION = 0L;
108 
109   //320/5 = 64 characters
110   private static final int RANDOM_WIDTH = 320;
111   private static final int RANDOM_RADIX = 32;
112 
113   private static final Log LOG = LogFactory.getLog(SecureBulkLoadEndpoint.class);
114 
115   private final static FsPermission PERM_ALL_ACCESS = FsPermission.valueOf("-rwxrwxrwx");
116   private final static FsPermission PERM_HIDDEN = FsPermission.valueOf("-rwx--x--x");
117 
118   private SecureRandom random;
119   private FileSystem fs;
120   private Configuration conf;
121 
122   //two levels so it doesn't get deleted accidentally
123   //no sticky bit in Hadoop 1.0
124   private Path baseStagingDir;
125 
126   private RegionCoprocessorEnvironment env;
127 
128   private UserProvider userProvider;
129 
130   @Override
131   public void start(CoprocessorEnvironment env) {
132     this.env = (RegionCoprocessorEnvironment)env;
133     random = new SecureRandom();
134     conf = env.getConfiguration();
135     baseStagingDir = SecureBulkLoadUtil.getBaseStagingDir(conf);
136     this.userProvider = UserProvider.instantiate(conf);
137 
138     try {
139       fs = FileSystem.get(conf);
140       fs.mkdirs(baseStagingDir, PERM_HIDDEN);
141       fs.setPermission(baseStagingDir, PERM_HIDDEN);
142       //no sticky bit in hadoop-1.0, making directory nonempty so it never gets erased
143       fs.mkdirs(new Path(baseStagingDir,"DONOTERASE"), PERM_HIDDEN);
144       FileStatus status = fs.getFileStatus(baseStagingDir);
145       if(status == null) {
146         throw new IllegalStateException("Failed to create staging directory");
147       }
148       if(!status.getPermission().equals(PERM_HIDDEN)) {
149         throw new IllegalStateException(
150             "Directory already exists but permissions aren't set to '-rwx--x--x' ");
151       }
152     } catch (IOException e) {
153       throw new IllegalStateException("Failed to get FileSystem instance",e);
154     }
155   }
156 
157   @Override
158   public void stop(CoprocessorEnvironment env) throws IOException {
159   }
160 
161   @Override
162   public void prepareBulkLoad(RpcController controller,
163                                                  PrepareBulkLoadRequest request,
164                                                  RpcCallback<PrepareBulkLoadResponse> done){
165     try {
166       List<BulkLoadObserver> bulkLoadObservers = getBulkLoadObservers();
167 
168       if(bulkLoadObservers != null) {
169         ObserverContext<RegionCoprocessorEnvironment> ctx =
170                                            new ObserverContext<RegionCoprocessorEnvironment>();
171         ctx.prepare(env);
172 
173         for(BulkLoadObserver bulkLoadObserver : bulkLoadObservers) {
174           bulkLoadObserver.prePrepareBulkLoad(ctx, request);
175         }
176       }
177 
178       String bulkToken = createStagingDir(baseStagingDir,
179           getActiveUser(), ProtobufUtil.toTableName(request.getTableName())).toString();
180       done.run(PrepareBulkLoadResponse.newBuilder().setBulkToken(bulkToken).build());
181     } catch (IOException e) {
182       ResponseConverter.setControllerException(controller, e);
183     }
184     done.run(null);
185   }
186 
187   @Override
188   public void cleanupBulkLoad(RpcController controller,
189                               CleanupBulkLoadRequest request,
190                               RpcCallback<CleanupBulkLoadResponse> done) {
191     try {
192       List<BulkLoadObserver> bulkLoadObservers = getBulkLoadObservers();
193 
194       if(bulkLoadObservers != null) {
195         ObserverContext<RegionCoprocessorEnvironment> ctx =
196                                            new ObserverContext<RegionCoprocessorEnvironment>();
197         ctx.prepare(env);
198 
199         for(BulkLoadObserver bulkLoadObserver : bulkLoadObservers) {
200           bulkLoadObserver.preCleanupBulkLoad(ctx, request);
201         }
202       }
203 
204       fs.delete(new Path(request.getBulkToken()), true);
205       done.run(CleanupBulkLoadResponse.newBuilder().build());
206     } catch (IOException e) {
207       ResponseConverter.setControllerException(controller, e);
208     }
209     done.run(null);
210   }
211 
212   @Override
213   public void secureBulkLoadHFiles(RpcController controller,
214                                    SecureBulkLoadHFilesRequest request,
215                                    RpcCallback<SecureBulkLoadHFilesResponse> done) {
216     final List<Pair<byte[], String>> familyPaths = new ArrayList<Pair<byte[], String>>();
217     for(ClientProtos.BulkLoadHFileRequest.FamilyPath el : request.getFamilyPathList()) {
218       familyPaths.add(new Pair(el.getFamily().toByteArray(),el.getPath()));
219     }
220 
221     Token userToken = null;
222     if (userProvider.isHadoopSecurityEnabled()) {
223       userToken = new Token(request.getFsToken().getIdentifier().toByteArray(), request.getFsToken()
224               .getPassword().toByteArray(), new Text(request.getFsToken().getKind()), new Text(
225               request.getFsToken().getService()));
226     }
227     final String bulkToken = request.getBulkToken();
228     User user = getActiveUser();
229     final UserGroupInformation ugi = user.getUGI();
230     if(userToken != null) {
231       ugi.addToken(userToken);
232     } else if (userProvider.isHadoopSecurityEnabled()) {
233       //we allow this to pass through in "simple" security mode
234       //for mini cluster testing
235       ResponseConverter.setControllerException(controller,
236           new DoNotRetryIOException("User token cannot be null"));
237       done.run(SecureBulkLoadHFilesResponse.newBuilder().setLoaded(false).build());
238       return;
239     }
240 
241     Region region = env.getRegion();
242     boolean bypass = false;
243     if (region.getCoprocessorHost() != null) {
244       try {
245         bypass = region.getCoprocessorHost().preBulkLoadHFile(familyPaths);
246       } catch (IOException e) {
247         ResponseConverter.setControllerException(controller, e);
248         done.run(SecureBulkLoadHFilesResponse.newBuilder().setLoaded(false).build());
249         return;
250       }
251     }
252     boolean loaded = false;
253     if (!bypass) {
254       // Get the target fs (HBase region server fs) delegation token
255       // Since we have checked the permission via 'preBulkLoadHFile', now let's give
256       // the 'request user' necessary token to operate on the target fs.
257       // After this point the 'doAs' user will hold two tokens, one for the source fs
258       // ('request user'), another for the target fs (HBase region server principal).
259       if (userProvider.isHadoopSecurityEnabled()) {
260         FsDelegationToken targetfsDelegationToken = new FsDelegationToken(userProvider, "renewer");
261         try {
262           targetfsDelegationToken.acquireDelegationToken(fs);
263         } catch (IOException e) {
264           ResponseConverter.setControllerException(controller, e);
265           done.run(SecureBulkLoadHFilesResponse.newBuilder().setLoaded(false).build());
266           return;
267         }
268         Token<?> targetFsToken = targetfsDelegationToken.getUserToken();
269         if (targetFsToken != null
270             && (userToken == null || !targetFsToken.getService().equals(userToken.getService()))) {
271           ugi.addToken(targetFsToken);
272         }
273       }
274 
275       loaded = ugi.doAs(new PrivilegedAction<Boolean>() {
276         @Override
277         public Boolean run() {
278           FileSystem fs = null;
279           try {
280             Configuration conf = env.getConfiguration();
281             fs = FileSystem.get(conf);
282             for(Pair<byte[], String> el: familyPaths) {
283               Path stageFamily = new Path(bulkToken, Bytes.toString(el.getFirst()));
284               if(!fs.exists(stageFamily)) {
285                 fs.mkdirs(stageFamily);
286                 fs.setPermission(stageFamily, PERM_ALL_ACCESS);
287               }
288             }
289             //We call bulkLoadHFiles as requesting user
290             //To enable access prior to staging
291             return env.getRegion().bulkLoadHFiles(familyPaths, true,
292                 new SecureBulkLoadListener(fs, bulkToken, conf));
293           } catch (Exception e) {
294             LOG.error("Failed to complete bulk load", e);
295           }
296           return false;
297         }
298       });
299     }
300     if (region.getCoprocessorHost() != null) {
301       try {
302         loaded = region.getCoprocessorHost().postBulkLoadHFile(familyPaths, loaded);
303       } catch (IOException e) {
304         ResponseConverter.setControllerException(controller, e);
305         done.run(SecureBulkLoadHFilesResponse.newBuilder().setLoaded(false).build());
306         return;
307       }
308     }
309     done.run(SecureBulkLoadHFilesResponse.newBuilder().setLoaded(loaded).build());
310   }
311 
312   private List<BulkLoadObserver> getBulkLoadObservers() {
313     List<BulkLoadObserver> coprocessorList =
314               this.env.getRegion().getCoprocessorHost().findCoprocessors(BulkLoadObserver.class);
315 
316     return coprocessorList;
317   }
318 
319   private Path createStagingDir(Path baseDir,
320                                 User user,
321                                 TableName tableName) throws IOException {
322     String tblName = tableName.getNameAsString().replace(":", "_");
323     String randomDir = user.getShortName()+"__"+ tblName +"__"+
324         (new BigInteger(RANDOM_WIDTH, random).toString(RANDOM_RADIX));
325     return createStagingDir(baseDir, user, randomDir);
326   }
327 
328   private Path createStagingDir(Path baseDir,
329                                 User user,
330                                 String randomDir) throws IOException {
331     Path p = new Path(baseDir, randomDir);
332     fs.mkdirs(p, PERM_ALL_ACCESS);
333     fs.setPermission(p, PERM_ALL_ACCESS);
334     return p;
335   }
336 
337   private User getActiveUser() {
338     User user = RpcServer.getRequestUser();
339     if (user == null) {
340       return null;
341     }
342 
343     //this is for testing
344     if (userProvider.isHadoopSecurityEnabled()
345         && "simple".equalsIgnoreCase(conf.get(User.HBASE_SECURITY_CONF_KEY))) {
346       return User.createUserForTesting(conf, user.getShortName(), new String[]{});
347     }
348 
349     return user;
350   }
351 
352   @Override
353   public Service getService() {
354     return this;
355   }
356 
357   private static class SecureBulkLoadListener implements BulkLoadListener {
358     // Target filesystem
359     private FileSystem fs;
360     private String stagingDir;
361     private Configuration conf;
362     // Source filesystem
363     private FileSystem srcFs = null;
364     private Map<String, FsPermission> origPermissions = null;
365 
366     public SecureBulkLoadListener(FileSystem fs, String stagingDir, Configuration conf) {
367       this.fs = fs;
368       this.stagingDir = stagingDir;
369       this.conf = conf;
370       this.origPermissions = new HashMap<String, FsPermission>();
371     }
372 
373     @Override
374     public String prepareBulkLoad(final byte[] family, final String srcPath) throws IOException {
375       Path p = new Path(srcPath);
376       Path stageP = new Path(stagingDir, new Path(Bytes.toString(family), p.getName()));
377 
378       // In case of Replication for bulk load files, hfiles are already copied in staging directory
379       if (p.equals(stageP)) {
380         LOG.debug(p.getName()
381             + " is already available in staging directory. Skipping copy or rename.");
382         return stageP.toString();
383       }
384 
385       if (srcFs == null) {
386         srcFs = FileSystem.get(p.toUri(), conf);
387       }
388 
389       if(!isFile(p)) {
390         throw new IOException("Path does not reference a file: " + p);
391       }
392 
393       // Check to see if the source and target filesystems are the same
394       if (!FSHDFSUtils.isSameHdfs(conf, srcFs, fs)) {
395         LOG.debug("Bulk-load file " + srcPath + " is on different filesystem than " +
396             "the destination filesystem. Copying file over to destination staging dir.");
397         FileUtil.copy(srcFs, p, fs, stageP, false, conf);
398       } else {
399         LOG.debug("Moving " + p + " to " + stageP);
400         FileStatus origFileStatus = fs.getFileStatus(p);
401         origPermissions.put(srcPath, origFileStatus.getPermission());
402         if(!fs.rename(p, stageP)) {
403           throw new IOException("Failed to move HFile: " + p + " to " + stageP);
404         }
405       }
406       fs.setPermission(stageP, PERM_ALL_ACCESS);
407       return stageP.toString();
408     }
409 
410     @Override
411     public void doneBulkLoad(byte[] family, String srcPath) throws IOException {
412       LOG.debug("Bulk Load done for: " + srcPath);
413     }
414 
415     @Override
416     public void failedBulkLoad(final byte[] family, final String srcPath) throws IOException {
417       if (!FSHDFSUtils.isSameHdfs(conf, srcFs, fs)) {
418         // files are copied so no need to move them back
419         return;
420       }
421       Path p = new Path(srcPath);
422       Path stageP = new Path(stagingDir,
423           new Path(Bytes.toString(family), p.getName()));
424 
425       // In case of Replication for bulk load files, hfiles are not renamed by end point during
426       // prepare stage, so no need of rename here again
427       if (p.equals(stageP)) {
428         LOG.debug(p.getName() + " is already available in source directory. Skipping rename.");
429         return;
430       }
431 
432       LOG.debug("Moving " + stageP + " back to " + p);
433       if(!fs.rename(stageP, p))
434         throw new IOException("Failed to move HFile: " + stageP + " to " + p);
435 
436       // restore original permission
437       if (origPermissions.containsKey(srcPath)) {
438         fs.setPermission(p, origPermissions.get(srcPath));
439       } else {
440         LOG.warn("Can't find previous permission for path=" + srcPath);
441       }
442     }
443 
444     /**
445      * Check if the path is referencing a file.
446      * This is mainly needed to avoid symlinks.
447      * @param p
448      * @return true if the p is a file
449      * @throws IOException
450      */
451     private boolean isFile(Path p) throws IOException {
452       FileStatus status = srcFs.getFileStatus(p);
453       boolean isFile = !status.isDirectory();
454       try {
455         isFile = isFile && !(Boolean)Methods.call(FileStatus.class, status, "isSymlink", null, null);
456       } catch (Exception e) {
457       }
458       return isFile;
459     }
460   }
461 }