cloud security

Implementing Cloud DLP for Data Protection

Implementing Cloud Data Loss Prevention (DLP) using Amazon Macie, Azure Information Protection, and Google Cloud DLP API to discover, classify, and protect sensitive data across cloud storage, databases, and data pipelines.

cloud-securitydata-classificationdata-protectiondlpmacieprivacy
Install this skill
npx skills add mukul975/Anthropic-Cybersecurity-Skills
Framework mappings

When to Use

  • When compliance frameworks (GDPR, HIPAA, PCI DSS) require automated sensitive data discovery and protection
  • When building data governance programs that classify and label data across cloud storage
  • When implementing data loss prevention controls for cloud-based data pipelines
  • When auditing cloud environments for unprotected sensitive data (PII, PHI, financial data)
  • When integrating DLP scanning into CI/CD pipelines to prevent sensitive data from reaching production

Do not use for endpoint DLP (use Microsoft Purview or Symantec DLP agents), for email DLP (use Microsoft 365 DLP or Google Workspace DLP), or for network-level data exfiltration prevention (use VPC endpoint policies and network firewalls).

Prerequisites

  • Amazon Macie enabled with appropriate S3 bucket permissions
  • Google Cloud DLP API enabled (gcloud services enable dlp.googleapis.com)
  • Azure Information Protection or Microsoft Purview configured
  • IAM permissions for DLP service administration and data access
  • Knowledge of data sensitivity categories relevant to the organization (PII, PHI, PCI, proprietary)

Workflow

Step 1: Deploy Amazon Macie for S3 Data Discovery

Enable Macie and configure automated sensitive data discovery jobs for S3 buckets.

# Enable Amazon Macie
aws macie2 enable-macie
 
# List all S3 buckets Macie can scan
aws macie2 describe-buckets \
  --query 'buckets[*].[bucketName,classifiableSizeInBytes,unclassifiableObjectCount.total]' \
  --output table
 
# Create a classification job for specific buckets
aws macie2 create-classification-job \
  --job-type SCHEDULED \
  --name "weekly-pii-scan" \
  --schedule-frequency-details '{"weekly":{"dayOfWeek":"MONDAY"}}' \
  --s3-job-definition '{
    "bucketDefinitions": [{
      "accountId": "ACCOUNT_ID",
      "buckets": ["customer-data-bucket", "analytics-data-lake", "backup-bucket"]
    }],
    "scoping": {
      "includes": {
        "and": [{
          "simpleScopeTerm": {
            "key": "OBJECT_EXTENSION",
            "values": ["csv", "json", "parquet", "txt", "xlsx"],
            "comparator": "EQ"
          }
        }]
      }
    }
  }' \
  --managed-data-identifier-ids '["SSN","CREDIT_CARD_NUMBER","EMAIL_ADDRESS","AWS_CREDENTIALS","PHONE_NUMBER"]'
 
# Create custom data identifier for internal employee IDs
aws macie2 create-custom-data-identifier \
  --name "EmployeeID" \
  --regex "EMP-[0-9]{6}" \
  --description "Internal employee ID format"
 
# Check job status and results
aws macie2 list-classification-jobs \
  --query 'items[*].[name,jobStatus,statistics.approximateNumberOfObjectsToProcess]' \
  --output table

Step 2: Configure Google Cloud DLP API for Data Inspection

Use Google Cloud DLP to inspect and de-identify sensitive data across GCP resources.

# Inspect a Cloud Storage bucket for sensitive data
gcloud dlp inspect-content \
  --content-type=TEXT_PLAIN \
  --min-likelihood=LIKELY \
  --info-types=PHONE_NUMBER,EMAIL_ADDRESS,CREDIT_CARD_NUMBER,US_SOCIAL_SECURITY_NUMBER \
  --storage-type=CLOUD_STORAGE \
  --gcs-uri="gs://sensitive-data-bucket/data/*.csv"
 
# Create an inspection job for BigQuery
cat > dlp-job.json << 'EOF'
{
  "inspectJob": {
    "storageConfig": {
      "bigQueryOptions": {
        "tableReference": {
          "projectId": "PROJECT_ID",
          "datasetId": "customer_data",
          "tableId": "transactions"
        },
        "sampleMethod": "RANDOM_START",
        "rowsLimit": 10000
      }
    },
    "inspectConfig": {
      "infoTypes": [
        {"name": "CREDIT_CARD_NUMBER"},
        {"name": "US_SOCIAL_SECURITY_NUMBER"},
        {"name": "EMAIL_ADDRESS"},
        {"name": "PHONE_NUMBER"},
        {"name": "PERSON_NAME"}
      ],
      "minLikelihood": "LIKELY",
      "limits": {"maxFindingsPerRequest": 1000}
    },
    "actions": [{
      "saveFindings": {
        "outputConfig": {
          "table": {
            "projectId": "PROJECT_ID",
            "datasetId": "dlp_results",
            "tableId": "findings"
          }
        }
      }
    }]
  }
}
EOF
 
gcloud dlp jobs create --project=PROJECT_ID --body-from-file=dlp-job.json

Step 3: Implement Data De-identification with Cloud DLP

Configure de-identification transforms to mask, tokenize, or redact sensitive data.

# deidentify_pipeline.py - De-identify sensitive data using Google Cloud DLP
from google.cloud import dlp_v2
 
def deidentify_data(project_id, text):
    """De-identify PII in text using Cloud DLP."""
    client = dlp_v2.DlpServiceClient()
 
    inspect_config = {
        "info_types": [
            {"name": "EMAIL_ADDRESS"},
            {"name": "PHONE_NUMBER"},
            {"name": "CREDIT_CARD_NUMBER"},
            {"name": "US_SOCIAL_SECURITY_NUMBER"},
        ],
        "min_likelihood": dlp_v2.Likelihood.LIKELY,
    }
 
    deidentify_config = {
        "info_type_transformations": {
            "transformations": [
                {
                    "info_types": [{"name": "EMAIL_ADDRESS"}],
                    "primitive_transformation": {
                        "character_mask_config": {
                            "masking_character": "*",
                            "number_to_mask": 0,
                            "characters_to_ignore": [
                                {"common_characters_to_ignore": "PUNCTUATION"}
                            ],
                        }
                    },
                },
                {
                    "info_types": [{"name": "CREDIT_CARD_NUMBER"}],
                    "primitive_transformation": {
                        "crypto_replace_ffx_fpe_config": {
                            "crypto_key": {
                                "kms_wrapped": {
                                    "wrapped_key": "WRAPPED_KEY_BASE64",
                                    "crypto_key_name": "projects/PROJECT/locations/global/keyRings/dlp/cryptoKeys/tokenization",
                                }
                            },
                            "common_alphabet": "NUMERIC",
                        }
                    },
                },
                {
                    "info_types": [{"name": "US_SOCIAL_SECURITY_NUMBER"}],
                    "primitive_transformation": {
                        "redact_config": {}
                    },
                },
            ]
        }
    }
 
    item = {"value": text}
    parent = f"projects/{project_id}/locations/global"
 
    response = client.deidentify_content(
        request={
            "parent": parent,
            "deidentify_config": deidentify_config,
            "inspect_config": inspect_config,
            "item": item,
        }
    )
    return response.item.value

Step 4: Configure Azure Information Protection

Set up sensitivity labels and DLP policies in Microsoft Purview for Azure resources.

# Connect to Microsoft Purview compliance
Connect-IPPSSession
 
# Create sensitivity labels
New-Label -DisplayName "Confidential - PII" \
  -Name "Confidential-PII" \
  -Tooltip "Contains personally identifiable information" \
  -ContentType "File, Email"
 
New-Label -DisplayName "Highly Confidential - Financial" \
  -Name "HighlyConfidential-Financial" \
  -Tooltip "Contains financial data subject to PCI DSS" \
  -ContentType "File, Email"
 
# Create auto-labeling policy for Azure Storage
New-AutoSensitivityLabelPolicy -Name "Auto-Label-PII" \
  -ExchangeLocation All \
  -SharePointLocation All \
  -OneDriveLocation All \
  -Mode Enable
 
New-AutoSensitivityLabelRule -Policy "Auto-Label-PII" \
  -Name "Detect-SSN" \
  -ContentContainsSensitiveInformation @{
    Name = "U.S. Social Security Number (SSN)";
    MinCount = 1;
    MinConfidence = 85
  } \
  -ApplySensitivityLabel "Confidential-PII"
# Azure: Configure DLP policy for Storage accounts
az security assessment create \
  --name "storage-sensitive-data" \
  --assessed-resource-type "Microsoft.Storage/storageAccounts"
 
# Enable Microsoft Defender for Storage with sensitive data threat detection
az security pricing create --name StorageAccounts --tier standard \
  --subplan DefenderForStorageV2 \
  --extensions '[{"name":"SensitiveDataDiscovery","isEnabled":"True"}]'

Step 5: Integrate DLP into Data Pipelines

Add DLP scanning to ETL and data pipeline workflows to prevent sensitive data leakage.

# pipeline_dlp_gate.py - DLP gate for data pipelines
import boto3
import json
 
macie_client = boto3.client('macie2')
s3_client = boto3.client('s3')
 
def scan_pipeline_output(bucket, prefix):
    """Scan pipeline output data for sensitive content before promotion."""
    job_response = macie_client.create_classification_job(
        jobType='ONE_TIME',
        name=f'pipeline-scan-{prefix}',
        s3JobDefinition={
            'bucketDefinitions': [{
                'accountId': boto3.client('sts').get_caller_identity()['Account'],
                'buckets': [bucket]
            }],
            'scoping': {
                'includes': {
                    'and': [{
                        'simpleScopeTerm': {
                            'key': 'OBJECT_KEY',
                            'comparator': 'STARTS_WITH',
                            'values': [prefix]
                        }
                    }]
                }
            }
        },
        managedDataIdentifierSelector='ALL'
    )
 
    return job_response['jobId']
 
def check_scan_results(job_id):
    """Check if DLP scan found sensitive data."""
    response = macie_client.list_findings(
        findingCriteria={
            'criterion': {
                'classificationDetails.jobId': {'eq': [job_id]},
                'severity.description': {'eq': ['High', 'Critical']}
            }
        }
    )
    return len(response.get('findingIds', [])) > 0
 
def gate_decision(bucket, prefix):
    """DLP gate: block pipeline if sensitive data found."""
    job_id = scan_pipeline_output(bucket, prefix)
    has_sensitive_data = check_scan_results(job_id)
 
    if has_sensitive_data:
        return {
            'decision': 'BLOCK',
            'reason': 'Sensitive data detected in pipeline output',
            'action': 'Apply de-identification before promoting to production'
        }
    return {'decision': 'ALLOW', 'reason': 'No sensitive data detected'}

Step 6: Monitor DLP Findings and Generate Reports

Aggregate DLP findings across cloud providers and generate compliance reports.

# Macie: Get finding statistics
aws macie2 get-finding-statistics \
  --group-by "severity.description" \
  --finding-criteria '{"criterion":{"category":{"eq":["CLASSIFICATION"]}}}'
 
# Macie: List findings by sensitivity type
aws macie2 list-findings \
  --finding-criteria '{
    "criterion": {
      "classificationDetails.result.sensitiveData.category": {"eq": ["PERSONAL_INFORMATION"]},
      "severity.description": {"eq": ["High"]}
    }
  }' \
  --sort-criteria '{"attributeName": "updatedAt", "orderBy": "DESC"}'
 
# GCP DLP: List job results
gcloud dlp jobs list --project=PROJECT_ID --filter="state=DONE" \
  --format="table(name, createTime, inspectDetails.result.processedBytes, inspectDetails.result.totalEstimatedTransformations)"
 
# Export Macie findings to S3 for compliance reporting
aws macie2 create-findings-report \
  --finding-criteria '{"criterion":{"category":{"eq":["CLASSIFICATION"]}}}' \
  --sort-criteria '{"attributeName":"severity.score","orderBy":"DESC"}'

Key Concepts

Term Definition
Data Loss Prevention Security controls and technologies that detect and prevent unauthorized disclosure of sensitive data from cloud environments
Amazon Macie AWS service using machine learning to discover, classify, and protect sensitive data stored in S3 buckets
Google Cloud DLP GCP API for inspecting, classifying, and de-identifying sensitive data across Cloud Storage, BigQuery, and Datastore
Data De-identification Transforming sensitive data using masking, tokenization, encryption, or redaction to remove identifying characteristics while preserving utility
Sensitivity Label Classification tag applied to data (Confidential, Highly Confidential) that triggers DLP policy enforcement and access controls
Custom Data Identifier Organization-specific pattern (regex or keyword) added to DLP services to detect proprietary sensitive data formats

Tools & Systems

  • Amazon Macie: ML-powered sensitive data discovery and classification for S3 with automated finding generation
  • Google Cloud DLP API: Programmable API for inspecting, classifying, de-identifying, and redacting sensitive data
  • Microsoft Purview: Data governance platform with sensitivity labeling, auto-classification, and DLP policy enforcement
  • Azure Information Protection: Data classification and labeling service integrated with Microsoft 365 and Azure storage
  • Nightfall AI: Third-party cloud DLP tool supporting scanning across SaaS applications and cloud infrastructure

Common Scenarios

Scenario: Discovering PII in an Unprotected S3 Data Lake

Context: A compliance audit reveals that the analytics team's S3 data lake contains customer PII (names, emails, SSNs) in CSV files without encryption or access controls. The organization must classify all data and implement DLP controls.

Approach:

  1. Enable Macie and create a one-time classification job against the data lake bucket
  2. Review Macie findings to identify which objects contain PII and what types
  3. Create custom data identifiers for organization-specific formats (employee IDs, account numbers)
  4. Implement a weekly scheduled Macie job for ongoing discovery
  5. Build a data pipeline gate that scans new data before promotion to the data lake
  6. Apply de-identification transforms (masking SSNs, tokenizing emails) for analytics use cases
  7. Configure S3 bucket policies to restrict access to classified data to authorized roles only

Pitfalls: Macie charges per GB scanned. Large data lakes can generate significant costs. Use scoping rules to focus on high-risk object types (CSV, JSON, Parquet) and exclude known-safe formats (compressed archives, binary files). De-identification must preserve data utility for analytics while removing re-identification risk.

Output Format

Cloud DLP Compliance Report
==============================
Organization: Acme Corp
Scan Period: 2026-02-01 to 2026-02-23
Environments: AWS (12 buckets), GCP (3 datasets), Azure (5 storage accounts)
 
DATA DISCOVERY SUMMARY:
  Total objects/records scanned:    2,847,000
  Objects with sensitive data:        45,200 (1.6%)
  Unique sensitivity categories:      8
 
SENSITIVE DATA FINDINGS:
  PII (names, emails, phone):       23,400 objects
  Financial (credit cards, bank):     8,700 objects
  Health (PHI, medical records):      3,200 objects
  Credentials (API keys, tokens):     1,400 objects
  Government ID (SSN, passport):      5,800 objects
  Custom (employee ID, account):      2,700 objects
 
FINDINGS BY SEVERITY:
  Critical:    1,400 (exposed credentials)
  High:       14,200 (unprotected PII/PHI)
  Medium:     18,600 (standard PII)
  Low:        11,000 (non-sensitive patterns)
 
PROTECTION STATUS:
  Data with encryption at rest:       78%
  Data with access controls:          65%
  Data with sensitivity labels:       12%
  Pipeline data with DLP gates:       30%
 
REMEDIATION ACTIONS:
  Objects quarantined:                1,400
  De-identification applied:          8,200
  Access controls tightened:         14,200
  Sensitivity labels applied:        45,200
Source materials

References and resources

Everything below is rendered for inspection. Script files are read-only and never run.

References 1

api-reference.md2.1 KB

API Reference: Implementing Cloud DLP for Data Protection

Libraries

google-cloud-dlp (Google Cloud DLP)

  • Install: pip install google-cloud-dlp
  • Docs: https://cloud.google.com/dlp/docs/reference/libraries
  • DlpServiceClient() -- Create DLP client
  • inspect_content(parent, inspect_config, item) -- Scan content for sensitive data
  • deidentify_content(parent, deidentify_config, item) -- Mask/redact sensitive data
  • create_inspect_template() -- Reusable inspection configuration
  • create_dlp_job() -- Scan Cloud Storage, BigQuery, Datastore

boto3 -- Amazon Macie

GCP DLP Info Types

Category Info Types
PII PERSON_NAME, EMAIL_ADDRESS, PHONE_NUMBER, DATE_OF_BIRTH
Financial CREDIT_CARD_NUMBER, IBAN_CODE, SWIFT_CODE
US-specific US_SOCIAL_SECURITY_NUMBER, US_DRIVERS_LICENSE_NUMBER
Health US_HEALTHCARE_NPI, MEDICAL_RECORD_NUMBER

De-identification Methods

  • CharacterMaskConfig -- Replace characters with mask symbol
  • CryptoReplaceFfxFpeConfig -- Format-preserving encryption
  • RedactConfig -- Remove sensitive content entirely
  • ReplaceWithInfoTypeConfig -- Replace with info type name

Macie Finding Types

  • SensitiveData:S3Object/Personal -- PII found
  • SensitiveData:S3Object/Financial -- Financial data found
  • SensitiveData:S3Object/Credentials -- Credentials detected
  • Policy:IAMUser/S3BucketPublic -- Public bucket with sensitive data

External References

Scripts 1

agent.py7.2 KB
Display-only source. This catalog never executes bundled scripts.
#!/usr/bin/env python3
"""Cloud DLP agent for sensitive data discovery using Google Cloud DLP and AWS Macie."""

import json
import argparse
from datetime import datetime

try:
    import boto3
    from botocore.exceptions import ClientError
except ImportError:
    boto3 = None

try:
    from google.cloud import dlp_v2
except ImportError:
    dlp_v2 = None


INFO_TYPES_PII = [
    "PERSON_NAME", "EMAIL_ADDRESS", "PHONE_NUMBER", "US_SOCIAL_SECURITY_NUMBER",
    "CREDIT_CARD_NUMBER", "US_DRIVERS_LICENSE_NUMBER", "DATE_OF_BIRTH",
    "STREET_ADDRESS", "IP_ADDRESS", "PASSPORT",
]

INFO_TYPES_FINANCIAL = [
    "CREDIT_CARD_NUMBER", "IBAN_CODE", "SWIFT_CODE",
    "US_BANK_ROUTING_MICR", "US_EMPLOYER_IDENTIFICATION_NUMBER",
]

INFO_TYPES_HEALTH = [
    "US_HEALTHCARE_NPI", "US_DEA_NUMBER", "MEDICAL_RECORD_NUMBER",
]


def scan_text_with_gcp_dlp(project_id, text, info_types=None):
    """Scan text content for sensitive data using Google Cloud DLP."""
    if dlp_v2 is None:
        print("[!] Install google-cloud-dlp: pip install google-cloud-dlp")
        return None
    client = dlp_v2.DlpServiceClient()
    parent = f"projects/{project_id}"
    if info_types is None:
        info_types = INFO_TYPES_PII
    inspect_config = {
        "info_types": [{"name": it} for it in info_types],
        "min_likelihood": dlp_v2.Likelihood.LIKELY,
        "include_quote": True,
        "limits": {"max_findings_per_request": 50},
    }
    item = {"value": text}
    response = client.inspect_content(
        request={"parent": parent, "inspect_config": inspect_config, "item": item})
    findings = []
    for f in response.result.findings:
        findings.append({
            "info_type": f.info_type.name,
            "likelihood": dlp_v2.Likelihood(f.likelihood).name,
            "quote": f.quote[:50] + "..." if len(f.quote) > 50 else f.quote,
            "location": {"start": f.location.byte_range.start, "end": f.location.byte_range.end},
        })
    return findings


def deidentify_text_with_gcp(project_id, text, info_types=None):
    """De-identify sensitive data in text using masking."""
    if dlp_v2 is None:
        return None
    client = dlp_v2.DlpServiceClient()
    parent = f"projects/{project_id}"
    if info_types is None:
        info_types = INFO_TYPES_PII
    deidentify_config = {
        "info_type_transformations": {
            "transformations": [{
                "primitive_transformation": {
                    "character_mask_config": {"masking_character": "*", "number_to_mask": 0}
                },
                "info_types": [{"name": it} for it in info_types],
            }]
        }
    }
    inspect_config = {"info_types": [{"name": it} for it in info_types]}
    item = {"value": text}
    response = client.deidentify_content(
        request={"parent": parent, "deidentify_config": deidentify_config,
                 "inspect_config": inspect_config, "item": item})
    return response.item.value


def enable_macie(region="us-east-1"):
    """Enable Amazon Macie for S3 sensitive data discovery."""
    if boto3 is None:
        print("[!] Install boto3: pip install boto3")
        return None
    client = boto3.client("macie2", region_name=region)
    try:
        client.enable_macie(status="ENABLED", findingPublishingFrequency="FIFTEEN_MINUTES")
        return {"status": "enabled"}
    except ClientError as e:
        if "already enabled" in str(e).lower():
            return {"status": "already_enabled"}
        return {"error": str(e)}


def create_macie_classification_job(region, bucket_names, job_name):
    """Create a Macie classification job to scan S3 buckets."""
    if boto3 is None:
        return None
    client = boto3.client("macie2", region_name=region)
    try:
        resp = client.create_classification_job(
            jobType="ONE_TIME", name=job_name,
            s3JobDefinition={
                "bucketDefinitions": [{"accountId": boto3.client("sts").get_caller_identity()["Account"],
                                       "buckets": bucket_names}]
            },
            description=f"DLP scan for sensitive data in {', '.join(bucket_names)}")
        return {"job_id": resp["jobId"], "status": "created"}
    except ClientError as e:
        return {"error": str(e)}


def get_macie_findings(region="us-east-1", max_results=50):
    """Retrieve Macie findings for sensitive data discoveries."""
    if boto3 is None:
        return []
    client = boto3.client("macie2", region_name=region)
    try:
        resp = client.list_findings(
            sortCriteria={"attributeName": "severity.score", "orderBy": "DESC"},
            maxResults=max_results)
        finding_ids = resp.get("findingIds", [])
        if not finding_ids:
            return []
        details = client.get_findings(findingIds=finding_ids)
        return [{"id": f["id"], "type": f["type"], "severity": f["severity"]["score"],
                 "title": f["title"], "bucket": f.get("resourcesAffected", {}).get(
                     "s3Bucket", {}).get("name", ""),
                 "count": f.get("count", 1)}
                for f in details.get("findings", [])]
    except ClientError as e:
        return [{"error": str(e)}]


def run_dlp_report(project_id=None, region="us-east-1"):
    """Generate a DLP discovery report."""
    print(f"\n{'='*60}")
    print(f"  CLOUD DLP DATA PROTECTION REPORT")
    print(f"  Generated: {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC")
    print(f"{'='*60}\n")

    if boto3:
        print(f"--- AWS MACIE STATUS ---")
        macie_status = enable_macie(region)
        print(f"  Macie: {macie_status}")
        findings = get_macie_findings(region)
        print(f"  Findings: {len(findings)}")
        for f in findings[:5]:
            print(f"  [{f.get('severity', 'N/A')}] {f.get('title', 'N/A')} - {f.get('bucket', 'N/A')}")

    if dlp_v2 and project_id:
        print(f"\n--- GCP DLP SCAN ---")
        sample = "Contact John Doe at john@example.com, SSN 123-45-6789, CC 4111-1111-1111-1111"
        findings = scan_text_with_gcp_dlp(project_id, sample)
        if findings:
            for f in findings:
                print(f"  [{f['likelihood']}] {f['info_type']}: {f['quote']}")

    print(f"\n{'='*60}\n")


def main():
    parser = argparse.ArgumentParser(description="Cloud DLP Data Protection Agent")
    parser.add_argument("--gcp-project", help="GCP project ID for DLP API")
    parser.add_argument("--aws-region", default="us-east-1", help="AWS region for Macie")
    parser.add_argument("--scan-text", help="Text to scan for sensitive data")
    parser.add_argument("--scan-buckets", nargs="+", help="S3 bucket names to scan with Macie")
    parser.add_argument("--report", action="store_true", help="Generate DLP report")
    parser.add_argument("--output", help="Save report to JSON")
    args = parser.parse_args()

    if args.scan_text and args.gcp_project:
        findings = scan_text_with_gcp_dlp(args.gcp_project, args.scan_text)
        print(json.dumps(findings, indent=2))
    elif args.scan_buckets:
        result = create_macie_classification_job(args.aws_region, args.scan_buckets, "dlp-agent-scan")
        print(json.dumps(result, indent=2))
    elif args.report:
        run_dlp_report(args.gcp_project, args.aws_region)
    else:
        parser.print_help()


if __name__ == "__main__":
    main()
Keep exploring