diff --git a/packages/smithy-aws-core/.changes/next-release/smithy-aws-core-feature-9e2d74d0c5724eacbee1b1af6260ab54.json b/packages/smithy-aws-core/.changes/next-release/smithy-aws-core-feature-9e2d74d0c5724eacbee1b1af6260ab54.json new file mode 100644 index 000000000..e5bcf9b7a --- /dev/null +++ b/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." +} diff --git a/packages/smithy-aws-core/pyproject.toml b/packages/smithy-aws-core/pyproject.toml index cb2f732cf..8439bc665 100644 --- a/packages/smithy-aws-core/pyproject.toml +++ b/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:ProfileProcessCredentialsProvider" [build-system] requires = ["hatchling"] diff --git a/packages/smithy-aws-core/src/smithy_aws_core/identity/__init__.py b/packages/smithy-aws-core/src/smithy_aws_core/identity/__init__.py index 7db4add9c..2fe508952 100644 --- a/packages/smithy-aws-core/src/smithy_aws_core/identity/__init__.py +++ b/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 ProfileProcessCredentialsProvider 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", + "ProfileProcessCredentialsProvider", "ProfileSessionCredentialsProvider", "ProfileStaticCredentialsProvider", "SharedConfigProvider", diff --git a/packages/smithy-aws-core/src/smithy_aws_core/identity/chain/providers/process.py b/packages/smithy-aws-core/src/smithy_aws_core/identity/chain/providers/process.py new file mode 100644 index 000000000..0403e0b22 --- /dev/null +++ b/packages/smithy-aws-core/src/smithy_aws_core/identity/chain/providers/process.py @@ -0,0 +1,123 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +import shlex +import sys + +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" + + +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) + return shlex.split(command) + + +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 ValueError(f"No closing quotation in string: {command}") + + if backslashes: + argument.extend("\\" * backslashes) + if argument_started: + arguments.append("".join(argument)) + + return arguments + + +class ProfileProcessCredentialsProvider: + """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), + ) + ) diff --git a/packages/smithy-aws-core/src/smithy_aws_core/identity/process.py b/packages/smithy-aws-core/src/smithy_aws_core/identity/process.py new file mode 100644 index 000000000..0a2ce3d8d --- /dev/null +++ b/packages/smithy-aws-core/src/smithy_aws_core/identity/process.py @@ -0,0 +1,141 @@ +# 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 + + version = creds.get("Version") + if version != 1: + raise SmithyIdentityError( + f"Unsupported version '{version}' for credential process provider, supported versions: 1" + ) + access_key_id = creds.get("AccessKeyId") + secret_access_key = creds.get("SecretAccessKey") + session_token = creds.get("SessionToken") + expiration = creds.get("Expiration") + # Prefer the process output's AccountId, falling back to the profile's + # aws_account_id when the process omits it. + account_id = creds.get("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 + + async def invalidate(self) -> None: + """Discard cached credentials so the next resolution reruns the process.""" + self._credentials = None diff --git a/packages/smithy-aws-core/tests/unit/identity/chain/providers/test_process.py b/packages/smithy-aws-core/tests/unit/identity/chain/providers/test_process.py new file mode 100644 index 000000000..f6a2b873e --- /dev/null +++ b/packages/smithy-aws-core/tests/unit/identity/chain/providers/test_process.py @@ -0,0 +1,217 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +# pyright: reportPrivateUsage=false +import asyncio +import json +import subprocess +from collections.abc import Awaitable, Callable +from unittest.mock import AsyncMock, patch + +import pytest +from smithy_aws_core.config.merged_config import MergedConfig +from smithy_aws_core.identity.chain.provider import ChainSetup +from smithy_aws_core.identity.chain.providers.process import ( + ProfileProcessCredentialsProvider, + _split_process_command, +) +from smithy_aws_core.identity.process import ProcessCredentialsResolver + +from .conftest import OtherIdentity + + +@pytest.mark.parametrize( + ("command", "expected"), + [ + ("", []), + ("spam eggs", ["spam", "eggs"]), + ("spam\teggs", ["spam", "eggs"]), + ("spam\neggs", ["spam\neggs"]), + ('""', [""]), + ('" "', [" "]), + ('"\t"', ["\t"]), + (r"spam \\", ["spam", r"\\"]), + (r"\\", [r"\\"]), + (r"\\ ", [r"\\"]), + (r"\\ ", [r"\\"]), + (r"\"", ['"']), + ( + r"C:\Tools\awscreds.exe --profile dev", + [r"C:\Tools\awscreds.exe", "--profile", "dev"], + ), + ( + r'"C:\Program Files\awscreds.exe" --profile "test profile"', + [r"C:\Program Files\awscreds.exe", "--profile", "test profile"], + ), + (r'"abc" d e', ["abc", "d", "e"]), + (r'a\\b d"e f"g h', [r"a\\b", "de fg", "h"]), + (r"a\\\"b c d", ['a\\"b', "c", "d"]), + (r'a\\\\"b c" d e', [r"a\\b c", "d", "e"]), + ], +) +def test_split_process_command_windows( + command: str, + expected: list[str], +) -> None: + assert _split_process_command(command, platform="win32") == expected + + +@pytest.mark.parametrize( + "arguments", + [ + [r"C:\Tools\awscreds.exe", "--profile", "dev"], + [r"C:\Program Files\awscreds.exe", "--profile", "test profile"], + ["credential-helper", "", "embedded space"], + ["credential-helper", 'embedded"quote', "trailing\\"], + ["credential-helper", r"multiple\\backslashes", r'backslash\\"quote'], + ], +) +def test_split_process_command_windows_round_trips_python_arguments( + arguments: list[str], +) -> None: + command = subprocess.list2cmdline(arguments) + + assert _split_process_command(command, platform="win32") == arguments + + +@pytest.mark.parametrize("platform", ["darwin", "linux"]) +def test_split_process_command_posix(platform: str) -> None: + command = r'/opt/My\ Tools/awscreds --profile "test profile"' + + assert _split_process_command(command, platform=platform) == [ + "/opt/My Tools/awscreds", + "--profile", + "test profile", + ] + + +@pytest.mark.parametrize("platform", ["darwin", "linux", "win32"]) +def test_split_process_command_rejects_unclosed_quote(platform: str) -> None: + with pytest.raises(ValueError, match="No closing quotation"): + _split_process_command('"credential-helper', platform=platform) + + +async def test_ignores_non_aws_identity_type( + setup_provider: Callable[..., Awaitable[ChainSetup]], + merged_config: Callable[..., MergedConfig], +) -> None: + provider = ProfileProcessCredentialsProvider() + + setup = await setup_provider( + provider, + identity_type=OtherIdentity, + config_file=merged_config( + {"default": {"credential_process": "credential-helper"}} + ), + profile_name="default", + ) + + assert setup.resolvers == () + assert not setup.terminal + + +async def test_requires_active_profile( + setup_provider: Callable[..., Awaitable[ChainSetup]], +) -> None: + setup = await setup_provider(ProfileProcessCredentialsProvider()) + + assert setup.resolvers == () + assert not setup.terminal + + +async def test_missing_process_does_not_register( + setup_provider: Callable[..., Awaitable[ChainSetup]], + merged_config: Callable[..., MergedConfig], +) -> None: + setup = await setup_provider( + ProfileProcessCredentialsProvider(), + config_file=merged_config({"default": {}}), + profile_name="default", + ) + + assert setup.resolvers == () + assert not setup.terminal + + +async def test_registers_terminal_resolver( + setup_provider: Callable[..., Awaitable[ChainSetup]], + merged_config: Callable[..., MergedConfig], +) -> None: + setup = await setup_provider( + ProfileProcessCredentialsProvider(), + config_file=merged_config( + { + "default": { + "credential_process": ( + 'credential-helper --profile "test profile" --format json' + ) + } + } + ), + profile_name="default", + ) + + assert setup.terminal + assert len(setup.resolvers) == 1 + assert setup.resolvers[0].provider_name == "ProfileCredentialProcess" + assert isinstance(setup.resolvers[0].resolver, ProcessCredentialsResolver) + + process = AsyncMock() + process.returncode = 0 + process.communicate.return_value = ( + json.dumps( + { + "Version": 1, + "AccessKeyId": "akid", + "SecretAccessKey": "secret", + } + ).encode(), + b"", + ) + with patch("asyncio.create_subprocess_exec", return_value=process) as mock_exec: + identity = await setup.resolvers[0].get_identity(properties={}) + + assert identity.access_key_id == "akid" + mock_exec.assert_called_once_with( + "credential-helper", + "--profile", + "test profile", + "--format", + "json", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + + +async def test_account_id_falls_back_to_profile_config( + setup_provider: Callable[..., Awaitable[ChainSetup]], + merged_config: Callable[..., MergedConfig], +) -> None: + setup = await setup_provider( + ProfileProcessCredentialsProvider(), + config_file=merged_config( + { + "default": { + "credential_process": "credential-helper", + "aws_account_id": "123456789012", + } + } + ), + profile_name="default", + ) + + process = AsyncMock() + process.returncode = 0 + process.communicate.return_value = ( + json.dumps( + { + "Version": 1, + "AccessKeyId": "akid", + "SecretAccessKey": "secret", + } + ).encode(), + b"", + ) + with patch("asyncio.create_subprocess_exec", return_value=process): + identity = await setup.resolvers[0].get_identity(properties={}) + + assert identity.account_id == "123456789012" diff --git a/packages/smithy-aws-core/tests/unit/identity/test_process.py b/packages/smithy-aws-core/tests/unit/identity/test_process.py new file mode 100644 index 000000000..352ebc334 --- /dev/null +++ b/packages/smithy-aws-core/tests/unit/identity/test_process.py @@ -0,0 +1,410 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: Apache-2.0 +import asyncio +import json +import traceback +from datetime import UTC, datetime, timedelta +from unittest.mock import AsyncMock, Mock, patch + +import pytest +from smithy_aws_core.identity.process import ProcessCredentialsResolver +from smithy_core.exceptions import SmithyIdentityError + +ISO8601 = "%Y-%m-%dT%H:%M:%SZ" + +DEFAULT_RESPONSE_DATA = { + "Version": 1, + "AccessKeyId": "foo", + "SecretAccessKey": "bar", + "SessionToken": "baz", +} + + +@pytest.mark.parametrize("command", [[], None, "mock-process", ["mock-process", 1]]) +def test_resolver_invalid_command(command: object): + with pytest.raises(ValueError, match="command must be a non-empty list"): + ProcessCredentialsResolver(command) # type: ignore[arg-type] + + +def mock_subprocess(returncode: int, stdout: bytes, stderr: bytes = b""): + """Helper to mock asyncio.create_subprocess_exec""" + process = AsyncMock() + process.returncode = returncode + process.communicate.return_value = (stdout, stderr) + return process + + +async def test_valid_credentials_with_session_token(): + resp_body = json.dumps(DEFAULT_RESPONSE_DATA) + process = mock_subprocess(0, resp_body.encode("utf-8")) + + with patch("asyncio.create_subprocess_exec", return_value=process): + resolver = ProcessCredentialsResolver(["mock-process"]) + identity = await resolver.get_identity(properties={}) + + assert identity.access_key_id == "foo" + assert identity.secret_access_key == "bar" + assert identity.session_token == "baz" + assert identity.expiration is None + assert identity.account_id is None + + +async def test_valid_credentials_without_session_token(): + resp_data = { + "Version": 1, + "AccessKeyId": "foo", + "SecretAccessKey": "bar", + } + resp_body = json.dumps(resp_data) + process = mock_subprocess(0, resp_body.encode("utf-8")) + + with patch("asyncio.create_subprocess_exec", return_value=process): + resolver = ProcessCredentialsResolver(["mock-process"]) + identity = await resolver.get_identity(properties={}) + + assert identity.access_key_id == "foo" + assert identity.secret_access_key == "bar" + assert identity.session_token is None + + +async def test_credentials_with_expiration(): + current_time = datetime.now(UTC) + timedelta(minutes=10) + resp_data = dict(DEFAULT_RESPONSE_DATA) + resp_data["Expiration"] = current_time.strftime(ISO8601) + + resp_body = json.dumps(resp_data) + process = mock_subprocess(0, resp_body.encode("utf-8")) + + with patch("asyncio.create_subprocess_exec", return_value=process): + resolver = ProcessCredentialsResolver(["mock-process"]) + identity = await resolver.get_identity(properties={}) + + assert identity.expiration is not None + assert identity.expiration.tzinfo == UTC + + +async def test_credentials_with_non_utc_expiration(): + """Test that non-UTC expiration timestamps are correctly converted to UTC.""" + # 2026-03-16T10:00:00+05:00 should become 2026-03-16T05:00:00 UTC + resp_data = dict(DEFAULT_RESPONSE_DATA) + resp_data["Expiration"] = "2026-03-16T10:00:00+05:00" + + resp_body = json.dumps(resp_data) + process = mock_subprocess(0, resp_body.encode("utf-8")) + + with patch("asyncio.create_subprocess_exec", return_value=process): + resolver = ProcessCredentialsResolver(["mock-process"]) + identity = await resolver.get_identity(properties={}) + + assert identity.expiration is not None + assert identity.expiration.tzinfo == UTC + assert identity.expiration == datetime(2026, 3, 16, 5, 0, 0, tzinfo=UTC) + + +async def test_invalid_expiration_string(): + resp_data = dict(DEFAULT_RESPONSE_DATA) + resp_data["Expiration"] = "not-a-timestamp" + + resp_body = json.dumps(resp_data) + process = mock_subprocess(0, resp_body.encode("utf-8")) + + with patch("asyncio.create_subprocess_exec", return_value=process): + resolver = ProcessCredentialsResolver(["mock-process"]) + with pytest.raises( + SmithyIdentityError, + match="Invalid credential process Expiration; expected an ISO 8601 string", + ): + await resolver.get_identity(properties={}) + + +async def test_non_string_expiration(): + resp_data = dict(DEFAULT_RESPONSE_DATA) + resp_data["Expiration"] = 12345 + + resp_body = json.dumps(resp_data) + process = mock_subprocess(0, resp_body.encode("utf-8")) + + with patch("asyncio.create_subprocess_exec", return_value=process): + resolver = ProcessCredentialsResolver(["mock-process"]) + with pytest.raises( + SmithyIdentityError, + match="Invalid credential process Expiration; expected an ISO 8601 string", + ): + await resolver.get_identity(properties={}) + + +async def test_credentials_with_account_id(): + resp_data = dict(DEFAULT_RESPONSE_DATA) + resp_data["AccountId"] = "123456789012" + + resp_body = json.dumps(resp_data) + process = mock_subprocess(0, resp_body.encode("utf-8")) + + with patch("asyncio.create_subprocess_exec", return_value=process): + resolver = ProcessCredentialsResolver(["mock-process"]) + identity = await resolver.get_identity(properties={}) + + assert identity.account_id == "123456789012" + + +async def test_account_id_falls_back_to_configured_value(): + """The configured account_id is used when the process omits AccountId.""" + resp_body = json.dumps(DEFAULT_RESPONSE_DATA) + process = mock_subprocess(0, resp_body.encode("utf-8")) + + with patch("asyncio.create_subprocess_exec", return_value=process): + resolver = ProcessCredentialsResolver( + ["mock-process"], account_id="123456789012" + ) + identity = await resolver.get_identity(properties={}) + + assert identity.account_id == "123456789012" + + +async def test_process_account_id_takes_precedence_over_configured_value(): + """The process output's AccountId wins over the configured fallback.""" + resp_data = dict(DEFAULT_RESPONSE_DATA) + resp_data["AccountId"] = "111111111111" + + resp_body = json.dumps(resp_data) + process = mock_subprocess(0, resp_body.encode("utf-8")) + + with patch("asyncio.create_subprocess_exec", return_value=process): + resolver = ProcessCredentialsResolver( + ["mock-process"], account_id="222222222222" + ) + identity = await resolver.get_identity(properties={}) + + assert identity.account_id == "111111111111" + + +async def test_non_zero_exit_code(): + process = mock_subprocess(1, b"", b"Process error message") + + with patch("asyncio.create_subprocess_exec", return_value=process): + resolver = ProcessCredentialsResolver(["mock-process"]) + with pytest.raises( + SmithyIdentityError, + match="exit code 1: Process error message", + ): + await resolver.get_identity(properties={}) + + +@pytest.mark.parametrize( + "resp_data", + [ + {"Version": 1, "SecretAccessKey": "bar"}, + {"Version": 1, "AccessKeyId": "foo"}, + ], +) +async def test_missing_required_credentials(resp_data: dict[str, object]): + resp_body = json.dumps(resp_data) + process = mock_subprocess(0, resp_body.encode("utf-8")) + + with patch("asyncio.create_subprocess_exec", return_value=process): + resolver = ProcessCredentialsResolver(["mock-process"]) + with pytest.raises( + SmithyIdentityError, + match="AccessKeyId and SecretAccessKey are required", + ): + await resolver.get_identity(properties={}) + + +async def test_invalid_version(): + resp_data = dict(DEFAULT_RESPONSE_DATA) + resp_data["Version"] = 2 + + resp_body = json.dumps(resp_data) + process = mock_subprocess(0, resp_body.encode("utf-8")) + + with patch("asyncio.create_subprocess_exec", return_value=process): + resolver = ProcessCredentialsResolver(["mock-process"]) + with pytest.raises(SmithyIdentityError, match="Unsupported version '2'"): + await resolver.get_identity(properties={}) + + +async def test_missing_version(): + resp_data = { + "AccessKeyId": "foo", + "SecretAccessKey": "bar", + } + resp_body = json.dumps(resp_data) + process = mock_subprocess(0, resp_body.encode("utf-8")) + + with patch("asyncio.create_subprocess_exec", return_value=process): + resolver = ProcessCredentialsResolver(["mock-process"]) + with pytest.raises(SmithyIdentityError, match="Unsupported version 'None'"): + await resolver.get_identity(properties={}) + + +async def test_invalid_json(): + process = mock_subprocess(0, b'{"SecretAccessKey": "json-secret"') + + with patch("asyncio.create_subprocess_exec", return_value=process): + resolver = ProcessCredentialsResolver(["mock-process"]) + with pytest.raises( + SmithyIdentityError, + match="Credential process output is not valid JSON at line 1, column", + ) as exc_info: + await resolver.get_identity(properties={}) + + rendered = "".join(traceback.format_exception(exc_info.value)) + assert "json-secret" not in rendered + + +async def test_invalid_utf8(): + process = mock_subprocess(0, b'{"SecretAccessKey": "utf8-secret"}\xff') + + with patch("asyncio.create_subprocess_exec", return_value=process): + resolver = ProcessCredentialsResolver(["mock-process"]) + with pytest.raises( + SmithyIdentityError, + match="Credential process output is not valid UTF-8 at byte", + ) as exc_info: + await resolver.get_identity(properties={}) + + rendered = "".join(traceback.format_exception(exc_info.value)) + assert "utf8-secret" not in rendered + + +async def test_process_timeout(): + process = AsyncMock() + process.returncode = None + process.kill = Mock() + process.wait = AsyncMock() + + with ( + patch("asyncio.create_subprocess_exec", return_value=process), + patch("asyncio.wait_for", side_effect=TimeoutError), + ): + resolver = ProcessCredentialsResolver(["mock-process"], timeout=1) + with pytest.raises(SmithyIdentityError, match="timed out after 1 seconds"): + await resolver.get_identity(properties={}) + + process.kill.assert_called_once_with() + process.wait.assert_awaited_once_with() + + +async def test_process_startup_failure_raises_smithy_identity_error(): + with patch( + "asyncio.create_subprocess_exec", + side_effect=FileNotFoundError("No such file or directory"), + ): + resolver = ProcessCredentialsResolver(["missing-process"]) + with pytest.raises(SmithyIdentityError, match="failed to start"): + await resolver.get_identity(properties={}) + + +async def test_long_term_credentials_cached(): + """Test that credentials without expiration are cached indefinitely.""" + resp_body = json.dumps(DEFAULT_RESPONSE_DATA) + process = mock_subprocess(0, resp_body.encode("utf-8")) + + with patch("asyncio.create_subprocess_exec", return_value=process) as mock_exec: + resolver = ProcessCredentialsResolver(["mock-process"]) + identity_one = await resolver.get_identity(properties={}) + identity_two = await resolver.get_identity(properties={}) + + # Process should only be called once + assert mock_exec.call_count == 1 + # Should return the same identity instance + assert identity_one is identity_two + + +async def test_temporary_credentials_cached_when_valid(): + """Test that temporary credentials are cached when not expired.""" + current_time = datetime.now(UTC) + timedelta(minutes=10) + resp_data = dict(DEFAULT_RESPONSE_DATA) + resp_data["Expiration"] = current_time.strftime(ISO8601) + + resp_body = json.dumps(resp_data) + process = mock_subprocess(0, resp_body.encode("utf-8")) + + with patch("asyncio.create_subprocess_exec", return_value=process) as mock_exec: + resolver = ProcessCredentialsResolver(["mock-process"]) + identity_one = await resolver.get_identity(properties={}) + identity_two = await resolver.get_identity(properties={}) + + # Process should only be called once + assert mock_exec.call_count == 1 + # Should return the same identity instance + assert identity_one is identity_two + + +async def test_expired_credentials_refreshed(): + """Test that expired credentials are refreshed.""" + expired_time = datetime.now(UTC) - timedelta(minutes=10) + initial_data = dict(DEFAULT_RESPONSE_DATA) + initial_data["Expiration"] = expired_time.strftime(ISO8601) + + refreshed_time = datetime.now(UTC) + timedelta(minutes=10) + refreshed_data = { + "Version": 1, + "AccessKeyId": "foo-refreshed", + "SecretAccessKey": "bar-refreshed", + "SessionToken": "baz-refreshed", + "Expiration": refreshed_time.strftime(ISO8601), + } + + first_process = mock_subprocess(0, json.dumps(initial_data).encode("utf-8")) + second_process = mock_subprocess(0, json.dumps(refreshed_data).encode("utf-8")) + + with patch( + "asyncio.create_subprocess_exec", + side_effect=[first_process, second_process], + ) as mock_exec: + resolver = ProcessCredentialsResolver(["mock-process"]) + identity_one = await resolver.get_identity(properties={}) + identity_two = await resolver.get_identity(properties={}) + + # Process should be called twice (once for initial, once for refresh) + assert mock_exec.call_count == 2 + # Should be different instances + assert identity_one is not identity_two + assert identity_one.access_key_id == "foo" + assert identity_one.secret_access_key == "bar" + assert identity_one.session_token == "baz" + assert identity_two.access_key_id == "foo-refreshed" + assert identity_two.secret_access_key == "bar-refreshed" + assert identity_two.session_token == "baz-refreshed" + + +async def test_invalidate_clears_cached_credentials(): + resp_body = json.dumps(DEFAULT_RESPONSE_DATA) + first_process = mock_subprocess(0, resp_body.encode("utf-8")) + second_process = mock_subprocess(0, resp_body.encode("utf-8")) + + with patch( + "asyncio.create_subprocess_exec", + side_effect=[first_process, second_process], + ) as mock_exec: + resolver = ProcessCredentialsResolver(["mock-process"]) + identity_one = await resolver.get_identity(properties={}) + await resolver.invalidate() + identity_two = await resolver.get_identity(properties={}) + + assert mock_exec.call_count == 2 + assert identity_one is not identity_two + + +async def test_command_with_multiple_args(): + """Test that commands with multiple arguments are passed correctly.""" + resp_body = json.dumps(DEFAULT_RESPONSE_DATA) + process = mock_subprocess(0, resp_body.encode("utf-8")) + + with patch("asyncio.create_subprocess_exec", return_value=process) as mock_exec: + resolver = ProcessCredentialsResolver( + ["aws-credential-helper", "--profile", "test", "--format", "json"] + ) + await resolver.get_identity(properties={}) + + # Verify the command was called with all arguments + mock_exec.assert_called_once_with( + "aws-credential-helper", + "--profile", + "test", + "--format", + "json", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + )