Merge pull request #498 from fresh-borzoni/sasl-scram-support

SASL Auth implementation(PLAIN + SCRAM)

Author: Argonus

AUTH.md

@@ -0,0 +1,148 @@
+# SASL Authentication
+
+KafkaEx supports SASL authentication for secure Kafka clusters. Multiple mechanisms are available with flexible configuration options.
+
+## Supported Mechanisms
+
+- **PLAIN** - Simple username/password (requires SSL/TLS)
+- **SCRAM-SHA-256** - Secure challenge-response authentication (Kafka 0.10.2+)
+- **SCRAM-SHA-512** - Secure challenge-response with stronger hash (Kafka 0.10.2+)
+
+## Configuration
+
+### Via Application Config
+
+```elixir
+# config/config.exs
+config :kafka_ex,
+  brokers: [{"localhost", 9292}],
+  use_ssl: true,
+  ssl_options: [
+    verify: :verify_peer,
+    cacertfile: "/path/to/ca-cert"
+  ],
+  sasl: %{
+    mechanism: :scram,
+    username: System.get_env("KAFKA_USERNAME"),
+    password: System.get_env("KAFKA_PASSWORD"),
+    mechanism_opts: %{algo: :sha256}  # :sha256 or :sha512
+  }
+```
+
+### Via Worker Options
+
+```elixir
+opts = [
+  uris: [{"broker1", 9092}, {"broker2", 9092}],
+  use_ssl: true,
+  ssl_options: [verify: :verify_none],
+  auth: KafkaEx.Auth.Config.new(%{
+    mechanism: :plain,
+    username: "alice",
+    password: "secret123"
+  })
+]
+
+{:ok, pid} = KafkaEx.create_worker(:my_worker, opts)
+```
+
+### Docker Compose Setup
+
+The project includes Docker configurations for testing SASL authentication:
+
+```bash
+# Start Kafka with SASL enabled
+docker-compose up -d
+
+# Ports:
+# 9092 - No authentication (SSL)
+# 9192 - SASL/PLAIN (SSL)
+# 9292 - SASL/SCRAM (SSL)
+```
+
+## Security Considerations
+
+- Always use SSL/TLS with PLAIN mechanism - plain text passwords must be encrypted in transit
+- Use environment variables for credentials - never hardcode passwords
+- SCRAM is preferred over PLAIN when both are available
+
+### Minimum Kafka Versions
+
+- PLAIN: Kafka 0.9.0+
+- SCRAM: Kafka 0.10.2+
+
+## Testing with Different Mechanisms
+
+```elixir
+# Test PLAIN authentication
+config :kafka_ex,
+  brokers: [{"localhost", 9192}],
+  use_ssl: true,
+  ssl_options: [verify: :verify_none],
+  sasl: %{mechanism: :plain, username: "test", password: "secret"}
+
+# Test SCRAM-SHA-256
+config :kafka_ex,
+  brokers: [{"localhost", 9292}],
+  use_ssl: true,
+  ssl_options: [verify: :verify_none],
+  sasl: %{
+    mechanism: :scram,
+    username: "test",
+    password: "secret",
+    mechanism_opts: %{algo: :sha256}
+  }
+```
+
+## Integration with Existing Code
+
+SASL authentication is transparent to the rest of your KafkaEx usage:
+
+```elixir
+# Once configured, use KafkaEx normally
+KafkaEx.metadata()
+KafkaEx.produce("my-topic", 0, "message")
+messages = KafkaEx.fetch("my-topic", 0, offset: 0)
+```
+
+## Troubleshooting
+
+- **Connection refused**: Ensure you're using the correct port for SASL (9192 for PLAIN, 9292 for SCRAM in test setup)
+- **Authentication failed**: Check credentials and ensure the user exists in Kafka with the correct SASL mechanism configured
+- **SSL handshake error**: Verify SSL certificates or use verify: :verify_none for testing (not production!)
+- **Unsupported mechanism**: Ensure your Kafka version supports the mechanism (SCRAM requires 0.10.2+)
+
+## Advanced: Custom Authentication
+
+For OAuth or custom mechanisms, implement the `KafkaEx.Auth.Mechanism` behaviour:
+
+```elixir
+defmodule MyAuth do
+  @behaviour KafkaEx.Auth.Mechanism
+
+  def mechanism_name(_), do: "OAUTHBEARER"
+
+  def authenticate(config, send_fun) do
+    # Custom authentication logic
+    :ok
+  end
+end
+```
+
+## Implementation Notes
+
+### Version Compatibility
+
+The SASL implementation handles different Kafka versions appropriately:
+
+- Kafka 0.9.x: Skips API versions call (not supported)
+- Kafka 0.10.0-0.10.1: Queries API versions, supports PLAIN only
+- Kafka 0.10.2+: Full support including SCRAM mechanisms
+
+### Technical Details
+
+- Authentication occurs immediately after socket creation
+- The implementation handles packet mode switching between raw and length-prefixed formats
+- Correlation IDs are used to match requests with responses
+- Server signatures are validated in SCRAM authentication
+- Passwords are never logged and are redacted in inspect output

README.md

@@ -355,6 +355,47 @@ KafkaEx.produce(produce_request)
 
 Compression is handled automatically on the consuming/fetching end.
 
+## SASL Authentication
+
+KafkaEx supports connecting to secure Kafka clusters with SASL mechanisms.
+
+Example:
+
+```elixir
+# config/config.exs
+config :kafka_ex,
+  brokers: [{"localhost", 9292}],
+  use_ssl: true,
+  ssl_options: [verify: :verify_none],
+  sasl: %{
+    mechanism: :scram,
+    username: System.get_env("KAFKA_USERNAME"),
+    password: System.get_env("KAFKA_PASSWORD"),
+    mechanism_opts: %{algo: :sha256}  # or :sha512
+  }
+```
+
+Or via worker options:
+
+```elixir
+{:ok, _pid} = KafkaEx.create_worker(:sasl_worker, [
+  uris: [{"localhost", 9292}],
+  use_ssl: true,
+  ssl_options: [verify: :verify_none],
+  auth: KafkaEx.Auth.Config.new(%{
+    mechanism: :plain,
+    username: "alice",
+    password: "secret123"
+  })
+])
+```
+
+> ✅ Use SSL/TLS with PLAIN (never send passwords in cleartext).  
+> ✅ Prefer SCRAM over PLAIN when supported.  
+> ✅ PLAIN requires Kafka 0.9.0+, SCRAM requires 0.10.2+.  
+
+👉 See [AUTH.md](./AUTH.md) for full details, configuration examples, and troubleshooting tips.
+
 ## Testing
 
 It is strongly recommended to test using the Dockerized test cluster described
@@ -366,13 +407,18 @@ asynchronous issues, the test suite sometimes fails on the first try.
 ### Dockerized Test Cluster
 
 Testing KafkaEx requires a local SSL-enabled Kafka cluster with 3 nodes: one
-node listening on each port 9092, 9093, and 9093.  The easiest way to do this
+node listening on appropriate port. The easiest way to do this
 is using the scripts in
 this repository that utilize [Docker](https://www.docker.com) and
 [Docker Compose](https://www.docker.com/products/docker-compose) (both of which
 are freely available).  This is the method we use for our CI testing of
 KafkaEx.
 
+Ports:
+9092-9094 - No authentication (SSL)
+9192-9194 - SASL/PLAIN (SSL)
+9292-9294 - SASL/SCRAM (SSL)
+
 To launch the included test cluster, run
 
 ```

config/config.exs

@@ -79,6 +79,19 @@ config :kafka_ex,
   # use "kayrock" for the new client
   kafka_version: "0.10.1"
 
+  # SASL Authentication (optional)
+
+  # Configure SASL credentials and mechanism
+  # sasl: %{
+  #   mechanism: :scram,  # :plain or :scram
+  #   username: "kafka_user",         # USE ENV VARS and don't expose secrets
+  #   password: "kafka_password",     # USE ENV VARS and don't expose secrets
+  #   mechanism_opts: %{algorithm: :sha256}  # For SCRAM only, :sha256 or :sha512
+  # }
+  #
+  # Note: SASL/PLAIN requires SSL to be enabled for security
+  
+
 env_config = Path.expand("#{Mix.env()}.exs", __DIR__)
 
 if File.exists?(env_config) do