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  package org.apache.hadoop.hbase.io.hfile;
19  
20  import static org.junit.Assert.*;
21  
22  import java.io.DataInputStream;
23  import java.io.DataOutputStream;
24  import java.io.IOException;
25  import java.security.SecureRandom;
26  import java.util.List;
27  import java.util.UUID;
28  
29  import org.apache.commons.logging.Log;
30  import org.apache.commons.logging.LogFactory;
31  import org.apache.hadoop.conf.Configuration;
32  import org.apache.hadoop.fs.FSDataInputStream;
33  import org.apache.hadoop.fs.FSDataOutputStream;
34  import org.apache.hadoop.fs.FileSystem;
35  import org.apache.hadoop.fs.Path;
36  import org.apache.hadoop.hbase.HBaseTestingUtility;
37  import org.apache.hadoop.hbase.HConstants;
38  import org.apache.hadoop.hbase.KeyValue;
39  import org.apache.hadoop.hbase.SmallTests;
40  import org.apache.hadoop.hbase.io.compress.Compression;
41  import org.apache.hadoop.hbase.io.crypto.Cipher;
42  import org.apache.hadoop.hbase.io.crypto.Encryption;
43  import org.apache.hadoop.hbase.io.crypto.KeyProviderForTesting;
44  import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
45  import org.apache.hadoop.hbase.util.Bytes;
46  import org.apache.hadoop.hbase.util.test.RedundantKVGenerator;
47  
48  import org.junit.BeforeClass;
49  import org.junit.Test;
50  import org.junit.experimental.categories.Category;
51  
52  @Category(SmallTests.class)
53  public class TestHFileEncryption {
54    private static final Log LOG = LogFactory.getLog(TestHFileEncryption.class);
55    private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
56    private static final SecureRandom RNG = new SecureRandom();
57  
58    private static FileSystem fs;
59    private static Encryption.Context cryptoContext;
60  
61    @BeforeClass
62    public static void setUp() throws Exception {
63      Configuration conf = TEST_UTIL.getConfiguration();
64      conf.set(HConstants.CRYPTO_KEYPROVIDER_CONF_KEY, KeyProviderForTesting.class.getName());
65      conf.set(HConstants.CRYPTO_MASTERKEY_NAME_CONF_KEY, "hbase");
66      conf.setInt("hfile.format.version", 3);
67  
68      fs = FileSystem.get(conf);
69  
70      cryptoContext = Encryption.newContext(conf);
71      Cipher aes = Encryption.getCipher(conf, "AES");
72      assertNotNull(aes);
73      cryptoContext.setCipher(aes);
74      byte[] key = new byte[aes.getKeyLength()];
75      RNG.nextBytes(key);
76      cryptoContext.setKey(key);
77    }
78  
79    private int writeBlock(FSDataOutputStream os, HFileContext fileContext, int size)
80        throws IOException {
81      HFileBlock.Writer hbw = new HFileBlock.Writer(null, fileContext);
82      DataOutputStream dos = hbw.startWriting(BlockType.DATA);
83      for (int j = 0; j < size; j++) {
84        dos.writeInt(j);
85      }
86      hbw.writeHeaderAndData(os);
87      LOG.info("Wrote a block at " + os.getPos() + " with" +
88          " onDiskSizeWithHeader=" + hbw.getOnDiskSizeWithHeader() +
89          " uncompressedSizeWithoutHeader=" + hbw.getOnDiskSizeWithoutHeader() +
90          " uncompressedSizeWithoutHeader=" + hbw.getUncompressedSizeWithoutHeader());
91      return hbw.getOnDiskSizeWithHeader();
92    }
93  
94    private long readAndVerifyBlock(long pos, HFileBlock.FSReaderV2 hbr, int size)
95        throws IOException {
96      HFileBlock b = hbr.readBlockData(pos, -1, -1, false);
97      assertEquals(0, HFile.getChecksumFailuresCount());
98      b.sanityCheck();
99      LOG.info("Read a block at " + pos + " with" +
100         " onDiskSizeWithHeader=" + b.getOnDiskSizeWithHeader() +
101         " uncompressedSizeWithoutHeader=" + b.getOnDiskSizeWithoutHeader() +
102         " uncompressedSizeWithoutHeader=" + b.getUncompressedSizeWithoutHeader());
103     DataInputStream dis = b.getByteStream();
104     for (int i = 0; i < size; i++) {
105       int read = dis.readInt();
106       if (read != i) {
107         fail("Block data corrupt at element " + i);
108       }
109     }
110     return b.getOnDiskSizeWithHeader();
111   }
112 
113   @Test(timeout=20000)
114   public void testDataBlockEncryption() throws IOException {
115     final int blocks = 10;
116     final int[] blockSizes = new int[blocks];
117     for (int i = 0; i < blocks; i++) {
118       blockSizes[i] = (1024 + RNG.nextInt(1024 * 63)) / Bytes.SIZEOF_INT;
119     }
120     for (Compression.Algorithm compression : TestHFileBlock.COMPRESSION_ALGORITHMS) {
121       Path path = new Path(TEST_UTIL.getDataTestDir(), "block_v3_" + compression + "_AES");
122       LOG.info("testDataBlockEncryption: encryption=AES compression=" + compression);
123       long totalSize = 0;
124       HFileContext fileContext = new HFileContextBuilder()
125         .withCompression(compression)
126         .withEncryptionContext(cryptoContext)
127         .build();
128       FSDataOutputStream os = fs.create(path);
129       try {
130         for (int i = 0; i < blocks; i++) {
131           totalSize += writeBlock(os, fileContext, blockSizes[i]);
132         }
133       } finally {
134         os.close();
135       }
136       FSDataInputStream is = fs.open(path);
137       try {
138         HFileBlock.FSReaderV2 hbr = new HFileBlock.FSReaderV2(is, totalSize, fileContext);
139         long pos = 0;
140         for (int i = 0; i < blocks; i++) {
141           pos += readAndVerifyBlock(pos, hbr, blockSizes[i]);
142         }
143       } finally {
144         is.close();
145       }
146     }
147   }
148 
149   @Test(timeout=20000)
150   public void testHFileEncryptionMetadata() throws Exception {
151     Configuration conf = TEST_UTIL.getConfiguration();
152     CacheConfig cacheConf = new CacheConfig(conf);
153 
154     HFileContext fileContext = new HFileContextBuilder()
155     .withEncryptionContext(cryptoContext)
156     .build();
157 
158     // write a simple encrypted hfile
159     Path path = new Path(TEST_UTIL.getDataTestDir(), "cryptometa.hfile");
160     FSDataOutputStream out = fs.create(path);
161     HFile.Writer writer = HFile.getWriterFactory(conf, cacheConf)
162       .withOutputStream(out)
163       .withFileContext(fileContext)
164       .create();
165     writer.append("foo".getBytes(), "value".getBytes());
166     writer.close();
167     out.close();
168 
169     // read it back in and validate correct crypto metadata
170     HFile.Reader reader = HFile.createReader(fs, path, cacheConf, conf);
171     reader.loadFileInfo();
172     FixedFileTrailer trailer = reader.getTrailer();
173     assertNotNull(trailer.getEncryptionKey());
174     Encryption.Context readerContext = reader.getFileContext().getEncryptionContext();
175     assertEquals(readerContext.getCipher().getName(), cryptoContext.getCipher().getName());
176     assertTrue(Bytes.equals(readerContext.getKeyBytes(),
177       cryptoContext.getKeyBytes()));
178   }
179 
180   @Test(timeout=60000)
181   public void testHFileEncryption() throws Exception {
182     // Create 1000 random test KVs
183     RedundantKVGenerator generator = new RedundantKVGenerator();
184     List<KeyValue> testKvs = generator.generateTestKeyValues(1000);
185 
186     // Iterate through data block encoding and compression combinations
187     Configuration conf = TEST_UTIL.getConfiguration();
188     CacheConfig cacheConf = new CacheConfig(conf);
189     for (DataBlockEncoding encoding: DataBlockEncoding.values()) {
190       for (Compression.Algorithm compression: TestHFileBlock.COMPRESSION_ALGORITHMS) {
191         HFileContext fileContext = new HFileContextBuilder()
192           .withBlockSize(4096) // small blocks
193           .withEncryptionContext(cryptoContext)
194           .withCompression(compression)
195           .withDataBlockEncoding(encoding)
196           .build();
197         // write a new test HFile
198         LOG.info("Writing with " + fileContext);
199         Path path = new Path(TEST_UTIL.getDataTestDir(), UUID.randomUUID().toString() + ".hfile");
200         FSDataOutputStream out = fs.create(path);
201         HFile.Writer writer = HFile.getWriterFactory(conf, cacheConf)
202           .withOutputStream(out)
203           .withFileContext(fileContext)
204           .create();
205         for (KeyValue kv: testKvs) {
206           writer.append(kv);
207         }
208         writer.close();
209         out.close();
210 
211         // read it back in
212         LOG.info("Reading with " + fileContext);
213         HFile.Reader reader = HFile.createReader(fs, path, cacheConf, conf);
214         reader.loadFileInfo();
215         FixedFileTrailer trailer = reader.getTrailer();
216         assertNotNull(trailer.getEncryptionKey());
217         HFileScanner scanner = reader.getScanner(false, false);
218         assertTrue("Initial seekTo failed", scanner.seekTo());
219         int i = 0;
220         do {
221           KeyValue kv = scanner.getKeyValue();
222           assertTrue("Read back an unexpected or invalid KV", testKvs.contains(kv));
223           i++;
224         } while (scanner.next());
225         reader.close();
226 
227         assertEquals("Did not read back as many KVs as written", i, testKvs.size());
228 
229         // Test random seeks with pread
230         LOG.info("Random seeking with " + fileContext);
231         reader = HFile.createReader(fs, path, cacheConf, conf);
232         scanner = reader.getScanner(false, true);
233         assertTrue("Initial seekTo failed", scanner.seekTo());
234         for (i = 0; i < 100; i++) {
235           KeyValue kv = testKvs.get(RNG.nextInt(testKvs.size()));
236           assertEquals("Unable to find KV as expected: " + kv, scanner.seekTo(kv.getKey()), 0);
237         }
238         reader.close();
239       }
240     }
241   }
242 
243 }