| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more contributor license |
| * agreements. See the NOTICE file distributed with this work for additional information regarding |
| * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance with the License. You may obtain a |
| * copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software distributed under the License |
| * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express |
| * or implied. See the License for the specific language governing permissions and limitations under |
| * the License. |
| */ |
| package org.apache.geode.cache.client.internal; |
| |
| import static org.apache.geode.distributed.ConfigurationProperties.CONFLATE_EVENTS; |
| import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_AUTH_INIT; |
| |
| import java.io.DataInputStream; |
| import java.io.DataOutput; |
| import java.io.DataOutputStream; |
| import java.io.EOFException; |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.net.Socket; |
| import java.util.ArrayList; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Properties; |
| |
| import javax.net.ssl.SSLSocket; |
| |
| import org.apache.commons.lang3.StringUtils; |
| |
| import org.apache.geode.CancelCriterion; |
| import org.apache.geode.DataSerializer; |
| import org.apache.geode.GemFireConfigException; |
| import org.apache.geode.InternalGemFireException; |
| import org.apache.geode.annotations.internal.MutableForTesting; |
| import org.apache.geode.cache.GatewayConfigurationException; |
| import org.apache.geode.cache.client.ServerRefusedConnectionException; |
| import org.apache.geode.distributed.DistributedMember; |
| import org.apache.geode.distributed.internal.DistributionConfig; |
| import org.apache.geode.distributed.internal.DistributionManager; |
| import org.apache.geode.distributed.internal.InternalDistributedSystem; |
| import org.apache.geode.distributed.internal.LonerDistributionManager; |
| import org.apache.geode.distributed.internal.ServerLocation; |
| import org.apache.geode.distributed.internal.membership.InternalDistributedMember; |
| import org.apache.geode.internal.HeapDataOutputStream; |
| import org.apache.geode.internal.InternalDataSerializer; |
| import org.apache.geode.internal.InternalInstantiator; |
| import org.apache.geode.internal.cache.tier.ClientSideHandshake; |
| import org.apache.geode.internal.cache.tier.CommunicationMode; |
| import org.apache.geode.internal.cache.tier.Encryptor; |
| import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID; |
| import org.apache.geode.internal.cache.tier.sockets.EncryptorImpl; |
| import org.apache.geode.internal.cache.tier.sockets.Handshake; |
| import org.apache.geode.internal.cache.tier.sockets.ServerQueueStatus; |
| import org.apache.geode.internal.security.SecurityService; |
| import org.apache.geode.internal.serialization.ByteArrayDataInput; |
| import org.apache.geode.internal.serialization.KnownVersion; |
| import org.apache.geode.internal.serialization.StaticSerialization; |
| import org.apache.geode.internal.serialization.VersionedDataInputStream; |
| import org.apache.geode.internal.serialization.VersionedDataOutputStream; |
| import org.apache.geode.internal.serialization.Versioning; |
| import org.apache.geode.internal.serialization.VersioningIO; |
| import org.apache.geode.security.AuthenticationFailedException; |
| import org.apache.geode.security.AuthenticationRequiredException; |
| import org.apache.geode.security.GemFireSecurityException; |
| |
| public class ClientSideHandshakeImpl extends Handshake implements ClientSideHandshake { |
| /** |
| * If true, the client has configured multi-user security, meaning that each thread holds its own |
| * security principal. |
| */ |
| private final boolean multiuserSecureMode; |
| |
| /** |
| * Another test hook, holding a version ordinal that is higher than CURRENT |
| */ |
| @MutableForTesting |
| private static short overrideClientVersion = -1; |
| |
| private final byte replyCode; |
| |
| @Override |
| protected byte getReplyCode() { |
| return replyCode; |
| } |
| |
| public ClientSideHandshakeImpl(ClientProxyMembershipID proxyId, |
| InternalDistributedSystem distributedSystem, SecurityService securityService, |
| boolean multiuserSecureMode) { |
| this.multiuserSecureMode = multiuserSecureMode; |
| id = proxyId; |
| system = distributedSystem; |
| this.securityService = securityService; |
| replyCode = REPLY_OK; |
| setOverrides(); |
| credentials = null; |
| encryptor = new EncryptorImpl(distributedSystem.getSecurityLogWriter()); |
| } |
| |
| /** |
| * Clone a HandShake to be used in creating other connections |
| */ |
| public ClientSideHandshakeImpl(ClientSideHandshakeImpl handshake) { |
| super(handshake); |
| multiuserSecureMode = handshake.multiuserSecureMode; |
| replyCode = handshake.getReplyCode(); |
| } |
| |
| public static void setVersionForTesting(short ver) { |
| if (ver > KnownVersion.CURRENT_ORDINAL) { |
| overrideClientVersion = ver; |
| } else { |
| currentClientVersion = |
| Versioning.getKnownVersionOrDefault(Versioning.getVersion(ver), |
| KnownVersion.CURRENT); |
| overrideClientVersion = -1; |
| } |
| } |
| |
| private void setOverrides() { |
| clientConflation = determineClientConflation(); |
| |
| // As of May 2009 ( GFE 6.0 ): |
| // Note that this.clientVersion is used by server side for accepting |
| // handshakes. |
| // Client side handshake code uses this.currentClientVersion which can be |
| // set via tests. |
| if (currentClientVersion.isNotOlderThan(KnownVersion.GFE_603)) { |
| overrides = new byte[] {clientConflation}; |
| } |
| } |
| |
| // used by the client side |
| private byte determineClientConflation() { |
| byte result = CONFLATION_DEFAULT; |
| |
| String clientConflationValue = system.getProperties().getProperty(CONFLATE_EVENTS); |
| if (DistributionConfig.CLIENT_CONFLATION_PROP_VALUE_ON |
| .equalsIgnoreCase(clientConflationValue)) { |
| result = CONFLATION_ON; |
| } else if (DistributionConfig.CLIENT_CONFLATION_PROP_VALUE_OFF |
| .equalsIgnoreCase(clientConflationValue)) { |
| result = CONFLATION_OFF; |
| } |
| return result; |
| } |
| |
| |
| /** |
| * Return fake, temporary DistributedMember to represent the other vm this vm is connecting to |
| * |
| * @param sock the socket this handshake is operating on |
| * @return temporary id to reprent the other vm |
| */ |
| private InternalDistributedMember getIDForSocket(Socket sock) { |
| return new InternalDistributedMember(sock.getInetAddress(), sock.getPort(), false); |
| } |
| |
| /** |
| * Client-side handshake with a Server |
| */ |
| @Override |
| public ServerQueueStatus handshakeWithServer(Connection conn, ServerLocation location, |
| CommunicationMode communicationMode) throws IOException, AuthenticationRequiredException, |
| AuthenticationFailedException, ServerRefusedConnectionException { |
| try { |
| Socket sock = conn.getSocket(); |
| DataOutputStream dos = new DataOutputStream(sock.getOutputStream()); |
| final InputStream in = sock.getInputStream(); |
| DataInputStream dis = new DataInputStream(in); |
| InternalDistributedMember member = getIDForSocket(sock); |
| // if running in a loner system, use the new port number in the ID to |
| // help differentiate from other clients |
| DistributionManager dm = ((InternalDistributedSystem) system).getDistributionManager(); |
| InternalDistributedMember idm = dm.getDistributionManagerId(); |
| synchronized (idm) { |
| if (idm.getMembershipPort() == 0 && dm instanceof LonerDistributionManager) { |
| int port = sock.getLocalPort(); |
| ((LonerDistributionManager) dm).updateLonerPort(port); |
| id.updateID(dm.getDistributionManagerId()); |
| } |
| } |
| if (communicationMode.isWAN()) { |
| credentials = getCredentials(member); |
| } |
| byte intermediateAcceptanceCode = write(dos, dis, communicationMode, REPLY_OK, |
| clientReadTimeout, null, credentials, member, false); |
| |
| String authInit = system.getProperties().getProperty(SECURITY_CLIENT_AUTH_INIT); |
| if (!communicationMode.isWAN() |
| && intermediateAcceptanceCode != REPLY_AUTH_NOT_REQUIRED |
| && (StringUtils.isNotBlank(authInit) || multiuserSecureMode)) { |
| location.compareAndSetRequiresCredentials(true); |
| } |
| // Read the acceptance code |
| byte acceptanceCode = dis.readByte(); |
| if (acceptanceCode == (byte) 21 && !(sock instanceof SSLSocket)) { |
| // This is likely the case of server setup with SSL and client not using |
| // SSL |
| throw new AuthenticationRequiredException( |
| "Server expecting SSL connection"); |
| } |
| if (acceptanceCode == REPLY_SERVER_IS_LOCATOR) { |
| throw new GemFireConfigException("Improperly configured client detected. " + "Server at " |
| + location + " is actually a locator. Use addPoolLocator to configure locators."); |
| } |
| |
| // Successful handshake for GATEWAY_TO_GATEWAY mode sets the peer version in connection |
| if (communicationMode.isWAN() && !(acceptanceCode == REPLY_EXCEPTION_AUTHENTICATION_REQUIRED |
| || acceptanceCode == REPLY_EXCEPTION_AUTHENTICATION_FAILED)) { |
| short wanSiteVersion = VersioningIO.readOrdinal(dis); |
| conn.setWanSiteVersion(wanSiteVersion); |
| // establish a versioned stream for the other site, if necessary |
| if (wanSiteVersion < KnownVersion.CURRENT_ORDINAL) { |
| dis = new VersionedDataInputStream(dis, Versioning |
| .getKnownVersionOrDefault(Versioning.getVersion(wanSiteVersion), |
| KnownVersion.CURRENT)); |
| } |
| } |
| |
| // No need to check for return value since DataInputStream already throws |
| // EOFException in case of EOF |
| final byte endpointType = dis.readByte(); |
| final int queueSize = dis.readInt(); |
| |
| member = readServerMember(dis); |
| |
| final ServerQueueStatus serverQStatus = |
| new ServerQueueStatus(endpointType, queueSize, member); |
| |
| // Read the message (if any) |
| readMessage(dis, dos, acceptanceCode, member); |
| |
| // Read delta-propagation property value from server. |
| // [sumedh] Static variable below? Client can connect to different |
| // DSes with different values of this. It shoule be a member variable. |
| if (!communicationMode.isWAN() && currentClientVersion.isNotOlderThan(KnownVersion.GFE_61)) { |
| ((InternalDistributedSystem) system).setDeltaEnabledOnServer(dis.readBoolean()); |
| } |
| |
| // validate that the remote side has a different distributed system id. |
| if (communicationMode.isWAN() |
| && KnownVersion.GFE_66 |
| .compareTo(Versioning.getVersion(conn.getWanSiteVersion())) <= 0 |
| && currentClientVersion.isNotOlderThan(KnownVersion.GFE_66)) { |
| int remoteDistributedSystemId = in.read(); |
| int localDistributedSystemId = |
| ((InternalDistributedSystem) system).getDistributionManager().getDistributedSystemId(); |
| if (localDistributedSystemId >= 0 |
| && localDistributedSystemId == remoteDistributedSystemId) { |
| throw new GatewayConfigurationException( |
| "Remote WAN site's distributed system id " + remoteDistributedSystemId |
| + " matches this sites distributed system id " + localDistributedSystemId); |
| } |
| } |
| // Read the PDX registry size from the remote size |
| if (communicationMode.isWAN() |
| && KnownVersion.GFE_80 |
| .compareTo(Versioning.getVersion(conn.getWanSiteVersion())) <= 0 |
| && currentClientVersion.isNotOlderThan(KnownVersion.GFE_80)) { |
| int remotePdxSize = dis.readInt(); |
| serverQStatus.setPdxSize(remotePdxSize); |
| } |
| |
| return serverQStatus; |
| } catch (IOException ex) { |
| CancelCriterion stopper = system.getCancelCriterion(); |
| stopper.checkCancelInProgress(null); |
| throw ex; |
| } |
| } |
| |
| private InternalDistributedMember readServerMember(DataInputStream p_dis) throws IOException { |
| |
| byte[] memberBytes = DataSerializer.readByteArray(p_dis); |
| KnownVersion v = StaticSerialization.getVersionForDataStreamOrNull(p_dis); |
| ByteArrayDataInput dis = new ByteArrayDataInput(memberBytes, v); |
| try { |
| return DataSerializer.readObject(dis); |
| } catch (EOFException e) { |
| throw e; |
| } catch (Exception e) { |
| throw new InternalGemFireException( |
| "Unable to deserialize member", e); |
| } |
| } |
| |
| |
| /** |
| * Used by client-side CacheClientUpdater to handshake with a server in order to receive messages |
| * generated by subscriptions (register-interest, continuous query) |
| */ |
| @Override |
| public ServerQueueStatus handshakeWithSubscriptionFeed(Socket sock, boolean isPrimary) |
| throws IOException, AuthenticationRequiredException, AuthenticationFailedException, |
| ServerRefusedConnectionException, ClassNotFoundException { |
| try { |
| final DataOutputStream dos = new DataOutputStream(sock.getOutputStream()); |
| final InputStream in = sock.getInputStream(); |
| final DataInputStream dis = new DataInputStream(in); |
| final DistributedMember member = getIDForSocket(sock); |
| if (!multiuserSecureMode) { |
| credentials = getCredentials(member); |
| } |
| final CommunicationMode mode = isPrimary ? CommunicationMode.PrimaryServerToClient |
| : CommunicationMode.SecondaryServerToClient; |
| write(dos, dis, mode, REPLY_OK, 0, new ArrayList<>(), credentials, member, true); |
| |
| // Wait here for a reply before continuing. This ensures that the client |
| // updater is registered with the server before continuing. |
| final byte acceptanceCode = dis.readByte(); |
| if (acceptanceCode == (byte) 21 && !(sock instanceof SSLSocket)) { |
| // This is likely the case of server setup with SSL and client not using |
| // SSL |
| throw new AuthenticationRequiredException( |
| "Server expecting SSL connection"); |
| } |
| |
| final byte endpointType = dis.readByte(); |
| final int queueSize = dis.readInt(); |
| |
| // Read the message (if any) |
| readMessage(dis, dos, acceptanceCode, member); |
| |
| // nothing more to be done for older clients used in tests |
| // there is a difference in serializer map registration for >= 6.5.1.6 |
| // clients but that is not used in tests |
| if (currentClientVersion.isOlderThan(KnownVersion.GFE_61)) { |
| return new ServerQueueStatus(endpointType, queueSize, member); |
| } |
| |
| final Map<Integer, List<String>> instantiatorMap = DataSerializer.readHashMap(dis); |
| for (final Map.Entry<Integer, List<String>> entry : instantiatorMap.entrySet()) { |
| final Integer id = entry.getKey(); |
| final List<String> instantiatorArguments = entry.getValue(); |
| InternalInstantiator.register(instantiatorArguments.get(0), instantiatorArguments.get(1), |
| id, false); |
| } |
| |
| final Map<Integer, String> dataSerializersMap = DataSerializer.readHashMap(dis); |
| for (final Map.Entry<Integer, String> entry : dataSerializersMap.entrySet()) { |
| InternalDataSerializer.register(entry.getValue(), false, null, null, entry.getKey()); |
| } |
| final Map<Integer, List<String>> dsToSupportedClassNames = DataSerializer.readHashMap(dis); |
| InternalDataSerializer.updateSupportedClassesMap(dsToSupportedClassNames); |
| |
| // the server's ping interval is only sent to subscription feeds so we can't read it as |
| // part of a "standard" server response along with the other status data. |
| final int pingInterval = dis.readInt(); |
| return new ServerQueueStatus(endpointType, queueSize, member, pingInterval); |
| |
| } catch (IOException | ClassNotFoundException ex) { |
| CancelCriterion stopper = system.getCancelCriterion(); |
| stopper.checkCancelInProgress(null); |
| throw ex; |
| } |
| } |
| |
| /** |
| * client-to-server handshake. Nothing is sent to the server prior to invoking this method. |
| */ |
| private byte write(DataOutputStream dos, DataInputStream dis, CommunicationMode communicationMode, |
| int replyCode, int readTimeout, List<String> ports, Properties p_credentials, |
| DistributedMember member, boolean isCallbackConnection) throws IOException { |
| HeapDataOutputStream hdos = new HeapDataOutputStream(32, KnownVersion.CURRENT); |
| byte acceptanceCode = -1; |
| try { |
| hdos.writeByte(communicationMode.getModeNumber()); |
| if (overrideClientVersion > 0) { |
| // for testing |
| VersioningIO.writeOrdinal(hdos, overrideClientVersion, true); |
| } else { |
| VersioningIO.writeOrdinal(hdos, currentClientVersion.ordinal(), true); |
| } |
| |
| hdos.writeByte(replyCode); |
| if (ports != null) { |
| hdos.writeInt(ports.size()); |
| for (String port : ports) { |
| hdos.writeInt(Integer.parseInt(port)); |
| } |
| } else { |
| hdos.writeInt(readTimeout); |
| } |
| // we do not know the receiver's version at this point, but the on-wire |
| // form of InternalDistributedMember changed in 9.0, so we must serialize |
| // it using the previous version |
| DataOutput idOut = new VersionedDataOutputStream(hdos, KnownVersion.GFE_81); |
| DataSerializer.writeObject(id, idOut); |
| |
| if (currentClientVersion.isNotOlderThan(KnownVersion.GFE_603)) { |
| byte[] overrides = getOverrides(); |
| for (final byte override : overrides) { |
| hdos.writeByte(override); |
| } |
| } else { |
| // write the client conflation setting byte |
| if (setClientConflationForTesting) { |
| hdos.writeByte(clientConflationForTesting); |
| } else { |
| hdos.writeByte(clientConflation); |
| } |
| } |
| |
| if (isCallbackConnection || communicationMode.isWAN()) { |
| if (isCallbackConnection && multiuserSecureMode && !communicationMode.isWAN()) { |
| hdos.writeByte(SECURITY_MULTIUSER_NOTIFICATIONCHANNEL); |
| hdos.flush(); |
| dos.write(hdos.toByteArray()); |
| dos.flush(); |
| } else { |
| writeCredentials(dos, dis, p_credentials, ports != null, member, hdos); |
| } |
| } else { |
| String authInitMethod = system.getProperties().getProperty(SECURITY_CLIENT_AUTH_INIT); |
| acceptanceCode = writeCredential(dos, dis, authInitMethod, ports != null, member, hdos); |
| } |
| } finally { |
| hdos.close(); |
| } |
| return acceptanceCode; |
| } |
| |
| @Override |
| protected byte writeCredential(DataOutputStream dos, DataInputStream dis, String authInit, |
| boolean isNotification, DistributedMember member, HeapDataOutputStream heapdos) |
| throws IOException, GemFireSecurityException { |
| |
| if (!multiuserSecureMode && (authInit == null || authInit.length() == 0)) { |
| // No credentials indicator |
| heapdos.writeByte(CREDENTIALS_NONE); |
| heapdos.flush(); |
| dos.write(heapdos.toByteArray()); |
| dos.flush(); |
| return -1; |
| } |
| |
| return super.writeCredential(dos, dis, authInit, isNotification, member, heapdos); |
| } |
| |
| @Override |
| public Encryptor getEncryptor() { |
| return encryptor; |
| } |
| } |
