let msk_clusters = Resources.*[
	Type == "AWS::MSK::Cluster"
]

rule msk_encrypt_in_transit_tls when %msk_clusters !empty {
	%msk_clusters {
		Properties {
			EncryptionInfo !exists OR
			EncryptionInfo exists
			when EncryptionInfo exists {
				EncryptionInfo {
					EncryptionInTransit !exists OR
					EncryptionInTransit exists
					when EncryptionInTransit exists {
						EncryptionInTransit {
							ClientBroker !exists OR
							ClientBroker exists
							when ClientBroker exists {
								ClientBroker == "TLS" <<Only encryption in transit (Client-broker) is disabled.>>
							}
						}
					}
				}
			}
		}
	}
}

rule msk_encrypt_within_cluster when %msk_clusters !empty {
	%msk_clusters {
		Properties {
			EncryptionInfo !exists OR
			EncryptionInfo exists
			when EncryptionInfo exists {
				EncryptionInfo {
					EncryptionInTransit !exists OR
					EncryptionInTransit exists
					when EncryptionInTransit exists {
						EncryptionInTransit {
							InCluster !exists OR
							InCluster exists
							when InCluster exists {
								InCluster == true <<Encryption in-transit is disabled within the cluster nodes.>>
							}
						}
					}
				}
			}
		}
	}
}

rule msk_broker_logs_enabled when %msk_clusters !empty {
	%msk_clusters {
		Properties {
			LoggingInfo exists <<LoggingInfo is not configured.>>
			when LoggingInfo exists {
				LoggingInfo {
					BrokerLogs {
						CloudWatchLogs exists OR
						Firehose exists OR
						S3 exists <<No broker logs delivery configured.>>

						when CloudWatchLogs exists OR
						Firehose exists OR
						S3 exists {
							CloudWatchLogs.Enabled == true OR
							Firehose.Enabled == true OR
							S3.Enabled == true
						}
					}
				}
			}
		}
	}
}

rule msk_iam_authentication when %msk_clusters !empty {
	%msk_clusters {
		Properties {
			ClientAuthentication exists <<ClientAuthentication is not configured.>>
			when ClientAuthentication exists {
				ClientAuthentication {
					Sasl exists <<Sasl is not configured.>>
					when Sasl exists {
						Sasl {
							Iam exists <<Iam is not configured.>>
							when Iam exists {
								Iam {
									Enabled == true <<IAM authentication is disabled for Apache Kafka actions.>>
								}
							}
						}
					}
				}
			}
		}
	}
}