Authentication

Four mechanisms: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, and OAUTHBEARER. The same fields work on KafkaConfig and KafkaConsumerConfig.

Auth failures fail fast. A wrong password or rejected token raises KafkaConnectionError with the broker’s actual SASL error at construction time, not a timeout thirty seconds later.

Username and password

PLAIN and SCRAM need a username, a password, and a security protocol that includes SASL:

from pyrula.kafka import KafkaConfig

cfg = KafkaConfig(
    bootstrap_servers="broker:9093",
    security_protocol="SASL_SSL",
    sasl_mechanism="SCRAM-SHA-512",   # or "PLAIN", "SCRAM-SHA-256"
    sasl_username="svc-orders",
    sasl_password="...",
)

Use SASL_SSL against anything reachable from outside your network. SASL_PLAINTEXT sends credentials in the clear and belongs in local development only.

OAUTHBEARER

For brokers that authenticate with OAuth 2.0 bearer tokens. You supply a callback that returns a token and its expiry; the runtime caches it and only calls back when the token is close to expiring, no matter how many broker connections it opens.

cfg = KafkaConfig(
    bootstrap_servers="broker:9092",
    security_protocol="SASL_SSL",
    sasl_mechanism="OAUTHBEARER",
    oauth_cb=my_token_source,   # () -> (token: str, expiry_epoch_secs: float)
)

You rarely need to write the callback yourself. Two helpers cover the common providers.

OIDC client credentials (Confluent Cloud, Azure Event Hubs, Keycloak)

make_oidc_cb does the standard client-credentials grant against your identity provider’s token endpoint. Standard library only, no extra dependencies:

from pyrula.kafka.oauth import make_oidc_cb

cfg = KafkaConfig(
    bootstrap_servers="pkc-xxxxx.us-east-1.aws.confluent.cloud:9092",
    security_protocol="SASL_SSL",
    sasl_mechanism="OAUTHBEARER",
    oauth_cb=make_oidc_cb(
        token_url="https://login.example.com/oauth2/token",
        client_id="my-client",
        client_secret="...",
        scope="kafka",          # optional
    ),
)

AWS MSK IAM

MSK’s IAM auth is OAUTHBEARER with a SigV4-signed token. make_msk_iam_cb wraps the official AWS signer; install the extra first:

pip install pyrula-kafka[msk]

from pyrula.kafka.oauth import make_msk_iam_cb

cfg = KafkaConfig(
    bootstrap_servers="b-1.cluster.xxxxx.kafka.us-east-1.amazonaws.com:9098",
    security_protocol="SASL_SSL",
    sasl_mechanism="OAUTHBEARER",
    oauth_cb=make_msk_iam_cb(region="us-east-1"),
)

Credentials resolve through the normal boto3 chain: environment variables, ~/.aws/credentials, instance profiles.

Token lifecycle

The callback’s contract is () -> (token, expiry_epoch_secs). Tokens are cached and refreshed when within 30 seconds (or 20 percent of remaining lifetime) of expiry. Brokers authenticate per connection; if a token expires while a connection is open, the broker eventually closes it and the reconnect picks up a fresh token. There is no mid-connection re-authentication.

TLS

ssl_ca_location overrides the system CA bundle. Client certificates (mTLS) use ssl_certificate_location and ssl_key_location, which must be set together.

cfg = KafkaConfig(
    bootstrap_servers="broker:9094",
    security_protocol="SSL",
    ssl_ca_location="/etc/pki/kafka-ca.pem",
    ssl_certificate_location="/etc/pki/client.pem",
    ssl_key_location="/etc/pki/client.key",
)

Migrating from confluent-kafka

pyrula.kafka.compat.confluent translates sasl.* config keys, including the oauth_cb(config_str) callback convention. One difference: librdkafka’s built-in token retriever (sasl.oauthbearer.config on its own) has no equivalent here, so that combination is rejected with a pointer to the helpers above rather than silently dropped.