AI Marketing Operations: Data Security and Permission Models

import re
from typing import Tuple, Dict

# =====================================================================
# ENTERPRISE MARKETING OPERATIONS: PII SCRUBBING MIDDLEWARE
# =====================================================================

class MarketingPIIScrubber:
    def __init__(self):
        # High-precision regular expressions for common PII patterns
        self.patterns = {
            "EMAIL": r'[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+',
            "PHONE": r'(\+?\d{1,3}[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}',
            "IP_ADDR": r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}',
            "CREDIT_CARD": r'(?:\d{4}[-\s]?){3}\d{4}'
        }

    def sanitize_payload(self, text: str) -> Tuple[str, Dict[str, str]]:
        '''
        Replaces sensitive customer PII with reversible synthetic tokens.
        Returns the sanitized text and an encrypted mapping vault.
        '''
        mapping_vault = {}
        sanitized = text

        for pii_type, regex in self.patterns.items():
            counter = 1
            matches = list(set(re.findall(regex, sanitized)))
            for match in matches:
                # If regex returned tuples (from capturing groups), select full string
                match_val = match[0] if isinstance(match, tuple) else match
                if not match_val:
                    continue
                placeholder = f"[{pii_type}_{counter}]"
                mapping_vault[placeholder] = match_val
                sanitized = sanitized.replace(match_val, placeholder)
                counter += 1

        return sanitized, mapping_vault

    def restore_payload(self, sanitized_text: str, mapping_vault: Dict[str, str]) -> str:
        '''
        Restores original customer values locally after receiving LLM response.
        '''
        restored = sanitized_text
        for placeholder, original in mapping_vault.items():
            restored = restored.replace(placeholder, original)
        return restored

# Example Execution
if __name__ == "__main__":
    scrubber = MarketingPIIScrubber()
    
    raw_lead_inbound = (
        "Customer Johnathan Doe ([email protected], Phone: +1-555-839-2041) "
        "reported that his recent ad campaign payment from IP 192.168.1.45 failed on Card 4111-2222-3333-4444. "
        "Please generate an executive customer support reply addressing this incident."
    )

    print("--- RAW MARKETING PAYLOAD (UNSAFE FOR PUBLIC LLM) ---")
    print(raw_lead_inbound)

    sanitized, vault = scrubber.sanitize_payload(raw_lead_inbound)
    print("
--- SANITIZED PAYLOAD DISPATCHED TO LLM ---")
    print(sanitized)
    print(f"
Local Vault Tokens Stored: {len(vault)} items")

    # Simulate LLM Response using sanitized tokens
    simulated_llm_response = (
        "Dear [PERSON], we have investigated the billing issue reported from [IP_ADDR_1]. "
        "We have securely updated your contact record for [EMAIL_1] and can confirm that "
        "card ending in 4444 will not be charged again."
    )

    restored_response = scrubber.restore_payload(simulated_llm_response, vault)
    print("
--- RESTORED RESPONSE DELIVERED TO CUSTOMER ---")
    print(restored_response)

# TERRAFORM (HCL): ZERO-TRUST MARKETING OPERATIONS CREDENTIAL VAULT
provider "aws" {
  region = "us-east-1"
}

# KMS Key for Marketing Secret Encryption with Automated 365-Day Rotation
resource "aws_kms_key" "marketing_vault_key" {
  description             = "KMS Key for Growth Marketing API Tokens and CAPI Credentials"
  deletion_window_in_days = 30
  enable_key_rotation     = true

  tags = {
    Environment = "Production"
    Team        = "Growth-Engineering"
    Compliance  = "SOC2-Type-II"
  }
}

# AWS Secrets Manager Secret for Meta Marketing API Long-Lived Token
resource "aws_secretsmanager_secret" "meta_ads_api_token" {
  name                    = "marketing/production/meta_ads_api_token"
  kms_key_id              = aws_kms_key.marketing_vault_key.arn
  recovery_window_in_days = 7

  tags = {
    Role = "Ad-Automation-Runner"
  }
}

# Scoped IAM Policy: Media Buyer Automation Service Can Only Read, Not Delete
resource "aws_iam_policy" "marketing_runner_least_privilege" {
  name        = "MarketingAutomationRunnerPolicy"
  description = "Allows ephemeral read-only access to scoped marketing secrets"

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "secretsmanager:GetSecretValue",
          "secretsmanager:DescribeSecret"
        ]
        Resource = aws_secretsmanager_secret.meta_ads_api_token.arn
      },
      {
        Effect = "Allow"
        Action = [
          "kms:Decrypt"
        ]
        Resource = aws_kms_key.marketing_vault_key.arn
      }
    ]
  })
}