Skip to content

Kafka ScramAuth

onetl.connection.db_connection.kafka.kafka_scram_auth.KafkaScramAuth

Bases: KafkaAuth, GenericOptions

Connect to Kafka using sasl.mechanism="SCRAM-SHA-256" or sasl.mechanism="SCRAM-SHA-512".

For more details see Kafka Documentation.

Added in 0.9.0

Examples

Auth in Kafka with SCRAM-SHA-256 mechanism:

from onetl.connection import Kafka

auth = Kafka.ScramAuth(
    user="me",
    password="abc",
    digest="SHA-256",
)
Auth in Kafka with SCRAM-SHA-512 mechanism and some custom SASL options passed to Kafka client config:

from onetl.connection import Kafka

auth = Kafka.ScramAuth.parse(
    {
        "user": "me",
        "password": "abc",
        "digest": "SHA-512",
        # options with `sasl.login.` prefix are passed to Kafka client config as-is
        "sasl.login.class": "com.example.CustomScramLogin",
    }
)
Source code in onetl/connection/db_connection/kafka/kafka_scram_auth.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
class KafkaScramAuth(KafkaAuth, GenericOptions):
    """
    Connect to Kafka using `sasl.mechanism="SCRAM-SHA-256"` or `sasl.mechanism="SCRAM-SHA-512"`.

    For more details see [Kafka Documentation](https://kafka.apache.org/documentation/#security_sasl_scram_clientconfig).

    !!! success "Added in 0.9.0"

    Examples
    --------

    Auth in Kafka with `SCRAM-SHA-256` mechanism:

    ```python
    from onetl.connection import Kafka

    auth = Kafka.ScramAuth(
        user="me",
        password="abc",
        digest="SHA-256",
    )
    ```
    Auth in Kafka with `SCRAM-SHA-512` mechanism and some custom SASL options passed to Kafka client config:

    ```python
    from onetl.connection import Kafka

    auth = Kafka.ScramAuth.parse(
        {
            "user": "me",
            "password": "abc",
            "digest": "SHA-512",
            # options with `sasl.login.` prefix are passed to Kafka client config as-is
            "sasl.login.class": "com.example.CustomScramLogin",
        }
    )
    ```
    """

    user: str = Field(alias="username")
    password: SecretStr
    digest: Literal["SHA-256", "SHA-512"]

    class Config:
        strip_prefixes = ("kafka.",)
        # https://kafka.apache.org/documentation/#producerconfigs_sasl.login.class
        known_options = frozenset(("sasl.login.*",))
        prohibited_options = frozenset(("sasl.mechanism", "sasl.jaas.config"))
        extra = "allow"

    def get_jaas_conf(self) -> str:
        return (
            "org.apache.kafka.common.security.scram.ScramLoginModule required "
            f'username="{self.user}" '
            f'password="{self.password.get_secret_value()}";'
        )

    def get_options(self, kafka: Kafka) -> dict:
        result = {
            key: value for key, value in self.dict(by_alias=True, exclude_none=True).items() if key.startswith("sasl.")
        }
        result.update(
            {
                "sasl.mechanism": f"SCRAM-{self.digest}",
                "sasl.jaas.config": self.get_jaas_conf(),
            },
        )
        return stringify(result)

    def cleanup(self, kafka: Kafka) -> None:
        # nothing to cleanup
        pass

parse(options) classmethod

If a parameter inherited from the ReadOptions class was passed, then it will be returned unchanged. If a Dict object was passed it will be converted to ReadOptions.

Otherwise, an exception will be raised

Source code in onetl/impl/generic_options.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
@classmethod
def parse(
    cls,
    options: GenericOptions | dict | None,
) -> Self:
    """
    If a parameter inherited from the ReadOptions class was passed, then it will be returned unchanged.
    If a Dict object was passed it will be converted to ReadOptions.

    Otherwise, an exception will be raised
    """

    if not options:
        return cls()

    if isinstance(options, dict):
        return cls.parse_obj(options)

    if not isinstance(options, cls):
        msg = f"{options.__class__.__name__} is not a {cls.__name__} instance"
        raise TypeError(msg)

    return options