Back to Checks
Kinesis stream is encrypted at rest with KMS
kinesis_stream_encrypted_at_rest
Service: kinesisSeverity: high
OfficialBuilt-inVerified
Categories:encryption
Resource Type:AwsKinesisStream
Amazon Kinesis Data Streams with server-side encryption use AWS KMS to protect records at rest. The evaluation determines whether a stream has SSE-KMS configured with a KMS key; streams lacking KMS-based at rest encryption are identified.
Risk
Without SSE-KMS, records in shards may be exposed in plaintext if storage, backups, or analytics exports are accessed, undermining confidentiality. Absence of KMS controls also reduces integrity and oversight by removing key policies, rotation, and audit trails-enabling covert data exfiltration or insider misuse.
Run this check with Prowler
Run in Prowler CloudRun
prowler aws --checks kinesis_stream_encrypted_at_restRecommendation
Enable SSE-KMS on all streams.
- Use customer-managed keys for rotation and ownership
- Enforce least privilege on KMS grants; limit cross-account use
- Monitor key usage and require encryption in CI/CD
Remediation
CLI
aws kinesis start-stream-encryption --stream-name <KINESIS_STREAM_NAME> --encryption-type KMS --key-id alias/aws/kinesisTerraform
# Enable KMS encryption on a Kinesis stream
resource "aws_kinesis_stream" "<example_resource_name>" {
name = "<example_resource_name>"
shard_count = 1
encryption_type = "KMS" # Critical: enables KMS encryption at rest
kms_key_id = "alias/aws/kinesis" # Critical: uses AWS managed Kinesis KMS key
}NativeIaC
# CloudFormation: enable KMS encryption on a Kinesis stream
Resources:
<example_resource_name>:
Type: AWS::Kinesis::Stream
Properties:
ShardCount: 1
StreamEncryption:
EncryptionType: KMS # Critical: enables KMS encryption at rest
KeyId: alias/aws/kinesis # Critical: uses AWS managed Kinesis KMS keyOther
1Open the AWS Console and go to Amazon Kinesis > Data streams
2Select the stream
3On the Details tab, click Edit in Server-side encryption
4Select Enabled
5Choose the (Default) aws/kinesis KMS key
6Click Save
Check Code
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.kinesis.kinesis_client import kinesis_client
from prowler.providers.aws.services.kinesis.kinesis_service import EncryptionType
class kinesis_stream_encrypted_at_rest(Check):
def execute(self):
findings = []
for stream in kinesis_client.streams.values():
report = Check_Report_AWS(metadata=self.metadata(), resource=stream)
report.status = "FAIL"
report.status_extended = (
f"Kinesis Stream {stream.name} is not encrypted at rest."
)
if stream.encrypted_at_rest == EncryptionType.KMS:
report.status = "PASS"
report.status_extended = (
f"Kinesis Stream {stream.name} is encrypted at rest."
)
findings.append(report)
return findings