WIP: Add Sasl authentication - PLAIN #129

ramrengaswamy · 2022-04-21T02:36:38Z

Closes #125

Describe your proposed changes here.

Added support for PLAIN SASL auth.
Added a couple of new requests SaslHandshakeRequest and SaslAuthenticateRequest
Added a couple of primitives Bytes (not used) and CompactBytes.
Added a SaslConfig struct and that can be set in the ClientBuilder and used during connection establishment.
I've read the contributing section of the project CONTRIBUTING.md.
Signed CLA (if not already signed).

crepererum · 2022-04-21T08:24:40Z

src/protocol/messages/sasl_auth.rs

+{
+    fn read_versioned(reader: &mut R, version: ApiVersion) -> Result<Self, ReadVersionedError> {
+        let v = version.0 .0;
+        assert!(v <= 1);


This is not in line w/ API_VERSION_RANGE: you don't support version 0.

crepererum · 2022-04-21T08:25:59Z

src/protocol/messages/sasl_auth.rs

+        version: ApiVersion,
+    ) -> Result<(), WriteVersionedError> {
+        let v = version.0 .0;
+        assert!(v == 2);


Why is version 1 not supported?

Edit: reading https://kafka.apache.org/protocol#sasl_handshake again I think we can also support version 0 for the SaslAuthenticate message.

I am not super familiar with the protocol.
Based on this:
If SaslHandshakeRequest version is v0, a series of SASL client and server tokens corresponding to the mechanism are sent as opaque packets without wrapping the messages with Kafka protocol headers.
I wasn't sure if this library could handle messages without Kafka protocol headers.

The sentence you're quoting is about the SaslHandshakeRequest (for which I agree that we should only support version 1), not about the SaslAuthenticateRequest. The versions for these two request-response types are different. The version selection will be done by the Messenger.

I realize that your comment was regarding SaslAuthenticate. The issue I ran in to here was that the type of the fields got modified through the different versions. For e.g. auth_bytes was BYTES in version 0 but become COMPACT_BYTES from version 1 onwards. Is there an example for this is handled anywhere else in the library?

I see. TBH we don't have a beautiful solution of it. I would go for Bytes as a type within the struct and then do some type conversion during read/write. See this example:

rskafka/src/protocol/messages/delete_records.rs

Lines 63 to 66 in ad8de3c

#[derive(Debug)]

pub struct DeleteRequestTopic {

/// The topic name.

pub name: String_,

Here the name is an ordinary string, but is compact after a certain version. This is then handled here:

rskafka/src/protocol/messages/delete_records.rs

Lines 89 to 93 in ad8de3c

if v >= 2 {

CompactStringRef(&self.name.0).write(writer)?

} else {

self.name.write(writer)?;

}

I think you might want to introduce CompactBytesRef for that, so you don't have to copy the payload during type conversion. This is similar how it is done for CompactString:

rskafka/src/protocol/primitives.rs

Lines 362 to 365 in ad8de3c

/// Represents a string whose length is expressed as a variable-length integer rather than a fixed 2-byte length.

#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Default)]

#[cfg_attr(test, derive(proptest_derive::Arbitrary))]

pub struct CompactString(pub String);

rskafka/src/protocol/primitives.rs

Lines 392 to 399 in ad8de3c

impl<W> WriteType<W> for CompactString

where

W: Write,

{

fn write(&self, writer: &mut W) -> Result<(), WriteError> {

CompactStringRef(&self.0).write(writer)

}

}

(note how the "write" operation is only implemented once, because CompactString can use CompactStringRef)

rskafka/src/protocol/primitives.rs

Lines 401 to 417 in ad8de3c

/// Same as [`CompactString`] but contains referenced data.

///

/// This only supports writing.

#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]

pub struct CompactStringRef<'a>(pub &'a str);

impl<'a, W> WriteType<W> for CompactStringRef<'a>

where

W: Write,

{

fn write(&self, writer: &mut W) -> Result<(), WriteError> {

let len = u64::try_from(self.0.len() + 1).map_err(WriteError::Overflow)?;

UnsignedVarint(len).write(writer)?;

writer.write_all(self.0.as_bytes())?;

Ok(())

}

}

Neat! I will follow a similar pattern.

crepererum · 2022-04-21T08:28:38Z

src/protocol/messages/sasl_auth.rs

+#[derive(Debug)]
+pub struct SaslHandshakeRequest {
+    pub mechanism: String_,
+}


Would be nice if you could copy over the docstrings for the official message description. If a field was added at a specific version, this should also be documented, e.g.

/// Some description as written in <https://kafka.apache.org/protocol> /// /// Added in version X. pub my_field: String_,

crepererum · 2022-04-21T08:29:48Z

src/protocol/primitives.rs

 }

+#[derive(Debug)]
+pub struct Bytes(pub Vec<u8>);


Looks good. Please copy over the docs from https://kafka.apache.org/protocol#protocol_types and add a roundtrip test.

crepererum · 2022-04-21T08:33:42Z

src/protocol/messages/sasl_auth.rs

+#[derive(Debug, Clone)]
+pub struct SaslConfig {
+    pub username: String,
+    pub password: String,
+}


This is not really a part of the underlying protocol but our own API. Should probably go to src/connection/transport/sasl.rs. This would also allow you to use this type in CliendBuilder::sasl_config w/o leaking protocol details. I would expect that a struct/enum instead of username+password parameters are also more future-proof (in light of all the different SASL auth methods).

crepererum · 2022-04-21T08:35:50Z

src/messenger.rs

+    pub async fn sasl_handshake(&self, sasl_config: &SaslConfig) -> Result<(),  RequestError> {
+        let req = SaslHandshakeRequest::new();
+        self.request(req).await?;
+        let req = SaslAuthenticateRequest::new(sasl_config);
+        self.request(req).await?;
+        Ok(())
+    }


So if there is a SaslConfig, then we need to call this each time after sync_versions. I think you can just grep throw the code and see where is this done. Maybe you wanna make the two methods private and instead offer a setup method that does both, so we don't forget to call one or the other a certain callsides.

crepererum · 2022-04-21T08:38:52Z

@ramrengaswamy this is a good start 👍

ramrengaswamy · 2022-04-21T18:12:08Z

src/connection.rs


        let messenger = Arc::new(Messenger::new(BufStream::new(transport), max_message_size));
        messenger.sync_versions().await?;
+        if let Some(sasl_config) = sasl_config {


@crepererum I didn't refactor this into a separate setup function because I wasn't sure SaslConfig should leak into the Messenger. Right now SaslConfig is known to transport, connection and ClientBuilder.

Looks good to me. I've just checked: this is the only place where we call sync_versions and also the place where we set up SOCKS5 and TLS. So I think this is a good spot to call sasl_handshake 👍

ramrengaswamy · 2022-04-25T18:04:34Z

@crepererum ... Ready for another look.
This is my first real rust program so also let me know if some code doesn't look "rusty" :)
I will fix those.

crepererum

Looks solid 👍

Now how do we test this? 😁

crepererum · 2022-04-26T13:38:36Z

src/client/mod.rs

+    /// Setup SASL username and password. Mechanism is assumed to be PLAIN.
+    pub fn sasl_config(mut self, username: &str, password: &str) -> Self {
+        self.sasl_config = Some(SaslConfig::Plain{
+            username: username.to_string(),
+            password: password.to_string(),
+        });
+        self
+    }


I would take SaslConfig as an argument here so we can extend the config in the future w/o overloading the ClientBuilder w/ too many methods.

Had to pub use crate::connection::SaslConfig in client/mod.rs to publicly expose SaslConfig

crepererum · 2022-04-26T13:41:40Z

src/connection/transport/sasl.rs

+                let mut auth: Vec<u8> = Vec::new();
+                auth.push(0);
+                auth.extend(username.bytes());
+                auth.push(0);
+                auth.extend(password.bytes());


Can you add a link/reference for this binary format?

I think for PLAIN it's OK to hand-roll this, in the future however I would like to offload this to another crate because I don't wanna deal w/ all the SASL crypto stuff in rskafka.

Are you okay depending on a pure rust crate such as this one? https://docs.rs/sasl/latest/sasl/
The other SASL packages are rust wrappers around C implementations.

crepererum · 2022-04-26T13:42:42Z

src/protocol/messages/sasl_msg.rs

+    ///
+    /// Added in version 0.


I think you don't have to mention that a field was always there ("Added in version 0."), but it's OK.

Got rid of it to keep it consistent with the rest of the codebase.

crepererum · 2022-04-26T13:44:05Z

src/connection/transport/sasl.rs

+}
+
+impl SaslConfig {
+    pub fn auth_bytes(&self) -> Vec<u8> {


Suggested change

pub fn auth_bytes(&self) -> Vec<u8> {

pub(crate) fn auth_bytes(&self) -> Vec<u8> {

This is not a public API that falls under semantic versioning and that users should rely on. Same goes for fn mechanism.

ramrengaswamy · 2022-04-26T18:40:27Z

I am currently testing these changes against a simple application that I am running against a Confluent stream.
I am not sure if it is possible to run a local Kafka instance with SASL auth enabled. Will look into it.

ramrengaswamy · 2022-04-27T17:23:51Z

@crepererum - Seems that the local Kafka and Redpanda have sasl enabled. So I added a test_sasl that will trigger the SASL handshake and authenticate exchanges in the integration tests.

crepererum · 2022-04-28T08:50:38Z

src/messenger.rs

+    pub async fn sasl_handshake(
+        &self,
+        mechanism: &str,
+        auth_bytes: Vec<u8>,
+    ) -> Result<(), RequestError> {
+        let req = SaslHandshakeRequest::new(mechanism);
+        self.request(req).await?;
+        let req = SaslAuthenticateRequest::new(auth_bytes);
+        self.request(req).await?;
+        Ok(())
+    }
 }


I was really surprised that this "just works" in our CI tests. In short: It doesn't. You have to check the error codes within the messages (SaslHandshakeResponse::error_code and SaslAuthenticateResponse::error_code) and return an error if the codes are set.

You may wonder how I found that out that quickly: Wireshark:

Ugh, can't believe I missed that! Fixing.

ramrengaswamy · 2022-04-29T01:11:05Z

@crepererum tl;dr - I got the sasl integration test working locally but ran into an issue getting to work on CircleCI.

The following command will run all Kafka integration tests (including sasl)
TEST_INTEGRATION=1 TEST_DELETE_RECORDS=1 KAFKA_CONNECT=localhost:9094 KAFKA_SASL_CONNECT=localhost:9097 cargo test

To get Sasl integration test to work I made the following changes:

Added a new SECURE listener to all the brokers and configured its security to SASL_PLAINTEXT.
Added a kafka_jaas.conf file to get the brokers to boot up successfully and handle simple password auth.
Introduced a new environment variable KAFKA_SASL_CONNECT that is used by test_sasl to connect to the SECURE listener.

I have not used CircleCI before. It seems there is no way to volume mount into a Docker container.
So we would have to use this workaround to copy kafka_jaas.conf into CircleCI and make it accessible to the brokers.

crepererum · 2022-04-29T11:17:25Z

@ramrengaswamy Didn't have time to look into this today, sorry. I'll be out of office for two weeks, I'll get back to this on the 16th of May.

crepererum · 2022-05-04T15:02:41Z

BTW: we're testing against Kafka and really redpanda. Could you have a look if you can get SASL to work with redpanda as well?

crepererum · 2022-05-31T14:21:50Z

@ramrengaswamy any progress on this?

seijik42 · 2023-02-26T10:13:32Z

I tried to contribute this by fixing the test for Redpanda but seems like they don't support SASL/PLAIN at the moment. (only SASL/SCRAM)
https://docs.redpanda.com/docs/manage/security/authentication/#saslscram
Is it ok to simply skip test_sasl when envvar TEST_BROKER_IMPL=redpanda ?

My PR will be like this. #198

toondaey · 2023-08-07T08:39:46Z

Is there any progress on this? Or is there anything that can be done to help with this?

crepererum · 2023-08-07T09:03:11Z

@toondaey I think the latest attempt is #198 . If you could pick that one up (just create a new PR) and make sure that the tests also run in CircleCI, I think we could easily drive it to completion.

toondaey · 2023-08-07T09:54:47Z

@toondaey I think the latest attempt is #198 . If you could pick that one up (just create a new PR) and make sure that the tests also run in CircleCI, I think we could easily drive it to completion.

Sure. I'll give it a whirl.

crepererum · 2023-08-08T15:31:45Z

This is now supported, see #216.

Neustradamus · 2023-08-17T23:10:41Z

Linked to:

WIP: Add Sasl authentication - PLAIN

da8f5e0

crepererum reviewed Apr 21, 2022

View reviewed changes

ramrengaswamy added 2 commits April 21, 2022 14:01

[WIP]: Address review comments.

d6099cf

[WIP] - Rust format

aef2979

ramrengaswamy commented Apr 21, 2022

View reviewed changes

Ram Kumar Rengaswamy and others added 3 commits April 21, 2022 14:20

Merge branch 'influxdata:main' into main

28ce752

WIP: Handle version 0 and 1 for SaslAuthenticateRequest

717e395

Merge branch 'main' into main

58456de

crepererum reviewed Apr 26, 2022

View reviewed changes

WIP - Expose SaslConfig struct from Client

bd72176

ramrengaswamy added 3 commits April 27, 2022 09:55

WIP: Fix clippy errors

ae6692c

WIP: Rust fmt checks

12c2bf4

WIP: Add client SASL test

1ae6b61

crepererum reviewed Apr 28, 2022

View reviewed changes

WIP: Run Sasl integration test for Kafka

05effaf

Ram Kumar Rengaswamy added 2 commits May 12, 2022 14:07

Merge branch 'influxdata:main' into main

6559148

Merge branch 'influxdata:main' into main

a7d336c

crepererum mentioned this pull request Nov 7, 2022

Support for authorization methods. #125

Closed

crepererum closed this Aug 8, 2023

This was referenced Aug 17, 2023

WIP: Add Sasl authentication - PLAIN (No Redpand test) #198

Closed

feat: sasl auth impl #216

Merged

	#[derive(Debug)]
	pub struct DeleteRequestTopic {
	/// The topic name.
	pub name: String_,

	if v >= 2 {
	CompactStringRef(&self.name.0).write(writer)?
	} else {
	self.name.write(writer)?;
	}

	/// Represents a string whose length is expressed as a variable-length integer rather than a fixed 2-byte length.
	#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Default)]
	#[cfg_attr(test, derive(proptest_derive::Arbitrary))]
	pub struct CompactString(pub String);

	impl<W> WriteType<W> for CompactString
	where
	W: Write,
	{
	fn write(&self, writer: &mut W) -> Result<(), WriteError> {
	CompactStringRef(&self.0).write(writer)
	}
	}

	/// Same as [`CompactString`] but contains referenced data.
	///
	/// This only supports writing.
	#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
	pub struct CompactStringRef<'a>(pub &'a str);

	impl<'a, W> WriteType<W> for CompactStringRef<'a>
	where
	W: Write,
	{
	fn write(&self, writer: &mut W) -> Result<(), WriteError> {
	let len = u64::try_from(self.0.len() + 1).map_err(WriteError::Overflow)?;
	UnsignedVarint(len).write(writer)?;
	writer.write_all(self.0.as_bytes())?;
	Ok(())
	}
	}

	pub fn auth_bytes(&self) -> Vec<u8> {
	pub(crate) fn auth_bytes(&self) -> Vec<u8> {

WIP: Add Sasl authentication - PLAIN #129

WIP: Add Sasl authentication - PLAIN #129

Uh oh!

Conversation

ramrengaswamy commented Apr 21, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

crepererum commented Apr 21, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ramrengaswamy commented Apr 25, 2022

Uh oh!

crepererum left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ramrengaswamy commented Apr 26, 2022

Uh oh!

ramrengaswamy commented Apr 27, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ramrengaswamy commented Apr 29, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

crepererum commented Apr 29, 2022

Uh oh!

crepererum commented May 4, 2022

Uh oh!

crepererum commented May 31, 2022

Uh oh!

seijik42 commented Feb 26, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

toondaey commented Aug 7, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

crepererum commented Aug 7, 2023

Uh oh!

toondaey commented Aug 7, 2023

Uh oh!

ramrengaswamy commented Apr 29, 2022 •

edited

Loading

seijik42 commented Feb 26, 2023 •

edited

Loading

toondaey commented Aug 7, 2023 •

edited

Loading