Source code

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.io.hfile;
019
020import static org.junit.Assert.assertEquals;
021import static org.junit.Assert.assertFalse;
022import static org.junit.Assert.assertNotNull;
023import static org.junit.Assert.assertTrue;
024import static org.junit.Assert.fail;
025
026import java.io.DataInputStream;
027import java.io.DataOutputStream;
028import java.io.IOException;
029import java.security.SecureRandom;
030import java.util.List;
031import org.apache.hadoop.conf.Configuration;
032import org.apache.hadoop.fs.FSDataInputStream;
033import org.apache.hadoop.fs.FSDataOutputStream;
034import org.apache.hadoop.fs.FileSystem;
035import org.apache.hadoop.fs.Path;
036import org.apache.hadoop.hbase.Cell;
037import org.apache.hadoop.hbase.HBaseClassTestRule;
038import org.apache.hadoop.hbase.HBaseTestingUtility;
039import org.apache.hadoop.hbase.HConstants;
040import org.apache.hadoop.hbase.KeyValue;
041import org.apache.hadoop.hbase.KeyValueUtil;
042import org.apache.hadoop.hbase.io.ByteBuffAllocator;
043import org.apache.hadoop.hbase.io.FSDataInputStreamWrapper;
044import org.apache.hadoop.hbase.io.compress.Compression;
045import org.apache.hadoop.hbase.io.crypto.Cipher;
046import org.apache.hadoop.hbase.io.crypto.Encryption;
047import org.apache.hadoop.hbase.io.crypto.KeyProviderForTesting;
048import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
049import org.apache.hadoop.hbase.testclassification.IOTests;
050import org.apache.hadoop.hbase.testclassification.SmallTests;
051import org.apache.hadoop.hbase.util.Bytes;
052import org.apache.hadoop.hbase.util.RedundantKVGenerator;
053import org.junit.BeforeClass;
054import org.junit.ClassRule;
055import org.junit.Test;
056import org.junit.experimental.categories.Category;
057import org.slf4j.Logger;
058import org.slf4j.LoggerFactory;
059
060@Category({IOTests.class, SmallTests.class})
061public class TestHFileEncryption {
062
063  @ClassRule
064  public static final HBaseClassTestRule CLASS_RULE =
065      HBaseClassTestRule.forClass(TestHFileEncryption.class);
066
067  private static final Logger LOG = LoggerFactory.getLogger(TestHFileEncryption.class);
068  private static final HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
069  private static final SecureRandom RNG = new SecureRandom();
070
071  private static FileSystem fs;
072  private static Encryption.Context cryptoContext;
073
074  @BeforeClass
075  public static void setUp() throws Exception {
076    Configuration conf = TEST_UTIL.getConfiguration();
077    // Disable block cache in this test.
078    conf.setFloat(HConstants.HFILE_BLOCK_CACHE_SIZE_KEY, 0.0f);
079    conf.set(HConstants.CRYPTO_KEYPROVIDER_CONF_KEY, KeyProviderForTesting.class.getName());
080    conf.set(HConstants.CRYPTO_MASTERKEY_NAME_CONF_KEY, "hbase");
081    conf.setInt("hfile.format.version", 3);
082
083    fs = FileSystem.get(conf);
084
085    cryptoContext = Encryption.newContext(conf);
086    String algorithm =
087        conf.get(HConstants.CRYPTO_KEY_ALGORITHM_CONF_KEY, HConstants.CIPHER_AES);
088    Cipher aes = Encryption.getCipher(conf, algorithm);
089    assertNotNull(aes);
090    cryptoContext.setCipher(aes);
091    byte[] key = new byte[aes.getKeyLength()];
092    RNG.nextBytes(key);
093    cryptoContext.setKey(key);
094  }
095
096  private int writeBlock(FSDataOutputStream os, HFileContext fileContext, int size)
097      throws IOException {
098    HFileBlock.Writer hbw = new HFileBlock.Writer(null, fileContext);
099    DataOutputStream dos = hbw.startWriting(BlockType.DATA);
100    for (int j = 0; j < size; j++) {
101      dos.writeInt(j);
102    }
103    hbw.writeHeaderAndData(os);
104    LOG.info("Wrote a block at " + os.getPos() + " with" +
105        " onDiskSizeWithHeader=" + hbw.getOnDiskSizeWithHeader() +
106        " uncompressedSizeWithoutHeader=" + hbw.getOnDiskSizeWithoutHeader() +
107        " uncompressedSizeWithoutHeader=" + hbw.getUncompressedSizeWithoutHeader());
108    return hbw.getOnDiskSizeWithHeader();
109  }
110
111  private long readAndVerifyBlock(long pos, HFileContext ctx, HFileBlock.FSReaderImpl hbr, int size)
112      throws IOException {
113    HFileBlock b = hbr.readBlockData(pos, -1, false, false, true);
114    assertEquals(0, HFile.getAndResetChecksumFailuresCount());
115    b.sanityCheck();
116    assertFalse(b.isUnpacked());
117    b = b.unpack(ctx, hbr);
118    LOG.info("Read a block at " + pos + " with" +
119        " onDiskSizeWithHeader=" + b.getOnDiskSizeWithHeader() +
120        " uncompressedSizeWithoutHeader=" + b.getOnDiskSizeWithoutHeader() +
121        " uncompressedSizeWithoutHeader=" + b.getUncompressedSizeWithoutHeader());
122    DataInputStream dis = b.getByteStream();
123    for (int i = 0; i < size; i++) {
124      int read = dis.readInt();
125      if (read != i) {
126        fail("Block data corrupt at element " + i);
127      }
128    }
129    return b.getOnDiskSizeWithHeader();
130  }
131
132  @Test
133  public void testDataBlockEncryption() throws IOException {
134    final int blocks = 10;
135    final int[] blockSizes = new int[blocks];
136    for (int i = 0; i < blocks; i++) {
137      blockSizes[i] = (1024 + RNG.nextInt(1024 * 63)) / Bytes.SIZEOF_INT;
138    }
139    for (Compression.Algorithm compression : TestHFileBlock.COMPRESSION_ALGORITHMS) {
140      Path path = new Path(TEST_UTIL.getDataTestDir(), "block_v3_" + compression + "_AES");
141      LOG.info("testDataBlockEncryption: encryption=AES compression=" + compression);
142      long totalSize = 0;
143      HFileContext fileContext = new HFileContextBuilder()
144        .withCompression(compression)
145        .withEncryptionContext(cryptoContext)
146        .build();
147      FSDataOutputStream os = fs.create(path);
148      try {
149        for (int i = 0; i < blocks; i++) {
150          totalSize += writeBlock(os, fileContext, blockSizes[i]);
151        }
152      } finally {
153        os.close();
154      }
155      FSDataInputStream is = fs.open(path);
156      ReaderContext context = new ReaderContextBuilder()
157          .withInputStreamWrapper(new FSDataInputStreamWrapper(is))
158          .withFilePath(path)
159          .withFileSystem(fs)
160          .withFileSize(totalSize).build();
161      try {
162        HFileBlock.FSReaderImpl hbr = new HFileBlock.FSReaderImpl(context, fileContext,
163            ByteBuffAllocator.HEAP);
164        long pos = 0;
165        for (int i = 0; i < blocks; i++) {
166          pos += readAndVerifyBlock(pos, fileContext, hbr, blockSizes[i]);
167        }
168      } finally {
169        is.close();
170      }
171    }
172  }
173
174  @Test
175  public void testHFileEncryptionMetadata() throws Exception {
176    Configuration conf = TEST_UTIL.getConfiguration();
177    CacheConfig cacheConf = new CacheConfig(conf);
178    HFileContext fileContext = new HFileContextBuilder()
179        .withEncryptionContext(cryptoContext)
180        .build();
181
182    // write a simple encrypted hfile
183    Path path = new Path(TEST_UTIL.getDataTestDir(), "cryptometa.hfile");
184    FSDataOutputStream out = fs.create(path);
185    HFile.Writer writer = HFile.getWriterFactory(conf, cacheConf)
186        .withOutputStream(out)
187        .withFileContext(fileContext)
188        .create();
189    try {
190      KeyValue kv = new KeyValue("foo".getBytes(), "f1".getBytes(), null, "value".getBytes());
191      writer.append(kv);
192    } finally {
193      writer.close();
194      out.close();
195    }
196
197    // read it back in and validate correct crypto metadata
198    HFile.Reader reader = HFile.createReader(fs, path, cacheConf, true, conf);
199    try {
200      FixedFileTrailer trailer = reader.getTrailer();
201      assertNotNull(trailer.getEncryptionKey());
202      Encryption.Context readerContext = reader.getFileContext().getEncryptionContext();
203      assertEquals(readerContext.getCipher().getName(), cryptoContext.getCipher().getName());
204      assertTrue(Bytes.equals(readerContext.getKeyBytes(),
205          cryptoContext.getKeyBytes()));
206    } finally {
207      reader.close();
208    }
209  }
210
211  @Test
212  public void testHFileEncryption() throws Exception {
213    // Create 1000 random test KVs
214    RedundantKVGenerator generator = new RedundantKVGenerator();
215    List<KeyValue> testKvs = generator.generateTestKeyValues(1000);
216
217    // Iterate through data block encoding and compression combinations
218    Configuration conf = TEST_UTIL.getConfiguration();
219    CacheConfig cacheConf = new CacheConfig(conf);
220    for (DataBlockEncoding encoding: DataBlockEncoding.values()) {
221      for (Compression.Algorithm compression: TestHFileBlock.COMPRESSION_ALGORITHMS) {
222        HFileContext fileContext = new HFileContextBuilder()
223          .withBlockSize(4096) // small blocks
224          .withEncryptionContext(cryptoContext)
225          .withCompression(compression)
226          .withDataBlockEncoding(encoding)
227          .build();
228        // write a new test HFile
229        LOG.info("Writing with " + fileContext);
230        Path path = new Path(TEST_UTIL.getDataTestDir(),
231                        TEST_UTIL.getRandomUUID().toString() + ".hfile");
232        FSDataOutputStream out = fs.create(path);
233        HFile.Writer writer = HFile.getWriterFactory(conf, cacheConf)
234          .withOutputStream(out)
235          .withFileContext(fileContext)
236          .create();
237        try {
238          for (KeyValue kv: testKvs) {
239            writer.append(kv);
240          }
241        } finally {
242          writer.close();
243          out.close();
244        }
245
246        // read it back in
247        LOG.info("Reading with " + fileContext);
248        int i = 0;
249        HFileScanner scanner = null;
250        HFile.Reader reader = HFile.createReader(fs, path, cacheConf, true, conf);
251        try {
252          FixedFileTrailer trailer = reader.getTrailer();
253          assertNotNull(trailer.getEncryptionKey());
254          scanner = reader.getScanner(false, false);
255          assertTrue("Initial seekTo failed", scanner.seekTo());
256          do {
257            Cell kv = scanner.getCell();
258            assertTrue("Read back an unexpected or invalid KV",
259              testKvs.contains(KeyValueUtil.ensureKeyValue(kv)));
260            i++;
261          } while (scanner.next());
262        } finally {
263          reader.close();
264          scanner.close();
265        }
266
267        assertEquals("Did not read back as many KVs as written", i, testKvs.size());
268
269        // Test random seeks with pread
270        LOG.info("Random seeking with " + fileContext);
271        reader = HFile.createReader(fs, path, cacheConf, true, conf);
272        try {
273          scanner = reader.getScanner(false, true);
274          assertTrue("Initial seekTo failed", scanner.seekTo());
275          for (i = 0; i < 100; i++) {
276            KeyValue kv = testKvs.get(RNG.nextInt(testKvs.size()));
277            assertEquals("Unable to find KV as expected: " + kv, 0, scanner.seekTo(kv));
278          }
279        } finally {
280          scanner.close();
281          reader.close();
282        }
283      }
284    }
285  }
286
287}