Extend AuthenticationStepsOperator for non-blocking resource access

AuthenticationStepsOperator now uses DataBufferUtils to load credentials from a Resource for Suppliers that are instance of ResourceCredentialSupplier. Generic suppliers are called on the boundedElastic Scheduler to offload potentially blocking calls to a different thread.

Resolves gh-586.

spring-vault-core/src/main/java/org/springframework/vault/authentication/AuthenticationSteps.java

@@ -90,7 +90,7 @@ public class AuthenticationSteps {
 
 		Assert.notNull(token, "Vault token must not be null");
 
-		return new AuthenticationSteps(new SupplierStep<>(() -> token, AuthenticationSteps.HEAD));
+		return new AuthenticationSteps(new ScalarValueStep<>(token, AuthenticationSteps.HEAD));
 	}
 
 	/**
@@ -106,10 +106,26 @@ public class AuthenticationSteps {
 		return new AuthenticationSteps(new HttpRequestNode<>(request, AuthenticationSteps.HEAD));
 	}
 
+	/**
+	 * Start flow composition from a scalar {@code value}.
+	 * @param value the value to be used from this {@link Node}, must not be
+	 * {@literal null}.
+	 * @return the first {@link Node}.
+	 * @since 2.3
+	 */
+	public static <T> Node<T> fromValue(T value) {
+
+		Assert.notNull(value, "Value must not be null");
+
+		return new ScalarValueStep<>(value, AuthenticationSteps.HEAD);
+	}
+
 	/**
 	 * Start flow composition from a {@link Supplier}.
 	 * @param supplier supplier function that will produce the flow value, must not be
-	 * {@literal null}.
+	 * {@literal null}. Infrastructure components evaluating authentication steps may
+	 * inspect the given {@link java.util.function.Supplier} for an optimized approach to
+	 * obtain its value.
 	 * @return the first {@link Node}.
 	 */
 	public static <T> Node<T> fromSupplier(Supplier<T> supplier) {
@@ -642,6 +658,47 @@ public class AuthenticationSteps {
 
 	}
 
+	static final class ScalarValueStep<T> extends Node<T> implements PathAware {
+
+		private final T value;
+
+		private final Node<?> previous;
+
+		ScalarValueStep(T value, Node<?> previous) {
+			this.value = value;
+			this.previous = previous;
+		}
+
+		@Override
+		public String toString() {
+			return "Value: " + this.value.toString();
+		}
+
+		public T get() {
+			return this.value;
+		}
+
+		public Node<?> getPrevious() {
+			return this.previous;
+		}
+
+		@Override
+		public boolean equals(Object o) {
+			if (this == o)
+				return true;
+			if (!(o instanceof ScalarValueStep))
+				return false;
+			ScalarValueStep<?> that = (ScalarValueStep<?>) o;
+			return this.value.equals(that.value) && this.previous.equals(that.previous);
+		}
+
+		@Override
+		public int hashCode() {
+			return Objects.hash(this.value, this.previous);
+		}
+
+	}
+
 	static final class SupplierStep<T> extends Node<T> implements PathAware {
 
 		private final Supplier<T> supplier;

spring-vault-core/src/main/java/org/springframework/vault/authentication/AuthenticationStepsExecutor.java

@@ -29,6 +29,7 @@ import org.springframework.vault.authentication.AuthenticationSteps.MapStep;
 import org.springframework.vault.authentication.AuthenticationSteps.Node;
 import org.springframework.vault.authentication.AuthenticationSteps.OnNextStep;
 import org.springframework.vault.authentication.AuthenticationSteps.Pair;
+import org.springframework.vault.authentication.AuthenticationSteps.ScalarValueStep;
 import org.springframework.vault.authentication.AuthenticationSteps.SupplierStep;
 import org.springframework.vault.authentication.AuthenticationSteps.ZipStep;
 import org.springframework.vault.client.VaultResponses;
@@ -118,6 +119,10 @@ public class AuthenticationStepsExecutor implements ClientAuthentication {
 					state = doOnNext((OnNextStep<Object>) o, state);
 				}
 
+				if (o instanceof ScalarValueStep<?>) {
+					state = doScalarValueStep((ScalarValueStep<Object>) o);
+				}
+
 				if (o instanceof SupplierStep<?>) {
 					state = doSupplierStep((SupplierStep<Object>) o);
 				}
@@ -139,6 +144,10 @@ public class AuthenticationStepsExecutor implements ClientAuthentication {
 		return state;
 	}
 
+	private static Object doScalarValueStep(ScalarValueStep<Object> scalarValueStep) {
+		return scalarValueStep.get();
+	}
+
 	private static Object doSupplierStep(SupplierStep<Object> supplierStep) {
 		return supplierStep.get();
 	}

spring-vault-core/src/main/java/org/springframework/vault/authentication/AuthenticationStepsOperator.java

@@ -15,13 +15,19 @@
  */
 package org.springframework.vault.authentication;
 
+import java.io.IOException;
 import java.util.List;
 import java.util.Map.Entry;
+import java.util.function.Supplier;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
 import org.springframework.http.HttpEntity;
 import org.springframework.util.Assert;
 import org.springframework.vault.VaultException;
@@ -31,6 +37,7 @@ import org.springframework.vault.authentication.AuthenticationSteps.MapStep;
 import org.springframework.vault.authentication.AuthenticationSteps.Node;
 import org.springframework.vault.authentication.AuthenticationSteps.OnNextStep;
 import org.springframework.vault.authentication.AuthenticationSteps.Pair;
+import org.springframework.vault.authentication.AuthenticationSteps.ScalarValueStep;
 import org.springframework.vault.authentication.AuthenticationSteps.SupplierStep;
 import org.springframework.vault.authentication.AuthenticationSteps.ZipStep;
 import org.springframework.vault.support.VaultResponse;
@@ -45,6 +52,13 @@ import org.springframework.web.reactive.function.client.WebClient.RequestBodySpe
  * This class uses {@link WebClient} for non-blocking and reactive HTTP access. The
  * {@link AuthenticationSteps authentication flow} is materialized as reactive sequence
  * postponing execution until {@link Mono#subscribe() subscription}.
+ * <p>
+ * {@link Supplier Supplier} instances are inspected for their type.
+ * {@link ResourceCredentialSupplier} instances are loaded through
+ * {@link DataBufferUtils#read(org.springframework.core.io.Resource, org.springframework.core.io.buffer.DataBufferFactory, int)
+ * DataBufferUtils} to use non-blocking I/O for file access. {@link Supplier#get() Calls}
+ * to generic supplier types are offloaded to a {@link Schedulers#boundedElastic()
+ * scheduler} to avoid blocking calls on reactive worker/eventloop threads.
  *
  * @author Mark Paluch
  * @since 2.0
@@ -58,6 +72,8 @@ public class AuthenticationStepsOperator implements VaultTokenSupplier {
 
 	private final WebClient webClient;
 
+	private final DataBufferFactory factory = new DefaultDataBufferFactory();
+
 	/**
 	 * Create a new {@link AuthenticationStepsOperator} given {@link AuthenticationSteps}
 	 * and {@link WebClient}.
@@ -98,6 +114,7 @@ public class AuthenticationStepsOperator implements VaultTokenSupplier {
 		}).onErrorMap(t -> new VaultLoginException("Cannot retrieve VaultToken from authentication chain", t));
 	}
 
+	@SuppressWarnings("unchecked")
 	private Mono<Object> createMono(Iterable<Node<?>> steps) {
 
 		Mono<Object> state = Mono.just(Undefinded.INSTANCE);
@@ -125,8 +142,12 @@ public class AuthenticationStepsOperator implements VaultTokenSupplier {
 				state = state.doOnNext(stateObject -> doOnNext((OnNextStep<Object>) o, stateObject));
 			}
 
+			if (o instanceof ScalarValueStep<?>) {
+				state = state.map(stateObject -> doScalarValueStep((ScalarValueStep<Object>) o));
+			}
+
 			if (o instanceof SupplierStep<?>) {
-				state = state.map(stateObject -> doSupplierStep((SupplierStep<Object>) o));
+				state = state.flatMap(stateObject -> doSupplierStepLater((SupplierStep<Object>) o));
 			}
 
 			if (logger.isDebugEnabled()) {
@@ -136,8 +157,29 @@ public class AuthenticationStepsOperator implements VaultTokenSupplier {
 		return state;
 	}
 
-	private static Object doSupplierStep(SupplierStep<Object> supplierStep) {
-		return supplierStep.get();
+	private static Object doScalarValueStep(ScalarValueStep<Object> scalarValueStep) {
+		return scalarValueStep.get();
+	}
+
+	private Mono<Object> doSupplierStepLater(SupplierStep<Object> supplierStep) {
+
+		Supplier<?> supplier = supplierStep.getSupplier();
+
+		if (!(supplier instanceof ResourceCredentialSupplier)) {
+			return Mono.fromSupplier(supplierStep.getSupplier()).subscribeOn(Schedulers.boundedElastic());
+		}
+
+		ResourceCredentialSupplier resourceSupplier = (ResourceCredentialSupplier) supplier;
+
+		return DataBufferUtils.join(DataBufferUtils.read(resourceSupplier.getResource(), this.factory, 4096))
+				.map(dataBuffer -> {
+					String result = dataBuffer.toString(ResourceCredentialSupplier.CHARSET);
+					DataBufferUtils.release(dataBuffer);
+					return (Object) result;
+				}).onErrorMap(IOException.class,
+						e -> new VaultException(
+								String.format("Credential retrieval from %s failed", resourceSupplier.getResource()),
+								e));
 	}
 
 	private static Object doMapStep(MapStep<Object, Object> o, Object state) {

spring-vault-core/src/main/java/org/springframework/vault/authentication/CredentialSupplier.java

@@ -38,8 +38,7 @@ public interface CredentialSupplier extends Supplier<String> {
 
 	/**
 	 * Retrieve a cached {@link CredentialSupplier} that obtains the credential early and
-	 * reuses the token for each {@link #get()} call. This is useful to prevent I/O
-	 * operations in e.g. reactive usage.
+	 * reuses the token for each {@link #get()} call.
 	 * <p>
 	 * Reusing a cached token can lead to authentication failures if the credential
 	 * expires.

spring-vault-core/src/main/java/org/springframework/vault/authentication/ResourceCredentialSupplier.java

@@ -18,6 +18,7 @@ package org.springframework.vault.authentication;
 import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
+import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
 
 import org.springframework.core.io.FileSystemResource;
@@ -35,6 +36,8 @@ import org.springframework.vault.VaultException;
  */
 public class ResourceCredentialSupplier implements CredentialSupplier {
 
+	static final Charset CHARSET = StandardCharsets.US_ASCII;
+
 	private final Resource resource;
 
 	/**
@@ -74,13 +77,17 @@ public class ResourceCredentialSupplier implements CredentialSupplier {
 	public String get() {
 
 		try {
-			return new String(readToken(this.resource), StandardCharsets.US_ASCII);
+			return new String(readToken(this.resource), CHARSET);
 		}
 		catch (IOException e) {
 			throw new VaultException(String.format("Credential retrieval from %s failed", this.resource), e);
 		}
 	}
 
+	Resource getResource() {
+		return this.resource;
+	}
+
 	/**
 	 * Read the token from {@link Resource}.
 	 * @param resource the resource to read from, must not be {@literal null}.

spring-vault-core/src/test/java/org/springframework/vault/authentication/AuthenticationStepsExecutorUnitTests.java

@@ -20,6 +20,8 @@ import java.net.URI;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.core.io.ByteArrayResource;
+import org.springframework.core.io.ClassPathResource;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.MediaType;
@@ -80,6 +82,26 @@ class AuthenticationStepsExecutorUnitTests {
 		assertThat(login(steps)).isEqualTo(VaultToken.of("my-token"));
 	}
 
+	@Test
+	void fileResourceCredentialSupplierShouldBeLoaded() {
+
+		AuthenticationSteps steps = AuthenticationSteps
+				.fromSupplier(new ResourceCredentialSupplier(new ClassPathResource("kube-jwt-token")))
+				.login(VaultToken::of);
+
+		assertThat(login(steps).getToken()).startsWith("eyJhbGciOiJSUz");
+	}
+
+	@Test
+	void inputStreamResourceCredentialSupplierShouldBeLoaded() {
+
+		AuthenticationSteps steps = AuthenticationSteps
+				.fromSupplier(new ResourceCredentialSupplier(new ByteArrayResource("eyJhbGciOiJSUz".getBytes())))
+				.login(VaultToken::of);
+
+		assertThat(login(steps).getToken()).startsWith("eyJhbGciOiJSUz");
+	}
+
 	@Test
 	void justLoginRequestShouldLogin() {
 

spring-vault-core/src/test/java/org/springframework/vault/authentication/AuthenticationStepsOperatorUnitTests.java

@@ -15,11 +15,15 @@
  */
 package org.springframework.vault.authentication;
 
-import org.junit.jupiter.api.BeforeEach;
+import java.io.IOException;
+import java.io.InputStream;
+
 import org.junit.jupiter.api.Test;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
+import org.springframework.core.io.ByteArrayResource;
+import org.springframework.core.io.ClassPathResource;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
@@ -27,11 +31,13 @@ import org.springframework.http.client.reactive.ClientHttpConnector;
 import org.springframework.http.client.reactive.ClientHttpRequest;
 import org.springframework.mock.http.client.reactive.MockClientHttpRequest;
 import org.springframework.mock.http.client.reactive.MockClientHttpResponse;
+import org.springframework.vault.VaultException;
 import org.springframework.vault.authentication.AuthenticationSteps.Node;
 import org.springframework.vault.support.VaultResponse;
 import org.springframework.vault.support.VaultToken;
 import org.springframework.web.reactive.function.client.WebClient;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.springframework.vault.authentication.AuthenticationSteps.HttpRequestBuilder.post;
 
 /**
@@ -41,10 +47,6 @@ import static org.springframework.vault.authentication.AuthenticationSteps.HttpR
  */
 class AuthenticationStepsOperatorUnitTests {
 
-	@BeforeEach
-	void before() {
-	}
-
 	@Test
 	void justTokenShouldLogin() {
 
@@ -65,6 +67,58 @@ class AuthenticationStepsOperatorUnitTests {
 				.verifyComplete();
 	}
 
+	@Test
+	void fileResourceCredentialSupplierShouldBeLoaded() {
+
+		AuthenticationSteps steps = AuthenticationSteps
+				.fromSupplier(new ResourceCredentialSupplier(new ClassPathResource("kube-jwt-token")))
+				.login(VaultToken::of);
+
+		login(steps).as(StepVerifier::create) //
+				.consumeNextWith(actual -> {
+					assertThat(actual.getToken()).startsWith("eyJhbGciOiJSUz");
+				}).verifyComplete();
+	}
+
+	@Test
+	void absentFileResourceCredentialSupplierShouldFail() {
+
+		AuthenticationSteps steps = AuthenticationSteps
+				.fromSupplier(new ResourceCredentialSupplier(new ByteArrayResource("eyJhbGciOiJSUz".getBytes()) {
+					@Override
+					public InputStream getInputStream() throws IOException {
+						throw new IOException("Oops!");
+					}
+				})).login(VaultToken::of);
+
+		login(steps).as(StepVerifier::create) //
+				.verifyError(VaultException.class);
+	}
+
+	@Test
+	void inputStreamResourceCredentialSupplierShouldBeLoaded() {
+
+		AuthenticationSteps steps = AuthenticationSteps
+				.fromSupplier(new ResourceCredentialSupplier(new ByteArrayResource("eyJhbGciOiJSUz".getBytes())))
+				.login(VaultToken::of);
+
+		login(steps).as(StepVerifier::create) //
+				.consumeNextWith(actual -> {
+					assertThat(actual.getToken()).startsWith("eyJhbGciOiJSUz");
+				}).verifyComplete();
+	}
+
+	@Test
+	void anyCredentialSupplierShouldBeLoaded() {
+
+		AuthenticationSteps steps = AuthenticationSteps.fromSupplier(() -> "eyJhbGciOiJSUz").login(VaultToken::of);
+
+		login(steps).as(StepVerifier::create) //
+				.consumeNextWith(actual -> {
+					assertThat(actual.getToken()).startsWith("eyJhbGciOiJSUz");
+				}).verifyComplete();
+	}
+
 	@Test
 	void justLoginRequestShouldLogin() {
 

src/main/asciidoc/new-features.adoc

@@ -9,6 +9,7 @@
 * `VaultKeyValueMetadataOperations` for Key-Value metadata interaction.
 * Support for `transform` backend (Enterprise Feature).
 * Documentation of <<vault.core.secret-engines,how to use Vault secret backends>>.
+* Login credentials for Kubernetes and PCF authentication are reloaded for each login attempt.
 
 [[new-features.2-2-0]]
 === What's new in Spring Vault 2.2