feat(backend): add SASL/TLS support to the Kafka message queue backend

[harshitg927]

Overview

Adds SASL and TLS support to the Kafka message queue backend.

Current behavior: the Kafka backend only supported PLAINTEXT. The security config fields (securityProtocol, saslMechanism, saslUsername, saslPassword, tlsCaCertPath) were parsed but any non-plaintext value was explicitly rejected - SASL/TLS was deferred.

Future behavior: SASL and TLS are now wired into both the producer and consumer (reader) franz-go clients:

• SASL - PLAIN, SCRAM-SHA-256, and SCRAM-SHA-512 mechanisms.
• TLS - server-authenticated TLS (min TLS 1.2) against the system roots, with an optional custom CA via tlsCaCertPath.
• securityProtocol (PLAINTEXT / SSL / SASL_PLAINTEXT / SASL_SSL) is the single authoritative selector; the PLAINTEXT path is unchanged.

Additional Information

Any additional information that may be useful for reviewers to know

Key design decisions:

• securityProtocol is authoritative - the SASL_ prefix enables SASL and the _SSL suffix enables TLS. The redundant tlsEnabled flag was removed to avoid a config field that could only ever cause an error.
• PLAIN is rejected without TLS - PLAIN transmits the password in cleartext, so it is only allowed over a TLS-encrypted channel (SASL_SSL). SCRAM never exposes the password and is permitted over plaintext.
• Fail-fast validation - inconsistent security settings (missing/invalid mechanism, missing credentials, stray fields that contradict the protocol) are rejected before any client is built. Validation is enforced inside kafkaSecurityOptions, so it applies regardless of which constructor is used.
• No secret leakage - credentials are never logged or included in error messages.
• Out of scope: mutual TLS (client certificates), OAUTHBEARER/AWS SASL, and live credential rotation (creds are captured at startup; rotating a mounted Secret requires a pod restart, the standard k8s flow).

How to Test

Please provide instructions on how to test the changes made in this pull request

# Build, vet, and unit tests (validation logic, no broker required)
go build ./...
go vet ./adapters/backend/v1/... ./config/...
go test -short ./adapters/backend/v1/... ./config/...

# Integration tests against a live broker (requires Docker; uses testcontainers/redpanda)
go test -run 'TestKafkaSecurity_' -count=1 ./adapters/backend/v1/...

Integration coverage added:

• TestKafkaSecurity_SASLRoundTrip - SASL (SASL_PLAINTEXT + SCRAM-SHA-256) produce/consume round-trip.
• TestKafkaSecurity_TLSRoundTrip - TLS (SSL) round-trip against a broker with a self-signed cert.
• TestKafkaSecurity_SASLTLSRoundTrip - the enterprise SASL_SSL path (SASL over TLS).
• TestKafkaSecurity_TLSRejectsUntrustedBroker - negative test proving TLS verification is enforced (untrusted CA ⇒ connection refused).

Note: each test passes individually; running the entire Kafka integration suite in one process can intermittently hit container exited with code 139 (redpanda SIGSEGV at startup) on memory-constrained machines - this is an infra limitation, not a code issue.

Related issues/PRs:

• Related to Support for alternative message queues #151

Checklist before requesting a review

• My code follows the style guidelines of this project
• I have commented on my code, particularly in hard-to-understand areas
• I have performed a self-review of my code
• If it is a core feature, I have added thorough tests.
• New and existing unit tests pass locally with my changes

Please open the PR against the dev branch (Unless the PR contains only documentation changes)

Summary by CodeRabbit

• New Features
  • Added end-to-end Kafka security support for PLAINTEXT, TLS, SASL, and SASL-over-TLS connections.
  • Security settings are now applied when constructing Kafka producer/consumer clients, including optional custom CA trust and SASL mechanisms (PLAIN/SCRAM).
• Bug Fixes
  • Improved Kafka security validation with clearer invalid-case coverage (including lowercase protocol values) and correct acceptance of valid plaintext/SSL/SASL combinations.
  • TLS connections now fail reliably when the broker is not trusted.
• Tests
  • Added integration coverage using a secured Kafka-compatible broker for positive round-trips and a negative untrusted-CA scenario.
• Documentation / Config
  • tlsEnabled has been removed; securityProtocol is the single source of truth for selecting plaintext/TLS/SASL behavior.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 1b6a8bc6-7425-4d33-b480-f311ff11daf4

📥 Commits

Reviewing files that changed from the base of the PR and between 31c076a and 102e0ca.

📒 Files selected for processing (2)

• adapters/backend/v1/kafka_integration_test.go
• adapters/backend/v1/kafka_security_integration_test.go

🚧 Files skipped from review as they are similar to previous changes (1)

• adapters/backend/v1/kafka_security_integration_test.go

---

📝 Walkthrough

Walkthrough

Kafka security configuration now selects plaintext, TLS, and SASL modes through SecurityProtocol. Validation and franz-go option construction support SASL and TLS clients, while producer/reader wiring and Redpanda integration tests cover secure message flows.

Changes

Kafka security support

Layer / File(s)	Summary
Security contract and option construction config/config.go, adapters/backend/v1/kafka_security.go, adapters/backend/v1/factory_test.go	SecurityProtocol becomes the authoritative selector; validation covers protocol, SASL, and TLS consistency, and security options are constructed for supported mechanisms and certificates.
Producer and reader client wiring adapters/backend/v1/kafka.go	Producer and reader creation builds security options and passes them to franz-go clients, returning configuration errors during client construction.
Secured broker round-trip coverage adapters/backend/v1/kafka_security_integration_test.go, adapters/backend/v1/kafka_integration_test.go	Redpanda tests cover SASL, TLS, SASL over TLS, message round trips, untrusted broker certificates, and reliable container cleanup.

Estimated code review effort: 4 (Complex) | ~45 minutes

Sequence Diagram(s)

sequenceDiagram
  participant KafkaConfig
  participant KafkaAdapter
  participant kafkaSecurityOptions
  participant Redpanda
  KafkaConfig->>KafkaAdapter: configure securityProtocol and credentials
  KafkaAdapter->>kafkaSecurityOptions: validate and build SASL/TLS options
  kafkaSecurityOptions-->>KafkaAdapter: return client options
  KafkaAdapter->>Redpanda: produce and consume secured message
  Redpanda-->>KafkaAdapter: return record

Loading

Possibly related PRs

• kubescape/synchronizer#166: Both changes modify Kafka security configuration wiring and validation in the Kafka adapter.

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly matches the main change: adding SASL/TLS support to the Kafka backend.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[harshitg927]

@matthyx PTAL

[matthyx]

Review — blocker: the new TLS integration tests fail CI deterministically

Nice, well-structured change: securityProtocol as the single authoritative selector, fail-fast validation centralized in kafkaSecurityOptions, PLAIN-without-TLS rejected, no secret leakage, and real round-trip + negative TLS coverage. The core logic reviews cleanly and go build/go vet are fine. One thing blocks merge.

Blocker — pr-created / test / Create cross-platform build is red

The three new TLS integration tests fail every run; the redpanda container SIGSEGVs at startup:

wait for readiness: internal check: container exited with code 139
--- FAIL: TestKafkaSecurity_TLSRoundTrip (0.72s)
--- FAIL: TestKafkaSecurity_SASLTLSRoundTrip (0.77s)
--- FAIL: TestKafkaSecurity_TLSRejectsUntrustedBroker (0.79s)
FAIL	github.com/kubescape/synchronizer/adapters/backend/v1	93.406s

(run 30154054261)

This is not the intermittent infra flake the PR description anticipates. In the same run:

• TestKafkaSecurity_SASLRoundTrip (SASL_PLAINTEXT, no TLS) — PASS
• every pre-existing plaintext redpanda test (TestKafkaMessageProducer_*, TestKafkaMessageReader_*) — PASS
• all three tests that pass redpanda.WithTLS(...) — FAIL, code 139, ~0.7s each

So the crash is deterministic and specific to the TLS-configured broker, i.e. the WithTLS(certPEM, keyPEM) bootstrap (self-signed ECDSA cert) makes redpanda v24.2.7 crash on boot in the CI environment — not a random OOM.

These tests are gated only by testing.Short() (requireIntegration → if testing.Short() { t.Skip(...) }), and the build/test job runs go test ./... without -short, so they run unconditionally and keep the required check red.

To unblock, pick one:

1. Fix the TLS broker startup so the container boots in CI — e.g. try a newer redpanda image, an RSA cert instead of ECDSA, or bump the container's memory/--smp/--overprovisioned flags (redpanda/seastar is known to SIGSEGV on constrained CI runners, and the TLS listener pushes it over). This keeps the coverage you added.
2. Move the container-backed security tests behind the integration build tag (the tests/ suite already uses --tags=integration) and run them in a dedicated lane, so the default go test ./... build job stays green. Note that -short gating alone is insufficient here, since the build job doesn't pass -short.

Not blocking (context)

The component-tests (TC01–TC13) failures are unrelated to this change — they fail in ~13s with fatal: Authentication failed for https://github.com/armosec/db-migrations.git because DB_MIGRATIONS_GH_USER is empty for fork PRs (secrets aren't exposed to forks). Those will go green when run from an in-repo branch.

Minor

• Removing the TLSEnabled field is a config-schema change, but it's safe: nothing in the repo (charts/configuration/*.json) references tlsEnabled, and it was previously only ever an error, so no working config regresses. Since it's silently ignored now (unknown mapstructure key) rather than rejected, a one-line mention in the PR/CHANGELOG for anyone who set it would be a courtesy.

[harshitg927]

@matthyx Thanks for the review, this was a real blocker.

I fixed it by updating the TLS test cert helper to generate RSA-2048 certs instead of ECDSA P-256 in synchronizer/adapters/backend/v1/kafka_security_integration_test.go. The Redpanda TLS tests in the testcontainers module use RSA as well, and with that change the broker now starts cleanly in the container-backed TLS cases.

I intentionally kept these tests in the default go test lane rather than moving them behind an integration build tag, so we preserve the TLS round-trip and untrusted-broker coverage on the required check.

Verified locally with:

• go test ./adapters/backend/v1/ -count=1 -run 'TestKafkaSecurity_' -timeout 10m
• go test ./adapters/backend/v1/ -count=1 -run 'TestKafkaSecurity|TestValidateKafka|TestFactory' -short

Also noted on the minor point: tlsEnabled is gone in favor of securityProtocol.

[matthyx]

Re-check after 31c076a — still red, and I found the actual root cause

The RSA cert change didn't help: run 30343208700 fails identically, same container exited with code 139 on the same three TLS tests. My earlier guess that the cert algorithm was to blame was wrong — sorry for sending you down that path. I dug in properly this time and reproduced the underlying defect locally.

Root cause: every redpanda container leaks, because Terminate gets a cancelled context

Both helpers tear the broker down with the test's own cancellable context:

// kafka_security_integration_test.go:47
t.Cleanup(func() { _ = container.Terminate(ctx) })
// kafka_integration_test.go:36
_ = container.Terminate(ctx)

and every test sets that context up as

ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()

Go runs a test's deferred calls before its t.Cleanup functions. So cancel() always fires first, and Terminate is then called with an already-cancelled context. The error is discarded by _ =, so it fails silently:

Terminate() returned error: stop: container stop: Post "http://.../containers/2dee8dd4.../stop?t=10": context canceled
running redpanda containers after: 1
LEAK CONFIRMED: 1 container(s) still running after test cleanup

That's why the CI log prints 🐳 Stopping container: … but never the matching ✅ Container stopped / 🚫 Container terminated lines that a successful teardown emits.

Running the full kafka suite locally on this branch, all 9 brokers stay running, each holding --smp=1 --memory=1G:

still running: 0ff016ebd0d4 Up 1 second
still running: 8a4265250f4a Up 2 seconds
... (9 total)
LEAKED redpanda containers still running after the kafka suite: 9

They're only reaped by ryuk when the test process exits — long after the damage is done.

Why this shows up as "TLS is broken"

The failures line up exactly with container ordering, not with TLS being wrong:

#	test	CI
1–5	TestKafkaMessageProducer_*, TestKafkaMessageReader_*	PASS
6	TestKafkaSecurity_SASLRoundTrip	PASS
7	TestKafkaSecurity_TLSRoundTrip	FAIL 139
8	TestKafkaSecurity_SASLTLSRoundTrip	FAIL 139
9	TestKafkaSecurity_TLSRejectsUntrustedBroker	FAIL 139

By test 7 there are six live brokers still squatting on the runner. The three TLS tests just happen to be numbers 7, 8 and 9, so they're the ones that hit the wall — a 4-core/16 GB GitHub runner runs out of headroom and seastar dies at boot. Nothing about the TLS path itself is faulty, which is also why swapping ECDSA→RSA changed nothing.

This also explains the note in the PR description: individually each test is container #1 and always passes; the whole suite in one process accumulates the leak and blows up. It's not flaky infra.

The fix

Terminate with a context that isn't tied to the test's lifetime:

// kafka_security_integration_test.go:47
t.Cleanup(func() { _ = container.Terminate(context.Background()) })

// kafka_integration_test.go:36
_ = container.Terminate(context.Background())

Verified on this branch — leaked containers go from 9 to 0 and the suite stays green:

--- PASS: TestKafkaSecurity_SASLRoundTrip (1.94s)
--- PASS: TestKafkaSecurity_TLSRoundTrip (1.99s)
--- PASS: TestKafkaSecurity_SASLTLSRoundTrip (2.00s)
--- PASS: TestKafkaSecurity_TLSRejectsUntrustedBroker (1.52s)
LEAKED redpanda containers still running after the kafka suite: 0
ok  	github.com/kubescape/synchronizer/adapters/backend/v1	25.020s

Worth surfacing the teardown error instead of swallowing it (require.NoError in the cleanup, or at least t.Logf) so the next leak isn't silent.

One caveat, so you can weigh it: I could not reproduce the code-139 crash itself on my machine (24 cores / 30 GB simply absorbs nine stray brokers; I also tried pinning the container to 2 CPUs / 2 GB and it booted fine). The leak and the fix are directly measured; the leak → exhaustion → SIGSEGV link on the 4-core/16 GB runner is inference from the ordering above. CI is the confirmation — but the leak is a real defect regardless of whether it turns out to be the whole story.

Note the leak pattern predates this PR (kafka_integration_test.go came in with #166); this PR is just what pushes the container count past what the runner tolerates. Fixing both helpers is the right move.

Still unrelated

component-tests (TC01–TC13) continue to fail on fatal: Authentication failed for .../db-migrations.git — fork PRs don't get repo secrets. Not yours to fix.