Isolate SeaweedFS control APIs from tenant actions

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
+    app: seaweedfs

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',

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    
+    return response

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):

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

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:

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'])]