diff --git a/deploy/seaweedfs/seaweedfs-svc.yaml b/deploy/seaweedfs/seaweedfs-svc.yaml index 0e3133f..420456d 100644 --- a/deploy/seaweedfs/seaweedfs-svc.yaml +++ b/deploy/seaweedfs/seaweedfs-svc.yaml @@ -21,22 +21,19 @@ metadata: name: seaweedfs namespace: openserverless spec: - type: NodePort + type: ClusterIP ports: - name: s3-api port: 9000 targetPort: 9000 - nodePort: 32090 protocol: TCP - name: filer port: 9090 targetPort: 9090 - nodePort: 32091 protocol: TCP - name: master port: 9333 targetPort: 9333 - nodePort: 32093 protocol: TCP selector: - app: seaweedfs \ No newline at end of file + app: seaweedfs diff --git a/openserverless/datastore_security.py b/openserverless/datastore_security.py index 12925a0..6ecfbce 100644 --- a/openserverless/datastore_security.py +++ b/openserverless/datastore_security.py @@ -93,6 +93,21 @@ def policy(component): 'app.kubernetes.io/instance': 'openserverless-prometheus'}}}, {'podSelector': {'matchLabels': {'name': 'openserverless-operator'}}}], 'ports': [{'port': 9091, 'protocol': 'TCP'}]}] + elif component == 'seaweedfs': + selector = {'app': 'seaweedfs'} + def trusted_peer(labels): + return {'podSelector': { + 'matchLabels': labels, + 'matchExpressions': [ + {'key': 'user-action-pod', 'operator': 'DoesNotExist'}]}} + trusted = [ + trusted_peer({'name': 'openserverless-operator'}), + trusted_peer(selector)] + ingress = [ + {'ports': [{'port': 9000, 'protocol': 'TCP'}]}, + {'from': trusted, + 'ports': [{'port': 9090, 'protocol': 'TCP'}, + {'port': 9333, 'protocol': 'TCP'}]}] else: raise ValueError('Unknown datastore component') return {'apiVersion': 'networking.k8s.io/v1', 'kind': 'NetworkPolicy', diff --git a/openserverless/kopf_util.py b/openserverless/kopf_util.py index 25a0f22..b0a6c52 100644 --- a/openserverless/kopf_util.py +++ b/openserverless/kopf_util.py @@ -69,9 +69,10 @@ def check_seaweedfs_ingresses(response: dict, item: dict): """ Forces an update of seaweed-ingresses if needed """ - if(item['path']=='spec.seaweedfs.ingress.s3-enabled' or item['path']=='spec.seaweedfs.ingress.console-enabled'): - if(item['new']): - response["seaweedfs-ingresses"]="update" + if item['path'] == 'spec.seaweedfs.ingress.console-enabled': + response["seaweedfs-ingresses"] = "update" + elif item['path'] == 'spec.seaweedfs.ingress.s3-enabled' and item['new']: + response["seaweedfs-ingresses"] = "update" def check_registry_ingresses(response: dict, item: dict): """ @@ -170,4 +171,4 @@ def detect_wsku_changes(kopf_diff): response = {} evaluate_wsku_differences(response, differences) - return response \ No newline at end of file + return response diff --git a/openserverless/main.py b/openserverless/main.py index 699a900..77741d3 100644 --- a/openserverless/main.py +++ b/openserverless/main.py @@ -402,6 +402,7 @@ def whisk_update(spec, status, namespace, diff, name, **kwargs): messaging_security.reconcile(owner) monitoring.reconcile(owner) + seaweedfs.reconcile(owner) patcher.patch(diff, status, owner, name) @kopf.on.resume('openserverless.org', 'v1', 'whisks') @@ -411,6 +412,8 @@ def whisk_resume(spec, status, name, **kwargs): messaging_security.reconcile(kube.get(f"wsk/{name}")) if cfg.get('components.monitoring'): monitoring.reconcile(kube.get(f"wsk/{name}")) + if cfg.get('components.seaweedfs'): + seaweedfs.reconcile(kube.get(f"wsk/{name}")) operator_util.whisk_post_resume(name) def runtimes_filter(name, type, **kwargs): diff --git a/openserverless/seaweedfs_deploy.py b/openserverless/seaweedfs_deploy.py index d0f9c1c..28dfc02 100644 --- a/openserverless/seaweedfs_deploy.py +++ b/openserverless/seaweedfs_deploy.py @@ -23,6 +23,7 @@ import openserverless.openwhisk as openwhisk import openserverless.seaweedfs_ingress as seaweedfs_ingress import openserverless.operator_util as operator_util +import openserverless.datastore_security as datastore_security from openserverless.user_config import UserConfig from openserverless.user_metadata import UserMetadata @@ -31,6 +32,13 @@ def _get_seaweedfs_service(): return util.get_service("{.items[?(@.spec.selector.app == 'seaweedfs')]}") +def _render_seaweedfs_spec(data): + tplp = ["pvc-attach.yaml"] + if data['affinity'] or data['tolerations']: + tplp.append("affinity-tolerance-sts-core-attach.yaml") + kust = kus.patchTemplates("seaweedfs", tplp, data) + return kus.kustom_list("seaweedfs", kust, templates=[], data=data) + def _add_seaweedfs_user_metadata(ucfg: UserConfig, user_metadata:UserMetadata): """ adds entries for seaweedfs connectivity S3_HOST, S3_PORT, S3_ACCESS_KEY, S3_SECRET_KEY @@ -65,19 +73,17 @@ def create(owner=None): data = util.get_seaweedfs_config_data() - tplp = ["pvc-attach.yaml"] - - if(data['affinity'] or data['tolerations']): - tplp.append("affinity-tolerance-sts-core-attach.yaml") - - kust = kus.patchTemplates("seaweedfs", tplp, data) - spec = kus.kustom_list("seaweedfs", kust, templates=[], data=data) + spec = _render_seaweedfs_spec(data) + policy = datastore_security.policy('seaweedfs') + spec['items'].append(policy) if owner: kopf.append_owner_reference(spec['items'], owner) else: cfg.put("state.seaweedfs.spec", spec) + # Establish the deny-by-default boundary before creating any listener. + datastore_security.apply(policy) res = kube.apply(spec) # dynamically detect seaweedfs pod and wait for readiness @@ -92,6 +98,28 @@ def create(owner=None): logging.info("*** configured seaweedfs standalone") return res +def reconcile(owner=None): + """Secure an existing SeaweedFS deployment without recreating tenant data.""" + if not cfg.get('components.seaweedfs'): + return None + if not owner or not owner.get('metadata', {}).get('uid'): + raise RuntimeError('Whisk owner is required to reconcile SeaweedFS isolation') + + data = util.get_seaweedfs_config_data() + rendered = _render_seaweedfs_spec(data) + services = [item for item in rendered['items'] + if item['kind'] == 'Service' and item['metadata']['name'] == 'seaweedfs'] + if len(services) != 1: + raise RuntimeError('Could not render the SeaweedFS service') + + policy = datastore_security.policy('seaweedfs') + resources = [policy, services[0]] + kopf.append_owner_reference(resources, owner) + datastore_security.apply(policy) + result = kube.apply(services[0]) + seaweedfs_ingress.create_seaweedfs_ingresses(data, owner) + return result + def _annotate_nuv_metadata(data): """ annotate openserverless configmap with entries for S3 connectivity S3_ENDPOINT, S3_PORT, S3_ACCESS_KEY, S3_SECRET_KEY diff --git a/openserverless/seaweedfs_ingress.py b/openserverless/seaweedfs_ingress.py index 87c0861..7f44403 100644 --- a/openserverless/seaweedfs_ingress.py +++ b/openserverless/seaweedfs_ingress.py @@ -119,6 +119,13 @@ def create_seaweedfs_ingresses(data, owner=None): apihost_url = apihost_util.get_apihost(runtime) res = "" + # The Filer API has no tenant authentication boundary and must never be + # exposed publicly. Remove resources and metadata left by older releases. + ingress_class = util.get_ingress_class(runtime) + delete_seaweedfs_ingress(runtime, "openserverless", ingress_class, + "seaweedfs-filer", owner) + openwhisk.annotate("s3_console_url-") + if data['seaweedfs_s3_ingress_enabled']: s3_hostname_url = get_seaweedfs_ingress_hostname(runtime, apihost_url,"s3",data['seaweedfs_s3_ingress_hostname']) res += create_s3_ingress_endpoint(data, runtime, s3_hostname_url, owner) @@ -127,11 +134,7 @@ def create_seaweedfs_ingresses(data, owner=None): openwhisk.annotate(f"s3_api_url={s3_hostname_url}") if data['seaweedfs_console_ingress_enabled']: - filer_hostname_url = get_seaweedfs_ingress_hostname(runtime, apihost_url,"filer",data['seaweedfs_console_ingress_hostname']) - res += create_console_ingress_endpoint(data, runtime, filer_hostname_url, owner) - - if res: - openwhisk.annotate(f"s3_console_url={filer_hostname_url}") + logging.warning("SeaweedFS Filer ingress is disabled because the Filer API has no tenant authentication") return res @@ -145,13 +148,17 @@ def delete_seaweedfs_ingress(runtime, namespace, ingress_class, type, owner=None try: res = "" if(runtime=='openshift'): - res = kube.kubectl("delete", "route",endpoint.api_route_name(namespace,type)) + res = kube.kubectl("delete", "route",endpoint.api_route_name(namespace,type), + "--ignore-not-found") return res - res += kube.kubectl("delete", "ingress",endpoint.api_ingress_name(namespace,type)) + res += kube.kubectl("delete", "ingress",endpoint.api_ingress_name(namespace,type), + "--ignore-not-found") if(ingress_class == 'traefik'): - res = kube.kubectl("delete", util.get_traefik_middleware_resource(),endpoint.api_middleware_ingress_name(namespace,type)) + res += kube.kubectl("delete", util.get_traefik_middleware_resource(), + endpoint.api_middleware_ingress_name(namespace,type), + "--ignore-not-found") return res except Exception as e: diff --git a/tests/test_datastore_security.py b/tests/test_datastore_security.py index 5b0701d..4688432 100644 --- a/tests/test_datastore_security.py +++ b/tests/test_datastore_security.py @@ -27,6 +27,9 @@ class SecurityTests(unittest.TestCase): + def tearDown(self): + config.clean() + def test_debug_logs_never_contain_values(self): config.clean() config.configure({'etcd': {'root': {'password': 'sentinel-password'}}, @@ -97,6 +100,103 @@ def test_monitoring_does_not_allow_all_peers(self): self.assertTrue(all(peer['podSelector']['matchLabels'] for peer in metrics[0]['from'])) self.assertFalse(any('namespaceSelector' in peer for peer in metrics[0]['from'])) + def test_seaweedfs_service_has_no_node_ports(self): + import yaml + root = Path(__file__).resolve().parents[1] + service = yaml.safe_load((root/'deploy/seaweedfs/seaweedfs-svc.yaml').read_text()) + self.assertEqual(service['spec']['type'], 'ClusterIP') + self.assertTrue(all('nodePort' not in port for port in service['spec']['ports'])) + + def test_seaweedfs_filer_and_master_are_restricted(self): + rules = security.policy('seaweedfs')['spec']['ingress'] + s3 = [rule for rule in rules if any(port['port'] == 9000 for port in rule['ports'])] + control = [rule for rule in rules if any(port['port'] == 9090 for port in rule['ports'])] + self.assertEqual(s3, [{'ports': [{'port': 9000, 'protocol': 'TCP'}]}]) + self.assertEqual(len(control), 1) + self.assertEqual({port['port'] for port in control[0]['ports']}, {9090, 9333}) + self.assertEqual(control[0]['from'], [ + {'podSelector': { + 'matchLabels': {'name': 'openserverless-operator'}, + 'matchExpressions': [ + {'key': 'user-action-pod', 'operator': 'DoesNotExist'}]}}, + {'podSelector': { + 'matchLabels': {'app': 'seaweedfs'}, + 'matchExpressions': [ + {'key': 'user-action-pod', 'operator': 'DoesNotExist'}]}}]) + + def test_seaweedfs_action_cannot_impersonate_trusted_peers(self): + control = security.policy('seaweedfs')['spec']['ingress'][1] + for peer in control['from']: + selector = peer['podSelector'] + self.assertIn({'key': 'user-action-pod', 'operator': 'DoesNotExist'}, + selector['matchExpressions']) + + def test_seaweedfs_create_applies_isolation_policy(self): + from openserverless import seaweedfs_deploy + spec = {'apiVersion': 'v1', 'kind': 'List', 'items': []} + with patch.object(seaweedfs_deploy.util, 'get_seaweedfs_config_data', return_value={ + 'affinity': False, 'tolerations': False}), \ + patch.object(seaweedfs_deploy.kus, 'patchTemplates'), \ + patch.object(seaweedfs_deploy.kus, 'kustom_list', return_value=spec), \ + patch.object(seaweedfs_deploy.cfg, 'put'), \ + patch.object(seaweedfs_deploy.kube, 'apply') as apply, \ + patch.object(seaweedfs_deploy.util, 'wait_for_pod_ready'), \ + patch.object(seaweedfs_deploy.seaweedfs_ingress, 'create_seaweedfs_ingresses'), \ + patch.object(seaweedfs_deploy.util, 'wait_for_http'), \ + patch.object(seaweedfs_deploy, 'create_seaweedfs_nuv_storage'): + seaweedfs_deploy.create() + self.assertIn(security.policy('seaweedfs'), spec['items']) + self.assertEqual(apply.call_args_list[0].args[0]['items'], + [security.policy('seaweedfs')]) + + def test_existing_cluster_reconcile_secures_before_service_update(self): + from openserverless import seaweedfs_deploy + owner = {'apiVersion': 'openserverless.org/v1', 'kind': 'Whisk', + 'metadata': {'name': 'controller', 'uid': 'synthetic'}} + service = {'apiVersion': 'v1', 'kind': 'Service', + 'metadata': {'name': 'seaweedfs'}, + 'spec': {'type': 'ClusterIP', 'ports': []}} + events = [] + config.configure({'components': {'seaweedfs': True}}) + with patch.object(seaweedfs_deploy.util, 'get_seaweedfs_config_data', return_value={}), \ + patch.object(seaweedfs_deploy, '_render_seaweedfs_spec', return_value={ + 'apiVersion': 'v1', 'kind': 'List', 'items': [service]}), \ + patch.object(seaweedfs_deploy.datastore_security, 'apply', + side_effect=lambda obj: events.append(('policy', obj))), \ + patch.object(seaweedfs_deploy.kube, 'apply', + side_effect=lambda obj: events.append(('service', obj))), \ + patch.object(seaweedfs_deploy.seaweedfs_ingress, + 'create_seaweedfs_ingresses', + side_effect=lambda data, obj: events.append(('ingress', obj))): + seaweedfs_deploy.reconcile(owner) + self.assertEqual([event[0] for event in events], ['policy', 'service', 'ingress']) + self.assertEqual(events[0][1]['kind'], 'NetworkPolicy') + self.assertEqual(events[1][1]['spec']['type'], 'ClusterIP') + + def test_seaweedfs_console_ingress_is_removed_not_created(self): + from openserverless import seaweedfs_ingress + data = {'seaweedfs_s3_ingress_enabled': False, + 'seaweedfs_console_ingress_enabled': True} + with patch.object(seaweedfs_ingress.cfg, 'get', return_value='k3s'), \ + patch.object(seaweedfs_ingress.apihost_util, 'get_apihost', return_value='miniops.me'), \ + patch.object(seaweedfs_ingress.util, 'get_ingress_class', return_value='traefik'), \ + patch.object(seaweedfs_ingress, 'delete_seaweedfs_ingress') as delete, \ + patch.object(seaweedfs_ingress, 'create_console_ingress_endpoint') as create, \ + patch.object(seaweedfs_ingress.openwhisk, 'annotate') as annotate: + with self.assertLogs(level=logging.WARNING): + seaweedfs_ingress.create_seaweedfs_ingresses(data) + delete.assert_called_once_with('k3s', 'openserverless', 'traefik', + 'seaweedfs-filer', None) + create.assert_not_called() + annotate.assert_called_once_with('s3_console_url-') + + def test_disabling_console_schedules_legacy_ingress_cleanup(self): + from openserverless import kopf_util + response = {} + kopf_util.check_seaweedfs_ingresses(response, { + 'path': 'spec.seaweedfs.ingress.console-enabled', 'new': False}) + self.assertEqual(response, {'seaweedfs-ingresses': 'update'}) + def test_etcd_peer_port_is_restricted_to_members(self): rules = security.policy('etcd')['spec']['ingress'] peer_rule = [rule for rule in rules if any(p['port'] == 2380 for p in rule['ports'])]