From f56e900d9f503cf3d8ccc7f3c6e90e6ed180b8eb Mon Sep 17 00:00:00 2001 From: Artem Vysochyn Date: Sat, 28 Sep 2024 09:46:50 +0300 Subject: [PATCH] Enhance auth interfaces (#850) * Added CredentialsCodec * Disabled unstable - test_remote_node_died_mono_never * Moved CredentialsSupplier to ServiceTransport --- .../services/auth/CredentialsCodec.java | 96 +++++++++++++++++++ .../services/auth/CredentialsSupplier.java | 14 --- .../transport/api/ServiceTransport.java | 11 +++ .../auth/CompositeProfileAuthExample.java | 2 +- .../auth/PrincipalMapperAuthExample.java | 2 +- .../auth/ServiceTransportAuthExample.java | 2 +- .../rsocket/RSocketClientTransport.java | 2 +- .../rsocket/RSocketServiceTransport.java | 1 - .../rsocket/RSocketServiceTransportTest.java | 2 + 9 files changed, 113 insertions(+), 19 deletions(-) create mode 100644 services-api/src/main/java/io/scalecube/services/auth/CredentialsCodec.java delete mode 100644 services-api/src/main/java/io/scalecube/services/auth/CredentialsSupplier.java diff --git a/services-api/src/main/java/io/scalecube/services/auth/CredentialsCodec.java b/services-api/src/main/java/io/scalecube/services/auth/CredentialsCodec.java new file mode 100644 index 000000000..e4c44256a --- /dev/null +++ b/services-api/src/main/java/io/scalecube/services/auth/CredentialsCodec.java @@ -0,0 +1,96 @@ +package io.scalecube.services.auth; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.InputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.io.OutputStream; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Objects; +import reactor.core.Exceptions; + +public class CredentialsCodec { + + private CredentialsCodec() { + // Do not instantiate + } + + /** + * Encodes the given credentials to the given stream. + * + * @param stream stream + * @param credentials credentials + */ + public static void encode(OutputStream stream, Map credentials) { + if (credentials == null) { + return; + } + Objects.requireNonNull(stream, "output stream"); + try (ObjectOutputStream out = new ObjectOutputStream(stream)) { + // credentials + out.writeInt(credentials.size()); + for (Entry entry : credentials.entrySet()) { + out.writeUTF(entry.getKey()); + out.writeObject(entry.getValue()); // value is nullable + } + + out.flush(); + } catch (Throwable th) { + throw Exceptions.propagate(th); + } + } + + /** + * Encodes the given credentials to a byte array. + * + * @param credentials credentials + * @return byte array representation of credentials + */ + public static byte[] toByteArray(Map credentials) { + if (credentials == null || credentials.isEmpty()) { + return new byte[0]; + } + ByteArrayOutputStream output = new ByteArrayOutputStream(); + encode(output, credentials); + return output.toByteArray(); + } + + /** + * Decodes the given stream to credentials as {@code Map}. + * + * @return credentials + */ + public static Map decode(InputStream stream) { + Objects.requireNonNull(stream, "input stream"); + try (ObjectInputStream in = new ObjectInputStream(stream)) { + // credentials + int credentialsSize = in.readInt(); + Map credentials = new HashMap<>(credentialsSize); + for (int i = 0; i < credentialsSize; i++) { + String key = in.readUTF(); + String value = (String) in.readObject(); // value is nullable + credentials.put(key, value); + } + return Collections.unmodifiableMap(credentials); + } catch (Throwable th) { + throw Exceptions.propagate(th); + } + } + + /** + * Decodes the given byte array to credentials as {@code Map}. + * + * @return credentials + */ + public static Map decode(byte[] bytes) { + if (bytes == null || bytes.length == 0) { + return Collections.emptyMap(); + } + ByteArrayInputStream input = new ByteArrayInputStream(bytes); + return decode(input); + } +} diff --git a/services-api/src/main/java/io/scalecube/services/auth/CredentialsSupplier.java b/services-api/src/main/java/io/scalecube/services/auth/CredentialsSupplier.java deleted file mode 100644 index 5eff4dfd9..000000000 --- a/services-api/src/main/java/io/scalecube/services/auth/CredentialsSupplier.java +++ /dev/null @@ -1,14 +0,0 @@ -package io.scalecube.services.auth; - -import io.scalecube.services.ServiceReference; -import java.util.Map; -import java.util.function.Function; -import reactor.core.publisher.Mono; - -/** - * Returns credentials for the given {@link ServiceReference}. Credentials are being returned in - * most generic form which is {@code Map}. - */ -@FunctionalInterface -public interface CredentialsSupplier - extends Function>> {} diff --git a/services-api/src/main/java/io/scalecube/services/transport/api/ServiceTransport.java b/services-api/src/main/java/io/scalecube/services/transport/api/ServiceTransport.java index 8c9364562..c4007a590 100644 --- a/services-api/src/main/java/io/scalecube/services/transport/api/ServiceTransport.java +++ b/services-api/src/main/java/io/scalecube/services/transport/api/ServiceTransport.java @@ -1,6 +1,10 @@ package io.scalecube.services.transport.api; +import io.scalecube.services.ServiceReference; import io.scalecube.services.registry.api.ServiceRegistry; +import java.util.Map; +import java.util.function.Function; +import reactor.core.publisher.Mono; public interface ServiceTransport { @@ -28,4 +32,11 @@ public interface ServiceTransport { /** Shutdowns transport and release occupied resources. */ void stop(); + + /** + * Returns credentials for the given {@link ServiceReference}. Credentials are being returned in + * most generic form which is {@code Map}. + */ + @FunctionalInterface + interface CredentialsSupplier extends Function>> {} } 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 index 8d7aba427..b67d63fee 100644 --- 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 @@ -7,9 +7,9 @@ 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.api.ServiceTransport.CredentialsSupplier; import io.scalecube.services.transport.rsocket.RSocketServiceTransport; import io.scalecube.transport.netty.websocket.WebsocketTransportFactory; import java.time.Duration; diff --git a/services-examples/src/main/java/io/scalecube/services/examples/auth/PrincipalMapperAuthExample.java b/services-examples/src/main/java/io/scalecube/services/examples/auth/PrincipalMapperAuthExample.java index 5c97ac099..abbf515c9 100644 --- a/services-examples/src/main/java/io/scalecube/services/examples/auth/PrincipalMapperAuthExample.java +++ b/services-examples/src/main/java/io/scalecube/services/examples/auth/PrincipalMapperAuthExample.java @@ -4,9 +4,9 @@ import io.scalecube.services.ServiceEndpoint; import io.scalecube.services.ServiceInfo; 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.api.ServiceTransport.CredentialsSupplier; import io.scalecube.services.transport.rsocket.RSocketServiceTransport; import io.scalecube.transport.netty.websocket.WebsocketTransportFactory; import java.time.Duration; 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 index 23f5d0cb0..52ea9b101 100644 --- 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 @@ -3,9 +3,9 @@ import io.scalecube.services.Microservices; import io.scalecube.services.ServiceEndpoint; 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.api.ServiceTransport.CredentialsSupplier; import io.scalecube.services.transport.rsocket.RSocketServiceTransport; import io.scalecube.transport.netty.websocket.WebsocketTransportFactory; import java.time.Duration; diff --git a/services-transport-parent/services-transport-rsocket/src/main/java/io/scalecube/services/transport/rsocket/RSocketClientTransport.java b/services-transport-parent/services-transport-rsocket/src/main/java/io/scalecube/services/transport/rsocket/RSocketClientTransport.java index b97a6847f..41f2efb7e 100644 --- a/services-transport-parent/services-transport-rsocket/src/main/java/io/scalecube/services/transport/rsocket/RSocketClientTransport.java +++ b/services-transport-parent/services-transport-rsocket/src/main/java/io/scalecube/services/transport/rsocket/RSocketClientTransport.java @@ -10,7 +10,6 @@ import io.rsocket.util.ByteBufPayload; import io.scalecube.services.Address; import io.scalecube.services.ServiceReference; -import io.scalecube.services.auth.CredentialsSupplier; import io.scalecube.services.exceptions.MessageCodecException; import io.scalecube.services.exceptions.ServiceException; import io.scalecube.services.exceptions.UnauthorizedException; @@ -18,6 +17,7 @@ import io.scalecube.services.transport.api.ClientTransport; import io.scalecube.services.transport.api.DataCodec; import io.scalecube.services.transport.api.HeadersCodec; +import io.scalecube.services.transport.api.ServiceTransport.CredentialsSupplier; import java.util.Collection; import java.util.Collections; import java.util.Map; diff --git a/services-transport-parent/services-transport-rsocket/src/main/java/io/scalecube/services/transport/rsocket/RSocketServiceTransport.java b/services-transport-parent/services-transport-rsocket/src/main/java/io/scalecube/services/transport/rsocket/RSocketServiceTransport.java index 41b684942..a1aac61ef 100644 --- a/services-transport-parent/services-transport-rsocket/src/main/java/io/scalecube/services/transport/rsocket/RSocketServiceTransport.java +++ b/services-transport-parent/services-transport-rsocket/src/main/java/io/scalecube/services/transport/rsocket/RSocketServiceTransport.java @@ -7,7 +7,6 @@ import io.netty.util.concurrent.DefaultThreadFactory; import io.netty.util.concurrent.Future; import io.scalecube.services.auth.Authenticator; -import io.scalecube.services.auth.CredentialsSupplier; import io.scalecube.services.exceptions.ConnectionClosedException; import io.scalecube.services.registry.api.ServiceRegistry; import io.scalecube.services.transport.api.ClientTransport; diff --git a/services/src/test/java/io/scalecube/services/transport/rsocket/RSocketServiceTransportTest.java b/services/src/test/java/io/scalecube/services/transport/rsocket/RSocketServiceTransportTest.java index 37fa6bc5f..0c41f4ab6 100644 --- a/services/src/test/java/io/scalecube/services/transport/rsocket/RSocketServiceTransportTest.java +++ b/services/src/test/java/io/scalecube/services/transport/rsocket/RSocketServiceTransportTest.java @@ -22,6 +22,7 @@ import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import reactor.core.Disposable; import reactor.core.publisher.Mono; @@ -78,6 +79,7 @@ public void cleanUp() { } } + @Disabled @Test public void test_remote_node_died_mono_never() throws Exception { int batchSize = 1;