PCI-DSS Testing for Payment Systems: Validate Cardholder Data Security

# tests/pci/test_scope_reduction.py

def test_raw_pan_never_reaches_our_systems():
    """Card numbers must go directly to payment processor, not our servers."""
    
    # Monitor network traffic during checkout
    with capture_network_traffic() as traffic:
        complete_checkout_with_card(
            number="4111111111111111",
            expiry="12/25",
            cvv="123"
        )
    
    # Our application logs and requests should never contain the PAN
    requests_to_our_servers = [
        req for req in traffic.requests
        if "my-app.com" in req.url and "stripe.com" not in req.url
    ]
    
    for request in requests_to_our_servers:
        assert "4111111111111111" not in str(request.body), \
            f"PAN detected in request to our servers: {request.url}"
        
        # Check response too
        assert "4111111111111111" not in str(request.response_body), \
            "PAN detected in response from our servers"

def test_we_store_only_tokens_not_pans():
    """Database should contain payment tokens, never raw card numbers."""
    complete_checkout_with_card(number="4111111111111111")
    
    # Check database for raw PANs
    all_payment_records = db.execute(
        "SELECT * FROM payment_methods"
    ).fetchall()
    
    for record in all_payment_records:
        record_str = str(record)
        
        # Should never find a 16-digit card number
        import re
        pan_pattern = r'\b4[0-9]{15}\b'  # Visa-style 16 digit number
        assert not re.search(pan_pattern, record_str), \
            f"Possible raw PAN found in payment record: {record['id']}"
    
    # Tokens from Stripe look like tok_xxx or pm_xxx
    latest_payment = db.execute(
        "SELECT * FROM payment_methods ORDER BY created_at DESC LIMIT 1"
    ).fetchone()
    assert latest_payment["token"].startswith(("tok_", "pm_", "src_")), \
        "Payment record doesn't look like a valid token"

def test_stored_pan_is_masked():
    """If PANs are stored at all, they must be masked."""
    payment_method = get_payment_method(id="pm_123")
    
    displayed_pan = payment_method["last_four"]
    
    # Should show only last 4 digits
    assert len(displayed_pan) == 4, \
        f"More than last 4 digits stored: {displayed_pan}"
    
    # Should not start with card number digits
    assert not displayed_pan.startswith("4111"), \
        "PAN stored with more than last 4 digits"

def test_cvv_never_stored():
    """CVV/CVC must never be stored after authorization."""
    # Complete a transaction (CVV provided)
    transaction = complete_purchase(cvv="123")
    
    # Search everywhere for the CVV
    db_records = db.execute(
        f"SELECT * FROM transactions WHERE id = ?",
        (transaction.id,)
    ).fetchone()
    
    assert "123" not in str(db_records), "CVV found in transaction record"
    
    audit_logs = get_audit_logs(transaction_id=transaction.id)
    for log in audit_logs:
        assert "123" not in log.message, "CVV found in audit log"

def test_cardholder_data_encrypted_with_strong_cryptography():
    """Any stored cardholder data must use strong cryptography."""
    encryption_config = get_encryption_config()
    
    # PCI requires strong cryptography (AES-256, RSA-2048+, or equivalent)
    assert encryption_config.algorithm in ["AES-256", "AES-256-GCM", "RSA-4096"], \
        f"Weak encryption algorithm: {encryption_config.algorithm}"
    
    assert encryption_config.key_length >= 256, \
        f"Insufficient key length: {encryption_config.key_length} bits"

def test_cardholder_data_environment_network_segmentation():
    """CDE systems must be isolated from other network segments."""
    
    # From a non-CDE host, attempt to connect to CDE systems
    non_cde_host = get_test_host(environment="non_cde")
    cde_systems = get_cde_systems()
    
    for cde_system in cde_systems:
        connection_result = attempt_connection(
            from_host=non_cde_host,
            to_host=cde_system,
            port=5432  # PostgreSQL
        )
        
        assert not connection_result.success, \
            f"Non-CDE host can reach CDE system {cde_system}: network segmentation failing"

def test_firewall_rules_deny_unnecessary_traffic():
    """Only explicitly allowed traffic should reach the CDE."""
    allowed_ports = {80, 443, 8443}  # Only HTTPS
    
    cde_host = get_any_cde_host()
    
    for port in range(1, 10000):
        if port in allowed_ports:
            continue
        
        result = port_scan(host=cde_host, port=port)
        
        if result.open:
            # Some ports may be legitimately open (SSH for management)
            assert port in ALLOWED_MANAGEMENT_PORTS, \
                f"Unexpected open port {port} on CDE host {cde_host}"

def test_no_direct_internet_access_from_cde():
    """CDE systems must not have direct internet access."""
    cde_host = get_any_cde_host()
    
    # Attempt to reach an external host from within the CDE
    result = execute_on_host(
        cde_host,
        command="curl --max-time 5 https://checkip.amazonaws.com"
    )
    
    assert result.return_code != 0, \
        "CDE system has direct internet access (should route through proxy)"

def test_no_known_vulnerabilities_in_cde_systems():
    """CDE systems must not have high/critical CVEs."""
    cde_hosts = get_all_cde_hosts()
    
    violations = []
    for host in cde_hosts:
        scan_results = run_vulnerability_scan(host)
        
        critical_vulns = [
            v for v in scan_results.vulnerabilities
            if v.cvss_score >= 7.0  # High/Critical threshold
        ]
        
        for vuln in critical_vulns:
            violations.append(
                f"{host}: {vuln.cve_id} (CVSS {vuln.cvss_score}) - {vuln.title}"
            )
    
    assert len(violations) == 0, \
        f"High/Critical vulnerabilities in CDE:\n" + "\n".join(violations)

def test_web_application_firewall_blocks_common_attacks():
    """WAF must block OWASP Top 10 attack patterns."""
    attack_payloads = {
        "sql_injection": "' OR '1'='1",
        "xss": "<script>alert('xss')</script>",
        "command_injection": "; ls -la",
        "path_traversal": "../../etc/passwd",
    }
    
    for attack_name, payload in attack_payloads.items():
        response = httpx.get(
            "https://payment.my-app.com/checkout",
            params={"search": payload}
        )
        
        assert response.status_code in [400, 403], \
            f"WAF did not block {attack_name} attack (status: {response.status_code})"

def test_payment_data_access_requires_business_need():
    """Access to cardholder data must be need-to-know only."""
    
    # Customer service rep should not see full transaction history
    cs_token = get_token(role="customer_service")
    response = api_client.get(
        "/api/admin/all-transactions",
        headers={"Authorization": f"Bearer {cs_token}"}
    )
    assert response.status_code == 403, \
        "Customer service can access all transaction data"

def test_unique_ids_for_all_cde_users():
    """Every user accessing CDE must have a unique ID (no shared accounts)."""
    cde_users = get_users_with_cde_access()
    
    emails = [u.email for u in cde_users]
    assert len(emails) == len(set(emails)), "Duplicate user accounts detected"
    
    # Check for shared/generic accounts
    generic_names = ["admin", "shared", "service", "test", "backup"]
    for user in cde_users:
        assert not any(name in user.username.lower() for name in generic_names), \
            f"Generic/shared account with CDE access: {user.username}"

def test_passwords_meet_complexity_requirements():
    """Passwords for CDE access must meet PCI-DSS complexity requirements."""
    # PCI requires: min 12 chars, upper+lower+number+special, changes every 90 days
    
    weak_passwords = [
        "password123",     # No uppercase or special
        "P@ss1",           # Too short
        "Password123",     # No special character
    ]
    
    for weak_password in weak_passwords:
        response = api_client.post("/api/auth/change-password", json={
            "new_password": weak_password
        }, headers=auth_headers)
        
        assert response.status_code == 400, \
            f"Weak password accepted: {weak_password}"

def run_daily_pci_checks():
    """Daily PCI compliance validation."""
    checks = [
        ("No raw PANs in database", test_no_raw_pan_in_database),
        ("CVV not stored", test_cvv_not_stored_post_auth),
        ("Network segmentation intact", test_network_segmentation_quick_check),
        ("Vulnerability scan current", test_scan_results_are_recent),
        ("Access logs active", test_audit_logging_active),
        ("Encryption keys rotated", test_key_rotation_within_policy),
    ]
    
    for check_name, check_fn in checks:
        result = run_check_safely(check_fn)
        log_pci_compliance_result(check_name, result)
        
        if not result.passed:
            notify_security_team(f"PCI check failed: {check_name}\n{result.details}")

schedule.every().day.at("02:00").do(run_daily_pci_checks)