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

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

@@ -0,0 +1,62 @@
+package org.atsign.client.api;
+
+import lombok.*;
+
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * The identity a connection authenticates as (its {@code atSign}, {@code keys} and {@code config})
+ * together with the single-use challenge from the {@code from:} that is issued as the first command
+ * once the connection is ready.
+ *
+ * <p>
+ * The context is created by the builder (see
+ * {@code AtCommandExecutors#createCommandExecutor}) and closed over by the {@code onReady}
+ * consumers it wires, so the {@code from:} sender can retain the challenge and the authentication
+ * that follows on the same connection can reuse it rather than issuing a second {@code from:}.
+ * The command executor itself is pure transport and knows nothing about this context.
+ *
+ * <p>
+ * The identity fields are fixed for the life of the context. The challenge is per-connection state:
+ * {@link #setChallenge(String) retained} when the initial {@code from:} completes and
+ * {@link #consumeChallenge() consumed} at most once (the server's {@code from:} challenge is
+ * single-use). On reconnect the ready sequence re-runs, so a fresh challenge overwrites any
+ * previous
+ * one before it is consumed.
+ */
+@Value
+public class AtCommandExecutorContext {
+
+  AtSign atSign;
+
+  AtKeys keys;
+
+  Map<String, Object> config;
+
+  @Getter(AccessLevel.NONE)
+  @EqualsAndHashCode.Exclude
+  @ToString.Exclude
+  AtomicReference<String> challenge = new AtomicReference<>();
+
+  /**
+   * Retains the challenge returned by the initial {@code from:}, so the authentication that follows
+   * on the same connection can reuse it.
+   *
+   * @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 it; anything else on the
+   * same connection 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);
+  }
+}

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

@@ -1,7 +1,17 @@
 package org.atsign.client.impl;
 
 
-import static org.atsign.client.impl.common.Preconditions.checkNotNull;
+import lombok.Builder;
+import lombok.extern.slf4j.Slf4j;
+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.commands.AuthenticationCommands;
+import org.atsign.client.impl.common.ReconnectStrategy;
+import org.atsign.client.impl.common.SimpleReconnectStrategy;
+import org.atsign.client.impl.exceptions.AtException;
+import org.atsign.client.impl.netty.NettyAtCommandExecutor;
 
 import java.io.InputStream;
 import java.net.URL;
@@ -12,17 +22,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
-import org.atsign.client.api.AtCommandExecutor;
-import org.atsign.client.api.AtKeys;
-import org.atsign.client.api.AtSign;
-import org.atsign.client.impl.commands.AuthenticationCommands;
-import org.atsign.client.impl.common.ReconnectStrategy;
-import org.atsign.client.impl.common.SimpleReconnectStrategy;
-import org.atsign.client.impl.exceptions.AtException;
-import org.atsign.client.impl.netty.NettyAtCommandExecutor;
-
-import lombok.Builder;
-import lombok.extern.slf4j.Slf4j;
+import static org.atsign.client.impl.common.Preconditions.checkNotNull;
 
 /**
  * Utility methods / builders for instantiating {@link AtCommandExecutor} implementations
@@ -42,8 +42,10 @@
  *
  * <b>NOTE:</b> If the url is prefixed with proxy (e.g. proxy:host:port) then the builder
  * will automatically attempt to connect to an At Server at host:port.
- * <b>NOTE:</b> If atSign and keys are provided then the builder
- * will automatically configure the {@link AtCommandExecutor} to authenticate with PKAM.
+ * <b>NOTE:</b> If an atSign is provided then the builder issues {@code from:@atSign} as the first
+ * command once connected (so proxies / gateways can route the connection); if keys are also
+ * provided it then authenticates the {@link AtCommandExecutor} with PKAM, reusing that
+ * {@code from:}'s challenge.
  * <b>NOTE:</b> If reconnect is not set then the builder will default to a
  * {@link SimpleReconnectStrategy}
  */
@@ -69,14 +71,18 @@ public static AtCommandExecutor createCommandExecutor(String url,
       checkNotNull(atSign, "atSign not set");
     }
 
+    // the context is closed over by the onReady consumers the builder wires (see createOnReady); the
+    // command executor itself stays pure transport and knows nothing about it
+    AtCommandExecutorContext context = new AtCommandExecutorContext(atSign, keys, createClientConfig(config));
+
     return NettyAtCommandExecutor.builder()
         .endpoint(AtEndpointSuppliers.builder().url(url).atSign(atSign).build())
         .isVerbose(isVerbose)
         .timeoutMillis(defaultIfNotSet(timeoutMillis, DEFAULT_TIMEOUT_MILLIS))
         .awaitReadyMillis(defaultIfNotSet(awaitReadyMillis, DEFAULT_TIMEOUT_MILLIS))
         .reconnect(defaultIfNotSet(reconnect, SimpleReconnectStrategy.builder().build()))
         .queueLimit(queueLimit)
-        .onReady(defaultIfNotSet(onReady, createOnReady(atSign, keys, createClientConfig(config))))
+        .onReady(defaultIfNotSet(onReady, createOnReady(context)))
         .build();
   }
 
@@ -128,14 +134,21 @@ 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) {
-    Consumer<AtCommandExecutor> onReady;
-    if (atSign != null && keys != null) {
-      onReady = AuthenticationCommands.pkamAuthenticator(atSign, keys, config);
-    } else {
-      onReady = c -> {
+  /**
+   * The default protocol for a newly-ready connection. A connection with no atSign (e.g. one talking
+   * to the atDirectory / root server) sends nothing. Every connection that has an atSign issues
+   * {@code from:@atSign} first so that proxies / gateways can route it; if keys are also present it
+   * then authenticates with PKAM, reusing the challenge from that initial {@code from:}.
+   */
+  private static Consumer<AtCommandExecutor> createOnReady(AtCommandExecutorContext context) {
+    if (context.getAtSign() == null) {
+      return c -> {
       };
     }
+    Consumer<AtCommandExecutor> onReady = AuthenticationCommands.sendFrom(context);
+    if (context.getKeys() != null) {
+      onReady = onReady.andThen(AuthenticationCommands.pkamAuthenticator(context));
+    }
     return onReady;
   }
 

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

@@ -4,16 +4,16 @@
 import java.util.concurrent.TimeUnit;
 
 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.AtEndpointSuppliers;
+import org.atsign.client.impl.AtCommandExecutors;
+import org.atsign.client.impl.AtCommandExecutors.AtCommandExecutorBuilder;
 import org.atsign.client.impl.commands.AuthenticationCommands;
+import org.atsign.client.impl.common.ReconnectStrategy;
 import org.atsign.client.impl.common.SimpleReconnectStrategy;
 import org.atsign.client.impl.exceptions.AtClientConfigException;
 import org.atsign.client.impl.exceptions.AtException;
-import org.atsign.client.impl.netty.NettyAtCommandExecutor;
-import org.atsign.client.impl.netty.NettyAtCommandExecutor.NettyAtCommandExecutorBuilder;
 import org.atsign.client.impl.util.KeysUtils;
 
 import picocli.CommandLine.Option;
@@ -87,25 +87,69 @@ protected static String ensureNotNull(String value, String defaultValue) {
   }
 
   protected AtCommandExecutor createConnection(String rootUrl, AtSign atSign, int retries) throws AtException {
-    return createCommandExecutorBuilder(rootUrl, atSign, retries, verbose).build();
+    // no keys: the builder wires an onReady that issues from:@atSign only (so proxies can route it)
+    return connectionBuilder(rootUrl, retries, verbose).atSign(atSign).build();
   }
 
   protected AtCommandExecutor createAuthenticatedConnection(String rootUrl, AtSign atSign, int retries)
       throws AtException {
-    return createCommandExecutorBuilder(rootUrl, atSign, retries, verbose)
-        .onReady(AuthenticationCommands.pkamAuthenticator(atSign, getKeys(), null))
+    // keys present: the builder wires from:@atSign followed by PKAM (reusing the from: challenge)
+    return connectionBuilder(rootUrl, retries, verbose).atSign(atSign).keys(getKeys()).build();
+  }
+
+  /**
+   * A connection that issues from:@atSign on connect (so proxies / gateways can route it), wiring the
+   * {@code from:} over the given {@code context} so the challenge it retains is the one an imperative
+   * flow driving the connection (e.g. onboarding: scan, then CRAM, then PKAM) later consumes. No
+   * on-ready authentication is wired — the caller authenticates imperatively.
+   *
+   * <p>
+   * Reconnect is disabled: because the imperative auth reuses the connect-time {@code from:}
+   * challenge, a mid-flow drop must abort (the one-shot flow is then rerun) rather than reconnect and
+   * authenticate on the new connection with a challenge the previous connection issued — the new
+   * connection's server would reject it. (The challenge is not cleared on close, so reconnect-and-
+   * reuse would be stale.)
+   */
+  protected AtCommandExecutor createConnectionSendingFrom(AtCommandExecutorContext context) throws AtException {
+    return AtCommandExecutors.builder()
+        .url(rootUrl)
+        .atSign(context.getAtSign())
+        .onReady(AuthenticationCommands.sendFrom(context))
+        .reconnect(ReconnectStrategy.NONE)
+        .isVerbose(verbose)
         .build();
   }
 
-  private static NettyAtCommandExecutorBuilder createCommandExecutorBuilder(String rootUrl, AtSign atSign, int retries,
-                                                                            boolean verbose) {
-    AtEndpointSupplier endpoint = AtEndpointSuppliers.builder().url(rootUrl).atSign(atSign).build();
+  /**
+   * A connection that does NOT issue {@code from:} on connect — its initial {@code from:} is issued
+   * by the caller's own first command — for flows that authenticate imperatively and must control the
+   * timing (e.g. a pending-retry loop), where a connect-time {@code from:} challenge could go stale
+   * before it is used. No onReady {@code from:} is wired, so the caller's first command (its
+   * authentication) establishes the atSign for proxies / gateways.
+   */
+  protected AtCommandExecutor createConnectionDeferringFrom(String rootUrl, AtSign atSign, int retries)
+      throws AtException {
+    return connectionBuilder(rootUrl, retries, verbose).atSign(atSign).onReady(executor -> {
+    }).build();
+  }
+
+  /**
+   * Creates an {@link AtCommandExecutorContext} carrying this CLI's atSign and the given {@code keys}
+   * (the identity being onboarded), to be passed to
+   * {@link #createConnectionSendingFrom(AtCommandExecutorContext)} and threaded into the
+   * imperative onboarding flow, which reuses the connection's {@code from:} challenge.
+   */
+  protected AtCommandExecutorContext newConnectionContext(AtKeys keys) {
+    return new AtCommandExecutorContext(atSign, keys, AtCommandExecutors.createClientConfig(null));
+  }
+
+  private static AtCommandExecutorBuilder connectionBuilder(String rootUrl, int retries, boolean verbose) {
     SimpleReconnectStrategy reconnect = SimpleReconnectStrategy.builder()
         .maxReconnectRetries(retries)
         .reconnectPauseMillis(TimeUnit.SECONDS.toMillis(2))
         .build();
-    return NettyAtCommandExecutor.builder()
-        .endpoint(endpoint)
+    return AtCommandExecutors.builder()
+        .url(rootUrl)
         .reconnect(reconnect)
         .isVerbose(verbose);
   }

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

@@ -14,6 +14,7 @@
 
 import org.atsign.client.api.AtKeys;
 import org.atsign.client.api.AtCommandExecutor;
+import org.atsign.client.api.AtCommandExecutorContext;
 import org.atsign.client.impl.commands.EnrollCommands;
 import org.atsign.client.impl.commands.ScanCommands;
 import org.atsign.client.impl.common.EnrollmentId;
@@ -180,24 +181,23 @@ public Activate addNamespace(String namespace, String accessControl) {
   }
 
   public EnrollmentId onboard() throws Exception {
-    try (AtCommandExecutor executor = createConnection(rootUrl, atSign, connectionRetries)) {
-      return onboard(executor);
+    AtCommandExecutorContext context = newConnectionContext(generateAtKeys(true));
+    try (AtCommandExecutor executor = createConnectionSendingFrom(context)) {
+      return onboard(executor, context);
     }
   }
 
-  public EnrollmentId onboard(AtCommandExecutor executor) throws Exception {
+  public EnrollmentId onboard(AtCommandExecutor executor, AtCommandExecutorContext context) throws Exception {
     File file = getAtKeysFile(keysFile, atSign);
     if (!overwriteKeysFile) {
       checkNotExists(file);
     }
-    AtKeys keys = generateAtKeys(true);
-    keys = EnrollCommands.onboard(executor,
-                                  atSign,
-                                  keys,
-                                  cramSecret,
-                                  ensureNotNull(appName, DEFAULT_FIRST_APP),
-                                  ensureNotNull(deviceName, DEFAULT_FIRST_DEVICE),
-                                  deleteCramKey);
+    AtKeys keys = EnrollCommands.onboard(executor,
+                                         context,
+                                         cramSecret,
+                                         ensureNotNull(appName, DEFAULT_FIRST_APP),
+                                         ensureNotNull(deviceName, DEFAULT_FIRST_DEVICE),
+                                         deleteCramKey);
     saveKeys(keys, file);
     return enrollmentId;
   }
@@ -292,7 +292,9 @@ public EnrollmentId enroll(AtCommandExecutor executor) throws Exception {
   }
 
   public void complete() throws Exception {
-    try (AtCommandExecutor executor = createConnection(rootUrl, atSign, connectionRetries)) {
+    // complete authenticates imperatively (and retries on "pending"), so it issues its own from:
+    // rather than reusing a connect-time challenge that could go stale before the retried PKAM
+    try (AtCommandExecutor executor = createConnectionDeferringFrom(rootUrl, atSign, connectionRetries)) {
       complete(executor);
     }
   }
@@ -304,7 +306,8 @@ public void complete(AtCommandExecutor executor) throws Exception {
   }
 
   public void complete(int retries, long sleepDuration, TimeUnit sleepUnit) throws Exception {
-    try (AtCommandExecutor executor = createConnection(rootUrl, atSign, connectionRetries)) {
+    // see complete(): imperative, retried PKAM issues its own from:
+    try (AtCommandExecutor executor = createConnectionDeferringFrom(rootUrl, atSign, connectionRetries)) {
       complete(executor, retries, sleepDuration, sleepUnit);
     }
   }