diff --git a/internal/app/cfsec/adapter/aws/adapt.go b/internal/app/cfsec/adapter/aws/adapt.go
index 2f842ab8..209702dc 100644
--- a/internal/app/cfsec/adapter/aws/adapt.go
+++ b/internal/app/cfsec/adapter/aws/adapt.go
@@ -21,6 +21,7 @@ import (
 	"github.com/aquasecurity/cfsec/internal/app/cfsec/adapter/aws/elasticsearch"
 	"github.com/aquasecurity/cfsec/internal/app/cfsec/adapter/aws/elb"
 	"github.com/aquasecurity/cfsec/internal/app/cfsec/adapter/aws/iam"
+	"github.com/aquasecurity/cfsec/internal/app/cfsec/adapter/aws/msk"
 	"github.com/aquasecurity/cfsec/internal/app/cfsec/adapter/aws/neptune"
 	"github.com/aquasecurity/cfsec/internal/app/cfsec/adapter/aws/rds"
 	"github.com/aquasecurity/cfsec/internal/app/cfsec/adapter/aws/redshift"
@@ -56,6 +57,7 @@ func Adapt(cfFile parser.FileContext) aws.AWS {
 		ElastiCache:   elasticache.Adapt(cfFile),
 		Elasticsearch: elasticsearch.Adapt(cfFile),
 		ELB:           elb.Adapt(cfFile),
+		MSK:           msk.Adapt(cfFile),
 		Neptune:       neptune.Adapt(cfFile),
 		RDS:           rds.Adapt(cfFile),
 		Redshift:      redshift.Adapt(cfFile),
diff --git a/internal/app/cfsec/adapter/aws/msk/msk.go b/internal/app/cfsec/adapter/aws/msk/msk.go
new file mode 100644
index 00000000..8b3635b9
--- /dev/null
+++ b/internal/app/cfsec/adapter/aws/msk/msk.go
@@ -0,0 +1,66 @@
+package msk
+
+import (
+	"github.com/aquasecurity/cfsec/internal/app/cfsec/parser"
+	"github.com/aquasecurity/defsec/provider/aws/msk"
+	"github.com/aquasecurity/defsec/types"
+)
+
+func Adapt(cfFile parser.FileContext) msk.MSK {
+	return msk.MSK{
+		Clusters: getClusters(cfFile),
+	}
+}
+
+func getClusters(ctx parser.FileContext) (clusters []msk.Cluster) {
+	for _, clusterResource := range ctx.GetResourceByType("AWS::MSK::Cluster") {
+
+		var cluster msk.Cluster
+
+		if brokerProp := clusterResource.GetProperty("EncryptionInfo.EncryptionInTransit.ClientBroker"); brokerProp.IsString() {
+			cluster.EncryptionInTransit.ClientBroker = brokerProp.AsStringValue()
+		} else {
+			cluster.EncryptionInTransit.ClientBroker = types.StringDefault("TLS", clusterResource.Metadata())
+		}
+
+		if logsProp := clusterResource.GetProperty("LoggingInfo.BrokerLogs"); logsProp.IsNotNil() {
+			if cloudwatchProp := logsProp.GetProperty("CloudWatchLogs"); cloudwatchProp.IsNotNil() {
+				if enableProp := cloudwatchProp.GetProperty("Enabled"); enableProp.IsBool() {
+					cluster.Logging.Broker.Cloudwatch.Enabled = enableProp.AsBoolValue()
+				} else {
+					cluster.Logging.Broker.Cloudwatch.Enabled = types.BoolDefault(false, cloudwatchProp.Metadata())
+				}
+			} else {
+				cluster.Logging.Broker.Cloudwatch.Enabled = types.BoolDefault(false, logsProp.Metadata())
+			}
+
+			if firehoseProp := logsProp.GetProperty("Firehose"); firehoseProp.IsNotNil() {
+				if enableProp := firehoseProp.GetProperty("Enabled"); enableProp.IsBool() {
+					cluster.Logging.Broker.Firehose.Enabled = enableProp.AsBoolValue()
+				} else {
+					cluster.Logging.Broker.Firehose.Enabled = types.BoolDefault(false, firehoseProp.Metadata())
+				}
+			} else {
+				cluster.Logging.Broker.Firehose.Enabled = types.BoolDefault(false, logsProp.Metadata())
+			}
+
+			if s3Prop := logsProp.GetProperty("S3"); s3Prop.IsNotNil() {
+				if enableProp := s3Prop.GetProperty("Enabled"); enableProp.IsBool() {
+					cluster.Logging.Broker.S3.Enabled = enableProp.AsBoolValue()
+				} else {
+					cluster.Logging.Broker.S3.Enabled = types.BoolDefault(false, s3Prop.Metadata())
+				}
+			} else {
+				cluster.Logging.Broker.S3.Enabled = types.BoolDefault(false, logsProp.Metadata())
+			}
+
+		} else {
+			cluster.Logging.Broker.Cloudwatch.Enabled = types.BoolDefault(false, clusterResource.Metadata())
+			cluster.Logging.Broker.Firehose.Enabled = types.BoolDefault(false, clusterResource.Metadata())
+			cluster.Logging.Broker.S3.Enabled = types.BoolDefault(false, clusterResource.Metadata())
+		}
+
+		clusters = append(clusters, cluster)
+	}
+	return clusters
+}
diff --git a/internal/app/cfsec/rules/aws/msk/enable_in_transit_encryption_rule.go b/internal/app/cfsec/rules/aws/msk/enable_in_transit_encryption_rule.go
new file mode 100644
index 00000000..5030b119
--- /dev/null
+++ b/internal/app/cfsec/rules/aws/msk/enable_in_transit_encryption_rule.go
@@ -0,0 +1,42 @@
+package rds
+
+import (
+	"github.com/aquasecurity/cfsec/internal/app/cfsec/rule"
+	"github.com/aquasecurity/cfsec/internal/app/cfsec/scanner"
+	"github.com/aquasecurity/defsec/rules/aws/msk"
+)
+
+func init() {
+	scanner.RegisterCheckRule(rule.Rule{
+
+		BadExample: []string{
+			`---
+AWSTemplateFormatVersion: 2010-09-09
+Description: Bad example
+Resources:
+  Cluster:
+    Type: AWS::MSK::Cluster
+    Properties:
+      EncryptionInfo:
+        EncryptionInTransit:
+          ClientBroker: "TLS_PLAINTEXT"
+
+`,
+		},
+
+		GoodExample: []string{
+			`---
+AWSTemplateFormatVersion: 2010-09-09
+Description: Good example
+Resources:
+  Cluster:
+    Type: AWS::MSK::Cluster
+    Properties:
+      EncryptionInfo:
+        EncryptionInTransit:
+          ClientBroker: "TLS"
+`,
+		},
+		Base: msk.CheckEnableInTransitEncryption,
+	})
+}
diff --git a/internal/app/cfsec/rules/aws/msk/enable_in_transit_encryption_rule_test.go b/internal/app/cfsec/rules/aws/msk/enable_in_transit_encryption_rule_test.go
new file mode 100644
index 00000000..90bed0b6
--- /dev/null
+++ b/internal/app/cfsec/rules/aws/msk/enable_in_transit_encryption_rule_test.go
@@ -0,0 +1,17 @@
+package rds
+
+import (
+	"testing"
+
+	"github.com/aquasecurity/cfsec/internal/app/cfsec/test"
+)
+
+func Test_MSK_InTransit_FailureExamples(t *testing.T) {
+	expectedCode := "aws-msk-enable-in-transit-encryption"
+	test.RunFailureExamplesTest(t, expectedCode)
+}
+
+func Test_MSK_InTransit_SuccessExamples(t *testing.T) {
+	expectedCode := "aws-msk-enable-in-transit-encryption"
+	test.RunPassingExamplesTest(t, expectedCode)
+}
diff --git a/internal/app/cfsec/rules/aws/msk/enable_logging_rule.go b/internal/app/cfsec/rules/aws/msk/enable_logging_rule.go
new file mode 100644
index 00000000..ce804415
--- /dev/null
+++ b/internal/app/cfsec/rules/aws/msk/enable_logging_rule.go
@@ -0,0 +1,46 @@
+package rds
+
+import (
+	"github.com/aquasecurity/cfsec/internal/app/cfsec/rule"
+	"github.com/aquasecurity/cfsec/internal/app/cfsec/scanner"
+	"github.com/aquasecurity/defsec/rules/aws/msk"
+)
+
+func init() {
+	scanner.RegisterCheckRule(rule.Rule{
+
+		BadExample: []string{
+			`---
+AWSTemplateFormatVersion: 2010-09-09
+Description: Bad example
+Resources:
+  Cluster:
+    Type: AWS::MSK::Cluster
+    Properties:
+      LoggingInfo:
+        BrokerLogs:
+          CloudWatchLogs:
+            Enabled: false
+
+`,
+		},
+
+		GoodExample: []string{
+			`---
+AWSTemplateFormatVersion: 2010-09-09
+Description: Good example
+Resources:
+  Cluster:
+    Type: AWS::MSK::Cluster
+    Properties:
+      LoggingInfo:
+        BrokerLogs:
+          S3:
+            Enabled: true
+
+
+`,
+		},
+		Base: msk.CheckEnableLogging,
+	})
+}
diff --git a/internal/app/cfsec/rules/aws/msk/enable_logging_rule_test.go b/internal/app/cfsec/rules/aws/msk/enable_logging_rule_test.go
new file mode 100644
index 00000000..d2757027
--- /dev/null
+++ b/internal/app/cfsec/rules/aws/msk/enable_logging_rule_test.go
@@ -0,0 +1,17 @@
+package rds
+
+import (
+	"testing"
+
+	"github.com/aquasecurity/cfsec/internal/app/cfsec/test"
+)
+
+func Test_MSK_Logs_FailureExamples(t *testing.T) {
+	expectedCode := "aws-msk-enable-logging"
+	test.RunFailureExamplesTest(t, expectedCode)
+}
+
+func Test_MSK_Logs_SuccessExamples(t *testing.T) {
+	expectedCode := "aws-msk-enable-logging"
+	test.RunPassingExamplesTest(t, expectedCode)
+}
diff --git a/vendor/github.com/aquasecurity/defsec/rules/aws/msk/enable_in_transit_encryption.go b/vendor/github.com/aquasecurity/defsec/rules/aws/msk/enable_in_transit_encryption.go
new file mode 100644
index 00000000..2021dc5e
--- /dev/null
+++ b/vendor/github.com/aquasecurity/defsec/rules/aws/msk/enable_in_transit_encryption.go
@@ -0,0 +1,42 @@
+package msk
+
+import (
+	"github.com/aquasecurity/defsec/provider"
+	"github.com/aquasecurity/defsec/provider/aws/msk"
+	"github.com/aquasecurity/defsec/rules"
+	"github.com/aquasecurity/defsec/severity"
+	"github.com/aquasecurity/defsec/state"
+)
+
+var CheckEnableInTransitEncryption = rules.Register(
+	rules.Rule{
+		Provider:    provider.AWSProvider,
+		Service:     "msk",
+		ShortCode:   "enable-in-transit-encryption",
+		Summary:     "A MSK cluster allows unencrypted data in transit.",
+		Impact:      "Intercepted data can be read in transit",
+		Resolution:  "Enable in transit encryption",
+		Explanation: `Encryption should be forced for Kafka clusters, including for communication between nodes. This ensure sensitive data is kept private.`,
+		Links: []string{
+			"https://docs.aws.amazon.com/msk/latest/developerguide/msk-encryption.html",
+		},
+		Severity: severity.High,
+	},
+	func(s *state.State) (results rules.Results) {
+		for _, cluster := range s.AWS.MSK.Clusters {
+			if cluster.EncryptionInTransit.ClientBroker.EqualTo(msk.ClientBrokerEncryptionPlaintext) {
+				results.Add(
+					"Cluster allows plaintext communication.",
+					cluster.EncryptionInTransit.ClientBroker,
+				)
+			} else if cluster.EncryptionInTransit.ClientBroker.EqualTo(msk.ClientBrokerEncryptionTLSOrPlaintext) {
+				results.Add(
+					"Cluster allows plaintext communication.",
+					cluster.EncryptionInTransit.ClientBroker,
+				)
+			}
+
+		}
+		return
+	},
+)
diff --git a/vendor/github.com/aquasecurity/defsec/rules/aws/msk/enable_logging.go b/vendor/github.com/aquasecurity/defsec/rules/aws/msk/enable_logging.go
new file mode 100644
index 00000000..e7e24610
--- /dev/null
+++ b/vendor/github.com/aquasecurity/defsec/rules/aws/msk/enable_logging.go
@@ -0,0 +1,45 @@
+package msk
+
+import (
+	"github.com/aquasecurity/defsec/provider"
+	"github.com/aquasecurity/defsec/rules"
+	"github.com/aquasecurity/defsec/severity"
+	"github.com/aquasecurity/defsec/state"
+)
+
+var CheckEnableLogging = rules.Register(
+	rules.Rule{
+		Provider:    provider.AWSProvider,
+		Service:     "msk",
+		ShortCode:   "enable-logging",
+		Summary:     "Ensure MSK Cluster logging is enabled",
+		Impact:      "Without logging it is difficult to trace issues",
+		Resolution:  "Enable logging",
+		Explanation: `Managed streaming for Kafka can log to Cloud Watch, Kinesis Firehose and S3, at least one of these locations should be logged to`,
+		Links:       []string{},
+		Severity:    severity.Medium,
+	},
+	func(s *state.State) (results rules.Results) {
+		for _, cluster := range s.AWS.MSK.Clusters {
+			brokerLogging := cluster.Logging.Broker
+
+			if brokerLogging.S3.Enabled.IsTrue() {
+				continue
+			}
+
+			if brokerLogging.Firehose.Enabled.IsTrue() {
+				continue
+			}
+
+			if brokerLogging.Cloudwatch.Enabled.IsTrue() {
+				continue
+			}
+
+			results.Add(
+				"Cluster does not ship logs to any service.",
+				brokerLogging.Cloudwatch.Enabled,
+			)
+		}
+		return
+	},
+)
diff --git a/vendor/modules.txt b/vendor/modules.txt
index d7bdaca9..7cfb70a5 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -110,6 +110,7 @@ github.com/aquasecurity/defsec/rules/aws/elasticache
 github.com/aquasecurity/defsec/rules/aws/elasticsearch
 github.com/aquasecurity/defsec/rules/aws/elb
 github.com/aquasecurity/defsec/rules/aws/iam
+github.com/aquasecurity/defsec/rules/aws/msk
 github.com/aquasecurity/defsec/rules/aws/neptune
 github.com/aquasecurity/defsec/rules/aws/rds
 github.com/aquasecurity/defsec/rules/aws/redshift