geode-core/src/distributedTest/java/org/apache/geode/security/AuthExpirationDistributedTest.java - geode - Git at Google

 /*
  *
  * 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.security;

 import static org.apache.geode.cache.query.dunit.SecurityTestUtils.createAndExecuteCQ;
 import static org.apache.geode.distributed.ConfigurationProperties.DURABLE_CLIENT_ID;
 import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_AUTH_INIT;
 import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_LOG_LEVEL;
 import static org.apache.geode.internal.lang.SystemProperty.DEFAULT_PREFIX;
 import static org.apache.geode.internal.lang.SystemPropertyHelper.RE_AUTHENTICATE_WAIT_TIME;
 import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
 import static org.assertj.core.api.Assertions.assertThat;

 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.IntStream;

 import org.junit.After;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.contrib.java.lang.system.RestoreSystemProperties;
 import org.junit.experimental.categories.Category;

 import org.apache.geode.cache.InterestResultPolicy;
 import org.apache.geode.cache.Region;
 import org.apache.geode.cache.RegionShortcut;
 import org.apache.geode.cache.client.ClientCache;
 import org.apache.geode.cache.client.ClientRegionShortcut;
 import org.apache.geode.cache.query.dunit.SecurityTestUtils.EventsCqListner;
 import org.apache.geode.cache.query.dunit.SecurityTestUtils.KeysCacheListener;
 import org.apache.geode.test.dunit.AsyncInvocation;
 import org.apache.geode.test.dunit.rules.ClientVM;
 import org.apache.geode.test.dunit.rules.ClusterStartupRule;
 import org.apache.geode.test.junit.categories.SecurityTest;
 import org.apache.geode.test.junit.rules.ServerStarterRule;

 @Category({SecurityTest.class})
 public class AuthExpirationDistributedTest {
   @Rule
   public ClusterStartupRule cluster = new ClusterStartupRule();

   @Rule
   public RestoreSystemProperties restore = new RestoreSystemProperties();

   @Rule
   public ServerStarterRule server = new ServerStarterRule()
       .withSecurityManager(ExpirableSecurityManager.class)
       .withProperty(SECURITY_LOG_LEVEL, "debug")
       .withSystemProperty(DEFAULT_PREFIX + RE_AUTHENTICATE_WAIT_TIME, "5000")
       .withRegion(RegionShortcut.REPLICATE, "region");

   private ClientVM clientVM;

   @Before
   public void before() throws Exception {
     // this is enabled to show how many times authorize call is made with each permission key
     getSecurityManager().setAllowDuplicate(true);
   }

   @After
   public void after() {
     if (clientVM != null) {
       clientVM.invoke(UpdatableUserAuthInitialize::reset);
     }
     getSecurityManager().close();
   }

   private static EventsCqListner CQLISTENER0;

   @Test
   public void cqClientWillReAuthenticateAutomatically() throws Exception {
     startClientWithCQ();
     Region<Object, Object> region = server.getCache().getRegion("/region");
     region.put("1", "value1");
     clientVM.invoke(() -> {
       await().untilAsserted(
           () -> assertThat(CQLISTENER0.getKeys())
               .asList()
               .containsExactly("1"));
     });

     // expire the current user
     getSecurityManager().addExpiredUser("user1");

     // update the user to be used before we try to send the 2nd event
     clientVM.invoke(() -> {
       UpdatableUserAuthInitialize.setUser("user2");
     });

     // do a second put, the event should be queued until client re-authenticate
     region.put("2", "value2");

     clientVM.invoke(() -> {
       // the client will eventually get the 2nd event
       await().untilAsserted(
           () -> assertThat(CQLISTENER0.getKeys())
               .asList()
               .containsExactly("1", "2"));
     });

     Map<String, List<String>> authorizedOps = getSecurityManager().getAuthorizedOps();
     assertThat(authorizedOps.keySet().size()).isEqualTo(2);
     assertThat(authorizedOps.get("user1")).asList().containsExactly("DATA:READ:region",
         "DATA:READ:region", "DATA:READ:region:1");
     assertThat(authorizedOps.get("user2")).asList().containsExactly("DATA:READ:region:2");

     Map<String, List<String>> unAuthorizedOps = getSecurityManager().getUnAuthorizedOps();
     assertThat(unAuthorizedOps.keySet().size()).isEqualTo(1);
     assertThat(unAuthorizedOps.get("user1")).asList().containsExactly("DATA:READ:region:2");
   }

   @Test
   public void registeredInterestForKeysAndValidatedTheyWereAllReceived_slowReAuth_policyDefault()
       throws Exception {
     int serverPort = server.getPort();
     clientVM = cluster.startClientVM(0,
         c -> c.withProperty(SECURITY_CLIENT_AUTH_INIT, UpdatableUserAuthInitialize.class.getName())
             .withPoolSubscription(true)
             .withProperty(SECURITY_LOG_LEVEL, "debug")
             .withServerConnection(serverPort));

     ClientVM client2 = cluster.startClientVM(1,
         c -> c.withProperty(SECURITY_CLIENT_AUTH_INIT, UpdatableUserAuthInitialize.class.getName())
             .withPoolSubscription(true)
             .withServerConnection(serverPort));

     clientVM.invoke(() -> {
       UpdatableUserAuthInitialize.setUser("user1");
       Region<Object, Object> region = ClusterStartupRule.getClientCache()
           .createClientRegionFactory(ClientRegionShortcut.CACHING_PROXY).create("region");

       // this test will succeed because when clients re-connects, it will re-register interest
       // a new queue will be created with all the data. Old queue is destroyed.
       region.registerInterestForAllKeys();
       UpdatableUserAuthInitialize.setUser("user11");
       // wait for time longer than server's max time to wait to re-authenticate
       UpdatableUserAuthInitialize.setWaitTime(6000);
     });

     AsyncInvocation<Void> invokePut = client2.invokeAsync(() -> {
       UpdatableUserAuthInitialize.setUser("user2");
       Region<Object, Object> region = ClusterStartupRule.getClientCache()
           .createClientRegionFactory(ClientRegionShortcut.CACHING_PROXY).create("region");
       IntStream.range(0, 100).forEach(i -> region.put("key" + i, "value" + i));
     });

     getSecurityManager().addExpiredUser("user1");
     invokePut.await();

     // make sure this client recovers and get all the events and will be able to do client operation
     clientVM.invoke(() -> {
       Region<Object, Object> region = ClusterStartupRule.getClientCache().getRegion("region");
       await().untilAsserted(
           () -> assertThat(region.keySet()).hasSize(100));
       region.put("key100", "value100");
     });

     // user1 should not be used to put any keys to the region
     assertThat(getSecurityManager().getAuthorizedOps().get("user1"))
         .containsExactly("DATA:READ:region");
     assertThat(getSecurityManager().getUnAuthorizedOps().get("user1"))
         .containsExactly("DATA:READ:region:key0");
   }

   @Test
   public void registeredInterestForKeysAndValidatedTheyWereAllReceived_slowReAuth_policyNone_durableClient()
       throws Exception {
     int serverPort = server.getPort();
     clientVM = cluster.startClientVM(0,
         c -> c.withProperty(SECURITY_CLIENT_AUTH_INIT, UpdatableUserAuthInitialize.class.getName())
             .withProperty(DURABLE_CLIENT_ID, "123456")
             .withPoolSubscription(true)
             .withServerConnection(serverPort));

     clientVM.invoke(() -> {
       UpdatableUserAuthInitialize.setUser("user1");
       ClientCache clientCache = ClusterStartupRule.getClientCache();
       Region<Object, Object> region = clientCache
           .createClientRegionFactory(ClientRegionShortcut.CACHING_PROXY).create("region");

       region.registerInterestForAllKeys(InterestResultPolicy.NONE, true);
       clientCache.readyForEvents();
       UpdatableUserAuthInitialize.setUser("user11");
       // wait for time longer than server's max time to wait to re-authenticate
       UpdatableUserAuthInitialize.setWaitTime(6000);
     });

     getSecurityManager().addExpiredUser("user1");
     Region<Object, Object> region = server.getCache().getRegion("/region");
     IntStream.range(0, 100).forEach(i -> region.put("key" + i, "value" + i));

     // make sure this client recovers and get all the events and will be able to do client operation
     clientVM.invoke(() -> {
       Region<Object, Object> clientRegion = ClusterStartupRule.getClientCache().getRegion("region");
       await().untilAsserted(() -> assertThat(clientRegion).hasSize(100));
       clientRegion.put("key100", "value100");
     });

     // user1 should not be used to put any keys to the region
     assertThat(getSecurityManager().getAuthorizedOps().get("user1"))
         .containsExactly("DATA:READ:region");
     assertThat(getSecurityManager().getUnAuthorizedOps().get("user1"))
         .containsExactly("DATA:READ:region:key0");
   }

   private static KeysCacheListener myListener = new KeysCacheListener();

   @Test
   public void registeredInterestForKeysAndValidatedTheyWereAllReceived_slowReAuth_policyNone_CacheListener_durableClient()
       throws Exception {
     int serverPort = server.getPort();
     clientVM = cluster.startClientVM(0,
         c -> c.withProperty(SECURITY_CLIENT_AUTH_INIT, UpdatableUserAuthInitialize.class.getName())
             .withProperty(DURABLE_CLIENT_ID, "123456")
             .withPoolSubscription(true)
             .withServerConnection(serverPort));


     clientVM.invoke(() -> {
       UpdatableUserAuthInitialize.setUser("user1");
       myListener = new KeysCacheListener();
       ClientCache clientCache = ClusterStartupRule.getClientCache();
       Region<Object, Object> region = clientCache
           .createClientRegionFactory(ClientRegionShortcut.PROXY)
           .addCacheListener(myListener).create("region");

       // use NONE policy to make sure the old messages still sticks around
       region.registerInterestForAllKeys(InterestResultPolicy.NONE, true);
       clientCache.readyForEvents();
       UpdatableUserAuthInitialize.setUser("user11");
       // wait for time longer than server's max time to wait to re-authenticate
       UpdatableUserAuthInitialize.setWaitTime(6000);
     });

     getSecurityManager().addExpiredUser("user1");
     Region<Object, Object> region = server.getCache().getRegion("/region");
     IntStream.range(0, 100).forEach(i -> region.put("key" + i, "value" + i));

     // make sure this client recovers and get all the events and will be able to do client operation
     clientVM.invoke(() -> {
       Region<Object, Object> clientRegion = ClusterStartupRule.getClientCache().getRegion("region");
       await().untilAsserted(() -> assertThat(myListener.keys).hasSize(100));
       clientRegion.put("key100", "value100");
     });

     // user1 should not be used to put any keys to the region
     assertThat(getSecurityManager().getAuthorizedOps().get("user1"))
         .containsExactly("DATA:READ:region");
     assertThat(getSecurityManager().getUnAuthorizedOps().get("user1"))
         .containsExactly("DATA:READ:region:key0");
   }


   @Test
   public void registeredInterestForKeysAndValidatedTheyWereAllReceived_slowReAuth_policyNone_nonDurableClient()
       throws Exception {
     int serverPort = server.getPort();
     clientVM = cluster.startClientVM(0,
         c -> c.withProperty(SECURITY_CLIENT_AUTH_INIT, UpdatableUserAuthInitialize.class.getName())
             .withPoolSubscription(true)
             .withServerConnection(serverPort));


     clientVM.invoke(() -> {
       UpdatableUserAuthInitialize.setUser("user1");
       ClientCache clientCache = ClusterStartupRule.getClientCache();
       Region<Object, Object> region = clientCache
           .createClientRegionFactory(ClientRegionShortcut.CACHING_PROXY).create("region");

       // use InterestResultPolicy.NONE to make sure the old queue is still around
       region.registerInterestForAllKeys(InterestResultPolicy.NONE);
       UpdatableUserAuthInitialize.setUser("user11");
       // wait for time longer than server's max time to wait to ree-authenticate
       UpdatableUserAuthInitialize.setWaitTime(6000);
     });

     getSecurityManager().addExpiredUser("user1");
     Region<Object, Object> region = server.getCache().getRegion("/region");
     IntStream.range(0, 100).forEach(i -> region.put("key" + i, "value" + i));

     // client will recover but there will be message loss
     clientVM.invoke(() -> {
       Region<Object, Object> clientRegion = ClusterStartupRule.getClientCache().getRegion("region");
       await().during(10, TimeUnit.SECONDS).untilAsserted(
           () -> assertThat(clientRegion.keySet()).hasSizeLessThan(100));
       clientRegion.put("key100", "value100");
     });

     // user1 should not be used to put any keys to the region
     assertThat(getSecurityManager().getAuthorizedOps().get("user1"))
         .containsExactly("DATA:READ:region");
     assertThat(getSecurityManager().getAuthorizedOps().get("user11"))
         .contains("DATA:WRITE:region:key100");
     assertThat(getSecurityManager().getUnAuthorizedOps().get("user1"))
         .containsExactly("DATA:READ:region:key0");
   }

   private void startClientWithCQ() throws Exception {
     int serverPort = server.getPort();
     clientVM = cluster.startClientVM(0,
         c -> c.withProperty(SECURITY_CLIENT_AUTH_INIT, UpdatableUserAuthInitialize.class.getName())
             .withCacheSetup(
                 ccf -> ccf.setPoolSubscriptionRedundancy(2).setPoolSubscriptionEnabled(true))
             .withServerConnection(serverPort));

     clientVM.invoke(() -> {
       UpdatableUserAuthInitialize.setUser("user1");
       CQLISTENER0 = createAndExecuteCQ(ClusterStartupRule.getClientCache().getQueryService(), "CQ1",
           "select * from /region");
     });
   }

   private ExpirableSecurityManager getSecurityManager() {
     return (ExpirableSecurityManager) server.getCache().getSecurityService().getSecurityManager();
   }


 }
	/*
	*
	* 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.security;

	import static org.apache.geode.cache.query.dunit.SecurityTestUtils.createAndExecuteCQ;
	import static org.apache.geode.distributed.ConfigurationProperties.DURABLE_CLIENT_ID;
	import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_AUTH_INIT;
	import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_LOG_LEVEL;
	import static org.apache.geode.internal.lang.SystemProperty.DEFAULT_PREFIX;
	import static org.apache.geode.internal.lang.SystemPropertyHelper.RE_AUTHENTICATE_WAIT_TIME;
	import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
	import static org.assertj.core.api.Assertions.assertThat;

	import java.util.List;
	import java.util.Map;
	import java.util.concurrent.TimeUnit;
	import java.util.stream.IntStream;

	import org.junit.After;
	import org.junit.Before;
	import org.junit.Rule;
	import org.junit.Test;
	import org.junit.contrib.java.lang.system.RestoreSystemProperties;
	import org.junit.experimental.categories.Category;

	import org.apache.geode.cache.InterestResultPolicy;
	import org.apache.geode.cache.Region;
	import org.apache.geode.cache.RegionShortcut;
	import org.apache.geode.cache.client.ClientCache;
	import org.apache.geode.cache.client.ClientRegionShortcut;
	import org.apache.geode.cache.query.dunit.SecurityTestUtils.EventsCqListner;
	import org.apache.geode.cache.query.dunit.SecurityTestUtils.KeysCacheListener;
	import org.apache.geode.test.dunit.AsyncInvocation;
	import org.apache.geode.test.dunit.rules.ClientVM;
	import org.apache.geode.test.dunit.rules.ClusterStartupRule;
	import org.apache.geode.test.junit.categories.SecurityTest;
	import org.apache.geode.test.junit.rules.ServerStarterRule;

	@Category({SecurityTest.class})
	public class AuthExpirationDistributedTest {
	@Rule
	public ClusterStartupRule cluster = new ClusterStartupRule();

	@Rule
	public RestoreSystemProperties restore = new RestoreSystemProperties();

	@Rule
	public ServerStarterRule server = new ServerStarterRule()
	.withSecurityManager(ExpirableSecurityManager.class)
	.withProperty(SECURITY_LOG_LEVEL, "debug")
	.withSystemProperty(DEFAULT_PREFIX + RE_AUTHENTICATE_WAIT_TIME, "5000")
	.withRegion(RegionShortcut.REPLICATE, "region");

	private ClientVM clientVM;

	@Before
	public void before() throws Exception {
	// this is enabled to show how many times authorize call is made with each permission key
	getSecurityManager().setAllowDuplicate(true);
	}

	@After
	public void after() {
	if (clientVM != null) {
	clientVM.invoke(UpdatableUserAuthInitialize::reset);
	}
	getSecurityManager().close();
	}

	private static EventsCqListner CQLISTENER0;

	@Test
	public void cqClientWillReAuthenticateAutomatically() throws Exception {
	startClientWithCQ();
	Region<Object, Object> region = server.getCache().getRegion("/region");
	region.put("1", "value1");
	clientVM.invoke(() -> {
	await().untilAsserted(
	() -> assertThat(CQLISTENER0.getKeys())
	.asList()
	.containsExactly("1"));
	});

	// expire the current user
	getSecurityManager().addExpiredUser("user1");

	// update the user to be used before we try to send the 2nd event
	clientVM.invoke(() -> {
	UpdatableUserAuthInitialize.setUser("user2");
	});

	// do a second put, the event should be queued until client re-authenticate
	region.put("2", "value2");

	clientVM.invoke(() -> {
	// the client will eventually get the 2nd event
	await().untilAsserted(
	() -> assertThat(CQLISTENER0.getKeys())
	.asList()
	.containsExactly("1", "2"));
	});

	Map<String, List<String>> authorizedOps = getSecurityManager().getAuthorizedOps();
	assertThat(authorizedOps.keySet().size()).isEqualTo(2);
	assertThat(authorizedOps.get("user1")).asList().containsExactly("DATA:READ:region",
	"DATA:READ:region", "DATA:READ:region:1");
	assertThat(authorizedOps.get("user2")).asList().containsExactly("DATA:READ:region:2");

	Map<String, List<String>> unAuthorizedOps = getSecurityManager().getUnAuthorizedOps();
	assertThat(unAuthorizedOps.keySet().size()).isEqualTo(1);
	assertThat(unAuthorizedOps.get("user1")).asList().containsExactly("DATA:READ:region:2");
	}

	@Test
	public void registeredInterestForKeysAndValidatedTheyWereAllReceived_slowReAuth_policyDefault()
	throws Exception {
	int serverPort = server.getPort();
	clientVM = cluster.startClientVM(0,
	c -> c.withProperty(SECURITY_CLIENT_AUTH_INIT, UpdatableUserAuthInitialize.class.getName())
	.withPoolSubscription(true)
	.withProperty(SECURITY_LOG_LEVEL, "debug")
	.withServerConnection(serverPort));

	ClientVM client2 = cluster.startClientVM(1,
	c -> c.withProperty(SECURITY_CLIENT_AUTH_INIT, UpdatableUserAuthInitialize.class.getName())
	.withPoolSubscription(true)
	.withServerConnection(serverPort));

	clientVM.invoke(() -> {
	UpdatableUserAuthInitialize.setUser("user1");
	Region<Object, Object> region = ClusterStartupRule.getClientCache()
	.createClientRegionFactory(ClientRegionShortcut.CACHING_PROXY).create("region");

	// this test will succeed because when clients re-connects, it will re-register interest
	// a new queue will be created with all the data. Old queue is destroyed.
	region.registerInterestForAllKeys();
	UpdatableUserAuthInitialize.setUser("user11");
	// wait for time longer than server's max time to wait to re-authenticate
	UpdatableUserAuthInitialize.setWaitTime(6000);
	});

	AsyncInvocation<Void> invokePut = client2.invokeAsync(() -> {
	UpdatableUserAuthInitialize.setUser("user2");
	Region<Object, Object> region = ClusterStartupRule.getClientCache()
	.createClientRegionFactory(ClientRegionShortcut.CACHING_PROXY).create("region");
	IntStream.range(0, 100).forEach(i -> region.put("key" + i, "value" + i));
	});

	getSecurityManager().addExpiredUser("user1");
	invokePut.await();

	// make sure this client recovers and get all the events and will be able to do client operation
	clientVM.invoke(() -> {
	Region<Object, Object> region = ClusterStartupRule.getClientCache().getRegion("region");
	await().untilAsserted(
	() -> assertThat(region.keySet()).hasSize(100));
	region.put("key100", "value100");
	});

	// user1 should not be used to put any keys to the region
	assertThat(getSecurityManager().getAuthorizedOps().get("user1"))
	.containsExactly("DATA:READ:region");
	assertThat(getSecurityManager().getUnAuthorizedOps().get("user1"))
	.containsExactly("DATA:READ:region:key0");
	}

	@Test
	public void registeredInterestForKeysAndValidatedTheyWereAllReceived_slowReAuth_policyNone_durableClient()
	throws Exception {
	int serverPort = server.getPort();
	clientVM = cluster.startClientVM(0,
	c -> c.withProperty(SECURITY_CLIENT_AUTH_INIT, UpdatableUserAuthInitialize.class.getName())
	.withProperty(DURABLE_CLIENT_ID, "123456")
	.withPoolSubscription(true)
	.withServerConnection(serverPort));

	clientVM.invoke(() -> {
	UpdatableUserAuthInitialize.setUser("user1");
	ClientCache clientCache = ClusterStartupRule.getClientCache();
	Region<Object, Object> region = clientCache
	.createClientRegionFactory(ClientRegionShortcut.CACHING_PROXY).create("region");

	region.registerInterestForAllKeys(InterestResultPolicy.NONE, true);
	clientCache.readyForEvents();
	UpdatableUserAuthInitialize.setUser("user11");
	// wait for time longer than server's max time to wait to re-authenticate
	UpdatableUserAuthInitialize.setWaitTime(6000);
	});

	getSecurityManager().addExpiredUser("user1");
	Region<Object, Object> region = server.getCache().getRegion("/region");
	IntStream.range(0, 100).forEach(i -> region.put("key" + i, "value" + i));

	// make sure this client recovers and get all the events and will be able to do client operation
	clientVM.invoke(() -> {
	Region<Object, Object> clientRegion = ClusterStartupRule.getClientCache().getRegion("region");
	await().untilAsserted(() -> assertThat(clientRegion).hasSize(100));
	clientRegion.put("key100", "value100");
	});

	// user1 should not be used to put any keys to the region
	assertThat(getSecurityManager().getAuthorizedOps().get("user1"))
	.containsExactly("DATA:READ:region");
	assertThat(getSecurityManager().getUnAuthorizedOps().get("user1"))
	.containsExactly("DATA:READ:region:key0");
	}

	private static KeysCacheListener myListener = new KeysCacheListener();

	@Test
	public void registeredInterestForKeysAndValidatedTheyWereAllReceived_slowReAuth_policyNone_CacheListener_durableClient()
	throws Exception {
	int serverPort = server.getPort();
	clientVM = cluster.startClientVM(0,
	c -> c.withProperty(SECURITY_CLIENT_AUTH_INIT, UpdatableUserAuthInitialize.class.getName())
	.withProperty(DURABLE_CLIENT_ID, "123456")
	.withPoolSubscription(true)
	.withServerConnection(serverPort));


	clientVM.invoke(() -> {
	UpdatableUserAuthInitialize.setUser("user1");
	myListener = new KeysCacheListener();
	ClientCache clientCache = ClusterStartupRule.getClientCache();
	Region<Object, Object> region = clientCache
	.createClientRegionFactory(ClientRegionShortcut.PROXY)
	.addCacheListener(myListener).create("region");

	// use NONE policy to make sure the old messages still sticks around
	region.registerInterestForAllKeys(InterestResultPolicy.NONE, true);
	clientCache.readyForEvents();
	UpdatableUserAuthInitialize.setUser("user11");
	// wait for time longer than server's max time to wait to re-authenticate
	UpdatableUserAuthInitialize.setWaitTime(6000);
	});

	getSecurityManager().addExpiredUser("user1");
	Region<Object, Object> region = server.getCache().getRegion("/region");
	IntStream.range(0, 100).forEach(i -> region.put("key" + i, "value" + i));

	// make sure this client recovers and get all the events and will be able to do client operation
	clientVM.invoke(() -> {
	Region<Object, Object> clientRegion = ClusterStartupRule.getClientCache().getRegion("region");
	await().untilAsserted(() -> assertThat(myListener.keys).hasSize(100));
	clientRegion.put("key100", "value100");
	});

	// user1 should not be used to put any keys to the region
	assertThat(getSecurityManager().getAuthorizedOps().get("user1"))
	.containsExactly("DATA:READ:region");
	assertThat(getSecurityManager().getUnAuthorizedOps().get("user1"))
	.containsExactly("DATA:READ:region:key0");
	}


	@Test
	public void registeredInterestForKeysAndValidatedTheyWereAllReceived_slowReAuth_policyNone_nonDurableClient()
	throws Exception {
	int serverPort = server.getPort();
	clientVM = cluster.startClientVM(0,
	c -> c.withProperty(SECURITY_CLIENT_AUTH_INIT, UpdatableUserAuthInitialize.class.getName())
	.withPoolSubscription(true)
	.withServerConnection(serverPort));


	clientVM.invoke(() -> {
	UpdatableUserAuthInitialize.setUser("user1");
	ClientCache clientCache = ClusterStartupRule.getClientCache();
	Region<Object, Object> region = clientCache
	.createClientRegionFactory(ClientRegionShortcut.CACHING_PROXY).create("region");

	// use InterestResultPolicy.NONE to make sure the old queue is still around
	region.registerInterestForAllKeys(InterestResultPolicy.NONE);
	UpdatableUserAuthInitialize.setUser("user11");
	// wait for time longer than server's max time to wait to ree-authenticate
	UpdatableUserAuthInitialize.setWaitTime(6000);
	});

	getSecurityManager().addExpiredUser("user1");
	Region<Object, Object> region = server.getCache().getRegion("/region");
	IntStream.range(0, 100).forEach(i -> region.put("key" + i, "value" + i));

	// client will recover but there will be message loss
	clientVM.invoke(() -> {
	Region<Object, Object> clientRegion = ClusterStartupRule.getClientCache().getRegion("region");
	await().during(10, TimeUnit.SECONDS).untilAsserted(
	() -> assertThat(clientRegion.keySet()).hasSizeLessThan(100));
	clientRegion.put("key100", "value100");
	});

	// user1 should not be used to put any keys to the region
	assertThat(getSecurityManager().getAuthorizedOps().get("user1"))
	.containsExactly("DATA:READ:region");
	assertThat(getSecurityManager().getAuthorizedOps().get("user11"))
	.contains("DATA:WRITE:region:key100");
	assertThat(getSecurityManager().getUnAuthorizedOps().get("user1"))
	.containsExactly("DATA:READ:region:key0");
	}

	private void startClientWithCQ() throws Exception {
	int serverPort = server.getPort();
	clientVM = cluster.startClientVM(0,
	c -> c.withProperty(SECURITY_CLIENT_AUTH_INIT, UpdatableUserAuthInitialize.class.getName())
	.withCacheSetup(
	ccf -> ccf.setPoolSubscriptionRedundancy(2).setPoolSubscriptionEnabled(true))
	.withServerConnection(serverPort));

	clientVM.invoke(() -> {
	UpdatableUserAuthInitialize.setUser("user1");
	CQLISTENER0 = createAndExecuteCQ(ClusterStartupRule.getClientCache().getQueryService(), "CQ1",
	"select * from /region");
	});
	}

	private ExpirableSecurityManager getSecurityManager() {
	return (ExpirableSecurityManager) server.getCache().getSecurityService().getSecurityManager();
	}


	}