This is an automated email from the ASF dual-hosted git repository.
sciabarracom pushed a commit to branch 0.9.0
in repository https://gitbox.apache.org/repos/asf/openserverless-operator.git
The following commit(s) were added to refs/heads/0.9.0 by this push:
new 245032c Isolate Kafka and ZooKeeper from tenant action pods on 0.9.0
(#145)
245032c is described below
commit 245032c9ee085402580f7094ef3db5b66226b831
Author: miki3421 <[email protected]>
AuthorDate: Thu Oct 22 10:41:21 2026 +0200
Isolate Kafka and ZooKeeper from tenant action pods on 0.9.0 (#145)
* Isolate Kafka and ZooKeeper from tenant action pods on 0.9.0
* Release unused Kind ingress NodePorts before integration tests
---------
Co-authored-by: Codex <[email protected]>
---
.github/workflows/check.yml | 9 +-
MESSAGING_SECURITY.md | 94 ++++++++
deploy/kafka/kafka-001-sts.yaml | 4 +-
deploy/kafka/kafka-004-networkpolicy.yaml | 51 ++++
deploy/zookeeper/zookeeper-003-sts.yaml | 2 +
deploy/zookeeper/zookeeper-005-networkpolicy.yaml | 50 ++++
openserverless/kafka.py | 7 +-
openserverless/main.py | 4 +
openserverless/messaging_security.py | 68 ++++++
openserverless/zookeeper.py | 5 +-
tests/test_messaging_security.py | 278 ++++++++++++++++++++++
11 files changed, 565 insertions(+), 7 deletions(-)
diff --git a/.github/workflows/check.yml b/.github/workflows/check.yml
index 965ca64..7f1d3c0 100644
--- a/.github/workflows/check.yml
+++ b/.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
diff --git a/MESSAGING_SECURITY.md b/MESSAGING_SECURITY.md
new file mode 100644
index 0000000..73f060a
--- /dev/null
+++ b/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.
diff --git a/deploy/kafka/kafka-001-sts.yaml b/deploy/kafka/kafka-001-sts.yaml
index 0c0f859..e8b7e36 100644
--- a/deploy/kafka/kafka-001-sts.yaml
+++ b/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
\ No newline at end of file
+ readOnly: true
diff --git a/deploy/kafka/kafka-004-networkpolicy.yaml
b/deploy/kafka/kafka-004-networkpolicy.yaml
new file mode 100644
index 0000000..b79276b
--- /dev/null
+++ b/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
diff --git a/deploy/zookeeper/zookeeper-003-sts.yaml
b/deploy/zookeeper/zookeeper-003-sts.yaml
index f2766c8..a27d21e 100644
--- a/deploy/zookeeper/zookeeper-003-sts.yaml
+++ b/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
diff --git a/deploy/zookeeper/zookeeper-005-networkpolicy.yaml
b/deploy/zookeeper/zookeeper-005-networkpolicy.yaml
new file mode 100644
index 0000000..d73fa41
--- /dev/null
+++ b/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
diff --git a/openserverless/kafka.py b/openserverless/kafka.py
index 2aad70b..36864b0 100644
--- a/openserverless/kafka.py
+++ b/openserverless/kafka.py
@@ -22,6 +22,7 @@ import openserverless.config as cfg
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
\ No newline at end of file
+ return res
diff --git a/openserverless/main.py b/openserverless/main.py
index 0f3e887..ca02d9d 100644
--- a/openserverless/main.py
+++ b/openserverless/main.py
@@ -29,6 +29,7 @@ import openserverless.issuer as issuer
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):
diff --git a/openserverless/messaging_security.py
b/openserverless/messaging_security.py
new file mode 100644
index 0000000..61fe17b
--- /dev/null
+++ b/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
diff --git a/openserverless/zookeeper.py b/openserverless/zookeeper.py
index 597ac01..5b3700c 100644
--- a/openserverless/zookeeper.py
+++ b/openserverless/zookeeper.py
@@ -21,6 +21,7 @@ import openserverless.kustomize as kus
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
diff --git a/tests/test_messaging_security.py b/tests/test_messaging_security.py
new file mode 100644
index 0000000..211de31
--- /dev/null
+++ b/tests/test_messaging_security.py
@@ -0,0 +1,278 @@
+# 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.
+
+"""Messaging isolation contract and component lifecycle regression tests."""
+
+import copy
+from pathlib import Path
+import unittest
+from unittest.mock import patch
+
+import yaml
+
+from openserverless import config, kafka, main, messaging_security, zookeeper
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def policy(component):
+ return yaml.safe_load(next((ROOT / 'deploy' /
component).glob('*networkpolicy.yaml')).read_text())
+
+
+def matches(selector, labels):
+ if any(labels.get(k) != v for k, v in selector.get('matchLabels',
{}).items()):
+ return False
+ for expression in selector.get('matchExpressions', []):
+ key, op = expression['key'], expression['operator']
+ if op == 'In' and labels.get(key) not in expression['values']:
+ return False
+ if op == 'DoesNotExist' and key in labels:
+ return False
+ return True
+
+
+def allowed(component, labels, port, namespace='openserverless'):
+ # Ingress peers without namespaceSelector select only the policy namespace.
+ if namespace != 'openserverless':
+ return False
+ return any(any(p['port'] == port for p in rule['ports']) and
+ any(matches(peer['podSelector'], labels) for peer in
rule['from'])
+ for rule in policy(component)['spec']['ingress'])
+
+
+class MessagingSecurityTests(unittest.TestCase):
+ def tearDown(self):
+ config.clean()
+
+ def
test_existing_cluster_resume_installs_isolation_before_post_resume(self):
+ events = []
+ owner = {'metadata': {'name': 'controller', 'uid': 'synthetic'}}
+ def configure(spec, **kwargs):
+ config.configure(spec)
+ with patch.object(main.operator_util, 'config_from_spec',
side_effect=configure), \
+ patch.object(main.kube, 'get', return_value=owner), \
+ patch.object(zookeeper, 'create', side_effect=lambda obj:
events.append(('zookeeper', obj))), \
+ patch.object(kafka, 'create', side_effect=lambda obj:
events.append(('kafka', obj))), \
+ patch.object(main.operator_util, 'whisk_post_resume',
side_effect=lambda name: events.append(('post', name))):
+ main.whisk_resume({'components': {'kafka': True, 'zookeeper':
True}}, {}, 'controller')
+ self.assertEqual(events, [('zookeeper', owner), ('kafka', owner),
('post', 'controller')])
+
+ def
test_resume_failure_propagates_and_never_reports_post_resume_success(self):
+ config.configure({'components': {'kafka': True, 'zookeeper': True}})
+ with patch.object(main.operator_util, 'config_from_spec'), \
+ patch.object(main.kube, 'get', return_value={'metadata': {'uid':
'synthetic'}}), \
+ patch.object(zookeeper, 'create',
side_effect=RuntimeError('policy denied')), \
+ patch.object(kafka, 'create') as broker, \
+ patch.object(main.operator_util, 'whisk_post_resume') as post:
+ with self.assertRaisesRegex(RuntimeError, 'policy denied'):
+ main.whisk_resume({}, {}, 'controller')
+ broker.assert_not_called()
+ post.assert_not_called()
+
+ def test_configuration_update_reconciles_before_existing_patcher(self):
+ events = []
+ config.configure({'components': {'kafka': True}})
+ owner = {'metadata': {'uid': 'synthetic'}}
+ with patch.object(main.operator_util, 'config_from_spec'), \
+ patch.object(main.kube, 'get', return_value=owner), \
+ patch.object(kafka, 'create', side_effect=lambda obj:
events.append(('kafka', obj))), \
+ patch.object(zookeeper, 'create') as zoo, \
+ patch.object(main.patcher, 'patch', side_effect=lambda *args:
events.append(('patch', args))):
+ main.whisk_update({}, {}, 'openserverless', (), 'controller')
+ self.assertEqual(events, [('kafka', owner), ('patch', ((), {}, owner,
'controller'))])
+ zoo.assert_not_called()
+
+ def test_missing_owner_aborts_before_any_component_mutation(self):
+ config.configure({'components': {'kafka': True, 'zookeeper': True}})
+ with patch.object(kafka, 'create') as broker, patch.object(zookeeper,
'create') as zoo:
+ for owner in [None, {}, {'metadata': {}}, {'metadata': {'name':
'controller'}}]:
+ with self.assertRaisesRegex(RuntimeError, 'Whisk owner'):
+ messaging_security.reconcile(owner)
+ broker.assert_not_called()
+ zoo.assert_not_called()
+
+ def test_slim_resume_does_not_query_owner_or_create_messaging(self):
+ config.configure({'components': {'kafka': False, 'zookeeper': False}})
+ with patch.object(main.operator_util, 'config_from_spec'), \
+ patch.object(main.kube, 'get') as get, \
+ patch.object(kafka, 'create') as broker, \
+ patch.object(zookeeper, 'create') as zoo, \
+ patch.object(main.operator_util, 'whisk_post_resume') as post:
+ main.whisk_resume({}, {}, 'controller')
+ messaging_security.reconcile(None)
+ get.assert_not_called()
+ broker.assert_not_called()
+ zoo.assert_not_called()
+ post.assert_called_once_with('controller')
+
+ def test_kafka_rollout_failure_stops_update_processing(self):
+ config.configure({'components': {'kafka': True}})
+ with patch.object(main.operator_util, 'config_from_spec'), \
+ patch.object(main.kube, 'get', return_value={'metadata': {'uid':
'synthetic'}}), \
+ patch.object(kafka, 'create', side_effect=RuntimeError('rollout
timed out')), \
+ patch.object(main.patcher, 'patch') as patcher:
+ with self.assertRaisesRegex(RuntimeError, 'rollout timed out'):
+ main.whisk_update({}, {}, 'openserverless', (), 'controller')
+ patcher.assert_not_called()
+
+ def test_tenant_and_unrelated_pods_denied_on_all_ports(self):
+ for component, ports in [('kafka', [9092, 9071, 9888]),
+ ('zookeeper', [2181, 2888, 3888])]:
+ for labels in [{}, {'user-action-pod': 'true'},
+ {'name': 'guest'}, {'name':
'openserverless-system-api'}]:
+ for port in ports:
+ with self.subTest(component=component, labels=labels,
port=port):
+ self.assertFalse(allowed(component, labels, port))
+
+ def test_action_cannot_impersonate_trusted_peer_or_monitoring(self):
+ for component, ports in [('kafka', [9092, 9071, 9888]),
+ ('zookeeper', [2181, 2888, 3888])]:
+ for name in ['controller', 'invoker', 'scheduler', 'kafka',
'zookeeper']:
+ for marker in ['true', 'false', '']:
+ labels = {'name': name, 'user-action-pod': marker,
+ 'app.kubernetes.io/name': 'prometheus',
+ 'app.kubernetes.io/instance':
'openserverless-prometheus'}
+ for port in ports:
+ self.assertFalse(allowed(component, labels, port))
+
+ def test_same_labels_in_other_namespace_are_denied(self):
+ for component in ['kafka', 'zookeeper']:
+ for name in ['controller', 'invoker', 'kafka', 'zookeeper']:
+ self.assertFalse(allowed(component, {'name': name}, 9092,
namespace='other'))
+ self.assertFalse(allowed(component, {'name': name}, 2181,
namespace='other'))
+
+ def test_control_plane_clients_preserve_required_connections(self):
+ for name in ['controller', 'invoker', 'scheduler', 'kafka']:
+ self.assertTrue(allowed('kafka', {'name': name}, 9092))
+ for name in ['kafka', 'invoker', 'zookeeper']:
+ self.assertTrue(allowed('zookeeper', {'name': name}, 2181))
+ for port in [2888, 3888]:
+ self.assertTrue(allowed('zookeeper', {'name': 'zookeeper'}, port))
+ for name in ['controller', 'invoker', 'scheduler', 'kafka']:
+ self.assertFalse(allowed('zookeeper', {'name': name}, port))
+
+ def test_metrics_does_not_grant_broker_access(self):
+ labels = {'app.kubernetes.io/name': 'prometheus',
+ 'app.kubernetes.io/instance': 'openserverless-prometheus'}
+ for port in [9071, 9888]:
+ self.assertTrue(allowed('kafka', labels, port))
+ self.assertFalse(allowed('kafka', labels, 9092))
+ self.assertFalse(allowed('zookeeper', labels, 2181))
+
+ def test_all_policy_rules_have_scoped_peers_and_ports(self):
+ for component in ['kafka', 'zookeeper']:
+ obj = policy(component)
+ self.assertEqual(obj['spec']['podSelector']['matchLabels'],
{'name': component})
+ self.assertEqual(obj['spec']['policyTypes'], ['Ingress'])
+ for rule in obj['spec']['ingress']:
+ self.assertTrue(rule['ports'])
+ self.assertTrue(rule['from'])
+ for peer in rule['from']:
+ self.assertEqual(set(peer), {'podSelector'})
+ self.assertTrue(peer['podSelector'])
+
+ def test_policy_apply_failure_never_applies_workloads(self):
+ for component in ['kafka', 'zookeeper']:
+ p = policy(component)
+ workload = {'kind': 'StatefulSet', 'metadata': {'name': component}}
+ spec = {'apiVersion': 'v1', 'kind': 'List', 'items': [workload, p]}
+ with patch.object(messaging_security.kube, 'apply',
side_effect=RuntimeError('denied')) as apply:
+ with self.assertRaisesRegex(RuntimeError, 'denied'):
+ messaging_security.apply(spec, component)
+ self.assertEqual(apply.call_count, 1)
+ self.assertEqual(apply.call_args.args[0]['items'], [p])
+
+ def test_missing_policy_fails_before_any_kubernetes_mutation(self):
+ for component in ['kafka', 'zookeeper']:
+ with patch.object(messaging_security.kube, 'apply') as apply:
+ with self.assertRaisesRegex(RuntimeError, 'Missing'):
+ messaging_security.apply({'items': []}, component)
+ apply.assert_not_called()
+
+ def
test_component_reconciliation_and_saved_deletion_spec_include_policy(self):
+ for component, module in [('kafka', kafka), ('zookeeper', zookeeper)]:
+ for owner in [None, {'apiVersion': 'openserverless.org/v1',
'kind': 'Whisk',
+ 'metadata': {'name': 'test', 'uid':
'synthetic'}}]:
+ config.clean()
+ config.put('openserverless.zookeeper.url',
'zookeeper-0.zookeeper:2181')
+ spec = {'apiVersion': 'v1', 'kind': 'List',
+ 'items': [{'kind': 'StatefulSet', 'metadata': {'name':
component}},
+ policy(component)]}
+ calls = []
+ with patch.object(module.cfg_util, 'get_' + component +
'_config_data',
+ return_value={'affinity': False,
'tolerations': False}), \
+ patch.object(module.kus, 'patchTemplates',
return_value=''), \
+ patch.object(module.kus, 'kustom_list',
return_value=spec), \
+ patch.object(module.util, 'wait_for_pod_ready'), \
+ patch.object(module, 'get_' + component +
'_connect_data', return_value=component + ':2181'), \
+ patch.object(messaging_security.kube, 'kubectl') as
kubectl, \
+ patch.object(messaging_security.kube, 'apply',
+ side_effect=lambda obj:
calls.append(copy.deepcopy(obj)) or 'applied'):
+ self.assertEqual(module.create(owner), 'applied')
+ self.assertTrue(all(item['kind'] == 'NetworkPolicy' for item
in calls[0]['items']))
+ self.assertEqual(calls[1]['items'], spec['items'])
+ kubectl.assert_called_once_with('rollout', 'status',
'statefulset/' + component, '--timeout=300s')
+ if owner:
+
self.assertTrue(all(item['metadata'].get('ownerReferences') for item in
calls[0]['items']))
+ else:
+ saved = config.get('state.' + component + '.spec')
+ self.assertTrue(any(item['kind'] == 'NetworkPolicy' for
item in saved['items']))
+ with patch.object(module.security, 'delete',
return_value='deleted') as delete:
+ self.assertEqual(module.delete(), 'deleted')
+ self.assertEqual(delete.call_args.args[0], saved)
+
+ def test_legacy_sessions_are_closed_by_statefulset_rollout(self):
+ for component in ['kafka', 'zookeeper']:
+ path = next((ROOT / 'deploy' / component).glob('*sts.yaml'))
+ obj = next(item for item in yaml.safe_load_all(path.read_text())
+ if item and item['kind'] == 'StatefulSet')
+ self.assertEqual(obj['spec']['template']['metadata']['annotations']
+ ['openserverless.org/messaging-isolation'], 'v1')
+
+ def test_deletion_keeps_policy_until_all_component_pods_are_gone(self):
+ for component in ['kafka', 'zookeeper']:
+ p = policy(component)
+ sts = {'kind': 'StatefulSet', 'metadata': {'name': component}}
+ events = []
+ def kubectl(*args):
+ events.append(args[0])
+ return 'pod/' + component + '-0' if args[0] == 'get' else ''
+ def delete(spec):
+ events.append(spec['items'][0]['kind'])
+ return 'deleted'
+ with patch.object(messaging_security.kube, 'kubectl',
side_effect=kubectl), \
+ patch.object(messaging_security.kube, 'delete',
side_effect=delete):
+ messaging_security.delete({'items': [p, sts]}, component)
+ self.assertEqual(events, ['StatefulSet', 'get', 'wait',
'NetworkPolicy'])
+
+ def test_delete_timeout_keeps_policy(self):
+ with patch.object(messaging_security.kube, 'kubectl',
+ side_effect=['pod/kafka-0', RuntimeError('still
terminating')]), \
+ patch.object(messaging_security.kube, 'delete',
return_value='deleted') as delete:
+ with self.assertRaisesRegex(RuntimeError, 'still terminating'):
+ messaging_security.delete({'items': [policy('kafka'), {'kind':
'StatefulSet'}]}, 'kafka')
+ self.assertEqual(delete.call_count, 1)
+
+ def test_legacy_saved_spec_without_policy_can_still_be_deleted(self):
+ with patch.object(messaging_security.kube, 'kubectl',
return_value=''), \
+ patch.object(messaging_security.kube, 'delete',
return_value='deleted') as delete:
+ self.assertEqual(messaging_security.delete({'items': [{'kind':
'StatefulSet'}]}, 'kafka'), 'deleted')
+ self.assertEqual(delete.call_count, 1)
+
+
+if __name__ == '__main__':
+ unittest.main()
