import org.apache.qpid.proton.engine.Sasl; //导入依赖的package包/类
private void performSaslSteps(Connection connection, InputStream in,
                              OutputStream out,
                              SaslMechanism mechanism) throws IOException, LoginException {
    Transport transport = connection.getTransport();
    Sasl sasl = transport.sasl();
    do {

        readFromNetwork(connection, in, () ->
            !(EnumSet.of(PN_SASL_PASS, PN_SASL_FAIL).contains(sasl.getState())
            || (sasl.getState() == PN_SASL_STEP && sasl.pending() > 0)));

        if (sasl.pending() > 0) {
            byte[] challenge = new byte[sasl.pending()];
            byte[] response = mechanism.getResponse(challenge);
            if (sasl.getState() == PN_SASL_STEP) {
                sasl.send(response, 0, response.length);
                writeToNetwork(connection, out);
            }
        }

    } while (sasl.getState() == PN_SASL_STEP);
}
 

import org.apache.qpid.proton.engine.Sasl; //导入依赖的package包/类
@Override
public void process(final Handler<Boolean> completionHandler) {
    if (sasl == null) {
        throw new IllegalStateException("Init was not called with the associated transport");
    }
    // Note we do not record the identity with which the client authenticated, nor to we take any notice of
    // an alternative identity passed in the response
    boolean done = false;
    String[] remoteMechanisms = sasl.getRemoteMechanisms();
    if (remoteMechanisms.length > 0) {
        String chosen = remoteMechanisms[0];
        if (ProtonSaslExternalImpl.MECH_NAME.equals(chosen)) {
            // TODO - should handle the case of no initial response per the SASL spec, (i.e. send an empty challenge)
            // however this was causing errors in some clients
            // Missing initial response can be detected with: sasl.recv(new byte[0], 0, 0) == -1
            sasl.done(Sasl.SaslOutcome.PN_SASL_OK);
            succeeded = true;
        } else {
            sasl.done(Sasl.SaslOutcome.PN_SASL_AUTH);
        }
        done = true;
    }
    completionHandler.handle(done);
}
 

import org.apache.qpid.proton.engine.Sasl; //导入依赖的package包/类
@Override
public Sasl sasl()
{
    if(_sasl == null)
    {
        if(_processingStarted)
        {
            throw new IllegalStateException("Sasl can't be initiated after transport has started processing");
        }

        init();
        _sasl = new SaslImpl(this, _remoteMaxFrameSize);
        TransportWrapper transportWrapper = _sasl.wrap(_inputProcessor, _outputProcessor);
        _inputProcessor = transportWrapper;
        _outputProcessor = transportWrapper;
    }
    return _sasl;

}
 

import org.apache.qpid.proton.engine.Sasl; //导入依赖的package包/类
@Override
public void onSaslInit(Sasl s, Transport t)
{
    assertArrayEquals("Server should now know the client's chosen mechanism.",
            new String[]{TESTMECH1}, s.getRemoteMechanisms());

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

    assertArrayEquals("Server should now know the client's initial response.",
            INITIAL_RESPONSE_BYTES, serverReceivedInitialBytes);

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

    assertFalse("Should not have already received init", initReceived.getAndSet(true));
}
 

import org.apache.qpid.proton.engine.Sasl; //导入依赖的package包/类
@Override
public void channelActive(ChannelHandlerContext ctx) {
    synchronized (lock) {
        System.out.println("ACTIVE");
        transport = Transport.Factory.create();
        transport.setContext(ctx);

        Sasl sasl = transport.sasl();
        sasl.setMechanisms("ANONYMOUS");
        sasl.server();
        sasl.done(Sasl.PN_SASL_OK);

        connection = Connection.Factory.create();
        collector = Collector.Factory.create();
        connection.collect(collector);
        transport.bind(connection);

        // XXX: really we should fire both of these off of an
        //      initial transport event
        write(ctx);
        int capacity = transport.capacity();
        if (capacity > 0) {
            ctx.read();
        }
    }
}
 

import org.apache.qpid.proton.engine.Sasl; //导入依赖的package包/类
void init() throws NetworkException
{
    synchronized (_lock)
    {
        _collector = Collector.Factory.create();
        _transport = Transport.Factory.create();
        _connection = Connection.Factory.create();
        String id = _settings.getId();
        _connection.collect(_collector);
        _connection.setContainer(id == null || id.trim().equals("") ? UUID.randomUUID().toString() : id);
        _connection.setHostname(_settings.getHost());
        _transport.bind(_connection);
        Sasl sasl = _transport.sasl();
        sasl.client();
        sasl.setMechanisms(new String[] { "ANONYMOUS" });
        _connection.open();
        _connected.set(true);
        write();
    }
}
 

import org.apache.qpid.proton.engine.Sasl; //导入依赖的package包/类
public void handleSaslMechanisms(Sasl sasl, Transport transport) {
    try {
        String[] remoteMechanisms = sasl.getRemoteMechanisms();
        if (remoteMechanisms != null && remoteMechanisms.length != 0) {
            try {
                mechanism = mechanismFinder.apply(remoteMechanisms);
            } catch (JMSSecurityRuntimeException jmssre){
                recordFailure("Could not find a suitable SASL mechanism. " + jmssre.getMessage(), jmssre);
                return;
            }

            byte[] response = mechanism.getInitialResponse();
            if (response != null) {
                sasl.send(response, 0, response.length);
            }
            sasl.setMechanisms(mechanism.getName());
        }
    } catch (Throwable error) {
        recordFailure("Exception while processing SASL init: " + error.getMessage(), error);
    }
}
 

import org.apache.qpid.proton.engine.Sasl; //导入依赖的package包/类
public void handleSaslOutcome(Sasl sasl, Transport transport) {
    try {
        switch (sasl.getState()) {
            case PN_SASL_FAIL:
                handleSaslFail();
                break;
            case PN_SASL_PASS:
                handleSaslCompletion(sasl);
                break;
            default:
                break;
        }
    } catch (Throwable error) {
        recordFailure(error.getMessage(), error);
    }
}
 

import org.apache.qpid.proton.engine.Sasl; //导入依赖的package包/类
public void processSaslExchange() {
    if (protonSasl.getRemoteMechanisms().length > 0) {

        String[] mechanisms = protonSasl.getRemoteMechanisms();
        if (mechanisms != null && mechanisms.length > 0) {
            LOG.debug("SASL [{}} Handshake started.", mechanisms[0]);

            if (mechanisms[0].equalsIgnoreCase("PLAIN")) {
                byte[] data = new byte[protonSasl.pending()];
                protonSasl.recv(data, 0, data.length);
                protonSasl.done(Sasl.SaslOutcome.PN_SASL_OK);
            } else if (mechanisms[0].equalsIgnoreCase("ANONYMOUS")) {
                protonSasl.done(Sasl.SaslOutcome.PN_SASL_OK);
            } else {
                protonSasl.done(Sasl.SaslOutcome.PN_SASL_PERM);
            }

        } else {
            LOG.info("SASL: could not find supported mechanism");
            protonSasl.done(Sasl.SaslOutcome.PN_SASL_PERM);
        }
    }
}
 

import org.apache.qpid.proton.engine.Sasl; //导入依赖的package包/类
public AmqpConnection(AmqpProvider provider, Connection protonConnection, Sasl sasl, JmsConnectionInfo info) {
    super(info, protonConnection);

    this.provider = provider;
    this.remoteURI = provider.getRemoteURI();

    if (sasl != null) {
        this.authenticator = new AmqpSaslAuthenticator(sasl, info);
    }

    this.info.getConnectionId().setProviderHint(this);

    this.queuePrefix = info.getQueuePrefix();
    this.topicPrefix = info.getTopicPrefix();
    this.tempQueuePrefix = info.getTempQueuePrefix();
    this.tempTopicPrefix = info.getTempTopicPrefix();

    // Create a Session for this connection that is used for Temporary Destinations
    // and perhaps later on management and advisory monitoring.
    JmsSessionInfo sessionInfo = new JmsSessionInfo(this.info, -1);
    sessionInfo.setAcknowledgementMode(Session.AUTO_ACKNOWLEDGE);

    this.connectionSession = new AmqpSession(this, sessionInfo);
}
 

import org.apache.qpid.proton.engine.Sasl; //导入依赖的package包/类
private SaslMechanism chooseSaslMechanismAndSendInit(Connection connection, InputStream in, OutputStream out) throws LoginException, IOException {

        Transport transport = connection.getTransport();
        Sasl sasl = transport.sasl();
        SaslMechanism mechanism = null;
        // read from network until we get a sasl-mechanisms
        readFromNetwork(connection, in, () -> sasl.getState() == PN_SASL_IDLE && sasl.getRemoteMechanisms().length == 0);

        for (SaslMechanismFactory factory : saslFactories) {
            if (Arrays.asList(sasl.getRemoteMechanisms()).contains(factory.getName())) {
                mechanism = factory.newInstance(callbackHandler, sharedState, options);
                if (mechanism != null) {
                    sasl.setRemoteHostname(saslHostname);
                    sasl.setMechanisms(factory.getName());
                    byte[] initialResponse = mechanism.getResponse(null);
                    if (initialResponse != null && initialResponse.length != 0) {
                        sasl.send(initialResponse, 0, initialResponse.length);
                    }
                    break;
                }
            }
        }

        if (mechanism == null) {
            throw new LoginException("Unable to authenticate using SASL delegation, no supported mechanisms");
        }

        writeToNetwork(connection, out);
        return mechanism;
    }
 

import org.apache.qpid.proton.engine.Sasl; //导入依赖的package包/类
public void selected() throws IOException {
    SocketChannel sock = socket.accept();
    System.out.println("ACCEPTED: " + sock);
    Connection conn = Connection.Factory.create();
    conn.collect(collector);
    Transport transport = Transport.Factory.create();
    Sasl sasl = transport.sasl();
    sasl.setMechanisms("ANONYMOUS");
    sasl.server();
    sasl.done(Sasl.PN_SASL_OK);
    transport.bind(conn);
    new ChannelHandler(sock, SelectionKey.OP_READ, transport);
}
 

import org.apache.qpid.proton.engine.Sasl; //导入依赖的package包/类
private static Transport makeTransport(Connection conn) {
    Transport transport = Transport.Factory.create();
    Sasl sasl = transport.sasl();
    sasl.setMechanisms("ANONYMOUS");
    sasl.client();
    transport.bind(conn);
    return transport;
}
 

import org.apache.qpid.proton.engine.Sasl; //导入依赖的package包/类
private void handleOpen(Reactor reactor, Event event) {
    Connection connection = event.getConnection();
    if (connection.getRemoteState() != EndpointState.UNINITIALIZED) {
        return;
    }
    // Outgoing Reactor connections set the virtual host automatically using the
    // following rules:
    String vhost = connection.getHostname();
    if (vhost == null) {
        // setHostname never called, use the host from the connection's
        // socket address as the default virtual host:
        String conAddr = reactor.getConnectionAddress(connection);
        if (conAddr != null) {
            Address addr = new Address(conAddr);
            connection.setHostname(addr.getHost());
        }
    } else if (vhost.isEmpty()) {
        // setHostname called explictly with a null string. This allows
        // the application to completely avoid sending a virtual host
        // name
        connection.setHostname(null);
    } else {
        // setHostname set by application - use it.
    }
    Transport transport = Proton.transport();

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

    if (reactor.getOptions().isEnableSaslByDefault()) {
        Sasl sasl = transport.sasl();
        sasl.client();
        sasl.setMechanisms("ANONYMOUS");
    }

    transport.bind(connection);
}
 

import org.apache.qpid.proton.engine.Sasl; //导入依赖的package包/类
/** 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());
}
 

import org.apache.qpid.proton.engine.Sasl; //导入依赖的package包/类
/**
 *  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());

}
 

import org.apache.qpid.proton.engine.Sasl; //导入依赖的package包/类
@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));
}
 

import org.apache.qpid.proton.engine.Sasl; //导入依赖的package包/类
@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));
}
 

import org.apache.qpid.proton.engine.Sasl; //导入依赖的package包/类
@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));
}
 

import org.apache.qpid.proton.engine.Sasl; //导入依赖的package包/类
@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));
}