Isolate Kafka and ZooKeeper from tenant action pods on 0.9.0

.github/workflows/check.yml

@@ -69,7 +69,14 @@ jobs:
       - name: BuildAndLoad
         run: task build-and-load
       - name: AddIngress
-        run: task kind:ingress
+        run: |
+          task kind:ingress
+          # Kind ingress uses hostPort 80/443. Release its unused NodePorts so
+          # dynamic allocation cannot occupy fixed test ports such as 30984.
+          kubectl -n ingress-nginx patch service ingress-nginx-controller --type=strategic \
+            -p '{"spec":{"type":"ClusterIP","allocateLoadBalancerNodePorts":null,"ports":[{"port":80,"nodePort":null},{"port":443,"nodePort":null}]}}'
+          test -z "$(kubectl -n ingress-nginx get service ingress-nginx-controller -o jsonpath='{.spec.ports[*].nodePort}')"
+          kubectl get services --all-namespaces
       - name: Unit Test
         run: task utest
         continue-on-error: false

MESSAGING_SECURITY.md

@@ -0,0 +1,94 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with this
+work for additional information regarding copyright ownership. The ASF
+licenses this file to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+License for the specific language governing permissions and limitations
+under the License.
+-->
+
+# Kafka and ZooKeeper tenant isolation
+
+The full deployment runs untrusted action pods alongside the shared Kafka and
+ZooKeeper services. Those services are private control-plane endpoints; tenant
+actions must not connect to them, even when the action knows their DNS names or
+pod IPs.
+
+The operator installs ingress NetworkPolicies before reconciling either
+component. Kafka's broker port accepts only controller, invoker, scheduler and
+Kafka pods in the same namespace. Kafka metrics accept only the installation's
+Prometheus pods. ZooKeeper's client port accepts Kafka, invoker and ZooKeeper
+pods; quorum ports accept only ZooKeeper peers. Every allowed peer selector
+also excludes pods carrying the `user-action-pod` label, regardless of its value.
+The invoker controls action pod labels; action annotations cannot grant these
+network permissions.
+
+The policies are included in rendered and saved component specs and inherit the
+Whisk owner reference. Policy application failure stops workload reconciliation.
+A pod-template revision rolls legacy Kafka and ZooKeeper pods once, closing
+connections established before the boundary was installed. Reconciliation waits
+for the rollout. When called with a saved component spec, the explicit deletion
+helper keeps the policy until all component pods have disappeared; a deletion
+timeout retains the policy.
+
+Normal owner-backed Whisk deletion currently relies on Kubernetes owner garbage
+collection, which does not guarantee that the policy outlives the messaging pods.
+The ordered helper is not reached when the process-local saved spec is absent.
+This teardown race is a known, accepted limitation of this change; policy-last
+deletion is not guaranteed during Whisk uninstall. Normal operation and automatic
+upgrade isolation are covered by the verification below.
+
+## Deployment requirements
+
+The cluster must enforce Kubernetes NetworkPolicy. The presence of a policy API
+object alone is not proof of enforcement. K3s with its network-policy controller
+enabled is supported by this boundary; an installation with
+`--disable-network-policy`, or a CNI without policy support, needs an enforcing
+network provider before this protection is effective. Do not claim tenant
+isolation on such a cluster without a live denied-connection test.
+
+For an existing full installation, the patched operator automatically reconciles
+enabled ZooKeeper and Kafka components when it resumes or processes a Whisk
+configuration update. The supported `ops update operator` flow therefore installs
+the boundary without manual internal calls. Reconciliation applies ZooKeeper
+before Kafka, propagates errors to the operator handler, and waits for each
+StatefulSet rollout before continuing normal post-resume or update processing.
+A missing Whisk owner aborts reconciliation. Slim installations do not deploy
+messaging components. Repeating a restart preserves existing policies and does
+not roll unchanged pod templates. Plan for the single-replica service restart on
+the first upgrade that installs this boundary. Do not remove the policies to work
+around an unrelated connectivity failure. Kubernetes NetworkPolicies are
+additive: an administrator-installed broad allow policy can reopen access.
+
+This patch preserves broker protocols, component versions and PVC contents. It
+enforces the ordinary tenant-action boundary at the network layer; it does not
+add TLS, Kafka SASL/ACLs or ZooKeeper authentication. Host/node administrators,
+Kubernetes administrators and compromised allowed control-plane pods remain
+trusted. Service-level credentials would provide additional protection against
+those separate paths. Slim deployments do not create these messaging components.
+
+## Verification
+
+Run `python -m unittest discover -s tests -p test_messaging_security.py` in the
+operator environment. Live acceptance must additionally demonstrate:
+
+- A normal action in each of two tenants cannot reach Kafka or ZooKeeper by
+  Service name, pod DNS name or direct pod IP, or Kafka metrics ports.
+- A pod in another namespace cannot gain access by copying trusted labels.
+- Action-labelled pods remain denied even with a trusted component label.
+- Controller/invoker Kafka traffic, invoker ZooKeeper traffic, broker-to-
+  ZooKeeper traffic and Prometheus metric scraping still work.
+- Invoker restart preserves its assigned ID and successful tenant invocations.
+- Updating an existing installation through `ops update operator` installs both
+  policies and rolls legacy messaging pods without manual reconciliation;
+  repeating the operator restart preserves the messaging pod identities.
+- Alpha/Beta application, Redis Lua, Streamer and scheduled actions still work
+  without editing their code.

deploy/kafka/kafka-001-sts.yaml

@@ -33,6 +33,8 @@ spec:
       name: kafka
   template:
     metadata:
+      annotations:
+        openserverless.org/messaging-isolation: "v1"
       labels:
         name: kafka
         app: kafka
@@ -104,4 +106,4 @@ spec:
         volumeMounts:
         - mountPath: /kafka
           name: kafka-pvc
-          readOnly: true
+          readOnly: true

deploy/kafka/kafka-004-networkpolicy.yaml

@@ -0,0 +1,51 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+metadata:
+  name: openserverless-kafka-isolation
+  namespace: openserverless
+spec:
+  podSelector:
+    matchLabels:
+      name: kafka
+  policyTypes: [Ingress]
+  ingress:
+    - from:
+        - podSelector:
+            matchExpressions:
+              - key: name
+                operator: In
+                values: [controller, invoker, scheduler, kafka]
+              - key: user-action-pod
+                operator: DoesNotExist
+      ports:
+        - port: 9092
+          protocol: TCP
+    - from:
+        - podSelector:
+            matchLabels:
+              app.kubernetes.io/name: prometheus
+              app.kubernetes.io/instance: openserverless-prometheus
+            matchExpressions:
+              - key: user-action-pod
+                operator: DoesNotExist
+      ports:
+        - port: 9071
+          protocol: TCP
+        - port: 9888
+          protocol: TCP

deploy/zookeeper/zookeeper-003-sts.yaml

@@ -32,6 +32,8 @@ spec:
       name: zookeeper
   template:
     metadata:
+      annotations:
+        openserverless.org/messaging-isolation: "v1"
       labels:
         name: zookeeper
         app: zookeeper

deploy/zookeeper/zookeeper-005-networkpolicy.yaml

@@ -0,0 +1,50 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+metadata:
+  name: openserverless-zookeeper-isolation
+  namespace: openserverless
+spec:
+  podSelector:
+    matchLabels:
+      name: zookeeper
+  policyTypes: [Ingress]
+  ingress:
+    - from:
+        - podSelector:
+            matchExpressions:
+              - key: name
+                operator: In
+                values: [kafka, invoker, zookeeper]
+              - key: user-action-pod
+                operator: DoesNotExist
+      ports:
+        - port: 2181
+          protocol: TCP
+    - from:
+        - podSelector:
+            matchLabels:
+              name: zookeeper
+            matchExpressions:
+              - key: user-action-pod
+                operator: DoesNotExist
+      ports:
+        - port: 2888
+          protocol: TCP
+        - port: 3888
+          protocol: TCP

openserverless/kafka.py

@@ -22,6 +22,7 @@
 import openserverless.enterprise_util as cfg_util
 import openserverless.enterprise_util as cfg_util
 import openserverless.util as util
+import openserverless.messaging_security as security
 
 def get_kafka_pod_name():
     pod_name = kube.kubectl("get", "pods", jsonpath="{.items[?(@.metadata.labels.app == 'kafka')].metadata.name}")
@@ -62,7 +63,7 @@ def create(owner=None):
         cfg.put("state.kafka.spec", spec)
 
 
-    res = kube.apply(spec)
+    res = security.apply(spec, 'kafka')
 
     # dynamically detect kafka pod and wait for readiness
     util.wait_for_pod_ready("{.items[?(@.metadata.labels.app == 'kafka')].metadata.name}")
@@ -76,6 +77,6 @@ def delete():
     spec = cfg.get("state.kafka.spec")
     res = False
     if spec:
-        res = kube.delete(spec)
+        res = security.delete(spec, 'kafka')
         logging.info(f"delete kafka: {res}")
-    return res
+    return res

openserverless/main.py

@@ -29,6 +29,7 @@
 import openserverless.endpoint as endpoint
 import openserverless.zookeeper as zookeeper
 import openserverless.kafka as kafka
+import openserverless.messaging_security as messaging_security
 import openserverless.invoker as invoker
 import openserverless.patcher as patcher
 import openserverless.storage_static as static
@@ -399,11 +400,14 @@ def whisk_update(spec, status, namespace, diff, name, **kwargs):
     operator_util.config_from_spec(spec,handler_type="on_update")
     owner = kube.get(f"wsk/{name}")
 
+    messaging_security.reconcile(owner)
     patcher.patch(diff, status, owner, name)
 
 @kopf.on.resume('openserverless.org', 'v1', 'whisks')
 def whisk_resume(spec, status, name, **kwargs):   
     operator_util.config_from_spec(spec, handler_type="on_resume")
+    if cfg.get('components.zookeeper') or cfg.get('components.kafka'):
+        messaging_security.reconcile(kube.get(f"wsk/{name}"))
     operator_util.whisk_post_resume(name)
 
 def runtimes_filter(name, type, **kwargs):

openserverless/messaging_security.py

@@ -0,0 +1,68 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Apply the messaging network boundary before creating or updating workloads."""
+
+import openserverless.kube as kube
+import openserverless.config as cfg
+
+
+def reconcile(owner):
+    """Install the boundary on existing clusters during operator resume/update."""
+    # Import lazily: component create functions use this module's apply helper.
+    from openserverless import kafka, zookeeper
+
+    enabled = [(name, module) for name, module in
+               [('zookeeper', zookeeper), ('kafka', kafka)]
+               if cfg.get('components.' + name)]
+    if not enabled:
+        return
+    # Never turn a failed owner lookup into an ownerless component deployment.
+    if not owner or not owner.get('metadata', {}).get('uid'):
+        raise RuntimeError('Cannot reconcile messaging isolation without the Whisk owner')
+    for name, module in enabled:
+        module.create(owner)
+
+
+def apply(spec, component):
+    policy_name = 'openserverless-' + component + '-isolation'
+    policies = [item for item in spec['items']
+                if item['kind'] == 'NetworkPolicy']
+    if not any(item['metadata']['name'] == policy_name for item in policies):
+        raise RuntimeError('Missing ' + component + ' isolation policy')
+    # Do not create/reconcile an exposed service if policy application fails.
+    # The complete spec still owns the policy for reconciliation and deletion.
+    kube.apply({'apiVersion': 'v1', 'kind': 'List', 'items': policies})
+    result = kube.apply(spec)
+    # The template boundary revision rolls legacy pods to close existing
+    # anonymous sessions as well as blocking new connections.
+    kube.kubectl('rollout', 'status', 'statefulset/' + component, '--timeout=300s')
+    return result
+
+
+def delete(spec, component):
+    policies = [item for item in spec['items'] if item['kind'] == 'NetworkPolicy']
+    workloads = [item for item in spec['items'] if item['kind'] != 'NetworkPolicy']
+    result = kube.delete({'apiVersion': 'v1', 'kind': 'List', 'items': workloads})
+    # StatefulSet deletion can return while its dependent pods are terminating.
+    # Keep the boundary until those pods are actually gone; fail closed on error.
+    pods = kube.kubectl('get', 'pods', '-l', 'name=' + component, '-o', 'name').split()
+    if pods:
+        kube.kubectl('wait', '--for=delete', *pods, '--timeout=300s')
+    if policies:
+        result += kube.delete({'apiVersion': 'v1', 'kind': 'List', 'items': policies})
+    return result

openserverless/zookeeper.py

@@ -21,6 +21,7 @@
 import openserverless.config as cfg
 import openserverless.enterprise_util as cfg_util
 import openserverless.util as util
+import openserverless.messaging_security as security
 
 def get_zookeeper_pod_name():
     pod_name = kube.kubectl("get", "pods", jsonpath="{.items[?(@.metadata.labels.app == 'zookeeper')].metadata.name}")
@@ -56,7 +57,7 @@ def create(owner=None):
         cfg.put("state.zookeeper.spec", spec)
 
 
-    res = kube.apply(spec)
+    res = security.apply(spec, 'zookeeper')
 
     # dynamically detect zookeeper pod and wait for readiness
     util.wait_for_pod_ready("{.items[?(@.metadata.labels.app == 'zookeeper')].metadata.name}")    
@@ -70,7 +71,7 @@ def delete():
     spec = cfg.get("state.zookeeper.spec")
     res = False
     if spec:
-        res = kube.delete(spec)
+        res = security.delete(spec, 'zookeeper')
         logging.info(f"delete zookeeper: {res}")
     return res