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  
19  package org.apache.hadoop.hbase.security;
20  
21  import java.io.BufferedInputStream;
22  import java.io.BufferedOutputStream;
23  import java.io.DataInputStream;
24  import java.io.DataOutputStream;
25  import java.io.IOException;
26  import java.io.InputStream;
27  import java.io.OutputStream;
28  
29  import javax.security.auth.callback.Callback;
30  import javax.security.auth.callback.CallbackHandler;
31  import javax.security.auth.callback.NameCallback;
32  import javax.security.auth.callback.PasswordCallback;
33  import javax.security.auth.callback.UnsupportedCallbackException;
34  import javax.security.sasl.RealmCallback;
35  import javax.security.sasl.RealmChoiceCallback;
36  import javax.security.sasl.Sasl;
37  import javax.security.sasl.SaslClient;
38  import javax.security.sasl.SaslException;
39  
40  import org.apache.commons.logging.Log;
41  import org.apache.commons.logging.LogFactory;
42  import org.apache.hadoop.hbase.classification.InterfaceAudience;
43  import org.apache.hadoop.io.WritableUtils;
44  import org.apache.hadoop.ipc.RemoteException;
45  import org.apache.hadoop.security.SaslInputStream;
46  import org.apache.hadoop.security.SaslOutputStream;
47  import org.apache.hadoop.security.token.Token;
48  import org.apache.hadoop.security.token.TokenIdentifier;
49  
50  import com.google.common.annotations.VisibleForTesting;
51  
52  /**
53   * A utility class that encapsulates SASL logic for RPC client.
54   * Copied from <code>org.apache.hadoop.security</code>
55   */
56  @InterfaceAudience.Private
57  public class HBaseSaslRpcClient {
58    public static final Log LOG = LogFactory.getLog(HBaseSaslRpcClient.class);
59  
60    private final SaslClient saslClient;
61    private final boolean fallbackAllowed;
62    /**
63     * Create a HBaseSaslRpcClient for an authentication method
64     * 
65     * @param method
66     *          the requested authentication method
67     * @param token
68     *          token to use if needed by the authentication method
69     * @param serverPrincipal
70     *          the server principal that we are trying to set the connection up to
71     * @param fallbackAllowed
72     *          does the client allow fallback to simple authentication
73     * @throws IOException
74     */
75    public HBaseSaslRpcClient(AuthMethod method,
76        Token<? extends TokenIdentifier> token, String serverPrincipal, boolean fallbackAllowed)
77        throws IOException {
78      this(method, token, serverPrincipal, fallbackAllowed, "authentication"); 
79    }
80    /**
81     * Create a HBaseSaslRpcClient for an authentication method
82     * 
83     * @param method
84     *          the requested authentication method
85     * @param token
86     *          token to use if needed by the authentication method
87     * @param serverPrincipal
88     *          the server principal that we are trying to set the connection up to
89     * @param fallbackAllowed
90     *          does the client allow fallback to simple authentication
91     * @param rpcProtection
92     *          the protection level ("authentication", "integrity" or "privacy")
93     * @throws IOException
94     */
95    public HBaseSaslRpcClient(AuthMethod method,
96        Token<? extends TokenIdentifier> token, String serverPrincipal, boolean fallbackAllowed,
97        String rpcProtection) throws IOException {
98      this.fallbackAllowed = fallbackAllowed;
99      SaslUtil.initSaslProperties(rpcProtection);
100     switch (method) {
101     case DIGEST:
102       if (LOG.isDebugEnabled())
103         LOG.debug("Creating SASL " + AuthMethod.DIGEST.getMechanismName()
104             + " client to authenticate to service at " + token.getService());
105       saslClient = createDigestSaslClient(
106           new String[] { AuthMethod.DIGEST.getMechanismName() },
107           SaslUtil.SASL_DEFAULT_REALM, new SaslClientCallbackHandler(token));
108       break;
109     case KERBEROS:
110       if (LOG.isDebugEnabled()) {
111         LOG
112             .debug("Creating SASL " + AuthMethod.KERBEROS.getMechanismName()
113                 + " client. Server's Kerberos principal name is "
114                 + serverPrincipal);
115       }
116       if (serverPrincipal == null || serverPrincipal.length() == 0) {
117         throw new IOException(
118             "Failed to specify server's Kerberos principal name");
119       }
120       String names[] = SaslUtil.splitKerberosName(serverPrincipal);
121       if (names.length != 3) {
122         throw new IOException(
123           "Kerberos principal does not have the expected format: "
124                 + serverPrincipal);
125       }
126       saslClient = createKerberosSaslClient(
127           new String[] { AuthMethod.KERBEROS.getMechanismName() },
128           names[0], names[1]);
129       break;
130     default:
131       throw new IOException("Unknown authentication method " + method);
132     }
133     if (saslClient == null)
134       throw new IOException("Unable to find SASL client implementation");
135   }
136 
137   protected SaslClient createDigestSaslClient(String[] mechanismNames, 
138       String saslDefaultRealm, CallbackHandler saslClientCallbackHandler) 
139       throws IOException {
140     return Sasl.createSaslClient(mechanismNames, null, null, saslDefaultRealm,
141         SaslUtil.SASL_PROPS, saslClientCallbackHandler);
142   }
143 
144   protected SaslClient createKerberosSaslClient(String[] mechanismNames,
145       String userFirstPart, String userSecondPart) throws IOException {
146     return Sasl.createSaslClient(mechanismNames, null, userFirstPart, 
147         userSecondPart, SaslUtil.SASL_PROPS, null);
148   }
149 
150   private static void readStatus(DataInputStream inStream) throws IOException {
151     int status = inStream.readInt(); // read status
152     if (status != SaslStatus.SUCCESS.state) {
153       throw new RemoteException(WritableUtils.readString(inStream),
154           WritableUtils.readString(inStream));
155     }
156   }
157   
158   /**
159    * Do client side SASL authentication with server via the given InputStream
160    * and OutputStream
161    * 
162    * @param inS
163    *          InputStream to use
164    * @param outS
165    *          OutputStream to use
166    * @return true if connection is set up, or false if needs to switch 
167    *             to simple Auth.
168    * @throws IOException
169    */
170   public boolean saslConnect(InputStream inS, OutputStream outS)
171       throws IOException {
172     DataInputStream inStream = new DataInputStream(new BufferedInputStream(inS));
173     DataOutputStream outStream = new DataOutputStream(new BufferedOutputStream(
174         outS));
175 
176     try {
177       byte[] saslToken = new byte[0];
178       if (saslClient.hasInitialResponse())
179         saslToken = saslClient.evaluateChallenge(saslToken);
180       if (saslToken != null) {
181         outStream.writeInt(saslToken.length);
182         outStream.write(saslToken, 0, saslToken.length);
183         outStream.flush();
184         if (LOG.isDebugEnabled())
185           LOG.debug("Have sent token of size " + saslToken.length
186               + " from initSASLContext.");
187       }
188       if (!saslClient.isComplete()) {
189         readStatus(inStream);
190         int len = inStream.readInt();
191         if (len == SaslUtil.SWITCH_TO_SIMPLE_AUTH) {
192           if (!fallbackAllowed) {
193             throw new IOException("Server asks us to fall back to SIMPLE auth, " +
194                 "but this client is configured to only allow secure connections.");
195           }
196           if (LOG.isDebugEnabled()) {
197             LOG.debug("Server asks us to fall back to simple auth.");
198           }
199           saslClient.dispose();
200           return false;
201         }
202         saslToken = new byte[len];
203         if (LOG.isDebugEnabled())
204           LOG.debug("Will read input token of size " + saslToken.length
205               + " for processing by initSASLContext");
206         inStream.readFully(saslToken);
207       }
208 
209       while (!saslClient.isComplete()) {
210         saslToken = saslClient.evaluateChallenge(saslToken);
211         if (saslToken != null) {
212           if (LOG.isDebugEnabled())
213             LOG.debug("Will send token of size " + saslToken.length
214                 + " from initSASLContext.");
215           outStream.writeInt(saslToken.length);
216           outStream.write(saslToken, 0, saslToken.length);
217           outStream.flush();
218         }
219         if (!saslClient.isComplete()) {
220           readStatus(inStream);
221           saslToken = new byte[inStream.readInt()];
222           if (LOG.isDebugEnabled())
223             LOG.debug("Will read input token of size " + saslToken.length
224                 + " for processing by initSASLContext");
225           inStream.readFully(saslToken);
226         }
227       }
228       if (LOG.isDebugEnabled()) {
229         LOG.debug("SASL client context established. Negotiated QoP: "
230             + saslClient.getNegotiatedProperty(Sasl.QOP));
231       }
232       return true;
233     } catch (IOException e) {
234       try {
235         saslClient.dispose();
236       } catch (SaslException ignored) {
237         // ignore further exceptions during cleanup
238       }
239       throw e;
240     }
241   }
242 
243   /**
244    * Get a SASL wrapped InputStream. Can be called only after saslConnect() has
245    * been called.
246    * 
247    * @param in
248    *          the InputStream to wrap
249    * @return a SASL wrapped InputStream
250    * @throws IOException
251    */
252   public InputStream getInputStream(InputStream in) throws IOException {
253     if (!saslClient.isComplete()) {
254       throw new IOException("Sasl authentication exchange hasn't completed yet");
255     }
256     return new SaslInputStream(in, saslClient);
257   }
258 
259   /**
260    * Get a SASL wrapped OutputStream. Can be called only after saslConnect() has
261    * been called.
262    * 
263    * @param out
264    *          the OutputStream to wrap
265    * @return a SASL wrapped OutputStream
266    * @throws IOException
267    */
268   public OutputStream getOutputStream(OutputStream out) throws IOException {
269     if (!saslClient.isComplete()) {
270       throw new IOException("Sasl authentication exchange hasn't completed yet");
271     }
272     return new SaslOutputStream(out, saslClient);
273   }
274 
275   /** Release resources used by wrapped saslClient */
276   public void dispose() throws SaslException {
277     saslClient.dispose();
278   }
279 
280   @VisibleForTesting
281   static class SaslClientCallbackHandler implements CallbackHandler {
282     private final String userName;
283     private final char[] userPassword;
284 
285     public SaslClientCallbackHandler(Token<? extends TokenIdentifier> token) {
286       this.userName = SaslUtil.encodeIdentifier(token.getIdentifier());
287       this.userPassword = SaslUtil.encodePassword(token.getPassword());
288     }
289 
290     public void handle(Callback[] callbacks)
291         throws UnsupportedCallbackException {
292       NameCallback nc = null;
293       PasswordCallback pc = null;
294       RealmCallback rc = null;
295       for (Callback callback : callbacks) {
296         if (callback instanceof RealmChoiceCallback) {
297           continue;
298         } else if (callback instanceof NameCallback) {
299           nc = (NameCallback) callback;
300         } else if (callback instanceof PasswordCallback) {
301           pc = (PasswordCallback) callback;
302         } else if (callback instanceof RealmCallback) {
303           rc = (RealmCallback) callback;
304         } else {
305           throw new UnsupportedCallbackException(callback,
306               "Unrecognized SASL client callback");
307         }
308       }
309       if (nc != null) {
310         if (LOG.isDebugEnabled())
311           LOG.debug("SASL client callback: setting username: " + userName);
312         nc.setName(userName);
313       }
314       if (pc != null) {
315         if (LOG.isDebugEnabled())
316           LOG.debug("SASL client callback: setting userPassword");
317         pc.setPassword(userPassword);
318       }
319       if (rc != null) {
320         if (LOG.isDebugEnabled())
321           LOG.debug("SASL client callback: setting realm: "
322               + rc.getDefaultText());
323         rc.setText(rc.getDefaultText());
324       }
325     }
326   }
327 }