Skip to content

fix(kafka): add strimzi-no-tls and redpanda-no-tls providers #181

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
catinspace-au merged 1 commit into from
Nov 11, 2026
Merged

fix(kafka): add strimzi-no-tls and redpanda-no-tls providers #181

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions docs/transport/kafka.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -220,6 +220,18 @@ Every blocking metadata, watermark, committed-offset and offsets-for-times call
| `internal_sasl_scram(brokers, user, pw)` | SASL_PLAINTEXT + SCRAM-SHA-512 (in-cluster, TLS-terminated upstream). |
| `get_default_config()` | `config_from_env()` + `ADMIN_DEFAULTS`. |

`scalo.kafka.providers.provider_config(provider, brokers, user, pw)` derives the protocol and mechanism from a provider key, so neither is hand-set:

| Provider | `security.protocol` | `sasl.mechanisms` |
|---|---|---|
| `strimzi`, `redpanda`, `msk`, `redpanda-cloud` | `SASL_SSL` | `SCRAM-SHA-512` |
| `strimzi-no-tls`, `redpanda-no-tls` | `SASL_PLAINTEXT` | `SCRAM-SHA-512` |
| `confluent-cloud` | `SASL_SSL` | `PLAIN` |
| `msk_iam` | `SASL_SSL` | `OAUTHBEARER` |
| `plaintext` | `PLAINTEXT` | none |

A `-no-tls` key names a self-hosted broker's TLS-off listener. Only SCRAM gets one, because its challenge-response never sends the password. `scalo.kafka.contract.DEFAULT_ALLOWED` leaves both out.

### Identity propagation

Headers carry whatever you pass to `producer.send(headers=...)`. The
Expand Down
23 changes: 19 additions & 4 deletions src/scalo/kafka/providers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@

# (security_protocol, sasl_mechanism) pairs.
_SCRAM = ("SASL_SSL", "SCRAM-SHA-512")
_SCRAM_NO_TLS = ("SASL_PLAINTEXT", "SCRAM-SHA-512")
_PLAIN = ("SASL_SSL", "PLAIN")

# Local dev with no auth.
Expand All @@ -76,6 +77,10 @@
DERIVATION: dict[str, tuple[str, str]] = {
"strimzi": _SCRAM, # self-hosted Apache Kafka / Strimzi
"redpanda": _SCRAM, # self-hosted Redpanda
# The same brokers on a TLS-off listener. SCRAM is challenge-response, so the
# password never crosses it. No PLAIN or OAUTHBEARER provider has such a key.
"strimzi-no-tls": _SCRAM_NO_TLS,
"redpanda-no-tls": _SCRAM_NO_TLS,
"msk": _SCRAM, # AWS MSK provisioned (SCRAM; serverless = msk_iam)
"redpanda-cloud": _SCRAM, # Redpanda Cloud (SASL_SSL + SCRAM-256/512)
"confluent-cloud": _PLAIN, # Confluent Cloud: no SCRAM -> API-key PLAIN over TLS
Expand All @@ -93,6 +98,8 @@ def derive(provider: str) -> tuple[str, str]:
"""Return ``(security_protocol, sasl_mechanism)`` for a provider.

``plaintext`` -> ``("PLAINTEXT", "")`` (local dev, no auth).
``*-no-tls`` -> ``("SASL_PLAINTEXT", "SCRAM-SHA-512")`` (a self-hosted
broker's TLS-off listener).
``msk_iam`` -> ``("SASL_SSL", "OAUTHBEARER")`` (quarantined IAM token path;
the caller wires the AWS MSK IAM callback, not user/pass).
unknown -> ``KafkaProviderError``.
Expand Down Expand Up @@ -163,7 +170,7 @@ def provider_config(
# section is the FULL abstraction: one open `KafkaProvider` behind which
# every provider-specific concern (auth, capabilities, metadata plane, schema
# registry, transport tuning) is normalised. `KnownProvider` implements it
# for the seven built-in providers using the SAME `derive()` table above, so
# for the built-in providers using the SAME `derive()` table above, so
# there is exactly one source of truth for the auth facts.


Expand Down Expand Up @@ -268,6 +275,8 @@ class KnownProvider(Enum):

STRIMZI = "strimzi" # Self-hosted Apache Kafka / Strimzi - SCRAM-SHA-512
REDPANDA = "redpanda" # Self-hosted Redpanda - SCRAM-SHA-512
STRIMZI_NO_TLS = "strimzi-no-tls" # Strimzi's TLS-off listener - SASL_PLAINTEXT + SCRAM-SHA-512
REDPANDA_NO_TLS = "redpanda-no-tls" # Redpanda's TLS-off listener - SASL_PLAINTEXT + SCRAM-SHA-512
MSK = "msk" # AWS MSK provisioned - SCRAM-SHA-512 (serverless is IAM-only: MSK_IAM)
REDPANDA_CLOUD = "redpanda-cloud" # Redpanda Cloud - SASL_SSL + SCRAM-SHA-512
CONFLUENT_CLOUD = "confluent-cloud" # Confluent Cloud - no SCRAM, API-key PLAIN over TLS
Expand Down Expand Up @@ -321,7 +330,7 @@ def capabilities(self) -> ProviderCapabilities:
requires_tls=True,
auth_kind=AuthKind.USER_PASSWORD,
)
if self in (KnownProvider.STRIMZI, KnownProvider.REDPANDA):
if self in _SELF_HOSTED:
return self_hosted
if self is KnownProvider.MSK:
# Provisioned MSK: fixed broker-hour capacity (MSK Serverless is
Expand All @@ -341,7 +350,7 @@ def metadata_mode(self) -> MetadataMode:
"""Return this provider's metadata plane."""
# Redpanda is KRaft-native (no ZooKeeper); Strimzi + local dev default
# to KRaft on modern Kafka; provisioned MSK is KRaft-forward.
if self in (KnownProvider.REDPANDA, KnownProvider.STRIMZI, KnownProvider.PLAINTEXT, KnownProvider.MSK):
if self in _SELF_HOSTED or self in (KnownProvider.PLAINTEXT, KnownProvider.MSK):
return MetadataMode.KRAFT
# The clouds hide their metadata plane entirely.
return MetadataMode.MANAGED
Expand All @@ -350,10 +359,16 @@ def schema_registry(self) -> SchemaRegistry:
"""Return the schema registry this provider bundles, if any."""
if self is KnownProvider.CONFLUENT_CLOUD:
return SchemaRegistry.CONFLUENT
if self in (KnownProvider.REDPANDA, KnownProvider.REDPANDA_CLOUD):
if self in (KnownProvider.REDPANDA, KnownProvider.REDPANDA_NO_TLS, KnownProvider.REDPANDA_CLOUD):
return SchemaRegistry.REDPANDA
return SchemaRegistry.NONE

def transport_overrides(self) -> dict[str, str]:
"""Return no overrides, because no built-in provider needs transport tuning."""
return {}


# Brokers the operator runs, on either listener.
_SELF_HOSTED = frozenset(
{KnownProvider.STRIMZI, KnownProvider.REDPANDA, KnownProvider.STRIMZI_NO_TLS, KnownProvider.REDPANDA_NO_TLS}
)
8 changes: 7 additions & 1 deletion tests/unit/test_kafka_abstraction.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,8 @@
CANONICAL_TABLE = [
("strimzi", "SASL_SSL", "SCRAM-SHA-512"),
("redpanda", "SASL_SSL", "SCRAM-SHA-512"),
("strimzi-no-tls", "SASL_PLAINTEXT", "SCRAM-SHA-512"),
("redpanda-no-tls", "SASL_PLAINTEXT", "SCRAM-SHA-512"),
("msk", "SASL_SSL", "SCRAM-SHA-512"),
("redpanda-cloud", "SASL_SSL", "SCRAM-SHA-512"),
("confluent-cloud", "SASL_SSL", "PLAIN"),
Expand Down Expand Up @@ -70,7 +72,7 @@ def test_managed_providers_are_always_on(self):
assert caps.requires_tls, f"{key} mandates TLS"

def test_self_hosted_is_not_managed(self):
for key in ("strimzi", "redpanda", "plaintext"):
for key in ("strimzi", "redpanda", "strimzi-no-tls", "redpanda-no-tls", "plaintext"):
caps = KnownProvider.parse(key).capabilities()
assert not caps.managed, f"{key} is self-hosted"
assert not caps.always_on
Expand Down Expand Up @@ -101,6 +103,8 @@ def test_kraft_native_and_self_hosted(self):
assert KnownProvider.STRIMZI.metadata_mode() == MetadataMode.KRAFT
assert KnownProvider.MSK.metadata_mode() == MetadataMode.KRAFT
assert KnownProvider.PLAINTEXT.metadata_mode() == MetadataMode.KRAFT
assert KnownProvider.STRIMZI_NO_TLS.metadata_mode() == MetadataMode.KRAFT
assert KnownProvider.REDPANDA_NO_TLS.metadata_mode() == MetadataMode.KRAFT

def test_clouds_are_managed_hidden(self):
for key in ("confluent-cloud", "redpanda-cloud", "msk_iam"):
Expand All @@ -112,6 +116,8 @@ def test_schema_registry_kind_per_provider(self):
assert KnownProvider.CONFLUENT_CLOUD.schema_registry() == SchemaRegistry.CONFLUENT
assert KnownProvider.REDPANDA_CLOUD.schema_registry() == SchemaRegistry.REDPANDA
assert KnownProvider.REDPANDA.schema_registry() == SchemaRegistry.REDPANDA
assert KnownProvider.REDPANDA_NO_TLS.schema_registry() == SchemaRegistry.REDPANDA
assert KnownProvider.STRIMZI_NO_TLS.schema_registry() == SchemaRegistry.NONE
assert KnownProvider.STRIMZI.schema_registry() == SchemaRegistry.NONE
assert KnownProvider.MSK.schema_registry() == SchemaRegistry.NONE
assert KnownProvider.PLAINTEXT.schema_registry() == SchemaRegistry.NONE
Expand Down
8 changes: 8 additions & 0 deletions tests/unit/test_kafka_providers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,8 @@
CANONICAL_TABLE = {
"strimzi": ("SASL_SSL", "SCRAM-SHA-512"),
"redpanda": ("SASL_SSL", "SCRAM-SHA-512"),
"strimzi-no-tls": ("SASL_PLAINTEXT", "SCRAM-SHA-512"),
"redpanda-no-tls": ("SASL_PLAINTEXT", "SCRAM-SHA-512"),
"msk": ("SASL_SSL", "SCRAM-SHA-512"),
"redpanda-cloud": ("SASL_SSL", "SCRAM-SHA-512"),
"confluent-cloud": ("SASL_SSL", "PLAIN"),
Expand Down Expand Up @@ -72,6 +74,12 @@ def test_owned_broker_builds_scram(self):
cfg = provider_config("strimzi", "b:9092", "u", "p")
assert cfg["sasl.mechanisms"] == "SCRAM-SHA-512"

def test_no_tls_listener_builds_scram_over_plaintext(self):
cfg = provider_config("strimzi-no-tls", "b:9092", "u", "p", verify_ssl=False)
assert cfg["security.protocol"] == "SASL_PLAINTEXT"
assert cfg["sasl.mechanisms"] == "SCRAM-SHA-512"
assert "enable.ssl.certificate.verification" not in cfg

def test_plaintext_has_no_sasl_keys(self):
cfg = provider_config("plaintext", "b:9092")
assert cfg["security.protocol"] == "PLAINTEXT"
Expand Down
Loading