feat(auth): send from: on connect and reuse its challenge

at_client/src/main/java/org/atsign/client/api/AtCommandExecutor.java

@@ -83,4 +83,18 @@ default CompletableFuture<Void> send(String command, Consumer<String> consumer)
    * @return this (to allow chaining / fluent style invocation)
    */
   AtCommandExecutor onReady(Consumer<AtCommandExecutor> consumer);
+
+  /**
+   * Returns this executor's {@link AtCommandExecutorContext authentication context}: the identity it
+   * authenticates as together with the single-use challenge from the {@code from:} it issued as its
+   * first command once ready. The authentication commands reuse that challenge rather than sending a
+   * second {@code from:}, and read the identity from the same context. Never {@code null} — an
+   * executor that was not configured with an atSign (and so issued no initial {@code from:}) returns
+   * {@link AtCommandExecutorContext#EMPTY}, whose challenge is always {@code null}.
+   *
+   * @return this executor's authentication context (never {@code null})
+   */
+  default AtCommandExecutorContext getContext() {
+    return AtCommandExecutorContext.EMPTY;
+  }
 }

at_client/src/main/java/org/atsign/client/api/AtCommandExecutorContext.java

@@ -0,0 +1,100 @@
+package org.atsign.client.api;
+
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * The authentication context an {@link AtCommandExecutor} carries for the life of a connection: the
+ * identity it authenticates as ({@link #getAtSign() atSign}, {@link #getKeys() keys},
+ * {@link #getConfig() config}) together with the single-use challenge from the {@code from:} the
+ * executor issues as its first command once ready.
+ *
+ * <p>
+ * Grouping these behind {@link AtCommandExecutor#getContext()} lets the authentication commands
+ * take
+ * exactly what they need from one accessor rather than the executor interface growing a separate
+ * method per field.
+ *
+ * <p>
+ * The identity fields are fixed for the life of the context. The challenge is session state: the
+ * executor {@link #setChallenge(String) retains} it once the initial {@code from:} completes,
+ * callers {@link #consumeChallenge() consume} it at most once (the server's {@code from:} challenge
+ * is single-use), and the executor {@link #clearChallenge() clears} it on disconnect — a challenge
+ * is only valid for the server session that issued it.
+ */
+public class AtCommandExecutorContext {
+
+  /**
+   * A context with no identity and no challenge, returned by executors that were not configured with
+   * an atSign (and so issue no initial {@code from:}). {@link #consumeChallenge()} always yields
+   * {@code null}, so authentication falls back to sending its own {@code from:}.
+   */
+  public static final AtCommandExecutorContext EMPTY = new AtCommandExecutorContext(null, null, null);
+
+  private final AtSign atSign;
+
+  private final AtKeys keys;
+
+  private final Map<String, Object> config;
+
+  private final AtomicReference<String> challenge = new AtomicReference<>();
+
+  public AtCommandExecutorContext(AtSign atSign, AtKeys keys, Map<String, Object> config) {
+    this.atSign = atSign;
+    this.keys = keys;
+    this.config = config;
+  }
+
+  /**
+   * @return the atSign this executor authenticates as, or {@code null} if none was configured
+   */
+  public AtSign getAtSign() {
+    return atSign;
+  }
+
+  /**
+   * @return the keys this executor authenticates with, or {@code null} if none were configured (e.g.
+   *         an onboarding executor whose keys are generated mid-flow and supplied to the command
+   *         directly)
+   */
+  public AtKeys getKeys() {
+    return keys;
+  }
+
+  /**
+   * @return the config sent in the {@code from:} command, or {@code null} if none was configured
+   */
+  public Map<String, Object> getConfig() {
+    return config;
+  }
+
+  /**
+   * Retains the challenge returned by the initial {@code from:}. Called by the executor once that
+   * command completes on the ready thread.
+   *
+   * @param challenge the challenge from the server's {@code from:} response
+   */
+  public void setChallenge(String challenge) {
+    this.challenge.set(challenge);
+  }
+
+  /**
+   * Returns the retained {@code from:} challenge and clears it, so it is consumed at most once.
+   * Whichever authentication (CRAM or PKAM) sends its digest first consumes the challenge; a second
+   * authentication on the same connection (e.g. onboarding, which does CRAM then PKAM) gets
+   * {@code null} and must issue its own {@code from:}.
+   *
+   * @return the retained challenge, or {@code null} if none is available
+   */
+  public String consumeChallenge() {
+    return challenge.getAndSet(null);
+  }
+
+  /**
+   * Clears any retained challenge. Called by the executor on disconnect — a challenge is only valid
+   * for the server session that issued it, so it must not survive into the next connection.
+   */
+  public void clearChallenge() {
+    challenge.set(null);
+  }
+}

at_client/src/main/java/org/atsign/client/impl/AtClientImpl.java

@@ -122,6 +122,8 @@ public void startMonitor() {
   @Override
   public void stopMonitor() {
     isMonitoring.compareAndSet(true, false);
+    // this client owns its atSign/keys/config, so authenticate with those directly rather than via
+    // the executor's context (the executor is injected and may not carry this client's identity)
     executor.onReady(AuthenticationCommands.pkamAuthenticator(atSign, keys, config));
   }
 

at_client/src/main/java/org/atsign/client/impl/AtCommandExecutors.java

@@ -76,7 +76,10 @@ public static AtCommandExecutor createCommandExecutor(String url,
         .awaitReadyMillis(defaultIfNotSet(awaitReadyMillis, DEFAULT_TIMEOUT_MILLIS))
         .reconnect(defaultIfNotSet(reconnect, SimpleReconnectStrategy.builder().build()))
         .queueLimit(queueLimit)
-        .onReady(defaultIfNotSet(onReady, createOnReady(atSign, keys, createClientConfig(config))))
+        .atSign(atSign)
+        .keys(keys)
+        .clientConfig(createClientConfig(config))
+        .onReady(defaultIfNotSet(onReady, createOnReady(atSign, keys)))
         .build();
   }
 
@@ -128,10 +131,12 @@ public static Map<String, Object> createClientConfig(Map<String, Object> config)
     return result;
   }
 
-  private static Consumer<AtCommandExecutor> createOnReady(AtSign atSign, AtKeys keys, Map<String, Object> config) {
+  private static Consumer<AtCommandExecutor> createOnReady(AtSign atSign, AtKeys keys) {
     Consumer<AtCommandExecutor> onReady;
     if (atSign != null && keys != null) {
-      onReady = AuthenticationCommands.pkamAuthenticator(atSign, keys, config);
+      // the executor is built with this atSign/keys/config in its context, so the authenticator
+      // reads the identity (and reuses the initial from: challenge) from there
+      onReady = AuthenticationCommands.pkamAuthenticator();
     } else {
       onReady = c -> {
       };

at_client/src/main/java/org/atsign/client/impl/cli/AbstractCli.java

@@ -106,6 +106,7 @@ private static NettyAtCommandExecutorBuilder createCommandExecutorBuilder(String
         .build();
     return NettyAtCommandExecutor.builder()
         .endpoint(endpoint)
+        .atSign(atSign)
         .reconnect(reconnect)
         .isVerbose(verbose);
   }

at_client/src/main/java/org/atsign/client/impl/commands/AuthenticationCommands.java

@@ -15,6 +15,7 @@
 
 import org.atsign.client.api.AtKeys;
 import org.atsign.client.api.AtCommandExecutor;
+import org.atsign.client.api.AtCommandExecutorContext;
 import org.atsign.client.impl.util.EncryptionUtils;
 import org.atsign.client.impl.exceptions.AtException;
 import org.atsign.client.api.AtSign;
@@ -30,6 +31,31 @@ public static Consumer<AtCommandExecutor> pkamAuthenticator(AtSign atSign, AtKey
     return throwOnReadyException(executor -> authenticateWithPkam(executor, atSign, keys, config));
   }
 
+  /**
+   * Returns an onReady consumer that authenticates using the identity carried by the executor's
+   * {@link AtCommandExecutorContext context} — the atSign, keys and config it was built with. This
+   * is the standard client path, where that identity is fixed for the connection. Onboarding, whose
+   * keys are generated mid-flow, uses {@link #pkamAuthenticator(AtSign, AtKeys, Map)} with explicit
+   * keys instead.
+   *
+   * @return an onReady consumer performing PKAM authentication from the executor's context
+   */
+  public static Consumer<AtCommandExecutor> pkamAuthenticator() {
+    return throwOnReadyException(AuthenticationCommands::authenticateWithPkam);
+  }
+
+  /**
+   * Implements the protocol workflow / sequence for PKAM authentication, taking the atSign, keys and
+   * config from the executor's {@link AtCommandExecutorContext context}.
+   *
+   * @param executor The executor with which to send the commands; its context supplies the identity.
+   * @throws AtException If authentication fails.
+   */
+  public static void authenticateWithPkam(AtCommandExecutor executor) throws AtException {
+    AtCommandExecutorContext context = executor.getContext();
+    authenticateWithPkam(executor, context.getAtSign(), context.getKeys(), context.getConfig());
+  }
+
   /**
    * Implements the protocol workflow / sequence for PKAM authentication.
    *
@@ -59,10 +85,14 @@ public static void authenticateWithPkam(AtCommandExecutor executor,
       throws AtException {
     try {
 
+      // reuse the challenge from the initial from: if the executor already sent one, otherwise
       // send a from command and expect to receive a challenge
-      String fromCommand = CommandBuilders.fromCommandBuilder().atSign(atSign).config(config).build();
-      String fromResponse = executor.sendSync(fromCommand);
-      String challenge = matchDataStringNoWhitespace(throwExceptionIfError(fromResponse));
+      String challenge = consumeFromChallenge(executor);
+      if (challenge == null) {
+        String fromCommand = CommandBuilders.fromCommandBuilder().atSign(atSign).config(config).build();
+        String fromResponse = executor.sendSync(fromCommand);
+        challenge = matchDataStringNoWhitespace(throwExceptionIfError(fromResponse));
+      }
 
       // send a pkam command with the signed challenge
       String signature = EncryptionUtils.signSHA256RSA(challenge, keys.getApkamPrivateKey());
@@ -95,10 +125,14 @@ public static void authenticateWithCram(AtCommandExecutor executor, AtSign atSig
       throws AtException {
     try {
 
+      // reuse the challenge from the initial from: if the executor already sent one, otherwise
       // send a from command and expect to receive a challenge
-      String fromCommand = CommandBuilders.fromCommandBuilder().atSign(atSign).build();
-      String fromResponse = executor.sendSync(fromCommand);
-      String challenge = matchDataStringNoWhitespace(throwExceptionIfError(fromResponse));
+      String challenge = consumeFromChallenge(executor);
+      if (challenge == null) {
+        String fromCommand = CommandBuilders.fromCommandBuilder().atSign(atSign).build();
+        String fromResponse = executor.sendSync(fromCommand);
+        challenge = matchDataStringNoWhitespace(throwExceptionIfError(fromResponse));
+      }
 
       // send a cram command
       String cramDigest = createDigest(cramSecret, challenge);
@@ -124,5 +158,16 @@ private static String createDigest(String cramSecret, String challenge) throws A
     }
   }
 
-
+  /**
+   * Returns and clears the single-use {@code from:} challenge the executor retained after issuing its
+   * initial {@code from:}, or {@code null} if none is available. A {@code null} context is treated as
+   * "no retained challenge" so authentication falls back to sending its own {@code from:} — the same
+   * behaviour the interface's {@code getContext()} default (an EMPTY context) gives, kept null-safe
+   * so test doubles that leave {@code getContext()} unstubbed behave as they did before the context
+   * replaced the former {@code getFromChallenge()} accessor.
+   */
+  private static String consumeFromChallenge(AtCommandExecutor executor) {
+    AtCommandExecutorContext context = executor.getContext();
+    return context == null ? null : context.consumeChallenge();
+  }
 }

at_client/src/main/java/org/atsign/client/impl/netty/NettyAtCommandExecutor.java

@@ -2,12 +2,15 @@
 
 import static java.util.concurrent.CompletableFuture.failedFuture;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.atsign.client.impl.commands.DataResponses.matchDataStringNoWhitespace;
+import static org.atsign.client.impl.commands.ErrorResponses.throwExceptionIfError;
 import static org.atsign.client.impl.common.CommandElement.isPrompt;
 import static org.atsign.client.impl.common.Preconditions.checkNotNull;
 
 import java.io.IOException;
 import java.time.Clock;
 import java.util.Collection;
+import java.util.Map;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
@@ -18,7 +21,11 @@
 import javax.net.ssl.SSLException;
 
 import org.atsign.client.api.AtCommandExecutor;
+import org.atsign.client.api.AtCommandExecutorContext;
+import org.atsign.client.api.AtKeys;
+import org.atsign.client.api.AtSign;
 import org.atsign.client.impl.AtEndpointSupplier;
+import org.atsign.client.impl.commands.CommandBuilders;
 import org.atsign.client.impl.common.ReconnectStrategy;
 import org.atsign.client.impl.common.CommandElement;
 import org.atsign.client.impl.common.CommandQueue;
@@ -102,6 +109,8 @@ String toLowerCase() {
 
   private volatile long lastReadMillis;
 
+  private final AtCommandExecutorContext context;
+
   /**
    * Builder method for instantiating instances of a Netty based implementation of
    * {@link AtCommandExecutor}
@@ -135,9 +144,13 @@ protected NettyAtCommandExecutor(AtEndpointSupplier endpoint,
                                    Integer queueLimit,
                                    Long awaitReadyMillis,
                                    Boolean isVerbose,
+                                   AtSign atSign,
+                                   AtKeys keys,
+                                   Map<String, Object> clientConfig,
                                    Clock clock,
                                    Logger log)
       throws AtException {
+    this.context = new AtCommandExecutorContext(atSign, keys, clientConfig);
     this.endpointSupplier = checkNotNull(endpoint, "endpoint is not set");
     this.maxFrameLength = defaultIfUnset(maxFrameLength, DEFAULT_MAX_FRAME_LENGTH);
     this.reconnectStrategy = defaultIfNull(reconnect, ReconnectStrategy.NONE);
@@ -466,6 +479,7 @@ private Runnable createOnReadyRunnable() {
       threadFactory.markCurrentThreadOnReadyThread();
       try {
         readyingLock.lock();
+        sendFromIfRequired();
         onReadyConsumer.get().accept(this);
       } catch (AtOnReadyException e) {
         log.error("onReady exception", e);
@@ -484,6 +498,48 @@ private Runnable createOnReadyRunnable() {
     };
   }
 
+  /**
+   * Sends {@code from:@atSign} as the first command once the connection is ready, so the
+   * connection's atSign is established before any other verb (including a {@code scan} sent by
+   * an onReady consumer prior to authenticating). The challenge returned by the server is
+   * retained so that CRAM / PKAM authentication can reuse it instead of issuing a second
+   * {@code from:}. When no atSign was supplied to the builder this is a no-op and the context's
+   * {@link AtCommandExecutorContext#consumeChallenge() challenge} stays {@code null}.
+   *
+   * <p>
+   * Runs on the onReady thread, where {@link #sendSync(String)} is permitted. On reconnect
+   * the ready sequence re-runs, so the challenge is refreshed on each connect.
+   */
+  private void sendFromIfRequired() {
+    AtSign atSign = context.getAtSign();
+    if (atSign == null) {
+      return;
+    }
+    try {
+      String fromCommand = CommandBuilders.fromCommandBuilder()
+          .atSign(atSign)
+          .config(context.getConfig())
+          .build();
+      String fromResponse = sendSync(fromCommand);
+      String challenge = matchDataStringNoWhitespace(throwExceptionIfError(fromResponse));
+      context.setChallenge(challenge);
+    } catch (AtException | ExecutionException | InterruptedException | RuntimeException e) {
+      throw new AtOnReadyException("from command failed : " + e.getMessage(), e);
+    }
+  }
+
+  /**
+   * Returns this executor's authentication context — the identity it authenticates as plus the
+   * single-use challenge from the initial {@code from:}. The challenge is retained by
+   * {@link #sendFromIfRequired()} on the ready thread, consumed at most once by whichever
+   * authentication sends its digest first, and cleared on disconnect (see {@code channelInactive})
+   * so a caller can never consume a challenge from a connection that has since dropped.
+   */
+  @Override
+  public AtCommandExecutorContext getContext() {
+    return context;
+  }
+
   private void onResponse(String msg) {
     if (isVerbose) {
       log.info("RCVD: {}", msg);
@@ -522,6 +578,9 @@ public void exceptionCaught(ChannelHandlerContext context, Throwable cause) {
     @Override
     public void channelInactive(ChannelHandlerContext context) {
       channel = null;
+      // a from: challenge is only valid for the server session that just ended (this method's
+      // `context` parameter is the Netty ChannelHandlerContext, so qualify the executor's field)
+      NettyAtCommandExecutor.this.context.clearChallenge();
       if (status.get().isClosedOrClosing()) {
         log.debug("connection closed");
       } else {

at_client/src/test/java/org/atsign/client/impl/commands/TestExecutorBuilder.java

@@ -13,6 +13,7 @@
 import java.util.regex.Pattern;
 
 import org.atsign.client.api.AtCommandExecutor;
+import org.atsign.client.api.AtCommandExecutorContext;
 import org.atsign.client.impl.util.JsonUtils;
 import org.mockito.Mockito;
 import org.mockito.stubbing.Answer;
@@ -62,6 +63,9 @@ public TestExecutorBuilder stub(Pattern command, Exception ex) {
 
   public AtCommandExecutor build() throws ExecutionException, InterruptedException {
     AtCommandExecutor mock = Mockito.mock(AtCommandExecutor.class);
+    // an EMPTY context yields no challenge, so authentication sends its own from: (which the tests
+    // stub) rather than reusing an initial from: challenge this mock never issued
+    when(mock.getContext()).thenReturn(AtCommandExecutorContext.EMPTY);
     when(mock.sendSync(anyString())).thenAnswer((Answer<String>) invocation -> {
       String command = invocation.getArgument(0);
       for (Map.Entry<Pattern, Object> entry : mapping.entrySet()) {