Repository: getmoto/moto
Pull request #7519 (2024-03-25 04:21:31)

SecretsManager: Add BatchGetSecretValue implementation
Hi all, I've been working on some stuff these days and stumbled upon this [issue](https://github.com/localstack/localstack/issues/9982) on the LocalStack board. 
So, I decided to contribute by providing an implementation for the batch_get_secret_value feature.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/getmoto/moto/pull/7519?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) Report
Attention: Patch coverage is `95.55556%` with `2 lines` in your changes are missing coverage. Please review.
> Project coverage is 94.07%. Comparing base [(`9aef694`)](https://app.codecov.io/gh/getmoto/moto/commit/9aef6944da6a6f1b84be99cefb34f1724e6ca58f?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) to head [(`819daea`)](https://app.codecov.io/gh/getmoto/moto/pull/7519?dropdown=coverage&src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).
> Report is 137 commits behind head on master.

| [Files](https://app.codecov.io/gh/getmoto/moto/pull/7519?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | Patch % | Lines |
|---|---|---|
| [moto/secretsmanager/models.py](https://app.codecov.io/gh/getmoto/moto/pull/7519?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto#diff-bW90by9zZWNyZXRzbWFuYWdlci9tb2RlbHMucHk=) | 94.73% | [2 Missing :warning: ](https://app.codecov.io/gh/getmoto/moto/pull/7519?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #7519      +/-   ##
==========================================
- Coverage   95.88%   94.07%   -1.81%     
==========================================
  Files         843     1070     +227     
  Lines       82578    91863    +9285     
==========================================
+ Hits        79178    86419    +7241     
- Misses       3400     5444    +2044     
```

| [Flag](https://app.codecov.io/gh/getmoto/moto/pull/7519/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | Coverage Δ | |
|---|---|---|
| [servertests](https://app.codecov.io/gh/getmoto/moto/pull/7519/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `29.56% <8.88%> (-2.96%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/getmoto/moto/pull/7519/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `94.04% <95.55%> (-1.81%)` | :arrow_down: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto#carryforward-flags-in-the-pull-request-comment) to find out more.


</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/getmoto/moto/pull/7519?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto).

hpaskalev: The missing tests codecov was complaining about should be now in place.

Patch:
diff --git a/moto/secretsmanager/models.py b/moto/secretsmanager/models.py
index 036501145..6f38516aa 100644
--- a/moto/secretsmanager/models.py
+++ b/moto/secretsmanager/models.py
@@ -28,6 +28,8 @@ from .list_secrets.filters import (
 )
 from .utils import get_secret_name_from_partial_arn, random_password, secret_arn
 
+MAX_RESULTS_DEFAULT = 100
+
 
 def filter_primary_region(secret: "FakeSecret", values: List[str]) -> bool:
     if isinstance(secret, FakeSecret):
@@ -470,6 +472,53 @@ class SecretsManagerBackend(BaseBackend):
 
         return response_data
 
+    def batch_get_secret_value(
+        self,
+        secret_id_list: Optional[List[str]] = None,
+        filters: Optional[List[Dict[str, Any]]] = None,
+        max_results: Optional[int] = None,
+        next_token: Optional[str] = None,
+    ) -> Tuple[List[Dict[str, Any]], List[Any], Optional[str]]:
+        secret_list = []
+        errors: List[Any] = []
+        if secret_id_list and filters:
+            raise InvalidParameterException(
+                "Either 'SecretIdList' or 'Filters' must be provided, but not both."
+            )
+
+        if max_results and not filters:
+            raise InvalidParameterException(
+                "'Filters' not specified. 'Filters' must also be specified when 'MaxResults' is provided."
+            )
+
+        if secret_id_list:
+            for secret_id in secret_id_list:
+                # TODO perhaps there should be a check if the secret id is valid identifier
+                # and add an error to the list if not
+                try:
+                    # TODO investigate the behaviour when the secret doesn't exist or has been deleted,
+                    # might need to add an error to the list
+                    secret_list.append(self.get_secret_value(secret_id, "", ""))
+                except (SecretNotFoundException, InvalidRequestException):
+                    pass
+
+        if filters:
+            for secret in self.secrets.values():
+                if _matches(secret, filters):
+                    if isinstance(secret, FakeSecret):
+                        secret_list.append(
+                            self.get_secret_value(secret.secret_id, "", "")
+                        )
+                    elif isinstance(secret, ReplicaSecret):
+                        secret_list.append(
+                            self.get_secret_value(secret.source.secret_id, "", "")
+                        )
+
+        secret_page, new_next_token = self._get_secret_values_page_and_next_token(
+            secret_list, max_results, next_token
+        )
+        return secret_page, errors, new_next_token
+
     def update_secret(
         self,
         secret_id: str,
@@ -850,20 +899,17 @@ class SecretsManagerBackend(BaseBackend):
     def list_secrets(
         self,
         filters: List[Dict[str, Any]],
-        max_results: int = 100,
+        max_results: int = MAX_RESULTS_DEFAULT,
         next_token: Optional[str] = None,
     ) -> Tuple[List[Dict[str, Any]], Optional[str]]:
-        secret_list = []
+        secret_list: List[Dict[str, Any]] = []
         for secret in self.secrets.values():
             if _matches(secret, filters):
                 secret_list.append(secret.to_dict())
 
-        starting_point = int(next_token or 0)
-        ending_point = starting_point + int(max_results or 100)
-        secret_page = secret_list[starting_point:ending_point]
-        new_next_token = str(ending_point) if ending_point < len(secret_list) else None
-
-        return secret_page, new_next_token
+        return self._get_secret_values_page_and_next_token(
+            secret_list, max_results, next_token
+        )
 
     def delete_secret(
         self,
@@ -1094,5 +1140,18 @@ class SecretsManagerBackend(BaseBackend):
         statuses = [replica.config for replica in secret.replicas]
         return secret_id, statuses
 
+    def _get_secret_values_page_and_next_token(
+        self,
+        secret_list: List[Dict[str, Any]],
+        max_results: Optional[int],
+        next_token: Optional[str],
+    ) -> Tuple[List[Dict[str, Any]], Optional[str]]:
+        starting_point = int(next_token or 0)
+        ending_point = starting_point + int(max_results or MAX_RESULTS_DEFAULT)
+        secret_page = secret_list[starting_point:ending_point]
+        new_next_token = str(ending_point) if ending_point < len(secret_list) else None
+
+        return secret_page, new_next_token
+
 
 secretsmanager_backends = BackendDict(SecretsManagerBackend, "secretsmanager")
diff --git a/moto/secretsmanager/responses.py b/moto/secretsmanager/responses.py
index f1b107408..72a7403fa 100644
--- a/moto/secretsmanager/responses.py
+++ b/moto/secretsmanager/responses.py
@@ -49,6 +49,22 @@ class SecretsManagerResponse(BaseResponse):
         )
         return json.dumps(value)
 
+    def batch_get_secret_value(self) -> str:
+        secret_id_list = self._get_param("SecretIdList", if_none=[])
+        filters = self._get_param("Filters", if_none=[])
+        max_results = self._get_int_param("MaxResults")
+        next_token = self._get_param("NextToken")
+
+        secret_values, errors, next_token = self.backend.batch_get_secret_value(
+            secret_id_list=secret_id_list,
+            filters=filters,
+            max_results=max_results,
+            next_token=next_token,
+        )
+        return json.dumps(
+            dict(SecretValues=secret_values, Errors=errors, NextToken=next_token)
+        )
+
     def create_secret(self) -> str:
         name = self._get_param("Name")
         secret_string = self._get_param("SecretString")


Test patch:
diff --git a/tests/test_secretsmanager/test_secretsmanager.py b/tests/test_secretsmanager/test_secretsmanager.py
index 8d2d1f7ff..e386aff25 100644
--- a/tests/test_secretsmanager/test_secretsmanager.py
+++ b/tests/test_secretsmanager/test_secretsmanager.py
@@ -181,6 +181,127 @@ def test_get_secret_version_stage_mismatch():
     ) == cm.value.response["Error"]["Message"]
 
 
+@mock_aws
+def test_batch_get_secret_value_for_secret_id_list_with_matches():
+    conn = boto3.client("secretsmanager", region_name="us-west-2")
+
+    secret_a = conn.create_secret(Name="test-secret-a", SecretString="secret")
+    secret_b = conn.create_secret(Name="test-secret-b", SecretString="secret")
+
+    secrets_batch = conn.batch_get_secret_value(
+        SecretIdList=["test-secret-a", "test-secret-b"]
+    )
+    matched = [
+        secret
+        for secret in secrets_batch["SecretValues"]
+        if secret["ARN"] in [secret_a["ARN"], secret_b["ARN"]]
+    ]
+
+    assert len(matched) == len(secrets_batch["SecretValues"]) == 2
+
+
+@mock_aws
+def test_batch_get_secret_value_for_secret_id_list_without_matches():
+    conn = boto3.client("secretsmanager", region_name="us-west-2")
+
+    conn.create_secret(Name="test-secret-a", SecretString="secret")
+
+    secrets_batch = conn.batch_get_secret_value(
+        SecretIdList=["test-secret-b", "test-secret-c"]
+    )
+    assert len(secrets_batch["SecretValues"]) == 0
+
+
+@mock_aws
+def test_batch_get_secret_value_with_filters():
+    conn = boto3.client("secretsmanager", region_name="us-west-2")
+
+    secret_a = conn.create_secret(Name="test-secret-a", SecretString="secret")
+    secret_b = conn.create_secret(Name="test-secret-b", SecretString="secret")
+    conn.create_secret(Name="test-secret-c", SecretString="secret")
+
+    secrets_batch = conn.batch_get_secret_value(
+        Filters=[{"Key": "name", "Values": [secret_a["Name"], secret_b["Name"]]}]
+    )
+
+    assert [sec["ARN"] for sec in secrets_batch["SecretValues"]] == [
+        secret_a["ARN"],
+        secret_b["ARN"],
+    ]
+
+
+@mock_aws
+def test_batch_get_secret_value_with_both_secret_id_list_and_filters():
+    conn = boto3.client("secretsmanager", region_name="us-west-2")
+
+    with pytest.raises(ClientError) as exc:
+        conn.batch_get_secret_value(
+            Filters=[{"Key": "name", "Values": ["test-secret-a", "test-secret-b"]}],
+            SecretIdList=["foo", "bar"],
+        )
+
+    err = exc.value.response["Error"]
+    assert err["Code"] == "InvalidParameterException"
+    assert (
+        "Either 'SecretIdList' or 'Filters' must be provided, but not both."
+        in err["Message"]
+    )
+
+
+@mock_aws
+def test_batch_get_secret_value_with_max_results_and_no_filters():
+    conn = boto3.client("secretsmanager", region_name="us-west-2")
+
+    conn = boto3.client("secretsmanager", region_name="us-west-2")
+    with pytest.raises(ClientError) as exc:
+        conn.batch_get_secret_value(MaxResults=10, SecretIdList=["foo", "bar"])
+
+    err = exc.value.response["Error"]
+    assert err["Code"] == "InvalidParameterException"
+    assert (
+        "'Filters' not specified. 'Filters' must also be specified when 'MaxResults' is provided."
+        in err["Message"]
+    )
+
+
+@mock_aws
+def test_batch_get_secret_value_binary():
+    conn = boto3.client("secretsmanager", region_name="us-west-2")
+
+    secret_a = conn.create_secret(Name="test-secret-a", SecretBinary="secretA")
+    secret_b = conn.create_secret(Name="test-secret-b", SecretBinary="secretB")
+    conn.create_secret(Name="test-secret-c", SecretBinary="secretC")
+
+    secrets_batch = conn.batch_get_secret_value(
+        Filters=[{"Key": "name", "Values": [secret_a["Name"], secret_b["Name"]]}]
+    )
+    matched = [
+        secret
+        for secret in secrets_batch["SecretValues"]
+        if secret["ARN"] == secret_a["ARN"]
+        and secret["SecretBinary"] == b"secretA"
+        or secret["ARN"] == secret_b["ARN"]
+        and secret["SecretBinary"] == b"secretB"
+    ]
+    assert len(matched) == 2
+
+
+@mock_aws
+def test_batch_get_secret_value_missing_value():
+    conn = boto3.client("secretsmanager", region_name="us-east-2")
+
+    secret_a = conn.create_secret(Name="test-secret-a")
+    secret_b = conn.create_secret(Name="test-secret-b")
+
+    with pytest.raises(ClientError) as exc:
+        conn.batch_get_secret_value(
+            Filters=[{"Key": "name", "Values": [secret_a["Name"], secret_b["Name"]]}]
+        )
+
+    err = exc.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"
+
+
 @mock_aws
 def test_create_secret():
     conn = boto3.client("secretsmanager", region_name="us-east-1")
diff --git a/tests/test_secretsmanager/test_server.py b/tests/test_secretsmanager/test_server.py
index d6b5ae8ec..43c90388a 100644
--- a/tests/test_secretsmanager/test_server.py
+++ b/tests/test_secretsmanager/test_server.py
@@ -912,6 +912,120 @@ def test_update_secret_version_stage_validation():
     assert resp.status_code == 400
 
 
+@mock_aws
+def test_batch_get_secret_value_for_secret_id_list_with_matches():
+    backend = server.create_backend_app("secretsmanager")
+    test_client = backend.test_client()
+
+    test_client.post(
+        "/",
+        data={"Name": "db/username", "SecretString": "foo"},
+        headers={"X-Amz-Target": "secretsmanager.CreateSecret"},
+    )
+
+    test_client.post(
+        "/",
+        data={"Name": "db/password", "SecretString": "foo-password"},
+        headers={"X-Amz-Target": "secretsmanager.CreateSecret"},
+    )
+
+    batch_get_secret_values = test_client.post(
+        "/",
+        json={"SecretIdList": ["db/username", "db/password"]},
+        headers={"X-Amz-Target": "secretsmanager.BatchGetSecretValue"},
+    )
+
+    json_data = json.loads(batch_get_secret_values.data.decode("utf-8"))
+    matched = [
+        s
+        for s in json_data["SecretValues"]
+        if (s["Name"] == "db/username" and s["SecretString"] == "foo")
+        or (s["Name"] == "db/password" and s["SecretString"] == "foo-password")
+    ]
+    assert len(matched) == len(json_data["SecretValues"]) == 2
+
+
+@mock_aws
+def test_batch_get_secret_value_for_secret_id_list_with_no_matches():
+    backend = server.create_backend_app("secretsmanager")
+    test_client = backend.test_client()
+
+    test_client.post(
+        "/",
+        data={"Name": "db/foo", "SecretString": "bar"},
+        headers={"X-Amz-Target": "secretsmanager.CreateSecret"},
+    )
+
+    batch_get_secret_values = test_client.post(
+        "/",
+        json={"SecretIdList": ["db/username", "db/password"]},
+        headers={"X-Amz-Target": "secretsmanager.BatchGetSecretValue"},
+    )
+
+    json_data = json.loads(batch_get_secret_values.data.decode("utf-8"))
+    assert len(json_data["SecretValues"]) == 0
+
+
+@mock_aws
+def test_batch_get_secret_value_with_both_secret_id_list_and_filters():
+    backend = server.create_backend_app("secretsmanager")
+    test_client = backend.test_client()
+
+    batch_get_secret_values = test_client.post(
+        "/",
+        json={
+            "SecretIdList": ["db/username", "db/password"],
+            "Filters": [{"Key": "description", "Values": ["foo"]}],
+        },
+        headers={"X-Amz-Target": "secretsmanager.BatchGetSecretValue"},
+    )
+
+    json_data = json.loads(batch_get_secret_values.data.decode("utf-8"))
+    assert (
+        json_data["message"]
+        == "Either 'SecretIdList' or 'Filters' must be provided, but not both."
+    )
+    assert json_data["__type"] == "InvalidParameterException"
+    assert batch_get_secret_values.status_code == 400
+
+
+@mock_aws
+def test_batch_get_secret_value_with_filters():
+    backend = server.create_backend_app("secretsmanager")
+    test_client = backend.test_client()
+
+    test_client.post(
+        "/",
+        data={"Name": "db/foo", "SecretString": "bar", "Description": "foo-secret"},
+        headers={"X-Amz-Target": "secretsmanager.CreateSecret"},
+    )
+
+    test_client.post(
+        "/",
+        data={"Name": "db/bar", "SecretString": "foo", "Description": "foo-secret"},
+        headers={"X-Amz-Target": "secretsmanager.CreateSecret"},
+    )
+
+    batch_get_secret_values = test_client.post(
+        "/",
+        json={
+            "Filters": [{"Key": "description", "Values": ["foo-secret"]}],
+        },
+        headers={"X-Amz-Target": "secretsmanager.BatchGetSecretValue"},
+    )
+
+    json_data = json.loads(batch_get_secret_values.data.decode("utf-8"))
+    matched = [
+        s
+        for s in json_data["SecretValues"]
+        if (s["Name"] == "db/foo" and s["SecretString"] == "bar")
+        or (s["Name"] == "db/bar" and s["SecretString"] == "foo")
+    ]
+
+    json_data = json.loads(batch_get_secret_values.data.decode("utf-8"))
+    assert len(json_data["SecretValues"]) == len(matched) == 2
+
+
 #
 # The following tests should work, but fail on the embedded dict in
 # RotationRules. The error message suggests a problem deeper in the code, which