Repository: prowler-cloud/prowler
Pull request: feat(new_check): cloudwatch_log_group_no_secrets_in_logs
### Context

Added a check to look for secrets in CloudWatch logs


### Description

Altered the CloudWatch Logs class to get log events for each log group. There are 2 parameters that determine the number of events that are collected. 
* number_of_log_streams_to_retrieve determines the number of log streams per log group to get. The latest log streams are retrieved
* events_per_log_stream_threshold determines the threshold for the number of events before it stops collecting more events. It will retrieve at most events_per_log_stream_threshold + events_per_log_stream_threshold-1 events. Could alter this to retrieve exactly events_per_log_stream_threshold, but it didn't seem to make a performance difference. Can discuss this further over slack if you want.
I think that these parameters should be configured in the config file.

The check will output all the events into a temp file, and run detect-secrets on them. If a secret is detected, it will then look to see if it can give more specific information on what lines the secrets were detected on for that event. I do this by using json.loads on the event message, and then doing a json.dumps with the indent parameter set to 2. If there are multiple lines after this, then the event is run through detect-secrets again to determine which lines of the secrets are being flagged on. I found this useful for logs being generated for the aws-controltower/CloudTrailLogs log group which outputs logs as json messages. 

The convert_to_cloudwatch_timestamp_format function in the check could move to the cloudwatch_service file, and the detect_secrets_scan function and potentially the SecretsDict could move to the utils file.


### License

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.


Problem statement:
Implement a new check for secrets in CloudWatch Logs
**Is your feature request related to a problem? Please describe.**
Suggestion to implement a new check that performs the same secrets detection done for ECS task definitions/lambda code etc, but done against logs stored in CloudWatch to detect sensitive information being stored in logs in plaintext.

**Describe the solution you'd like**
It is likely that it is not feasible to pull all the log streams for each cloudwatch log group. I suggest that ~5-10 streams are pulled, and run through secrets-detector. Something like the below could be altered and incorporated into the logs_client - with the logGroupName being a variable (for log_group in self.log_groups).

```python
     stream_response = client.describe_log_streams(
        logGroupName="/aws/lambda/lambdaFnName", # Can be dynamic
        orderBy='LastEventTime',                 # For the latest events
        limit=5
        )

    for log_stream in stream_response["logStreams"]:
        logStreamName = log_stream["logStreamName"]

        response = client.get_log_events(
             logGroupName="/aws/lambda/lambdaFnName",
             logStreamName=logStreamName
        )
```

**Describe alternatives you've considered**
None

**Additional context**
None

Hints from issue discussion:
If the team agrees that this is something worth implementing then Id be happy doing it - I would just like that someone else handles the metadata for the check.
For sure @Fennerr , great idea! I could help you with the metadata 💪🏼

Commit message: Update prowler/providers/aws/services/cloudwatch/cloudwatch_no_secerts_in_logs/cloudwatch_no_secerts_in_logs.metadata.json

Co-authored-by: Pepe Fagoaga <pepe@verica.io>

Patch to review:
diff --git a/prowler/providers/aws/services/cloudwatch/cloudwatch_no_secerts_in_logs/__init__.py b/prowler/providers/aws/services/cloudwatch/cloudwatch_no_secerts_in_logs/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/prowler/providers/aws/services/cloudwatch/cloudwatch_no_secerts_in_logs/cloudwatch_no_secerts_in_logs.metadata.json b/prowler/providers/aws/services/cloudwatch/cloudwatch_no_secerts_in_logs/cloudwatch_no_secerts_in_logs.metadata.json
new file mode 100644
index 0000000000..0d8dde3cc0
--- /dev/null
+++ b/prowler/providers/aws/services/cloudwatch/cloudwatch_no_secerts_in_logs/cloudwatch_no_secerts_in_logs.metadata.json
@@ -0,0 +1,39 @@
+{
+  "Provider": "aws",
+  "CheckID": "cloudwatch_no_secrets_in_logs",
+  "CheckTitle": "Check if secrets exists in CloudWatch logs.",
+  "CheckType": [
+    "Protect",
+    "Secure development"
+  ],
+  "ServiceName": "cloudwatch",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:cloudwatch:region:account-id:log-group/resource-id",
+  "Severity": "medium",
+  "ResourceType": "AwsCloudTrailLogGroup",
+  "Description": "Check if secrets exists in CloudWatch logs",
+  "Risk": "Storing sensitive data in CloudWatch logs could allow an attacker with read-only access to escalate their privileges or gain unauthorised access to systems.",
+  "RelatedUrl": "https://docs.aws.amazon.com/awscloudtrail/latest/userguide/cloudwatch-alarms-for-cloudtrail.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "It is recommended that sensitive information is not logged to CloudWatch logs. Alternatively, sensitive data may be masked using a protection policy",
+      "Url": "https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/mask-sensitive-log-data.html"
+    }
+  },
+  "Categories": [
+    "secrets"
+  ],
+  "Tags": {
+    "Tag1Key": "value",
+    "Tag2Key": "value"
+  },
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
diff --git a/prowler/providers/aws/services/cloudwatch/cloudwatch_no_secerts_in_logs/cloudwatch_no_secerts_in_logs.py b/prowler/providers/aws/services/cloudwatch/cloudwatch_no_secerts_in_logs/cloudwatch_no_secerts_in_logs.py
new file mode 100644
index 0000000000..b66b839f4a
--- /dev/null
+++ b/prowler/providers/aws/services/cloudwatch/cloudwatch_no_secerts_in_logs/cloudwatch_no_secerts_in_logs.py
@@ -0,0 +1,145 @@
+import os
+import tempfile
+from datetime import datetime, timezone
+from json import dumps, loads
+
+from detect_secrets import SecretsCollection
+from detect_secrets.settings import default_settings
+
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.cloudwatch.cloudwatch_client import (
+    cloudwatch_client,
+)
+from prowler.providers.aws.services.cloudwatch.logs_client import logs_client
+
+
+class cloudwatch_no_secerts_in_logs(Check):
+    def execute(self):
+        findings = []
+        for log_group in logs_client.log_groups:
+            report = Check_Report_AWS(self.metadata())
+            report.status = "PASS"
+            report.status_extended = f"No secrets found in {log_group.name} log group."
+            report.region = cloudwatch_client.region
+            report.resource_id = log_group.arn
+            log_group_secrets = []
+            if log_group.log_streams:
+                for log_stream_name in log_group.log_streams:
+                    log_stream_secrets = {}
+                    log_stream_data = "\n".join(
+                        [
+                            dumps(event["message"])
+                            for event in log_group.log_streams[log_stream_name]
+                        ]
+                    )
+                    log_stream_secrets_output = detect_secrets_scan(log_stream_data)
+
+                    if log_stream_secrets_output:
+                        for secret in log_stream_secrets_output:
+                            flagged_event = log_group.log_streams[log_stream_name][
+                                secret["line_number"] - 1
+                            ]
+                            cloudwatch_timestamp = (
+                                convert_to_cloudwatch_timestamp_format(
+                                    flagged_event["timestamp"]
+                                )
+                            )
+                            if cloudwatch_timestamp not in log_stream_secrets.keys():
+                                log_stream_secrets[cloudwatch_timestamp] = SecretsDict()
+
+                            try:
+                                log_event_data = dumps(
+                                    loads(flagged_event["message"]), indent=2
+                                )
+                            except:
+                                log_event_data = dumps(
+                                    flagged_event["message"], indent=2
+                                )
+                            if len(log_event_data.split("\n")) > 1:
+                                # Can get more informative output if there is more than 1 line.
+                                # Will rescan just this event to get the type of secret and the line number
+                                event_detect_secrets_output = detect_secrets_scan(
+                                    log_event_data
+                                )
+                                [
+                                    log_stream_secrets[cloudwatch_timestamp].add_secret(
+                                        secret["line_number"], secret["type"]
+                                    )
+                                    for secret in event_detect_secrets_output
+                                ]
+                            else:
+                                log_stream_secrets[cloudwatch_timestamp].add_secret(
+                                    1, secret["type"]
+                                )
+                    if log_stream_secrets:
+                        secrets_string = "; ".join(
+                            [
+                                f"At {timestamp} - {log_stream_secrets[timestamp].to_string()}"
+                                for timestamp in log_stream_secrets
+                            ]
+                        )
+                        log_group_secrets.append(
+                            f"In log stream {log_stream_name}: {secrets_string}"
+                        )
+            if log_group_secrets:
+                secrets_string = ". ".join(log_group_secrets)
+                report.status = "FAIL"
+                report.status_extended = (
+                    f"Potential secrets found in log group. {secrets_string}"
+                )
+            findings.append(report)
+        return findings
+
+
+def convert_to_cloudwatch_timestamp_format(epoch_time):
+    date_time = datetime.fromtimestamp(
+        epoch_time / 1000, datetime.now(timezone.utc).astimezone().tzinfo
+    )
+    datetime_str = date_time.strftime(
+        "%Y-%m-%dT%H:%M:%S.!%f!%z"
+    )  # use exclamation marks as placeholders to convert datetime str to cloudwatch timestamp str
+    datetime_parts = datetime_str.split("!")
+    return (
+        datetime_parts[0]
+        + datetime_parts[1][:-3]
+        + datetime_parts[2][:-2]
+        + ":"
+        + datetime_parts[2][-2:]
+    )  # Removes the microseconds, and places a ':' character in the timezone offset
+
+
+def detect_secrets_scan(data):
+    # Should move this to the utils file, but will leave that for a merge request after this one is accepted
+    temp_data_file = tempfile.NamedTemporaryFile(delete=False)
+    temp_data_file.write(bytes(data, encoding="raw_unicode_escape"))
+    temp_data_file.close()
+
+    secrets = SecretsCollection()
+    with default_settings():
+        secrets.scan_file(temp_data_file.name)
+    os.remove(temp_data_file.name)
+
+    detect_secrets_output = secrets.json()
+    if detect_secrets_output:
+        return detect_secrets_output[temp_data_file.name]
+    else:
+        return None
+
+
+class SecretsDict(dict):
+    # Using this dict to remove duplicates of the secret type showing up multiple times on the same line
+    # Also includes the to_string method
+    def add_secret(self, line_number, secret_type):
+        if line_number not in self.keys():
+            self[line_number] = [secret_type]
+        else:
+            if secret_type not in self[line_number]:
+                self[line_number] += [secret_type]
+
+    def to_string(self):
+        return ", ".join(
+            [
+                f"{','.join(secret_types)} on line {line_number}"
+                for line_number, secret_types in sorted(self.items())
+            ]
+        )
diff --git a/prowler/providers/aws/services/cloudwatch/cloudwatch_service.py b/prowler/providers/aws/services/cloudwatch/cloudwatch_service.py
index ba7aeefd6c..88a287f1b4 100644
--- a/prowler/providers/aws/services/cloudwatch/cloudwatch_service.py
+++ b/prowler/providers/aws/services/cloudwatch/cloudwatch_service.py
@@ -75,8 +75,15 @@ def __init__(self, audit_info):
         self.regional_clients = generate_regional_clients(self.service, audit_info)
         self.metric_filters = []
         self.log_groups = []
-        self.__threading_call__(self.__describe_metric_filters__)
+        # self.__threading_call__(self.__describe_metric_filters__)
         self.__threading_call__(self.__describe_log_groups__)
+        if "cloudwatch_no_secerts_in_logs" in audit_info.audit_metadata.expected_checks:
+            # Can look at defining these parameters in the config file
+            self.number_of_log_streams_to_retrieve = (
+                2  # The number of log streams to retrieve for each log group.
+            )
+            self.events_per_log_stream_threshold = 100  # The threshold for number of log events to get per stream. Will retrieve until this limit is hit, or no more events are available.
+            self.__threading_call__(self.__get_log_events__)
 
     def __get_session__(self):
         return self.session
@@ -91,7 +98,7 @@ def __threading_call__(self, call):
             t.join()
 
     def __describe_metric_filters__(self, regional_client):
-        logger.info("CloudWatch Logs- Describing metric filters...")
+        logger.info("CloudWatch Logs - Describing metric filters...")
         try:
             describe_metric_filters_paginator = regional_client.get_paginator(
                 "describe_metric_filters"
@@ -116,7 +123,7 @@ def __describe_metric_filters__(self, regional_client):
             )
 
     def __describe_log_groups__(self, regional_client):
-        logger.info("CloudWatch Logs- Describing log groups...")
+        logger.info("CloudWatch Logs - Describing log groups...")
         try:
             describe_log_groups_paginator = regional_client.get_paginator(
                 "describe_log_groups"
@@ -146,6 +153,69 @@ def __describe_log_groups__(self, regional_client):
                 f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
 
+    def __get_log_events__(self, regional_client):
+        regional_log_groups = [
+            log_group
+            for log_group in self.log_groups
+            if log_group.region == regional_client.region
+        ]
+        total_log_groups = len(regional_log_groups)
+        logger.info(
+            f"CloudWatch Logs - Retrieving log events for {total_log_groups} log groups in {regional_client.region}..."
+        )
+        try:
+            # for log_group in self.log_groups:
+            for count, log_group in enumerate(regional_log_groups, start=1):
+                # if not log_group.region==regional_client.region:
+                #     continue
+
+                # Don't need a paginator, unless more than 50 streams are retrieved
+                log_streams = regional_client.describe_log_streams(
+                    logGroupName=log_group.name,
+                    orderBy="LastEventTime",
+                    limit=self.number_of_log_streams_to_retrieve,
+                )
+
+                for log_stream in log_streams["logStreams"]:
+                    log_stream_name = log_stream["logStreamName"]
+                    log_events = regional_client.get_log_events(
+                        logGroupName=log_group.name,
+                        logStreamName=log_stream_name,
+                        limit=self.events_per_log_stream_threshold,  # defaults to as many as can fit in 1MB, up to 10,000
+                    )
+                    events = log_events["events"]
+                    previous_backward_token = ""
+
+                    while previous_backward_token != log_events["nextBackwardToken"]:
+                        if len(events) >= self.events_per_log_stream_threshold:
+                            break
+
+                        previous_backward_token = log_events["nextBackwardToken"]
+                        log_events = regional_client.get_log_events(
+                            logGroupName=log_group.name,
+                            logStreamName=log_stream_name,
+                            nextToken=log_events["nextBackwardToken"],
+                            limit=self.events_per_log_stream_threshold,  # -len(events) # Could include this to limit it to the threshold, but I dont think there will be a performance issue.
+                        )
+                        events = events + log_events["events"]
+                    logger.info(
+                        f"CloudWatch Logs - Found {len(events)} log events for {log_stream_name} stream for log group {log_group.name} in {regional_client.region}..."
+                    )
+                    log_group.log_streams[log_stream_name] = events
+
+                if count % 10 == 0:
+                    logger.info(
+                        f"CloudWatch Logs - Retrieved log events for {count}/{total_log_groups} log groups in {regional_client.region}..."
+                    )
+
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+        logger.info(
+            f"CloudWatch Logs - Finished retrieving log events in {regional_client.region}..."
+        )
+
 
 @dataclass
 class MetricAlarm:
@@ -200,6 +270,9 @@ class LogGroup:
     retention_days: int
     kms_id: str
     region: str
+    log_streams: dict[
+        str, list[str]
+    ]  # Log stream name as the key, array of events as the value
 
     def __init__(
         self,
@@ -214,3 +287,4 @@ def __init__(
         self.retention_days = retention_days
         self.kms_id = kms_id
         self.region = region
+        self.log_streams = {}


Code review comments:
diff_hunk: @@ -75,8 +75,15 @@ def __init__(self, audit_info):
         self.regional_clients = generate_regional_clients(self.service, audit_info)
         self.metric_filters = []
         self.log_groups = []
-        self.__threading_call__(self.__describe_metric_filters__)
+        # self.__threading_call__(self.__describe_metric_filters__)
         self.__threading_call__(self.__describe_log_groups__)
+        if "cloudwatch_no_secerts_in_logs" in audit_info.audit_metadata.expected_checks:
path: prowler/providers/aws/services/cloudwatch/cloudwatch_service.py
text: @user1:
```suggestion
        if "cloudwatch_no_secrets_in_logs" in audit_info.audit_metadata.expected_checks:
```