feat: add check for secrets in CodePipeline pipeline definitions

prowler/providers/aws/services/codepipeline/codepipeline_pipeline_no_secrets_in_definition/codepipeline_pipeline_no_secrets_in_definition.metadata.json

@@ -0,0 +1,44 @@
+{
+  "Provider": "aws",
+  "CheckID": "codepipeline_pipeline_no_secrets_in_definition",
+  "CheckTitle": "CodePipeline pipeline definition has no sensitive credentials",
+  "CheckType": [
+    "Software and Configuration Checks/AWS Security Best Practices",
+    "TTPs/Credential Access",
+    "Effects/Data Exposure",
+    "Sensitive Data Identifications/Security"
+  ],
+  "ServiceName": "codepipeline",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:codepipeline:region:account-id:pipeline-name",
+  "Severity": "high",
+  "ResourceType": "AwsCodePipelinePipeline",
+  "ResourceGroup": "devops",
+  "Description": "AWS CodePipeline pipeline definitions are inspected for hardcoded secrets, such as keys, tokens, passwords, or credentials embedded directly in stage action configuration values.",
+  "Risk": "Plaintext secrets in CodePipeline pipeline definitions can be viewed by users with pipeline read permissions and may leak through logs or exported definitions. Exposed credentials can enable unauthorized access to source repositories, build systems, and deployment targets.",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://docs.aws.amazon.com/codepipeline/latest/APIReference/API_GetPipeline.html",
+    "https://docs.aws.amazon.com/codepipeline/latest/userguide/pipeline-structure.html",
+    "https://docs.prowler.com/developer-guide/secret-scanning-checks"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "1. Review the CodePipeline pipeline definition.\n2. Remove hardcoded credentials from stage action configurations.\n3. Store secrets in AWS Secrets Manager or AWS Systems Manager Parameter Store.\n4. Grant the pipeline role least-privilege access to retrieve secrets securely at runtime.\n5. Rotate any exposed credentials.",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Avoid embedding secrets in CodePipeline pipeline definitions. Store sensitive values in AWS Secrets Manager or AWS Systems Manager Parameter Store and reference them securely at runtime with least-privilege IAM permissions.",
+      "Url": "https://hub.prowler.com/check/codepipeline_pipeline_no_secrets_in_definition"
+    }
+  },
+  "Categories": [
+    "secrets",
+    "ci-cd"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

prowler/providers/aws/services/codepipeline/codepipeline_pipeline_no_secrets_in_definition/codepipeline_pipeline_no_secrets_in_definition.py

@@ -0,0 +1,118 @@
+import json
+
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.lib.utils.utils import (
+    SecretsScanError,
+    annotate_verified_secrets,
+    detect_secrets_scan_batch,
+)
+from prowler.providers.aws.services.codepipeline.codepipeline_client import (
+    codepipeline_client,
+)
+
+
+class codepipeline_pipeline_no_secrets_in_definition(Check):
+    """Check that AWS CodePipeline pipeline definitions contain no hardcoded secrets."""
+
+    def execute(self) -> list[Check_Report_AWS]:
+        """Execute the CodePipeline definition secret scan.
+
+        Scans the action configurations of every CodePipeline pipeline definition
+        for hardcoded secrets (keys, tokens, passwords, credentials) using the
+        batched Kingfisher secret scanner.
+
+        Returns:
+            list[Check_Report_AWS]: One report per CodePipeline pipeline, with
+                status PASS when no secrets are found, FAIL when potential
+                secrets are detected, or MANUAL when the scanner cannot produce
+                a trustworthy result.
+        """
+        findings = []
+        secrets_ignore_patterns = codepipeline_client.audit_config.get(
+            "secrets_ignore_patterns", []
+        )
+        validate = codepipeline_client.audit_config.get("secrets_validate", False)
+        pipelines = list(codepipeline_client.pipelines.values())
+        line_context_by_pipeline = {}
+        payloads = []
+        for pipeline_index, pipeline in enumerate(pipelines):
+            payload, line_context = _build_definition_payload(pipeline.definition)
+            line_context_by_pipeline[pipeline_index] = line_context
+            if payload:
+                payloads.append((pipeline_index, payload))
+
+        scan_error = None
+        try:
+            batch_results = detect_secrets_scan_batch(
+                payloads, excluded_secrets=secrets_ignore_patterns, validate=validate
+            )
+        except SecretsScanError as error:
+            batch_results = {}
+            scan_error = error
+
+        for pipeline_index, pipeline in enumerate(pipelines):
+            report = Check_Report_AWS(metadata=self.metadata(), resource=pipeline)
+            report.resource_tags = pipeline.tags
+            report.status = "PASS"
+            report.status_extended = (
+                f"No secrets found in CodePipeline {pipeline.name} definition."
+            )
+
+            line_context = line_context_by_pipeline.get(pipeline_index, {})
+            if line_context:
+                if scan_error:
+                    report.status = "MANUAL"
+                    report.status_extended = (
+                        f"Could not scan CodePipeline {pipeline.name} definition "
+                        f"for secrets: {scan_error}; manual review is required."
+                    )
+                    findings.append(report)
+                    continue
+
+                detect_secrets_output = batch_results.get(pipeline_index)
+                if detect_secrets_output:
+                    secrets_string = ", ".join(
+                        [
+                            f"{secret['type']} in {line_context.get(secret['line_number'], 'definition')}"
+                            for secret in detect_secrets_output
+                        ]
+                    )
+                    report.status = "FAIL"
+                    report.status_extended = (
+                        f"Potential {'secrets' if len(detect_secrets_output) > 1 else 'secret'} "
+                        f"found in CodePipeline {pipeline.name} definition -> {secrets_string}."
+                    )
+                    annotate_verified_secrets(report, detect_secrets_output)
+
+            findings.append(report)
+        return findings
+
+
+def _build_definition_payload(definition: list) -> tuple[str, dict[int, str]]:
+    """Build a line-oriented scan payload and map each line to a definition field.
+
+    Iterates over every stage/action configuration in the pipeline definition and
+    emits one JSON line per configuration value so that Kingfisher can scan each
+    value independently and map findings back to the originating stage/action.
+    """
+    lines = []
+    line_context = {}
+
+    def add_line(context: str, value) -> None:
+        if value is None:
+            return
+        lines.append(json.dumps({context: value}))
+        line_context[len(lines)] = context
+
+    for stage in definition:
+        stage_name = stage.get("name", "stage")
+        for action in stage.get("actions", []):
+            action_name = action.get("name", "action")
+            configuration = action.get("configuration", {})
+            for config_key, config_value in configuration.items():
+                add_line(
+                    f"stage {stage_name} action {action_name} configuration {config_key}",
+                    config_value,
+                )
+
+    return "\n".join(lines), line_context

prowler/providers/aws/services/codepipeline/codepipeline_service.py

@@ -77,7 +77,8 @@ def _list_pipelines(self, regional_client):
     def _get_pipeline_state(self, pipeline):
         """Retrieves the current state of a pipeline.
 
-        Gets detailed information about a pipeline including its source configuration.
+        Gets detailed information about a pipeline including its source configuration
+        and full definition (stages and actions).
 
         Args:
             pipeline: Pipeline object to retrieve state for.
@@ -96,6 +97,7 @@ def _get_pipeline_state(self, pipeline):
                 repository_id=repository_id,
                 configuration=source_info["configuration"],
             )
+            pipeline.definition = pipeline_info["pipeline"].get("stages", [])
         except ClientError as error:
             logger.error(
                 f"{pipeline.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -154,11 +156,13 @@ class Pipeline(BaseModel):
         arn: The ARN (Amazon Resource Name) of the pipeline.
         region: The AWS region where the pipeline exists.
         source: Optional Source object containing source configuration.
+        definition: Optional pipeline definition containing stages and actions.
         tags: Optional list of pipeline tags.
     """
 
     name: str
     arn: str
     region: str
     source: Optional[Source] = None
+    definition: Optional[list] = []
     tags: Optional[list] = []