RegisterInterest61 extends BaseCommand { @Override public void cmdExecute(Message clientMessage, ServerConnection serverConnection, SecurityService securityService, long start) throws IOException, InterruptedException { Part regionNamePart, keyPart; String regionName; Object key; serverConnection.setAsTrue(REQUIRES_RESPONSE); serverConnection.setAsTrue(REQUIRES_CHUNKED_RESPONSE); regionNamePart = clientMessage.getPart(0); InterestResultPolicy policy; int interestType = clientMessage.getPart(1).getInt(); try { policy = (InterestResultPolicy) clientMessage.getPart(2).getObject(); } catch (Exception e) { writeChunkedException(clientMessage, e, serverConnection); serverConnection.setAsTrue(RESPONDED); return; } boolean isDurable; try { Part durablePart = clientMessage.getPart(3); byte[] durablePartBytes = (byte[]) durablePart.getObject(); isDurable = durablePartBytes[0] == 0x01; } catch (Exception e) { writeChunkedException(clientMessage, e, serverConnection); serverConnection.setAsTrue(RESPONDED); return; } byte[] regionDataPolicyPartBytes; boolean serializeValues = false; try { Part regionDataPolicyPart = clientMessage.getPart(clientMessage.getNumberOfParts() - 1); regionDataPolicyPartBytes = (byte[]) regionDataPolicyPart.getObject(); if (serverConnection.getClientVersion().isNotOlderThan(KnownVersion.GFE_80)) { serializeValues = regionDataPolicyPartBytes[1] == (byte) 0x01; } } catch (Exception e) { writeChunkedException(clientMessage, e, serverConnection); serverConnection.setAsTrue(RESPONDED); return; } keyPart = clientMessage.getPart(4); regionName = regionNamePart.getCachedString(); try { key = keyPart.getStringOrObject(); } catch (Exception e) { writeChunkedException(clientMessage, e, serverConnection); serverConnection.setAsTrue(RESPONDED); return; } boolean sendUpdatesAsInvalidates = false; if (clientMessage.getNumberOfParts() > 5) { try { Part notifyPart = clientMessage.getPart(5); byte[] notifyPartBytes = (byte[]) notifyPart.getObject(); sendUpdatesAsInvalidates = notifyPartBytes[0] == 0x01; } catch (Exception e) { writeChunkedException(clientMessage, e, serverConnection); serverConnection.setAsTrue(RESPONDED); return; } } if (logger.isDebugEnabled()) { logger.debug( "{}: Received register interest 61 request ({} bytes) from {} for region {} key {}", serverConnection.getName(), clientMessage.getPayloadLength(), serverConnection.getSocketString(), regionName, key); } if (VMOTION_DURING_REGISTER_INTEREST_FLAG) { VMotionObserver vmo = VMotionObserverHolder.getInstance(); vmo.vMotionBeforeRegisterInterest(); } if (key == null || regionName == null) { String message = null; if (key == null) { message = "The input key for the register interest request is null"; } if (regionName == null) { message = "The input region name for the register interest request is null."; } logger.warn("{}: {}", serverConnection.getName(), message); writeChunkedErrorResponse(clientMessage, MessageType.REGISTER_INTEREST_DATA_ERROR, message, serverConnection); serverConnection.setAsTrue(RESPONDED); return; } LocalRegion region = (LocalRegion) serverConnection.getCache().getRegion(regionName); if (region == null) { logger.info("{}: Region named {} was not found during register interest request.", new Object[] {serverConnection.getName(), regionName}); } try { if (interestType == InterestType.REGULAR_EXPRESSION) { securityService.authorize(Resource.DATA, Operation.READ, regionName); } else { securityService.authorize(Resource.DATA, Operation.READ, regionName, key); } AuthorizeRequest authorizeRequest = serverConnection.getAuthzRequest(); if (authorizeRequest != null) { if (!DynamicRegionFactory.regionIsDynamicRegionList(regionName)) { RegisterInterestOperationContext registerContext = authorizeRequest.registerInterestAuthorize(regionName, key, interestType, policy); key = registerContext.getKey(); } } serverConnection.getAcceptor().getCacheClientNotifier().registerClientInterest(regionName, key, serverConnection.getProxyID(), interestType, isDurable, sendUpdatesAsInvalidates, true, regionDataPolicyPartBytes[0], true); } catch (Exception e) { checkForInterrupt(serverConnection, e); writeChunkedException(clientMessage, e, serverConnection); serverConnection.setAsTrue(RESPONDED); return; } CacheClientProxy ccp = serverConnection.getAcceptor().getCacheClientNotifier() .getClientProxy(serverConnection.getProxyID()); if (ccp == null) { IOException ioException = new IOException( "CacheClientProxy for this client is no longer on the server"); writeChunkedException(clientMessage, ioException, serverConnection); serverConnection.setAsTrue(RESPONDED); return; } boolean isPrimary = ccp.isPrimary(); ChunkedMessage chunkedResponseMsg = serverConnection.getRegisterInterestResponseMessage(); if (!isPrimary) { chunkedResponseMsg.setMessageType(MessageType.RESPONSE_FROM_SECONDARY); chunkedResponseMsg.setTransactionId(clientMessage.getTransactionId()); chunkedResponseMsg.sendHeader(); chunkedResponseMsg.setLastChunk(true); if (logger.isDebugEnabled()) { logger.debug( "{}: Sending register interest response chunk from secondary for region: {} for key: {} chunk=<{}>", serverConnection.getName(), regionName, key, chunkedResponseMsg); } chunkedResponseMsg.sendChunk(serverConnection); } else { chunkedResponseMsg.setMessageType(MessageType.RESPONSE_FROM_PRIMARY); chunkedResponseMsg.setTransactionId(clientMessage.getTransactionId()); chunkedResponseMsg.sendHeader(); try { if (region.getDistributionManager() instanceof LonerDistributionManager && region instanceof PartitionedRegion) { throw new IllegalStateException( "Should not register interest for a partitioned region when mcast-port is 0 and no locator is present"); } fillAndSendRegisterInterestResponseChunks(region, key, interestType, serializeValues, policy, serverConnection); serverConnection.setAsTrue(RESPONDED); } catch (Exception e) { writeChunkedException(clientMessage, e, serverConnection, chunkedResponseMsg); serverConnection.setAsTrue(RESPONDED); return; } if (logger.isDebugEnabled()) { logger.debug( "{}: Sent register interest response for region {} key {}", serverConnection.getName(), regionName, key); } } } RegisterInterest61(); static Command getCommand(); @Override void cmdExecute(Message clientMessage, ServerConnection serverConnection,
      SecurityService securityService, long start); static final boolean VMOTION_DURING_REGISTER_INTEREST_FLAG; }
@Test public void noSecurityShouldSucceed() throws Exception { when(this.securityService.isClientSecurityRequired()).thenReturn(false); this.registerInterest61.cmdExecute(this.message, this.serverConnection, this.securityService, 0); verify(this.chunkedResponseMessage).sendChunk(this.serverConnection); }
@Test public void integratedSecurityShouldSucceedIfAuthorized() throws Exception { when(this.securityService.isClientSecurityRequired()).thenReturn(true); when(this.securityService.isIntegratedSecurity()).thenReturn(true); this.registerInterest61.cmdExecute(this.message, this.serverConnection, this.securityService, 0); verify(this.securityService).authorize(Resource.DATA, Operation.READ, REGION_NAME, KEY); verify(this.chunkedResponseMessage).sendChunk(this.serverConnection); }
@Test public void integratedSecurityShouldThrowIfNotAuthorized() throws Exception { when(this.securityService.isClientSecurityRequired()).thenReturn(true); when(this.securityService.isIntegratedSecurity()).thenReturn(true); doThrow(new NotAuthorizedException("")).when(this.securityService).authorize(Resource.DATA, Operation.READ, REGION_NAME, KEY); this.registerInterest61.cmdExecute(this.message, this.serverConnection, this.securityService, 0); verify(this.securityService).authorize(Resource.DATA, Operation.READ, REGION_NAME, KEY); verify(this.chunkedResponseMessage).sendChunk(this.serverConnection); }
@Test public void oldSecurityShouldSucceedIfAuthorized() throws Exception { when(this.securityService.isClientSecurityRequired()).thenReturn(true); when(this.securityService.isIntegratedSecurity()).thenReturn(false); this.registerInterest61.cmdExecute(this.message, this.serverConnection, this.securityService, 0); verify(this.authzRequest).registerInterestAuthorize(eq(REGION_NAME), eq(KEY), anyInt(), any()); verify(this.chunkedResponseMessage).sendChunk(this.serverConnection); }
@Test public void oldSecurityShouldFailIfNotAuthorized() throws Exception { when(this.securityService.isClientSecurityRequired()).thenReturn(true); when(this.securityService.isIntegratedSecurity()).thenReturn(false); doThrow(new NotAuthorizedException("")).when(this.authzRequest) .registerInterestAuthorize(eq(REGION_NAME), eq(KEY), anyInt(), any()); this.registerInterest61.cmdExecute(this.message, this.serverConnection, this.securityService, 0); verify(this.authzRequest).registerInterestAuthorize(eq(REGION_NAME), eq(KEY), anyInt(), any()); ArgumentCaptor<NotAuthorizedException> argument = ArgumentCaptor.forClass(NotAuthorizedException.class); verify(this.chunkedResponseMessage).addObjPart(argument.capture()); assertThat(argument.getValue()).isExactlyInstanceOf(NotAuthorizedException.class); verify(this.chunkedResponseMessage).sendChunk(this.serverConnection); }