diff --git a/prowler/providers/aws/services/kinesis/kinesis_service.py b/prowler/providers/aws/services/kinesis/kinesis_service.py
new file mode 100644
index 0000000000..3c49bc5370
--- /dev/null
+++ b/prowler/providers/aws/services/kinesis/kinesis_service.py
@@ -0,0 +1,70 @@
+from enum import Enum
+from typing import Optional
+
+from pydantic import BaseModel
+
+from prowler.lib.logger import logger
+from prowler.lib.scan_filters.scan_filters import is_resource_filtered
+from prowler.providers.aws.lib.service.service import AWSService
+
+
+class Kinesis(AWSService):
+    def __init__(self, provider):
+        # Call AWSService's __init__
+        super().__init__(__class__.__name__, provider)
+        self.streams = {}
+        self.__threading_call__(self._list_streams)
+        self.__threading_call__(self._list_tags_for_stream, self.streams.values())
+
+    def _list_streams(self, regional_client):
+        logger.info("Kinesis - Listing Kinesis Streams...")
+        try:
+            list_streams_paginator = regional_client.get_paginator("list_streams")
+            for page in list_streams_paginator.paginate():
+                for stream in page["StreamSummaries"]:
+                    arn = stream["StreamARN"]
+                    if not self.audit_resources or (
+                        is_resource_filtered(arn, self.audit_resources)
+                    ):
+                        self.streams[arn] = Stream(
+                            arn=arn,
+                            name=stream["StreamName"],
+                            region=regional_client.region,
+                            status=StreamStatus(stream.get("StreamStatus", "ACTIVE")),
+                        )
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+    def _list_tags_for_stream(self, stream):
+        logger.info(f"Kinesis - Listing tags for Stream {stream.name}...")
+        try:
+            stream.tags = (
+                self.regional_clients[stream.region]
+                .list_tags_for_stream(StreamName=stream.name)
+                .get("Tags", [])
+            )
+        except Exception as error:
+            logger.error(
+                f"{stream.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+
+class StreamStatus(Enum):
+    """Enum for Kinesis Stream Status"""
+
+    ACTIVE = "ACTIVE"
+    CREATING = "CREATING"
+    DELETING = "DELETING"
+    UPDATING = "UPDATING"
+
+
+class Stream(BaseModel):
+    """Model for Kinesis Stream"""
+
+    arn: str
+    region: str
+    name: str
+    status: StreamStatus
+    tags: Optional[list]
diff --git a/tests/providers/aws/services/kinesis/kinesis_service_test.py b/tests/providers/aws/services/kinesis/kinesis_service_test.py
new file mode 100644
index 0000000000..21dbd70804
--- /dev/null
+++ b/tests/providers/aws/services/kinesis/kinesis_service_test.py
@@ -0,0 +1,74 @@
+from unittest.mock import patch
+
+import botocore
+from moto import mock_aws
+
+from prowler.providers.aws.services.kinesis.kinesis_service import Kinesis, StreamStatus
+from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider
+
+make_api_call = botocore.client.BaseClient._make_api_call
+
+
+def mock_make_api_call(self, operation_name, kwarg):
+    if operation_name == "ListStreams":
+        return {
+            "StreamNames": ["test-stream"],
+            "StreamSummaries": [
+                {
+                    "StreamName": "test-stream",
+                    "StreamARN": "arn:aws:kinesis:us-east-1:123456789012:stream/test-stream",
+                    "StreamStatus": "ACTIVE",
+                }
+            ],
+        }
+    if operation_name == "DescribeStream":
+        return {
+            "StreamDescription": {
+                "StreamName": "test-stream",
+                "StreamARN": "arn:aws:kinesis:us-east-1:123456789012:stream/test-stream",
+                "StreamStatus": "ACTIVE",
+                "Tags": [{"Key": "test_tag", "Value": "test_value"}],
+            }
+        }
+    if operation_name == "ListTagsForStream":
+        return {"Tags": [{"Key": "test_tag", "Value": "test_value"}]}
+    return make_api_call(self, operation_name, kwarg)
+
+
+# Patch every AWS call using Boto3
+@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
+class Test_Kinesis_Service:
+    # Test Kinesis Client
+    @mock_aws
+    def test_get_client(self):
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+        kinesis = Kinesis(aws_provider)
+        assert (
+            kinesis.regional_clients[AWS_REGION_US_EAST_1].__class__.__name__
+            == "Kinesis"
+        )
+
+    # Test Kinesis Session
+    def test__get_session__(self):
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+        kinesis = Kinesis(aws_provider)
+        assert kinesis.session.__class__.__name__ == "Session"
+
+    # Test Kinesis Service
+    @mock_aws
+    def test__get_service__(self):
+        kinesis = Kinesis(set_mocked_aws_provider())
+        assert kinesis.service == "kinesis"
+
+    @mock_aws
+    def test_list_streamscomplete(self):
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+        kinesis = Kinesis(aws_provider)
+
+        arn = "arn:aws:kinesis:us-east-1:123456789012:stream/test-stream"
+        assert len(kinesis.streams) == 1
+        assert kinesis.streams[arn].name == "test-stream"
+        assert kinesis.streams[arn].status == StreamStatus.ACTIVE
+        assert kinesis.streams[arn].tags == [{"Key": "test_tag", "Value": "test_value"}]
+        assert kinesis.streams[arn].region == AWS_REGION_US_EAST_1
+        assert kinesis.streams[arn].arn == arn