PCI-DSS Compliance: Complete Implementation Guide for Payment Systems

# PCI-DSS compliant network architecture

class PCI_DSS_Architecture:
    """
    Compliant network design:
    
    Internet
        │
        ▼
    ┌─────────────────┐
    │  Firewall       │ (Requirement 1)
    └────────┬────────┘
             │
    ┌────────▼────────────────────────────┐
    │  DMZ (Demilitarized Zone)           │
    │  - Web servers                      │
    │  - API gateways                     │
    │  - Load balancers                   │
    └────────┬────────────────────────────┘
             │
    ┌────────▼────────────────────────────┐
    │  Internal Network (Segmented)       │
    │  - Payment processing servers       │
    │  - Database servers                 │
    │  - Admin workstations               │
    └────────┬────────────────────────────┘
             │
    ┌────────▼────────────────────────────┐
    │  Cardholder Data Environment (CDE)  │
    │  - Isolated, encrypted              │
    │  - Limited access                   │
    │  - Monitored 24/7                   │
    └─────────────────────────────────────┘
    """
    pass

from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2
import base64
import hashlib

class CardholderDataProtection:
    """Implement PCI-DSS data protection requirements"""
    
    def __init__(self, encryption_key: str):
        # Derive encryption key
        kdf = PBKDF2(
            algorithm=hashes.SHA256(),
            length=32,
            salt=b'pci_dss_salt_123',
            iterations=100000,
        )
        key = base64.urlsafe_b64encode(kdf.derive(encryption_key.encode()))
        self.cipher = Fernet(key)
    
    def tokenize_card(self, card_number: str) -> str:
        """Replace card number with token (Requirement 3)"""
        
        # Generate unique token
        token = f"tok_{hashlib.sha256(card_number.encode()).hexdigest()[:16]}"
        
        # Store mapping securely (encrypted)
        encrypted_card = self.cipher.encrypt(card_number.encode())
        
        # Store in secure vault (not in main database)
        # vault.store(token, encrypted_card)
        
        return token
    
    def encrypt_card_data(self, card_data: dict) -> dict:
        """Encrypt sensitive card data (Requirement 3)"""
        
        sensitive_fields = ['pan', 'cvv', 'expiration_date']
        encrypted_data = card_data.copy()
        
        for field in sensitive_fields:
            if field in encrypted_data:
                value = str(encrypted_data[field])
                encrypted_value = self.cipher.encrypt(value.encode())
                encrypted_data[field] = encrypted_value.decode()
        
        return encrypted_data
    
    def hash_card_number(self, card_number: str) -> str:
        """Hash card number for storage (Requirement 3)"""
        
        # Use SHA-256 with salt
        salt = b'pci_dss_salt'
        hashed = hashlib.pbkdf2_hmac(
            'sha256',
            card_number.encode(),
            salt,
            100000
        )
        
        return hashed.hex()
    
    def mask_card_number(self, card_number: str) -> str:
        """Mask card number for display (Requirement 3)"""
        
        # Show only last 4 digits
        return f"****-****-****-{card_number[-4:]}"
    
    def validate_encryption(self, data: dict) -> bool:
        """Validate encryption is in place (Requirement 3)"""
        
        # Check that sensitive fields are encrypted
        sensitive_fields = ['pan', 'cvv', 'expiration_date']
        
        for field in sensitive_fields:
            if field in data:
                # Encrypted data should start with specific prefix
                if not isinstance(data[field], str) or not data[field].startswith('gAAAAAA'):
                    return False
        
        return True

# Usage
protection = CardholderDataProtection('your_master_key')

# Tokenize card
token = protection.tokenize_card('4532-1234-5678-9010')
print(f"Token: {token}")

# Encrypt card data
card_data = {
    'pan': '4532123456789010',
    'cvv': '123',
    'expiration_date': '12/25'
}

encrypted = protection.encrypt_card_data(card_data)
print(f"Encrypted: {encrypted}")

# Mask for display
masked = protection.mask_card_number('4532123456789010')
print(f"Masked: {masked}")

from enum import Enum
from datetime import datetime, timedelta

class UserRole(Enum):
    ADMIN = "admin"
    PAYMENT_PROCESSOR = "payment_processor"
    AUDITOR = "auditor"
    DEVELOPER = "developer"

class AccessControl:
    """Implement PCI-DSS access control (Requirement 7 & 8)"""
    
    def __init__(self, db_connection):
        self.db = db_connection
    
    def authenticate_user(self, username: str, password: str,
                         mfa_code: str) -> bool:
        """Authenticate with MFA (Requirement 8)"""
        
        # Verify username and password
        user = self.db.query(
            "SELECT * FROM users WHERE username = ?",
            (username,)
        )
        
        if not user:
            return False
        
        # Verify password (use bcrypt)
        import bcrypt
        if not bcrypt.checkpw(password.encode(), user[0]['password_hash']):
            return False
        
        # Verify MFA
        if not self._verify_mfa(user[0]['id'], mfa_code):
            return False
        
        # Log authentication
        self._log_access('authentication_success', username)
        
        return True
    
    def _verify_mfa(self, user_id: str, mfa_code: str) -> bool:
        """Verify MFA code"""
        
        import pyotp
        
        user = self.db.query("SELECT * FROM users WHERE id = ?", (user_id,))
        secret = user[0]['mfa_secret']
        
        totp = pyotp.TOTP(secret)
        return totp.verify(mfa_code)
    
    def check_access(self, user_id: str, resource: str,
                    action: str) -> bool:
        """Check if user has access (Requirement 7)"""
        
        user = self.db.query("SELECT * FROM users WHERE id = ?", (user_id,))
        role = user[0]['role']
        
        # Define role-based permissions
        permissions = {
            UserRole.ADMIN: ['read', 'write', 'delete', 'audit'],
            UserRole.PAYMENT_PROCESSOR: ['read', 'write'],
            UserRole.AUDITOR: ['read', 'audit'],
            UserRole.DEVELOPER: ['read']
        }
        
        allowed_actions = permissions.get(UserRole[role], [])
        
        if action not in allowed_actions:
            self._log_access('access_denied', user_id, resource, action)
            return False
        
        self._log_access('access_granted', user_id, resource, action)
        return True
    
    def enforce_password_policy(self, password: str) -> bool:
        """Enforce strong password policy (Requirement 8)"""
        
        # Minimum 8 characters
        if len(password) < 8:
            return False
        
        # Must contain uppercase, lowercase, number, special char
        has_upper = any(c.isupper() for c in password)
        has_lower = any(c.islower() for c in password)
        has_digit = any(c.isdigit() for c in password)
        has_special = any(c in '!@#$%^&*' for c in password)
        
        return has_upper and has_lower and has_digit and has_special
    
    def _log_access(self, action: str, user_id: str,
                   resource: str = None, action_type: str = None):
        """Log all access (Requirement 10)"""
        
        self.db.insert('audit_logs', {
            'timestamp': datetime.now(),
            'action': action,
            'user_id': user_id,
            'resource': resource,
            'action_type': action_type,
            'ip_address': None  # Set from request context
        })

# Usage
access_control = AccessControl(db)

# Authenticate user
authenticated = access_control.authenticate_user(
    username='[email protected]',
    password='SecurePass123!',
    mfa_code='123456'
)

# Check access
has_access = access_control.check_access(
    user_id='user_123',
    resource='cardholder_data',
    action='read'
)

# Validate password
is_strong = access_control.enforce_password_policy('SecurePass123!')

class AuditLogger:
    """Implement comprehensive audit logging (Requirement 10)"""
    
    def __init__(self, db_connection):
        self.db = db_connection
    
    def log_cardholder_data_access(self, user_id: str, action: str,
                                  data_type: str, record_count: int):
        """Log access to cardholder data"""
        
        self.db.insert('audit_logs', {
            'timestamp': datetime.now(),
            'user_id': user_id,
            'action': action,
            'data_type': data_type,
            'record_count': record_count,
            'ip_address': None,
            'status': 'success'
        })
    
    def log_failed_access_attempt(self, user_id: str, reason: str):
        """Log failed access attempts"""
        
        self.db.insert('audit_logs', {
            'timestamp': datetime.now(),
            'user_id': user_id,
            'action': 'failed_access',
            'reason': reason,
            'status': 'failed'
        })
    
    def log_configuration_change(self, user_id: str, component: str,
                                change_description: str):
        """Log system configuration changes"""
        
        self.db.insert('audit_logs', {
            'timestamp': datetime.now(),
            'user_id': user_id,
            'action': 'configuration_change',
            'component': component,
            'change_description': change_description
        })
    
    def generate_audit_report(self, start_date: str,
                             end_date: str) -> dict:
        """Generate audit report for compliance"""
        
        logs = self.db.query(
            """SELECT * FROM audit_logs 
               WHERE timestamp BETWEEN ? AND ?""",
            (start_date, end_date)
        )
        
        return {
            'total_events': len(logs),
            'failed_access_attempts': len([l for l in logs if l['status'] == 'failed']),
            'configuration_changes': len([l for l in logs if l['action'] == 'configuration_change']),
            'data_access_events': len([l for l in logs if l['action'] == 'access']),
            'logs': logs
        }

# Usage
logger = AuditLogger(db)

# Log access
logger.log_cardholder_data_access(
    user_id='user_123',
    action='read',
    data_type='cardholder_data',
    record_count=100
)

# Generate report
report = logger.generate_audit_report('2025-01-01', '2025-01-31')
print(f"Total events: {report['total_events']}")