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 <https://kafka.apache.org/documentation/#security_sasl_scram_clientconfig>_.

.. versionadded:: 0.9.0

Examples

Auth in Kafka with SCRAM-SHA-256 mechanism:

.. code:: 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:

.. code:: 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",
    }
)
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
94
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>`_.

    .. versionadded:: 0.9.0

    Examples
    --------

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

    .. code:: 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:

    .. code:: 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 = {"sasl.login.*"}
        prohibited_options = {"sasl.mechanism", "sasl.jaas.config"}
        extra = "allow"

    def get_jaas_conf(self) -> str:  # noqa: WPS473
        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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
@classmethod
def parse(
    cls: type[T],
    options: GenericOptions | dict | None,
) -> T:
    """
    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):
        raise TypeError(
            f"{options.__class__.__name__} is not a {cls.__name__} instance",
        )

    return options