Back to Checks
aws

AWS Batch job definitions have no secrets in environment variables or command parameters

batch_job_definition_no_secrets
Service: batchSeverity: high
OfficialBuilt-inVerified
Categories:secrets
Resource Type:AwsBatchJobDefinition

AWS Batch job definitions are analyzed for plaintext secrets placed in container environment variables and command parameters. It identifies values that resemble credentials (keys, tokens, passwords) within job definitions.

Risk

Exposed secrets in env vars or command parameters undermine confidentiality via logs, job metadata, and introspection.

With container or read-only API access, attackers can reuse credentials to read databases, modify records (integrity), pivot to other services, and trigger outages or unauthorized costs (availability).

Run this check with Prowler

Run in Prowler Cloud
Run
prowler aws --checks batch_job_definition_no_secrets

Recommendation

Store secrets in AWS Secrets Manager or SSM Parameter Store and inject them at runtime instead of plaintext env vars.

Apply least privilege via job role, enable regular rotation, avoid logging secret values, and prefer ephemeral credentials for downstream services.

Remediation

CLI

aws batch register-job-definition --job-definition-name <JOB_DEFINITION_NAME> --type container --container-properties '{"image":"<IMAGE>","secrets":[{"name":"<SECRET_NAME>","valueFrom":"arn:aws:secretsmanager:<REGION>:<ACCOUNT_ID>:secret:<SECRET_NAME>-<RANDOM>"}]}'  # Register a new revision without plaintext secrets; reference Secrets Manager or SSM Parameter Store via valueFrom

Terraform

resource "aws_batch_job_definition" "<example_resource_name>" {
  name = "<example_resource_name>"
  type = "container"

  container_properties = jsonencode({
    image = "<image>"
    environment = [
      {
        name  = "DB_PASSWORD"
        value = var.db_password  # Use variable from Secrets Manager or SSM
      }
    ]
  })
}

NativeIaC

Resources:
  <example_resource_name>:
    Type: AWS::Batch::JobDefinition
    Properties:
      Type: container
      ContainerProperties:
        Image: <image>
        Environment:
          - Name: DB_PASSWORD
            Value: !Ref <secret_parameter>  # Reference SSM Parameter or Secrets Manager

Other

1In the AWS Console, go to Batch > Job Definitions and open your job definition
2Create a new revision
3Remove any sensitive values from Environment variables and command parameters
4Reference secrets from AWS Secrets Manager or SSM Parameter Store instead
5Save to create the new revision
6Update any Batch job queues to use the new job definition revision

Check Code

from json import dumps

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.batch.batch_client import batch_client


class batch_job_definition_no_secrets(Check):
    """Detect secrets in AWS Batch job definition environment variables and commands."""

    def execute(self) -> list[Check_Report_AWS]:
        """Scan job definitions for hardcoded secrets in env vars and commands."""
        findings = []

        secrets_ignore_patterns = batch_client.audit_config.get(
            "secrets_ignore_patterns", []
        )
        validate = batch_client.audit_config.get("secrets_validate", False)

        job_definitions = list(batch_client.job_definitions.values())

        def scan_payloads():
            """Yield index-keyed payloads for each env var and the command."""
            for jd_index, job_definition in enumerate(job_definitions):
                container = job_definition.container_properties

                for env_index, env_var in enumerate(container.environment):
                    yield (jd_index, env_index), dumps(
                        {env_var.name: env_var.value}, indent=2
                    )

                if container.command:
                    yield (
                        (jd_index, "command"),
                        " ".join(container.command),
                    )

        scan_error = None
        try:
            batch_results = detect_secrets_scan_batch(
                scan_payloads(),
                excluded_secrets=secrets_ignore_patterns,
                validate=validate,
            )
        except SecretsScanError as error:
            batch_results = {}
            scan_error = error

        for jd_index, job_definition in enumerate(job_definitions):
            report = Check_Report_AWS(
                metadata=self.metadata(),
                resource=job_definition,
            )

            report.resource_id = f"{job_definition.name}:{job_definition.revision}"
            report.status = "PASS"

            extended_status_parts = []
            all_secrets = []

            container = job_definition.container_properties

            if scan_error and (container.environment or container.command):
                report.status = "MANUAL"
                report.status_extended = (
                    f"Could not scan Batch job definition "
                    f"{job_definition.name} with revision "
                    f"{job_definition.revision} for secrets: "
                    f"{scan_error}; manual review is required."
                )
                findings.append(report)
                continue

            for env_index, env_var in enumerate(container.environment):
                env_secrets = batch_results.get((jd_index, env_index))
                if env_secrets:
                    all_secrets.extend(env_secrets)
                    secrets_string = ", ".join(
                        f"{secret['type']} on the environment variable {env_var.name}"
                        for secret in env_secrets
                    )
                    extended_status_parts.append(
                        f"Secrets in environment variables -> {secrets_string}"
                    )

            if container.command:
                command_secrets = batch_results.get((jd_index, "command"))
                if command_secrets:
                    all_secrets.extend(command_secrets)
                    secrets_string = ", ".join(
                        secret["type"] for secret in command_secrets
                    )
                    extended_status_parts.append(
                        f"Secrets in command -> {secrets_string}"
                    )

            if extended_status_parts:
                report.status = "FAIL"
                report.status_extended = (
                    f"Potential secrets found in Batch job definition "
                    f"{job_definition.name} with revision "
                    f"{job_definition.revision}: "
                    + "; ".join(extended_status_parts)
                    + "."
                )
                annotate_verified_secrets(report, all_secrets)
            else:
                report.status_extended = (
                    f"No secrets found in Batch job definition "
                    f"{job_definition.name} with revision "
                    f"{job_definition.revision}."
                )

            findings.append(report)

        return findings