ExecuteFunction66 extends BaseCommand { @Override public void cmdExecute(final Message clientMessage, final ServerConnection serverConnection, final SecurityService securityService, long start) throws IOException { Object function = null; Object args; MemberMappedArgument memberMappedArg = null; String[] groups; byte hasResult = 0; byte functionState; boolean isReexecute = false; boolean allMembers; boolean ignoreFailedMembers; int functionTimeout = DEFAULT_CLIENT_FUNCTION_TIMEOUT; try { byte[] bytes = clientMessage.getPart(0).getSerializedForm(); functionState = bytes[0]; if (bytes.length >= 5 && serverConnection.getClientVersion().ordinal() >= KnownVersion.GFE_8009.ordinal()) { functionTimeout = Part.decodeInt(bytes, 1); } if (functionState == AbstractExecution.HA_HASRESULT_NO_OPTIMIZEFORWRITE_REEXECUTE) { functionState = AbstractExecution.HA_HASRESULT_NO_OPTIMIZEFORWRITE; isReexecute = true; } else if (functionState == AbstractExecution.HA_HASRESULT_OPTIMIZEFORWRITE_REEXECUTE) { functionState = AbstractExecution.HA_HASRESULT_OPTIMIZEFORWRITE; isReexecute = true; } if (functionState != 1) { hasResult = (byte) ((functionState & 2) - 1); } else { hasResult = functionState; } if (hasResult == 1) { serverConnection.setAsTrue(REQUIRES_RESPONSE); serverConnection.setAsTrue(REQUIRES_CHUNKED_RESPONSE); } function = clientMessage.getPart(1).getStringOrObject(); args = clientMessage.getPart(2).getObject(); Part part = clientMessage.getPart(3); if (part != null) { memberMappedArg = (MemberMappedArgument) part.getObject(); } groups = getGroups(clientMessage); allMembers = getAllMembers(clientMessage); ignoreFailedMembers = getIgnoreFailedMembers(clientMessage); } catch (ClassNotFoundException e) { logger.warn("Exception on server while executing function: {}", function, e); if (hasResult == 1) { writeChunkedException(clientMessage, e, serverConnection); } else { writeException(clientMessage, e, false, serverConnection); } serverConnection.setAsTrue(RESPONDED); return; } if (function == null) { String message = "The input function for the execute function request is null"; logger.warn("{} : {}", serverConnection.getName(), message); sendError(hasResult, clientMessage, message, serverConnection); return; } try { Function<?> functionObject; if (function instanceof String) { functionObject = internalFunctionExecutionService.getFunction((String) function); if (functionObject == null) { String message = String .format("Function named %s is not registered to FunctionService", function); logger.warn("{}: {}", serverConnection.getName(), message); sendError(hasResult, clientMessage, message, serverConnection); return; } else { byte functionStateOnServerSide = AbstractExecution.getFunctionState(functionObject.isHA(), functionObject.hasResult(), functionObject.optimizeForWrite()); if (logger.isDebugEnabled()) { logger.debug("Function State on server side: {} on client: {}", functionStateOnServerSide, functionState); } if (functionStateOnServerSide != functionState) { String message = String .format("Function attributes at client and server don't match for %s", function); logger.warn("{}: {}", serverConnection.getName(), message); sendError(hasResult, clientMessage, message, serverConnection); return; } } } else { functionObject = (Function) function; } FunctionStats stats = FunctionStatsManager.getFunctionStats(functionObject.getId()); functionObject.getRequiredPermissions(null, args).forEach(securityService::authorize); AuthorizeRequest authzRequest = serverConnection.getAuthzRequest(); ExecuteFunctionOperationContext executeContext = null; if (authzRequest != null) { executeContext = authzRequest.executeFunctionAuthorize(functionObject.getId(), null, null, args, functionObject.optimizeForWrite()); } ChunkedMessage chunkedMessage = serverConnection.getFunctionResponseMessage(); chunkedMessage.setTransactionId(clientMessage.getTransactionId()); ServerToClientFunctionResultSender resultSender = serverToClientFunctionResultSender65Factory.create(chunkedMessage, MessageType.EXECUTE_FUNCTION_RESULT, serverConnection, functionObject, executeContext); FunctionContext context; InternalCache cache = serverConnection.getCache(); InternalDistributedMember localVM = (InternalDistributedMember) cache.getDistributedSystem().getDistributedMember(); if (memberMappedArg != null) { context = functionContextImplFactory.create(cache, functionObject.getId(), memberMappedArg.getArgumentsForMember(localVM.getId()), resultSender, isReexecute); } else { context = functionContextImplFactory.create(cache, functionObject.getId(), args, resultSender, isReexecute); } ServerSideHandshake handshake = serverConnection.getHandshake(); int earlierClientReadTimeout = handshake.getClientReadTimeout(); handshake.setClientReadTimeout(functionTimeout); try { if (logger.isDebugEnabled()) { logger.debug("Executing Function on Server: {} with context: {}", serverConnection, context); } LowMemoryException lowMemoryException = cache.getInternalResourceManager().getHeapMonitor() .createLowMemoryIfNeeded(functionObject, cache.getMyId()); if (lowMemoryException != null) { sendException(hasResult, clientMessage, lowMemoryException.getMessage(), serverConnection, lowMemoryException); return; } DistributionManager dm = cache.getDistributionManager(); if (groups != null && groups.length > 0) { executeFunctionOnGroups(function, args, groups, allMembers, functionObject, resultSender, ignoreFailedMembers); } else { executeFunctionLocally(functionObject, context, (ServerToClientFunctionResultSender65) resultSender, dm, stats); } if (!functionObject.hasResult()) { writeReply(clientMessage, serverConnection); } } catch (FunctionException e) { throw e; } catch (Exception e) { throw new FunctionException(e); } finally { handshake.setClientReadTimeout(earlierClientReadTimeout); } } catch (IOException e) { logger.warn("Exception on server while executing function: {}}", function, e); String message = "Server could not send the reply"; sendException(hasResult, clientMessage, message, serverConnection, e); } catch (InternalFunctionInvocationTargetException e) { if (logger.isDebugEnabled()) { logger.debug("Exception on server while executing function: {}", function, e); } sendException(hasResult, clientMessage, e.getMessage(), serverConnection, e); } catch (Exception e) { logger.warn("Exception on server while executing function: {}", function, e); sendException(hasResult, clientMessage, e.getMessage(), serverConnection, e); } } ExecuteFunction66();   ExecuteFunction66(InternalFunctionExecutionService internalFunctionExecutionService,
      ServerToClientFunctionResultSender65Factory serverToClientFunctionResultSender65Factory,
      FunctionContextImplFactory functionContextImplFactory); static Command getCommand(); @Override void cmdExecute(final Message clientMessage, final ServerConnection serverConnection,
      final SecurityService securityService, long start);  }
@Test public void nonSecureShouldSucceed() throws Exception { when(oldClientSupportService.getThrowable(any(), any())).thenReturn(mock(Throwable.class)); when(securityService.isClientSecurityRequired()).thenReturn(false); executeFunction.cmdExecute(message, serverConnection, securityService, 0); verify(serverToClientFunctionResultSender65Factory).create(eq(functionResponseMessage), anyInt(), any(), any(), any()); }
@Test public void withIntegratedSecurityShouldSucceedIfAuthorized() throws Exception { when(oldClientSupportService.getThrowable(any(), any())).thenReturn(mock(Throwable.class)); when(securityService.isClientSecurityRequired()).thenReturn(true); when(securityService.isIntegratedSecurity()).thenReturn(true); executeFunction.cmdExecute(message, serverConnection, securityService, 0); verify(securityService).authorize(ResourcePermissions.DATA_WRITE); verify(serverToClientFunctionResultSender65Factory).create(eq(functionResponseMessage), anyInt(), any(), any(), any()); }
@Test public void withIntegratedSecurityShouldThrowIfNotAuthorized() { when(securityService.isClientSecurityRequired()).thenReturn(true); when(securityService.isIntegratedSecurity()).thenReturn(true); doThrow(new NotAuthorizedException("")).when(securityService) .authorize(ResourcePermissions.DATA_WRITE); assertThatThrownBy(() -> executeFunction.cmdExecute(message, serverConnection, securityService, 0)).isExactlyInstanceOf(NullPointerException.class); verify(securityService).authorize(ResourcePermissions.DATA_WRITE); verifyNoMoreInteractions(serverToClientFunctionResultSender65Factory); }
@Test public void withOldSecurityShouldSucceedIfAuthorized() throws Exception { when(oldClientSupportService.getThrowable(any(), any())).thenReturn(mock(Throwable.class)); when(securityService.isClientSecurityRequired()).thenReturn(true); when(securityService.isIntegratedSecurity()).thenReturn(false); executeFunction.cmdExecute(message, serverConnection, securityService, 0); verify(authzRequest).executeFunctionAuthorize(eq(FUNCTION_ID), any(), any(), any(), eq(false)); verify(serverToClientFunctionResultSender65Factory).create(eq(functionResponseMessage), anyInt(), any(), any(), any()); }
@Test public void withOldSecurityShouldThrowIfNotAuthorized() { when(securityService.isClientSecurityRequired()).thenReturn(true); when(securityService.isIntegratedSecurity()).thenReturn(false); doThrow(new NotAuthorizedException("")).when(authzRequest) .executeFunctionAuthorize(eq(FUNCTION_ID), any(), any(), any(), eq(false)); assertThatThrownBy(() -> executeFunction.cmdExecute(message, serverConnection, securityService, 0)).isExactlyInstanceOf(NullPointerException.class); verify(securityService).authorize(ResourcePermissions.DATA_WRITE); verifyNoMoreInteractions(serverToClientFunctionResultSender65Factory); }