1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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.lang.StringUtils;
26 import org.apache.commons.logging.Log;
27 import org.apache.commons.logging.LogFactory;
28 import org.apache.hadoop.hbase.classification.InterfaceAudience;
29 import org.apache.hadoop.hbase.client.Connection;
30 import org.apache.hadoop.conf.Configuration;
31 import org.apache.hadoop.fs.FileStatus;
32 import org.apache.hadoop.fs.FileSystem;
33 import org.apache.hadoop.fs.FileUtil;
34 import org.apache.hadoop.fs.Path;
35 import org.apache.hadoop.fs.permission.FsPermission;
36 import org.apache.hadoop.hbase.Coprocessor;
37 import org.apache.hadoop.hbase.CoprocessorEnvironment;
38 import org.apache.hadoop.hbase.TableName;
39 import org.apache.hadoop.hbase.DoNotRetryIOException;
40 import org.apache.hadoop.hbase.coprocessor.BulkLoadObserver;
41 import org.apache.hadoop.hbase.coprocessor.CoprocessorService;
42 import org.apache.hadoop.hbase.coprocessor.ObserverContext;
43 import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
44 import org.apache.hadoop.hbase.ipc.RpcServer;
45 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
46 import org.apache.hadoop.hbase.protobuf.ResponseConverter;
47 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
48 import org.apache.hadoop.hbase.protobuf.generated.SecureBulkLoadProtos.SecureBulkLoadService;
49 import org.apache.hadoop.hbase.protobuf.generated.SecureBulkLoadProtos.PrepareBulkLoadRequest;
50 import org.apache.hadoop.hbase.protobuf.generated.SecureBulkLoadProtos.PrepareBulkLoadResponse;
51 import org.apache.hadoop.hbase.protobuf.generated.SecureBulkLoadProtos.CleanupBulkLoadRequest;
52 import org.apache.hadoop.hbase.protobuf.generated.SecureBulkLoadProtos.CleanupBulkLoadResponse;
53 import org.apache.hadoop.hbase.protobuf.generated.SecureBulkLoadProtos.SecureBulkLoadHFilesRequest;
54 import org.apache.hadoop.hbase.protobuf.generated.SecureBulkLoadProtos.SecureBulkLoadHFilesResponse;
55 import org.apache.hadoop.hbase.regionserver.Region;
56 import org.apache.hadoop.hbase.regionserver.Region.BulkLoadListener;
57 import org.apache.hadoop.hbase.regionserver.RegionServerServices;
58 import org.apache.hadoop.hbase.security.SecureBulkLoadUtil;
59 import org.apache.hadoop.hbase.security.User;
60 import org.apache.hadoop.hbase.security.UserProvider;
61 import org.apache.hadoop.hbase.security.token.FsDelegationToken;
62 import org.apache.hadoop.hbase.security.token.TokenUtil;
63 import org.apache.hadoop.hbase.util.Bytes;
64 import org.apache.hadoop.hbase.util.FSHDFSUtils;
65 import org.apache.hadoop.hbase.util.Methods;
66 import org.apache.hadoop.hbase.util.Pair;
67 import org.apache.hadoop.io.Text;
68 import org.apache.hadoop.security.UserGroupInformation;
69 import org.apache.hadoop.security.token.Token;
70
71 import java.io.IOException;
72 import java.math.BigInteger;
73 import java.security.PrivilegedAction;
74 import java.security.SecureRandom;
75 import java.util.ArrayList;
76 import java.util.Arrays;
77 import java.util.HashMap;
78 import java.util.HashSet;
79 import java.util.List;
80 import java.util.Map;
81 import java.util.Set;
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109 @InterfaceAudience.Private
110 public class SecureBulkLoadEndpoint extends SecureBulkLoadService
111 implements CoprocessorService, Coprocessor {
112
113 public static final long VERSION = 0L;
114
115
116 private static final int RANDOM_WIDTH = 320;
117 private static final int RANDOM_RADIX = 32;
118
119 private static final Log LOG = LogFactory.getLog(SecureBulkLoadEndpoint.class);
120
121 private final static FsPermission PERM_ALL_ACCESS = FsPermission.valueOf("-rwxrwxrwx");
122 private final static FsPermission PERM_HIDDEN = FsPermission.valueOf("-rwx--x--x");
123
124 public static final String FS_WITHOUT_SUPPORT_PERMISSION_KEY =
125 "hbase.secure.bulkload.fs.permission.lacking";
126 public static final String FS_WITHOUT_SUPPORT_PERMISSION_DEFAULT =
127 "s3,s3a,s3n,wasb,wasbs,swift,adfs,abfs,viewfs";
128
129 private SecureRandom random;
130 private FileSystem fs;
131 private Configuration conf;
132
133
134
135 private Path baseStagingDir;
136
137 private RegionCoprocessorEnvironment env;
138 private Connection conn;
139
140 private UserProvider userProvider;
141
142 @Override
143 public void start(CoprocessorEnvironment env) {
144 this.env = (RegionCoprocessorEnvironment)env;
145 random = new SecureRandom();
146 RegionServerServices svc = ((RegionCoprocessorEnvironment)env).getRegionServerServices();
147 this.conn = svc.getConnection();
148 conf = env.getConfiguration();
149 baseStagingDir = SecureBulkLoadUtil.getBaseStagingDir(conf);
150 this.userProvider = UserProvider.instantiate(conf);
151 Set<String> fsSet = getFileSystemSchemesWithoutPermissionSupport(conf);
152
153 try {
154 fs = baseStagingDir.getFileSystem(conf);
155 if (!fs.exists(baseStagingDir)) {
156 fs.mkdirs(baseStagingDir, PERM_HIDDEN);
157 }
158 FileStatus status = fs.getFileStatus(baseStagingDir);
159 if (status == null) {
160 throw new IllegalStateException("Failed to create staging directory");
161 }
162
163
164
165 if (!status.getPermission().equals(PERM_HIDDEN)) {
166 fs.setPermission(baseStagingDir, PERM_HIDDEN);
167 status = fs.getFileStatus(baseStagingDir);
168 }
169
170
171 Path doNotEraseDir = new Path(baseStagingDir, "DONOTERASE");
172 if (!fs.exists(doNotEraseDir)) {
173 fs.mkdirs(doNotEraseDir, PERM_HIDDEN);
174 fs.setPermission(doNotEraseDir, PERM_HIDDEN);
175 }
176
177 String scheme = fs.getScheme().toLowerCase();
178 if (!fsSet.contains(scheme) && !status.getPermission().equals(PERM_HIDDEN)) {
179 throw new IllegalStateException(
180 "Staging directory of " + baseStagingDir + " already exists but permissions aren't set to '-rwx--x--x' ");
181 }
182 } catch (IOException e) {
183 throw new IllegalStateException("Failed to get FileSystem instance",e);
184 }
185 }
186
187 Set<String> getFileSystemSchemesWithoutPermissionSupport(Configuration conf) {
188 final String value = conf.get(
189 FS_WITHOUT_SUPPORT_PERMISSION_KEY, FS_WITHOUT_SUPPORT_PERMISSION_DEFAULT);
190 return new HashSet<String>(Arrays.asList(StringUtils.split(value, ',')));
191 }
192
193 @Override
194 public void stop(CoprocessorEnvironment env) throws IOException {
195 }
196
197 @Override
198 public void prepareBulkLoad(RpcController controller,
199 PrepareBulkLoadRequest request,
200 RpcCallback<PrepareBulkLoadResponse> done){
201 try {
202 List<BulkLoadObserver> bulkLoadObservers = getBulkLoadObservers();
203
204 if(bulkLoadObservers != null) {
205 ObserverContext<RegionCoprocessorEnvironment> ctx =
206 new ObserverContext<RegionCoprocessorEnvironment>();
207 ctx.prepare(env);
208
209 for(BulkLoadObserver bulkLoadObserver : bulkLoadObservers) {
210 bulkLoadObserver.prePrepareBulkLoad(ctx, request);
211 }
212 }
213
214 String bulkToken = createStagingDir(baseStagingDir,
215 getActiveUser(), ProtobufUtil.toTableName(request.getTableName())).toString();
216 done.run(PrepareBulkLoadResponse.newBuilder().setBulkToken(bulkToken).build());
217 } catch (IOException e) {
218 ResponseConverter.setControllerException(controller, e);
219 }
220 done.run(null);
221 }
222
223 @Override
224 public void cleanupBulkLoad(RpcController controller,
225 CleanupBulkLoadRequest request,
226 RpcCallback<CleanupBulkLoadResponse> done) {
227 try {
228 List<BulkLoadObserver> bulkLoadObservers = getBulkLoadObservers();
229
230 if(bulkLoadObservers != null) {
231 ObserverContext<RegionCoprocessorEnvironment> ctx =
232 new ObserverContext<RegionCoprocessorEnvironment>();
233 ctx.prepare(env);
234
235 for(BulkLoadObserver bulkLoadObserver : bulkLoadObservers) {
236 bulkLoadObserver.preCleanupBulkLoad(ctx, request);
237 }
238 }
239
240 Path path = new Path(request.getBulkToken());
241 if (!fs.delete(path, true)) {
242 if (fs.exists(path)) {
243 throw new IOException("Failed to clean up " + path);
244 }
245 }
246 LOG.info("Cleaned up " + path + " successfully.");
247 done.run(CleanupBulkLoadResponse.newBuilder().build());
248 } catch (IOException e) {
249 ResponseConverter.setControllerException(controller, e);
250 } finally {
251 UserGroupInformation ugi = getActiveUser().getUGI();
252 try {
253 if (!UserGroupInformation.getLoginUser().equals(ugi)) {
254 FileSystem.closeAllForUGI(ugi);
255 }
256 } catch (IOException e) {
257 LOG.error("Failed to close FileSystem for: " + ugi, e);
258 }
259 }
260 done.run(null);
261 }
262
263 @Override
264 public void secureBulkLoadHFiles(RpcController controller,
265 SecureBulkLoadHFilesRequest request,
266 RpcCallback<SecureBulkLoadHFilesResponse> done) {
267 final List<Pair<byte[], String>> familyPaths = new ArrayList<Pair<byte[], String>>();
268 for(ClientProtos.BulkLoadHFileRequest.FamilyPath el : request.getFamilyPathList()) {
269 familyPaths.add(new Pair(el.getFamily().toByteArray(),el.getPath()));
270 }
271
272 Token userToken = null;
273 if (userProvider.isHadoopSecurityEnabled()) {
274 userToken = new Token(request.getFsToken().getIdentifier().toByteArray(), request.getFsToken()
275 .getPassword().toByteArray(), new Text(request.getFsToken().getKind()), new Text(
276 request.getFsToken().getService()));
277 }
278 final String bulkToken = request.getBulkToken();
279 User user = getActiveUser();
280 final UserGroupInformation ugi = user.getUGI();
281 if (userProvider.isHadoopSecurityEnabled()) {
282 try {
283 Token tok = TokenUtil.obtainToken(conn);
284 if (tok != null) {
285 boolean b = ugi.addToken(tok);
286 LOG.debug("extra token added " + tok + " ret=" + b);
287 }
288 } catch (IOException ioe) {
289 LOG.warn("unable to add token", ioe);
290 }
291 }
292 if (userToken != null) {
293 ugi.addToken(userToken);
294 } else if (userProvider.isHadoopSecurityEnabled()) {
295
296
297 ResponseConverter.setControllerException(controller,
298 new DoNotRetryIOException("User token cannot be null"));
299 done.run(SecureBulkLoadHFilesResponse.newBuilder().setLoaded(false).build());
300 return;
301 }
302
303 Region region = env.getRegion();
304 boolean bypass = false;
305 if (region.getCoprocessorHost() != null) {
306 try {
307 bypass = region.getCoprocessorHost().preBulkLoadHFile(familyPaths);
308 } catch (IOException e) {
309 ResponseConverter.setControllerException(controller, e);
310 done.run(SecureBulkLoadHFilesResponse.newBuilder().setLoaded(false).build());
311 return;
312 }
313 }
314 boolean loaded = false;
315 if (!bypass) {
316
317
318
319
320
321 if (userProvider.isHadoopSecurityEnabled()) {
322 FsDelegationToken targetfsDelegationToken = new FsDelegationToken(userProvider, "renewer");
323 try {
324 targetfsDelegationToken.acquireDelegationToken(fs);
325 } catch (IOException e) {
326 ResponseConverter.setControllerException(controller, e);
327 done.run(SecureBulkLoadHFilesResponse.newBuilder().setLoaded(false).build());
328 return;
329 }
330 Token<?> targetFsToken = targetfsDelegationToken.getUserToken();
331 if (targetFsToken != null
332 && (userToken == null || !targetFsToken.getService().equals(userToken.getService()))) {
333 ugi.addToken(targetFsToken);
334 }
335 }
336
337 loaded = ugi.doAs(new PrivilegedAction<Boolean>() {
338 @Override
339 public Boolean run() {
340 FileSystem fs = null;
341 try {
342 Configuration conf = env.getConfiguration();
343 fs = FileSystem.get(conf);
344 for(Pair<byte[], String> el: familyPaths) {
345 Path stageFamily = new Path(bulkToken, Bytes.toString(el.getFirst()));
346 if(!fs.exists(stageFamily)) {
347 fs.mkdirs(stageFamily);
348 fs.setPermission(stageFamily, PERM_ALL_ACCESS);
349 }
350 }
351
352
353 return env.getRegion().bulkLoadHFiles(familyPaths, true,
354 new SecureBulkLoadListener(fs, bulkToken, conf));
355 } catch (Exception e) {
356 LOG.error("Failed to complete bulk load", e);
357 }
358 return false;
359 }
360 });
361 }
362 if (region.getCoprocessorHost() != null) {
363 try {
364 loaded = region.getCoprocessorHost().postBulkLoadHFile(familyPaths, loaded);
365 } catch (IOException e) {
366 ResponseConverter.setControllerException(controller, e);
367 done.run(SecureBulkLoadHFilesResponse.newBuilder().setLoaded(false).build());
368 return;
369 }
370 }
371 done.run(SecureBulkLoadHFilesResponse.newBuilder().setLoaded(loaded).build());
372 }
373
374 private List<BulkLoadObserver> getBulkLoadObservers() {
375 List<BulkLoadObserver> coprocessorList =
376 this.env.getRegion().getCoprocessorHost().findCoprocessors(BulkLoadObserver.class);
377
378 return coprocessorList;
379 }
380
381 private Path createStagingDir(Path baseDir,
382 User user,
383 TableName tableName) throws IOException {
384 String tblName = tableName.getNameAsString().replace(":", "_");
385 String randomDir = user.getShortName()+"__"+ tblName +"__"+
386 (new BigInteger(RANDOM_WIDTH, random).toString(RANDOM_RADIX));
387 return createStagingDir(baseDir, user, randomDir);
388 }
389
390 private Path createStagingDir(Path baseDir,
391 User user,
392 String randomDir) throws IOException {
393 Path p = new Path(baseDir, randomDir);
394 fs.mkdirs(p, PERM_ALL_ACCESS);
395 fs.setPermission(p, PERM_ALL_ACCESS);
396 return p;
397 }
398
399 private User getActiveUser() {
400 User user = RpcServer.getRequestUser();
401 if (user == null) {
402 return null;
403 }
404
405
406 if (userProvider.isHadoopSecurityEnabled()
407 && "simple".equalsIgnoreCase(conf.get(User.HBASE_SECURITY_CONF_KEY))) {
408 return User.createUserForTesting(conf, user.getShortName(), new String[]{});
409 }
410
411 return user;
412 }
413
414 @Override
415 public Service getService() {
416 return this;
417 }
418
419 private static class SecureBulkLoadListener implements BulkLoadListener {
420
421 private FileSystem fs;
422 private String stagingDir;
423 private Configuration conf;
424
425 private FileSystem srcFs = null;
426 private Map<String, FsPermission> origPermissions = null;
427
428 public SecureBulkLoadListener(FileSystem fs, String stagingDir, Configuration conf) {
429 this.fs = fs;
430 this.stagingDir = stagingDir;
431 this.conf = conf;
432 this.origPermissions = new HashMap<String, FsPermission>();
433 }
434
435 @Override
436 public String prepareBulkLoad(final byte[] family, final String srcPath) throws IOException {
437 Path p = new Path(srcPath);
438 Path stageP = new Path(stagingDir, new Path(Bytes.toString(family), p.getName()));
439
440
441 if (p.equals(stageP)) {
442 LOG.debug(p.getName()
443 + " is already available in staging directory. Skipping copy or rename.");
444 return stageP.toString();
445 }
446
447 if (srcFs == null) {
448 srcFs = FileSystem.newInstance(p.toUri(), conf);
449 }
450
451 if(!isFile(p)) {
452 throw new IOException("Path does not reference a file: " + p);
453 }
454
455
456 if (!FSHDFSUtils.isSameHdfs(conf, srcFs, fs)) {
457 LOG.debug("Bulk-load file " + srcPath + " is on different filesystem than " +
458 "the destination filesystem. Copying file over to destination staging dir.");
459 FileUtil.copy(srcFs, p, fs, stageP, false, conf);
460 } else {
461 LOG.debug("Moving " + p + " to " + stageP);
462 FileStatus origFileStatus = fs.getFileStatus(p);
463 origPermissions.put(srcPath, origFileStatus.getPermission());
464 if(!fs.rename(p, stageP)) {
465 throw new IOException("Failed to move HFile: " + p + " to " + stageP);
466 }
467 }
468 fs.setPermission(stageP, PERM_ALL_ACCESS);
469 return stageP.toString();
470 }
471
472 @Override
473 public void doneBulkLoad(byte[] family, String srcPath) throws IOException {
474 LOG.debug("Bulk Load done for: " + srcPath);
475 closeSrcFs();
476 }
477
478 private void closeSrcFs() throws IOException {
479 if (srcFs != null) {
480 srcFs.close();
481 srcFs = null;
482 }
483 }
484
485 @Override
486 public void failedBulkLoad(final byte[] family, final String srcPath) throws IOException {
487 try {
488 Path p = new Path(srcPath);
489 if (srcFs == null) {
490 srcFs = FileSystem.newInstance(p.toUri(), conf);
491 }
492 if (!FSHDFSUtils.isSameHdfs(conf, srcFs, fs)) {
493
494 return;
495 }
496 Path stageP = new Path(stagingDir, new Path(Bytes.toString(family), p.getName()));
497
498
499
500 if (p.equals(stageP)) {
501 LOG.debug(p.getName() + " is already available in source directory. Skipping rename.");
502 return;
503 }
504
505 LOG.debug("Moving " + stageP + " back to " + p);
506 if (!fs.rename(stageP, p))
507 throw new IOException("Failed to move HFile: " + stageP + " to " + p);
508
509
510 if (origPermissions.containsKey(srcPath)) {
511 fs.setPermission(p, origPermissions.get(srcPath));
512 } else {
513 LOG.warn("Can't find previous permission for path=" + srcPath);
514 }
515 } finally {
516 closeSrcFs();
517 }
518 }
519
520
521
522
523
524
525
526
527 private boolean isFile(Path p) throws IOException {
528 FileStatus status = srcFs.getFileStatus(p);
529 boolean isFile = !status.isDirectory();
530 try {
531 isFile = isFile && !(Boolean)Methods.call(FileStatus.class, status, "isSymlink", null, null);
532 } catch (Exception e) {
533 }
534 return isFile;
535 }
536 }
537 }