[controller] Manual store deletion only can be done by specific user group.

internal/venice-common/src/main/java/com/linkedin/venice/acl/AccessController.java

@@ -48,6 +48,16 @@ public interface AccessController {
    */
   boolean isAllowlistUsers(X509Certificate clientCert, String resource, String method);
 
+  /**
+   * Check whether the client is the allowlist admin users for store deletion.
+   *
+   * @param clientCert the X509Certificate submitted by client
+   * @param resource the resource being requested;
+   * @param method the operation (GET, POST, ...) to perform against the resource
+   * @return true if the client is admin
+   */
+  boolean isAllowlistUsersForStoreDeletion(X509Certificate clientCert, String resource, String method);
+
   /**
    * Get principal Id from client certificate.
    * @param clientCert the X509Certificate submitted by client

services/venice-controller/src/main/java/com/linkedin/venice/controller/server/AbstractRoute.java

@@ -150,6 +150,23 @@ protected boolean isAllowListUser(Request request) {
     return accessController.get().isAllowlistUsers(certificate, storeName, HTTP_GET);
   }
 
+  /**
+   * Check whether the user is within the admin users allowlist for store deletion.
+   */
+  protected boolean isAllowListUserForStoreDeletion(Request request) {
+    if (!isAclEnabled()) {
+      /**
+       * Grant access if it's not required to check ACL.
+       * {@link accessController} will be empty if ACL is not enabled.
+       */
+      return true;
+    }
+    X509Certificate certificate = getCertificate(request);
+
+    String storeName = request.queryParamOrDefault(NAME, STORE_UNKNOWN);
+    return accessController.get().isAllowlistUsersForStoreDeletion(certificate, storeName, HTTP_GET);
+  }
+
   /**
    * @return whether SSL is enabled
    */

services/venice-controller/src/main/java/com/linkedin/venice/controller/server/StoresRoutes.java

@@ -506,6 +506,12 @@ public Route deleteStore(Admin admin) {
     return new VeniceRouteHandler<TrackableControllerResponse>(TrackableControllerResponse.class) {
       @Override
       public void internalHandle(Request request, TrackableControllerResponse veniceResponse) {
+        // This is to limit the manual store deletion from admin tool without https to a specific allow list users.
+        if (!isSslEnabled()
+            && !checkIsAllowListUser(request, veniceResponse, () -> isAllowListUserForStoreDeletion(request))) {
+          return;
+        }
+
         // Only allow allowlist users to run this command
         if (!checkIsAllowListUser(request, veniceResponse, () -> isAllowListUser(request))) {
           return;

services/venice-controller/src/test/java/com/linkedin/venice/controller/server/StoreRoutesTest.java

@@ -1,12 +1,16 @@
 package com.linkedin.venice.controller.server;
 
+import static com.linkedin.venice.HttpConstants.HTTP_GET;
+import static com.linkedin.venice.VeniceConstants.CONTROLLER_SSL_CERTIFICATE_ATTRIBUTE_NAME;
 import static com.linkedin.venice.exceptions.ErrorType.INCORRECT_CONTROLLER;
 import static com.linkedin.venice.exceptions.ErrorType.STORE_NOT_FOUND;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doCallRealMethod;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 
+import com.linkedin.venice.acl.DynamicAccessController;
 import com.linkedin.venice.controller.Admin;
 import com.linkedin.venice.controller.VeniceHelixAdmin;
 import com.linkedin.venice.controller.VeniceParentHelixAdmin;
@@ -21,10 +25,12 @@
 import com.linkedin.venice.meta.ZKStore;
 import com.linkedin.venice.pubsub.PubSubTopicRepository;
 import com.linkedin.venice.utils.ObjectMapperFactory;
+import java.security.cert.X509Certificate;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
+import javax.servlet.http.HttpServletRequest;
 import org.testng.Assert;
 import org.testng.annotations.Test;
 import spark.QueryParamsMap;
@@ -216,4 +222,43 @@ public void testGetStore() throws Exception {
             StoreResponse.class);
     Assert.assertEquals(response.getStore().getMaxRecordSizeBytes(), testMaxRecordSizeBytesValue);
   }
+
+  @Test
+  public void testDeleteStoreAclChecking() throws Exception {
+    final Store testStore = new ZKStore(
+        TEST_STORE_NAME,
+        "owner",
+        System.currentTimeMillis(),
+        PersistenceType.IN_MEMORY,
+        RoutingStrategy.CONSISTENT_HASH,
+        ReadStrategy.ANY_OF_ONLINE,
+        OfflinePushStrategy.WAIT_N_MINUS_ONE_REPLCIA_PER_PARTITION,
+        1);
+
+    final int testMaxRecordSizeBytesValue = 33333;
+    final Admin mockAdmin = mock(VeniceParentHelixAdmin.class);
+    doReturn(true).when(mockAdmin).isLeaderControllerFor(TEST_CLUSTER);
+    doReturn(testStore).when(mockAdmin).getStore(TEST_CLUSTER, TEST_STORE_NAME);
+
+    final Request request = mock(Request.class);
+    doReturn(TEST_CLUSTER).when(request).queryParams(eq(ControllerApiConstants.CLUSTER));
+    doReturn(TEST_STORE_NAME).when(request).queryParamOrDefault(eq(ControllerApiConstants.NAME), eq("STORE_UNKNOWN"));
+    HttpServletRequest rawRequest = mock(HttpServletRequest.class);
+    doReturn(rawRequest).when(request).raw();
+    X509Certificate[] certificates = new X509Certificate[1];
+    doReturn(certificates).when(rawRequest).getAttribute(eq(CONTROLLER_SSL_CERTIFICATE_ATTRIBUTE_NAME));
+
+    DynamicAccessController accessController = mock(DynamicAccessController.class);
+
+    doReturn(false).when(accessController).isAllowlistUsersForStoreDeletion(any(), eq(TEST_STORE_NAME), eq(HTTP_GET));
+
+    final StoresRoutes storesRoutes = new StoresRoutes(false, Optional.of(accessController), pubSubTopicRepository);
+    final StoreResponse response = ObjectMapperFactory.getInstance()
+        .readValue(
+            storesRoutes.deleteStore(mockAdmin).handle(request, mock(Response.class)).toString(),
+            StoreResponse.class);
+    Assert.assertTrue(response.isError());
+    Assert.assertTrue(response.getError().startsWith(VeniceRouteHandler.ACL_CHECK_FAILURE_WARN_MESSAGE_PREFIX));
+  }
+
 }

services/venice-router/src/test/java/com/linkedin/venice/router/acl/RouterStoreAclHandlerTest.java

@@ -245,6 +245,11 @@ public boolean isAllowlistUsers(X509Certificate clientCert, String resource, Str
       return true;
     }
 
+    @Override
+    public boolean isAllowlistUsersForStoreDeletion(X509Certificate clientCert, String resource, String method) {
+      return isAllowlistUsers(clientCert, resource, method);
+    }
+
     @Override
     public String getPrincipalId(X509Certificate clientCert) {
       assertNotNull(clientCert, resourceType.toString());

services/venice-server/src/test/java/com/linkedin/venice/listener/ServerStoreAclHandlerTest.java

@@ -92,6 +92,11 @@ public boolean isAllowlistUsers(X509Certificate clientCert, String resource, Str
       return true;
     }
 
+    @Override
+    public boolean isAllowlistUsersForStoreDeletion(X509Certificate clientCert, String resource, String method) {
+      return this.isAllowlistUsers(clientCert, resource, method);
+    }
+
     @Override
     public String getPrincipalId(X509Certificate clientCert) {
       assertNotNull(clientCert, queryAction.toString());