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.security;
019
020import static org.apache.hadoop.hbase.ipc.TestProtobufRpcServiceImpl.SERVICE;
021import static org.apache.hadoop.hbase.ipc.TestProtobufRpcServiceImpl.newBlockingStub;
022import static org.junit.jupiter.api.Assertions.assertEquals;
023import static org.junit.jupiter.api.Assertions.assertNull;
024import static org.mockito.Mockito.mock;
025import static org.mockito.Mockito.when;
026
027import java.io.File;
028import java.io.IOException;
029import java.net.InetSocketAddress;
030import java.nio.file.Path;
031import java.nio.file.Paths;
032import java.security.Security;
033import java.time.Duration;
034import java.util.ArrayList;
035import java.util.List;
036import java.util.Objects;
037import java.util.concurrent.CountDownLatch;
038import java.util.concurrent.TimeUnit;
039import java.util.stream.Stream;
040import org.apache.commons.io.FileUtils;
041import org.apache.hadoop.conf.Configuration;
042import org.apache.hadoop.hbase.HBaseCommonTestingUtil;
043import org.apache.hadoop.hbase.HBaseConfiguration;
044import org.apache.hadoop.hbase.HBaseParameterizedTestTemplate;
045import org.apache.hadoop.hbase.HBaseServerBase;
046import org.apache.hadoop.hbase.ServerName;
047import org.apache.hadoop.hbase.io.FileChangeWatcher;
048import org.apache.hadoop.hbase.io.crypto.tls.KeyStoreFileType;
049import org.apache.hadoop.hbase.io.crypto.tls.X509KeyType;
050import org.apache.hadoop.hbase.io.crypto.tls.X509TestContext;
051import org.apache.hadoop.hbase.io.crypto.tls.X509TestContextProvider;
052import org.apache.hadoop.hbase.io.crypto.tls.X509Util;
053import org.apache.hadoop.hbase.ipc.AbstractRpcClient;
054import org.apache.hadoop.hbase.ipc.FifoRpcScheduler;
055import org.apache.hadoop.hbase.ipc.HBaseRpcController;
056import org.apache.hadoop.hbase.ipc.HBaseRpcControllerImpl;
057import org.apache.hadoop.hbase.ipc.NettyRpcClient;
058import org.apache.hadoop.hbase.ipc.NettyRpcServer;
059import org.apache.hadoop.hbase.ipc.RpcScheduler;
060import org.apache.hadoop.hbase.ipc.RpcServer;
061import org.apache.hadoop.hbase.net.Address;
062import org.apache.hadoop.hbase.testclassification.RPCTests;
063import org.apache.hadoop.hbase.testclassification.SmallTests;
064import org.apache.hadoop.hbase.util.NettyEventLoopGroupConfig;
065import org.bouncycastle.jce.provider.BouncyCastleProvider;
066import org.junit.jupiter.api.AfterAll;
067import org.junit.jupiter.api.AfterEach;
068import org.junit.jupiter.api.BeforeAll;
069import org.junit.jupiter.api.BeforeEach;
070import org.junit.jupiter.api.Tag;
071import org.junit.jupiter.api.TestTemplate;
072import org.junit.jupiter.params.provider.Arguments;
073import org.slf4j.Logger;
074import org.slf4j.LoggerFactory;
075
076import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
077
078import org.apache.hadoop.hbase.shaded.ipc.protobuf.generated.TestProtos;
079import org.apache.hadoop.hbase.shaded.ipc.protobuf.generated.TestRpcServiceProtos;
080
081@Tag(RPCTests.TAG)
082@Tag(SmallTests.TAG)
083@HBaseParameterizedTestTemplate(name = "{index}: keyType={0}, storeFileType={1}")
084public class TestNettyTLSIPCFileWatcher {
085
086  private static final Logger LOG = LoggerFactory.getLogger(TestNettyTLSIPCFileWatcher.class);
087
088  private static final Configuration CONF = HBaseConfiguration.create();
089  private static final HBaseCommonTestingUtil UTIL = new HBaseCommonTestingUtil(CONF);
090  private static HBaseServerBase<?> SERVER;
091  private static X509TestContextProvider PROVIDER;
092  private static NettyEventLoopGroupConfig EVENT_LOOP_GROUP_CONFIG;
093
094  private X509TestContext x509TestContext;
095
096  private X509KeyType keyType;
097
098  private KeyStoreFileType storeFileType;
099
100  public static Stream<Arguments> parameters() {
101    List<Arguments> params = new ArrayList<>();
102    for (X509KeyType caKeyType : X509KeyType.values()) {
103      for (KeyStoreFileType ks : KeyStoreFileType.values()) {
104        params.add(Arguments.of(caKeyType, ks));
105      }
106    }
107    return params.stream();
108  }
109
110  public TestNettyTLSIPCFileWatcher(X509KeyType keyType, KeyStoreFileType storeFileType) {
111    this.keyType = keyType;
112    this.storeFileType = storeFileType;
113  }
114
115  @BeforeAll
116  public static void setUpBeforeClass() throws IOException {
117    Security.addProvider(new BouncyCastleProvider());
118    File dir =
119      new File(UTIL.getDataTestDir(TestNettyTLSIPCFileWatcher.class.getSimpleName()).toString())
120        .getCanonicalFile();
121    FileUtils.forceMkdir(dir);
122    // server must enable tls
123    CONF.setBoolean(X509Util.HBASE_SERVER_NETTY_TLS_ENABLED, true);
124    PROVIDER = new X509TestContextProvider(CONF, dir);
125    EVENT_LOOP_GROUP_CONFIG =
126      NettyEventLoopGroupConfig.setup(CONF, TestNettyTLSIPCFileWatcher.class.getSimpleName());
127    SERVER = mock(HBaseServerBase.class);
128    when(SERVER.getEventLoopGroupConfig()).thenReturn(EVENT_LOOP_GROUP_CONFIG);
129  }
130
131  @AfterAll
132  public static void tearDownAfterClass() throws InterruptedException {
133    Security.removeProvider(BouncyCastleProvider.PROVIDER_NAME);
134    EVENT_LOOP_GROUP_CONFIG.group().shutdownGracefully().sync();
135    UTIL.cleanupTestDir();
136  }
137
138  @BeforeEach
139  public void setUp() throws Exception {
140    x509TestContext = PROVIDER.get(keyType, keyType, "keyPa$$word".toCharArray());
141    x509TestContext.setConfigurations(storeFileType, storeFileType);
142    CONF.setBoolean(X509Util.HBASE_SERVER_NETTY_TLS_SUPPORTPLAINTEXT, false);
143    CONF.setBoolean(X509Util.HBASE_CLIENT_NETTY_TLS_ENABLED, true);
144    CONF.setBoolean(X509Util.TLS_CERT_RELOAD, true);
145    CONF.setLong(X509Util.HBASE_TLS_FILEPOLL_INTERVAL_MILLIS, 10);
146  }
147
148  @AfterEach
149  public void tearDown() {
150    x509TestContext.clearConfigurations();
151    x509TestContext.getConf().unset(X509Util.TLS_CONFIG_OCSP);
152    x509TestContext.getConf().unset(X509Util.TLS_CONFIG_CLR);
153    x509TestContext.getConf().unset(X509Util.TLS_CONFIG_PROTOCOL);
154    x509TestContext.getConf().unset(X509Util.HBASE_TLS_FILEPOLL_INTERVAL_MILLIS);
155    System.clearProperty("com.sun.net.ssl.checkRevocation");
156    System.clearProperty("com.sun.security.enableCRLDP");
157    Security.setProperty("ocsp.enable", Boolean.FALSE.toString());
158    Security.setProperty("com.sun.security.enableCRLDP", Boolean.FALSE.toString());
159  }
160
161  @TestTemplate
162  public void testReplaceServerKeystore() throws Exception {
163    Configuration clientConf = new Configuration(CONF);
164    RpcServer rpcServer = createRpcServer("testRpcServer",
165      Lists.newArrayList(new RpcServer.BlockingServiceAndInterface(SERVICE, null)),
166      new InetSocketAddress("localhost", 0), CONF, new FifoRpcScheduler(CONF, 1));
167
168    try {
169      rpcServer.start();
170
171      try (AbstractRpcClient<?> client = new NettyRpcClient(clientConf)) {
172        TestRpcServiceProtos.TestProtobufRpcProto.BlockingInterface stub =
173          newBlockingStub(client, rpcServer.getListenerAddress());
174        HBaseRpcController pcrc = new HBaseRpcControllerImpl();
175        String message = "hello";
176        assertEquals(message,
177          stub.echo(pcrc, TestProtos.EchoRequestProto.newBuilder().setMessage(message).build())
178            .getMessage());
179        assertNull(pcrc.cellScanner());
180      }
181
182      // truststore file change latch
183      final CountDownLatch latch = new CountDownLatch(1);
184      final Path trustStorePath = Paths.get(CONF.get(X509Util.TLS_CONFIG_TRUSTSTORE_LOCATION));
185      createAndStartFileWatcher(trustStorePath, latch, Duration.ofMillis(20));
186
187      // Replace keystore
188      x509TestContext.regenerateStores(keyType, keyType, storeFileType, storeFileType);
189
190      if (!latch.await(1, TimeUnit.SECONDS)) {
191        throw new AssertionError("Timed out waiting for truststore file to be changed");
192      }
193
194      try (AbstractRpcClient<?> client = new NettyRpcClient(clientConf)) {
195        TestRpcServiceProtos.TestProtobufRpcProto.BlockingInterface stub =
196          newBlockingStub(client, rpcServer.getListenerAddress());
197        HBaseRpcController pcrc = new HBaseRpcControllerImpl();
198        String message = "hello";
199        assertEquals(message,
200          stub.echo(pcrc, TestProtos.EchoRequestProto.newBuilder().setMessage(message).build())
201            .getMessage());
202        assertNull(pcrc.cellScanner());
203      }
204
205    } finally {
206      rpcServer.stop();
207    }
208  }
209
210  @TestTemplate
211  public void testReplaceClientAndServerKeystore() throws Exception {
212    Configuration clientConf = new Configuration(CONF);
213    RpcServer rpcServer = createRpcServer("testRpcServer",
214      Lists.newArrayList(new RpcServer.BlockingServiceAndInterface(SERVICE, null)),
215      new InetSocketAddress("localhost", 0), CONF, new FifoRpcScheduler(CONF, 1));
216
217    try {
218      rpcServer.start();
219
220      try (AbstractRpcClient<?> client = new NettyRpcClient(clientConf)) {
221        TestRpcServiceProtos.TestProtobufRpcProto.BlockingInterface stub =
222          newBlockingStub(client, rpcServer.getListenerAddress());
223        HBaseRpcController pcrc = new HBaseRpcControllerImpl();
224        String message = "hello";
225        assertEquals(message,
226          stub.echo(pcrc, TestProtos.EchoRequestProto.newBuilder().setMessage(message).build())
227            .getMessage());
228        assertNull(pcrc.cellScanner());
229
230        // truststore file change latch
231        final CountDownLatch latch = new CountDownLatch(1);
232
233        final Path trustStorePath = Paths.get(CONF.get(X509Util.TLS_CONFIG_TRUSTSTORE_LOCATION));
234        createAndStartFileWatcher(trustStorePath, latch, Duration.ofMillis(20));
235
236        // Replace keystore and cancel client connections
237        x509TestContext.regenerateStores(keyType, keyType, storeFileType, storeFileType);
238        client.cancelConnections(
239          ServerName.valueOf(Address.fromSocketAddress(rpcServer.getListenerAddress()), 0L));
240
241        if (!latch.await(1, TimeUnit.SECONDS)) {
242          throw new AssertionError("Timed out waiting for truststore file to be changed");
243        }
244
245        assertEquals(message,
246          stub.echo(pcrc, TestProtos.EchoRequestProto.newBuilder().setMessage(message).build())
247            .getMessage());
248        assertNull(pcrc.cellScanner());
249      }
250    } finally {
251      rpcServer.stop();
252    }
253  }
254
255  private RpcServer createRpcServer(String name,
256    List<RpcServer.BlockingServiceAndInterface> services, InetSocketAddress bindAddress,
257    Configuration conf, RpcScheduler scheduler) throws IOException {
258    return new NettyRpcServer(SERVER, name, services, bindAddress, conf, scheduler, true);
259  }
260
261  private void createAndStartFileWatcher(Path trustStorePath, CountDownLatch latch,
262    Duration duration) throws IOException {
263    FileChangeWatcher fileChangeWatcher = new FileChangeWatcher(trustStorePath,
264      Objects.toString(trustStorePath.getFileName()), duration, watchEventFilePath -> {
265        LOG.info("File " + watchEventFilePath.getFileName() + " has been changed.");
266        latch.countDown();
267      });
268    fileChangeWatcher.start();
269  }
270}