feat(cli): add supports subcommand and xtest env auth fallbacks

packages/otdf-python/src/otdf_python/cli.py

@@ -9,6 +9,7 @@
 import contextlib
 import json
 import logging
+import os
 import sys
 from dataclasses import asdict
 from importlib import metadata
@@ -113,7 +114,14 @@ def load_client_credentials(creds_file_path: str) -> tuple[str, str]:
 
 
 def _configure_auth(builder: SDKBuilder, args) -> None:
-    """Configure authentication on the SDK builder."""
+    """Configure authentication on the SDK builder.
+
+    Preference order:
+    1. Explicit --client-id / --client-secret flags
+    2. --with-client-creds-file
+    3. --auth clientId:clientSecret
+    4. CLIENTID / CLIENTSECRET environment variables (xtest / local platform)
+    """
     if args.client_id and args.client_secret:
         builder.client_secret(args.client_id, args.client_secret)
     elif hasattr(args, "with_client_creds_file") and args.with_client_creds_file:
@@ -127,10 +135,14 @@ def _configure_auth(builder: SDKBuilder, args) -> None:
                 f"Auth expects <clientId>:<clientSecret>, received {args.auth}",
             )
         builder.client_secret(auth_parts[0], auth_parts[1])
+    elif os.environ.get("CLIENTID") and os.environ.get("CLIENTSECRET"):
+        builder.client_secret(os.environ["CLIENTID"], os.environ["CLIENTSECRET"])
+        logger.debug("Using CLIENTID/CLIENTSECRET from environment")
     else:
         raise CLIError(
             "CRITICAL",
-            "Authentication required: provide --with-client-creds-file OR --client-id and --client-secret",
+            "Authentication required: provide --with-client-creds-file, "
+            "--client-id/--client-secret, or set CLIENTID and CLIENTSECRET env vars",
         )
 
 
@@ -149,36 +161,103 @@ def _configure_kas_allowlist(builder: SDKBuilder, args) -> None:
 
 
 def build_sdk(args) -> SDK:
-    """Build SDK instance from CLI arguments."""
+    """Build SDK instance from CLI arguments.
+
+    Falls back to PLATFORMURL / KCFULLURL / KASURL environment variables
+    (xtest contract) when flags are omitted, so shims need not pass secrets
+    on argv.
+    """
     builder = SDKBuilder()
 
-    if args.platform_url:
-        builder.set_platform_endpoint(args.platform_url)
+    platform_url = getattr(args, "platform_url", None) or os.environ.get("PLATFORMURL")
+    oidc_endpoint = getattr(args, "oidc_endpoint", None) or os.environ.get("KCFULLURL")
+    # kas-endpoint may be comma-separated; also accept single KASURL env.
+    if not getattr(args, "kas_endpoint", None) and os.environ.get("KASURL"):
+        args.kas_endpoint = os.environ["KASURL"]
+
+    if platform_url:
+        builder.set_platform_endpoint(platform_url)
         # Auto-detect HTTP URLs and enable plaintext mode
-        if args.platform_url.startswith("http://") and (
-            not hasattr(args, "plaintext") or not args.plaintext
-        ):
+        if platform_url.startswith("http://") and not getattr(args, "plaintext", None):
             logger.debug(
-                f"Auto-detected HTTP URL {args.platform_url}, enabling plaintext mode"
+                f"Auto-detected HTTP URL {platform_url}, enabling plaintext mode"
             )
             builder.use_insecure_plaintext_connection(True)
+        # Keep args.platform_url set for create_tdf_config / nano KAS derivation
+        args.platform_url = platform_url
 
-    if args.oidc_endpoint:
-        builder.set_issuer_endpoint(args.oidc_endpoint)
+    if oidc_endpoint:
+        builder.set_issuer_endpoint(oidc_endpoint)
+        args.oidc_endpoint = oidc_endpoint
 
     _configure_auth(builder, args)
 
-    if hasattr(args, "plaintext") and args.plaintext:
+    if getattr(args, "plaintext", None):
         builder.use_insecure_plaintext_connection(True)
 
-    if args.insecure:
+    if getattr(args, "insecure", None):
         builder.use_insecure_skip_verify(True)
 
     _configure_kas_allowlist(builder, args)
 
     return builder.build()
 
 
+# Features the Python SDK currently exercises honestly in community xtest Stage-1.
+# Keep conservative: only advertise what encrypt/decrypt paths actually honor.
+_SUPPORTED_FEATURES: frozenset[str] = frozenset(
+    {
+        "autoconfigure",
+        "connectrpc",
+        "hexless",
+        "kasallowlist",
+    }
+)
+
+_KNOWN_FEATURES: frozenset[str] = frozenset(
+    {
+        "assertions",
+        "assertion_verification",
+        "attribute_traversal",
+        "audit_logging",
+        "autoconfigure",
+        "better-messages-2024",
+        "bulk_rewrap",
+        "connectrpc",
+        "dpop",
+        "dpop_nonce_challenge",
+        "ecwrap",
+        "hexless",
+        "hexaflexible",
+        "kasallowlist",
+        "key_management",
+        "mechanism-rsa-4096",
+        "mechanism-ec-curves-384-521",
+        "mechanism-xwing",
+        "mechanism-secpmlkem",
+        "mechanism-mlkem",
+        "ns_grants",
+        "obligations",
+    }
+)
+
+
+def cmd_supports(args) -> int:
+    """Check if a feature is supported (xtest CLI contract).
+
+    Returns:
+        0 if supported, 1 if not supported, 2 if unknown.
+
+    """
+    feature = getattr(args, "feature", None)
+    if feature not in _KNOWN_FEATURES:
+        logger.error(f"Unknown feature: {feature}")
+        return 2
+    if feature in _SUPPORTED_FEATURES:
+        return 0
+    return 1
+
+
 def create_tdf_config(sdk: SDK, args) -> TDFConfig:
     """Create TDF configuration from CLI arguments."""
     attributes = (
@@ -190,10 +269,13 @@ def create_tdf_config(sdk: SDK, args) -> TDFConfig:
     config = sdk.new_tdf_config(attributes=attributes)
 
     if hasattr(args, "kas_endpoint") and args.kas_endpoint:
-        # Add KAS endpoints
+        # Explicit endpoints replace the platform-derived default; extending
+        # would give the same KAS two key access objects in the manifest.
         kas_endpoints = parse_kas_endpoints(args.kas_endpoint)
-        kas_info_list = [KASInfo(url=kas_url) for kas_url in kas_endpoints]
-        config.kas_info_list.extend(kas_info_list)
+        config.kas_info_list = [
+            KASInfo(url=kas_url, default=(i == 0))
+            for i, kas_url in enumerate(kas_endpoints)
+        ]
 
     if hasattr(args, "mime_type") and args.mime_type:
         config.mime_type = args.mime_type
@@ -550,9 +632,31 @@ def create_parser() -> argparse.ArgumentParser:
     )
     inspect_parser.add_argument("file", help="Path to encrypted file")
 
+    # Supports command (xtest feature probe)
+    supports_parser = subparsers.add_parser(
+        "supports",
+        help="Check if a feature is supported (exit 0=yes, 1=no, 2=unknown)",
+    )
+    supports_parser.add_argument("feature", help="Feature name (e.g. ecwrap, hexless)")
+
     return parser
 
 
+def _dispatch_command(parser, args) -> None:
+    """Route a parsed command to its handler."""
+    if args.command == "encrypt":
+        cmd_encrypt(args)
+    elif args.command == "decrypt":
+        cmd_decrypt(args)
+    elif args.command == "inspect":
+        cmd_inspect(args)
+    elif args.command == "supports":
+        sys.exit(cmd_supports(args))
+    else:
+        parser.print_help()
+        sys.exit(1)
+
+
 def main():
     """Execute the CLI entry point."""
     parser = create_parser()
@@ -567,16 +671,7 @@ def main():
         sys.exit(1)
 
     try:
-        if args.command == "encrypt":
-            cmd_encrypt(args)
-        elif args.command == "decrypt":
-            cmd_decrypt(args)
-        elif args.command == "inspect":
-            cmd_inspect(args)
-        else:
-            parser.print_help()
-            sys.exit(1)
-
+        _dispatch_command(parser, args)
     except CLIError as e:
         logger.error(f"{e.level}: {e.message}")
         if e.cause:

packages/otdf-python/src/otdf_python/kas_client.py

@@ -703,13 +703,17 @@ def _unwrap_with_connect_rpc(
             session_key_type: Optional session key type (RSA_KEY_TYPE or EC_KEY_TYPE)
 
         """
-        # Get access token for authentication if token source is available
+        # Rewrap always requires authentication — fail fast rather than send
+        # an unauthenticated request KAS will reject with a misleading
+        # "missing authorization header".
         access_token = None
         if self.token_source:
             try:
                 access_token = self.token_source()
             except Exception as e:
-                logging.warning(f"Failed to get access token: {e}")
+                raise SDKException(
+                    f"Failed to get access token for KAS rewrap: {e}"
+                ) from e
 
         # Normalize the URL
         normalized_kas_url = self._normalize_kas_url(key_access.url)

packages/otdf-python/src/otdf_python/sdk_builder.py

@@ -290,6 +290,50 @@ def _discover_token_endpoint_from_platform(self) -> None:
 
         self._discover_token_endpoint_from_issuer(platform_issuer)
 
+    def _candidate_issuer_urls(self) -> list[str]:
+        """Issuer URLs to try for OIDC discovery, in order.
+
+        An endpoint that already names a realm (contains ``/realms/``) is
+        used verbatim — appending another realm segment would double the
+        path (…/realms/opentdf/realms/opentdf) and 404. A bare Keycloak
+        base URL gets the default realm appended, with the bare URL as a
+        fallback for non-Keycloak issuers.
+        """
+        base = (self.issuer_endpoint or "").rstrip("/")
+        if not base:
+            return []
+        if "/realms/" in base:
+            return [base]
+        return [f"{base}/realms/opentdf", base]
+
+    def _discover_token_endpoint_from_issuer_endpoint(self) -> None:
+        """Discover the token endpoint from the configured issuer endpoint.
+
+        Raises:
+            AutoConfigureException: If discovery fails for every candidate
+
+        """
+
+        def try_issuer(issuer_url: str) -> Exception | None:
+            try:
+                self._discover_token_endpoint_from_issuer(issuer_url)
+            except Exception as e:
+                return e
+            return None
+
+        last_error: Exception | None = None
+        for issuer_url in self._candidate_issuer_urls():
+            last_error = try_issuer(issuer_url)
+            if last_error is None:
+                return
+        if last_error is not None:
+            raise AutoConfigureException(
+                f"Error during token endpoint discovery: {last_error!s}"
+            ) from last_error
+        raise AutoConfigureException(
+            "Issuer endpoint must be configured for OIDC token discovery"
+        )
+
     def _discover_token_endpoint_from_issuer(self, issuer_url: str) -> None:
         """Discover token endpoint using OIDC discovery from issuer.
 
@@ -335,9 +379,7 @@ def _discover_token_endpoint(self) -> None:
                 # If platform fails and we have an explicit issuer, try that
                 if self.issuer_endpoint:
                     try:
-                        realm_name = "opentdf"  # Default realm name
-                        issuer_url = f"{self.issuer_endpoint}/realms/{realm_name}"
-                        self._discover_token_endpoint_from_issuer(issuer_url)
+                        self._discover_token_endpoint_from_issuer_endpoint()
                         return
                     except Exception:
                         # Re-raise the original platform error
@@ -348,9 +390,7 @@ def _discover_token_endpoint(self) -> None:
 
         # Fall back to explicit issuer endpoint
         if self.issuer_endpoint:
-            realm_name = "opentdf"  # Default realm name
-            issuer_url = f"{self.issuer_endpoint}/realms/{realm_name}"
-            self._discover_token_endpoint_from_issuer(issuer_url)
+            self._discover_token_endpoint_from_issuer_endpoint()
             return
 
         raise AutoConfigureException(

packages/otdf-python/src/otdf_python/tdf.py

@@ -77,6 +77,19 @@ def _validate_kas_infos(self, kas_infos):
         if not isinstance(kas_infos, list):
             kas_infos = [kas_infos]
 
+        # One key access object per KAS: drop entries that only repeat an
+        # earlier URL (e.g. a platform-derived default plus the same KAS
+        # passed explicitly).
+        seen_urls = set()
+        unique_kas_infos = []
+        for kas in kas_infos:
+            url_key = (getattr(kas, "url", "") or "").rstrip("/")
+            if url_key in seen_urls:
+                continue
+            seen_urls.add(url_key)
+            unique_kas_infos.append(kas)
+        kas_infos = unique_kas_infos
+
         validated_kas_infos = []
         for kas in kas_infos:
             # If public key is missing, try to fetch it from the KAS service

tests/test_cli_supports.py

@@ -0,0 +1,44 @@
+"""Unit tests for the CLI `supports` subcommand (community xtest contract)."""
+
+import subprocess
+import sys
+from types import SimpleNamespace
+
+from otdf_python.cli import cmd_supports
+
+
+def test_cmd_supports_returns_codes_in_process():
+    """cmd_supports returns exit codes without calling sys.exit (review feedback)."""
+    assert cmd_supports(SimpleNamespace(feature="autoconfigure")) == 0
+    assert cmd_supports(SimpleNamespace(feature="ecwrap")) == 1
+    assert cmd_supports(SimpleNamespace(feature="not-a-real-feature")) == 2
+
+
+def test_supports_autoconfigure_exit_0():
+    r = subprocess.run(
+        [sys.executable, "-m", "otdf_python", "supports", "autoconfigure"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert r.returncode == 0, r.stderr
+
+
+def test_supports_ecwrap_exit_1():
+    r = subprocess.run(
+        [sys.executable, "-m", "otdf_python", "supports", "ecwrap"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert r.returncode == 1, r.stderr
+
+
+def test_supports_unknown_exit_2():
+    r = subprocess.run(
+        [sys.executable, "-m", "otdf_python", "supports", "not-a-real-feature"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert r.returncode == 2