fix(secret-resolver): update secret resolver logic when set SERVICE_BINDING_ROOT (#121)

Author: cassiofariasmachado

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "sap-cloud-sdk"
-version = "0.21.0"
+version = "0.21.1"
 description = "SAP Cloud SDK for Python"
 readme = "README.md"
 license = "Apache-2.0"

src/sap_cloud_sdk/core/secret_resolver/resolver.py

@@ -75,16 +75,14 @@ def _get_field_map(target: Any) -> Dict[str, Tuple[str, type]]:
     return mapping
 
 
-def _load_from_mount(
-    base_volume_mount: str, module: str, instance: str, target: Any
-) -> None:
+def _load_from_path(secret_dir: str, target: Any) -> None:
     """
-    Load secrets from files at:
-        {base_volume_mount}/{module}/{instance}/{field_key}
+    Load secrets from files directly in secret_dir into target dataclass.
 
-    Sets string attributes directly on the dataclass instance.
+    Reads each field key as a file name under secret_dir. Used for both the
+    servicebinding.io flat layout ($ROOT/<module>/<field>) and the legacy
+    three-level layout via :func:`_load_from_mount`.
     """
-    secret_dir = os.path.join(base_volume_mount, module, instance)
     _validate_path(secret_dir)
 
     field_map = _get_field_map(target)
@@ -106,6 +104,17 @@ def _load_from_mount(
         setattr(target, attr_name, content)
 
 
+def _load_from_mount(
+    base_volume_mount: str, module: str, instance: str, target: Any
+) -> None:
+    """
+    Load secrets from files at:
+        {base_volume_mount}/{module}/{instance}/{field_key}
+    """
+    secret_dir = os.path.join(base_volume_mount, module, instance)
+    _load_from_path(secret_dir, target)
+
+
 def _load_from_env(base_var_name: str, module: str, instance: str, target: Any) -> None:
     """
     Load secrets from environment variables with names:
@@ -133,17 +142,21 @@ def read_from_mount_and_fallback_to_env_var(
 ) -> None:
     """
     Load secrets for a given module and instance into the provided dataclass instance `target`.
-    Fallback order:
-      1. Mounted volume path: {base_volume_mount}/{module}/{instance}/{field_key}
-         (``SERVICE_BINDING_ROOT`` env var overrides ``base_volume_mount`` — see
-         :func:`resolve_base_mount`)
+
+    Fallback order when ``SERVICE_BINDING_ROOT`` is set:
+      1. Flat path: {SERVICE_BINDING_ROOT}/{module}/{field_key}  (servicebinding.io spec)
+      2. Full path: {SERVICE_BINDING_ROOT}/{module}/{instance}/{field_key}  (legacy convention)
+      3. Environment variables: {base_var_name}_{module}_{instance}_{field_key} (uppercased)
+
+    Fallback order when ``SERVICE_BINDING_ROOT`` is **not** set:
+      1. Full path: {base_volume_mount}/{module}/{instance}/{field_key}
       2. Environment variables: {base_var_name}_{module}_{instance}_{field_key} (uppercased)
 
     Raises:
       ValueError: If inputs are invalid or target is not a dataclass instance
       FileNotFoundError / NotADirectoryError / OSError: If mount path issues occur
       KeyError: If environment variables are missing on fallback
-      RuntimeError: If both mount and env var loading fail (aggregated error)
+      RuntimeError: If all strategies fail (aggregated error)
     """
     _validate_inputs(module, instance)
 
@@ -152,6 +165,15 @@ def read_from_mount_and_fallback_to_env_var(
     normalized_module = module.replace("-", "_")
     normalized_instance = instance.replace("-", "_")
 
+    # servicebinding.io: when SERVICE_BINDING_ROOT is explicitly set, try the flat path
+    # $ROOT/<module>/<field> before the legacy $ROOT/<module>/<instance>/<field> path.
+    if os.environ.get("SERVICE_BINDING_ROOT") is not None:
+        try:
+            _load_from_path(os.path.join(resolved_base_path, module), target)
+            return
+        except Exception as e:
+            errors.append(f"mount failed: {e};")
+
     try:
         _load_from_mount(resolved_base_path, module, instance, target)
         return

tests/core/unit/secret_resolver/unit/test_secret_resolver.py

@@ -199,7 +199,8 @@ def test_service_binding_root_overrides_base_mount(self, mock_file, mock_stat, m
         config = SampleConfig()
         read_from_mount_and_fallback_to_env_var("/etc/secrets/appfnd", "VAR", "module", "instance", config)
         first_call_path = mock_file.call_args_list[0][0][0]
-        assert first_call_path.startswith("/custom/root")
+        # With SERVICE_BINDING_ROOT set, flat path is tried first: $ROOT/<module>/<field>
+        assert first_call_path == "/custom/root/module/user"
 
     @patch.dict(os.environ, {}, clear=True)
     @patch('os.path.isdir', return_value=True)
@@ -214,4 +215,43 @@ def test_default_base_mount_used_when_no_service_binding_root(self, mock_file, m
         config = SampleConfig()
         read_from_mount_and_fallback_to_env_var("/etc/secrets/appfnd", "VAR", "module", "instance", config)
         first_call_path = mock_file.call_args_list[0][0][0]
-        assert first_call_path.startswith("/etc/secrets/appfnd")
+        # Without SERVICE_BINDING_ROOT, only the legacy $ROOT/<module>/<instance>/<field> path is tried
+        assert first_call_path == "/etc/secrets/appfnd/module/instance/user"
+
+    @patch.dict(os.environ, {"SERVICE_BINDING_ROOT": "/bindings"})
+    @patch('os.path.isdir', return_value=True)
+    @patch('os.stat')
+    @patch('builtins.open', new_callable=mock_open)
+    def test_service_binding_root_flat_path_success(self, mock_file, mock_stat, mock_isdir):
+        mock_file.side_effect = [
+            mock_open(read_data="flat_user").return_value,
+            mock_open(read_data="flat_pass").return_value,
+            mock_open(read_data="flat_endpoint").return_value,
+        ]
+        config = SampleConfig()
+        read_from_mount_and_fallback_to_env_var("/etc/secrets/appfnd", "VAR", "module", "instance", config)
+        first_call_path = mock_file.call_args_list[0][0][0]
+        # Flat path $ROOT/<module>/<field> is tried first
+        assert first_call_path == "/bindings/module/user"
+        assert config.username == "flat_user"
+
+    @patch.dict(os.environ, {"SERVICE_BINDING_ROOT": "/bindings"})
+    @patch('os.path.isdir', return_value=True)
+    @patch('os.stat')
+    def test_service_binding_root_flat_fails_falls_back_to_module_instance(self, mock_stat, mock_isdir):
+        # Flat path: directory exists but field files are not there (old AppFND structure)
+        # Full path: files are present under <module>/<instance>/
+        flat_not_found = FileNotFoundError("flat file missing")
+        mock_file_calls = [
+            flat_not_found,                                   # flat: <module>/user → not found
+            mock_open(read_data="legacy_user").return_value,  # full: <module>/<instance>/user
+            mock_open(read_data="legacy_pass").return_value,  # full: <module>/<instance>/password
+            mock_open(read_data="legacy_ep").return_value,    # full: <module>/<instance>/endpoint
+        ]
+        with patch('builtins.open', side_effect=mock_file_calls):
+            config = SampleConfig()
+            read_from_mount_and_fallback_to_env_var("/bindings", "VAR", "module", "instance", config)
+
+        assert config.username == "legacy_user"
+        assert config.password == "legacy_pass"
+        assert config.endpoint == "legacy_ep"