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.ipc;
019
020import static org.apache.hadoop.hbase.HConstants.RPC_HEADER;
021
022import io.opentelemetry.api.GlobalOpenTelemetry;
023import io.opentelemetry.api.trace.Span;
024import io.opentelemetry.context.Context;
025import io.opentelemetry.context.Scope;
026import io.opentelemetry.context.propagation.TextMapGetter;
027import java.io.ByteArrayInputStream;
028import java.io.Closeable;
029import java.io.DataOutputStream;
030import java.io.IOException;
031import java.net.InetAddress;
032import java.net.InetSocketAddress;
033import java.nio.ByteBuffer;
034import java.nio.channels.Channels;
035import java.nio.channels.ReadableByteChannel;
036import java.security.GeneralSecurityException;
037import java.util.Objects;
038import java.util.Properties;
039import org.apache.commons.crypto.cipher.CryptoCipherFactory;
040import org.apache.commons.crypto.random.CryptoRandom;
041import org.apache.commons.crypto.random.CryptoRandomFactory;
042import org.apache.hadoop.hbase.CellScanner;
043import org.apache.hadoop.hbase.DoNotRetryIOException;
044import org.apache.hadoop.hbase.client.VersionInfoUtil;
045import org.apache.hadoop.hbase.codec.Codec;
046import org.apache.hadoop.hbase.io.ByteBufferOutputStream;
047import org.apache.hadoop.hbase.io.crypto.aes.CryptoAES;
048import org.apache.hadoop.hbase.ipc.RpcServer.CallCleanup;
049import org.apache.hadoop.hbase.nio.ByteBuff;
050import org.apache.hadoop.hbase.nio.SingleByteBuff;
051import org.apache.hadoop.hbase.security.AccessDeniedException;
052import org.apache.hadoop.hbase.security.HBaseSaslRpcServer;
053import org.apache.hadoop.hbase.security.SaslStatus;
054import org.apache.hadoop.hbase.security.SaslUtil;
055import org.apache.hadoop.hbase.security.User;
056import org.apache.hadoop.hbase.security.provider.SaslServerAuthenticationProvider;
057import org.apache.hadoop.hbase.security.provider.SaslServerAuthenticationProviders;
058import org.apache.hadoop.hbase.security.provider.SimpleSaslServerAuthenticationProvider;
059import org.apache.hadoop.hbase.trace.TraceUtil;
060import org.apache.hadoop.hbase.util.Bytes;
061import org.apache.hadoop.io.BytesWritable;
062import org.apache.hadoop.io.IntWritable;
063import org.apache.hadoop.io.Writable;
064import org.apache.hadoop.io.WritableUtils;
065import org.apache.hadoop.io.compress.CompressionCodec;
066import org.apache.hadoop.security.UserGroupInformation;
067import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod;
068import org.apache.hadoop.security.authorize.AuthorizationException;
069import org.apache.hadoop.security.authorize.ProxyUsers;
070import org.apache.hadoop.security.token.SecretManager.InvalidToken;
071import org.apache.yetus.audience.InterfaceAudience;
072
073import org.apache.hbase.thirdparty.com.google.protobuf.BlockingService;
074import org.apache.hbase.thirdparty.com.google.protobuf.ByteInput;
075import org.apache.hbase.thirdparty.com.google.protobuf.ByteString;
076import org.apache.hbase.thirdparty.com.google.protobuf.CodedInputStream;
077import org.apache.hbase.thirdparty.com.google.protobuf.Descriptors.MethodDescriptor;
078import org.apache.hbase.thirdparty.com.google.protobuf.Message;
079import org.apache.hbase.thirdparty.com.google.protobuf.TextFormat;
080import org.apache.hbase.thirdparty.com.google.protobuf.UnsafeByteOperations;
081
082import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
083import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.VersionInfo;
084import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos;
085import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.ConnectionHeader;
086import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.RequestHeader;
087import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.ResponseHeader;
088import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.UserInformation;
089import org.apache.hadoop.hbase.shaded.protobuf.generated.TracingProtos.RPCTInfo;
090
091/** Reads calls from a connection and queues them for handling. */
092@edu.umd.cs.findbugs.annotations.SuppressWarnings(value = "VO_VOLATILE_INCREMENT",
093    justification = "False positive according to http://sourceforge.net/p/findbugs/bugs/1032/")
094@InterfaceAudience.Private
095abstract class ServerRpcConnection implements Closeable {
096
097  private static final TextMapGetter<RPCTInfo> getter = new RPCTInfoGetter();
098
099  protected final RpcServer rpcServer;
100  // If the connection header has been read or not.
101  protected boolean connectionHeaderRead = false;
102
103  protected CallCleanup callCleanup;
104
105  // Cache the remote host & port info so that even if the socket is
106  // disconnected, we can say where it used to connect to.
107  protected String hostAddress;
108  protected int remotePort;
109  protected InetAddress addr;
110  protected ConnectionHeader connectionHeader;
111
112  /**
113   * Codec the client asked use.
114   */
115  protected Codec codec;
116  /**
117   * Compression codec the client asked us use.
118   */
119  protected CompressionCodec compressionCodec;
120  protected BlockingService service;
121
122  protected SaslServerAuthenticationProvider provider;
123  protected boolean saslContextEstablished;
124  protected boolean skipInitialSaslHandshake;
125  private ByteBuffer unwrappedData;
126  // When is this set? FindBugs wants to know! Says NP
127  private ByteBuffer unwrappedDataLengthBuffer = ByteBuffer.allocate(4);
128  protected boolean useSasl;
129  protected HBaseSaslRpcServer saslServer;
130  protected CryptoAES cryptoAES;
131  protected boolean useWrap = false;
132  protected boolean useCryptoAesWrap = false;
133
134  // was authentication allowed with a fallback to simple auth
135  protected boolean authenticatedWithFallback;
136
137  protected boolean retryImmediatelySupported = false;
138
139  protected User user = null;
140  protected UserGroupInformation ugi = null;
141  protected SaslServerAuthenticationProviders saslProviders = null;
142
143  public ServerRpcConnection(RpcServer rpcServer) {
144    this.rpcServer = rpcServer;
145    this.callCleanup = null;
146    this.saslProviders = SaslServerAuthenticationProviders.getInstance(rpcServer.getConf());
147  }
148
149  @Override
150  public String toString() {
151    return getHostAddress() + ":" + remotePort;
152  }
153
154  public String getHostAddress() {
155    return hostAddress;
156  }
157
158  public InetAddress getHostInetAddress() {
159    return addr;
160  }
161
162  public int getRemotePort() {
163    return remotePort;
164  }
165
166  public VersionInfo getVersionInfo() {
167    if (connectionHeader.hasVersionInfo()) {
168      return connectionHeader.getVersionInfo();
169    }
170    return null;
171  }
172
173  private String getFatalConnectionString(final int version, final byte authByte) {
174    return "serverVersion=" + RpcServer.CURRENT_VERSION + ", clientVersion=" + version
175      + ", authMethod=" + authByte +
176      // The provider may be null if we failed to parse the header of the request
177      ", authName=" + (provider == null ? "unknown" : provider.getSaslAuthMethod().getName())
178      + " from " + toString();
179  }
180
181  /**
182   * Set up cell block codecs n
183   */
184  private void setupCellBlockCodecs(final ConnectionHeader header) throws FatalConnectionException {
185    // TODO: Plug in other supported decoders.
186    if (!header.hasCellBlockCodecClass()) return;
187    String className = header.getCellBlockCodecClass();
188    if (className == null || className.length() == 0) return;
189    try {
190      this.codec = (Codec) Class.forName(className).getDeclaredConstructor().newInstance();
191    } catch (Exception e) {
192      throw new UnsupportedCellCodecException(className, e);
193    }
194    if (!header.hasCellBlockCompressorClass()) return;
195    className = header.getCellBlockCompressorClass();
196    try {
197      this.compressionCodec =
198        (CompressionCodec) Class.forName(className).getDeclaredConstructor().newInstance();
199    } catch (Exception e) {
200      throw new UnsupportedCompressionCodecException(className, e);
201    }
202  }
203
204  /**
205   * Set up cipher for rpc encryption with Apache Commons Crypto n
206   */
207  private void setupCryptoCipher(final ConnectionHeader header,
208    RPCProtos.ConnectionHeaderResponse.Builder chrBuilder) throws FatalConnectionException {
209    // If simple auth, return
210    if (saslServer == null) return;
211    // check if rpc encryption with Crypto AES
212    String qop = saslServer.getNegotiatedQop();
213    boolean isEncryption = SaslUtil.QualityOfProtection.PRIVACY.getSaslQop().equalsIgnoreCase(qop);
214    boolean isCryptoAesEncryption = isEncryption
215      && this.rpcServer.conf.getBoolean("hbase.rpc.crypto.encryption.aes.enabled", false);
216    if (!isCryptoAesEncryption) return;
217    if (!header.hasRpcCryptoCipherTransformation()) return;
218    String transformation = header.getRpcCryptoCipherTransformation();
219    if (transformation == null || transformation.length() == 0) return;
220    // Negotiates AES based on complete saslServer.
221    // The Crypto metadata need to be encrypted and send to client.
222    Properties properties = new Properties();
223    // the property for SecureRandomFactory
224    properties.setProperty(CryptoRandomFactory.CLASSES_KEY,
225      this.rpcServer.conf.get("hbase.crypto.sasl.encryption.aes.crypto.random",
226        "org.apache.commons.crypto.random.JavaCryptoRandom"));
227    // the property for cipher class
228    properties.setProperty(CryptoCipherFactory.CLASSES_KEY,
229      this.rpcServer.conf.get("hbase.rpc.crypto.encryption.aes.cipher.class",
230        "org.apache.commons.crypto.cipher.JceCipher"));
231
232    int cipherKeyBits =
233      this.rpcServer.conf.getInt("hbase.rpc.crypto.encryption.aes.cipher.keySizeBits", 128);
234    // generate key and iv
235    if (cipherKeyBits % 8 != 0) {
236      throw new IllegalArgumentException(
237        "The AES cipher key size in bits" + " should be a multiple of byte");
238    }
239    int len = cipherKeyBits / 8;
240    byte[] inKey = new byte[len];
241    byte[] outKey = new byte[len];
242    byte[] inIv = new byte[len];
243    byte[] outIv = new byte[len];
244
245    try {
246      // generate the cipher meta data with SecureRandom
247      CryptoRandom secureRandom = CryptoRandomFactory.getCryptoRandom(properties);
248      secureRandom.nextBytes(inKey);
249      secureRandom.nextBytes(outKey);
250      secureRandom.nextBytes(inIv);
251      secureRandom.nextBytes(outIv);
252
253      // create CryptoAES for server
254      cryptoAES = new CryptoAES(transformation, properties, inKey, outKey, inIv, outIv);
255      // create SaslCipherMeta and send to client,
256      // for client, the [inKey, outKey], [inIv, outIv] should be reversed
257      RPCProtos.CryptoCipherMeta.Builder ccmBuilder = RPCProtos.CryptoCipherMeta.newBuilder();
258      ccmBuilder.setTransformation(transformation);
259      ccmBuilder.setInIv(getByteString(outIv));
260      ccmBuilder.setInKey(getByteString(outKey));
261      ccmBuilder.setOutIv(getByteString(inIv));
262      ccmBuilder.setOutKey(getByteString(inKey));
263      chrBuilder.setCryptoCipherMeta(ccmBuilder);
264      useCryptoAesWrap = true;
265    } catch (GeneralSecurityException | IOException ex) {
266      throw new UnsupportedCryptoException(ex.getMessage(), ex);
267    }
268  }
269
270  private ByteString getByteString(byte[] bytes) {
271    // return singleton to reduce object allocation
272    return (bytes.length == 0) ? ByteString.EMPTY : ByteString.copyFrom(bytes);
273  }
274
275  private UserGroupInformation createUser(ConnectionHeader head) {
276    UserGroupInformation ugi = null;
277
278    if (!head.hasUserInfo()) {
279      return null;
280    }
281    UserInformation userInfoProto = head.getUserInfo();
282    String effectiveUser = null;
283    if (userInfoProto.hasEffectiveUser()) {
284      effectiveUser = userInfoProto.getEffectiveUser();
285    }
286    String realUser = null;
287    if (userInfoProto.hasRealUser()) {
288      realUser = userInfoProto.getRealUser();
289    }
290    if (effectiveUser != null) {
291      if (realUser != null) {
292        UserGroupInformation realUserUgi = UserGroupInformation.createRemoteUser(realUser);
293        ugi = UserGroupInformation.createProxyUser(effectiveUser, realUserUgi);
294      } else {
295        ugi = UserGroupInformation.createRemoteUser(effectiveUser);
296      }
297    }
298    return ugi;
299  }
300
301  protected final void disposeSasl() {
302    if (saslServer != null) {
303      saslServer.dispose();
304      saslServer = null;
305    }
306  }
307
308  /**
309   * No protobuf encoding of raw sasl messages
310   */
311  protected final void doRawSaslReply(SaslStatus status, Writable rv, String errorClass,
312    String error) throws IOException {
313    BufferChain bc;
314    // In my testing, have noticed that sasl messages are usually
315    // in the ballpark of 100-200. That's why the initial capacity is 256.
316    try (ByteBufferOutputStream saslResponse = new ByteBufferOutputStream(256);
317      DataOutputStream out = new DataOutputStream(saslResponse)) {
318      out.writeInt(status.state); // write status
319      if (status == SaslStatus.SUCCESS) {
320        rv.write(out);
321      } else {
322        WritableUtils.writeString(out, errorClass);
323        WritableUtils.writeString(out, error);
324      }
325      bc = new BufferChain(saslResponse.getByteBuffer());
326    }
327    doRespond(() -> bc);
328  }
329
330  public void saslReadAndProcess(ByteBuff saslToken) throws IOException, InterruptedException {
331    if (saslContextEstablished) {
332      RpcServer.LOG.trace("Read input token of size={} for processing by saslServer.unwrap()",
333        saslToken.limit());
334      if (!useWrap) {
335        processOneRpc(saslToken);
336      } else {
337        byte[] b = saslToken.hasArray() ? saslToken.array() : saslToken.toBytes();
338        byte[] plaintextData;
339        if (useCryptoAesWrap) {
340          // unwrap with CryptoAES
341          plaintextData = cryptoAES.unwrap(b, 0, b.length);
342        } else {
343          plaintextData = saslServer.unwrap(b, 0, b.length);
344        }
345        processUnwrappedData(plaintextData);
346      }
347    } else {
348      byte[] replyToken;
349      try {
350        if (saslServer == null) {
351          try {
352            saslServer =
353              new HBaseSaslRpcServer(provider, rpcServer.saslProps, rpcServer.secretManager);
354          } catch (Exception e) {
355            RpcServer.LOG.error("Error when trying to create instance of HBaseSaslRpcServer "
356              + "with sasl provider: " + provider, e);
357            throw e;
358          }
359          RpcServer.LOG.debug("Created SASL server with mechanism={}",
360            provider.getSaslAuthMethod().getAuthMethod());
361        }
362        RpcServer.LOG.debug(
363          "Read input token of size={} for processing by saslServer." + "evaluateResponse()",
364          saslToken.limit());
365        replyToken = saslServer
366          .evaluateResponse(saslToken.hasArray() ? saslToken.array() : saslToken.toBytes());
367      } catch (IOException e) {
368        RpcServer.LOG.debug("Failed to execute SASL handshake", e);
369        IOException sendToClient = e;
370        Throwable cause = e;
371        while (cause != null) {
372          if (cause instanceof InvalidToken) {
373            sendToClient = (InvalidToken) cause;
374            break;
375          }
376          cause = cause.getCause();
377        }
378        doRawSaslReply(SaslStatus.ERROR, null, sendToClient.getClass().getName(),
379          sendToClient.getLocalizedMessage());
380        this.rpcServer.metrics.authenticationFailure();
381        String clientIP = this.toString();
382        // attempting user could be null
383        RpcServer.AUDITLOG.warn("{}{}: {}", RpcServer.AUTH_FAILED_FOR, clientIP,
384          saslServer.getAttemptingUser());
385        throw e;
386      }
387      if (replyToken != null) {
388        if (RpcServer.LOG.isDebugEnabled()) {
389          RpcServer.LOG.debug("Will send token of size " + replyToken.length + " from saslServer.");
390        }
391        doRawSaslReply(SaslStatus.SUCCESS, new BytesWritable(replyToken), null, null);
392      }
393      if (saslServer.isComplete()) {
394        String qop = saslServer.getNegotiatedQop();
395        useWrap = qop != null && !"auth".equalsIgnoreCase(qop);
396        ugi =
397          provider.getAuthorizedUgi(saslServer.getAuthorizationID(), this.rpcServer.secretManager);
398        RpcServer.LOG.debug(
399          "SASL server context established. Authenticated client: {}. Negotiated QoP is {}", ugi,
400          qop);
401        this.rpcServer.metrics.authenticationSuccess();
402        RpcServer.AUDITLOG.info(RpcServer.AUTH_SUCCESSFUL_FOR + ugi);
403        saslContextEstablished = true;
404      }
405    }
406  }
407
408  private void processUnwrappedData(byte[] inBuf) throws IOException, InterruptedException {
409    ReadableByteChannel ch = Channels.newChannel(new ByteArrayInputStream(inBuf));
410    // Read all RPCs contained in the inBuf, even partial ones
411    while (true) {
412      int count;
413      if (unwrappedDataLengthBuffer.remaining() > 0) {
414        count = this.rpcServer.channelRead(ch, unwrappedDataLengthBuffer);
415        if (count <= 0 || unwrappedDataLengthBuffer.remaining() > 0) return;
416      }
417
418      if (unwrappedData == null) {
419        unwrappedDataLengthBuffer.flip();
420        int unwrappedDataLength = unwrappedDataLengthBuffer.getInt();
421
422        if (unwrappedDataLength == RpcClient.PING_CALL_ID) {
423          if (RpcServer.LOG.isDebugEnabled()) RpcServer.LOG.debug("Received ping message");
424          unwrappedDataLengthBuffer.clear();
425          continue; // ping message
426        }
427        unwrappedData = ByteBuffer.allocate(unwrappedDataLength);
428      }
429
430      count = this.rpcServer.channelRead(ch, unwrappedData);
431      if (count <= 0 || unwrappedData.remaining() > 0) return;
432
433      if (unwrappedData.remaining() == 0) {
434        unwrappedDataLengthBuffer.clear();
435        unwrappedData.flip();
436        processOneRpc(new SingleByteBuff(unwrappedData));
437        unwrappedData = null;
438      }
439    }
440  }
441
442  public void processOneRpc(ByteBuff buf) throws IOException, InterruptedException {
443    if (connectionHeaderRead) {
444      processRequest(buf);
445    } else {
446      processConnectionHeader(buf);
447      this.connectionHeaderRead = true;
448      if (rpcServer.needAuthorization() && !authorizeConnection()) {
449        // Throw FatalConnectionException wrapping ACE so client does right thing and closes
450        // down the connection instead of trying to read non-existent retun.
451        throw new AccessDeniedException("Connection from " + this + " for service "
452          + connectionHeader.getServiceName() + " is unauthorized for user: " + ugi);
453      }
454      this.user = this.rpcServer.userProvider.create(this.ugi);
455    }
456  }
457
458  private boolean authorizeConnection() throws IOException {
459    try {
460      // If auth method is DIGEST, the token was obtained by the
461      // real user for the effective user, therefore not required to
462      // authorize real user. doAs is allowed only for simple or kerberos
463      // authentication
464      if (ugi != null && ugi.getRealUser() != null && provider.supportsProtocolAuthentication()) {
465        ProxyUsers.authorize(ugi, this.getHostAddress(), this.rpcServer.conf);
466      }
467      this.rpcServer.authorize(ugi, connectionHeader, getHostInetAddress());
468      this.rpcServer.metrics.authorizationSuccess();
469    } catch (AuthorizationException ae) {
470      if (RpcServer.LOG.isDebugEnabled()) {
471        RpcServer.LOG.debug("Connection authorization failed: " + ae.getMessage(), ae);
472      }
473      this.rpcServer.metrics.authorizationFailure();
474      doRespond(getErrorResponse(ae.getMessage(), new AccessDeniedException(ae)));
475      return false;
476    }
477    return true;
478  }
479
480  // Reads the connection header following version
481  private void processConnectionHeader(ByteBuff buf) throws IOException {
482    if (buf.hasArray()) {
483      this.connectionHeader = ConnectionHeader.parseFrom(buf.array());
484    } else {
485      CodedInputStream cis = UnsafeByteOperations
486        .unsafeWrap(new ByteBuffByteInput(buf, 0, buf.limit()), 0, buf.limit()).newCodedInput();
487      cis.enableAliasing(true);
488      this.connectionHeader = ConnectionHeader.parseFrom(cis);
489    }
490    String serviceName = connectionHeader.getServiceName();
491    if (serviceName == null) throw new EmptyServiceNameException();
492    this.service = RpcServer.getService(this.rpcServer.services, serviceName);
493    if (this.service == null) throw new UnknownServiceException(serviceName);
494    setupCellBlockCodecs(this.connectionHeader);
495    RPCProtos.ConnectionHeaderResponse.Builder chrBuilder =
496      RPCProtos.ConnectionHeaderResponse.newBuilder();
497    setupCryptoCipher(this.connectionHeader, chrBuilder);
498    responseConnectionHeader(chrBuilder);
499    UserGroupInformation protocolUser = createUser(connectionHeader);
500    if (!useSasl) {
501      ugi = protocolUser;
502      if (ugi != null) {
503        ugi.setAuthenticationMethod(AuthenticationMethod.SIMPLE);
504      }
505      // audit logging for SASL authenticated users happens in saslReadAndProcess()
506      if (authenticatedWithFallback) {
507        RpcServer.LOG.warn("Allowed fallback to SIMPLE auth for {} connecting from {}", ugi,
508          getHostAddress());
509      }
510    } else {
511      // user is authenticated
512      ugi.setAuthenticationMethod(provider.getSaslAuthMethod().getAuthMethod());
513      // Now we check if this is a proxy user case. If the protocol user is
514      // different from the 'user', it is a proxy user scenario. However,
515      // this is not allowed if user authenticated with DIGEST.
516      if ((protocolUser != null) && (!protocolUser.getUserName().equals(ugi.getUserName()))) {
517        if (!provider.supportsProtocolAuthentication()) {
518          // Not allowed to doAs if token authentication is used
519          throw new AccessDeniedException("Authenticated user (" + ugi
520            + ") doesn't match what the client claims to be (" + protocolUser + ")");
521        } else {
522          // Effective user can be different from authenticated user
523          // for simple auth or kerberos auth
524          // The user is the real user. Now we create a proxy user
525          UserGroupInformation realUser = ugi;
526          ugi = UserGroupInformation.createProxyUser(protocolUser.getUserName(), realUser);
527          // Now the user is a proxy user, set Authentication method Proxy.
528          ugi.setAuthenticationMethod(AuthenticationMethod.PROXY);
529        }
530      }
531    }
532    String version;
533    if (this.connectionHeader.hasVersionInfo()) {
534      // see if this connection will support RetryImmediatelyException
535      this.retryImmediatelySupported = VersionInfoUtil.hasMinimumVersion(getVersionInfo(), 1, 2);
536      version = this.connectionHeader.getVersionInfo().getVersion();
537    } else {
538      version = "UNKNOWN";
539    }
540    RpcServer.AUDITLOG.info("Connection from {}:{}, version={}, sasl={}, ugi={}, service={}",
541      this.hostAddress, this.remotePort, version, this.useSasl, this.ugi, serviceName);
542  }
543
544  /**
545   * Send the response for connection header
546   */
547  private void responseConnectionHeader(RPCProtos.ConnectionHeaderResponse.Builder chrBuilder)
548    throws FatalConnectionException {
549    // Response the connection header if Crypto AES is enabled
550    if (!chrBuilder.hasCryptoCipherMeta()) return;
551    try {
552      byte[] connectionHeaderResBytes = chrBuilder.build().toByteArray();
553      // encrypt the Crypto AES cipher meta data with sasl server, and send to client
554      byte[] unwrapped = new byte[connectionHeaderResBytes.length + 4];
555      Bytes.putBytes(unwrapped, 0, Bytes.toBytes(connectionHeaderResBytes.length), 0, 4);
556      Bytes.putBytes(unwrapped, 4, connectionHeaderResBytes, 0, connectionHeaderResBytes.length);
557      byte[] wrapped = saslServer.wrap(unwrapped, 0, unwrapped.length);
558      BufferChain bc;
559      try (ByteBufferOutputStream response = new ByteBufferOutputStream(wrapped.length + 4);
560        DataOutputStream out = new DataOutputStream(response)) {
561        out.writeInt(wrapped.length);
562        out.write(wrapped);
563        bc = new BufferChain(response.getByteBuffer());
564      }
565      doRespond(() -> bc);
566    } catch (IOException ex) {
567      throw new UnsupportedCryptoException(ex.getMessage(), ex);
568    }
569  }
570
571  protected abstract void doRespond(RpcResponse resp) throws IOException;
572
573  /**
574   * n * Has the request header and the request param and optionally encoded data buffer all in this
575   * one array. nn
576   */
577  protected void processRequest(ByteBuff buf) throws IOException, InterruptedException {
578    long totalRequestSize = buf.limit();
579    int offset = 0;
580    // Here we read in the header. We avoid having pb
581    // do its default 4k allocation for CodedInputStream. We force it to use
582    // backing array.
583    CodedInputStream cis;
584    if (buf.hasArray()) {
585      cis = UnsafeByteOperations.unsafeWrap(buf.array(), 0, buf.limit()).newCodedInput();
586    } else {
587      cis = UnsafeByteOperations
588        .unsafeWrap(new ByteBuffByteInput(buf, 0, buf.limit()), 0, buf.limit()).newCodedInput();
589    }
590    cis.enableAliasing(true);
591    int headerSize = cis.readRawVarint32();
592    offset = cis.getTotalBytesRead();
593    Message.Builder builder = RequestHeader.newBuilder();
594    ProtobufUtil.mergeFrom(builder, cis, headerSize);
595    RequestHeader header = (RequestHeader) builder.build();
596    offset += headerSize;
597    Context traceCtx = GlobalOpenTelemetry.getPropagators().getTextMapPropagator()
598      .extract(Context.current(), header.getTraceInfo(), getter);
599
600    // n.b. Management of this Span instance is a little odd. Most exit paths from this try scope
601    // are early-exits due to error cases. There's only one success path, the asynchronous call to
602    // RpcScheduler#dispatch. The success path assumes ownership of the span, which is represented
603    // by null-ing out the reference in this scope. All other paths end the span. Thus, and in
604    // order to avoid accidentally orphaning the span, the call to Span#end happens in a finally
605    // block iff the span is non-null.
606    Span span = TraceUtil.createRemoteSpan("RpcServer.process", traceCtx);
607    try (Scope ignored = span.makeCurrent()) {
608      int id = header.getCallId();
609      if (RpcServer.LOG.isTraceEnabled()) {
610        RpcServer.LOG.trace("RequestHeader " + TextFormat.shortDebugString(header)
611          + " totalRequestSize: " + totalRequestSize + " bytes");
612      }
613      // Enforcing the call queue size, this triggers a retry in the client
614      // This is a bit late to be doing this check - we have already read in the
615      // total request.
616      if (
617        (totalRequestSize + this.rpcServer.callQueueSizeInBytes.sum())
618            > this.rpcServer.maxQueueSizeInBytes
619      ) {
620        final ServerCall<?> callTooBig = createCall(id, this.service, null, null, null, null,
621          totalRequestSize, null, 0, this.callCleanup);
622        this.rpcServer.metrics.exception(RpcServer.CALL_QUEUE_TOO_BIG_EXCEPTION);
623        callTooBig.setResponse(null, null, RpcServer.CALL_QUEUE_TOO_BIG_EXCEPTION,
624          "Call queue is full on " + this.rpcServer.server.getServerName()
625            + ", is hbase.ipc.server.max.callqueue.size too small?");
626        TraceUtil.setError(span, RpcServer.CALL_QUEUE_TOO_BIG_EXCEPTION);
627        callTooBig.sendResponseIfReady();
628        return;
629      }
630      MethodDescriptor md = null;
631      Message param = null;
632      CellScanner cellScanner = null;
633      try {
634        if (header.hasRequestParam() && header.getRequestParam()) {
635          md = this.service.getDescriptorForType().findMethodByName(header.getMethodName());
636          if (md == null) {
637            throw new UnsupportedOperationException(header.getMethodName());
638          }
639          builder = this.service.getRequestPrototype(md).newBuilderForType();
640          cis.resetSizeCounter();
641          int paramSize = cis.readRawVarint32();
642          offset += cis.getTotalBytesRead();
643          if (builder != null) {
644            ProtobufUtil.mergeFrom(builder, cis, paramSize);
645            param = builder.build();
646          }
647          offset += paramSize;
648        } else {
649          // currently header must have request param, so we directly throw
650          // exception here
651          String msg = "Invalid request header: " + TextFormat.shortDebugString(header)
652            + ", should have param set in it";
653          RpcServer.LOG.warn(msg);
654          throw new DoNotRetryIOException(msg);
655        }
656        if (header.hasCellBlockMeta()) {
657          buf.position(offset);
658          ByteBuff dup = buf.duplicate();
659          dup.limit(offset + header.getCellBlockMeta().getLength());
660          cellScanner = this.rpcServer.cellBlockBuilder.createCellScannerReusingBuffers(this.codec,
661            this.compressionCodec, dup);
662        }
663      } catch (Throwable thrown) {
664        InetSocketAddress address = this.rpcServer.getListenerAddress();
665        String msg = (address != null ? address : "(channel closed)")
666          + " is unable to read call parameter from client " + getHostAddress();
667        RpcServer.LOG.warn(msg, thrown);
668
669        this.rpcServer.metrics.exception(thrown);
670
671        final Throwable responseThrowable;
672        if (thrown instanceof LinkageError) {
673          // probably the hbase hadoop version does not match the running hadoop version
674          responseThrowable = new DoNotRetryIOException(thrown);
675        } else if (thrown instanceof UnsupportedOperationException) {
676          // If the method is not present on the server, do not retry.
677          responseThrowable = new DoNotRetryIOException(thrown);
678        } else {
679          responseThrowable = thrown;
680        }
681
682        ServerCall<?> readParamsFailedCall = createCall(id, this.service, null, null, null, null,
683          totalRequestSize, null, 0, this.callCleanup);
684        readParamsFailedCall.setResponse(null, null, responseThrowable,
685          msg + "; " + responseThrowable.getMessage());
686        TraceUtil.setError(span, responseThrowable);
687        readParamsFailedCall.sendResponseIfReady();
688        return;
689      }
690
691      int timeout = 0;
692      if (header.hasTimeout() && header.getTimeout() > 0) {
693        timeout = Math.max(this.rpcServer.minClientRequestTimeout, header.getTimeout());
694      }
695      ServerCall<?> call = createCall(id, this.service, md, header, param, cellScanner,
696        totalRequestSize, this.addr, timeout, this.callCleanup);
697
698      if (this.rpcServer.scheduler.dispatch(new CallRunner(this.rpcServer, call))) {
699        // unset span do that it's not closed in the finally block
700        span = null;
701      } else {
702        this.rpcServer.callQueueSizeInBytes.add(-1 * call.getSize());
703        this.rpcServer.metrics.exception(RpcServer.CALL_QUEUE_TOO_BIG_EXCEPTION);
704        call.setResponse(null, null, RpcServer.CALL_QUEUE_TOO_BIG_EXCEPTION,
705          "Call queue is full on " + this.rpcServer.server.getServerName()
706            + ", too many items queued ?");
707        TraceUtil.setError(span, RpcServer.CALL_QUEUE_TOO_BIG_EXCEPTION);
708        call.sendResponseIfReady();
709      }
710    } finally {
711      if (span != null) {
712        span.end();
713      }
714    }
715  }
716
717  protected final RpcResponse getErrorResponse(String msg, Exception e) throws IOException {
718    ResponseHeader.Builder headerBuilder = ResponseHeader.newBuilder().setCallId(-1);
719    ServerCall.setExceptionResponse(e, msg, headerBuilder);
720    ByteBuffer headerBuf =
721      ServerCall.createHeaderAndMessageBytes(null, headerBuilder.build(), 0, null);
722    BufferChain buf = new BufferChain(headerBuf);
723    return () -> buf;
724  }
725
726  private void doBadPreambleHandling(String msg) throws IOException {
727    doBadPreambleHandling(msg, new FatalConnectionException(msg));
728  }
729
730  private void doBadPreambleHandling(String msg, Exception e) throws IOException {
731    SimpleRpcServer.LOG.warn(msg);
732    doRespond(getErrorResponse(msg, e));
733  }
734
735  protected final boolean processPreamble(ByteBuffer preambleBuffer) throws IOException {
736    assert preambleBuffer.remaining() == 6;
737    for (int i = 0; i < RPC_HEADER.length; i++) {
738      if (RPC_HEADER[i] != preambleBuffer.get()) {
739        doBadPreambleHandling(
740          "Expected HEADER=" + Bytes.toStringBinary(RPC_HEADER) + " but received HEADER="
741            + Bytes.toStringBinary(preambleBuffer.array(), 0, RPC_HEADER.length) + " from "
742            + toString());
743        return false;
744      }
745    }
746    int version = preambleBuffer.get() & 0xFF;
747    byte authbyte = preambleBuffer.get();
748
749    if (version != SimpleRpcServer.CURRENT_VERSION) {
750      String msg = getFatalConnectionString(version, authbyte);
751      doBadPreambleHandling(msg, new WrongVersionException(msg));
752      return false;
753    }
754    this.provider = this.saslProviders.selectProvider(authbyte);
755    if (this.provider == null) {
756      String msg = getFatalConnectionString(version, authbyte);
757      doBadPreambleHandling(msg, new BadAuthException(msg));
758      return false;
759    }
760    // TODO this is a wart while simple auth'n doesn't go through sasl.
761    if (this.rpcServer.isSecurityEnabled && isSimpleAuthentication()) {
762      if (this.rpcServer.allowFallbackToSimpleAuth) {
763        this.rpcServer.metrics.authenticationFallback();
764        authenticatedWithFallback = true;
765      } else {
766        AccessDeniedException ae = new AccessDeniedException("Authentication is required");
767        doRespond(getErrorResponse(ae.getMessage(), ae));
768        return false;
769      }
770    }
771    if (!this.rpcServer.isSecurityEnabled && !isSimpleAuthentication()) {
772      doRawSaslReply(SaslStatus.SUCCESS, new IntWritable(SaslUtil.SWITCH_TO_SIMPLE_AUTH), null,
773        null);
774      provider = saslProviders.getSimpleProvider();
775      // client has already sent the initial Sasl message and we
776      // should ignore it. Both client and server should fall back
777      // to simple auth from now on.
778      skipInitialSaslHandshake = true;
779    }
780    useSasl = !(provider instanceof SimpleSaslServerAuthenticationProvider);
781    return true;
782  }
783
784  boolean isSimpleAuthentication() {
785    return Objects.requireNonNull(provider) instanceof SimpleSaslServerAuthenticationProvider;
786  }
787
788  public abstract boolean isConnectionOpen();
789
790  public abstract ServerCall<?> createCall(int id, BlockingService service, MethodDescriptor md,
791    RequestHeader header, Message param, CellScanner cellScanner, long size,
792    InetAddress remoteAddress, int timeout, CallCleanup reqCleanup);
793
794  private static class ByteBuffByteInput extends ByteInput {
795
796    private ByteBuff buf;
797    private int offset;
798    private int length;
799
800    ByteBuffByteInput(ByteBuff buf, int offset, int length) {
801      this.buf = buf;
802      this.offset = offset;
803      this.length = length;
804    }
805
806    @Override
807    public byte read(int offset) {
808      return this.buf.get(getAbsoluteOffset(offset));
809    }
810
811    private int getAbsoluteOffset(int offset) {
812      return this.offset + offset;
813    }
814
815    @Override
816    public int read(int offset, byte[] out, int outOffset, int len) {
817      this.buf.get(getAbsoluteOffset(offset), out, outOffset, len);
818      return len;
819    }
820
821    @Override
822    public int read(int offset, ByteBuffer out) {
823      int len = out.remaining();
824      this.buf.get(out, getAbsoluteOffset(offset), len);
825      return len;
826    }
827
828    @Override
829    public int size() {
830      return this.length;
831    }
832  }
833}