SecretsManager - put/delete_resource_policy (#6049)

IMPLEMENTATION_COVERAGE.md

@@ -6055,11 +6055,11 @@
 
 ## secretsmanager
 <details>
-<summary>72% implemented</summary>
+<summary>81% implemented</summary>
 
 - [X] cancel_rotate_secret
 - [X] create_secret
-- [ ] delete_resource_policy
+- [X] delete_resource_policy
 - [X] delete_secret
 - [X] describe_secret
 - [X] get_random_password
@@ -6067,7 +6067,7 @@
 - [X] get_secret_value
 - [X] list_secret_version_ids
 - [X] list_secrets
-- [ ] put_resource_policy
+- [X] put_resource_policy
 - [X] put_secret_value
 - [ ] remove_regions_from_replication
 - [ ] replicate_secret_to_regions

docs/docs/services/secretsmanager.rst

@@ -27,7 +27,7 @@ secretsmanager
 
 - [X] cancel_rotate_secret
 - [X] create_secret
-- [ ] delete_resource_policy
+- [X] delete_resource_policy
 - [X] delete_secret
 - [X] describe_secret
 - [X] get_random_password
@@ -35,7 +35,11 @@ secretsmanager
 - [X] get_secret_value
 - [X] list_secret_version_ids
 - [X] list_secrets
-- [ ] put_resource_policy
+- [X] put_resource_policy
+  
+        The BlockPublicPolicy-parameter is not yet implemented
+        
+
 - [X] put_secret_value
 - [ ] remove_regions_from_replication
 - [ ] replicate_secret_to_regions

moto/secretsmanager/models.py

@@ -87,6 +87,7 @@ class FakeSecret:
         self.rotation_lambda_arn = ""
         self.auto_rotate_after_days = 0
         self.deleted_date = None
+        self.policy = None
 
     def update(
         self, description=None, tags=None, kms_key_id=None, last_changed_date=None
@@ -825,29 +826,37 @@ class SecretsManagerBackend(BaseBackend):
 
         return secret_id
 
-    @staticmethod
+    def put_resource_policy(self, secret_id: str, policy: str):
-    def get_resource_policy(secret_id):
+        """
-        resource_policy = {
+        The BlockPublicPolicy-parameter is not yet implemented
-            "Version": "2012-10-17",
+        """
-            "Statement": {
+        if not self._is_valid_identifier(secret_id):
-                "Effect": "Allow",
+            raise SecretNotFoundException()
-                "Principal": {
+
-                    "AWS": [
+        secret = self.secrets[secret_id]
-                        "arn:aws:iam::111122223333:root",
+        secret.policy = policy
-                        "arn:aws:iam::444455556666:root",
+        return secret.arn, secret.name
-                    ]
+
-                },
+    def get_resource_policy(self, secret_id):
-                "Action": ["secretsmanager:GetSecretValue"],
+        if not self._is_valid_identifier(secret_id):
-                "Resource": "*",
+            raise SecretNotFoundException()
-            },
+
+        secret = self.secrets[secret_id]
+        resp = {
+            "ARN": secret.arn,
+            "Name": secret.name,
         }
-        return json.dumps(
+        if secret.policy is not None:
-            {
+            resp["ResourcePolicy"] = secret.policy
-                "ARN": secret_id,
+        return json.dumps(resp)
-                "Name": secret_id,
+
-                "ResourcePolicy": json.dumps(resource_policy),
+    def delete_resource_policy(self, secret_id):
-            }
+        if not self._is_valid_identifier(secret_id):
-        )
+            raise SecretNotFoundException()
+
+        secret = self.secrets[secret_id]
+        secret.policy = None
+        return secret.arn, secret.name
 
 
 secretsmanager_backends = BackendDict(SecretsManagerBackend, "secretsmanager")

moto/secretsmanager/responses.py

@@ -174,6 +174,17 @@ class SecretsManagerResponse(BaseResponse):
         secret_id = self._get_param("SecretId")
         return self.backend.get_resource_policy(secret_id=secret_id)
 
+    def put_resource_policy(self):
+        secret_id = self._get_param("SecretId")
+        policy = self._get_param("ResourcePolicy")
+        arn, name = self.backend.put_resource_policy(secret_id, policy)
+        return json.dumps(dict(ARN=arn, Name=name))
+
+    def delete_resource_policy(self):
+        secret_id = self._get_param("SecretId")
+        arn, name = self.backend.delete_resource_policy(secret_id)
+        return json.dumps(dict(ARN=arn, Name=name))
+
     def tag_resource(self):
         secret_id = self._get_param("SecretId")
         tags = self._get_param("Tags", if_none=[])

tests/terraformtests/terraform-tests.success.txt

@@ -452,6 +452,14 @@ s3:
   - TestAccS3ObjectsDataSource_fetchOwner
 sagemaker:
   - TestAccSageMakerPrebuiltECRImageDataSource
+secretsmanager:
+  - TestAccSecretsManagerSecretDataSource_basic
+  - TestAccSecretsManagerSecretPolicy_
+  - TestAccSecretsManagerSecret_RecoveryWindowInDays_recreate
+  - TestAccSecretsManagerSecret_tags
+  - TestAccSecretsManagerSecret_kmsKeyID
+  - TestAccSecretsManagerSecret_withNamePrefix
+  - TestAccSecretsManagerSecret_rotationRules
 servicediscovery:
   - TestAccServiceDiscoveryDNSNamespaceDataSource
   - TestAccServiceDiscoveryHTTPNamespace

tests/test_secretsmanager/test_policy.py

@@ -0,0 +1,77 @@
+import boto3
+import json
+import pytest
+
+from botocore.exceptions import ClientError
+from moto import mock_secretsmanager
+
+
+@mock_secretsmanager
+def test_get_initial_policy():
+    client = boto3.client("secretsmanager", region_name="us-west-2")
+    client.create_secret(Name="test-secret")
+
+    resp = client.get_resource_policy(SecretId="test-secret")
+    assert resp.get("Name") == "test-secret"
+    assert "ARN" in resp
+    assert "ResourcePolicy" not in resp
+
+
+@mock_secretsmanager
+def test_put_resource_policy():
+    client = boto3.client("secretsmanager", region_name="us-west-2")
+    client.create_secret(Name="test-secret")
+
+    policy = {
+        "Statement": [
+            {
+                "Action": "secretsmanager:GetSecretValue",
+                "Effect": "Allow",
+                "Principal": {
+                    "AWS": "arn:aws:iam::123456789012:role/tf-acc-test-655046176950657276"
+                },
+                "Resource": "*",
+                "Sid": "EnableAllPermissions",
+            }
+        ],
+        "Version": "2012-10-17",
+    }
+    resp = client.put_resource_policy(
+        SecretId="test-secret", ResourcePolicy=json.dumps(policy)
+    )
+    assert "ARN" in resp
+    assert "Name" in resp
+
+    resp = client.get_resource_policy(SecretId="test-secret")
+    assert "ResourcePolicy" in resp
+    assert json.loads(resp["ResourcePolicy"]) == policy
+
+
+@mock_secretsmanager
+def test_delete_resource_policy():
+    client = boto3.client("secretsmanager", region_name="us-west-2")
+    client.create_secret(Name="test-secret")
+
+    client.put_resource_policy(SecretId="test-secret", ResourcePolicy="some policy")
+
+    client.delete_resource_policy(SecretId="test-secret")
+
+    resp = client.get_resource_policy(SecretId="test-secret")
+    assert "ResourcePolicy" not in resp
+
+
+@mock_secretsmanager
+def test_policies_for_unknown_secrets():
+    client = boto3.client("secretsmanager", region_name="us-west-2")
+
+    with pytest.raises(ClientError) as exc:
+        client.put_resource_policy(SecretId="unknown secret", ResourcePolicy="p")
+    assert exc.value.response["Error"]["Code"] == "ResourceNotFoundException"
+
+    with pytest.raises(ClientError) as exc:
+        client.get_resource_policy(SecretId="unknown secret")
+    assert exc.value.response["Error"]["Code"] == "ResourceNotFoundException"
+
+    with pytest.raises(ClientError) as exc:
+        client.delete_resource_policy(SecretId="unknown secret")
+    assert exc.value.response["Error"]["Code"] == "ResourceNotFoundException"