Add AWS Process Credential Resolver

packages/smithy-aws-core/.changes/next-release/smithy-aws-core-feature-9e2d74d0c5724eacbee1b1af6260ab54.json

@@ -0,0 +1,4 @@
+{
+  "type": "feature",
+  "description": "Added process credentials support to the default AWS identity chain through the active profile's `credential_process` setting."
+}

packages/smithy-aws-core/pyproject.toml

@@ -42,6 +42,7 @@ Environment = "smithy_aws_core.identity.chain.providers.environment:EnvironmentC
 SharedConfig = "smithy_aws_core.identity.chain.providers.shared_config:SharedConfigProvider"
 ProfileSessionKeys = "smithy_aws_core.identity.chain.providers.profile:ProfileSessionCredentialsProvider"
 ProfileStaticKeys = "smithy_aws_core.identity.chain.providers.profile:ProfileStaticCredentialsProvider"
+ProfileCredentialProcess = "smithy_aws_core.identity.chain.providers.process:ProfileCredentialProcessProvider"
 
 [build-system]
 requires = ["hatchling"]

packages/smithy-aws-core/src/smithy_aws_core/identity/__init__.py

@@ -4,6 +4,7 @@
 
 from .chain import IdentityChain, IdentityChainError, UnclaimedSource
 from .chain.providers.environment import EnvironmentCredentialsProvider
+from .chain.providers.process import ProfileCredentialProcessProvider
 from .chain.providers.profile import (
     ProfileSessionCredentialsProvider,
     ProfileStaticCredentialsProvider,
@@ -18,6 +19,7 @@
 from .container import ContainerCredentialsResolver
 from .environment import EnvironmentCredentialsResolver
 from .imds import IMDSCredentialsResolver
+from .process import ProcessCredentialsResolver
 from .static import StaticCredentialsResolver
 
 __all__ = (
@@ -30,6 +32,8 @@
     "IMDSCredentialsResolver",
     "IdentityChain",
     "IdentityChainError",
+    "ProcessCredentialsResolver",
+    "ProfileCredentialProcessProvider",
     "ProfileSessionCredentialsProvider",
     "ProfileStaticCredentialsProvider",
     "SharedConfigProvider",

packages/smithy-aws-core/src/smithy_aws_core/identity/chain/providers/process.py

@@ -0,0 +1,133 @@
+#  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#  SPDX-License-Identifier: Apache-2.0
+import shlex
+import sys
+
+from smithy_core.exceptions import SmithyError
+from smithy_core.interfaces.identity import Identity
+
+from ...components import AWSCredentialsIdentity
+from ...process import ProcessCredentialsResolver
+from ..ordering import Standard, StandardProvider
+from ..provider import ChainSetup
+
+_CREDENTIAL_PROCESS = "credential_process"
+_ACCOUNT_ID = "aws_account_id"
+
+
+class ProcessConfigurationError(SmithyError):
+    """Raised when a profile's credential process command is misconfigured."""
+
+
+def _split_process_command(
+    command: str,
+    *,
+    platform: str | None = None,
+) -> list[str]:
+    """Split a process command according to the host platform's quoting rules."""
+    if platform is None:
+        platform = sys.platform
+    if platform == "win32":
+        return _split_windows_command(command)
+    try:
+        return shlex.split(command)
+    except ValueError as e:
+        raise ProcessConfigurationError(
+            f"Could not parse credential process command: {e}"
+        ) from e
+
+
+def _split_windows_command(command: str) -> list[str]:
+    """Split a command using botocore's strict form of the Microsoft C runtime rules.
+
+    The underlying runtime rules are documented at:
+    https://learn.microsoft.com/en-us/cpp/cpp/main-function-command-line-args#parsing-c-command-line-arguments
+    """
+    arguments: list[str] = []
+    argument: list[str] = []
+    argument_started = False
+    in_quotes = False
+    backslashes = 0
+
+    for character in command:
+        if character == "\\":
+            # Delay emitting backslashes until we know whether a quote follows.
+            backslashes += 1
+            argument_started = True
+            continue
+
+        if character == '"':
+            # Pairs become literal backslashes; an odd remainder escapes the quote.
+            literal_backslashes, escaped_quote = divmod(backslashes, 2)
+            argument.extend("\\" * literal_backslashes)
+            backslashes = 0
+            argument_started = True
+            if escaped_quote:
+                argument.append('"')
+            else:
+                in_quotes = not in_quotes
+            continue
+
+        if backslashes:
+            # Without a following quote, backslashes are literal.
+            argument.extend("\\" * backslashes)
+            backslashes = 0
+
+        # Only spaces and tabs outside quotes delimit Windows arguments.
+        if character in (" ", "\t") and not in_quotes:
+            # This preserves empty quoted arguments while ignoring extra whitespace.
+            if argument_started:
+                arguments.append("".join(argument))
+                argument = []
+                argument_started = False
+            continue
+
+        argument.append(character)
+        argument_started = True
+
+    if in_quotes:
+        raise ProcessConfigurationError(f"No closing quotation in string: {command}")
+
+    if backslashes:
+        argument.extend("\\" * backslashes)
+    if argument_started:
+        arguments.append("".join(argument))
+
+    return arguments
+
+
+class ProfileCredentialProcessProvider:
+    """Adds a process credential resolver configured by the active profile."""
+
+    @property
+    def name(self) -> str:
+        """Return the canonical provider name."""
+        return StandardProvider.PROFILE_CREDENTIAL_PROCESS.canonical_name
+
+    @property
+    def ordering(self) -> Standard:
+        """Return the provider's standard chain position."""
+        return Standard(slot=StandardProvider.PROFILE_CREDENTIAL_PROCESS)
+
+    async def setup(self, identity_type: type[Identity], setup: ChainSetup) -> None:
+        """Add a resolver when the active profile configures a credential process."""
+        if identity_type is not AWSCredentialsIdentity:
+            return
+
+        config_file = setup.config_file
+        profile_name = setup.profile_name
+        if config_file is None or profile_name is None:
+            return
+
+        command = config_file.get(profile_name, _CREDENTIAL_PROCESS)
+        if not command:
+            return
+
+        # The process output's AccountId takes precedence; the profile's
+        # aws_account_id is only used as a fallback.
+        setup.add_terminal_resolver(
+            ProcessCredentialsResolver(
+                _split_process_command(command),
+                account_id=config_file.get(profile_name, _ACCOUNT_ID),
+            )
+        )

packages/smithy-aws-core/src/smithy_aws_core/identity/process.py

@@ -0,0 +1,154 @@
+#  Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#  SPDX-License-Identifier: Apache-2.0
+import asyncio
+import json
+from datetime import UTC, datetime
+from typing import TypeGuard, cast
+
+from smithy_core.aio.interfaces.identity import IdentityResolver
+from smithy_core.exceptions import SmithyIdentityError
+
+from .components import AWSCredentialsIdentity, AWSIdentityProperties
+
+
+def _is_command_list(command: object) -> TypeGuard[list[str]]:
+    if not isinstance(command, list) or not command:
+        return False
+    return all(isinstance(argument, str) for argument in cast(list[object], command))
+
+
+class ProcessCredentialsResolver(
+    IdentityResolver[AWSCredentialsIdentity, AWSIdentityProperties]
+):
+    """Resolves AWS Credentials from a process.
+
+    :param command: The process command and arguments to execute, as a
+        non-empty list of strings.
+    :param timeout: Maximum time in seconds to wait for the process to complete.
+    :param account_id: Fallback account ID to associate with the resolved
+        credentials when the process output does not include an ``AccountId``.
+    """
+
+    def __init__(
+        self,
+        command: list[str],
+        *,
+        timeout: float | None = None,
+        account_id: str | None = None,
+    ) -> None:
+        if not _is_command_list(command):
+            raise ValueError("command must be a non-empty list of strings")
+        self._command = list(command)
+        self._timeout = timeout
+        self._account_id = account_id
+        self._credentials: AWSCredentialsIdentity | None = None
+
+    async def get_identity(
+        self, *, properties: AWSIdentityProperties
+    ) -> AWSCredentialsIdentity:
+        if self._credentials is not None:
+            # Long-term credentials (no expiration) should always be reused
+            if self._credentials.expiration is None:
+                return self._credentials
+            # Temporary credentials should be reused if not expired
+            if datetime.now(UTC) < self._credentials.expiration:
+                return self._credentials
+
+        try:
+            process = await asyncio.create_subprocess_exec(
+                *self._command,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+        except OSError as e:
+            raise SmithyIdentityError(f"Credential process failed to start: {e}") from e
+
+        try:
+            stdout, stderr = await asyncio.wait_for(
+                process.communicate(), timeout=self._timeout
+            )
+        except TimeoutError as e:
+            if process.returncode is None:
+                try:
+                    process.kill()
+                except ProcessLookupError:
+                    pass
+                await process.wait()
+            raise SmithyIdentityError(
+                f"Credential process timed out after {self._timeout} seconds"
+            ) from e
+
+        if process.returncode != 0:
+            raise SmithyIdentityError(
+                f"Credential process failed with exit code {process.returncode}: "
+                f"{stderr.decode('utf-8', errors='replace')}"
+            )
+        # These exceptions retain the full process output, which may contain
+        # credentials. Suppress chaining to avoid exposing it in tracebacks.
+        try:
+            decoded = stdout.decode("utf-8")
+            creds = json.loads(decoded)
+        except UnicodeDecodeError as e:
+            raise SmithyIdentityError(
+                "Credential process output is not valid UTF-8 "
+                f"at byte {e.start}: {e.reason}"
+            ) from None
+        except json.JSONDecodeError as e:
+            raise SmithyIdentityError(
+                "Credential process output is not valid JSON "
+                f"at line {e.lineno}, column {e.colno}: {e.msg}"
+            ) from None
+
+        if not isinstance(creds, dict):
+            raise SmithyIdentityError(
+                "Credential process output must be a JSON object, "
+                f"got {type(creds).__name__}"
+            )
+        creds = cast(dict[str, object], creds)
+
+        version = creds.get("Version")
+        if version != 1:
+            raise SmithyIdentityError(
+                f"Unsupported version '{version}' for credential process provider, supported versions: 1"
+            )
+        access_key_id = self._get_string_field(creds, "AccessKeyId")
+        secret_access_key = self._get_string_field(creds, "SecretAccessKey")
+        session_token = self._get_string_field(creds, "SessionToken")
+        expiration = self._get_string_field(creds, "Expiration")
+        # Prefer the process output's AccountId, falling back to the profile's
+        # aws_account_id when the process omits it.
+        account_id = self._get_string_field(creds, "AccountId") or self._account_id
+
+        if expiration is not None:
+            try:
+                dt = datetime.fromisoformat(expiration)
+            except (TypeError, ValueError) as e:
+                raise SmithyIdentityError(
+                    "Invalid credential process Expiration; "
+                    f"expected an ISO 8601 string: {e}"
+                ) from e
+            expiration = dt.astimezone(UTC) if dt.tzinfo else dt.replace(tzinfo=UTC)
+
+        if access_key_id is None or secret_access_key is None:
+            raise SmithyIdentityError(
+                "AccessKeyId and SecretAccessKey are required for process credentials"
+            )
+
+        self._credentials = AWSCredentialsIdentity(
+            access_key_id=access_key_id,
+            secret_access_key=secret_access_key,
+            session_token=session_token,
+            expiration=expiration,
+            account_id=account_id,
+        )
+        return self._credentials
+
+    @staticmethod
+    def _get_string_field(creds: dict[str, object], key: str) -> str | None:
+        value = creds.get(key)
+        if value is not None and not isinstance(value, str):
+            raise SmithyIdentityError(
+                f"Credential process output field '{key}' must be a string, "
+                f"got {type(value).__name__}"
+            )
+        return value