chassis supports RBAC authentication

Author: chengyouling

clients/http-client-common/src/test/java/org/apache/servicecomb/http/client/common/AbstractAddressManagerTest.java

@@ -313,4 +313,16 @@ public void compareAndGetAddressTest() {
     Assertions.assertTrue(manager.compareAndGetAddress("192.168.20.16").isEmpty());
     Assertions.assertEquals("https://192.168.20.160:30100", manager.compareAndGetAddress("192.168.20.160"));
   }
+
+  @Test
+  public void AddressAffinityTest() {
+    List<String> testAddr = new ArrayList<>();
+    testAddr.add("https://192.168.20.160:30100?region=region1&availableZone=zone1");
+    testAddr.add("https://127.0.0.1:30100");
+    AbstractAddressManager manager = new AbstractAddressManager(testAddr, "region1", "zone1");
+    Assertions.assertEquals("https://192.168.20.160:30100", manager.address());
+
+    AbstractAddressManager manager2 = new AbstractAddressManager("default", testAddr, "region1", "zone1");
+    Assertions.assertEquals("https://192.168.20.160:30100", manager2.address());
+  }
 }

foundations/foundation-spi/src/main/java/org/apache/servicecomb/foundation/auth/AuthHeaderProvider.java

@@ -22,7 +22,7 @@
 
 public interface AuthHeaderProvider {
   /**
-   * Obtain RBAC authentication request header, host is the key of cache, spring-cloud module have specific usage of host
+   * Obtain RBAC authentication request header, host is the key of cache
    *
    * @param host engine address ip
    * @return auth headers

service-registry/registry-service-center/src/main/java/org/apache/servicecomb/serviceregistry/auth/TokenAuthHeaderProvider.java

@@ -27,7 +27,7 @@
 public class TokenAuthHeaderProvider implements AuthHeaderProvider {
   @Override
   public Map<String, String> authHeaders(String host) {
-    String token = TokenCacheManager.getInstance().getToken(RBACBootStrapService.DEFAULT_REGISTRY_NAME);
+    String token = TokenCacheManager.getInstance().getToken(host);
     if (StringUtils.isEmpty(token)) {
       return new HashMap<>();
     }

service-registry/registry-service-center/src/main/java/org/apache/servicecomb/serviceregistry/auth/TokenCacheManager.java

@@ -17,9 +17,8 @@
 
 package org.apache.servicecomb.serviceregistry.auth;
 
+import java.net.URI;
 import java.util.Map;
-import java.util.Objects;
-import java.util.Optional;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
@@ -28,13 +27,11 @@
 
 import org.apache.commons.lang3.StringUtils;
 import org.apache.servicecomb.foundation.auth.Cipher;
-import org.apache.servicecomb.foundation.common.concurrent.ConcurrentHashMapEx;
-import org.apache.servicecomb.http.client.event.EngineConnectChangedEvent;
+import org.apache.servicecomb.http.client.event.OperationEvents.UnAuthorizedOperationEvent;
 import org.apache.servicecomb.registry.api.event.ServiceCenterEventBus;
 import org.apache.servicecomb.service.center.client.ServiceCenterClient;
 import org.apache.servicecomb.service.center.client.model.RbacTokenRequest;
 import org.apache.servicecomb.service.center.client.model.RbacTokenResponse;
-import org.apache.servicecomb.serviceregistry.event.NotPermittedEvent;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -55,42 +52,35 @@ public final class TokenCacheManager {
 
   private static final TokenCacheManager INSTANCE = new TokenCacheManager();
 
+  private Map<String, ServiceCenterClient> serviceCenterClients;
 
-  private final Map<String, TokenCache> tokenCacheMap;
+  private TokenCache tokenCache;
 
-  private Map<String, ServiceCenterClient> serviceCenterClients;
+  private static final Object LOCK = new Object();
 
   public static TokenCacheManager getInstance() {
     return INSTANCE;
   }
 
   private TokenCacheManager() {
-    tokenCacheMap = new ConcurrentHashMapEx<>();
   }
 
   public void setServiceCenterClients(Map<String, ServiceCenterClient> serviceCenterClients) {
     this.serviceCenterClients = serviceCenterClients;
   }
 
   public void addTokenCache(String registryName, String accountName, String password, Cipher cipher) {
-    Objects.requireNonNull(registryName, "registryName should not be null!");
-    if (tokenCacheMap.containsKey(registryName)) {
-      LOGGER.warn("duplicate token cache registration for serviceRegistry[{}]", registryName);
-      return;
-    }
-
-    tokenCacheMap.put(registryName, new TokenCache(registryName, accountName, password, cipher));
+    tokenCache = new TokenCache(registryName, accountName, password, cipher);
   }
 
-  public String getToken(String registryName) {
-    return Optional.ofNullable(tokenCacheMap.get(registryName))
-        .map(TokenCache::getToken)
-        .orElse(null);
+  public String getToken(String host) {
+    if (tokenCache == null) {
+      return null;
+    }
+    return tokenCache.getToken(host);
   }
 
   public class TokenCache {
-    private static final String UN_AUTHORIZED_CODE_HALF_OPEN = "401302";
-
     private static final long TOKEN_REFRESH_TIME_IN_SECONDS = 20 * 60 * 1000;
 
     private final String registryName;
@@ -105,10 +95,6 @@ public class TokenCache {
 
     private final Cipher cipher;
 
-    private int lastStatusCode;
-
-    private String lastErrorCode;
-
     public TokenCache(String registryName, String accountName, String password,
         Cipher cipher) {
       this.registryName = registryName;
@@ -133,47 +119,41 @@ public void run() {
             .build(new CacheLoader<String, String>() {
               @Override
               public String load(String key) throws Exception {
-                return createHeaders();
+                return createHeaders(key);
               }
 
               @Override
               public ListenableFuture<String> reload(String key, String oldValue) throws Exception {
-                return Futures.submit(() -> createHeaders(), executorService);
+                return Futures.submit(() -> createHeaders(key), executorService);
               }
             });
         ServiceCenterEventBus.getEventBus().register(this);
       }
     }
 
     @Subscribe
-    public void onNotPermittedEvent(NotPermittedEvent event) {
-      this.executorService.submit(() -> {
-        if (lastStatusCode == Status.UNAUTHORIZED.getStatusCode() && UN_AUTHORIZED_CODE_HALF_OPEN
-            .equals(lastErrorCode)) {
-          cache.refresh(registryName);
-        }
-      });
+    public void onUnAuthorizedOperationEvent(UnAuthorizedOperationEvent event) {
+      LOGGER.warn("address {} unAuthorized, refresh cache token!", event.getAddress());
+      cache.refresh(getHostByAddress(event.getAddress()));
     }
 
-    @Subscribe
-    public void onEngineConnectChangedEvent(EngineConnectChangedEvent event) {
-      cache.refresh(registryName);
+    private String getHostByAddress(String address) {
+      try {
+        URI uri = URI.create(address);
+        return uri.getHost();
+      } catch (Exception e) {
+        LOGGER.error("get host by address [{}] error!", address, e);
+        return registryName;
+      }
     }
 
-    private String createHeaders() {
-      LOGGER.info("start to create RBAC headers");
-
+    private String createHeaders(String host) {
+      LOGGER.info("start to create RBAC headers for host: {}", host);
       ServiceCenterClient serviceCenterClient = serviceCenterClients.get(this.registryName);
-
       RbacTokenRequest request = new RbacTokenRequest();
       request.setName(accountName);
       request.setPassword(new String(cipher.decrypt(password.toCharArray())));
-
       RbacTokenResponse rbacTokenResponse = serviceCenterClient.queryToken(request, "");
-
-      this.lastStatusCode = rbacTokenResponse.getStatusCode();
-      this.lastErrorCode = rbacTokenResponse.getErrorCode();
-
       if (Status.UNAUTHORIZED.getStatusCode() == rbacTokenResponse.getStatusCode()
           || Status.FORBIDDEN.getStatusCode() == rbacTokenResponse.getStatusCode()) {
         // password wrong, do not try anymore
@@ -185,25 +165,36 @@ private String createHeaders() {
         LOGGER.warn("service center do not support RBAC token, you should not config account info");
         return INVALID_TOKEN;
       }
+      if (Status.INTERNAL_SERVER_ERROR.getStatusCode() == rbacTokenResponse.getStatusCode()) {
+        // return null for server_error, so the token information can be re-fetched on the next call.
+        // It will prompt 'CacheLoader returned null for key xxx'
+        LOGGER.warn("service center query RBAC token error!");
+        return null;
+      }
 
-      LOGGER.info("refresh token successfully {}", rbacTokenResponse.getStatusCode());
+      LOGGER.info("refresh host [{}] token successfully {}", host, rbacTokenResponse.getStatusCode());
       return rbacTokenResponse.getToken();
     }
 
     protected long refreshTime() {
       return TOKEN_REFRESH_TIME_IN_SECONDS;
     }
 
-    public String getToken() {
+    public String getToken(String host) {
       if (!enabled()) {
         return null;
       }
-
-      try {
-        return cache.get(registryName);
-      } catch (Exception e) {
-        LOGGER.error("failed to create token", e);
-        return null;
+      String address = host;
+      if (StringUtils.isEmpty(address)) {
+        address = registryName;
+      }
+      synchronized (LOCK) {
+        try {
+          return cache.get(address);
+        } catch (Exception e) {
+          LOGGER.error("failed to create token", e);
+          return null;
+        }
       }
     }
 

service-registry/registry-service-center/src/main/java/org/apache/servicecomb/serviceregistry/client/http/ServiceRegistryClientImpl.java

@@ -35,6 +35,7 @@
 import org.apache.servicecomb.foundation.common.net.IpPort;
 import org.apache.servicecomb.foundation.common.utils.JsonUtils;
 import org.apache.servicecomb.foundation.vertx.AsyncResultCallback;
+import org.apache.servicecomb.http.client.event.OperationEvents.UnAuthorizedOperationEvent;
 import org.apache.servicecomb.http.client.utils.ServiceCombServiceAvailableUtils;
 import org.apache.servicecomb.registry.api.event.MicroserviceInstanceChangedEvent;
 import org.apache.servicecomb.registry.api.registry.FindInstancesResponse;
@@ -152,7 +153,7 @@ public <T> Handler<RestResponse> syncHandler(CountDownLatch countDownLatch, Clas
         return;
       }
       holder.setStatusCode(response.statusCode());
-      sendUnAuthorizedEvent(response);
+      sendUnAuthorizedEvent(response, requestContext);
       response.exceptionHandler(e -> {
         LOGGER.error("error in processing response.", e);
         countDownLatch.countDown();
@@ -232,7 +233,7 @@ private Handler<RestResponse> syncHandlerEx(CountDownLatch countDownLatch, Holde
 
         return;
       }
-      sendUnAuthorizedEvent(response);
+      sendUnAuthorizedEvent(response, requestContext);
       response.exceptionHandler(e -> {
         LOGGER.error("error in processing response.", e);
         countDownLatch.countDown();
@@ -247,12 +248,20 @@ private Handler<RestResponse> syncHandlerEx(CountDownLatch countDownLatch, Holde
     };
   }
 
-  private void sendUnAuthorizedEvent(HttpClientResponse response) {
+  private void sendUnAuthorizedEvent(HttpClientResponse response, RequestContext requestContext) {
     if (response.statusCode() == Status.UNAUTHORIZED.getStatusCode()) {
-      ServiceCenterEventBus.getEventBus().post(new NotPermittedEvent());
+      ServiceCenterEventBus.getEventBus().post(new UnAuthorizedOperationEvent(getAddressWithProtocol(requestContext)));
     }
   }
 
+  private String getAddressWithProtocol(RequestContext requestContext) {
+    String ipAndPort = requestContext.getIpPort().toString();
+    if (ipAndPort.startsWith("http")) {
+      return ipAndPort;
+    }
+    return "https://" + ipAndPort;
+  }
+
   private Handler<RestResponse> syncHandlerForInstances(CountDownLatch countDownLatch,
       MicroserviceInstances mInstances) {
     return restResponse -> {
@@ -1007,7 +1016,7 @@ public <T> Handler<RestResponse> addressSyncHandler(CountDownLatch countDownLatc
         return;
       }
       holder.setStatusCode(response.statusCode());
-      sendUnAuthorizedEvent(response);
+      sendUnAuthorizedEvent(response, restResponse.getRequestContext());
       countDownLatch.countDown();
     };
   }