diff --git a/docs/transport/kafka.md b/docs/transport/kafka.md
index b5cfc61f..42c13d37 100644
--- a/docs/transport/kafka.md
+++ b/docs/transport/kafka.md
@@ -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
diff --git a/src/scalo/kafka/providers.py b/src/scalo/kafka/providers.py
index d0c250d4..b8b429c2 100644
--- a/src/scalo/kafka/providers.py
+++ b/src/scalo/kafka/providers.py
@@ -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.
@@ -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
@@ -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``.
@@ -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.
 
 
@@ -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
@@ -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
@@ -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
@@ -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}
+)
diff --git a/tests/unit/test_kafka_abstraction.py b/tests/unit/test_kafka_abstraction.py
index 9906ffa4..8c5cc63e 100644
--- a/tests/unit/test_kafka_abstraction.py
+++ b/tests/unit/test_kafka_abstraction.py
@@ -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"),
@@ -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
@@ -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"):
@@ -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
diff --git a/tests/unit/test_kafka_providers.py b/tests/unit/test_kafka_providers.py
index 184bbe0d..9cffc9e8 100644
--- a/tests/unit/test_kafka_providers.py
+++ b/tests/unit/test_kafka_providers.py
@@ -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"),
@@ -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"