Integrate mTLS and OPA with Strimzi

[mint-lmc]

Hi everyone,
I want to produce and consume to kafka from an external python client using mTLS for the authentication and OPA for the authorization.

• I followed this tutorial to use OPA https://strimzi.io/blog/2020/08/05/using-open-policy-agent-with-strimzi-and-apache-kafka/
• and the TLS certificates from the Strimzi documentation: https://strimzi.io/docs/operators/latest/full/configuring.html#certificates-and-secrets-formats-str

Broker configuration:

    authorization:
      type: opa
      url: http://10.42.0.208:8181/v1/data/kafka/authz/example/basic/allow
      expireAfterMs: 60000
    config:
      default.replication.factor: 1
      inter.broker.protocol.version: "3.1"
      min.insync.replicas: 1
      offsets.topic.replication.factor: 1
      transaction.state.log.min.isr: 1
      transaction.state.log.replication.factor: 1
    listeners:
    - name: plain
      port: 9092
      tls: false
      type: internal
    - authentication:
        type: tls
      name: tls
      port: 9093
      tls: true
      type: internal
    - authentication:
        type: tls
      name: external
      port: 9094
      tls: true
      type: nodeport

kafka user:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: john
  labels:
    strimzi.io/cluster: kafka-cluster
spec:
  authentication:
    type: tls

I have extracted the user certificates in PEM (.crt) format to use it with the kafka producer in python

Python code for the producer:

from kafka import KafkaProducer
import json
from bson import json_util

bootstrap_server = 'localhost:32672'

print("Producing messages to Kafka topic ...")
producer = KafkaProducer(
		       bootstrap_servers=bootstrap_server,
                       security_protocol="SSL",
                       ssl_cafile="ca.crt",
		       ssl_certfile="user.crt",
		       ssl_keyfile="user.key",
                       ssl_password="SPf6WnxZzdRz",
		       ssl_check_hostname=False,
		       value_serializer=lambda v: json.dumps(v).encode('ascii'),
		       key_serializer=lambda v: json.dumps(v).encode('ascii')
			)

producer.send("my-topic",
	       key={"key": 1},
		value={"message":"hello world"}
)
producer.flush()

Kafka user:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: matt
  labels:
    strimzi.io/cluster: kafka-cluster
spec:
  authentication:
    type: tls

Python code for the consumer:

from kafka import KafkaConsumer
import json
from bson import json_util

bootstrap_server = 'localhost:32672'

print("Consuming messages from Kafka topic ...")

consumer = KafkaConsumer(
		 bootstrap_servers=bootstrap_server,
                 ssl_cafile="ca.crt",
		 client_id ="matt",
		 security_protocol="SSL",
		 ssl_certfile="user.crt",
		 ssl_keyfile="user.key",
                 ssl_password= 'ZU9rZkLBrBYF',
		 ssl_check_hostname=False,
		 value_deserializer = lambda v: json.loads(v.decode('ascii')),
		 auto_offset_reset="earliest"
	)
consumer.subscribe(topics='my-topic')
for message in consumer:
  print ("%d:%d: v=%s" % (message.partition,
                          message.offset,
                          message.value))

I have tested the kafka producer and consumer in python with the ssl certificates and it s working fine.

I have deployed OPA server and added the OPA policies with Config map following the tutorial. The OPA server is working fine.
Once I add the authorization section to the broker configruation,

authorization:
      type: opa
      url: http://10.42.0.208:8181/v1/data/kafka/authz/example/basic/allow
      expireAfterMs: 60000

the producer is failing with this error :
kafka.errors.KafkaTimeoutError: KafkaTimeoutError: Failed to update metadata after 60.0 secs.

• Can we use the client certificates generated by Strimzi (ca.crt, user.crt, user.key) with an external client like python?

[scholzj]

You are not using the official Java client. So it is hard to know what different client errors mean. You should probably as a first thing check the broker logs. But if the client worked before without adding the authorization, then your problem is likely the authorization. Either it does not work or you are not allowed to produce/consume the messages.

PS: The blog post is 2 and half years old. I think the request sent to the OPA server changed since the blog post was written. So make sure that what you have is up to date.

[mint-lmc]

Hi again @scholzj ,

• I have tried to test the authorization flow with Simple ACL with the same authentication and broker configuration and everything is working fine,
• Regarding OPA, when I check the logs, the authorization is always denied ,
  ... "level":"info","msg":"Sent response.","req_id":1501,"req_method":"POST","req_path":"/v1/data/kafka/authz/allow","resp_body":"{\"decision_id\":\"e66199ef-e33b-45f8-a11f-d7781df7b1db\",\"result\":false}\n", ..
  It works only when I create an OPA policy that allows all requests allow := true

-Could you give me a simple example in rego where I can authorize only producer_1 to write in topic_1?

[scholzj]

I do not have such an example. Sorry.

[mint-lmc]

Thanks for the help.