allow only SPLUNK_AO_SF_API_TOKEN for CRUD use

Author: pradystar

src/splunk_ao/__init__.py

@@ -19,7 +19,7 @@
 from splunk_ao.collaborator import Collaborator, CollaboratorRole
 from splunk_ao.configuration import Configuration
 from splunk_ao.dataset import Dataset
-from splunk_ao.decorator import SplunkAODecorator, splunk_ao_context, log, start_session
+from splunk_ao.decorator import SplunkAODecorator, log, splunk_ao_context, start_session
 from splunk_ao.exceptions import (
     AuthenticationError,
     BadRequestError,
@@ -132,12 +132,12 @@
     "create_api_key",
     "delete_api_key",
     "enable_console_logging",
-    "splunk_ao_context",
     "get_agent_control_target",
     "get_tracing_headers",
     "is_dependency_available",
     "list_api_keys",
     "log",
     "setup_agent_control_bridge",
+    "splunk_ao_context",
     "start_session",
 ]

src/splunk_ao/config.py

@@ -44,6 +44,7 @@ async def arequest(self, method: RequestMethod, path: str, *args: Any, **kwargs:
     def stream_request(self, method: RequestMethod, path: str, *args: Any, **kwargs: Any) -> Iterator[Response]:
         return super().stream_request(method, self._prefixed(path), *args, **kwargs)
 
+
 # Mapping of SPLUNK_AO_* → GALILEO_* env var pairs used by the bridge.
 # Defined at module level so both _bridge_env_vars() and reset() can reference
 # the same authoritative list without duplication.
@@ -175,7 +176,8 @@ def _check_auth_config(kwargs: dict) -> str | None:
         message identifying what's missing.
 
         Auth methods supported by the underlying config model:
-          - SF tokens (o11y): SPLUNK_AO_SF_TOKEN and optional SPLUNK_AO_SF_API_TOKEN env vars
+          - SF tokens (o11y): SPLUNK_AO_REALM and at least one of
+            SPLUNK_AO_SF_TOKEN or SPLUNK_AO_SF_API_TOKEN env vars
           - API key (standalone): api_key kwarg or SPLUNK_AO_API_KEY env
           - Pre-exchanged JWT (standalone): jwt_token or SPLUNK_AO_JWT_TOKEN
           - SSO (paired): sso_id_token + sso_provider, both kwargs and env vars
@@ -192,7 +194,14 @@ def _val(kwarg_name: str, env_name: str) -> str | None:
                 return str(value)
             return os.environ.get(env_name)
 
-        if os.environ.get("SPLUNK_AO_SF_TOKEN") or os.environ.get("SPLUNK_AO_SF_API_TOKEN"):
+        realm = os.environ.get("SPLUNK_AO_REALM")
+        sf_token = os.environ.get("SPLUNK_AO_SF_TOKEN")
+        sf_api_token = os.environ.get("SPLUNK_AO_SF_API_TOKEN")
+        if realm or sf_token or sf_api_token:
+            if not realm:
+                return "O11y authentication requires SPLUNK_AO_REALM to be set."
+            if not sf_token and not sf_api_token:
+                return "O11y authentication requires SPLUNK_AO_SF_TOKEN or SPLUNK_AO_SF_API_TOKEN to be set."
             return None
 
         # Standalone methods — either alone is sufficient.
@@ -239,7 +248,8 @@ def _val(kwarg_name: str, env_name: str) -> str | None:
         # Nothing configured anywhere.
         return (
             "No Splunk AO authentication detected. Set one of: SPLUNK_AO_REALM with "
-            "SPLUNK_AO_SF_TOKEN; SPLUNK_AO_API_KEY; SPLUNK_AO_SSO_ID_TOKEN with SPLUNK_AO_SSO_PROVIDER; "
+            "SPLUNK_AO_SF_TOKEN or SPLUNK_AO_SF_API_TOKEN; SPLUNK_AO_API_KEY; "
+            "SPLUNK_AO_SSO_ID_TOKEN with SPLUNK_AO_SSO_PROVIDER; "
             "or SPLUNK_AO_USERNAME with SPLUNK_AO_PASSWORD. "
             "Alternatively, pass the equivalent kwargs to SplunkAOConfig.get(). "
             "See https://docs.splunk.com for setup instructions."

src/splunk_ao/deployment.py

@@ -51,11 +51,19 @@ class O11yConfig:
     """Configuration for a Splunk Observability Cloud deployment."""
 
     realm: str
-    sf_token: SecretStr
+    sf_token: SecretStr | None = None
     sf_api_token: SecretStr | None = None
 
     def __post_init__(self) -> None:
-        if not isinstance(self.sf_token, SecretStr):
+        missing = []
+        if not self.realm:
+            missing.append("SPLUNK_AO_REALM")
+        if self.sf_token is None and self.sf_api_token is None:
+            missing.append("one of SPLUNK_AO_SF_TOKEN or SPLUNK_AO_SF_API_TOKEN")
+        if missing:
+            raise MissingConfigurationError(f"O11y deployment requires {' and '.join(missing)} to be set.")
+
+        if self.sf_token is not None and not isinstance(self.sf_token, SecretStr):
             self.sf_token = SecretStr(self.sf_token)
         if self.sf_api_token is not None and not isinstance(self.sf_api_token, SecretStr):
             self.sf_api_token = SecretStr(self.sf_api_token)
@@ -65,16 +73,11 @@ def from_env(cls) -> "O11yConfig":
         """Load and validate o11y configuration from the environment."""
         realm = _env("SPLUNK_AO_REALM")
         sf_token = _env("SPLUNK_AO_SF_TOKEN")
-
-        if realm is None or sf_token is None:
-            missing = [
-                name for name, value in (("SPLUNK_AO_REALM", realm), ("SPLUNK_AO_SF_TOKEN", sf_token)) if value is None
-            ]
-            raise MissingConfigurationError(f"O11y deployment requires {' and '.join(missing)} to be set.")
-
         sf_api_token = _env("SPLUNK_AO_SF_API_TOKEN")
         return cls(
-            realm=realm, sf_token=SecretStr(sf_token), sf_api_token=SecretStr(sf_api_token) if sf_api_token else None
+            realm=realm or "",
+            sf_token=SecretStr(sf_token) if sf_token else None,
+            sf_api_token=SecretStr(sf_api_token) if sf_api_token else None,
         )
 
     @property
@@ -85,7 +88,20 @@ def otlp_endpoint(self) -> str:
     @property
     def crud_token(self) -> SecretStr:
         """Return the API token when set, otherwise the ingest token."""
-        return self.sf_api_token if self.sf_api_token is not None else self.sf_token
+        if self.sf_api_token is not None:
+            return self.sf_api_token
+        if self.sf_token is not None:
+            return self.sf_token
+        raise MissingConfigurationError("O11y CRUD requires SPLUNK_AO_SF_API_TOKEN or SPLUNK_AO_SF_TOKEN to be set.")
+
+    def require_ingest_token(self) -> SecretStr:
+        """Return the token required for OTLP trace export."""
+        if self.sf_token is None:
+            raise MissingConfigurationError(
+                "O11y OTLP trace export requires SPLUNK_AO_SF_TOKEN. "
+                "SPLUNK_AO_SF_API_TOKEN supports CRUD operations only."
+            )
+        return self.sf_token
 
     @property
     def api_root(self) -> str:

tests/test_deployment.py

@@ -101,6 +101,14 @@ def test_o11y_config_from_env() -> None:
     assert cfg.sf_api_token.get_secret_value() == "api-tok"
 
 
+def test_o11y_config_from_env_accepts_crud_only_api_token() -> None:
+    with env(SPLUNK_AO_REALM="us1", SPLUNK_AO_SF_API_TOKEN="api-tok"):
+        cfg = O11yConfig.from_env()
+
+    assert cfg.sf_token is None
+    assert cfg.crud_token.get_secret_value() == "api-tok"
+
+
 def test_otlp_endpoint_derived_from_realm() -> None:
     cfg = O11yConfig(realm="us1", sf_token="tok")
     assert cfg.otlp_endpoint == "https://ingest.us1.observability.splunkcloud.com/v2/trace/otlp"
@@ -116,16 +124,19 @@ def test_crud_token_falls_back_to_sf_token() -> None:
     assert cfg.crud_token.get_secret_value() == "tok"
 
 
-def test_missing_realm_raises() -> None:
-    with env(SPLUNK_AO_SF_TOKEN="tok"):
+@pytest.mark.parametrize("token_var", ["SPLUNK_AO_SF_TOKEN", "SPLUNK_AO_SF_API_TOKEN"])
+def test_missing_realm_raises(token_var: str) -> None:
+    with env(**{token_var: "tok"}):
         with pytest.raises(MissingConfigurationError, match="SPLUNK_AO_REALM"):
             O11yConfig.from_env()
 
 
-def test_missing_sf_token_raises() -> None:
+def test_missing_both_o11y_tokens_raises() -> None:
     with env(SPLUNK_AO_REALM="us1"):
-        with pytest.raises(MissingConfigurationError, match="SPLUNK_AO_SF_TOKEN"):
+        with pytest.raises(MissingConfigurationError) as exc_info:
             O11yConfig.from_env()
+    assert "SPLUNK_AO_SF_TOKEN" in str(exc_info.value)
+    assert "SPLUNK_AO_SF_API_TOKEN" in str(exc_info.value)
 
 
 def test_missing_o11y_config_names_both_required_variables() -> None:
@@ -134,6 +145,18 @@ def test_missing_o11y_config_names_both_required_variables() -> None:
             O11yConfig.from_env()
     assert "SPLUNK_AO_REALM" in str(exc_info.value)
     assert "SPLUNK_AO_SF_TOKEN" in str(exc_info.value)
+    assert "SPLUNK_AO_SF_API_TOKEN" in str(exc_info.value)
+
+
+def test_require_ingest_token_returns_sf_token() -> None:
+    cfg = O11yConfig(realm="us1", sf_token="ingest-tok", sf_api_token="api-tok")
+    assert cfg.require_ingest_token().get_secret_value() == "ingest-tok"
+
+
+def test_require_ingest_token_rejects_crud_only_config() -> None:
+    cfg = O11yConfig(realm="us1", sf_api_token="api-tok")
+    with pytest.raises(MissingConfigurationError, match="SPLUNK_AO_SF_TOKEN"):
+        cfg.require_ingest_token()
 
 
 def test_api_root_derives_from_realm() -> None:

tests/test_o11y_config.py

@@ -14,7 +14,7 @@
 from galileo_core.helpers.api_client import ApiClient
 from galileo_core.schemas.base_config import GalileoConfig
 from splunk_ao.config import O11yApiClient, SplunkAOConfig
-from splunk_ao.shared.exceptions import AmbiguousConfigurationError
+from splunk_ao.shared.exceptions import AmbiguousConfigurationError, MissingConfigurationError
 
 _CONFIG_ENV_VARS = (
     "SPLUNK_AO_REALM",
@@ -133,10 +133,29 @@ def fake_stream_request(
 
 @pytest.mark.parametrize("token_var", ["SPLUNK_AO_SF_TOKEN", "SPLUNK_AO_SF_API_TOKEN"])
 def test_o11y_auth_guard_accepts_environment_tokens(token_var: str) -> None:
-    with config_env(**{token_var: "tok"}):
+    with config_env(SPLUNK_AO_REALM="us1", **{token_var: "tok"}):
         assert SplunkAOConfig._check_auth_config({}) is None
 
 
+def test_o11y_auth_guard_requires_realm() -> None:
+    with config_env(SPLUNK_AO_SF_API_TOKEN="tok"):
+        assert "SPLUNK_AO_REALM" in (SplunkAOConfig._check_auth_config({}) or "")
+
+
+def test_o11y_get_with_api_token_but_no_realm_fails_clearly(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(SplunkAOConfig, "_instance", None)
+    with config_env(SPLUNK_AO_SF_API_TOKEN="tok"):
+        with pytest.raises(MissingConfigurationError, match="SPLUNK_AO_REALM"):
+            SplunkAOConfig.get()
+
+
+def test_o11y_auth_guard_requires_at_least_one_token() -> None:
+    with config_env(SPLUNK_AO_REALM="us1"):
+        error = SplunkAOConfig._check_auth_config({}) or ""
+    assert "SPLUNK_AO_SF_TOKEN" in error
+    assert "SPLUNK_AO_SF_API_TOKEN" in error
+
+
 def test_o11y_auth_guard_does_not_accept_token_kwargs() -> None:
     with config_env():
         assert SplunkAOConfig._check_auth_config({"sf_token": "tok"}) is not None
@@ -204,6 +223,27 @@ async def async_fail(*args: object, **kwargs: object) -> None:
         assert client.ssl_context is False
 
 
+def test_o11y_get_supports_crud_only_api_token(monkeypatch: pytest.MonkeyPatch) -> None:
+    def fail(*args: object, **kwargs: object) -> None:
+        raise AssertionError("o11y configuration must not use standalone validation")
+
+    async def async_fail(*args: object, **kwargs: object) -> None:
+        fail()
+
+    monkeypatch.setattr(SplunkAOConfig, "_instance", None)
+    monkeypatch.setattr(GalileoConfig, "get_jwt_token", staticmethod(fail))
+    monkeypatch.setattr(ApiClient, "make_request", staticmethod(async_fail))
+    monkeypatch.setattr(ApiClient, "request", fail)
+
+    with config_env(SPLUNK_AO_REALM="us1", SPLUNK_AO_SF_API_TOKEN="api-token"):
+        cfg = SplunkAOConfig.get(ssl_context=False)
+        client = cfg.api_client
+
+    assert cfg.jwt_token is None
+    assert isinstance(client, O11yApiClient)
+    assert client.auth_header == {"X-SF-Token": "api-token"}
+
+
 def test_ambiguous_environment_fails_before_config_construction(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(SplunkAOConfig, "_instance", None)
     with config_env(SPLUNK_AO_REALM="us1", SPLUNK_AO_SF_TOKEN="tok", SPLUNK_AO_API_KEY="key"):