fix: SASL_SSL without certs (env too long)
This commit is contained in:
+13
-41
@@ -7,9 +7,6 @@ Env vars (set by Terraform):
|
||||
KAFKA_TOPIC — topic name
|
||||
KAFKA_USERNAME — SASL username
|
||||
KAFKA_PASSWORD — SASL password
|
||||
KAFKA_CA_CRT — CA certificate (PEM)
|
||||
KAFKA_USER_CRT — user certificate (PEM)
|
||||
KAFKA_USER_KEY — user private key (PEM)
|
||||
PRODUCE_INTERVAL — seconds between events (default 3)
|
||||
"""
|
||||
|
||||
@@ -17,7 +14,6 @@ import json
|
||||
import os
|
||||
import random
|
||||
import ssl
|
||||
import tempfile
|
||||
import time
|
||||
import threading
|
||||
import uuid
|
||||
@@ -33,26 +29,8 @@ KAFKA_BROKERS = os.environ["KAFKA_BROKERS"]
|
||||
KAFKA_TOPIC = os.environ["KAFKA_TOPIC"]
|
||||
KAFKA_USERNAME = os.environ["KAFKA_USERNAME"]
|
||||
KAFKA_PASSWORD = os.environ["KAFKA_PASSWORD"]
|
||||
KAFKA_CA_CRT = os.environ.get("KAFKA_CA_CRT", "")
|
||||
KAFKA_USER_CRT = os.environ.get("KAFKA_USER_CRT", "")
|
||||
KAFKA_USER_KEY = os.environ.get("KAFKA_USER_KEY", "")
|
||||
PRODUCE_INTERVAL = float(os.environ.get("PRODUCE_INTERVAL", 3))
|
||||
|
||||
# --- TLS certs → temp files (kafka-python needs file paths) ---
|
||||
_cert_dir = tempfile.mkdtemp(prefix="kafka-certs-")
|
||||
|
||||
def _write_cert(name, content):
|
||||
if not content:
|
||||
return None
|
||||
path = os.path.join(_cert_dir, name)
|
||||
with open(path, "w") as f:
|
||||
f.write(content)
|
||||
return path
|
||||
|
||||
_ca_file = _write_cert("ca.crt", KAFKA_CA_CRT)
|
||||
_cert_file = _write_cert("user.crt", KAFKA_USER_CRT)
|
||||
_key_file = _write_cert("user.key", KAFKA_USER_KEY)
|
||||
|
||||
# --- IoT simulation config ---
|
||||
SENSORS = [
|
||||
{"id": "temp_living", "type": "temperature", "location": "living_room", "unit": "°C", "min": 18.0, "max": 28.0},
|
||||
@@ -73,25 +51,19 @@ errors_count = 0
|
||||
def get_producer():
|
||||
global producer
|
||||
if producer is None:
|
||||
kwargs = {
|
||||
"bootstrap_servers": KAFKA_BROKERS,
|
||||
"security_protocol": "SASL_SSL",
|
||||
"sasl_mechanism": "PLAIN",
|
||||
"sasl_plain_username": KAFKA_USERNAME,
|
||||
"sasl_plain_password": KAFKA_PASSWORD,
|
||||
"value_serializer": lambda v: json.dumps(v).encode("utf-8"),
|
||||
"key_serializer": lambda k: k.encode("utf-8") if k else None,
|
||||
}
|
||||
if _ca_file:
|
||||
kwargs["ssl_cafile"] = _ca_file
|
||||
if _cert_file:
|
||||
kwargs["ssl_certfile"] = _cert_file
|
||||
if _key_file:
|
||||
kwargs["ssl_keyfile"] = _key_file
|
||||
# Internal cluster communication — skip hostname check
|
||||
kwargs["ssl_check_hostname"] = False
|
||||
|
||||
producer = KafkaProducer(**kwargs)
|
||||
ctx = ssl.create_default_context()
|
||||
ctx.check_hostname = False
|
||||
ctx.verify_mode = ssl.CERT_NONE
|
||||
producer = KafkaProducer(
|
||||
bootstrap_servers=KAFKA_BROKERS,
|
||||
security_protocol="SASL_SSL",
|
||||
sasl_mechanism="PLAIN",
|
||||
sasl_plain_username=KAFKA_USERNAME,
|
||||
sasl_plain_password=KAFKA_PASSWORD,
|
||||
ssl_context=ctx,
|
||||
value_serializer=lambda v: json.dumps(v).encode("utf-8"),
|
||||
key_serializer=lambda k: k.encode("utf-8") if k else None,
|
||||
)
|
||||
return producer
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user