From 8130637aee08526daf069f7f2b6f04a73d29c5f5 Mon Sep 17 00:00:00 2001 From: Vikas Walunj Date: Tue, 22 Sep 2026 13:46:28 -0700 Subject: [PATCH 1/2] feat(payments): add RotatePaymentConnectorCredentials support Expose on-demand rotation of a connector's service-managed credentials, and surface the two connector fields the operation relies on. - `PaymentClient.rotate_payment_connector_credentials()` wraps the control plane operation, building the `credentialsToRotate` union and validating it against the model's constraints (non-empty, unique, known secret names) before the request goes out. - New `CoinbaseCdpSecret` constant (`API_KEY`, `WALLET_SECRET`) names the rotatable secrets; an enum rather than a bool so new kinds can be added without a breaking change. - Allowlist the operation for boto3 forwarding, so it is also reachable directly. - `get_payment_connector()` and `list_payment_connectors()` now return `provisionMode`, and `get_payment_connector()` returns `credentialsUpdatedAt` when the service supplies it. Both were previously dropped by the hand-built result dicts, so callers had no way to tell whether a connector's credentials are service-managed or how old they are. Rotation applies only to QUICK_CREATE connectors. MANUAL connectors are unaffected: the caller owns those credentials and rotates them with the payment provider directly. Tests - 16 new cases in tests/bedrock_agentcore/payments/test_client.py covering the union builder, enum/string inputs, de-duplication, each validation error, client token handling, ConflictException propagation, and the new response fields. - Full payments suite: 917 passed, 9 skipped; coverage 91%. Ruff lint and format clean. --- src/bedrock_agentcore/payments/README.md | 35 +++ src/bedrock_agentcore/payments/__init__.py | 2 + src/bedrock_agentcore/payments/client.py | 139 +++++++++- src/bedrock_agentcore/payments/constants.py | 11 + .../bedrock_agentcore/payments/test_client.py | 239 +++++++++++++++++- 5 files changed, 424 insertions(+), 2 deletions(-) diff --git a/src/bedrock_agentcore/payments/README.md b/src/bedrock_agentcore/payments/README.md index d4e858b7..eba7f1e9 100644 --- a/src/bedrock_agentcore/payments/README.md +++ b/src/bedrock_agentcore/payments/README.md @@ -238,6 +238,39 @@ user interaction, `wait_for_ready=True` cannot be used when creating a Quick Cre --- +### Rotating Connector Credentials + +Quick Create connectors have service-managed credentials, so you can replace them without signing in +to the payment provider: + +```python +from bedrock_agentcore.payments import CoinbaseCdpSecret, PaymentClient + +payment_client = PaymentClient(region_name="us-east-1") +result = payment_client.rotate_payment_connector_credentials( + payment_manager_id="payment-manager-id", + payment_connector_id="payment-connector-id", + secrets=[CoinbaseCdpSecret.API_KEY, CoinbaseCdpSecret.WALLET_SECRET], +) + +print(result["status"], result["updatedAt"]) +``` + +The rotation completes before the response returns, so there is no status to poll — on success the +connector stays `READY`, and on failure it is left unchanged and the request can be retried. Only one +rotation runs at a time per connector, so a concurrent call fails with `ConflictException`. + +Rotation replaces the credentials on the connector's credential provider, so every connector sharing +that provider is affected. Replace any copy of the previous credentials that you use outside +AgentCore. + +This applies only to connectors with a `provisionMode` of `QUICK_CREATE` — check +`get_payment_connector()["provisionMode"]` if you are unsure. For `MANUAL` connectors you own the +credentials: rotate them with the payment provider, then update the credential provider with the new +values. + +--- + ### Creating a Payment Instrument Create a payment instrument for a user. Below is an example creating an Ethereum-compatible embedded crypto wallet: @@ -907,6 +940,7 @@ except PaymentError as e: | `get_payment_connector()` | Retrieve payment connector details | | `list_payment_connectors()` | List payment connectors for a manager | | `update_payment_connector()` | Update a payment connector | +| `rotate_payment_connector_credentials()` | Rotate a connector's service-managed credentials | | `delete_payment_connector()` | Delete a payment connector | | `create_payment_manager_with_connector()` | One-step setup with automatic rollback | @@ -931,6 +965,7 @@ except PaymentError as e: | `PaymentConnectorStatus` | Payment connector statuses | | `PaymentConnectorType` | Supported connector types (CoinbaseCDP, StripePrivy) | | `PaymentConnectorProvisionMode` | Connector provisioning modes (MANUAL, QUICK_CREATE) | +| `CoinbaseCdpSecret` | Rotatable Coinbase CDP secrets (API_KEY, WALLET_SECRET) | | `PaymentsAuthorizerType` | Authorizer types (AWS_IAM, CUSTOM_JWT) | | `NETWORK_PREFERENCES` | Default blockchain network preference order | | `DEFAULT_MAX_RESULTS` | Default pagination limit (100) | diff --git a/src/bedrock_agentcore/payments/__init__.py b/src/bedrock_agentcore/payments/__init__.py index 0ab2a142..4ec5618d 100644 --- a/src/bedrock_agentcore/payments/__init__.py +++ b/src/bedrock_agentcore/payments/__init__.py @@ -3,6 +3,7 @@ from .client import PaymentClient from .constants import ( DEFAULT_MAX_RESULTS, + CoinbaseCdpSecret, PaymentConnectorProvisionMode, PaymentConnectorStatus, PaymentConnectorType, @@ -44,6 +45,7 @@ "PaymentConnectorStatus", "PaymentConnectorType", "PaymentConnectorProvisionMode", + "CoinbaseCdpSecret", "PaymentsAuthorizerType", "PaymentType", "DEFAULT_MAX_RESULTS", diff --git a/src/bedrock_agentcore/payments/client.py b/src/bedrock_agentcore/payments/client.py index fb657e77..e602471f 100644 --- a/src/bedrock_agentcore/payments/client.py +++ b/src/bedrock_agentcore/payments/client.py @@ -20,7 +20,12 @@ from bedrock_agentcore._utils.user_agent import build_user_agent_suffix from bedrock_agentcore.services.identity import IdentityClient -from .constants import PaymentConnectorProvisionMode, PaymentConnectorStatus +from .constants import ( + CoinbaseCdpSecret, + PaymentConnectorProvisionMode, + PaymentConnectorStatus, + PaymentConnectorType, +) logger = logging.getLogger(__name__) @@ -138,6 +143,7 @@ class PaymentClient: "list_payment_connectors", "update_payment_connector", "delete_payment_connector", + "rotate_payment_connector_credentials", } @staticmethod @@ -247,6 +253,59 @@ def _build_provider_config_input( f"Unsupported credential_provider_vendor: '{vendor}'. Supported vendors are: CoinbaseCDP, StripePrivy" ) + @staticmethod + def _build_rotation_config_input( + connector_type: Union[str, PaymentConnectorType], + secrets: List[Union[str, CoinbaseCdpSecret]], + ) -> Dict[str, Any]: + """Build the credentialsToRotate input for a credential rotation request. + + Args: + connector_type: The connector's type, which selects the provider to rotate for + secrets: The service-managed secrets to rotate. Accepts CoinbaseCdpSecret + members or their string values. Duplicates are dropped and the original + order is preserved. + + Returns: + Dictionary with the single provider entry for the connector type + + Raises: + ValueError: If the connector type does not support rotation, or if secrets is + empty or names an unknown secret + + Example: + For CoinbaseCDP connectors: + { + "coinbaseCDP": { + "secrets": ["API_KEY", "WALLET_SECRET"] + } + } + """ + normalized_type = connector_type.value if isinstance(connector_type, PaymentConnectorType) else connector_type + + if normalized_type != PaymentConnectorType.COINBASE_CDP.value: + raise ValueError( + f"Credential rotation is not supported for connector type: '{normalized_type}'. " + f"Supported types are: {PaymentConnectorType.COINBASE_CDP.value}" + ) + + if not secrets: + raise ValueError("secrets is required and must name at least one secret to rotate") + + supported_secrets = {secret.value for secret in CoinbaseCdpSecret} + normalized_secrets: List[str] = [] + for secret in secrets: + value = secret.value if isinstance(secret, CoinbaseCdpSecret) else secret + if value not in supported_secrets: + raise ValueError( + f"Unsupported CoinbaseCDP secret: '{value}'. " + f"Supported secrets are: {', '.join(sorted(supported_secrets))}" + ) + if value not in normalized_secrets: + normalized_secrets.append(value) + + return {"coinbaseCDP": {"secrets": normalized_secrets}} + def __init__( self, region_name: Optional[str] = None, @@ -794,6 +853,7 @@ def get_payment_connector(self, payment_manager_id: str, payment_connector_id: s "name": response.get("name"), "description": response.get("description"), "providerType": response.get("type"), + "provisionMode": response.get("provisionMode"), "status": response.get("status"), "createdAt": response.get("createdAt"), "updatedAt": response.get("lastUpdatedAt"), @@ -803,6 +863,10 @@ def get_payment_connector(self, payment_manager_id: str, payment_connector_id: s authorization_url ): result["authorizationUrl"] = authorization_url + # Present only for QUICK_CREATE connectors, whose credentials the service manages + credentials_updated_at = response.get("credentialsUpdatedAt") + if credentials_updated_at is not None: + result["credentialsUpdatedAt"] = credentials_updated_at return result except ClientError as e: @@ -849,6 +913,7 @@ def list_payment_connectors( "name": connector.get("name"), "description": connector.get("description"), "providerType": connector.get("type"), + "provisionMode": connector.get("provisionMode"), "status": connector.get("status"), "createdAt": connector.get("createdAt"), "updatedAt": connector.get("lastUpdatedAt"), @@ -990,6 +1055,78 @@ def get_connector_status(conn_id): logger.error("Failed to update payment connector: %s", e) raise + def rotate_payment_connector_credentials( + self, + payment_manager_id: str, + payment_connector_id: str, + secrets: List[Union[str, CoinbaseCdpSecret]], + connector_type: Union[str, PaymentConnectorType] = PaymentConnectorType.COINBASE_CDP, + client_token: Optional[str] = None, + ) -> Dict[str, Any]: + """Replace a payment connector's service-managed credentials with new ones. + + Use this only for connectors with a provision mode of QUICK_CREATE, whose + credentials the service issued and stores. For a connector with a provision mode of + MANUAL you own the credentials: rotate them with the payment provider, then call + UpdatePaymentCredentialProvider with the new values. + + The rotation completes before the response is returned, so there is no status to + poll. On success the connector stays READY. On failure the connector and its + existing credentials are left unchanged and the request can be retried. Only one + rotation runs at a time for a given connector, so a concurrent call fails with + ConflictException. + + Rotation replaces the credentials on the connector's credential provider, so every + connector that uses that provider is affected. Replace any copy of the previous + credentials that you use outside AgentCore. + + Args: + payment_manager_id: ID of the payment manager + payment_connector_id: ID of the connector whose credentials to rotate + secrets: The service-managed secrets to rotate (at least one). Accepts + CoinbaseCdpSecret members or their string values. + connector_type: The connector's type. Defaults to CoinbaseCDP, currently the + only type that supports rotation. + client_token: Optional idempotency token. If not provided, a UUID will be generated. + + Returns: + Dictionary with the connector IDs, its status after the rotation, and the + timestamp the rotation completed + + Raises: + ValueError: If the connector type does not support rotation, or if secrets is + empty or names an unknown secret + ClientError: If the rotation fails + """ + credentials_to_rotate = self._build_rotation_config_input(connector_type, secrets) + + if client_token is None: + client_token = str(uuid.uuid4()) + + try: + logger.info( + "Rotating credentials for payment connector: %s for manager %s", + payment_connector_id, + payment_manager_id, + ) + response = self.payments_cp_client.rotate_payment_connector_credentials( + paymentManagerId=payment_manager_id, + paymentConnectorId=payment_connector_id, + credentialsToRotate=credentials_to_rotate, + clientToken=client_token, + ) + + return { + "paymentConnectorId": response.get("paymentConnectorId"), + "paymentManagerId": response.get("paymentManagerId"), + "status": response.get("status"), + "updatedAt": response.get("lastUpdatedAt"), + } + + except ClientError as e: + logger.error("Failed to rotate payment connector credentials: %s", e) + raise + def create_payment_manager_with_connector( self, payment_manager_name: str, diff --git a/src/bedrock_agentcore/payments/constants.py b/src/bedrock_agentcore/payments/constants.py index 551116b8..15f43c87 100644 --- a/src/bedrock_agentcore/payments/constants.py +++ b/src/bedrock_agentcore/payments/constants.py @@ -51,6 +51,17 @@ class PaymentConnectorProvisionMode(Enum): QUICK_CREATE = "QUICK_CREATE" +class CoinbaseCdpSecret(Enum): + """Service-managed Coinbase CDP secrets that can be rotated. + + Only credentials the service provisioned (QUICK_CREATE provision mode) are + rotatable. Credentials you supplied yourself remain your responsibility. + """ + + API_KEY = "API_KEY" + WALLET_SECRET = "WALLET_SECRET" + + class PaymentType(Enum): """Payment protocols supported by ProcessPayment.""" diff --git a/tests/bedrock_agentcore/payments/test_client.py b/tests/bedrock_agentcore/payments/test_client.py index 799e3db9..e2183103 100644 --- a/tests/bedrock_agentcore/payments/test_client.py +++ b/tests/bedrock_agentcore/payments/test_client.py @@ -6,7 +6,12 @@ import pytest from botocore.exceptions import ClientError -from bedrock_agentcore.payments import PaymentClient, PaymentConnectorProvisionMode +from bedrock_agentcore.payments import ( + CoinbaseCdpSecret, + PaymentClient, + PaymentConnectorProvisionMode, + PaymentConnectorType, +) from bedrock_agentcore.payments.client import PaymentConnectorConfig # Get role ARN from environment variable, with fallback for testing @@ -1659,3 +1664,235 @@ def test_update_payment_connector_error_raises(self, mock_session, mock_boto3_cl payment_manager_id="pm-123", payment_connector_id="pc-123", ) + + +class TestBuildRotationConfigInput: + """Tests for _build_rotation_config_input static method.""" + + def test_coinbase_cdp_secrets(self): + """CoinbaseCDP produces the coinbaseCDP union member.""" + result = PaymentClient._build_rotation_config_input( + "CoinbaseCDP", [CoinbaseCdpSecret.API_KEY, CoinbaseCdpSecret.WALLET_SECRET] + ) + assert result == {"coinbaseCDP": {"secrets": ["API_KEY", "WALLET_SECRET"]}} + + def test_accepts_enum_type(self): + """connector_type accepts a PaymentConnectorType member.""" + result = PaymentClient._build_rotation_config_input( + PaymentConnectorType.COINBASE_CDP, [CoinbaseCdpSecret.API_KEY] + ) + assert result == {"coinbaseCDP": {"secrets": ["API_KEY"]}} + + def test_accepts_string_secrets(self): + """Raw secret strings are accepted alongside enum members.""" + result = PaymentClient._build_rotation_config_input("CoinbaseCDP", ["WALLET_SECRET", CoinbaseCdpSecret.API_KEY]) + assert result == {"coinbaseCDP": {"secrets": ["WALLET_SECRET", "API_KEY"]}} + + def test_duplicates_removed_order_preserved(self): + """Duplicate secrets are dropped, satisfying the model's uniqueItems constraint.""" + result = PaymentClient._build_rotation_config_input( + "CoinbaseCDP", + [CoinbaseCdpSecret.WALLET_SECRET, "WALLET_SECRET", CoinbaseCdpSecret.API_KEY], + ) + assert result == {"coinbaseCDP": {"secrets": ["WALLET_SECRET", "API_KEY"]}} + + def test_empty_secrets_raises(self): + """An empty secrets list is rejected before the request is sent.""" + with pytest.raises(ValueError, match="at least one secret"): + PaymentClient._build_rotation_config_input("CoinbaseCDP", []) + + def test_unknown_secret_raises(self): + """An unrecognized secret name is rejected.""" + with pytest.raises(ValueError, match="Unsupported CoinbaseCDP secret"): + PaymentClient._build_rotation_config_input("CoinbaseCDP", ["ROOT_PASSWORD"]) + + def test_unsupported_connector_type_raises(self): + """StripePrivy has no rotatable service-managed secrets.""" + with pytest.raises(ValueError, match="not supported for connector type"): + PaymentClient._build_rotation_config_input(PaymentConnectorType.STRIPE_PRIVY, [CoinbaseCdpSecret.API_KEY]) + + +class TestRotatePaymentConnectorCredentials: + """Tests for PaymentClient.rotate_payment_connector_credentials.""" + + @patch("bedrock_agentcore.payments.client.boto3.client") + @patch("bedrock_agentcore.payments.client.boto3.Session") + def test_rotate_success(self, mock_session, mock_boto3_client): + """Successful rotation returns the connector IDs, status and completion timestamp.""" + mock_session.return_value.region_name = "us-west-2" + mock_cp_client = MagicMock() + mock_boto3_client.return_value = mock_cp_client + mock_cp_client.rotate_payment_connector_credentials.return_value = { + "paymentConnectorId": "pc-123", + "paymentManagerId": "pm-123", + "status": "READY", + "lastUpdatedAt": "2026-09-22T00:00:00Z", + } + + client = PaymentClient(region_name="us-west-2") + result = client.rotate_payment_connector_credentials( + payment_manager_id="pm-123", + payment_connector_id="pc-123", + secrets=[CoinbaseCdpSecret.API_KEY], + ) + + call_kwargs = mock_cp_client.rotate_payment_connector_credentials.call_args[1] + assert call_kwargs["paymentManagerId"] == "pm-123" + assert call_kwargs["paymentConnectorId"] == "pc-123" + assert call_kwargs["credentialsToRotate"] == {"coinbaseCDP": {"secrets": ["API_KEY"]}} + + assert result == { + "paymentConnectorId": "pc-123", + "paymentManagerId": "pm-123", + "status": "READY", + "updatedAt": "2026-09-22T00:00:00Z", + } + + @patch("bedrock_agentcore.payments.client.boto3.client") + @patch("bedrock_agentcore.payments.client.boto3.Session") + def test_rotate_generates_client_token(self, mock_session, mock_boto3_client): + """A client token is generated when the caller does not supply one.""" + mock_session.return_value.region_name = "us-west-2" + mock_cp_client = MagicMock() + mock_boto3_client.return_value = mock_cp_client + mock_cp_client.rotate_payment_connector_credentials.return_value = {} + + client = PaymentClient(region_name="us-west-2") + client.rotate_payment_connector_credentials( + payment_manager_id="pm-123", + payment_connector_id="pc-123", + secrets=["API_KEY"], + ) + + call_kwargs = mock_cp_client.rotate_payment_connector_credentials.call_args[1] + assert call_kwargs["clientToken"] + + @patch("bedrock_agentcore.payments.client.boto3.client") + @patch("bedrock_agentcore.payments.client.boto3.Session") + def test_rotate_passes_client_token_through(self, mock_session, mock_boto3_client): + """A caller-supplied client token is used verbatim.""" + mock_session.return_value.region_name = "us-west-2" + mock_cp_client = MagicMock() + mock_boto3_client.return_value = mock_cp_client + mock_cp_client.rotate_payment_connector_credentials.return_value = {} + + client = PaymentClient(region_name="us-west-2") + client.rotate_payment_connector_credentials( + payment_manager_id="pm-123", + payment_connector_id="pc-123", + secrets=["API_KEY"], + client_token="token-abc", + ) + + call_kwargs = mock_cp_client.rotate_payment_connector_credentials.call_args[1] + assert call_kwargs["clientToken"] == "token-abc" + + @patch("bedrock_agentcore.payments.client.boto3.client") + @patch("bedrock_agentcore.payments.client.boto3.Session") + def test_rotate_validates_before_calling_service(self, mock_session, mock_boto3_client): + """Invalid input raises ValueError without issuing a request.""" + mock_session.return_value.region_name = "us-west-2" + mock_cp_client = MagicMock() + mock_boto3_client.return_value = mock_cp_client + + client = PaymentClient(region_name="us-west-2") + with pytest.raises(ValueError): + client.rotate_payment_connector_credentials( + payment_manager_id="pm-123", + payment_connector_id="pc-123", + secrets=[], + ) + + mock_cp_client.rotate_payment_connector_credentials.assert_not_called() + + @patch("bedrock_agentcore.payments.client.boto3.client") + @patch("bedrock_agentcore.payments.client.boto3.Session") + def test_rotate_conflict_raises(self, mock_session, mock_boto3_client): + """A concurrent rotation surfaces as ClientError.""" + mock_session.return_value.region_name = "us-west-2" + mock_cp_client = MagicMock() + mock_boto3_client.return_value = mock_cp_client + mock_cp_client.rotate_payment_connector_credentials.side_effect = ClientError( + {"Error": {"Code": "ConflictException", "Message": "Rotation already in progress"}}, + "RotatePaymentConnectorCredentials", + ) + + client = PaymentClient(region_name="us-west-2") + with pytest.raises(ClientError): + client.rotate_payment_connector_credentials( + payment_manager_id="pm-123", + payment_connector_id="pc-123", + secrets=["API_KEY"], + ) + + @patch("bedrock_agentcore.payments.client.boto3.client") + @patch("bedrock_agentcore.payments.client.boto3.Session") + def test_rotate_reachable_through_forwarding(self, mock_session, mock_boto3_client): + """The operation is allowlisted for direct boto3 forwarding.""" + mock_session.return_value.region_name = "us-west-2" + mock_boto3_client.return_value = MagicMock() + + assert "rotate_payment_connector_credentials" in PaymentClient._ALLOWED_PAYMENTS_CP_METHODS + + +class TestPaymentConnectorProvisionFields: + """Tests that provisionMode and credentialsUpdatedAt are surfaced to callers.""" + + @patch("bedrock_agentcore.payments.client.boto3.client") + @patch("bedrock_agentcore.payments.client.boto3.Session") + def test_get_connector_returns_provision_fields(self, mock_session, mock_boto3_client): + """get_payment_connector surfaces provisionMode and credentialsUpdatedAt.""" + mock_session.return_value.region_name = "us-west-2" + mock_cp_client = MagicMock() + mock_boto3_client.return_value = mock_cp_client + mock_cp_client.get_payment_connector.return_value = { + "paymentConnectorId": "pc-123", + "paymentManagerId": "pm-123", + "status": "READY", + "provisionMode": "QUICK_CREATE", + "credentialsUpdatedAt": "2026-09-22T00:00:00Z", + } + + client = PaymentClient(region_name="us-west-2") + result = client.get_payment_connector("pm-123", "pc-123") + + assert result["provisionMode"] == "QUICK_CREATE" + assert result["credentialsUpdatedAt"] == "2026-09-22T00:00:00Z" + + @patch("bedrock_agentcore.payments.client.boto3.client") + @patch("bedrock_agentcore.payments.client.boto3.Session") + def test_get_connector_omits_credentials_updated_at_when_absent(self, mock_session, mock_boto3_client): + """MANUAL connectors have no credentialsUpdatedAt, so the key is omitted.""" + mock_session.return_value.region_name = "us-west-2" + mock_cp_client = MagicMock() + mock_boto3_client.return_value = mock_cp_client + mock_cp_client.get_payment_connector.return_value = { + "paymentConnectorId": "pc-123", + "status": "READY", + "provisionMode": "MANUAL", + } + + client = PaymentClient(region_name="us-west-2") + result = client.get_payment_connector("pm-123", "pc-123") + + assert result["provisionMode"] == "MANUAL" + assert "credentialsUpdatedAt" not in result + + @patch("bedrock_agentcore.payments.client.boto3.client") + @patch("bedrock_agentcore.payments.client.boto3.Session") + def test_list_connectors_returns_provision_mode(self, mock_session, mock_boto3_client): + """list_payment_connectors surfaces provisionMode for each connector.""" + mock_session.return_value.region_name = "us-west-2" + mock_cp_client = MagicMock() + mock_boto3_client.return_value = mock_cp_client + mock_cp_client.list_payment_connectors.return_value = { + "paymentConnectors": [ + {"paymentConnectorId": "pc-1", "provisionMode": "QUICK_CREATE"}, + {"paymentConnectorId": "pc-2", "provisionMode": "MANUAL"}, + ] + } + + client = PaymentClient(region_name="us-west-2") + result = client.list_payment_connectors("pm-123") + + assert [c["provisionMode"] for c in result["paymentConnectors"]] == ["QUICK_CREATE", "MANUAL"] From f3e762155ab45124d71b16dd97688ded3681b13a Mon Sep 17 00:00:00 2001 From: Vikas Walunj Date: Tue, 22 Sep 2026 18:30:12 -0700 Subject: [PATCH 2/2] refactor(payments): let the service validate rotation secrets Addresses review feedback: the SDK was re-implementing constraints the control plane model already enforces, so a secret name the service adds would be rejected client-side until the SDK shipped a new release. Drops from `_build_rotation_config_input`: - the `CoinbaseCdpSecret` allowlist check (service enforces the enum) - the empty-list check (`@length(min: 1)`) - de-duplication (`@uniqueItems`) The builder now only normalizes enum members to their string values and wraps them in the union member. Invalid input surfaces as a `ValidationException` from the service rather than a local `ValueError`. Also removes the `connector_type` parameter rather than just its check. `CredentialRotationConfig` models exactly one member (`coinbaseCDP`), so the parameter selected from a set of one; keeping it without the check would silently build a `coinbaseCDP` payload for a caller who passed `StripePrivy`. The README example never passed it, so this is not a user-facing change. `_build_provider_config_input` keeps its vendor check: that one picks between two genuinely different payload shapes, so an unknown vendor has no correct output. Tests - `TestBuildRotationConfigInput`: 7 cases down to 4, with the two new ones pinning the pass-through contract (duplicates and empty lists go out on the wire unchanged). - `test_rotate_validates_before_calling_service` becomes `test_rotate_defers_validation_to_service`, asserting the service's ClientError propagates and the unvalidated payload was forwarded. - Payments + init + utils suites: 921 passed, 9 skipped. Ruff lint and format clean. --- src/bedrock_agentcore/payments/client.py | 48 ++++------------ .../bedrock_agentcore/payments/test_client.py | 55 +++++++------------ 2 files changed, 30 insertions(+), 73 deletions(-) diff --git a/src/bedrock_agentcore/payments/client.py b/src/bedrock_agentcore/payments/client.py index e602471f..82fc9ba8 100644 --- a/src/bedrock_agentcore/payments/client.py +++ b/src/bedrock_agentcore/payments/client.py @@ -24,7 +24,6 @@ CoinbaseCdpSecret, PaymentConnectorProvisionMode, PaymentConnectorStatus, - PaymentConnectorType, ) logger = logging.getLogger(__name__) @@ -255,24 +254,22 @@ def _build_provider_config_input( @staticmethod def _build_rotation_config_input( - connector_type: Union[str, PaymentConnectorType], secrets: List[Union[str, CoinbaseCdpSecret]], ) -> Dict[str, Any]: """Build the credentialsToRotate input for a credential rotation request. + CoinbaseCDP is the only provider CredentialRotationConfig currently models, so + there is nothing for the caller to select. The service validates the secret names + (CoinbaseCdpSecret enum), rejects an empty list (@length(min: 1)) and rejects + duplicates (@uniqueItems), so this only normalizes enum members to their values. + Args: - connector_type: The connector's type, which selects the provider to rotate for secrets: The service-managed secrets to rotate. Accepts CoinbaseCdpSecret - members or their string values. Duplicates are dropped and the original - order is preserved. + members or their string values. Returns: Dictionary with the single provider entry for the connector type - Raises: - ValueError: If the connector type does not support rotation, or if secrets is - empty or names an unknown secret - Example: For CoinbaseCDP connectors: { @@ -281,28 +278,7 @@ def _build_rotation_config_input( } } """ - normalized_type = connector_type.value if isinstance(connector_type, PaymentConnectorType) else connector_type - - if normalized_type != PaymentConnectorType.COINBASE_CDP.value: - raise ValueError( - f"Credential rotation is not supported for connector type: '{normalized_type}'. " - f"Supported types are: {PaymentConnectorType.COINBASE_CDP.value}" - ) - - if not secrets: - raise ValueError("secrets is required and must name at least one secret to rotate") - - supported_secrets = {secret.value for secret in CoinbaseCdpSecret} - normalized_secrets: List[str] = [] - for secret in secrets: - value = secret.value if isinstance(secret, CoinbaseCdpSecret) else secret - if value not in supported_secrets: - raise ValueError( - f"Unsupported CoinbaseCDP secret: '{value}'. " - f"Supported secrets are: {', '.join(sorted(supported_secrets))}" - ) - if value not in normalized_secrets: - normalized_secrets.append(value) + normalized_secrets = [secret.value if isinstance(secret, CoinbaseCdpSecret) else secret for secret in secrets] return {"coinbaseCDP": {"secrets": normalized_secrets}} @@ -1060,7 +1036,6 @@ def rotate_payment_connector_credentials( payment_manager_id: str, payment_connector_id: str, secrets: List[Union[str, CoinbaseCdpSecret]], - connector_type: Union[str, PaymentConnectorType] = PaymentConnectorType.COINBASE_CDP, client_token: Optional[str] = None, ) -> Dict[str, Any]: """Replace a payment connector's service-managed credentials with new ones. @@ -1085,8 +1060,6 @@ def rotate_payment_connector_credentials( payment_connector_id: ID of the connector whose credentials to rotate secrets: The service-managed secrets to rotate (at least one). Accepts CoinbaseCdpSecret members or their string values. - connector_type: The connector's type. Defaults to CoinbaseCDP, currently the - only type that supports rotation. client_token: Optional idempotency token. If not provided, a UUID will be generated. Returns: @@ -1094,11 +1067,10 @@ def rotate_payment_connector_credentials( timestamp the rotation completed Raises: - ValueError: If the connector type does not support rotation, or if secrets is - empty or names an unknown secret - ClientError: If the rotation fails + ClientError: If the rotation fails, including a ValidationException if secrets + is empty or names a secret the service does not support """ - credentials_to_rotate = self._build_rotation_config_input(connector_type, secrets) + credentials_to_rotate = self._build_rotation_config_input(secrets) if client_token is None: client_token = str(uuid.uuid4()) diff --git a/tests/bedrock_agentcore/payments/test_client.py b/tests/bedrock_agentcore/payments/test_client.py index e2183103..eaa16ece 100644 --- a/tests/bedrock_agentcore/payments/test_client.py +++ b/tests/bedrock_agentcore/payments/test_client.py @@ -10,7 +10,6 @@ CoinbaseCdpSecret, PaymentClient, PaymentConnectorProvisionMode, - PaymentConnectorType, ) from bedrock_agentcore.payments.client import PaymentConnectorConfig @@ -1670,46 +1669,27 @@ class TestBuildRotationConfigInput: """Tests for _build_rotation_config_input static method.""" def test_coinbase_cdp_secrets(self): - """CoinbaseCDP produces the coinbaseCDP union member.""" + """Enum members are normalized into the coinbaseCDP union member.""" result = PaymentClient._build_rotation_config_input( - "CoinbaseCDP", [CoinbaseCdpSecret.API_KEY, CoinbaseCdpSecret.WALLET_SECRET] + [CoinbaseCdpSecret.API_KEY, CoinbaseCdpSecret.WALLET_SECRET] ) assert result == {"coinbaseCDP": {"secrets": ["API_KEY", "WALLET_SECRET"]}} - def test_accepts_enum_type(self): - """connector_type accepts a PaymentConnectorType member.""" - result = PaymentClient._build_rotation_config_input( - PaymentConnectorType.COINBASE_CDP, [CoinbaseCdpSecret.API_KEY] - ) - assert result == {"coinbaseCDP": {"secrets": ["API_KEY"]}} - def test_accepts_string_secrets(self): """Raw secret strings are accepted alongside enum members.""" - result = PaymentClient._build_rotation_config_input("CoinbaseCDP", ["WALLET_SECRET", CoinbaseCdpSecret.API_KEY]) + result = PaymentClient._build_rotation_config_input(["WALLET_SECRET", CoinbaseCdpSecret.API_KEY]) assert result == {"coinbaseCDP": {"secrets": ["WALLET_SECRET", "API_KEY"]}} - def test_duplicates_removed_order_preserved(self): - """Duplicate secrets are dropped, satisfying the model's uniqueItems constraint.""" + def test_secrets_passed_through_verbatim(self): + """Secrets are not filtered or de-duplicated; the service validates them.""" result = PaymentClient._build_rotation_config_input( - "CoinbaseCDP", - [CoinbaseCdpSecret.WALLET_SECRET, "WALLET_SECRET", CoinbaseCdpSecret.API_KEY], + [CoinbaseCdpSecret.WALLET_SECRET, "WALLET_SECRET", "ROOT_PASSWORD"] ) - assert result == {"coinbaseCDP": {"secrets": ["WALLET_SECRET", "API_KEY"]}} - - def test_empty_secrets_raises(self): - """An empty secrets list is rejected before the request is sent.""" - with pytest.raises(ValueError, match="at least one secret"): - PaymentClient._build_rotation_config_input("CoinbaseCDP", []) - - def test_unknown_secret_raises(self): - """An unrecognized secret name is rejected.""" - with pytest.raises(ValueError, match="Unsupported CoinbaseCDP secret"): - PaymentClient._build_rotation_config_input("CoinbaseCDP", ["ROOT_PASSWORD"]) + assert result == {"coinbaseCDP": {"secrets": ["WALLET_SECRET", "WALLET_SECRET", "ROOT_PASSWORD"]}} - def test_unsupported_connector_type_raises(self): - """StripePrivy has no rotatable service-managed secrets.""" - with pytest.raises(ValueError, match="not supported for connector type"): - PaymentClient._build_rotation_config_input(PaymentConnectorType.STRIPE_PRIVY, [CoinbaseCdpSecret.API_KEY]) + def test_empty_secrets_passed_through(self): + """An empty list is sent as-is; the model's length constraint rejects it server-side.""" + assert PaymentClient._build_rotation_config_input([]) == {"coinbaseCDP": {"secrets": []}} class TestRotatePaymentConnectorCredentials: @@ -1789,21 +1769,26 @@ def test_rotate_passes_client_token_through(self, mock_session, mock_boto3_clien @patch("bedrock_agentcore.payments.client.boto3.client") @patch("bedrock_agentcore.payments.client.boto3.Session") - def test_rotate_validates_before_calling_service(self, mock_session, mock_boto3_client): - """Invalid input raises ValueError without issuing a request.""" + def test_rotate_defers_validation_to_service(self, mock_session, mock_boto3_client): + """Secret names are not validated client-side; the service is the single source of truth.""" mock_session.return_value.region_name = "us-west-2" mock_cp_client = MagicMock() mock_boto3_client.return_value = mock_cp_client + mock_cp_client.rotate_payment_connector_credentials.side_effect = ClientError( + {"Error": {"Code": "ValidationException", "Message": "Unsupported secret"}}, + "RotatePaymentConnectorCredentials", + ) client = PaymentClient(region_name="us-west-2") - with pytest.raises(ValueError): + with pytest.raises(ClientError): client.rotate_payment_connector_credentials( payment_manager_id="pm-123", payment_connector_id="pc-123", - secrets=[], + secrets=["ROOT_PASSWORD"], ) - mock_cp_client.rotate_payment_connector_credentials.assert_not_called() + call_kwargs = mock_cp_client.rotate_payment_connector_credentials.call_args[1] + assert call_kwargs["credentialsToRotate"] == {"coinbaseCDP": {"secrets": ["ROOT_PASSWORD"]}} @patch("bedrock_agentcore.payments.client.boto3.client") @patch("bedrock_agentcore.payments.client.boto3.Session")