Repository: getmoto/moto
Pull request #8463 (2025-01-03 21:44:34)

Kinesis resource based policy implementation
Fixes https://github.com/getmoto/moto/issues/8465

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/getmoto/moto/pull/8463?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 92.62%. Comparing base [(`ec277aa`)](https://app.codecov.io/gh/getmoto/moto/commit/ec277aa25294839f5b9ce5cd346955050a4bbe8d?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) to head [(`30b0c7f`)](https://app.codecov.io/gh/getmoto/moto/commit/30b0c7fb8e961cf26cf0678770bba45e45c4d450?dropdown=coverage&el=desc&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    #8463   +/-   ##
=======================================
  Coverage   92.62%   92.62%           
=======================================
  Files        1225     1225           
  Lines      105655   105685   +30     
=======================================
+ Hits        97865    97895   +30     
  Misses       7790     7790           
```

| [Flag](https://app.codecov.io/gh/getmoto/moto/pull/8463/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/8463/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `27.75% <23.33%> (-0.01%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/getmoto/moto/pull/8463/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=getmoto) | `92.60% <100.00%> (+<0.01%)` | :arrow_up: |

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/8463?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).


Patch:
diff --git a/moto/kinesis/models.py b/moto/kinesis/models.py
index 51a4a21e0..f3492006b 100644
--- a/moto/kinesis/models.py
+++ b/moto/kinesis/models.py
@@ -585,6 +585,7 @@ class KinesisBackend(BaseBackend):
     def __init__(self, region_name: str, account_id: str):
         super().__init__(region_name, account_id)
         self.streams: Dict[str, Stream] = OrderedDict()
+        self.resource_policies: Dict[str, str] = {}
 
     @staticmethod
     def default_vpc_endpoint_service(
@@ -1025,5 +1026,35 @@ class KinesisBackend(BaseBackend):
             explicit_hash_key="",
         )
 
+    def put_resource_policy(self, resource_arn: str, policy_doc: str) -> None:
+        """
+        Creates/updates resource policy and return policy object
+        """
+        self.resource_policies[resource_arn] = policy_doc
+
+    def delete_resource_policy(self, resource_arn: str) -> None:
+        """
+        Remove resource policy with a matching given resource arn.
+        """
+        stream_name = resource_arn.split("/")[-1]
+        if stream_name not in self.streams:
+            raise StreamNotFoundError(resource_arn, self.account_id)
+        if resource_arn not in self.resource_policies:
+            raise ResourceNotFoundError(
+                message=f"No resource policy found for resource ARN {resource_arn}."
+            )
+        del self.resource_policies[resource_arn]
+
+    def get_resource_policy(self, resource_arn: str) -> str:
+        """
+        Get resource policy with a matching given resource arn.
+        """
+        stream_name = resource_arn.split("/")[-1]
+        if stream_name not in self.streams:
+            raise StreamNotFoundError(resource_arn, self.account_id)
+        if resource_arn not in self.resource_policies:
+            return "{}"
+        return self.resource_policies[resource_arn]
+
 
 kinesis_backends = BackendDict(KinesisBackend, "kinesis")
diff --git a/moto/kinesis/responses.py b/moto/kinesis/responses.py
index ba873a7ff..e4d46410a 100644
--- a/moto/kinesis/responses.py
+++ b/moto/kinesis/responses.py
@@ -325,3 +325,24 @@ class KinesisResponse(BaseResponse):
         stream_mode = self._get_param("StreamModeDetails")
         self.kinesis_backend.update_stream_mode(stream_arn, stream_mode)
         return "{}"
+
+    def put_resource_policy(self) -> str:
+        resource_arn = self._get_param("ResourceARN")
+        policy = self._get_param("Policy")
+        self.kinesis_backend.put_resource_policy(
+            resource_arn=resource_arn,
+            policy_doc=policy,
+        )
+        return json.dumps(dict())
+
+    def get_resource_policy(self) -> str:
+        resource_arn = self._get_param("ResourceARN")
+        policy = self.kinesis_backend.get_resource_policy(resource_arn=resource_arn)
+        return json.dumps({"Policy": policy})
+
+    def delete_resource_policy(self) -> str:
+        resource_arn = self._get_param("ResourceARN")
+        self.kinesis_backend.delete_resource_policy(
+            resource_arn=resource_arn,
+        )
+        return json.dumps(dict())


Test patch:
diff --git a/tests/test_kinesis/test_kinesis_resource_policy.py b/tests/test_kinesis/test_kinesis_resource_policy.py
new file mode 100644
index 000000000..d7c4e6fe4
--- /dev/null
+++ b/tests/test_kinesis/test_kinesis_resource_policy.py
@@ -0,0 +1,109 @@
+import json
+
+import boto3
+import pytest
+from botocore.exceptions import ClientError
+
+from moto import mock_aws
+
+
+def get_policy_doc(action, stream_arn):
+    sts = boto3.client("sts", "us-east-1")
+    account_id = sts.get_caller_identity()["Account"]
+
+    policy_doc = {
+        "Version": "2012-10-17",
+        "Statement": [
+            {
+                "Principal": {"AWS": account_id},
+                "Effect": "Allow",
+                "Action": [action],
+                "Resource": f"{stream_arn}",
+            }
+        ],
+    }
+    return json.dumps(policy_doc)
+
+
+@mock_aws
+def test_put_resource_policy():
+    client = boto3.client("kinesis", region_name="us-west-2")
+    client.create_stream(StreamName="my-stream", ShardCount=2)
+    resp = client.describe_stream(StreamName="my-stream")
+    desc = resp["StreamDescription"]
+    stream_arn = desc["StreamARN"]
+
+    action = "kinesis:*"
+    json_policy_doc = get_policy_doc(action, stream_arn)
+    client.put_resource_policy(ResourceARN=stream_arn, Policy=json_policy_doc)
+    response = client.get_resource_policy(ResourceARN=stream_arn)
+
+    assert response["Policy"] == json_policy_doc
+
+
+@mock_aws
+def test_delete_resource_policy():
+    client = boto3.client("kinesis", region_name="us-west-2")
+    client.create_stream(StreamName="my-stream", ShardCount=2)
+    resp = client.describe_stream(StreamName="my-stream")
+    desc = resp["StreamDescription"]
+    stream_arn = desc["StreamARN"]
+    action = "kinesis:*"
+    json_policy_doc = get_policy_doc(action, stream_arn)
+    client.put_resource_policy(ResourceARN=stream_arn, Policy=json_policy_doc)
+    response = client.get_resource_policy(ResourceARN=stream_arn)
+    assert response["Policy"] == json_policy_doc
+
+    client.delete_resource_policy(ResourceARN=stream_arn)
+    response = client.get_resource_policy(ResourceARN=stream_arn)
+    assert response["Policy"] == "{}"
+
+
+@mock_aws
+def test_get_resource_policy_from_unknown_resource():
+    client = boto3.client("kinesis", region_name="us-west-2")
+    client.create_stream(StreamName="my-stream", ShardCount=2)
+    resp = client.describe_stream(StreamName="my-stream")
+    desc = resp["StreamDescription"]
+    stream_arn = desc["StreamARN"] + "unknown"
+    sts = boto3.client("sts", "us-east-1")
+    account_id = sts.get_caller_identity()["Account"]
+    with pytest.raises(ClientError) as exc:
+        client.get_resource_policy(ResourceARN=stream_arn)
+    err = exc.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"
+    assert (
+        err["Message"] == f"Stream {stream_arn} under account {account_id} not found."
+    )
+
+
+@mock_aws
+def test_delete_resource_policy_from_unknown_resource():
+    client = boto3.client("kinesis", region_name="us-west-2")
+    client.create_stream(StreamName="my-stream", ShardCount=2)
+    resp = client.describe_stream(StreamName="my-stream")
+    desc = resp["StreamDescription"]
+    stream_arn = desc["StreamARN"] + "unknown"
+    sts = boto3.client("sts", "us-east-1")
+    account_id = sts.get_caller_identity()["Account"]
+    with pytest.raises(ClientError) as exc:
+        client.delete_resource_policy(ResourceARN=stream_arn)
+    err = exc.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"
+    assert (
+        err["Message"] == f"Stream {stream_arn} under account {account_id} not found."
+    )
+
+
+@mock_aws
+def test_delete_resource_policy_from_resource_without_policy():
+    client = boto3.client("kinesis", region_name="us-west-2")
+    client.create_stream(StreamName="my-stream", ShardCount=2)
+    resp = client.describe_stream(StreamName="my-stream")
+    desc = resp["StreamDescription"]
+    stream_arn = desc["StreamARN"]
+    with pytest.raises(ClientError) as exc:
+        client.delete_resource_policy(ResourceARN=stream_arn)
+    err = exc.value.response["Error"]
+    assert err["Code"] == "ResourceNotFoundException"
+    assert err["Message"] == f"No resource policy found for resource ARN {stream_arn}."