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.security;
19  
20  import io.netty.buffer.ByteBuf;
21  import io.netty.channel.Channel;
22  import io.netty.channel.ChannelDuplexHandler;
23  import io.netty.channel.ChannelFuture;
24  import io.netty.channel.ChannelFutureListener;
25  import io.netty.channel.ChannelHandlerContext;
26  import io.netty.channel.ChannelPromise;
27  
28  import org.apache.commons.logging.Log;
29  import org.apache.commons.logging.LogFactory;
30  import org.apache.hadoop.hbase.classification.InterfaceAudience;
31  import org.apache.hadoop.ipc.RemoteException;
32  import org.apache.hadoop.security.UserGroupInformation;
33  import org.apache.hadoop.security.token.Token;
34  import org.apache.hadoop.security.token.TokenIdentifier;
35  
36  import javax.security.auth.callback.CallbackHandler;
37  import javax.security.sasl.Sasl;
38  import javax.security.sasl.SaslClient;
39  import javax.security.sasl.SaslException;
40  
41  import java.io.IOException;
42  import java.nio.charset.Charset;
43  import java.security.PrivilegedExceptionAction;
44  import java.util.Map;
45  import java.util.Random;
46  
47  /**
48   * Handles Sasl connections
49   */
50  @InterfaceAudience.Private
51  public class SaslClientHandler extends ChannelDuplexHandler {
52    private static final Log LOG = LogFactory.getLog(SaslClientHandler.class);
53  
54    private final boolean fallbackAllowed;
55  
56    private final UserGroupInformation ticket;
57  
58    /**
59     * Used for client or server's token to send or receive from each other.
60     */
61    private final SaslClient saslClient;
62    private final Map<String, String> saslProps;
63    private final SaslExceptionHandler exceptionHandler;
64    private final SaslSuccessfulConnectHandler successfulConnectHandler;
65    private byte[] saslToken;
66    private boolean firstRead = true;
67  
68    private int retryCount = 0;
69    private Random random;
70  
71    /**
72     * @param ticket                   the ugi
73     * @param method                   auth method
74     * @param token                    for Sasl
75     * @param serverPrincipal          Server's Kerberos principal name
76     * @param fallbackAllowed          True if server may also fall back to less secure connection
77     * @param rpcProtection            Quality of protection. Can be 'authentication', 'integrity' or
78     *                                 'privacy'.
79     * @throws java.io.IOException if handler could not be created
80     */
81    public SaslClientHandler(UserGroupInformation ticket, AuthMethod method,
82        Token<? extends TokenIdentifier> token, String serverPrincipal, boolean fallbackAllowed,
83        String rpcProtection, SaslExceptionHandler exceptionHandler,
84        SaslSuccessfulConnectHandler successfulConnectHandler) throws IOException {
85      this.ticket = ticket;
86      this.fallbackAllowed = fallbackAllowed;
87  
88      this.exceptionHandler = exceptionHandler;
89      this.successfulConnectHandler = successfulConnectHandler;
90  
91      saslProps = SaslUtil.initSaslProperties(rpcProtection);
92      switch (method) {
93      case DIGEST:
94        if (LOG.isDebugEnabled())
95          LOG.debug("Creating SASL " + AuthMethod.DIGEST.getMechanismName()
96              + " client to authenticate to service at " + token.getService());
97        saslClient = createDigestSaslClient(new String[] { AuthMethod.DIGEST.getMechanismName() },
98            SaslUtil.SASL_DEFAULT_REALM, new HBaseSaslRpcClient.SaslClientCallbackHandler(token));
99        break;
100     case KERBEROS:
101       if (LOG.isDebugEnabled()) {
102         LOG.debug("Creating SASL " + AuthMethod.KERBEROS.getMechanismName()
103             + " client. Server's Kerberos principal name is " + serverPrincipal);
104       }
105       if (serverPrincipal == null || serverPrincipal.isEmpty()) {
106         throw new IOException("Failed to specify server's Kerberos principal name");
107       }
108       String[] names = SaslUtil.splitKerberosName(serverPrincipal);
109       if (names.length != 3) {
110         throw new IOException(
111             "Kerberos principal does not have the expected format: " + serverPrincipal);
112       }
113       saslClient = createKerberosSaslClient(new String[] { AuthMethod.KERBEROS.getMechanismName() },
114           names[0], names[1]);
115       break;
116     default:
117       throw new IOException("Unknown authentication method " + method);
118     }
119     if (saslClient == null) {
120       throw new IOException("Unable to find SASL client implementation");
121     }
122   }
123 
124   /**
125    * Create a Digest Sasl client
126    */
127   protected SaslClient createDigestSaslClient(String[] mechanismNames, String saslDefaultRealm,
128       CallbackHandler saslClientCallbackHandler) throws IOException {
129     return Sasl.createSaslClient(mechanismNames, null, null, saslDefaultRealm, saslProps,
130         saslClientCallbackHandler);
131   }
132 
133   /**
134    * Create Kerberos client
135    *
136    * @param userFirstPart  first part of username
137    * @param userSecondPart second part of username
138    */
139   protected SaslClient createKerberosSaslClient(String[] mechanismNames, String userFirstPart,
140       String userSecondPart) throws IOException {
141     return Sasl
142         .createSaslClient(mechanismNames, null, userFirstPart, userSecondPart, saslProps,
143             null);
144   }
145 
146   @Override
147   public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
148     saslClient.dispose();
149   }
150 
151   private byte[] evaluateChallenge(final byte[] challenge) throws Exception {
152     return ticket.doAs(new PrivilegedExceptionAction<byte[]>() {
153 
154       @Override
155       public byte[] run() throws Exception {
156         return saslClient.evaluateChallenge(challenge);
157       }
158     });
159   }
160 
161   @Override
162   public void handlerAdded(final ChannelHandlerContext ctx) throws Exception {
163     saslToken = new byte[0];
164     if (saslClient.hasInitialResponse()) {
165       saslToken = evaluateChallenge(saslToken);
166     }
167     if (saslToken != null) {
168       writeSaslToken(ctx, saslToken);
169       if (LOG.isDebugEnabled()) {
170         LOG.debug("Have sent token of size " + saslToken.length + " from initSASLContext.");
171       }
172     }
173   }
174 
175   @Override
176   public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
177     ByteBuf in = (ByteBuf) msg;
178 
179     // If not complete, try to negotiate
180     if (!saslClient.isComplete()) {
181       while (!saslClient.isComplete() && in.isReadable()) {
182         readStatus(in);
183         int len = in.readInt();
184         if (firstRead) {
185           firstRead = false;
186           if (len == SaslUtil.SWITCH_TO_SIMPLE_AUTH) {
187             if (!fallbackAllowed) {
188               throw new IOException("Server asks us to fall back to SIMPLE auth, " + "but this "
189                   + "client is configured to only allow secure connections.");
190             }
191             if (LOG.isDebugEnabled()) {
192               LOG.debug("Server asks us to fall back to simple auth.");
193             }
194             saslClient.dispose();
195 
196             ctx.pipeline().remove(this);
197             successfulConnectHandler.onSuccess(ctx.channel());
198             return;
199           }
200         }
201         saslToken = new byte[len];
202         if (LOG.isDebugEnabled()) {
203           LOG.debug("Will read input token of size " + saslToken.length
204               + " for processing by initSASLContext");
205         }
206         in.readBytes(saslToken);
207 
208         saslToken = evaluateChallenge(saslToken);
209         if (saslToken != null) {
210           if (LOG.isDebugEnabled()) {
211             LOG.debug("Will send token of size " + saslToken.length + " from initSASLContext.");
212           }
213           writeSaslToken(ctx, saslToken);
214         }
215       }
216 
217       if (saslClient.isComplete()) {
218         String qop = (String) saslClient.getNegotiatedProperty(Sasl.QOP);
219 
220         if (LOG.isDebugEnabled()) {
221           LOG.debug("SASL client context established. Negotiated QoP: " + qop);
222         }
223 
224         boolean useWrap = qop != null && !"auth".equalsIgnoreCase(qop);
225 
226         if (!useWrap) {
227           ctx.pipeline().remove(this);
228         }
229         successfulConnectHandler.onSuccess(ctx.channel());
230       }
231     }
232     // Normal wrapped reading
233     else {
234       try {
235         int length = in.readInt();
236         if (LOG.isDebugEnabled()) {
237           LOG.debug("Actual length is " + length);
238         }
239         saslToken = new byte[length];
240         in.readBytes(saslToken);
241       } catch (IndexOutOfBoundsException e) {
242         return;
243       }
244       try {
245         ByteBuf b = ctx.channel().alloc().buffer(saslToken.length);
246 
247         b.writeBytes(saslClient.unwrap(saslToken, 0, saslToken.length));
248         ctx.fireChannelRead(b);
249 
250       } catch (SaslException se) {
251         try {
252           saslClient.dispose();
253         } catch (SaslException ignored) {
254           LOG.debug("Ignoring SASL exception", ignored);
255         }
256         throw se;
257       }
258     }
259   }
260 
261   private void writeSaslToken(final ChannelHandlerContext ctx, byte[] saslToken) {
262     ByteBuf b = ctx.alloc().buffer(4 + saslToken.length);
263     b.writeInt(saslToken.length);
264     b.writeBytes(saslToken, 0, saslToken.length);
265     ctx.writeAndFlush(b).addListener(new ChannelFutureListener() {
266       @Override
267       public void operationComplete(ChannelFuture future) throws Exception {
268         if (!future.isSuccess()) {
269           exceptionCaught(ctx, future.cause());
270         }
271       }
272     });
273   }
274 
275   /**
276    * Get the read status
277    */
278   private static void readStatus(ByteBuf inStream) throws RemoteException {
279     int status = inStream.readInt(); // read status
280     if (status != SaslStatus.SUCCESS.state) {
281       throw new RemoteException(inStream.toString(Charset.forName("UTF-8")),
282           inStream.toString(Charset.forName("UTF-8")));
283     }
284   }
285 
286   @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
287       throws Exception {
288     saslClient.dispose();
289 
290     ctx.close();
291 
292     if (this.random == null) {
293       this.random = new Random();
294     }
295     exceptionHandler.handle(this.retryCount++, this.random, cause);
296   }
297 
298   @Override
299   public void write(final ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
300       throws Exception {
301     // If not complete, try to negotiate
302     if (!saslClient.isComplete()) {
303       super.write(ctx, msg, promise);
304     } else {
305       ByteBuf in = (ByteBuf) msg;
306 
307       try {
308         saslToken = saslClient.wrap(in.array(), in.readerIndex(), in.readableBytes());
309       } catch (SaslException se) {
310         try {
311           saslClient.dispose();
312         } catch (SaslException ignored) {
313           LOG.debug("Ignoring SASL exception", ignored);
314         }
315         promise.setFailure(se);
316       }
317       if (saslToken != null) {
318         ByteBuf out = ctx.channel().alloc().buffer(4 + saslToken.length);
319         out.writeInt(saslToken.length);
320         out.writeBytes(saslToken, 0, saslToken.length);
321 
322         ctx.write(out).addListener(new ChannelFutureListener() {
323           @Override public void operationComplete(ChannelFuture future) throws Exception {
324             if (!future.isSuccess()) {
325               exceptionCaught(ctx, future.cause());
326             }
327           }
328         });
329 
330         saslToken = null;
331       }
332     }
333   }
334 
335   /**
336    * Handler for exceptions during Sasl connection
337    */
338   public interface SaslExceptionHandler {
339     /**
340      * Handle the exception
341      *
342      * @param retryCount current retry count
343      * @param random     to create new backoff with
344      */
345     public void handle(int retryCount, Random random, Throwable cause);
346   }
347 
348   /**
349    * Handler for successful connects
350    */
351   public interface SaslSuccessfulConnectHandler {
352     /**
353      * Runs on success
354      *
355      * @param channel which is successfully authenticated
356      */
357     public void onSuccess(Channel channel);
358   }
359 }