# Authentication

> SASL PLAIN, SCRAM, and OAUTHBEARER for managed Kafka: Confluent Cloud, AWS MSK, Azure Event Hubs.

Source: https://docs.pyrula.io/kafka/authentication/

PLAIN, SCRAM-SHA-256, SCRAM-SHA-512 and OAUTHBEARER, same fields on `KafkaConfig` and
`KafkaConsumerConfig`. A wrong password or rejected token raises `KafkaConnectionError`
with the broker's SASL error at construction.

With several `bootstrap_servers`, construction tries each in turn and uses the first that
accepts, each getting an equal share of the bootstrap timeout (`request_timeout_ms` for a
producer, `session_timeout_ms` for a consumer). A broker that rejects the credentials ends
the attempt at once (`SASL authentication failed`). If every address fails otherwise, the
error names each one.

A listener that closes the connection during the handshake usually speaks another protocol
than the client, such as an SSL client on a PLAINTEXT port. The error then says to check
`security_protocol` (`security.protocol` in the confluent shim). A PLAINTEXT client on a
SASL_PLAINTEXT listener gets no such hint.

## Username and password

```python
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="...",
)
```

`SASL_PLAINTEXT` sends credentials in the clear; use `SASL_SSL` off a trusted network.

## OAUTHBEARER

Supply a callback that returns a token and its expiry. The runtime caches the token and
calls back only near expiry, however many connections it opens.

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

A provider that needs SASL extensions (RFC 7628), such as Confluent Cloud's
`logicalCluster` and `identityPoolId`, returns confluent-kafka's
`(token, expiry_epoch_secs, principal, extensions)` with `extensions` a `dict[str, str]`;
the three-tuple is accepted too. The principal is not sent, as in librdkafka. Keys must be
ASCII letters (not `auth`) and values visible ASCII, space, tab, CR or LF, or empty;
anything else fails authentication with the reason.

Tokens refresh within 30 seconds (or 20 percent of remaining lifetime) of expiry. There is
no mid-connection re-authentication: the broker closes a connection whose token expired
and the reconnect picks up a fresh one.

### OIDC client credentials

For Confluent Cloud, Azure Event Hubs or Keycloak, `make_oidc_cb` does the
client-credentials grant against your token endpoint (stdlib only, 10 s HTTP timeout):

```python
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

`make_msk_iam_cb` wraps the AWS SigV4 signer. Install `"pyrula[kafka,msk]"`; credentials
resolve through the boto3 chain.

```python
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"),
)
```

## TLS

`ssl_ca_location` overrides the system CA bundle. mTLS uses `ssl_certificate_location` and
`ssl_key_location`, set together.

```python
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.*` keys, including the
`oauth_cb(config_str)` convention and its four-tuple answer. librdkafka's built-in token
retriever (`sasl.oauthbearer.config` alone) has no equivalent and is rejected with a
pointer to the helpers above.