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