fix: remove unsafe webdav option (#1168)

Closes #1167.

Remove unsafe option `bearer_token_command` from the WebDAV provider.

See also #964.

Author: leafty

components/renku_data_services/storage/rclone.py

@@ -14,7 +14,11 @@
 from renku_data_services import errors
 from renku_data_services.app_config import logging
 from renku_data_services.storage.constants import ENVIDAT_V1_PROVIDER
-from renku_data_services.storage.rclone_patches import BANNED_SFTP_OPTIONS, BANNED_STORAGE, apply_patches
+from renku_data_services.storage.rclone_patches import (
+    BANNED_OPTIONS,
+    BANNED_STORAGE,
+    apply_patches,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -412,9 +416,10 @@ def get_option_for_provider(self, name: str, provider: str | None) -> RCloneOpti
 
     def check_unsafe_option(self, name: str) -> None:
         """Check that the option is safe."""
-        if self.prefix != "sftp":
+        banned = BANNED_OPTIONS.get(self.prefix.lower(), None)
+        if banned is None:
             return None
-        if name in BANNED_SFTP_OPTIONS:
+        if name in banned:
             raise errors.ValidationError(message=f"The {name} option is not allowed.")
         return None
 

components/renku_data_services/storage/rclone_patches.py

@@ -38,11 +38,16 @@
     "zoho",
 }
 
-BANNED_SFTP_OPTIONS: Final[set[str]] = {
-    "key_file",  # path to a local file
-    "pubkey_file",  # path to a local file
-    "known_hosts_file",  # path to a local file
-    "ssh",  # arbitrary command to be executed
+BANNED_OPTIONS: Final[dict[str, set[str]]] = {
+    "sftp": {
+        "key_file",  # path to a local file
+        "pubkey_file",  # path to a local file
+        "known_hosts_file",  # path to a local file
+        "ssh",  # arbitrary command to be executed
+    },
+    "webdav": {
+        "bearer_token_command",  # arbitrary command to be executed
+    },
 }
 
 
@@ -252,14 +257,15 @@ def __patch_switchdrive_storage(spec: list[dict[str, Any]]) -> None:
     )
 
 
-def __patch_schema_remove_banned_sftp_options(spec: list[dict[str, Any]]) -> None:
-    """Remove unsafe SFTP options."""
-    sftp = find_storage(spec, "sftp")
-    options = []
-    for option in sftp["Options"]:
-        if option["Name"] not in BANNED_SFTP_OPTIONS:
-            options.append(option)
-    sftp["Options"] = options
+def __patch_schema_remove_banned_options(spec: list[dict[str, Any]]) -> None:
+    """Remove unsafe options."""
+    for storage_type, banned in BANNED_OPTIONS.items():
+        storage = find_storage(spec, storage_type)
+        options = []
+        for option in storage["Options"]:
+            if option["Name"] not in banned:
+                options.append(option)
+        storage["Options"] = options
 
 
 def __patch_schema_add_openbis_type(spec: list[dict[str, Any]]) -> None:
@@ -354,7 +360,7 @@ def apply_patches(spec: list[dict[str, Any]]) -> None:
         __patch_schema_remove_oauth_propeties,
         __patch_polybox_storage,
         __patch_switchdrive_storage,
-        __patch_schema_remove_banned_sftp_options,
+        __patch_schema_remove_banned_options,
         __patch_schema_add_openbis_type,
     ]
 

test/bases/renku_data_services/data_api/__snapshots__/test_storage.ambr

@@ -17415,18 +17415,6 @@
           'sensitive': True,
           'type': 'string',
         }),
-        dict({
-          'advanced': True,
-          'default': '',
-          'default_str': '',
-          'exclusive': False,
-          'help': 'Command to run to get a bearer token.',
-          'ispassword': False,
-          'name': 'bearer_token_command',
-          'required': False,
-          'sensitive': False,
-          'type': 'string',
-        }),
         dict({
           'advanced': True,
           'default': '',

test/bases/renku_data_services/data_api/test_storage.py

@@ -12,7 +12,7 @@
 from renku_data_services.data_api.dependencies import DependencyManager
 from renku_data_services.migrations.core import run_migrations_for_app
 from renku_data_services.storage.rclone import RCloneValidator
-from renku_data_services.storage.rclone_patches import BANNED_SFTP_OPTIONS, BANNED_STORAGE, OAUTH_PROVIDERS
+from renku_data_services.storage.rclone_patches import BANNED_OPTIONS, BANNED_STORAGE, OAUTH_PROVIDERS
 from renku_data_services.utils.core import get_openbis_session_token
 from test.utils import SanicReusableASGITestClient
 
@@ -725,7 +725,10 @@ async def test_storage_schema_patches(storage_test_client, snapshot) -> None:
     # check that unsafe SFTP options are removed
     sftp = next((e for e in schema if e["prefix"] == "sftp"), None)
     assert sftp
-    assert all(o["name"] not in BANNED_SFTP_OPTIONS for o in sftp["options"])
+    assert all(o["name"] not in BANNED_OPTIONS["sftp"] for o in sftp["options"])
+    webdav = next((e for e in schema if e["prefix"] == "webdav"), None)
+    assert webdav
+    assert all(o["name"] not in BANNED_OPTIONS["webdav"] for o in webdav["options"])
 
     # snapshot the schema
     assert schema == snapshot