ExecuteFunction extends BaseCommand { @Override public void cmdExecute(final Message clientMessage, final ServerConnection serverConnection, final SecurityService securityService, long start) throws IOException { Object function = null; Object args = null; MemberMappedArgument memberMappedArg = null; byte hasResult = 0; try { hasResult = clientMessage.getPart(0).getSerializedForm()[0]; 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(); } } catch (ClassNotFoundException e) { logger.warn("Exception on server while executing function: {}", function, e); if (hasResult == 1) { writeChunkedException(clientMessage, e, 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 { 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()); ResultSender resultSender = serverToClientFunctionResultSenderFactory.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); } else { context = functionContextImplFactory.create(cache, functionObject.getId(), args, resultSender); } ServerSideHandshake handshake = serverConnection.getHandshake(); int earlierClientReadTimeout = handshake.getClientReadTimeout(); handshake.setClientReadTimeout(0); long startExecution = stats.startFunctionExecution(functionObject.hasResult()); try { if (logger.isDebugEnabled()) { logger.debug("Executing Function on Server: {} with context: {}", serverConnection, context); } cache.getInternalResourceManager().getHeapMonitor().checkForLowMemory(functionObject, cache.getMyId()); functionObject.execute(context); stats.endFunctionExecution(startExecution, functionObject.hasResult()); } catch (FunctionException e) { stats.endFunctionExecutionWithException(startExecution, functionObject.hasResult()); throw e; } catch (Exception e) { stats.endFunctionExecutionWithException(startExecution, functionObject.hasResult()); 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); } } private  ExecuteFunction(); @VisibleForTesting  ExecuteFunction(InternalFunctionExecutionService internalFunctionExecutionService,
      ServerToClientFunctionResultSenderFactory serverToClientFunctionResultSenderFactory,
      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(securityService.isClientSecurityRequired()).thenReturn(false); executeFunction.cmdExecute(message, serverConnection, securityService, 0); verify(serverToClientFunctionResultSenderFactory).create(eq(functionResponseMessage), anyInt(), any(), any(), any()); }
@Test public void withIntegratedSecurityShouldSucceedIfAuthorized() throws Exception { when(securityService.isClientSecurityRequired()).thenReturn(true); when(securityService.isIntegratedSecurity()).thenReturn(true); executeFunction.cmdExecute(message, serverConnection, securityService, 0); verify(securityService).authorize(ResourcePermissions.DATA_WRITE); verify(serverToClientFunctionResultSenderFactory).create(eq(functionResponseMessage), anyInt(), any(), any(), any()); }
@Test public void withIntegratedSecurityShouldThrowIfNotAuthorized() throws Exception { when(securityService.isClientSecurityRequired()).thenReturn(true); when(securityService.isIntegratedSecurity()).thenReturn(true); doThrow(new NotAuthorizedException("")).when(securityService) .authorize(ResourcePermissions.DATA_WRITE); executeFunction.cmdExecute(message, serverConnection, securityService, 0); verify(securityService).authorize(ResourcePermissions.DATA_WRITE); verify(chunkedResponseMessage).sendChunk(serverConnection); verifyNoMoreInteractions(serverToClientFunctionResultSenderFactory); }
@Test public void withOldSecurityShouldSucceedIfAuthorized() throws Exception { when(securityService.isClientSecurityRequired()).thenReturn(true); when(securityService.isIntegratedSecurity()).thenReturn(false); executeFunction.cmdExecute(message, serverConnection, securityService, 0); verify(authorizeRequest).executeFunctionAuthorize(eq(FUNCTION_ID), any(), any(), any(), eq(false)); verify(serverToClientFunctionResultSenderFactory).create(eq(functionResponseMessage), anyInt(), any(), any(), any()); }
@Test public void withOldSecurityShouldThrowIfNotAuthorized() throws Exception { when(securityService.isClientSecurityRequired()).thenReturn(true); when(securityService.isIntegratedSecurity()).thenReturn(false); doThrow(new NotAuthorizedException("")).when(authorizeRequest) .executeFunctionAuthorize(eq(FUNCTION_ID), any(), any(), any(), eq(false)); executeFunction.cmdExecute(message, serverConnection, securityService, 0); verify(chunkedResponseMessage).sendChunk(serverConnection); verifyNoMoreInteractions(serverToClientFunctionResultSenderFactory); }