Enhanced ServiceMethodInvoker: added support of composite authentication contexts; added example

artem-v · artem-v · commit 4368e1e87a35 · 2021-10-30T02:04:36.000+03:00
diff --git a/services-api/src/main/java/io/scalecube/services/methods/ServiceMethodInvoker.java b/services-api/src/main/java/io/scalecube/services/methods/ServiceMethodInvoker.java
@@ -113,17 +113,17 @@ public Flux<ServiceMessage> invokeBidirectional(Publisher<ServiceMessage> publis
 
   private Mono<?> deferWithContextOne(ServiceMessage message, Object authData) {
     return Mono.deferContextual(context -> Mono.from(invoke(toRequest(message))))
-        .contextWrite(context -> enhanceContextWithPrincipal(authData, context));
+        .contextWrite(context -> enhanceContext(authData, context));
   }
 
   private Flux<?> deferWithContextMany(ServiceMessage message, Object authData) {
     return Flux.deferContextual(context -> Flux.from(invoke(toRequest(message))))
-        .contextWrite(context -> enhanceContextWithPrincipal(authData, context));
+        .contextWrite(context -> enhanceContext(authData, context));
   }
 
   private Flux<?> deferWithContextBidirectional(Flux<ServiceMessage> messages, Object authData) {
     return Flux.deferContextual(context -> messages.map(this::toRequest).transform(this::invoke))
-        .contextWrite(context -> enhanceContextWithPrincipal(authData, context));
+        .contextWrite(context -> enhanceContext(authData, context));
   }
 
   private Publisher<?> invoke(Object request) {
@@ -160,16 +160,19 @@ private Mono<Object> authenticate(ServiceMessage message, Context context) {
       return Mono.just(NULL_AUTH_CONTEXT);
     }
 
-    if (context.hasKey(AUTH_CONTEXT_KEY)) {
-      return Mono.just(context.get(AUTH_CONTEXT_KEY));
-    }
-
     if (authenticator == null) {
-      LOGGER.error("Authentication failed (auth context not found and authenticator not set)");
-      throw new UnauthorizedException("Authentication failed");
+      if (context.hasKey(AUTH_CONTEXT_KEY)) {
+        return Mono.just(context.get(AUTH_CONTEXT_KEY));
+      } else {
+        LOGGER.error("Authentication failed (auth context not found and authenticator not set)");
+        throw new UnauthorizedException("Authentication failed");
+      }
     }
 
-    return authenticator.apply(message.headers()).onErrorMap(this::toUnauthorizedException);
+    return authenticator
+        .apply(message.headers())
+        .switchIfEmpty(Mono.just(NULL_AUTH_CONTEXT))
+        .onErrorMap(this::toUnauthorizedException);
   }
 
   private UnauthorizedException toUnauthorizedException(Throwable th) {
@@ -181,12 +184,14 @@ private UnauthorizedException toUnauthorizedException(Throwable th) {
     }
   }
 
-  private Context enhanceContextWithPrincipal(Object authData, Context context) {
-    if (authData == NULL_AUTH_CONTEXT || authData == null) {
-      return context;
+  private Context enhanceContext(Object authData, Context context) {
+    if (authData == NULL_AUTH_CONTEXT || principalMapper == null) {
+      return context.put(AUTH_CONTEXT_KEY, authData);
     }
-    return context.put(
-        AUTH_CONTEXT_KEY, principalMapper != null ? principalMapper.apply(authData) : authData);
+
+    Object mappedData = principalMapper.apply(authData);
+
+    return context.put(AUTH_CONTEXT_KEY, mappedData != null ? mappedData : NULL_AUTH_CONTEXT);
   }
 
   private Object toRequest(ServiceMessage message) {
diff --git a/services-examples/src/main/java/io/scalecube/services/examples/auth/CompositeProfile.java b/services-examples/src/main/java/io/scalecube/services/examples/auth/CompositeProfile.java
@@ -0,0 +1,31 @@
+package io.scalecube.services.examples.auth;
+
+import java.util.StringJoiner;
+
+public class CompositeProfile {
+
+  private final ServiceEndpointProfile serviceEndpointProfile;
+  private final UserProfile userProfile;
+
+  public CompositeProfile(
+      ServiceEndpointProfile serviceEndpointProfile, UserProfile userProfile) {
+    this.serviceEndpointProfile = serviceEndpointProfile;
+    this.userProfile = userProfile;
+  }
+
+  public ServiceEndpointProfile serviceEndpointProfile() {
+    return serviceEndpointProfile;
+  }
+
+  public UserProfile userProfile() {
+    return userProfile;
+  }
+
+  @Override
+  public String toString() {
+    return new StringJoiner(", ", CompositeProfile.class.getSimpleName() + "[", "]")
+        .add("serviceEndpointProfile=" + serviceEndpointProfile)
+        .add("userProfile=" + userProfile)
+        .toString();
+  }
+}
diff --git a/services-examples/src/main/java/io/scalecube/services/examples/auth/CompositeProfileAuthExample.java b/services-examples/src/main/java/io/scalecube/services/examples/auth/CompositeProfileAuthExample.java
@@ -0,0 +1,113 @@
+package io.scalecube.services.examples.auth;
+
+import static io.scalecube.services.auth.MonoAuthUtil.deferWithPrincipal;
+
+import io.scalecube.services.Microservices;
+import io.scalecube.services.ServiceEndpoint;
+import io.scalecube.services.ServiceInfo;
+import io.scalecube.services.api.ServiceMessage;
+import io.scalecube.services.auth.Authenticator;
+import io.scalecube.services.auth.CredentialsSupplier;
+import io.scalecube.services.discovery.ScalecubeServiceDiscovery;
+import io.scalecube.services.exceptions.UnauthorizedException;
+import io.scalecube.services.transport.rsocket.RSocketServiceTransport;
+import io.scalecube.transport.netty.websocket.WebsocketTransportFactory;
+import java.time.Duration;
+import java.util.Collections;
+import reactor.core.publisher.Mono;
+
+public class CompositeProfileAuthExample {
+
+  /**
+   * Main program.
+   *
+   * @param args arguments
+   */
+  public static void main(String[] args) {
+    Microservices service =
+        Microservices.builder()
+            .discovery(
+                serviceEndpoint ->
+                    new ScalecubeServiceDiscovery()
+                        .transport(cfg -> cfg.transportFactory(new WebsocketTransportFactory()))
+                        .options(opts -> opts.metadata(serviceEndpoint)))
+            .transport(() -> new RSocketServiceTransport().authenticator(authenticator()))
+            .services(
+                call ->
+                    Collections.singletonList(
+                        ServiceInfo.fromServiceInstance(new SecuredServiceByCompositeProfileImpl())
+                            .authenticator(compositeAuthenticator())
+                            .build()))
+            .startAwait();
+
+    Microservices caller =
+        Microservices.builder()
+            .discovery(endpoint -> discovery(service, endpoint))
+            .transport(
+                () -> new RSocketServiceTransport().credentialsSupplier(credentialsSupplier()))
+            .startAwait();
+
+    ServiceMessage response =
+        caller
+            .call()
+            .requestOne(
+                ServiceMessage.builder()
+                    .qualifier("securedServiceByCompositeProfile/hello")
+                    .header("userProfile.name", "SEGA")
+                    .header("userProfile.role", "ADMIN")
+                    .data("hello world")
+                    .build(),
+                String.class)
+            .block(Duration.ofSeconds(3));
+
+    System.err.println("### Received 'caller' response: " + response.data());
+  }
+
+  private static Authenticator<ServiceEndpointProfile> authenticator() {
+    return headers -> {
+      String transportSessionKey = headers.get("transportSessionKey");
+
+      if ("asdf7hasd9hasd7fha8ds7fahsdf87".equals(transportSessionKey)) {
+        return Mono.just(new ServiceEndpointProfile("endpoint123", "operations"));
+      }
+
+      return Mono.error(
+          new UnauthorizedException("Authentication failed (transportSessionKey incorrect)"));
+    };
+  }
+
+  private static CredentialsSupplier credentialsSupplier() {
+    return service ->
+        Mono.just(
+            Collections.singletonMap("transportSessionKey", "asdf7hasd9hasd7fha8ds7fahsdf87"));
+  }
+
+  private static Authenticator<CompositeProfile> compositeAuthenticator() {
+    return headers ->
+        deferWithPrincipal(ServiceEndpointProfile.class)
+            .flatMap(
+                serviceEndpointProfile -> {
+
+                  // If userProfile not set then throw error
+                  if (!headers.containsKey("userProfile.name")
+                      || !headers.containsKey("userProfile.role")) {
+                    throw new UnauthorizedException("userProfile not found or invalid");
+                  }
+
+                  // Otherwise return new combined profile which will be stored under
+                  // AUTH_CONTEXT_KEY
+
+                  return Mono.just(
+                      new CompositeProfile(
+                          serviceEndpointProfile, UserProfile.fromHeaders(headers)));
+                });
+  }
+
+  private static ScalecubeServiceDiscovery discovery(
+      Microservices service, ServiceEndpoint endpoint) {
+    return new ScalecubeServiceDiscovery()
+        .transport(cfg -> cfg.transportFactory(new WebsocketTransportFactory()))
+        .options(opts -> opts.metadata(endpoint))
+        .membership(opts -> opts.seedMembers(service.discovery().address()));
+  }
+}
diff --git a/services-examples/src/main/java/io/scalecube/services/examples/auth/SecuredServiceByApiKey.java b/services-examples/src/main/java/io/scalecube/services/examples/auth/SecuredServiceByApiKey.java
@@ -6,7 +6,7 @@
 import reactor.core.publisher.Mono;
 
 @Secured
-@Service
+@Service("securedServiceByApiKey")
 public interface SecuredServiceByApiKey {
 
   @ServiceMethod
diff --git a/services-examples/src/main/java/io/scalecube/services/examples/auth/SecuredServiceByCompositeProfile.java b/services-examples/src/main/java/io/scalecube/services/examples/auth/SecuredServiceByCompositeProfile.java
@@ -0,0 +1,14 @@
+package io.scalecube.services.examples.auth;
+
+import io.scalecube.services.annotations.Service;
+import io.scalecube.services.annotations.ServiceMethod;
+import io.scalecube.services.auth.Secured;
+import reactor.core.publisher.Mono;
+
+@Secured
+@Service("securedServiceByCompositeProfile")
+public interface SecuredServiceByCompositeProfile {
+
+  @ServiceMethod
+  Mono<String> hello(String name);
+}
diff --git a/services-examples/src/main/java/io/scalecube/services/examples/auth/SecuredServiceByCompositeProfileImpl.java b/services-examples/src/main/java/io/scalecube/services/examples/auth/SecuredServiceByCompositeProfileImpl.java
@@ -0,0 +1,34 @@
+package io.scalecube.services.examples.auth;
+
+import io.scalecube.services.auth.MonoAuthUtil;
+import io.scalecube.services.exceptions.ForbiddenException;
+import reactor.core.publisher.Mono;
+
+public class SecuredServiceByCompositeProfileImpl implements SecuredServiceByCompositeProfile {
+
+  @Override
+  public Mono<String> hello(String name) {
+    return MonoAuthUtil.deferWithPrincipal(CompositeProfile.class)
+        .flatMap(
+            compositeProfile -> {
+              final UserProfile userProfile = compositeProfile.userProfile();
+              final ServiceEndpointProfile serviceEndpointProfile =
+                  compositeProfile.serviceEndpointProfile();
+              checkPermissions(userProfile);
+              return Mono.just(
+                  "Hello, name="
+                      + name
+                      + " (userProfile="
+                      + userProfile
+                      + ", serviceEndpointProfile="
+                      + serviceEndpointProfile
+                      + ")");
+            });
+  }
+
+  private void checkPermissions(UserProfile user) {
+    if (!user.role().equals("ADMIN")) {
+      throw new ForbiddenException("Forbidden");
+    }
+  }
+}
diff --git a/services-examples/src/main/java/io/scalecube/services/examples/auth/SecuredServiceByUserProfile.java b/services-examples/src/main/java/io/scalecube/services/examples/auth/SecuredServiceByUserProfile.java
@@ -6,7 +6,7 @@
 import reactor.core.publisher.Mono;
 
 @Secured
-@Service
+@Service("securedServiceByUserProfile")
 public interface SecuredServiceByUserProfile {
 
   @ServiceMethod
diff --git a/services-examples/src/main/java/io/scalecube/services/examples/auth/ServiceEndpointProfile.java b/services-examples/src/main/java/io/scalecube/services/examples/auth/ServiceEndpointProfile.java
@@ -0,0 +1,30 @@
+package io.scalecube.services.examples.auth;
+
+import java.util.StringJoiner;
+
+public class ServiceEndpointProfile {
+
+  private final String endpoint;
+  private final String serviceRole;
+
+  public ServiceEndpointProfile(String endpoint, String serviceRole) {
+    this.endpoint = endpoint;
+    this.serviceRole = serviceRole;
+  }
+
+  public String endpoint() {
+    return endpoint;
+  }
+
+  public String serviceRole() {
+    return serviceRole;
+  }
+
+  @Override
+  public String toString() {
+    return new StringJoiner(", ", ServiceEndpointProfile.class.getSimpleName() + "[", "]")
+        .add("endpoint='" + endpoint + "'")
+        .add("serviceRole='" + serviceRole + "'")
+        .toString();
+  }
+}
diff --git a/services-examples/src/main/java/io/scalecube/services/examples/auth/ServiceTransportAuthExample.java b/services-examples/src/main/java/io/scalecube/services/examples/auth/ServiceTransportAuthExample.java
@@ -35,7 +35,8 @@ public static void main(String[] args) {
     Microservices caller =
         Microservices.builder()
             .discovery(endpoint -> discovery(service, endpoint))
-            .transport(() -> new RSocketServiceTransport().credentialsSupplier(credsSupplier()))
+            .transport(
+                () -> new RSocketServiceTransport().credentialsSupplier(credentialsSupplier()))
             .startAwait();
 
     String response =
@@ -65,7 +66,7 @@ private static Authenticator<UserProfile> authenticator() {
     };
   }
 
-  private static CredentialsSupplier credsSupplier() {
+  private static CredentialsSupplier credentialsSupplier() {
     return service -> {
       HashMap<String, String> creds = new HashMap<>();
       creds.put("username", "Alice");
diff --git a/services-examples/src/main/java/io/scalecube/services/examples/auth/UserProfile.java b/services-examples/src/main/java/io/scalecube/services/examples/auth/UserProfile.java
@@ -1,5 +1,7 @@
 package io.scalecube.services.examples.auth;
 
+import java.util.Map;
+import java.util.Objects;
 import java.util.StringJoiner;
 
 public class UserProfile {
@@ -8,8 +10,12 @@ public class UserProfile {
   private final String role;
 
   public UserProfile(String name, String role) {
-    this.name = name;
-    this.role = role;
+    this.name = Objects.requireNonNull(name, "UserProfile.name");
+    this.role = Objects.requireNonNull(role, "UserProfile.role");
+  }
+
+  public static UserProfile fromHeaders(Map<String, String> headers) {
+    return new UserProfile(headers.get("userProfile.name"), headers.get("userProfile.role"));
   }
 
   public String name() {
diff --git a/services-transport-parent/services-transport-rsocket/src/main/java/io/scalecube/services/transport/rsocket/RSocketServiceAcceptor.java b/services-transport-parent/services-transport-rsocket/src/main/java/io/scalecube/services/transport/rsocket/RSocketServiceAcceptor.java
