From 3e71d0342888f7b56633531a4afc34e111b932fc Mon Sep 17 00:00:00 2001 From: miki3421 <42601846+miki3421@users.noreply.github.com> Date: Wed, 21 Oct 2026 15:07:29 +0200 Subject: [PATCH 1/2] Isolate Kafka and ZooKeeper from tenant action pods on 0.9.0 --- 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 + .../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 ++++++++++++++++++ 10 files changed, 557 insertions(+), 6 deletions(-) create mode 100644 MESSAGING_SECURITY.md create mode 100644 deploy/kafka/kafka-004-networkpolicy.yaml create mode 100644 deploy/zookeeper/zookeeper-005-networkpolicy.yaml create mode 100644 openserverless/messaging_security.py create mode 100644 tests/test_messaging_security.py 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 @@ + + +# 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.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.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.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() From 146e4639a896ce2efd7c1f391d1f279d7e20ce36 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 21 Oct 2026 22:08:22 +0200 Subject: [PATCH 2/2] Release unused Kind ingress NodePorts before integration tests --- .github/workflows/check.yml | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) 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