Method sendAndReceiveSASLSCRAMv0

broker.go:1692–1751

()

}

func (b *Broker) sendAndReceiveSASLSCRAMv0() error {
	if err := b.sendAndReceiveSASLHandshake(b.conf.Net.SASL.Mechanism, SASLHandshakeV0); err != nil {
		return err
	}

	scramClient := b.conf.Net.SASL.SCRAMClientGeneratorFunc()
	if err := scramClient.Begin(b.conf.Net.SASL.User, b.conf.Net.SASL.Password, b.conf.Net.SASL.SCRAMAuthzID); err != nil {
		return fmt.Errorf("failed to start SCRAM exchange with the server: %w", err)
	}

	msg, err := scramClient.Step("")
	if err != nil {
		return fmt.Errorf("failed to advance the SCRAM exchange: %w", err)
	}

	for !scramClient.Done() {
		requestTime := time.Now()
		// Will be decremented in updateIncomingCommunicationMetrics (except error)
		b.addRequestInFlightMetrics(1)
		length := len(msg)
		authBytes := make([]byte, length+4) // 4 byte length header + auth data
		binary.BigEndian.PutUint32(authBytes, uint32(length))
		copy(authBytes[4:], msg)
		_, err := b.write(authBytes)
		b.updateOutgoingCommunicationMetrics(length + 4)
		if err != nil {
			b.addRequestInFlightMetrics(-1)
			Logger.Printf("Failed to write SASL auth header to broker %s: %s\n", b.addr, err.Error())
			return err
		}
		b.correlationID++
		header := make([]byte, 4)
		_, err = b.readFull(header)
		if err != nil {
			b.addRequestInFlightMetrics(-1)
			Logger.Printf("Failed to read response header while authenticating with SASL to broker %s: %s\n", b.addr, err.Error())
			return err
		}
		payloadLength := binary.BigEndian.Uint32(header)
		if int64(payloadLength) > int64(MaxResponseSize) {
			return PacketDecodingError{fmt.Sprintf("SASL response of length %d too large", payloadLength)}
		}
		payload := make([]byte, int(payloadLength))
		n, err := b.readFull(payload)
		if err != nil {
			b.addRequestInFlightMetrics(-1)
			Logger.Printf("Failed to read response payload while authenticating with SASL to broker %s: %s\n", b.addr, err.Error())
			return err
		}
		b.updateIncomingCommunicationMetrics(n+4, time.Since(requestTime))
		msg, err = scramClient.Step(string(payload))
		if err != nil {
			Logger.Println("SASL authentication failed", err)
			return err
		}
	}

	DebugLogger.Println("SASL authentication succeeded")