feat: adapt secret resolver for customization

src/sap_cloud_sdk/core/secret_resolver/__init__.py

@@ -1,26 +1,64 @@
 """
-Secret resolver: load configuration/secrets from mounted files or environment variables
+Secret resolver: load configuration/secrets from mounted files or environment variables.
 
-Usage:
-    from dataclasses import dataclass, field
-    from sap_cloud_sdk.secret_resolver import read_from_mount_and_fallback_to_env_var
+Built-in resolvers and chain builder::
 
-    @dataclass
-    class MyConfig:
-        username: str = field(metadata={"secret": "username"})
-        password: str = field(metadata={"secret": "password"})
-        endpoint: str = "http://localhost"
+    from sap_cloud_sdk.core.secret_resolver import (
+        MountResolver,
+        EnvVarResolver,
+        ChainedResolver,
+    )
+
+    # Build a chain explicitly
+    resolver = ChainedResolver([MountResolver(), EnvVarResolver()])
+    resolver.resolve("destination", "default", binding)
+
+Legacy function-based API (still supported)::
+
+    from sap_cloud_sdk.core.secret_resolver import read_from_mount_and_fallback_to_env_var
 
-    cfg = MyConfig()
     read_from_mount_and_fallback_to_env_var(
         base_volume_mount="/etc/secrets/appfnd",
         base_var_name="CLOUD_SDK_CFG",
-        module="objectstore",
+        module="destination",
         instance="default",
-        target=cfg
+        target=binding,
     )
 """
 
-from .resolver import read_from_mount_and_fallback_to_env_var, resolve_base_mount
+from sap_cloud_sdk.core.secret_resolver.resolver import (
+    read_from_mount_and_fallback_to_env_var,
+)
+from sap_cloud_sdk.core.secret_resolver._resolvers import (
+    Resolver,
+    ChainedResolver,
+)
+
+from sap_cloud_sdk.core.secret_resolver.mount_resolver import (
+    MountResolver,
+    resolve_base_mount,
+)
+from sap_cloud_sdk.core.secret_resolver.env_resolver import EnvVarResolver
+
+from sap_cloud_sdk.core.secret_resolver.sdk_config import (
+    SdkConfig,
+    configure,
+    get_sdk_config,
+    get_resolver,
+)
 
-__all__ = ["read_from_mount_and_fallback_to_env_var", "resolve_base_mount"]
+__all__ = [
+    # Class-based API
+    "Resolver",
+    "MountResolver",
+    "EnvVarResolver",
+    "ChainedResolver",
+    # Global configuration
+    "SdkConfig",
+    "configure",
+    "get_sdk_config",
+    "get_resolver",
+    # Legacy function-based API
+    "read_from_mount_and_fallback_to_env_var",
+    "resolve_base_mount",
+]

src/sap_cloud_sdk/core/secret_resolver/_mapping.py

@@ -0,0 +1,32 @@
+"""Utilities for mapping dataclass fields to secret store keys."""
+
+from typing import Any, Dict, Tuple
+from dataclasses import fields, is_dataclass
+
+
+def _get_field_map(target: Any) -> dict[str, tuple[str, type]]:
+    """
+    Build a mapping from secret key -> (attribute_name, attribute_type) for a dataclass instance.
+
+    Priority:
+      1. Use field.metadata["secret"] if present as the key
+      2. Fallback to the lowercase dataclass field name
+    Only string-typed fields are supported.
+    """
+    if not is_dataclass(target) or isinstance(target, type):
+        raise TypeError("target must be a dataclass instance")
+
+    mapping: Dict[str, Tuple[str, type]] = {}
+    for f in fields(target):
+        # Only support string fields for secrets (consistent with Go SDK)
+        # Allow plain 'str' annotations; reject others to keep behavior predictable
+        if f.type is not str:
+            raise TypeError(
+                f"target field '{f.name}' is not a string (only str fields are supported)"
+            )
+        key = f.metadata.get("secret") if hasattr(f, "metadata") else None
+        if key and isinstance(key, str) and key.strip():
+            mapping[key] = (f.name, f.type)
+        else:
+            mapping[f.name.lower()] = (f.name, f.type)
+    return mapping

src/sap_cloud_sdk/core/secret_resolver/_resolvers.py

@@ -0,0 +1,92 @@
+"""BindingResolver protocol and built-in implementations.
+
+This module defines the core extensibility contract for secret resolution.
+Each resolver encapsulates one binding source. Compose them into an ordered
+chain via :class:`ChainedResolver` — the first resolver that succeeds wins.
+
+Protocol contract::
+
+    resolver.resolve(module, instance, target)
+
+- On success: populates ``target`` in-place, returns ``None``
+- On failure: raises any exception; the chain tries the next resolver
+"""
+
+from __future__ import annotations
+
+from dataclasses import fields, is_dataclass
+from typing import Any, Protocol, runtime_checkable
+
+
+@runtime_checkable
+class Resolver(Protocol):
+    """Contract for a single binding resolution strategy.
+
+    A ``BindingResolver`` reads credentials from one source and populates
+    ``target`` in-place. Implementations raise on failure so that a
+    :class:`ChainedResolver` can try the next strategy.
+
+    Any object implementing ``resolve`` with this signature satisfies the
+    protocol — no inheritance required.
+    """
+
+    def resolve(self, module: str, instance: str, target: Any) -> None:
+        """Populate ``target`` with credentials for ``module``/``instance``.
+
+        Args:
+            module: Service module name (e.g. ``"destination"``).
+            instance: Instance identifier (e.g. ``"default"``).
+            target: Dataclass instance whose ``str`` fields will be set.
+
+        Raises:
+            Any exception on failure; the caller determines how to handle it.
+        """
+        ...
+
+
+class ChainedResolver:
+    """Tries each resolver in order; returns on the first success.
+
+    Collects failure messages from each resolver and raises a
+    :class:`RuntimeError` with an aggregated report when all resolvers fail.
+
+    Args:
+        resolvers: Ordered list of :class:`BindingResolver` implementations to try.
+        base_var_name: Used only for the error guidance message.
+    """
+
+    def __init__(
+        self,
+        resolvers: list[Resolver],
+        base_var_name: str = "CLOUD_SDK_CFG",
+    ) -> None:
+        if not resolvers:
+            raise ValueError("resolvers list must not be empty")
+        self._resolvers = resolvers
+        self._base_var_name = base_var_name
+
+    def resolve(self, module: str, instance: str, target: Any) -> None:
+        """Try each resolver in order; raise on total failure."""
+        if not is_dataclass(target) or isinstance(target, type):
+            raise TypeError("target must be a dataclass instance")
+        for f in fields(target):
+            if f.type is not str and f.type != "str":
+                raise TypeError(
+                    f"target field {f.name!r} is not a string (only str fields are supported)"
+                )
+
+        errors: list[str] = []
+        for resolver in self._resolvers:
+            try:
+                resolver.resolve(module, instance, target)
+                return
+            except Exception as e:
+                label = type(resolver).__name__
+                errors.append(f"{label} failed: {e}")
+
+        raise RuntimeError(
+            f"module={module!r} instance={instance!r} failed to read secrets from all resolvers: "
+            f"{errors}. "
+            "Options: mount secrets under the service binding path, set environment variables "
+            f"like {self._base_var_name}_{module}_{instance}_<KEY> (uppercased), or set VCAP_SERVICES."
+        )

src/sap_cloud_sdk/core/secret_resolver/constants.py

@@ -3,3 +3,5 @@
 """
 
 BASE_MOUNT_PATH = "/etc/secrets/appfnd"
+BASE_VAR_NAME = "CLOUD_SDK_CFG"
+SERVICE_BINDING_ROOT = "SERVICE_BINDING_ROOT"

src/sap_cloud_sdk/core/secret_resolver/env_resolver.py

@@ -0,0 +1,47 @@
+"""Resolver that reads service binding secrets from environment variables."""
+
+import os
+from typing import Any
+
+from sap_cloud_sdk.core.secret_resolver._mapping import _get_field_map
+from sap_cloud_sdk.core.secret_resolver.constants import BASE_VAR_NAME
+
+
+class EnvVarResolver:
+    """Resolves bindings from environment variables.
+
+    Reads variables named ``{base_var_name}_{module}_{instance}_{field_key}``
+    (uppercased, hyphens in module/instance replaced with underscores).
+
+    Args:
+        base_var_name: Env var name prefix. Defaults to ``"CLOUD_SDK_CFG"``.
+    """
+
+    def __init__(self, base_var_name: str = BASE_VAR_NAME) -> None:
+        self._base_var_name = base_var_name
+
+    def resolve(self, module: str, instance: str, target: Any) -> None:
+        """Load secrets from environment variables."""
+        normalized_module = module.replace("-", "_")
+        normalized_instance = instance.replace("-", "_")
+        _load_from_env(
+            self._base_var_name, normalized_module, normalized_instance, target
+        )
+
+
+def _load_from_env(base_var_name: str, module: str, instance: str, target: Any) -> None:
+    """
+    Load secrets from environment variables with names:
+        {base_var_name}_{module}_{instance}_{field_key} (uppercased)
+    instance names have '-' replaced with '_' for env var compatibility.
+    """
+    field_map = _get_field_map(target)
+    prefix = f"{base_var_name}_{module}_{instance}".upper()
+
+    for key, (attr_name, _) in field_map.items():
+        var_name = f"{prefix}_{key}".upper()
+        value = os.environ.get(var_name)
+        if value is None:
+            # Align with Go: error if env var not found
+            raise KeyError(f"env var not found: {var_name}")
+        setattr(target, attr_name, value)

src/sap_cloud_sdk/core/secret_resolver/mount_resolver.py

@@ -0,0 +1,93 @@
+"""Resolver that reads service binding secrets from a mounted volume path."""
+
+import os
+from typing import Any
+
+from sap_cloud_sdk.core.secret_resolver._mapping import _get_field_map
+from sap_cloud_sdk.core.secret_resolver.constants import (
+    BASE_MOUNT_PATH,
+    SERVICE_BINDING_ROOT,
+)
+
+
+class MountResolver:
+    """Resolves bindings from a mounted volume path.
+
+    Reads secret files at ``{base_volume_mount}/{module}/{instance}/{field_key}``.
+    Respects the ``SERVICE_BINDING_ROOT`` environment variable (servicebinding.io
+    spec) as an override for ``base_volume_mount``.
+
+    Args:
+        base_volume_mount: Base path for mounted secrets. Defaults to
+            ``/etc/secrets/appfnd``.
+    """
+
+    def __init__(self, base_volume_mount: str = BASE_MOUNT_PATH) -> None:
+        self._base_volume_mount = base_volume_mount
+
+    def resolve(self, module: str, instance: str, target: Any) -> None:
+        """Load secrets from the mounted volume path."""
+        effective_base = resolve_base_mount(self._base_volume_mount)
+        _load_from_mount(effective_base, module, instance, target)
+
+
+def resolve_base_mount(base_volume_mount: str = BASE_MOUNT_PATH) -> str:
+    """Resolve the base mount path for service binding discovery.
+
+    Checks the ``SERVICE_BINDING_ROOT`` environment variable first (as defined
+    by the `servicebinding.io <https://servicebinding.io/spec/core/1.1.0/>`_
+    specification). Falls back to ``base_volume_mount`` when the env var is
+    absent.
+
+    Args:
+        base_volume_mount: Default base path used when ``SERVICE_BINDING_ROOT``
+            is not set. Defaults to ``/etc/secrets/appfnd``.
+
+    Returns:
+        The effective base path for secret mount resolution.
+    """
+    return os.environ.get(SERVICE_BINDING_ROOT, base_volume_mount)
+
+
+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}
+
+    Sets string attributes directly on the dataclass instance.
+    """
+    secret_dir = os.path.join(base_volume_mount, module, instance)
+    _validate_path(secret_dir)
+
+    field_map = _get_field_map(target)
+    for key, (attr_name, _) in field_map.items():
+        file_path = os.path.join(secret_dir, key)
+        try:
+            # Read entire file content as text; do not strip newlines to match Go behavior
+            with open(file_path, "r", encoding="utf-8") as f:
+                content = f.read()
+        except FileNotFoundError as e:
+            # Align with Go: surface precise file error
+            raise FileNotFoundError(
+                f"failed to read secret file {file_path}: {e}"
+            ) from e
+        except OSError as e:
+            raise OSError(f"failed to read secret file {file_path}: {e}") from e
+
+        # Set target field (string only)
+        setattr(target, attr_name, content)
+
+
+def _validate_path(path: str) -> None:
+    """Validate that the given path exists and is a directory."""
+    try:
+        _st = os.stat(path)
+    except FileNotFoundError as e:
+        raise FileNotFoundError(f"path does not exist: {path}") from e
+    except OSError as e:
+        raise OSError(f"cannot access path {path}: {e}") from e
+    # If exists, ensure it's a directory
+    if not os.path.isdir(path):
+        raise NotADirectoryError(f"path is not a directory: {path}")