feat(sdk): add Data Pipeline secret scanning check (#11821)

Co-authored-by: Daniel Barranquero <danielbo2001@gmail.com>
2026-07-23 12:31:54 +00:00 · 2026-07-09 15:21:05 +01:00
parent bf9c53a4c3
commit 0b36d08b92
11 changed files with 639 additions and 0 deletions
@@ -15,6 +15,9 @@
        "codebuild:BatchGet*",
        "codebuild:ListReportGroups",
        "cognito-idp:GetUserPoolMfaConfig",
+        "datapipeline:DescribePipelines",
+        "datapipeline:GetPipelineDefinition",
+        "datapipeline:ListPipelines",
        "dlm:Get*",
        "drs:Describe*",
        "ds:Get*",
@@ -109,6 +109,9 @@ Resources:
                  - "codebuild:BatchGet*"
                  - "codebuild:ListReportGroups"
                  - "cognito-idp:GetUserPoolMfaConfig"
+                  - "datapipeline:DescribePipelines"
+                  - "datapipeline:GetPipelineDefinition"
+                  - "datapipeline:ListPipelines"
                  - "dlm:Get*"
                  - "drs:Describe*"
                  - "ds:Get*"
@@ -0,0 +1 @@
+`datapipeline_pipeline_no_secrets_in_definition` check for AWS provider, scanning Data Pipeline object fields, parameter objects, and parameter values for hardcoded secrets with Kingfisher
@@ -0,0 +1,6 @@
+from prowler.providers.aws.services.datapipeline.datapipeline_service import (
+    DataPipeline,
+)
+from prowler.providers.common.provider import Provider
+
+datapipeline_client = DataPipeline(Provider.get_global_provider())
@@ -0,0 +1,43 @@
+{
+  "Provider": "aws",
+  "CheckID": "datapipeline_pipeline_no_secrets_in_definition",
+  "CheckTitle": "Data 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": "datapipeline",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:datapipeline:region:account-id:pipeline/pipeline-id",
+  "Severity": "high",
+  "ResourceType": "AwsDataPipelinePipeline",
+  "ResourceGroup": "analytics",
+  "Description": "AWS Data Pipeline definitions are inspected for hardcoded secrets, such as keys, tokens, passwords, or database credentials embedded directly in pipeline objects, parameters, or parameter values.",
+  "Risk": "Plaintext secrets in Data Pipeline definitions can be viewed by users with pipeline read permissions and may leak through scripts, SQL commands, logs, or exported definitions. Exposed credentials can enable unauthorized access to databases, storage, and downstream systems.",
+  "RelatedUrl": "",
+  "AdditionalURLs": [
+    "https://docs.aws.amazon.com/datapipeline/latest/APIReference/API_GetPipelineDefinition.html",
+    "https://docs.aws.amazon.com/datapipeline/latest/DeveloperGuide/dp-object-pipeline.html",
+    "https://docs.prowler.com/developer-guide/secret-scanning-checks"
+  ],
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "1. Review the Data Pipeline definition.\n2. Remove hardcoded credentials from pipeline objects, parameter objects, and parameter values.\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 AWS Data 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/datapipeline_pipeline_no_secrets_in_definition"
+    }
+  },
+  "Categories": [
+    "secrets"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": "AWS Data Pipeline has been closed to new customers since July 25, 2024. Accounts without pre-existing pipelines will not return findings for this check, as the service cannot be used by new customers."
+}
@@ -0,0 +1,112 @@
+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.datapipeline.datapipeline_client import (
+    datapipeline_client,
+)
+
+
+class datapipeline_pipeline_no_secrets_in_definition(Check):
+    """Check that AWS Data Pipeline definitions contain no hardcoded secrets."""
+
+    def execute(self) -> list[Check_Report_AWS]:
+        """Execute the Data Pipeline definition secret scan."""
+        findings = []
+        secrets_ignore_patterns = datapipeline_client.audit_config.get(
+            "secrets_ignore_patterns", []
+        )
+        validate = datapipeline_client.audit_config.get("secrets_validate", False)
+        pipelines = list(datapipeline_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 Data Pipeline {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 Data Pipeline {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 Data Pipeline {pipeline.name} definition -> {secrets_string}."
+                    )
+                    annotate_verified_secrets(report, detect_secrets_output)
+
+            findings.append(report)
+        return findings
+
+
+def _build_definition_payload(definition: dict) -> tuple[str, dict[int, str]]:
+    """Build a line-oriented scan payload and map each line to a definition field."""
+    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 pipeline_object in definition.get("pipelineObjects", []):
+        object_name = pipeline_object.get("name") or pipeline_object.get("id")
+        for field in pipeline_object.get("fields", []):
+            field_name = field.get("key")
+            field_value = field.get("stringValue") or field.get("refValue")
+            add_line(f"object {object_name} field {field_name}", field_value)
+
+    for parameter_object in definition.get("parameterObjects", []):
+        parameter_name = parameter_object.get("id")
+        for attribute in parameter_object.get("attributes", []):
+            attribute_name = attribute.get("key")
+            attribute_value = attribute.get("stringValue")
+            add_line(
+                f"parameter object {parameter_name} attribute {attribute_name}",
+                attribute_value,
+            )
+
+    for parameter_value in definition.get("parameterValues", []):
+        parameter_id = parameter_value.get("id")
+        add_line(f"parameter value {parameter_id}", parameter_value.get("stringValue"))
+
+    return "\n".join(lines), line_context
@@ -0,0 +1,96 @@
+from botocore.exceptions import ClientError
+from pydantic.v1 import BaseModel, Field
+
+from prowler.lib.logger import logger
+from prowler.lib.scan_filters.scan_filters import is_resource_filtered
+from prowler.providers.aws.lib.service.service import AWSService
+
+
+class Pipeline(BaseModel):
+    """Represents an AWS Data Pipeline pipeline."""
+
+    id: str
+    name: str
+    arn: str
+    region: str
+    definition: dict = Field(default_factory=dict)
+    tags: list[dict] = Field(default_factory=list)
+
+
+class DataPipeline(AWSService):
+    """AWS Data Pipeline service class to list pipelines and definitions."""
+
+    def __init__(self, provider):
+        """Initialize the AWS Data Pipeline service."""
+        super().__init__(__class__.__name__, provider)
+        self.pipelines = {}
+        self.__threading_call__(self._list_pipelines)
+        if self.pipelines:
+            self.__threading_call__(
+                self._get_pipeline_definition, self.pipelines.values()
+            )
+
+    def _list_pipelines(self, regional_client) -> None:
+        """List AWS Data Pipeline pipelines in a region."""
+        logger.info("DataPipeline - Listing pipelines...")
+        try:
+            list_pipelines_paginator = regional_client.get_paginator("list_pipelines")
+            for page in list_pipelines_paginator.paginate():
+                for pipeline in page.get("pipelineIdList", []):
+                    pipeline_id = pipeline.get("id")
+                    pipeline_name = pipeline.get("name", pipeline_id)
+                    pipeline_arn = (
+                        f"arn:{self.audited_partition}:datapipeline:"
+                        f"{regional_client.region}:{self.audited_account}:pipeline/{pipeline_id}"
+                    )
+                    if not self.audit_resources or is_resource_filtered(
+                        pipeline_arn, self.audit_resources
+                    ):
+                        self.pipelines[pipeline_arn] = Pipeline(
+                            id=pipeline_id,
+                            name=pipeline_name,
+                            arn=pipeline_arn,
+                            region=regional_client.region,
+                        )
+        except ClientError as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+    def _get_pipeline_definition(self, pipeline: Pipeline) -> None:
+        """Get the full definition for an AWS Data Pipeline pipeline."""
+        logger.info(f"DataPipeline - Getting definition for pipeline {pipeline.id}...")
+        try:
+            regional_client = self.regional_clients[pipeline.region]
+            try:
+                pipeline_descriptions = regional_client.describe_pipelines(
+                    pipelineIds=[pipeline.id]
+                ).get("pipelineDescriptionList", [])
+                if pipeline_descriptions:
+                    pipeline.tags = pipeline_descriptions[0].get("tags", [])
+            except ClientError as error:
+                logger.error(
+                    f"{pipeline.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+            except Exception as error:
+                logger.error(
+                    f"{pipeline.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+            definition = regional_client.get_pipeline_definition(pipelineId=pipeline.id)
+            pipeline.definition = {
+                "pipelineObjects": definition.get("pipelineObjects", []),
+                "parameterObjects": definition.get("parameterObjects", []),
+                "parameterValues": definition.get("parameterValues", []),
+            }
+        except ClientError as error:
+            logger.error(
+                f"{pipeline.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+        except Exception as error:
+            logger.error(
+                f"{pipeline.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
@@ -0,0 +1,254 @@
+from unittest import mock
+
+from prowler.lib.utils.utils import SecretsScanError
+from prowler.providers.aws.services.datapipeline.datapipeline_service import Pipeline
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+
+class Test_datapipeline_pipeline_no_secrets_in_definition:
+    def test_no_pipelines(self):
+        datapipeline_client = mock.MagicMock()
+        datapipeline_client.pipelines = {}
+        datapipeline_client.audit_config = {"secrets_ignore_patterns": []}
+
+        result = _execute_check(datapipeline_client)
+
+        assert len(result) == 0
+
+    def test_pipeline_with_no_secrets_in_definition(self):
+        pipeline = _build_pipeline(
+            definition={
+                "pipelineObjects": [
+                    {
+                        "id": "Default",
+                        "name": "Default",
+                        "fields": [
+                            {"key": "type", "stringValue": "Default"},
+                            {"key": "scheduleType", "stringValue": "cron"},
+                        ],
+                    }
+                ]
+            }
+        )
+        datapipeline_client = mock.MagicMock()
+        datapipeline_client.pipelines = {pipeline.arn: pipeline}
+        datapipeline_client.audit_config = {"secrets_ignore_patterns": []}
+
+        result = _execute_check(datapipeline_client)
+
+        assert len(result) == 1
+        assert result[0].status == "PASS"
+        assert (
+            result[0].status_extended
+            == "No secrets found in Data Pipeline test-pipeline definition."
+        )
+        assert result[0].region == AWS_REGION_US_EAST_1
+        assert result[0].resource_id == "df-1234567890"
+        assert result[0].resource_arn == pipeline.arn
+
+    def test_pipeline_with_secrets_in_object_field(self):
+        pipeline = _build_pipeline(
+            definition={
+                "pipelineObjects": [
+                    {
+                        "id": "SqlActivity",
+                        "name": "SqlActivity",
+                        "fields": [
+                            {"key": "type", "stringValue": "SqlActivity"},
+                            {
+                                "key": "script",
+                                "stringValue": "select * from users where token = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.dozjgNryP4J3jVmNHl0w5N_XgL0n3I9PlFUP0THsR8U'",
+                            },
+                        ],
+                    }
+                ]
+            }
+        )
+        datapipeline_client = mock.MagicMock()
+        datapipeline_client.pipelines = {pipeline.arn: pipeline}
+        datapipeline_client.audit_config = {"secrets_ignore_patterns": []}
+
+        result = _execute_check(datapipeline_client)
+
+        assert len(result) == 1
+        assert result[0].status == "FAIL"
+        assert "test-pipeline" in result[0].status_extended
+        assert "object SqlActivity field script" in result[0].status_extended
+        assert "eyJhbGciOiJIUzI1Ni" not in result[0].status_extended
+        assert result[0].region == AWS_REGION_US_EAST_1
+        assert result[0].resource_id == "df-1234567890"
+        assert result[0].resource_arn == pipeline.arn
+
+    def test_pipeline_with_secrets_in_parameter_value(self):
+        pipeline = _build_pipeline(
+            definition={
+                "parameterValues": [
+                    {
+                        "id": "databasePassword",
+                        "stringValue": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJwYXNzd29yZCI6InN1cGVyLXNlY3JldCJ9.zZ7_9wzLQfPy4TAAkpS6I8nRcEvuTnbwN7gGr1pH5fQ",
+                    }
+                ]
+            }
+        )
+        datapipeline_client = mock.MagicMock()
+        datapipeline_client.pipelines = {pipeline.arn: pipeline}
+        datapipeline_client.audit_config = {"secrets_ignore_patterns": []}
+
+        result = _execute_check(datapipeline_client)
+
+        assert len(result) == 1
+        assert result[0].status == "FAIL"
+        assert "parameter value databasePassword" in result[0].status_extended
+        assert "super-secret" not in result[0].status_extended
+
+    def test_pipeline_with_verified_secret_escalates_severity(self):
+        from prowler.lib.check.models import Severity
+
+        pipeline = _build_pipeline(
+            definition={
+                "parameterValues": [
+                    {
+                        "id": "databasePassword",
+                        "stringValue": "verified-secret-value",
+                    }
+                ]
+            }
+        )
+        datapipeline_client = mock.MagicMock()
+        datapipeline_client.pipelines = {pipeline.arn: pipeline}
+        datapipeline_client.audit_config = {
+            "secrets_ignore_patterns": [],
+            "secrets_validate": True,
+        }
+
+        result, scan_batch = _execute_check_with_mocked_scan(
+            datapipeline_client,
+            return_value={
+                0: [
+                    {
+                        "type": "Generic Password",
+                        "line_number": 1,
+                        "filename": "data",
+                        "hashed_secret": "x",
+                        "is_verified": True,
+                    }
+                ]
+            },
+        )
+
+        assert scan_batch.call_args.kwargs.get("validate") is True
+        assert len(result) == 1
+        assert result[0].status == "FAIL"
+        assert result[0].check_metadata.Severity == Severity.critical
+        assert "confirmed to be live" in result[0].status_extended
+
+    def test_scan_error_marks_all_scannable_pipelines_manual(self):
+        first_pipeline = _build_pipeline(
+            pipeline_id="df-first",
+            pipeline_name="first-pipeline",
+            definition={
+                "pipelineObjects": [
+                    {
+                        "id": "Default",
+                        "name": "Default",
+                        "fields": [{"key": "type", "stringValue": "Default"}],
+                    }
+                ]
+            },
+        )
+        second_pipeline = _build_pipeline(
+            pipeline_id="df-second",
+            pipeline_name="second-pipeline",
+            definition={
+                "parameterValues": [
+                    {"id": "databasePassword", "stringValue": "secret-value"}
+                ]
+            },
+        )
+        datapipeline_client = mock.MagicMock()
+        datapipeline_client.pipelines = {
+            first_pipeline.arn: first_pipeline,
+            second_pipeline.arn: second_pipeline,
+        }
+        datapipeline_client.audit_config = {"secrets_ignore_patterns": []}
+
+        result, scan_batch = _execute_check_with_mocked_scan(
+            datapipeline_client,
+            side_effect=SecretsScanError("scanner unavailable"),
+        )
+
+        scan_payloads = scan_batch.call_args.args[0]
+        assert len(scan_payloads) == 2
+        assert len(result) == 2
+        assert {report.status for report in result} == {"MANUAL"}
+        assert all(
+            "manual review is required" in report.status_extended for report in result
+        )
+
+
+def _build_pipeline(
+    definition: dict,
+    pipeline_id: str = "df-1234567890",
+    pipeline_name: str = "test-pipeline",
+) -> Pipeline:
+    pipeline_arn = (
+        f"arn:aws:datapipeline:{AWS_REGION_US_EAST_1}:"
+        f"{AWS_ACCOUNT_NUMBER}:pipeline/{pipeline_id}"
+    )
+    return Pipeline(
+        id=pipeline_id,
+        name=pipeline_name,
+        arn=pipeline_arn,
+        region=AWS_REGION_US_EAST_1,
+        definition=definition,
+    )
+
+
+def _execute_check(datapipeline_client):
+    aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+    with (
+        mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ),
+        mock.patch(
+            "prowler.providers.aws.services.datapipeline.datapipeline_pipeline_no_secrets_in_definition.datapipeline_pipeline_no_secrets_in_definition.datapipeline_client",
+            datapipeline_client,
+        ),
+    ):
+        from prowler.providers.aws.services.datapipeline.datapipeline_pipeline_no_secrets_in_definition.datapipeline_pipeline_no_secrets_in_definition import (
+            datapipeline_pipeline_no_secrets_in_definition,
+        )
+
+        check = datapipeline_pipeline_no_secrets_in_definition()
+        return check.execute()
+
+
+def _execute_check_with_mocked_scan(
+    datapipeline_client, return_value=None, side_effect=None
+):
+    aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+    with (
+        mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ),
+        mock.patch(
+            "prowler.providers.aws.services.datapipeline.datapipeline_pipeline_no_secrets_in_definition.datapipeline_pipeline_no_secrets_in_definition.datapipeline_client",
+            datapipeline_client,
+        ),
+    ):
+        import prowler.providers.aws.services.datapipeline.datapipeline_pipeline_no_secrets_in_definition.datapipeline_pipeline_no_secrets_in_definition as check_module
+
+        with mock.patch.object(
+            check_module,
+            "detect_secrets_scan_batch",
+            return_value=return_value,
+            side_effect=side_effect,
+        ) as scan_batch:
+            check = check_module.datapipeline_pipeline_no_secrets_in_definition()
+            return check.execute(), scan_batch
@@ -0,0 +1,121 @@
+from unittest.mock import patch
+
+import botocore
+from moto import mock_aws
+
+from prowler.providers.aws.services.datapipeline.datapipeline_service import (
+    DataPipeline,
+    Pipeline,
+)
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+pipeline_id = "df-1234567890"
+pipeline_name = "test-pipeline"
+pipeline_arn = (
+    f"arn:aws:datapipeline:{AWS_REGION_US_EAST_1}:"
+    f"{AWS_ACCOUNT_NUMBER}:pipeline/{pipeline_id}"
+)
+pipeline_definition = {
+    "pipelineObjects": [
+        {
+            "id": "Default",
+            "name": "Default",
+            "fields": [{"key": "type", "stringValue": "Default"}],
+        }
+    ],
+    "parameterObjects": [],
+    "parameterValues": [],
+}
+pipeline_tags = [{"key": "Environment", "value": "test"}]
+
+make_api_call = botocore.client.BaseClient._make_api_call
+
+
+def mock_make_api_call(self, operation_name, kwarg):
+    if operation_name == "ListPipelines":
+        return {"pipelineIdList": [{"id": pipeline_id, "name": pipeline_name}]}
+    if operation_name == "DescribePipelines":
+        return {
+            "pipelineDescriptionList": [
+                {
+                    "pipelineId": pipeline_id,
+                    "name": pipeline_name,
+                    "tags": pipeline_tags,
+                }
+            ]
+        }
+    if operation_name == "GetPipelineDefinition":
+        return pipeline_definition
+    return make_api_call(self, operation_name, kwarg)
+
+
+def mock_make_api_call_describe_fails(self, operation_name, kwarg):
+    if operation_name == "ListPipelines":
+        return {"pipelineIdList": [{"id": pipeline_id, "name": pipeline_name}]}
+    if operation_name == "DescribePipelines":
+        raise botocore.exceptions.ClientError(
+            {
+                "Error": {
+                    "Code": "AccessDeniedException",
+                    "Message": "Access denied",
+                }
+            },
+            operation_name,
+        )
+    if operation_name == "GetPipelineDefinition":
+        return pipeline_definition
+    return make_api_call(self, operation_name, kwarg)
+
+
+def mock_generate_regional_clients(provider, service):
+    regional_client = provider._session.current_session.client(
+        service, region_name=AWS_REGION_US_EAST_1
+    )
+    regional_client.region = AWS_REGION_US_EAST_1
+    return {AWS_REGION_US_EAST_1: regional_client}
+
+
+@patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call)
+@patch(
+    "prowler.providers.aws.aws_provider.AwsProvider.generate_regional_clients",
+    new=mock_generate_regional_clients,
+)
+class TestDataPipelineService:
+    @mock_aws
+    def test_datapipeline_service(self):
+        datapipeline = DataPipeline(set_mocked_aws_provider([AWS_REGION_US_EAST_1]))
+
+        assert datapipeline.session.__class__.__name__ == "Session"
+        assert datapipeline.service == "datapipeline"
+        assert len(datapipeline.pipelines) == 1
+        assert isinstance(datapipeline.pipelines[pipeline_arn], Pipeline)
+
+        pipeline = datapipeline.pipelines[pipeline_arn]
+        assert pipeline.id == pipeline_id
+        assert pipeline.name == pipeline_name
+        assert pipeline.arn == pipeline_arn
+        assert pipeline.region == AWS_REGION_US_EAST_1
+        assert pipeline.definition == pipeline_definition
+        assert pipeline.tags == pipeline_tags
+
+
+@patch(
+    "botocore.client.BaseClient._make_api_call", new=mock_make_api_call_describe_fails
+)
+@patch(
+    "prowler.providers.aws.aws_provider.AwsProvider.generate_regional_clients",
+    new=mock_generate_regional_clients,
+)
+class TestDataPipelineServiceDescribeFailure:
+    @mock_aws
+    def test_datapipeline_service_gets_definition_when_describe_fails(self):
+        datapipeline = DataPipeline(set_mocked_aws_provider([AWS_REGION_US_EAST_1]))
+
+        assert len(datapipeline.pipelines) == 1
+        pipeline = datapipeline.pipelines[pipeline_arn]
+        assert pipeline.definition == pipeline_definition
+        assert pipeline.tags == []
				`@@ -0,0 +1 @@`
				`datapipeline_pipeline_no_secrets_in_definition` check for AWS provider, scanning Data Pipeline object fields, parameter objects, and parameter values for hardcoded secrets with Kingfisher