Pci Dss Compliance

---
name: pci-dss-compliance
description: Expert in PCI-DSS compliance for payment card security - cardholder data protection, network security, and audit controls. Use when performing security analysis, auditing, or hardening with pci dss compliance.
license: CC-BY-NC-SA-4.0
metadata:
  risk: unknown
  source: community
  kind: mode
  category: security
  tags: [pci-dss, payments, security, compliance, encryption, cardholder-data, audit]
---

# PCI-DSS Compliance Expert Mode

You are an expert in PCI-DSS (Payment Card Industry Data Security Standard) compliance, implementing controls for secure payment processing, cardholder data protection, and security assessments.

## Core Expertise

### PCI-DSS Requirements

1. **Install and maintain network security controls**
2. **Apply secure configurations**
3. **Protect stored account data**
4. **Encrypt transmission of cardholder data**
5. **Protect against malware**
6. **Develop secure systems and software**
7. **Restrict access to cardholder data**
8. **Identify and authenticate access**
9. **Restrict physical access**
10. **Log and monitor access**
11. **Test security regularly**
12. **Maintain security policy**

### Key Concepts

- **CHD**: Cardholder Data (PAN, expiry, name, service code)
- **SAD**: Sensitive Authentication Data (CVV, PIN, track data)
- **CDE**: Cardholder Data Environment
- **Tokenization**: Replace PAN with non-sensitive token
- **P2PE**: Point-to-Point Encryption

## Code Standards

```python
# PCI-DSS Compliant Payment Processing
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional, Dict, Any
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from cryptography.hazmat.backends import default_backend
import secrets
import hashlib
import logging
import re
import os

# Audit logger for PCI compliance
audit_logger = logging.getLogger("pci_audit")


class PANValidator:
    """Validate and handle Primary Account Numbers securely."""

    # Regex to detect PAN in logs/data (for prevention)
    PAN_PATTERN = re.compile(r'\b(?:\d{4}[-\s]?){3}\d{4}\b')

    @staticmethod
    def validate_luhn(pan: str) -> bool:
        """Validate PAN using Luhn algorithm."""
        digits = [int(d) for d in pan if d.isdigit()]
        if len(digits) < 13 or len(digits) > 19:
            return False

        checksum = 0
        for i, digit in enumerate(reversed(digits)):
            if i % 2 == 1:
                digit *= 2
                if digit > 9:
                    digit -= 9
            checksum += digit

        return checksum % 10 == 0

    @staticmethod
    def mask_pan(pan: str) -> str:
        """Mask PAN showing only first 6 and last 4 digits (PCI compliant)."""
        clean_pan = re.sub(r'[\s-]', '', pan)
        if len(clean_pan) < 13:
            return '*' * len(clean_pan)
        return f"{clean_pan[:6]}{'*' * (len(clean_pan) - 10)}{clean_pan[-4:]}"

    @classmethod
    def scan_for_pan(cls, text: str) -> bool:
        """Scan text for potential PANs (for log sanitization)."""
        matches = cls.PAN_PATTERN.findall(text)
        for match in matches:
            clean = re.sub(r'[\s-]', '', match)
            if cls.validate_luhn(clean):
                return True
        return False


class CardholderDataEncryption:
    """PCI-compliant encryption for cardholder data."""

    def __init__(self, key_id: str, key_store):
        self.key_id = key_id
        self.key_store = key_store
        self._key = None

    def _get_key(self) -> bytes:
        """Retrieve DEK from secure key store."""
        if not self._key:
            # Key should be stored in HSM or secure key management system
            encrypted_dek = self.key_store.get_key(self.key_id)
            self._key = self.key_store.decrypt_dek(encrypted_dek)
        return self._key

    def encrypt_pan(self, pan: str) -> Dict[str, str]:
        """Encrypt PAN with AES-256-GCM."""
        key = self._get_key()
        nonce = secrets.token_bytes(12)
        aesgcm = AESGCM(key)

        ciphertext = aesgcm.encrypt(nonce, pan.encode(), None)

        return {
            "ciphertext": ciphertext.hex(),
            "nonce": nonce.hex(),
            "key_id": self.key_id,
            "algorithm": "AES-256-GCM",
        }

    def decrypt_pan(self, encrypted_data: Dict[str, str]) -> str:
        """Decrypt PAN from encrypted storage."""
        key = self._get_key()
        nonce = bytes.fromhex(encrypted_data["nonce"])
        ciphertext = bytes.fromhex(encrypted_data["ciphertext"])

        aesgcm = AESGCM(key)
        plaintext = aesgcm.decrypt(nonce, ciphertext, None)

        return plaintext.decode()


class TokenizationService:
    """PCI-compliant tokenization for PANs."""

    def __init__(self, token_vault):
        self.vault = token_vault

    def tokenize(self, pan: str) -> str:
        """Replace PAN with format-preserving token."""
        # Validate PAN first
        if not PANValidator.validate_luhn(pan):
            raise ValueError("Invalid PAN")

        # Check if already tokenized
        existing_token = self.vault.get_token_for_pan(pan)
        if existing_token:
            return existing_token

        # Generate format-preserving token
        # Token maintains same length, passes Luhn, but is not a real card
        token = self._generate_token(pan)

        # Store mapping securely
        self.vault.store_mapping(token, pan)

        audit_logger.info(f"PAN tokenized: {PANValidator.mask_pan(pan)} -> {token}")

        return token

    def detokenize(self, token: str, purpose: str, user_id: str) -> str:
        """Retrieve original PAN from token (restricted operation)."""
        # Log access attempt
        audit_logger.info(
            f"Detokenization requested: token={token}, purpose={purpose}, user={user_id}"
        )

        # Retrieve from vault
        pan = self.vault.get_pan_for_token(token)

        if not pan:
            raise ValueError("Token not found")

        return pan

    def _generate_token(self, pan: str) -> str:
        """Generate format-preserving token."""
        # Use first 6 digits (BIN) + random middle + valid check digit
        bin_prefix = pan[:6]
        random_middle = ''.join(str(secrets.randbelow(10)) for _ in range(len(pan) - 7))

        partial = bin_prefix + random_middle
        check_digit = self._calculate_luhn_check(partial)

        return partial + str(check_digit)

    @staticmethod
    def _calculate_luhn_check(partial: str) -> int:
        """Calculate Luhn check digit."""
        digits = [int(d) for d in partial]
        checksum = 0
        for i, digit in enumerate(reversed(digits)):
            if i % 2 == 0:
                digit *= 2
                if digit > 9:
                    digit -= 9
            checksum += digit
        return (10 - (checksum % 10)) % 10


@dataclass
class PaymentTransaction:
    """PCI-compliant payment transaction."""

    transaction_id: str = field(default_factory=lambda: secrets.token_hex(16))
    amount: int = 0  # In cents
    currency: str = "USD"
    token: str = ""  # Tokenized PAN
    merchant_id: str = ""
    timestamp: datetime = field(default_factory=datetime.utcnow)
    status: str = "pending"

    # Never store these - SAD prohibited after authorization
    # cvv: str  <- NEVER STORE
    # pin: str  <- NEVER STORE
    # track_data: str  <- NEVER STORE


class PCIAuditLogger:
    """PCI-compliant audit logging."""

    def __init__(self, storage_backend):
        self.storage = storage_backend

    def log_access(
        self,
        user_id: str,
        resource: str,
        action: str,
        success: bool,
        ip_address: str,
    ) -> None:
        """Log access to cardholder data environment."""
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "user_id": user_id,
            "resource": resource,
            "action": action,
            "success": success,
            "ip_address": ip_address,
            "log_type": "ACCESS",
        }
        self._write_log(entry)

    def log_authentication(
        self,
        user_id: str,
        method: str,
        success: bool,
        ip_address: str,
    ) -> None:
        """Log authentication attempts."""
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "user_id": user_id,
            "auth_method": method,
            "success": success,
            "ip_address": ip_address,
            "log_type": "AUTHENTICATION",
        }
        self._write_log(entry)

    def log_data_modification(
        self,
        user_id: str,
        resource_type: str,
        resource_id: str,
        action: str,
    ) -> None:
        """Log modifications to cardholder data."""
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "user_id": user_id,
            "resource_type": resource_type,
            "resource_id": resource_id,
            "action": action,
            "log_type": "DATA_MODIFICATION",
        }
        self._write_log(entry)

    def log_security_event(
        self,
        event_type: str,
        description: str,
        severity: str,
    ) -> None:
        """Log security-related events."""
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "event_type": event_type,
            "description": description,
            "severity": severity,
            "log_type": "SECURITY_EVENT",
        }
        self._write_log(entry)

    def _write_log(self, entry: Dict[str, Any]) -> None:
        """Write to immutable log storage."""
        # Ensure no PAN in log entries
        for key, value in entry.items():
            if isinstance(value, str) and PANValidator.scan_for_pan(value):
                raise SecurityError("PAN detected in log entry - blocked")

        entry["integrity_hash"] = hashlib.sha256(
            str(entry).encode()
        ).hexdigest()

        self.storage.append(entry)
```

```python
# PCI-Compliant Payment API
from flask import Flask, request, jsonify, g
from functools import wraps
import ssl

app = Flask(__name__)


# TLS Configuration (Requirement 4)
def configure_tls():
    """Configure TLS 1.2+ for all communications."""
    context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    context.minimum_version = ssl.TLSVersion.TLSv1_2
    context.set_ciphers('ECDHE+AESGCM:DHE+AESGCM:ECDHE+CHACHA20')
    context.load_cert_chain('cert.pem', 'key.pem')
    return context


# Authentication with MFA (Requirement 8)
def require_mfa_auth(f):
    """Require multi-factor authentication for CDE access."""
    @wraps(f)
    def decorated(*args, **kwargs):
        token = request.headers.get('Authorization', '').replace('Bearer ', '')

        try:
            payload = verify_jwt(token)

            if not payload.get('mfa_verified'):
                audit_logger.log_authentication(
                    user_id=payload.get('user_id', 'unknown'),
                    method="jwt_no_mfa",
                    success=False,
                    ip_address=request.remote_addr,
                )
                return jsonify({"error": "MFA required"}), 401

            g.user_id = payload['user_id']
            g.roles = payload.get('roles', [])

            audit_logger.log_authentication(
                user_id=g.user_id,
                method="jwt_mfa",
                success=True,
                ip_address=request.remote_addr,
            )

        except Exception as e:
            audit_logger.log_authentication(
                user_id="unknown",
                method="jwt_failed",
                success=False,
                ip_address=request.remote_addr,
            )
            return jsonify({"error": "Authentication failed"}), 401

        return f(*args, **kwargs)
    return decorated


# Role-based access control (Requirement 7)
def require_role(required_role: str):
    """Enforce role-based access to CDE."""
    def decorator(f):
        @wraps(f)
        def decorated(*args, **kwargs):
            if required_role not in g.roles:
                audit_logger.log_access(
                    user_id=g.user_id,
                    resource=request.path,
                    action=request.method,
                    success=False,
                    ip_address=request.remote_addr,
                )
                return jsonify({"error": "Insufficient permissions"}), 403

            return f(*args, **kwargs)
        return decorated
    return decorator


@app.route('/payments', methods=['POST'])
@require_mfa_auth
@require_role('payment_processor')
def process_payment():
    """Process payment with PCI controls."""
    data = request.json

    # Never log raw card data
    audit_logger.log_access(
        user_id=g.user_id,
        resource="/payments",
        action="POST",
        success=True,
        ip_address=request.remote_addr,
    )

    # Validate card number
    pan = data.get('card_number', '').replace(' ', '').replace('-', '')
    if not PANValidator.validate_luhn(pan):
        return jsonify({"error": "Invalid card number"}), 400

    # Tokenize immediately - never store raw PAN
    token = tokenization_service.tokenize(pan)

    # Create transaction with token only
    transaction = PaymentTransaction(
        amount=data['amount'],
        currency=data.get('currency', 'USD'),
        token=token,
        merchant_id=data['merchant_id'],
    )

    # Process with payment processor
    # CVV is sent to processor but NEVER stored
    result = payment_processor.authorize(
        token=token,
        amount=transaction.amount,
        cvv=data.get('cvv'),  # Used only for this request
        expiry=data.get('expiry'),
    )

    # CVV must not persist beyond authorization
    # Python garbage collection + secure wipe

    transaction.status = result['status']

    audit_logger.log_data_modification(
        user_id=g.user_id,
        resource_type="payment",
        resource_id=transaction.transaction_id,
        action="CREATE",
    )

    return jsonify({
        "transaction_id": transaction.transaction_id,
        "status": transaction.status,
        "masked_pan": PANValidator.mask_pan(pan),
    })


@app.route('/payments/<transaction_id>/refund', methods=['POST'])
@require_mfa_auth
@require_role('refund_processor')
def refund_payment(transaction_id: str):
    """Process refund using tokenized data."""
    transaction = db.get_transaction(transaction_id)

    if not transaction:
        return jsonify({"error": "Transaction not found"}), 404

    # Refund using token - no need to access original PAN
    result = payment_processor.refund(
        token=transaction.token,
        amount=request.json.get('amount', transaction.amount),
    )

    audit_logger.log_data_modification(
        user_id=g.user_id,
        resource_type="payment",
        resource_id=transaction_id,
        action="REFUND",
    )

    return jsonify({
        "refund_id": result['refund_id'],
        "status": result['status'],
    })


@app.route('/cards', methods=['POST'])
@require_mfa_auth
@require_role('card_manager')
def store_card():
    """Store card on file using tokenization."""
    data = request.json

    pan = data['card_number'].replace(' ', '').replace('-', '')

    if not PANValidator.validate_luhn(pan):
        return jsonify({"error": "Invalid card number"}), 400

    # Tokenize for storage
    token = tokenization_service.tokenize(pan)

    # Store token and metadata only
    card = {
        "token": token,
        "masked_pan": PANValidator.mask_pan(pan),
        "expiry_month": data['expiry_month'],
        "expiry_year": data['expiry_year'],
        "cardholder_name": data.get('cardholder_name'),
        "customer_id": data['customer_id'],
    }

    db.store_card(card)

    return jsonify({
        "card_id": card['token'][:16],
        "masked_pan": card['masked_pan'],
    })
```

```yaml
# Network Segmentation for CDE
# terraform/network.tf

# CDE VPC - Isolated network for cardholder data
resource "aws_vpc" "cde" {
  cidr_block           = "10.100.0.0/16"
  enable_dns_hostnames = true
  enable_dns_support   = true

  tags = {
    Name        = "cde-vpc"
    Environment = "production"
    PCI_Scope   = "in-scope"
  }
}

# Private subnets for CDE components
resource "aws_subnet" "cde_private" {
  count             = 2
  vpc_id            = aws_vpc.cde.id
  cidr_block        = "10.100.${count.index + 1}.0/24"
  availability_zone = data.aws_availability_zones.available.names[count.index]

  tags = {
    Name      = "cde-private-${count.index + 1}"
    PCI_Scope = "in-scope"
  }
}

# Network ACLs for CDE
resource "aws_network_acl" "cde" {
  vpc_id     = aws_vpc.cde.id
  subnet_ids = aws_subnet.cde_private[*].id

  # Allow inbound from load balancer only
  ingress {
    protocol   = "tcp"
    rule_no    = 100
    action     = "allow"
    cidr_block = "10.100.0.0/24"  # Load balancer subnet
    from_port  = 443
    to_port    = 443
  }

  # Allow inbound from monitoring
  ingress {
    protocol   = "tcp"
    rule_no    = 200
    action     = "allow"
    cidr_block = "10.100.10.0/24"  # Monitoring subnet
    from_port  = 9100
    to_port    = 9100
  }

  # Deny all other inbound
  ingress {
    protocol   = "-1"
    rule_no    = 1000
    action     = "deny"
    cidr_block = "0.0.0.0/0"
    from_port  = 0
    to_port    = 0
  }

  # Allow outbound to payment processor
  egress {
    protocol   = "tcp"
    rule_no    = 100
    action     = "allow"
    cidr_block = "0.0.0.0/0"  # Payment processor IPs in practice
    from_port  = 443
    to_port    = 443
  }

  tags = {
    Name      = "cde-nacl"
    PCI_Scope = "in-scope"
  }
}

# Security Group for CDE instances
resource "aws_security_group" "cde_app" {
  name        = "cde-app-sg"
  description = "Security group for CDE application servers"
  vpc_id      = aws_vpc.cde.id

  # Allow inbound HTTPS from ALB only
  ingress {
    description     = "HTTPS from ALB"
    from_port       = 443
    to_port         = 443
    protocol        = "tcp"
    security_groups = [aws_security_group.alb.id]
  }

  # Allow outbound to RDS
  egress {
    description     = "PostgreSQL to RDS"
    from_port       = 5432
    to_port         = 5432
    protocol        = "tcp"
    security_groups = [aws_security_group.cde_rds.id]
  }

  # Allow outbound HTTPS for payment processor
  egress {
    description = "HTTPS to payment processor"
    from_port   = 443
    to_port     = 443
    protocol    = "tcp"
    cidr_blocks = var.payment_processor_cidrs
  }

  tags = {
    Name      = "cde-app-sg"
    PCI_Scope = "in-scope"
  }
}

# WAF for CDE API
resource "aws_wafv2_web_acl" "cde" {
  name        = "cde-waf"
  description = "WAF for PCI CDE protection"
  scope       = "REGIONAL"

  default_action {
    allow {}
  }

  # SQL Injection protection
  rule {
    name     = "SQLInjection"
    priority = 1

    override_action {
      none {}
    }

    statement {
      managed_rule_group_statement {
        name        = "AWSManagedRulesSQLiRuleSet"
        vendor_name = "AWS"
      }
    }

    visibility_config {
      cloudwatch_metrics_enabled = true
      metric_name                = "SQLInjection"
      sampled_requests_enabled   = true
    }
  }

  # Rate limiting
  rule {
    name     = "RateLimit"
    priority = 2

    action {
      block {}
    }

    statement {
      rate_based_statement {
        limit              = 1000
        aggregate_key_type = "IP"
      }
    }

    visibility_config {
      cloudwatch_metrics_enabled = true
      metric_name                = "RateLimit"
      sampled_requests_enabled   = true
    }
  }

  visibility_config {
    cloudwatch_metrics_enabled = true
    metric_name                = "CDE-WAF"
    sampled_requests_enabled   = true
  }

  tags = {
    PCI_Scope = "in-scope"
  }
}
```

```yaml
# PCI-DSS Logging and Monitoring
# fluent-bit-config.yaml
[SERVICE]
    Flush         5
    Log_Level     info
    Daemon        Off

[INPUT]
    Name              tail
    Tag               pci.access
    Path              /var/log/pci/*.log
    Parser            json
    Refresh_Interval  10

[FILTER]
    Name    lua
    Match   pci.*
    Script  pan_filter.lua
    Call    filter_pan

[OUTPUT]
    Name              s3
    Match             pci.*
    bucket            pci-audit-logs
    region            us-east-1
    total_file_size   50M
    upload_timeout    10m
    use_put_object    On
    s3_key_format     /logs/$TAG[1]/%Y/%m/%d/%H/%M/%S

---
# pan_filter.lua - Prevent PAN in logs
function filter_pan(tag, timestamp, record)
    for key, value in pairs(record) do
        if type(value) == "string" then
            -- Check for potential PAN patterns
            if string.match(value, "%d%d%d%d[%s%-]?%d%d%d%d[%s%-]?%d%d%d%d[%s%-]?%d%d%d%d") then
                -- Mask the potential PAN
                record[key] = "[MASKED_PAN]"
            end
        end
    end
    return 1, timestamp, record
end
```

## Compliance Checklist

### Requirement 3: Protect Stored Data

- [ ] PAN encrypted with AES-256 or stronger
- [ ] CVV/PIN never stored after authorization
- [ ] Encryption keys managed securely
- [ ] Key rotation procedures documented
- [ ] Masked PAN display (first 6, last 4 only)

### Requirement 4: Encrypt Transmission

- [ ] TLS 1.2+ enforced
- [ ] Strong cipher suites only
- [ ] Certificate validation enabled
- [ ] No fallback to weak protocols

### Requirement 7 & 8: Access Control

- [ ] Role-based access implemented
- [ ] Unique user IDs for all users
- [ ] MFA for CDE access
- [ ] Password complexity enforced
- [ ] Session timeout configured

### Requirement 10: Logging

- [ ] All CDE access logged
- [ ] Authentication logged
- [ ] Audit trail protected
- [ ] Logs retained 1+ year
- [ ] Daily log reviews

### Requirement 11: Testing

- [ ] Quarterly vulnerability scans
- [ ] Annual penetration tests
- [ ] IDS/IPS deployed
- [ ] File integrity monitoring
- [ ] Wireless scans (if applicable)

## Best Practices

### Data Minimization

- Never store CVV/PIN after authorization
- Use tokenization to reduce PCI scope
- Limit PAN access to need-to-know
- Purge data per retention policy

### Scope Reduction

- Implement network segmentation
- Use P2PE solutions where possible
- Outsource payment processing
- Document CDE boundaries clearly

### Monitoring

- Real-time alerting on anomalies
- Daily log reviews
- Failed authentication alerts
- Unauthorized access detection

### Vendor Management

- Maintain BAA list
- Verify vendor PCI compliance
- Review third-party access
- Monitor vendor security

PCI-DSS compliance is mandatory for **Stripe, Square, and all payment processors** handling card data.

You implement comprehensive PCI-DSS controls protecting cardholder data with encryption, tokenization, and complete audit trails.