Kafka SSLProtocol

onetl.connection.db_connection.kafka.kafka_ssl_protocol.KafkaSSLProtocol

Bases: KafkaProtocol, GenericOptions

Connect to Kafka using SSL or SASL_SSL security protocols.

For more details see:

• Kafka Documentation
• IBM Documentation
• How to use PEM Certificates with Kafka

Added in 0.9.0

Examples

TLS (verify only server public certificate)mTLS (mutual certificate check of client and server)Custom Kafka client options

Pass PEM certificate as files located on Spark driver host:

from pathlib import Path

# Just read existing files located on host, and pass key and certificates as strings
protocol = Kafka.SSLProtocol(
    truststore_type="PEM",
    truststore_certificates=Path("/path/to/server.crt").read_text(),
)

Pass PEM certificate as raw string:

protocol = Kafka.SSLProtocol(
    truststore_type="PEM",
    truststore_certificates="-----BEGIN CERTIFICATE...\n...END CERTIFICATE-----",
)

Pass PEM key and certificates as files located on Spark driver host:

from pathlib import Path

# Just read existing files located on host, and pass key and certificates as strings
protocol = Kafka.SSLProtocol(
    keystore_type="PEM",
    keystore_certificate_chain=Path("path/to/user.crt").read_text(),
    keystore_key=Path("path/to/user.key").read_text(),
    truststore_type="PEM",
    truststore_certificates=Path("/path/to/server.crt").read_text(),
)

Pass PEM key and certificates as raw strings:

protocol = Kafka.SSLProtocol(
    keystore_type="PEM",
    keystore_certificate_chain="-----BEGIN CERTIFICATE...\n...END CERTIFICATE-----",
    keystore_key="-----BEGIN PRIVATE KEY...\n...END PRIVATE KEY-----",
    truststore_type="PEM",
    truststore_certificates="-----BEGIN CERTIFICATE...\n...END CERTIFICATE-----",
)

protocol = Kafka.SSLProtocol.parse(
    {
        # Just the same options as above, but using Kafka config naming with dots
        "ssl.keystore.type": "PEM",
        "ssl.keystore.certificate_chain": "-----BEGIN CERTIFICATE...\n...END CERTIFICATE-----",
        "ssl.keystore.key": "-----BEGIN PRIVATE KEY...\n...END PRIVATE KEY-----",
        "ssl.truststore.type": "PEM",
        "ssl.truststore.certificates": "-----BEGIN CERTIFICATE...\n...END CERTIFICATE-----",
        # Any option starting from "ssl." is passed to Kafka client as-is
        "ssl.protocol": "TLSv1.3",
    }
)

Not recommended

These options are error-prone and have several drawbacks, so it is not recommended to use them.

Passing PEM certificates as files:

• ENCRYPT user.key file with password "some password" using PKCS#8 scheme.
• Save encrypted key to file /path/to/user/encrypted_key_with_certificate_chain.pem.
• Then append user certificate to the end of this file.
• Deploy this file (and server certificate too) to EVERY host Spark could run (both driver and executors).
• Then pass file locations and password for key decryption to options below.

protocol = Kafka.SSLProtocol(
    keystore_type="PEM",
    keystore_location="/path/to/user/encrypted_key_with_certificate_chain.pem",
    key_password="some password",
    truststore_type="PEM",
    truststore_location="/path/to/server.crt",
)

Passing JKS (Java Key Store) location:

• Add user key and certificate to JKS keystore.
• Add server certificate to JKS truststore.
• This should be done on EVERY host Spark could run (both driver and executors).
• Pass keystore and truststore paths to options below, as well as passwords for accessing these stores:

protocol = Kafka.SSLProtocol(
    keystore_type="JKS",
    keystore_location="/usr/lib/jvm/default/lib/security/keystore.jks",
    keystore_password="changeit",
    truststore_type="JKS",
    truststore_location="/usr/lib/jvm/default/lib/security/truststore.jks",
    truststore_password="changeit",
)

Source code in onetl/connection/db_connection/kafka/kafka_ssl_protocol.py

class KafkaSSLProtocol(KafkaProtocol, GenericOptions):
    """
    Connect to Kafka using `SSL` or `SASL_SSL` security protocols.

    For more details see:

    * [Kafka Documentation](https://kafka.apache.org/documentation/#producerconfigs_ssl.keystore.location)
    * [IBM Documentation](https://www.ibm.com/docs/en/cloud-paks/cp-biz-automation/19.0.x?topic=fcee-kafka-using-ssl-kerberos-authentication)
    * [How to use PEM Certificates with Kafka](https://codingharbour.com/apache-kafka/using-pem-certificates-with-apache-kafka/)

    !!! success "Added in 0.9.0"

    Examples
    --------

    === "TLS (verify only server public certificate)"

        Pass PEM certificate as files located on Spark driver host:

        ```python
        from pathlib import Path

        # Just read existing files located on host, and pass key and certificates as strings
        protocol = Kafka.SSLProtocol(
            truststore_type="PEM",
            truststore_certificates=Path("/path/to/server.crt").read_text(),
        )
        ```
        Pass PEM certificate as raw string:

        ```python
        protocol = Kafka.SSLProtocol(
            truststore_type="PEM",
            truststore_certificates="-----BEGIN CERTIFICATE...\\n...END CERTIFICATE-----",
        )
        ```
    === "mTLS (mutual certificate check of client and server)"

        Pass PEM key and certificates as files located on Spark driver host:

        ```python
        from pathlib import Path

        # Just read existing files located on host, and pass key and certificates as strings
        protocol = Kafka.SSLProtocol(
            keystore_type="PEM",
            keystore_certificate_chain=Path("path/to/user.crt").read_text(),
            keystore_key=Path("path/to/user.key").read_text(),
            truststore_type="PEM",
            truststore_certificates=Path("/path/to/server.crt").read_text(),
        )
        ```
        Pass PEM key and certificates as raw strings:

        ```python
        protocol = Kafka.SSLProtocol(
            keystore_type="PEM",
            keystore_certificate_chain="-----BEGIN CERTIFICATE...\\n...END CERTIFICATE-----",
            keystore_key="-----BEGIN PRIVATE KEY...\\n...END PRIVATE KEY-----",
            truststore_type="PEM",
            truststore_certificates="-----BEGIN CERTIFICATE...\\n...END CERTIFICATE-----",
        )
        ```
    === "Custom Kafka client options"

        ```python
        protocol = Kafka.SSLProtocol.parse(
            {
                # Just the same options as above, but using Kafka config naming with dots
                "ssl.keystore.type": "PEM",
                "ssl.keystore.certificate_chain": "-----BEGIN CERTIFICATE...\\n...END CERTIFICATE-----",
                "ssl.keystore.key": "-----BEGIN PRIVATE KEY...\\n...END PRIVATE KEY-----",
                "ssl.truststore.type": "PEM",
                "ssl.truststore.certificates": "-----BEGIN CERTIFICATE...\\n...END CERTIFICATE-----",
                # Any option starting from "ssl." is passed to Kafka client as-is
                "ssl.protocol": "TLSv1.3",
            }
        )
        ```
    ??? note "Not recommended"

        These options are error-prone and have several drawbacks, so it is not recommended to use them.

        Passing PEM certificates as files:

        * ENCRYPT `user.key` file with password `"some password"` [using
          PKCS#8 scheme](https://www.mkssoftware.com/docs/man1/openssl_pkcs8.1.asp).
        * Save encrypted key to file `/path/to/user/encrypted_key_with_certificate_chain.pem`.
        * Then append user certificate to the end of this file.
        * Deploy this file (and server certificate too) to **EVERY** host Spark could run (both driver and executors).
        * Then pass file locations and password for key decryption to options below.

        ```python
        protocol = Kafka.SSLProtocol(
            keystore_type="PEM",
            keystore_location="/path/to/user/encrypted_key_with_certificate_chain.pem",
            key_password="some password",
            truststore_type="PEM",
            truststore_location="/path/to/server.crt",
        )
        ```
        Passing JKS (Java Key Store) location:

        * [Add user key and certificate to JKS keystore](https://stackoverflow.com/a/4326346).
        * [Add server certificate to JKS truststore](https://stackoverflow.com/a/373307).
        * This should be done on **EVERY** host Spark could run (both driver and executors).
        * Pass keystore and truststore paths to options below, as well as passwords for accessing these stores:

        ```python
        protocol = Kafka.SSLProtocol(
            keystore_type="JKS",
            keystore_location="/usr/lib/jvm/default/lib/security/keystore.jks",
            keystore_password="changeit",
            truststore_type="JKS",
            truststore_location="/usr/lib/jvm/default/lib/security/truststore.jks",
            truststore_password="changeit",
        )
        ```
    """

    keystore_type: Optional[str] = Field(  # type: ignore[literal-required]
        default=None,
        alias=avoid_alias("ssl.keystore.type"),
    )
    keystore_location: Optional[LocalPath] = Field(  # type: ignore[literal-required]
        default=None,
        alias=avoid_alias("ssl.keystore.location"),
    )
    keystore_password: Optional[SecretStr] = Field(  # type: ignore[literal-required]
        default=None,
        alias=avoid_alias("ssl.keystore.password"),
    )
    keystore_certificate_chain: Optional[str] = Field(  # type: ignore[literal-required]
        default=None,
        alias=avoid_alias("ssl.keystore.certificate.chain"),
        repr=False,
    )
    keystore_key: Optional[SecretStr] = Field(  # type: ignore[literal-required]
        default=None,
        alias=avoid_alias("ssl.keystore.key"),
    )

    # https://knowledge.informatica.com/s/article/145442?language=en_US
    key_password: Optional[SecretStr] = Field(  # type: ignore[literal-required]
        default=None,
        alias=avoid_alias("ssl.key.password"),
    )
    truststore_type: str = Field(alias=avoid_alias("ssl.truststore.type"))  # type: ignore[literal-required]
    truststore_location: Optional[LocalPath] = Field(  # type: ignore[literal-required]
        default=None,
        alias=avoid_alias("ssl.truststore.location"),
    )
    truststore_password: Optional[SecretStr] = Field(  # type: ignore[literal-required]
        default=None,
        alias=avoid_alias("ssl.truststore.password"),
    )
    truststore_certificates: Optional[str] = Field(  # type: ignore[literal-required]
        default=None,
        alias=avoid_alias("ssl.truststore.certificates"),
        repr=False,
    )

    class Config:
        known_options = frozenset(("ssl.*",))
        strip_prefixes = ("kafka.",)
        extra = "allow"

    def get_options(self, kafka: Kafka) -> dict:
        result = self.dict(by_alias=True, exclude_none=True)
        if kafka.auth:
            result["security.protocol"] = "SASL_SSL"
        else:
            result["security.protocol"] = "SSL"
        return stringify(result)

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

    @validator("keystore_location", "truststore_location")
    def validate_path(cls, value: LocalPath) -> Path:
        return is_file_readable(value)

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

@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