Skip to content

Commit e60ff8e

Browse files
dmacvicardmacvicar
authored
feat(kinesis): implement resource policies CRUD operations (#12488) (#12961)
1 parent 77a5785 commit e60ff8e

File tree

5 files changed

+266
-2
lines changed

5 files changed

+266
-2
lines changed

‎localstack-core/localstack/services/kinesis/models.py‎

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,18 @@
11
from collections import defaultdict
22

3-
from localstack.aws.api.kinesis import ConsumerDescription, MetricsName, StreamName
4-
from localstack.services.stores import AccountRegionBundle, BaseStore, LocalAttribute
3+
from localstack.aws.api.kinesis import (
4+
ConsumerDescription,
5+
MetricsName,
6+
Policy,
7+
ResourceARN,
8+
StreamName,
9+
)
10+
from localstack.services.stores import (
11+
AccountRegionBundle,
12+
BaseStore,
13+
CrossAccountAttribute,
14+
LocalAttribute,
15+
)
516

617

718
class KinesisStore(BaseStore):
@@ -13,5 +24,7 @@ class KinesisStore(BaseStore):
1324
default=lambda: defaultdict(set)
1425
)
1526

27+
resource_policies: dict[ResourceARN, Policy] = CrossAccountAttribute(default=dict)
28+
1629

1730
kinesis_stores = AccountRegionBundle("kinesis", KinesisStore)

‎localstack-core/localstack/services/kinesis/provider.py‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
1+
import json
12
import logging
23
import os
4+
import re
35
import time
46
from random import random
57

@@ -8,14 +10,18 @@
810
from localstack.aws.api.kinesis import (
911
ConsumerARN,
1012
Data,
13+
GetResourcePolicyOutput,
1114
HashKey,
1215
KinesisApi,
1316
PartitionKey,
17+
Policy,
1418
ProvisionedThroughputExceededException,
1519
PutRecordOutput,
1620
PutRecordsOutput,
1721
PutRecordsRequestEntryList,
1822
PutRecordsResultEntry,
23+
ResourceARN,
24+
ResourceNotFoundException,
1925
SequenceNumber,
2026
ShardId,
2127
StartingPosition,
@@ -24,6 +30,7 @@
2430
SubscribeToShardEvent,
2531
SubscribeToShardEventStream,
2632
SubscribeToShardOutput,
33+
ValidationException,
2734
)
2835
from localstack.aws.connect import connect_to
2936
from localstack.constants import LOCALHOST
@@ -39,6 +46,13 @@
3946
MAX_SUBSCRIPTION_SECONDS = 300
4047
SERVER_STARTUP_TIMEOUT = 120
4148

49+
DATA_STREAM_ARN_REGEX = re.compile(
50+
r"^arn:aws(?:-[a-z]+)*:kinesis:[a-z0-9-]+:\d{12}:stream\/[a-zA-Z0-9_.\-]+$"
51+
)
52+
CONSUMER_ARN_REGEX = re.compile(
53+
r"^arn:aws(?:-[a-z]+)*:kinesis:[a-z0-9-]+:\d{12}:stream\/[a-zA-Z0-9_.\-]+\/consumer\/[a-zA-Z0-9_.\-]+:\d+$"
54+
)
55+
4256

4357
def find_stream_for_consumer(consumer_arn):
4458
account_id = extract_account_id_from_arn(consumer_arn)
@@ -52,6 +66,11 @@ def find_stream_for_consumer(consumer_arn):
5266
raise Exception(f"Unable to find stream for stream consumer {consumer_arn}")
5367

5468

69+
def is_valid_kinesis_arn(resource_arn: ResourceARN) -> bool:
70+
"""Check if the provided ARN is a valid Kinesis ARN."""
71+
return bool(CONSUMER_ARN_REGEX.match(resource_arn) or DATA_STREAM_ARN_REGEX.match(resource_arn))
72+
73+
5574
class KinesisProvider(KinesisApi, ServiceLifecycleHook):
5675
server_manager: KinesisServerManager
5776

@@ -81,6 +100,60 @@ def get_forward_url(self, account_id: str, region_name: str) -> str:
81100
def get_store(account_id: str, region_name: str) -> KinesisStore:
82101
return kinesis_stores[account_id][region_name]
83102

103+
def put_resource_policy(
104+
self,
105+
context: RequestContext,
106+
resource_arn: ResourceARN,
107+
policy: Policy,
108+
**kwargs,
109+
) -> None:
110+
if not is_valid_kinesis_arn(resource_arn):
111+
raise ValidationException(f"invalid kinesis arn {resource_arn}")
112+
113+
kinesis = connect_to().kinesis
114+
try:
115+
kinesis.describe_stream_summary(StreamARN=resource_arn)
116+
except kinesis.exceptions.ResourceNotFoundException:
117+
raise ResourceNotFoundException(f"Stream with ARN {resource_arn} not found")
118+
119+
store = self.get_store(context.account_id, context.region)
120+
store.resource_policies[resource_arn] = policy
121+
122+
def get_resource_policy(
123+
self,
124+
context: RequestContext,
125+
resource_arn: ResourceARN,
126+
**kwargs,
127+
) -> GetResourcePolicyOutput:
128+
if not is_valid_kinesis_arn(resource_arn):
129+
raise ValidationException(f"invalid kinesis arn {resource_arn}")
130+
131+
kinesis = connect_to().kinesis
132+
try:
133+
kinesis.describe_stream_summary(StreamARN=resource_arn)
134+
except kinesis.exceptions.ResourceNotFoundException:
135+
raise ResourceNotFoundException(f"Stream with ARN {resource_arn} not found")
136+
137+
store = self.get_store(context.account_id, context.region)
138+
policy = store.resource_policies.get(resource_arn, json.dumps({}))
139+
return GetResourcePolicyOutput(Policy=policy)
140+
141+
def delete_resource_policy(
142+
self,
143+
context: RequestContext,
144+
resource_arn: ResourceARN,
145+
**kwargs,
146+
) -> None:
147+
if not is_valid_kinesis_arn(resource_arn):
148+
raise ValidationException(f"invalid kinesis arn {resource_arn}")
149+
150+
store = self.get_store(context.account_id, context.region)
151+
if resource_arn not in store.resource_policies:
152+
raise ResourceNotFoundException(
153+
f"No resource policy found for resource ARN {resource_arn}"
154+
)
155+
del store.resource_policies[resource_arn]
156+
84157
def subscribe_to_shard(
85158
self,
86159
context: RequestContext,

‎tests/aws/services/kinesis/test_kinesis.py‎

Lines changed: 89 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -134,6 +134,95 @@ def test_create_stream_without_shard_count(
134134

135135
snapshot.match("Shards", shards)
136136

137+
@markers.aws.validated
138+
def test_resource_policy_crud(
139+
self,
140+
account_id,
141+
kinesis_create_stream,
142+
wait_for_stream_ready,
143+
aws_client,
144+
snapshot,
145+
):
146+
"""Test complete CRUD cycle for Kinesis resource policies"""
147+
148+
stream_name = kinesis_create_stream()
149+
wait_for_stream_ready(stream_name)
150+
describe_stream = aws_client.kinesis.describe_stream(StreamName=stream_name)
151+
resource_arn = describe_stream["StreamDescription"]["StreamARN"]
152+
principal_arn = f"arn:aws:iam::{account_id}:root"
153+
154+
# retrieve, no policy yet
155+
resp = aws_client.kinesis.get_resource_policy(ResourceARN=resource_arn)
156+
snapshot.match("default_policy_if_not_set", resp)
157+
158+
# put
159+
policy = {
160+
"Version": "2012-10-17",
161+
"Statement": [
162+
{
163+
"Sid": "AllowCrossAccountWrite",
164+
"Effect": "Allow",
165+
"Principal": {"AWS": principal_arn},
166+
"Action": "kinesis:PutRecord",
167+
"Resource": resource_arn,
168+
}
169+
],
170+
}
171+
resp = aws_client.kinesis.put_resource_policy(
172+
ResourceARN=resource_arn, Policy=json.dumps(policy)
173+
)
174+
snapshot.match("put_resource_policy_if_not_set", resp)
175+
176+
# retrieve
177+
resp = aws_client.kinesis.get_resource_policy(ResourceARN=resource_arn)
178+
snapshot.match("get_resource_policy_after_set", resp)
179+
180+
# update
181+
updated_policy = {
182+
"Version": "2012-10-17",
183+
"Statement": [
184+
{
185+
"Sid": "AllowCrossAccountReadWrite",
186+
"Effect": "Allow",
187+
"Principal": {"AWS": principal_arn},
188+
"Action": ["kinesis:PutRecord", "kinesis:GetRecords"],
189+
"Resource": resource_arn,
190+
}
191+
],
192+
}
193+
resp = aws_client.kinesis.put_resource_policy(
194+
ResourceARN=resource_arn, Policy=json.dumps(updated_policy)
195+
)
196+
snapshot.match("update_resource_policy", resp)
197+
198+
# get the right policy after updating
199+
resp = aws_client.kinesis.get_resource_policy(ResourceARN=resource_arn)
200+
snapshot.match("get_resource_policy_after_update", resp)
201+
202+
# delete it
203+
resp = aws_client.kinesis.delete_resource_policy(ResourceARN=resource_arn)
204+
snapshot.match("delete_resource_policy", resp)
205+
206+
# get, policy should no longer exist
207+
resp = aws_client.kinesis.get_resource_policy(ResourceARN=resource_arn)
208+
snapshot.match("get_resource_policy_after_delete", resp)
209+
210+
# deleting non existent policy for a valid arn
211+
with pytest.raises(ClientError):
212+
aws_client.kinesis.delete_resource_policy(ResourceARN=resource_arn)
213+
214+
# put a policy for a non-existent stream
215+
not_existent_arn = "arn:aws:kinesis:us-east-1:000000000000:stream/non-existent-xxxxxx"
216+
policy["Statement"][0]["Resource"] = not_existent_arn
217+
with pytest.raises(ClientError):
218+
aws_client.kinesis.put_resource_policy(
219+
ResourceARN=not_existent_arn, Policy=json.dumps(policy)
220+
)
221+
222+
# TODO put a policy for an invalid stream arn, but it behaves differently
223+
# on localstack and AWS, as the later triggers end-point-resolution in botocore
224+
# and fails client side
225+
137226
@markers.aws.validated
138227
def test_stream_consumers(
139228
self,

‎tests/aws/services/kinesis/test_kinesis.snapshot.json‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -235,5 +235,85 @@
235235
"message": "Stream wrong-stream-name under account 111111111111 not found."
236236
}
237237
}
238+
},
239+
"tests/aws/services/kinesis/test_kinesis.py::TestKinesis::test_resource_policy_crud": {
240+
"recorded-date": "01-08-2025, 13:00:05",
241+
"recorded-content": {
242+
"default_policy_if_not_set": {
243+
"Policy": {},
244+
"ResponseMetadata": {
245+
"HTTPHeaders": {},
246+
"HTTPStatusCode": 200
247+
}
248+
},
249+
"put_resource_policy_if_not_set": {
250+
"ResponseMetadata": {
251+
"HTTPHeaders": {},
252+
"HTTPStatusCode": 200
253+
}
254+
},
255+
"get_resource_policy_after_set": {
256+
"Policy": {
257+
"Version": "2012-10-17",
258+
"Statement": [
259+
{
260+
"Sid": "AllowCrossAccountWrite",
261+
"Effect": "Allow",
262+
"Principal": {
263+
"AWS": "arn:<partition>:iam::111111111111:root"
264+
},
265+
"Action": "kinesis:PutRecord",
266+
"Resource": "arn:<partition>:kinesis:<region>:111111111111:<stream-name>"
267+
}
268+
]
269+
},
270+
"ResponseMetadata": {
271+
"HTTPHeaders": {},
272+
"HTTPStatusCode": 200
273+
}
274+
},
275+
"update_resource_policy": {
276+
"ResponseMetadata": {
277+
"HTTPHeaders": {},
278+
"HTTPStatusCode": 200
279+
}
280+
},
281+
"get_resource_policy_after_update": {
282+
"Policy": {
283+
"Version": "2012-10-17",
284+
"Statement": [
285+
{
286+
"Sid": "AllowCrossAccountReadWrite",
287+
"Effect": "Allow",
288+
"Principal": {
289+
"AWS": "arn:<partition>:iam::111111111111:root"
290+
},
291+
"Action": [
292+
"kinesis:PutRecord",
293+
"kinesis:GetRecords"
294+
],
295+
"Resource": "arn:<partition>:kinesis:<region>:111111111111:<stream-name>"
296+
}
297+
]
298+
},
299+
"ResponseMetadata": {
300+
"HTTPHeaders": {},
301+
"HTTPStatusCode": 200
302+
}
303+
},
304+
"delete_resource_policy": {
305+
"ResponseMetadata": {
306+
"HTTPHeaders": {},
307+
"HTTPStatusCode": 200
308+
}
309+
},
310+
"get_resource_policy_after_delete": {
311+
"Policy": {},
312+
"ResponseMetadata": {
313+
"HTTPHeaders": {},
314+
"HTTPStatusCode": 200
315+
}
316+
}
317+
}
238318
}
239319
}

‎tests/aws/services/kinesis/test_kinesis.validation.json‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,15 @@
2929
"tests/aws/services/kinesis/test_kinesis.py::TestKinesis::test_record_lifecycle_data_integrity": {
3030
"last_validated_date": "2022-08-25T10:39:44+00:00"
3131
},
32+
"tests/aws/services/kinesis/test_kinesis.py::TestKinesis::test_resource_policy_crud": {
33+
"last_validated_date": "2025-08-05T15:23:27+00:00",
34+
"durations_in_seconds": {
35+
"setup": 0.63,
36+
"call": 6.43,
37+
"teardown": 0.08,
38+
"total": 7.14
39+
}
40+
},
3241
"tests/aws/services/kinesis/test_kinesis.py::TestKinesis::test_stream_consumers": {
3342
"last_validated_date": "2022-08-26T08:23:46+00:00"
3443
},

0 commit comments

Comments
 (0)