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.apache.hadoop.hbase.security.HBaseKerberosUtils.getKeytabFileForTesting;
023import static org.apache.hadoop.hbase.security.HBaseKerberosUtils.getPrincipalForTesting;
024import static org.apache.hadoop.hbase.security.HBaseKerberosUtils.loginKerberosPrincipal;
025import static org.apache.hadoop.hbase.security.HBaseKerberosUtils.setSecuredConfiguration;
026import static org.apache.hadoop.hbase.security.provider.SaslClientAuthenticationProviders.SELECTOR_KEY;
027import static org.junit.Assert.assertEquals;
028import static org.junit.Assert.assertNotSame;
029import static org.junit.Assert.assertSame;
030import static org.junit.Assert.assertThrows;
031import static org.junit.Assert.fail;
032
033import java.io.File;
034import java.io.IOException;
035import java.lang.reflect.Field;
036import java.net.InetAddress;
037import java.net.InetSocketAddress;
038import java.util.ArrayList;
039import java.util.Collections;
040import java.util.Map;
041import javax.security.sasl.SaslClient;
042import javax.security.sasl.SaslException;
043import org.apache.commons.lang3.RandomStringUtils;
044import org.apache.hadoop.conf.Configuration;
045import org.apache.hadoop.hbase.HBaseTestingUtil;
046import org.apache.hadoop.hbase.HConstants;
047import org.apache.hadoop.hbase.ipc.FifoRpcScheduler;
048import org.apache.hadoop.hbase.ipc.RpcClient;
049import org.apache.hadoop.hbase.ipc.RpcClientFactory;
050import org.apache.hadoop.hbase.ipc.RpcServer;
051import org.apache.hadoop.hbase.ipc.RpcServerFactory;
052import org.apache.hadoop.hbase.ipc.RpcServerInterface;
053import org.apache.hadoop.hbase.security.provider.AuthenticationProviderSelector;
054import org.apache.hadoop.hbase.security.provider.BuiltInProviderSelector;
055import org.apache.hadoop.hbase.security.provider.SaslAuthMethod;
056import org.apache.hadoop.hbase.security.provider.SaslClientAuthenticationProvider;
057import org.apache.hadoop.hbase.util.Pair;
058import org.apache.hadoop.minikdc.MiniKdc;
059import org.apache.hadoop.security.UserGroupInformation;
060import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod;
061import org.apache.hadoop.security.token.Token;
062import org.apache.hadoop.security.token.TokenIdentifier;
063import org.junit.Test;
064import org.mockito.Mockito;
065
066import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
067import org.apache.hbase.thirdparty.com.google.protobuf.BlockingService;
068
069import org.apache.hadoop.hbase.shaded.ipc.protobuf.generated.TestProtos;
070import org.apache.hadoop.hbase.shaded.ipc.protobuf.generated.TestRpcServiceProtos.TestProtobufRpcProto.BlockingInterface;
071import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.UserInformation;
072
073public class AbstractTestSecureIPC {
074
075  protected static final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
076
077  protected static final File KEYTAB_FILE =
078    new File(TEST_UTIL.getDataTestDir("keytab").toUri().getPath());
079
080  protected static MiniKdc KDC;
081  protected static String HOST = "localhost";
082  protected static String PRINCIPAL;
083
084  protected String krbKeytab;
085  protected String krbPrincipal;
086  protected UserGroupInformation ugi;
087  protected Configuration clientConf;
088  protected Configuration serverConf;
089
090  protected static void initKDCAndConf() throws Exception {
091    KDC = TEST_UTIL.setupMiniKdc(KEYTAB_FILE);
092    PRINCIPAL = "hbase/" + HOST;
093    KDC.createPrincipal(KEYTAB_FILE, PRINCIPAL);
094    HBaseKerberosUtils.setPrincipalForTesting(PRINCIPAL + "@" + KDC.getRealm());
095    // set a smaller timeout and retry to speed up tests
096    TEST_UTIL.getConfiguration().setInt(RpcClient.SOCKET_TIMEOUT_READ, 2000);
097    TEST_UTIL.getConfiguration().setInt("hbase.security.relogin.maxretries", 1);
098  }
099
100  protected static void stopKDC() throws InterruptedException {
101    if (KDC != null) {
102      KDC.stop();
103    }
104  }
105
106  protected final void setUpPrincipalAndConf() throws Exception {
107    krbKeytab = getKeytabFileForTesting();
108    krbPrincipal = getPrincipalForTesting();
109    ugi = loginKerberosPrincipal(krbKeytab, krbPrincipal);
110    clientConf = new Configuration(TEST_UTIL.getConfiguration());
111    setSecuredConfiguration(clientConf);
112    serverConf = new Configuration(TEST_UTIL.getConfiguration());
113    setSecuredConfiguration(serverConf);
114  }
115
116  @Test
117  public void testRpcCallWithEnabledKerberosSaslAuth() throws Exception {
118    UserGroupInformation ugi2 = UserGroupInformation.getCurrentUser();
119
120    // check that the login user is okay:
121    assertSame(ugi2, ugi);
122    assertEquals(AuthenticationMethod.KERBEROS, ugi.getAuthenticationMethod());
123    assertEquals(krbPrincipal, ugi.getUserName());
124
125    callRpcService(User.create(ugi2));
126  }
127
128  @Test
129  public void testRpcCallWithEnabledKerberosSaslAuthCanonicalHostname() throws Exception {
130    UserGroupInformation ugi2 = UserGroupInformation.getCurrentUser();
131
132    // check that the login user is okay:
133    assertSame(ugi2, ugi);
134    assertEquals(AuthenticationMethod.KERBEROS, ugi.getAuthenticationMethod());
135    assertEquals(krbPrincipal, ugi.getUserName());
136
137    enableCanonicalHostnameTesting(clientConf, "localhost");
138    clientConf.setBoolean(
139      SecurityConstants.UNSAFE_HBASE_CLIENT_KERBEROS_HOSTNAME_DISABLE_REVERSEDNS, false);
140    clientConf.set(HBaseKerberosUtils.KRB_PRINCIPAL, "hbase/_HOST@" + KDC.getRealm());
141
142    callRpcService(User.create(ugi2));
143  }
144
145  @Test
146  public void testRpcCallWithEnabledKerberosSaslAuthNoCanonicalHostname() throws Exception {
147    UserGroupInformation ugi2 = UserGroupInformation.getCurrentUser();
148
149    // check that the login user is okay:
150    assertSame(ugi2, ugi);
151    assertEquals(AuthenticationMethod.KERBEROS, ugi.getAuthenticationMethod());
152    assertEquals(krbPrincipal, ugi.getUserName());
153
154    enableCanonicalHostnameTesting(clientConf, "127.0.0.1");
155    clientConf
156      .setBoolean(SecurityConstants.UNSAFE_HBASE_CLIENT_KERBEROS_HOSTNAME_DISABLE_REVERSEDNS, true);
157    clientConf.set(HBaseKerberosUtils.KRB_PRINCIPAL, "hbase/_HOST@" + KDC.getRealm());
158
159    callRpcService(User.create(ugi2));
160  }
161
162  private static void enableCanonicalHostnameTesting(Configuration conf, String canonicalHostname) {
163    conf.setClass(SELECTOR_KEY, CanonicalHostnameTestingAuthenticationProviderSelector.class,
164      AuthenticationProviderSelector.class);
165    conf.set(CanonicalHostnameTestingAuthenticationProviderSelector.CANONICAL_HOST_NAME_KEY,
166      canonicalHostname);
167  }
168
169  public static class CanonicalHostnameTestingAuthenticationProviderSelector
170    extends BuiltInProviderSelector {
171    private static final String CANONICAL_HOST_NAME_KEY =
172      "CanonicalHostnameTestingAuthenticationProviderSelector.canonicalHostName";
173
174    @Override
175    public Pair<SaslClientAuthenticationProvider, Token<? extends TokenIdentifier>>
176      selectProvider(String clusterId, User user) {
177      final Pair<SaslClientAuthenticationProvider, Token<? extends TokenIdentifier>> pair =
178        super.selectProvider(clusterId, user);
179      pair.setFirst(createCanonicalHostNameTestingProvider(pair.getFirst()));
180      return pair;
181    }
182
183    SaslClientAuthenticationProvider
184      createCanonicalHostNameTestingProvider(SaslClientAuthenticationProvider delegate) {
185      return new SaslClientAuthenticationProvider() {
186        @Override
187        public SaslClient createClient(Configuration conf, InetAddress serverAddr,
188          SecurityInfo securityInfo, Token<? extends TokenIdentifier> token,
189          boolean fallbackAllowed, Map<String, String> saslProps) throws IOException {
190          final String s = conf.get(CANONICAL_HOST_NAME_KEY);
191          if (s != null) {
192            try {
193              final Field canonicalHostName =
194                InetAddress.class.getDeclaredField("canonicalHostName");
195              canonicalHostName.setAccessible(true);
196              canonicalHostName.set(serverAddr, s);
197            } catch (NoSuchFieldException | IllegalAccessException e) {
198              throw new RuntimeException(e);
199            }
200          }
201
202          return delegate.createClient(conf, serverAddr, securityInfo, token, fallbackAllowed,
203            saslProps);
204        }
205
206        @Override
207        public UserInformation getUserInfo(User user) {
208          return delegate.getUserInfo(user);
209        }
210
211        @Override
212        public UserGroupInformation getRealUser(User ugi) {
213          return delegate.getRealUser(ugi);
214        }
215
216        @Override
217        public boolean canRetry() {
218          return delegate.canRetry();
219        }
220
221        @Override
222        public void relogin() throws IOException {
223          delegate.relogin();
224        }
225
226        @Override
227        public SaslAuthMethod getSaslAuthMethod() {
228          return delegate.getSaslAuthMethod();
229        }
230
231        @Override
232        public String getTokenKind() {
233          return delegate.getTokenKind();
234        }
235      };
236    }
237  }
238
239  @Test
240  public void testRpcFallbackToSimpleAuth() throws Exception {
241    String clientUsername = "testuser";
242    UserGroupInformation clientUgi =
243      UserGroupInformation.createUserForTesting(clientUsername, new String[] { clientUsername });
244
245    // check that the client user is insecure
246    assertNotSame(ugi, clientUgi);
247    assertEquals(AuthenticationMethod.SIMPLE, clientUgi.getAuthenticationMethod());
248    assertEquals(clientUsername, clientUgi.getUserName());
249
250    clientConf.set(User.HBASE_SECURITY_CONF_KEY, "simple");
251    serverConf.setBoolean(RpcServer.FALLBACK_TO_INSECURE_CLIENT_AUTH, true);
252    callRpcService(User.create(clientUgi));
253  }
254
255  private void setRpcProtection(String clientProtection, String serverProtection) {
256    clientConf.set("hbase.rpc.protection", clientProtection);
257    serverConf.set("hbase.rpc.protection", serverProtection);
258  }
259
260  /**
261   * Test various combinations of Server and Client qops.
262   */
263  @Test
264  public void testSaslWithCommonQop() throws Exception {
265    setRpcProtection("privacy,authentication", "authentication");
266    callRpcService(User.create(ugi));
267
268    setRpcProtection("authentication", "privacy,authentication");
269    callRpcService(User.create(ugi));
270
271    setRpcProtection("integrity,authentication", "privacy,authentication");
272    callRpcService(User.create(ugi));
273
274    setRpcProtection("integrity,authentication", "integrity,authentication");
275    callRpcService(User.create(ugi));
276
277    setRpcProtection("privacy,authentication", "privacy,authentication");
278    callRpcService(User.create(ugi));
279  }
280
281  @Test
282  public void testSaslNoCommonQop() throws Exception {
283    setRpcProtection("integrity", "privacy");
284    SaslException se = assertThrows(SaslException.class, () -> callRpcService(User.create(ugi)));
285    assertEquals("No common protection layer between client and server", se.getMessage());
286  }
287
288  /**
289   * Test sasl encryption with Crypto AES.
290   */
291  @Test
292  public void testSaslWithCryptoAES() throws Exception {
293    setRpcProtection("privacy", "privacy");
294    setCryptoAES("true", "true");
295    callRpcService(User.create(ugi));
296  }
297
298  /**
299   * Test various combinations of Server and Client configuration for Crypto AES.
300   */
301  @Test
302  public void testDifferentConfWithCryptoAES() throws Exception {
303    setRpcProtection("privacy", "privacy");
304
305    setCryptoAES("false", "true");
306    callRpcService(User.create(ugi));
307
308    setCryptoAES("true", "false");
309    try {
310      callRpcService(User.create(ugi));
311      fail("The exception should be thrown out for the rpc timeout.");
312    } catch (Exception e) {
313      // ignore the expected exception
314    }
315  }
316
317  private void setCryptoAES(String clientCryptoAES, String serverCryptoAES) {
318    clientConf.set("hbase.rpc.crypto.encryption.aes.enabled", clientCryptoAES);
319    serverConf.set("hbase.rpc.crypto.encryption.aes.enabled", serverCryptoAES);
320  }
321
322  /**
323   * Sets up a RPC Server and a Client. Does a RPC checks the result. If an exception is thrown from
324   * the stub, this function will throw root cause of that exception.
325   */
326  private void callRpcService(User clientUser) throws Exception {
327    SecurityInfo securityInfoMock = Mockito.mock(SecurityInfo.class);
328    Mockito.when(securityInfoMock.getServerPrincipal())
329      .thenReturn(HBaseKerberosUtils.KRB_PRINCIPAL);
330    SecurityInfo.addInfo("TestProtobufRpcProto", securityInfoMock);
331
332    InetSocketAddress isa = new InetSocketAddress(HOST, 0);
333
334    RpcServerInterface rpcServer = RpcServerFactory.createRpcServer(null, "AbstractTestSecureIPC",
335      Lists
336        .newArrayList(new RpcServer.BlockingServiceAndInterface((BlockingService) SERVICE, null)),
337      isa, serverConf, new FifoRpcScheduler(serverConf, 1));
338    rpcServer.start();
339    try (RpcClient rpcClient =
340      RpcClientFactory.createClient(clientConf, HConstants.DEFAULT_CLUSTER_ID.toString())) {
341      BlockingInterface stub =
342        newBlockingStub(rpcClient, rpcServer.getListenerAddress(), clientUser);
343      TestThread th1 = new TestThread(stub);
344      final Throwable exception[] = new Throwable[1];
345      Collections.synchronizedList(new ArrayList<Throwable>());
346      Thread.UncaughtExceptionHandler exceptionHandler = new Thread.UncaughtExceptionHandler() {
347        @Override
348        public void uncaughtException(Thread th, Throwable ex) {
349          exception[0] = ex;
350        }
351      };
352      th1.setUncaughtExceptionHandler(exceptionHandler);
353      th1.start();
354      th1.join();
355      if (exception[0] != null) {
356        // throw root cause.
357        while (exception[0].getCause() != null) {
358          exception[0] = exception[0].getCause();
359        }
360        throw (Exception) exception[0];
361      }
362    } finally {
363      rpcServer.stop();
364    }
365  }
366
367  public static class TestThread extends Thread {
368    private final BlockingInterface stub;
369
370    public TestThread(BlockingInterface stub) {
371      this.stub = stub;
372    }
373
374    @Override
375    public void run() {
376      try {
377        int[] messageSize = new int[] { 100, 1000, 10000 };
378        for (int i = 0; i < messageSize.length; i++) {
379          String input = RandomStringUtils.random(messageSize[i]);
380          String result =
381            stub.echo(null, TestProtos.EchoRequestProto.newBuilder().setMessage(input).build())
382              .getMessage();
383          assertEquals(input, result);
384        }
385      } catch (org.apache.hbase.thirdparty.com.google.protobuf.ServiceException e) {
386        throw new RuntimeException(e);
387      }
388    }
389  }
390}