private SaslOutcome getSaslOutcomeForErrorStatus(final int status) {
    final SaslOutcome saslOutcome;
    switch (status) {
        case HttpURLConnection.HTTP_BAD_REQUEST:
        case HttpURLConnection.HTTP_UNAUTHORIZED:
            // failed due to an authentication error
            saslOutcome = SaslOutcome.PN_SASL_AUTH;
            break;
        case HttpURLConnection.HTTP_INTERNAL_ERROR:
            // failed due to a system error
            saslOutcome = SaslOutcome.PN_SASL_SYS;
            break;
        case HttpURLConnection.HTTP_UNAVAILABLE:
            // failed due to a transient error
            saslOutcome = SaslOutcome.PN_SASL_TEMP;
            break;
        default:
            if (status >= 400 && status < 500) {
                // client error
                saslOutcome = SaslOutcome.PN_SASL_PERM;
            } else {
                saslOutcome = SaslOutcome.PN_SASL_TEMP;
            }
    }
    return saslOutcome;
}
 

@Override
public void process(Handler<Boolean> completionHandler) {
  if (sasl == null) {
    throw new IllegalStateException("Init was not called with the associated transport");
  }

  boolean done = false;
  String[] remoteMechanisms = sasl.getRemoteMechanisms();
  if (remoteMechanisms.length > 0) {
    String chosen = remoteMechanisms[0];
    if (ProtonSaslAnonymousImpl.MECH_NAME.equals(chosen)) {
      sasl.done(SaslOutcome.PN_SASL_OK);
      succeeded = true;
    } else {
      sasl.done(SaslOutcome.PN_SASL_AUTH);
    }
    done = true;
  }

  completionHandler.handle(done);
}
 

@Override
public void process(Handler<Boolean> processComplete) {
  boolean done = false;
  String[] remoteMechanisms = sasl.getRemoteMechanisms();
  if (remoteMechanisms.length > 0) {
    String chosenMech = remoteMechanisms[0];

    boolean success = false;
    if (PLAIN.equals(chosenMech)) {
      success = evaluatePlainResponse(sasl);
    }

    if (success) {
      succeeded = true;
      sasl.done(SaslOutcome.PN_SASL_OK);
      // Record any desired kind of auth detail in the connection attachments, just using a String here.
      protonConnection.attachments().set(AUTH_KEY, String.class, AUTH_VALUE);
    } else {
      sasl.done(SaslOutcome.PN_SASL_AUTH);
    }

    done = true;
  }

  processComplete.handle(done);
}
 

private void handleSaslFail(Sasl sasl) {
    StringBuilder message = new StringBuilder("Client failed to authenticate");
    if (mechanism != null) {
        message.append(" using SASL: ").append(mechanism.getName());
        if (mechanism.getAdditionalFailureInformation() != null) {
            message.append(" (").append(mechanism.getAdditionalFailureInformation()).append(")");
        }
    }

    SaslOutcome outcome = sasl.getOutcome();
    if (outcome.equals(SaslOutcome.PN_SASL_TEMP)) {
        message.append(", due to temporary system error.");
    }

    recordFailure(message.toString(), null, outcome.getCode());
}
 

@Test
public void testPeerSignalsAuthenticationFail() throws Exception {
    Mechanism mechanism = new TestSaslMechanism(INITIAL_RESPONSE);
    AmqpSaslAuthenticator authenticator = new AmqpSaslAuthenticator(mechanismName -> mechanism);

    authenticator.handleSaslMechanisms(sasl, transport);
    verifySaslMockReceived(sasl, INITIAL_RESPONSE);

    when(sasl.getState()).thenReturn(SaslState.PN_SASL_FAIL);
    when(sasl.getOutcome()).thenReturn(SaslOutcome.PN_SASL_AUTH);
    authenticator.handleSaslOutcome(sasl, transport);

    assertTrue(authenticator.isComplete());
    assertFalse(authenticator.wasSuccessful());
    assertNotNull(authenticator.getFailureCause());
    assertTrue(authenticator.getFailureCause().getMessage().contains("Client failed to authenticate"));
    assertFalse(authenticator.getFailureCause().getMessage().contains("due to temporary system error"));
}
 

@Test
public void testPeerSignalsAuthenticationSysTemp() throws Exception {
    Mechanism mechanism = new TestSaslMechanism(INITIAL_RESPONSE);
    AmqpSaslAuthenticator authenticator = new AmqpSaslAuthenticator(mechanismName -> mechanism);

    authenticator.handleSaslMechanisms(sasl, transport);
    verifySaslMockReceived(sasl, INITIAL_RESPONSE);

    when(sasl.getState()).thenReturn(SaslState.PN_SASL_FAIL);
    when(sasl.getOutcome()).thenReturn(SaslOutcome.PN_SASL_TEMP);
    authenticator.handleSaslOutcome(sasl, transport);

    assertTrue(authenticator.isComplete());
    assertFalse(authenticator.wasSuccessful());
    assertNotNull(authenticator.getFailureCause());
    assertTrue(authenticator.getFailureCause() instanceof ProviderConnectionSecuritySaslException);
    assertTrue(authenticator.getFailureCause().getMessage().contains("Client failed to authenticate"));
    assertTrue(authenticator.getFailureCause().getMessage().contains("due to temporary system error"));
}
 

/** 5.3.2 SASL Negotiation. ...challenge/response step can occur zero or more times*/
@Test
public void testOptionalChallengeResponseStepOmitted() throws Exception
{
    getClient().transport = Proton.transport();
    getServer().transport = Proton.transport();

    Sasl clientSasl = getClient().transport.sasl();
    clientSasl.client();
    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_NONE, clientSasl.getOutcome());

    Sasl serverSasl = getServer().transport.sasl();
    serverSasl.server();
    serverSasl.setMechanisms(TESTMECH1);
    assertEquals("Server should not yet know the remote's chosen mechanism.",
                 0,
                 serverSasl.getRemoteMechanisms().length);

    pumpClientToServer();
    pumpServerToClient();

    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_NONE, clientSasl.getOutcome());
    clientSasl.setMechanisms(TESTMECH1);

    pumpClientToServer();

    serverSasl.done(SaslOutcome.PN_SASL_OK);
    pumpServerToClient();

    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_OK, clientSasl.getOutcome());
}
 

/**
 *  5.3.3.6 Connection authentication failed due to an unspecified problem with the supplied credentials.
 */
@Test
public void testAuthenticationFails() throws Exception
{
    getClient().transport = Proton.transport();
    getServer().transport = Proton.transport();

    Sasl clientSasl = getClient().transport.sasl();
    clientSasl.client();
    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_NONE, clientSasl.getOutcome());

    Sasl serverSasl = getServer().transport.sasl();
    serverSasl.server();
    serverSasl.setMechanisms(TESTMECH1);

    pumpClientToServer();
    pumpServerToClient();

    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_NONE, clientSasl.getOutcome());
    clientSasl.setMechanisms(TESTMECH1);

    pumpClientToServer();

    serverSasl.done(SaslOutcome.PN_SASL_AUTH);
    pumpServerToClient();
    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_AUTH, clientSasl.getOutcome());

}
 

@Override
public void onSaslResponse(Sasl s, Transport t)
{
    byte[] serverReceivedResponseBytes = new byte[s.pending()];
    s.recv(serverReceivedResponseBytes, 0, serverReceivedResponseBytes.length);

    assertArrayEquals("Server should now know the client's response", RESPONSE_BYTES, serverReceivedResponseBytes);

    s.send(ADDITIONAL_DATA_BYTES, 0, ADDITIONAL_DATA_BYTES.length);
    s.done(SaslOutcome.PN_SASL_OK);

    assertFalse("Should not have already received response", responseReceived.getAndSet(true));
}
 

@Override
public void onSaslMechanisms(Sasl s, Transport t)
{
    assertArrayEquals("Client should now know the server's mechanisms.",
            new String[]{TESTMECH1, TESTMECH2}, s.getRemoteMechanisms());
    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_NONE, s.getOutcome());

    s.setMechanisms(TESTMECH1);
    s.send(INITIAL_RESPONSE_BYTES, 0, INITIAL_RESPONSE_BYTES.length);

    assertFalse("Should not have already received mechanisms", mechanismsReceived.getAndSet(true));
}
 

@Override
public void onSaslChallenge(Sasl s, Transport t)
{
    byte[] clientReceivedChallengeBytes = new byte[s.pending()];
    s.recv(clientReceivedChallengeBytes, 0, clientReceivedChallengeBytes.length);

    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_NONE, s.getOutcome());
    assertArrayEquals("Client should now know the server's challenge",
                      CHALLENGE_BYTES, clientReceivedChallengeBytes);

    s.send(RESPONSE_BYTES, 0, RESPONSE_BYTES.length);

    assertFalse("Should not have already received challenge", challengeReceived.getAndSet(true));
}
 

@Override
public void onSaslOutcome(Sasl s, Transport t)
{
    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_OK, s.getOutcome());

    byte[] clientReceivedAdditionalBytes = new byte[s.pending()];
    s.recv(clientReceivedAdditionalBytes, 0, clientReceivedAdditionalBytes.length);

    assertArrayEquals("Client should now know the server's outcome additional data", clientReceivedAdditionalBytes,
            clientReceivedAdditionalBytes);

    assertFalse("Should not have already received outcome", outcomeReceived.getAndSet(true));
}
 

@Override
public void process(Handler<Boolean> processComplete) {
  String[] remoteMechanisms = sasl.getRemoteMechanisms();
  if (remoteMechanisms.length > 0) {
    String chosenMech = remoteMechanisms[0];

    if (PLAIN.equals(chosenMech)) {
      Context context = Vertx.currentContext();

      byte[] response = new byte[sasl.pending()];
      sasl.recv(response, 0, response.length);

      // Signal process handling completed (with success/failure also in this case) only after the given delay.
      // The timer scheduling will use the same Context for the callback automatically in this case.
      context.owner().setTimer(completionDelay, x -> {
        if (passAuth) {
          succeeded = true;
          sasl.done(SaslOutcome.PN_SASL_OK);
        } else {
          sasl.done(SaslOutcome.PN_SASL_AUTH);
        }

        processComplete.handle(true);
      });
    } else {
      sasl.done(SaslOutcome.PN_SASL_AUTH);
      processComplete.handle(true);
    }
  } else {
    processComplete.handle(false);
  }
}
 

@Override
public void process(final Handler<Boolean> completionHandler) {
    final String[] remoteMechanisms = sasl.getRemoteMechanisms();
    if (remoteMechanisms.length == 0) {
        LOG.trace("client device provided an empty list of SASL mechanisms [hostname: {}, state: {}]",
                sasl.getHostname(), sasl.getState());
        completionHandler.handle(Boolean.FALSE);
        return;
    }

    final String remoteMechanism = remoteMechanisms[0];
    LOG.debug("client device wants to authenticate using SASL [mechanism: {}, host: {}, state: {}]",
            remoteMechanism, sasl.getHostname(), sasl.getState());

    final Context currentContext = Vertx.currentContext();

    final byte[] saslResponse = new byte[sasl.pending()];
    sasl.recv(saslResponse, 0, saslResponse.length);

    buildSaslResponseContext(remoteMechanism, saslResponse)
            .compose(saslResponseContext -> invokePreAuthenticationHandler(saslResponseContext, currentSpan))
            .compose(saslResponseContext -> verify(saslResponseContext))
            .onComplete(outcome -> {
                if (outcome.succeeded()) {
                    currentSpan.log("credentials verified successfully");
                    // add span to connection so that it can be used during the
                    // remaining connection establishment process
                    protonConnection.attachments().set(AmqpAdapterConstants.KEY_CURRENT_SPAN, Span.class,
                            currentSpan);
                    final Device authenticatedDevice = outcome.result();
                    protonConnection.attachments().set(AmqpAdapterConstants.KEY_CLIENT_DEVICE, Device.class,
                            authenticatedDevice);
                    succeeded = true;
                    sasl.done(SaslOutcome.PN_SASL_OK);

                } else {
                    TracingHelper.logError(currentSpan, outcome.cause());
                    currentSpan.finish();
                    LOG.debug("SASL handshake failed: {}", outcome.cause().getMessage());
                    sasl.done(SaslOutcome.PN_SASL_AUTH);
                }

                if (currentContext == null) {
                    completionHandler.handle(Boolean.TRUE);
                } else {
                    // invoke the completion handler on the calling context.
                    currentContext.runOnContext(action -> completionHandler.handle(Boolean.TRUE));
                }
            });
}
 

@Override
public void process(final Handler<Boolean> completionHandler) {

    final String[] remoteMechanisms = sasl.getRemoteMechanisms();

    if (remoteMechanisms.length == 0) {
        LOG.debug("client provided an empty list of SASL mechanisms [hostname: {}, state: {}]",
                sasl.getHostname(), sasl.getState().name());
        completionHandler.handle(false);
    } else {
        final String chosenMechanism = remoteMechanisms[0];
        LOG.debug("client wants to authenticate using SASL [mechanism: {}, host: {}, state: {}]",
                chosenMechanism, sasl.getHostname(), sasl.getState().name());

        final Promise<HonoUser> authTracker = Promise.promise();
        authTracker.future().onComplete(s -> {
            final SaslOutcome saslOutcome;
            if (s.succeeded()) {

                final HonoUser user = s.result();
                LOG.debug("authentication of client [authorization ID: {}] succeeded", user.getName());
                Constants.setClientPrincipal(protonConnection, user);
                succeeded = true;
                saslOutcome = SaslOutcome.PN_SASL_OK;

            } else {

                if (s.cause() instanceof ServiceInvocationException) {
                    final int status = ((ServiceInvocationException) s.cause()).getErrorCode();
                    LOG.debug("authentication check failed: {} (status {})", s.cause().getMessage(), status);
                    saslOutcome = getSaslOutcomeForErrorStatus(status);
                } else {
                    LOG.debug("authentication check failed (no status code given in exception)", s.cause());
                    saslOutcome = SaslOutcome.PN_SASL_TEMP;
                }

            }
            sasl.done(saslOutcome);
            completionHandler.handle(Boolean.TRUE);
        });

        final byte[] saslResponse = new byte[sasl.pending()];
        sasl.recv(saslResponse, 0, saslResponse.length);

        verify(chosenMechanism, saslResponse, authTracker);
    }
}
 

@Override
public void run(Selectable selectable) {
    Reactor reactor = selectable.getReactor();
    try {
        SocketChannel socketChannel = ((ServerSocketChannel)selectable.getChannel()).accept();
        if (socketChannel == null) {
            throw new ReactorInternalException("Selectable readable, but no socket to accept");
        }
        Handler handler = BaseHandler.getHandler(AcceptorImpl.this);
        if (handler == null) {
            handler = reactor.getHandler();
        }
        Connection conn = reactor.connection(handler);
        Record conn_recs = conn.attachments();
        conn_recs.set(CONNECTION_ACCEPTOR_KEY, Acceptor.class, AcceptorImpl.this);
        InetSocketAddress peerAddr = (InetSocketAddress)socketChannel.getRemoteAddress();
        if (peerAddr != null) {
            Address addr = new Address();
            addr.setHost(peerAddr.getHostString());
            addr.setPort(Integer.toString(peerAddr.getPort()));
            conn_recs.set(ReactorImpl.CONNECTION_PEER_ADDRESS_KEY, Address.class, addr);
        }
        Transport trans = Proton.transport();

        int maxFrameSizeOption = reactor.getOptions().getMaxFrameSize();
        if (maxFrameSizeOption != 0) {
            trans.setMaxFrameSize(maxFrameSizeOption);
        }

        if(reactor.getOptions().isEnableSaslByDefault()) {
            Sasl sasl = trans.sasl();
            sasl.server();
            sasl.setMechanisms("ANONYMOUS");
            sasl.done(SaslOutcome.PN_SASL_OK);
        }
        trans.bind(conn);
        IOHandler.selectableTransport(reactor, socketChannel.socket(), trans);
    } catch(IOException ioException) {
        sel.error();
    }
}
 

/** 5.3.2 SASL Negotiation. */
@Test
public void testSaslNegotiation() throws Exception
{
    getClient().transport = Proton.transport();
    getServer().transport = Proton.transport();

    Sasl clientSasl = getClient().transport.sasl();
    clientSasl.client();
    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_NONE, clientSasl.getOutcome());

    Sasl serverSasl = getServer().transport.sasl();
    serverSasl.server();
    serverSasl.setMechanisms(TESTMECH1, TESTMECH2);
    assertEquals("Server should not yet know the remote's chosen mechanism.",
                 0,
                 serverSasl.getRemoteMechanisms().length);

    pumpClientToServer();
    pumpServerToClient();

    assertArrayEquals("Client should now know the server's mechanisms.",
                      new String[]{TESTMECH1, TESTMECH2},
                      clientSasl.getRemoteMechanisms());
    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_NONE, clientSasl.getOutcome());
    clientSasl.setMechanisms(TESTMECH1);

    pumpClientToServer();

    assertArrayEquals("Server should now know the client's chosen mechanism.",
                      new String[]{TESTMECH1},
                      serverSasl.getRemoteMechanisms());

    serverSasl.send(CHALLENGE_BYTES, 0, CHALLENGE_BYTES.length);

    pumpServerToClient();

    byte[] clientReceivedChallengeBytes = new byte[clientSasl.pending()];
    clientSasl.recv(clientReceivedChallengeBytes, 0, clientReceivedChallengeBytes.length);

    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_NONE, clientSasl.getOutcome());
    assertArrayEquals("Client should now know the server's challenge",
                      CHALLENGE_BYTES,
                      clientReceivedChallengeBytes);

    clientSasl.send(RESPONSE_BYTES, 0, RESPONSE_BYTES.length);

    pumpClientToServer();

    byte[] serverReceivedResponseBytes = new byte[serverSasl.pending()];
    serverSasl.recv(serverReceivedResponseBytes, 0, serverReceivedResponseBytes.length);

    assertArrayEquals("Server should now know the client's response",
                      RESPONSE_BYTES,
                      serverReceivedResponseBytes);

    serverSasl.done(SaslOutcome.PN_SASL_OK);
    pumpServerToClient();

    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_OK, clientSasl.getOutcome());
}
 

/**
 *  5.3.3.5 The additional-data field carries additional data on successful authentication outcome as specified
 *  by the SASL specification [RFC4422].
 */
@Test
public void testOutcomeAdditionalData() throws Exception
{
    getClient().transport = Proton.transport();
    getServer().transport = Proton.transport();

    Sasl clientSasl = getClient().transport.sasl();
    clientSasl.client();
    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_NONE, clientSasl.getOutcome());

    Sasl serverSasl = getServer().transport.sasl();
    serverSasl.server();
    serverSasl.setMechanisms(TESTMECH1);

    pumpClientToServer();
    pumpServerToClient();

    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_NONE, clientSasl.getOutcome());
    clientSasl.setMechanisms(TESTMECH1);

    pumpClientToServer();

    serverSasl.send(CHALLENGE_BYTES, 0, CHALLENGE_BYTES.length);

    pumpServerToClient();

    byte[] clientReceivedChallengeBytes = new byte[clientSasl.pending()];
    clientSasl.recv(clientReceivedChallengeBytes, 0, clientReceivedChallengeBytes.length);

    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_NONE, clientSasl.getOutcome());
    clientSasl.send(RESPONSE_BYTES, 0, RESPONSE_BYTES.length);

    pumpClientToServer();

    byte[] serverReceivedResponseBytes = new byte[serverSasl.pending()];
    serverSasl.recv(serverReceivedResponseBytes, 0, serverReceivedResponseBytes.length);

    serverSasl.send(ADDITIONAL_DATA_BYTES, 0, ADDITIONAL_DATA_BYTES.length);
    serverSasl.done(SaslOutcome.PN_SASL_OK);
    pumpServerToClient();

    byte[] clientReceivedAdditionalDataBytes = new byte[clientSasl.pending()];
    clientSasl.recv(clientReceivedAdditionalDataBytes, 0, clientReceivedAdditionalDataBytes.length);

    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_OK, clientSasl.getOutcome());
    assertArrayEquals("Client should now know the serrver's additional-data",
                      ADDITIONAL_DATA_BYTES,
                      clientReceivedAdditionalDataBytes);
}
 

@Test
public void testSaslNegotiationWithConfiguredLargerFrameSize() throws Exception
{
    final byte[] largeInitialResponseBytesOrig = fillBytes("initialResponse", 1431);
    final byte[] largeChallengeBytesOrig = fillBytes("challenge", 1375);
    final byte[] largeResponseBytesOrig = fillBytes("response", 1282);
    final byte[] largeAdditionalBytesOrig = fillBytes("additionalData", 1529);

    getClient().transport = Proton.transport();
    getServer().transport = Proton.transport();

    // Configure transports to allow for larger initial frame sizes
    getClient().transport.setInitialRemoteMaxFrameSize(2048);
    getServer().transport.setInitialRemoteMaxFrameSize(2048);

    Sasl clientSasl = getClient().transport.sasl();
    clientSasl.client();

    Sasl serverSasl = getServer().transport.sasl();
    serverSasl.server();

    // Negotiate the mech
    serverSasl.setMechanisms(TESTMECH1, TESTMECH2);

    pumpClientToServer();
    pumpServerToClient();

    assertArrayEquals("Client should now know the server's mechanisms.", new String[] { TESTMECH1, TESTMECH2 },
            clientSasl.getRemoteMechanisms());
    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_NONE, clientSasl.getOutcome());

    // Select a mech, send large initial response along with it in sasl-init, verify server receives it
    clientSasl.setMechanisms(TESTMECH1);
    byte[] initialResponseBytes = Arrays.copyOf(largeInitialResponseBytesOrig, largeInitialResponseBytesOrig.length);
    clientSasl.send(initialResponseBytes, 0, initialResponseBytes.length);

    pumpClientToServer();

    assertArrayEquals("Server should now know the client's chosen mechanism.", new String[] { TESTMECH1 },
            serverSasl.getRemoteMechanisms());

    byte[] serverReceivedInitialResponseBytes = new byte[serverSasl.pending()];
    serverSasl.recv(serverReceivedInitialResponseBytes, 0, serverReceivedInitialResponseBytes.length);

    assertArrayEquals("Server should now know the clients initial response", largeInitialResponseBytesOrig,
            serverReceivedInitialResponseBytes);

    // Send a large challenge in a sasl-challenge, verify client receives it
    byte[] challengeBytes = Arrays.copyOf(largeChallengeBytesOrig, largeChallengeBytesOrig.length);
    serverSasl.send(challengeBytes, 0, challengeBytes.length);

    pumpServerToClient();

    byte[] clientReceivedChallengeBytes = new byte[clientSasl.pending()];
    clientSasl.recv(clientReceivedChallengeBytes, 0, clientReceivedChallengeBytes.length);

    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_NONE, clientSasl.getOutcome());
    assertArrayEquals("Client should now know the server's challenge", largeChallengeBytesOrig,
            clientReceivedChallengeBytes);

    // Send a large response in a sasl-response, verify server receives it
    byte[] responseBytes = Arrays.copyOf(largeResponseBytesOrig, largeResponseBytesOrig.length);
    clientSasl.send(responseBytes, 0, responseBytes.length);

    pumpClientToServer();

    byte[] serverReceivedResponseBytes = new byte[serverSasl.pending()];
    serverSasl.recv(serverReceivedResponseBytes, 0, serverReceivedResponseBytes.length);

    assertArrayEquals("Server should now know the client's response", largeResponseBytesOrig, serverReceivedResponseBytes);

    // Send an outcome with large additional data in a sasl-outcome, verify client receives it
    byte[] additionalBytes = Arrays.copyOf(largeAdditionalBytesOrig, largeAdditionalBytesOrig.length);
    serverSasl.send(additionalBytes, 0, additionalBytes.length);
    serverSasl.done(SaslOutcome.PN_SASL_OK);
    pumpServerToClient();

    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_OK, clientSasl.getOutcome());

    byte[] clientReceivedAdditionalBytes = new byte[clientSasl.pending()];
    clientSasl.recv(clientReceivedAdditionalBytes, 0, clientReceivedAdditionalBytes.length);

    assertArrayEquals("Client should now know the server's outcome additional data", largeAdditionalBytesOrig,
            clientReceivedAdditionalBytes);
}
 

@Test
public void testSaslNegotiationUsingListener() throws Exception
{
    getClient().transport = Proton.transport();
    getServer().transport = Proton.transport();

    AtomicBoolean mechanismsReceived = new AtomicBoolean();
    AtomicBoolean challengeReceived = new AtomicBoolean();
    AtomicBoolean outcomeReceived = new AtomicBoolean();

    Sasl clientSasl = getClient().transport.sasl();
    clientSasl.client();
    clientSasl.setListener(new ClientSaslHandling(mechanismsReceived, challengeReceived, outcomeReceived));

    AtomicBoolean initReceived = new AtomicBoolean();
    AtomicBoolean responseReceived = new AtomicBoolean();

    Sasl serverSasl = getServer().transport.sasl();
    serverSasl.server();
    serverSasl.setMechanisms(TESTMECH1, TESTMECH2);
    serverSasl.setListener(new ServerSaslHandling(initReceived, responseReceived));

    pumpClientToServer();
    pumpServerToClient();

    assertTrue("mechanisms were not received by client", mechanismsReceived.get());
    assertFalse("init was received by server", initReceived.get());

    pumpClientToServer();

    assertTrue("init was not received by server", initReceived.get());
    assertFalse("challenge was received by client", challengeReceived.get());

    pumpServerToClient();

    assertTrue("challenge was not received by client", challengeReceived.get());
    assertFalse("response was received by server", responseReceived.get());

    pumpClientToServer();

    assertTrue("response was received by server", responseReceived.get());
    assertFalse("outcome was received by client", outcomeReceived.get());

    pumpServerToClient();

    assertTrue("outcome was received by client", outcomeReceived.get());

    assertEquals("Unexpected SASL outcome at client", SaslOutcome.PN_SASL_OK, clientSasl.getOutcome());
}
 

private void recordFailure(String message, Throwable cause) {
    recordFailure(message, cause, SaslOutcome.PN_SASL_NONE.getCode());
}