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}