Repository: deepset-ai/haystack
Pull request: feat: Implement `Secret` for structured authentication
### Related Issues

- Fixes https://github.com/deepset-ai/haystack/issues/6851

### Proposed Changes:

Expose a `Secret` type to provide common logic for any component that requires a secret for authentication, with the following variants:
- `token` - Use a string literal.
- `env_var` - Resolve the token from the passed environment variable(s).

Token-based secrets are not serializable, i.e., they'll raise an error when serializing a component that uses it. Env vars are serializable.

### Example usage

```python
@component
class MyComponent:
	def __init__(self, api_key: Optional[Secret] = None, **kwargs):
		self.api_key = api_key
		self.backend = None

	def warm_up(self):
		# Call resolve_value to yield a single result. The semantics of the result is policy-dependent.
		# Currently, all supported policies will return a single string token.
		self.backend = SomeBackend(api_key=self.api_key.resolve_value() if self.api_key else None, ...)

	def to_dict(self):
		# Serialize the policy like any other (custom) data. If the policy is token-based, it will 
		# raise an error.
		return default_to_dict(self, api_key=self.api_key.to_dict() if self.api_key else None, ...)

	@classmethod
	def from_dict(cls, data):
		# Deserialize the policy data before passing it to the generic from_dict function.
		api_key_data = data["init_parameters"]["api_key"]
		api_key = Secret.from_dict(api_key_data) if api_key_data is not None else None
		data["init_parameters"]["api_key"] = api_key
		return default_from_dict(cls, data)

# No authentication.		
component = MyComponent(api_key=None)
# Token based authentication
component = MyComponent(api_key=Secret.from_token("sk-randomAPIkeyasdsa32ekasd32e"))
component.to_dict() # Error! Can't serialize authentication tokens
# Environment variable based authentication
component = MyComponent(api_key=Secret.from_env("OPENAI_API_KEY"))
component.to_dict() # This is fine
```

### How did you test it?

<!-- unit tests, integration tests, manual verification, instructions for manual tests -->
Unit tests

### Notes for the reviewer

<!-- E.g. point out section where the reviewer  -->
- A helper function to aid with deserialization will be introduced in a future PR.

### Checklist

- I have read the [contributors guidelines](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md) and the [code of conduct](https://github.com/deepset-ai/haystack/blob/main/code_of_conduct.txt)
- I have updated the related issue with new insights and changes
- I added unit tests and updated the docstrings
- I've used one of the [conventional commit types](https://www.conventionalcommits.org/en/v1.0.0/) for my PR title: `fix:`, `feat:`, `build:`, `chore:`, `ci:`, `docs:`, `style:`, `refactor:`, `perf:`, `test:`.
- I documented my code
- I ran [pre-commit hooks](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md#installation) and fixed any issue


Problem statement:
Component secret management - Implement a discriminated union that handles the main authentication usecases
Expose an `AuthPolicy` type to provide common logic for any component that requires authentication, with the following variants:
- `Token` - Use a string literal.
- `EnvVar` - Resolve the token from the passed environment variable.

Some backends support additional methods such as on-disk tokens (HF) and functions that yield tokens (Azure). We'll defer supporting them until there is enough demand (env vars and tokens should cover the grand majority of the use cases).

### What we get by implementing this:
- By having common logic, we sorta have our cake and eat it too - Any component that uses the `Token` policy will raise an error during serialization. The others can be serialized as usual.  
- We retain the flexibility of the current setup.
- Easier to maintain.
- Consistency.

Hints from issue discussion:

Commit message: feat: Implement `AuthPolicy` for structured authentication

Patch to review:
diff --git a/haystack/utils/__init__.py b/haystack/utils/__init__.py
index e4cdb87d60..62c13052fb 100644
--- a/haystack/utils/__init__.py
+++ b/haystack/utils/__init__.py
@@ -2,3 +2,4 @@
 from haystack.utils.requests_utils import request_with_retry
 from haystack.utils.filters import document_matches_filter
 from haystack.utils.device import ComponentDevice, DeviceType, Device, DeviceMap
+from haystack.utils.auth import AuthPolicy
diff --git a/haystack/utils/auth.py b/haystack/utils/auth.py
new file mode 100644
index 0000000000..16a4c2beff
--- /dev/null
+++ b/haystack/utils/auth.py
@@ -0,0 +1,194 @@
+from enum import Enum
+import os
+from typing import Any, Dict, List, Optional, Union
+from dataclasses import dataclass
+from abc import ABC, abstractmethod
+
+
+class AuthPolicyType(Enum):
+    TOKEN = "token"
+    ENV_VAR = "env_var"
+
+    def __str__(self):
+        return self.value
+
+    @staticmethod
+    def from_str(string: str) -> "AuthPolicyType":
+        map = {e.value: e for e in AuthPolicyType}
+        type = map.get(string)
+        if type is None:
+            raise ValueError(f"Unknown authentication policy type '{string}'")
+        return type
+
+
+@dataclass
+class AuthPolicy(ABC):
+    """
+    Provides a common interface for authentication policies.
+    """
+
+    _type: AuthPolicyType
+
+    def __init__(self, type: AuthPolicyType):
+        super().__init__()
+        self._type = type
+
+    @staticmethod
+    def from_token(token: str) -> "AuthPolicy":
+        """
+        Create a token authentication policy.
+        This policy cannot be serialized.
+
+        :param token:
+            The token to use for authentication.
+        """
+        return AuthPolicyToken(token)
+
+    @staticmethod
+    def from_env_var(env_vars: Union[str, List[str]], *, strict: bool = True) -> "AuthPolicy":
+        """
+        Create an environment variable authentication policy.
+        Accepts one or more environment variables and fetches
+        a string token from the first one that is set.
+
+        :param env_vars:
+            A single environment variable or an ordered list of
+            candidate environment variables.
+        :param strict:
+            Whether to raise an exception if none of the environment
+            variables are set.
+        """
+        if isinstance(env_vars, str):
+            env_vars = [env_vars]
+        return AuthPolicyEnvVar(env_vars, strict=strict)
+
+    def to_dict(self) -> Dict[str, Any]:
+        """
+        Convert the policy to a JSON-serializable dictionary.
+
+        :returns:
+            The serialized policy.
+        """
+        out = {"type": self._type.value}
+        inner = self._to_dict()
+        assert all(k not in inner for k in out.keys())
+        out.update(inner)
+        return out
+
+    @staticmethod
+    def from_dict(dict: Dict[str, Any]) -> "AuthPolicy":
+        """
+        Create a policy from a JSON-serializable dictionary.
+
+        :param dict:
+            The dictionary to create the policy from.
+        :returns:
+            The deserialized policy.
+        """
+        policy_map = {AuthPolicyType.TOKEN: AuthPolicyToken, AuthPolicyType.ENV_VAR: AuthPolicyEnvVar}
+        policy_type = AuthPolicyType.from_str(dict["type"])
+        return policy_map[policy_type]._from_dict(dict)  # type: ignore
+
+    @abstractmethod
+    def resolve_value(self) -> Optional[Any]:
+        """
+        Resolve the policy to an atomic value. The semantics
+        of the value is policy-dependent.
+
+        :returns:
+            The value of the policy, if any.
+        """
+        pass
+
+    @abstractmethod
+    def _to_dict(self) -> Dict[str, Any]:
+        pass
+
+    @staticmethod
+    @abstractmethod
+    def _from_dict(dict: Dict[str, Any]) -> "AuthPolicy":
+        pass
+
+
+@dataclass
+class AuthPolicyToken(AuthPolicy):
+    """
+    An authentication policy that uses a string token/API key.
+    This policy cannot be serialized.
+    """
+
+    _token: str
+
+    def __init__(self, token: str):
+        """
+        Create a token authentication policy.
+
+        :param token:
+            The token to use for authentication.
+        """
+        super().__init__(AuthPolicyType.TOKEN)
+        self._token = token
+
+        if len(token) == 0:
+            raise ValueError("Authentication token cannot be empty.")
+
+    def _to_dict(self) -> Dict[str, Any]:
+        raise ValueError(
+            "Cannot serialize token authentication policy. Use an alternative policy like environment variables."
+        )
+
+    @staticmethod
+    def _from_dict(dict: Dict[str, Any]) -> "AuthPolicy":
+        raise ValueError(
+            "Cannot deserialize token authentication policy. Use an alternative policy like environment variables."
+        )
+
+    def resolve_value(self) -> Optional[Any]:
+        return self._token
+
+
+@dataclass
+class AuthPolicyEnvVar(AuthPolicy):
+    """
+    An authentication policy that accepts one or more
+    environment variables and fetches a string token from
+    the first one that is set.
+    """
+
+    _env_vars: List[str]
+    _strict: bool
+
+    def __init__(self, env_vars: List[str], *, strict: bool = True):
+        """
+        Create an environment variable authentication policy.
+
+        :param env_vars:
+            Ordered list of candidate environment variables.
+        :param strict:
+            Whether to raise an exception if none of the environment
+            variables are set.
+        """
+        super().__init__(AuthPolicyType.ENV_VAR)
+        self._env_vars = list(env_vars)
+        self._strict = strict
+
+        if len(env_vars) == 0:
+            raise ValueError("One or more environment variables must be provided for the authentication policy.")
+
+    def _to_dict(self) -> Dict[str, Any]:
+        return {"env_vars": self._env_vars, "strict": self._strict}
+
+    @staticmethod
+    def _from_dict(dict: Dict[str, Any]) -> "AuthPolicy":
+        return AuthPolicyEnvVar(dict["env_vars"], strict=dict["strict"])
+
+    def resolve_value(self) -> Optional[Any]:
+        out = None
+        for env_var in self._env_vars:
+            value = os.getenv(env_var)
+            if value is not None:
+                out = value
+                break
+        if out is None and self._strict:
+            raise ValueError(f"None of the following authentication environment variables are set: {self._env_vars}")
+        return out
diff --git a/releasenotes/notes/auth-policy-for-components-d576a28135a224db.yaml b/releasenotes/notes/auth-policy-for-components-d576a28135a224db.yaml
new file mode 100644
index 0000000000..9feeca2497
--- /dev/null
+++ b/releasenotes/notes/auth-policy-for-components-d576a28135a224db.yaml
@@ -0,0 +1,6 @@
+---
+features:
+  - |
+    Expose an `AuthPolicy` type to provide consistent API for any component that requires authentication.
+    Currently supports string tokens and environment variables. Token-based policies are automatically
+    prevented from being serialized to disk (to prevent accidental leakage of secrets).
diff --git a/test/utils/test_auth.py b/test/utils/test_auth.py
new file mode 100644
index 0000000000..0c69860b8f
--- /dev/null
+++ b/test/utils/test_auth.py
@@ -0,0 +1,64 @@
+import os
+
+import pytest
+
+from haystack.utils.auth import AuthPolicy, AuthPolicyEnvVar, AuthPolicyType, AuthPolicyToken
+
+
+def test_auth_policy_type():
+    for e in AuthPolicyType:
+        assert e == AuthPolicyType.from_str(e.value)
+
+    with pytest.raises(ValueError, match="Unknown authentication policy type"):
+        AuthPolicyType.from_str("disk")
+
+
+def test_auth_policy_token():
+    policy = AuthPolicy.from_token("test-token")
+    assert policy._type == AuthPolicyType.TOKEN
+    assert isinstance(policy, AuthPolicyToken)
+    assert policy._token == "test-token"
+    assert policy.resolve_value() == "test-token"
+
+    with pytest.raises(ValueError, match="Cannot serialize token authentication policy"):
+        policy.to_dict()
+
+    with pytest.raises(ValueError, match="cannot be empty"):
+        AuthPolicy.from_token("")
+
+
+def test_auth_policy_env_var():
+    policy = AuthPolicy.from_env_var("TEST_ENV_VAR1")
+    os.environ["TEST_ENV_VAR1"] = "test-token"
+
+    assert policy._type == AuthPolicyType.ENV_VAR
+    assert isinstance(policy, AuthPolicyEnvVar)
+    assert policy._env_vars == ["TEST_ENV_VAR1"]
+    assert policy._strict is True
+    assert policy.resolve_value() == "test-token"
+
+    del os.environ["TEST_ENV_VAR1"]
+    with pytest.raises(ValueError, match="None of the following .* variables are set"):
+        policy.resolve_value()
+
+    policy = AuthPolicy.from_env_var("TEST_ENV_VAR2", strict=False)
+    assert policy._strict is False
+    assert policy.resolve_value() == None
+
+    policy = AuthPolicy.from_env_var(["TEST_ENV_VAR2", "TEST_ENV_VAR1"], strict=True)
+    assert policy._env_vars == ["TEST_ENV_VAR2", "TEST_ENV_VAR1"]
+    with pytest.raises(ValueError, match="None of the following .* variables are set"):
+        policy.resolve_value()
+    os.environ["TEST_ENV_VAR1"] = "test-token-2"
+    assert policy.resolve_value() == "test-token-2"
+    os.environ["TEST_ENV_VAR2"] = "test-token"
+    assert policy.resolve_value() == "test-token"
+
+    with pytest.raises(ValueError, match="One or more environment variables"):
+        AuthPolicy.from_env_var([])
+
+    assert policy.to_dict() == {"type": "env_var", "env_vars": ["TEST_ENV_VAR2", "TEST_ENV_VAR1"], "strict": True}
+    assert (
+        AuthPolicy.from_dict({"type": "env_var", "env_vars": ["TEST_ENV_VAR2", "TEST_ENV_VAR1"], "strict": True})
+        == policy
+    )


Code review comments:
diff_hunk: @@ -0,0 +1,194 @@
+from enum import Enum
+import os
+from typing import Any, Dict, List, Optional, Union
+from dataclasses import dataclass
+from abc import ABC, abstractmethod
+
+
+class AuthPolicyType(Enum):
+    TOKEN = "token"
+    ENV_VAR = "env_var"
+
+    def __str__(self):
+        return self.value
+
+    @staticmethod
+    def from_str(string: str) -> "AuthPolicyType":
+        map = {e.value: e for e in AuthPolicyType}
+        type = map.get(string)
+        if type is None:
+            raise ValueError(f"Unknown authentication policy type '{string}'")
+        return type
+
+
+@dataclass
+class AuthPolicy(ABC):
path: haystack/utils/auth.py
text: @author:
We could use a better name as "policy" might sound more restrictive that it needs to be (as this can be used to wrap usernames and passwords). Also, there is no logic on how the secret is used, so "policy" would also be a bit misleading.

Maybe `Secret`?

@author:
I've gone with `Secret`. If we prefer `AuthPolicy`, I can revert.