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