doc: Document OPA rego rules

Author: siegfriedweber

docs/modules/kafka/examples/security/01-setup.yaml

@@ -0,0 +1,20 @@
+---
+apiVersion: zookeeper.stackable.tech/v1alpha1
+kind: ZookeeperCluster
+metadata:
+  name: simple-zk
+spec:
+  image:
+    productVersion: 3.9.5
+  servers:
+    roleGroups:
+      default:
+        replicas: 1
+---
+apiVersion: zookeeper.stackable.tech/v1alpha1
+kind: ZookeeperZnode
+metadata:
+  name: simple-kafka-znode
+spec:
+  clusterRef:
+    name: simple-zk

docs/modules/kafka/examples/security/02-authorization-opa.yaml

@@ -0,0 +1,63 @@
+---
+apiVersion: opa.stackable.tech/v1alpha1
+kind: OpaCluster
+metadata:
+  name: simple-opa
+spec:
+  image:
+    productVersion: 1.16.2
+  servers:
+    config:
+      logging:
+        containers:
+          opa:
+            loggers:
+              decision:
+                level: INFO
+    roleGroups:
+      default: {}
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: kafka-rego-rules
+  labels:
+    opa.stackable.tech/bundle: "true"
+data:
+  kafka.rego: |
+    package kafka
+
+    is_internal_request if input.requestContext.listenerName == "INTERNAL"
+    is_external_request if not is_internal_request
+
+    # FQDNs of the brokers as regular expressions
+    brokers_fqdn := [
+      `local`,                                 # cluster domain
+      `cluster`,                               # cluster domain
+      `svc`,
+      `default`,                               # namespace
+      `simple-kafka-broker-default-headless`,  # Pod subdomain
+      `simple-kafka-broker-default-[0-9]+`,    # Pod name
+    ]
+
+    domain_components = [concat("", [`DC=`, dc]) | some dc in brokers_fqdn]
+    common_name = `CN=generated certificate for pod`
+    subject_dn_fields = array.concat(domain_components, [common_name])
+    brokers_subject_dn_pattern := concat("", ["^", concat(",", subject_dn_fields), "$"])
+
+    default allow := false
+
+    allow if {
+      is_internal_request
+
+      regex.match(
+        brokers_subject_dn_pattern,
+        input.requestContext.principal.name,
+      )
+    }
+
+    allow if {
+      is_external_request
+
+      # TODO Add your own rules
+    }

docs/modules/kafka/examples/security/03a-authorization-kafkacluster.yaml

@@ -0,0 +1,18 @@
+---
+apiVersion: kafka.stackable.tech/v1alpha1
+kind: KafkaCluster
+metadata:
+  name: simple-kafka
+spec:
+  image:
+    productVersion: 3.9.2
+  clusterConfig:
+    authorization:
+      opa:
+        configMapName: simple-opa
+        package: kafka
+    zookeeperConfigMapName: simple-kafka-znode
+  brokers:
+    roleGroups:
+      default:
+        replicas: 1

docs/modules/kafka/examples/security/03b-authorization-kafkacluster-custom-properties.yaml

@@ -0,0 +1,23 @@
+---
+apiVersion: kafka.stackable.tech/v1alpha1
+kind: KafkaCluster
+metadata:
+  name: simple-kafka
+spec:
+  image:
+    productVersion: 3.9.2
+  clusterConfig:
+    authorization:
+      opa:
+        configMapName: simple-opa
+        package: kafka
+    zookeeperConfigMapName: simple-kafka-znode
+  brokers:
+    configOverrides:
+      broker.properties:
+        opa.authorizer.cache.initial.capacity: "100"
+        opa.authorizer.cache.maximum.size: "100"
+        opa.authorizer.cache.expire.after.seconds: "10"
+    roleGroups:
+      default:
+        replicas: 1

docs/modules/kafka/pages/usage-guide/security.adoc

@@ -177,53 +177,24 @@ The package is optional and defaults to the `metadata.name` field:
 
 [source,yaml]
 ----
----
-apiVersion: kafka.stackable.tech/v1alpha1
-kind: KafkaCluster
-metadata:
-  name: simple-kafka
-spec:
-  image:
-    productVersion: 3.9.2
-  clusterConfig:
-    authorization:
-      opa:
-        configMapName: simple-opa
-        package: kafka
-    zookeeperConfigMapName: simple-kafka-znode
-  brokers:
-    roleGroups:
-      default:
-        replicas: 1
+include::example$security/03a-authorization-kafkacluster.yaml[]
 ----
 
 You can change some opa cache properties by overriding:
 
 [source,yaml]
 ----
----
-apiVersion: kafka.stackable.tech/v1alpha1
-kind: KafkaCluster
-metadata:
-  name: simple-kafka
-spec:
-  image:
-    productVersion: 3.9.2
-  clusterConfig:
-    authorization:
-      opa:
-        configMapName: simple-opa
-        package: kafka
-    zookeeperConfigMapName: simple-kafka-znode
-  brokers:
-    configOverrides:
-      broker.properties:
-        opa.authorizer.cache.initial.capacity: "100"
-        opa.authorizer.cache.maximum.size: "100"
-        opa.authorizer.cache.expire.after.seconds: "10"
-    roleGroups:
-      default:
-        replicas: 1
+include::example$security/03b-authorization-kafkacluster-custom-properties.yaml[]
 ----
 
 A full list of settings and their respective defaults can be found https://github.com/anderseknert/opa-kafka-plugin[here].
+
+The OPA Rego rules can authorize both external and internal communication.
+For internal communication, the principal name is the subject DN of the broker's certificate.
+The subject DN contains the FQDN of the pod and a generic common name, e.g. `DC=local,DC=cluster,DC=svc,DC=default,DC=simple-kafka-broker-default-headless,DC=simple-kafka-broker-default-0,CN=generated certificate for pod`.
+An OpaCluster and the Rego rules that verify the subject DNs could look as follows:
+
+[source,yaml]
+----
+include::example$security/02-authorization-opa.yaml[]
+----

tests/templates/kuttl/opa/20_opa.yaml.j2

@@ -9,7 +9,7 @@ data:
   kafka.rego: |
     package authz
 
-    # FQDNs of the brokers as regular expressions
+    # FQDN of the brokers as regular expressions
     brokers_fqdn := [
       `local`,
       `cluster`,
@@ -22,7 +22,7 @@ data:
     domain_components = [concat("", [`DC=`, dc]) | some dc in brokers_fqdn]
     common_name = `CN=generated certificate for pod`
     subject_dn_fields = array.concat(domain_components, [common_name])
-    brokers_subject_dn_pattern := concat(",", subject_dn_fields)
+    brokers_subject_dn_pattern := concat("", ["^", concat(",", subject_dn_fields), "$"])
 
     default allow := false