PCI DSS Testing for Payment Processing Applications

# tests/pci/test_cardholder_data_storage.py
import pytest
import re
import requests

BASE_URL = "https://api.example.com"

# Patterns that indicate raw card data
PAN_PATTERN = re.compile(r"\b(?:4[0-9]{12}(?:[0-9]{3})?|5[1-5][0-9]{14}|3[47][0-9]{13})\b")
CVV_PATTERN = re.compile(r"\b(?:cvv|cvc|cvv2|cvc2|security.?code)[:\s]*\d{3,4}\b", re.IGNORECASE)
TRACK_PATTERN = re.compile(r"%B\d{13,19}\^")  # Track 1 format

class TestCardholderDataStorage:

    def test_pan_is_masked_in_api_responses(self):
        """
        PCI DSS 3.3.1: PAN must be masked when displayed — only first 6 and last 4 digits.
        """
        admin_token = self._get_admin_token()
        resp = requests.get(
            f"{BASE_URL}/api/payment-methods",
            headers={"Authorization": f"Bearer {admin_token}"}
        )
        assert resp.status_code == 200
        methods = resp.json()["payment_methods"]

        for method in methods:
            if "pan" in method:
                pan = method["pan"]
                # Must be masked: first 6 + asterisks + last 4
                assert re.match(r"^\d{6}\*+\d{4}$", pan), \
                    f"PAN not properly masked: {pan}"
            if "card_number" in method:
                number = method["card_number"]
                assert re.match(r"^\d{4,6}\*+\d{4}$", number), \
                    f"Card number not properly masked: {number}"

    def test_cvv_is_never_stored(self):
        """
        PCI DSS 3.2.1: SAD including CVV must not be stored after authorization.
        Test by submitting a payment and then verifying the CVV is not retrievable.
        """
        user_token = self._get_user_token()

        # Make a payment (this will tokenize the card)
        payment_resp = requests.post(
            f"{BASE_URL}/api/payments",
            json={
                "amount": 100,
                "currency": "usd",
                "card": {
                    "number": "4111111111111111",
                    "exp_month": 12,
                    "exp_year": 2026,
                    "cvv": "123"
                }
            },
            headers={"Authorization": f"Bearer {user_token}"}
        )
        assert payment_resp.status_code in (200, 201)
        payment_id = payment_resp.json()["id"]

        # Retrieve the payment — CVV must not appear
        get_resp = requests.get(
            f"{BASE_URL}/api/payments/{payment_id}",
            headers={"Authorization": f"Bearer {user_token}"}
        )
        payment_data = str(get_resp.json())
        assert "123" not in payment_data or "cvv" not in payment_data.lower(), \
            "CVV appears to be stored and returned after authorization"

    def test_raw_pan_not_in_logs(self):
        """
        After processing a payment, scan logs to verify raw PAN is not present.
        """
        import subprocess

        # Process a test payment with a known card number
        user_token = self._get_user_token()
        requests.post(
            f"{BASE_URL}/api/payments",
            json={
                "amount": 100,
                "currency": "usd",
                "card": {
                    "number": "4111111111111111",
                    "exp_month": 12,
                    "exp_year": 2026,
                    "cvv": "999"
                }
            },
            headers={"Authorization": f"Bearer {user_token}"}
        )

        # Check application logs for raw PAN
        result = subprocess.run(
            ["kubectl", "logs", "--tail=500", "deployment/api-server"],
            capture_output=True, text=True
        )

        pan_matches = PAN_PATTERN.findall(result.stdout)
        assert len(pan_matches) == 0, \
            f"Raw PAN found in logs: {pan_matches}"

    def test_tokenization_replaces_pan(self):
        """
        Verify that after tokenization, the token is what is stored — not the PAN.
        """
        resp = requests.post(
            f"{BASE_URL}/api/payment-methods",
            json={
                "card": {
                    "number": "4111111111111111",
                    "exp_month": 12,
                    "exp_year": 2026,
                    "cvv": "123"
                }
            },
            headers={"Authorization": f"Bearer {self._get_user_token()}"}
        )
        assert resp.status_code == 201
        method = resp.json()

        # Stored reference must be a token, not a PAN
        assert "token" in method or "payment_method_id" in method
        # The token must not be a valid PAN
        stored_ref = method.get("token") or method.get("payment_method_id") or ""
        assert not PAN_PATTERN.match(stored_ref), \
            "Stored payment method reference appears to be a raw PAN"

# tests/pci/test_transmission_security.py
import ssl
import socket
import requests

class TestTransmissionSecurity:

    def test_payment_endpoint_rejects_http(self):
        """
        PCI DSS 4.2.1: Strong cryptography required for CHD transmission.
        HTTP must be rejected or redirected to HTTPS.
        """
        try:
            resp = requests.get(
                "http://api.example.com/api/payments",
                allow_redirects=False,
                timeout=5
            )
            # If we get here, either it redirects or rejects
            assert resp.status_code in (301, 302, 400, 403), \
                f"HTTP payment endpoint returned {resp.status_code} — not redirected or blocked"
        except requests.exceptions.ConnectionError:
            pass  # Connection refused — acceptable

    def test_tls_1_3_supported(self):
        """TLS 1.3 should be supported for PCI 4.0 compliance."""
        hostname = "api.example.com"
        port = 443

        context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        context.minimum_version = ssl.TLSVersion.TLSv1_3
        context.load_verify_locations(cafile="/etc/ssl/cert.pem")

        try:
            with socket.create_connection((hostname, port), timeout=10) as sock:
                with context.wrap_socket(sock, server_hostname=hostname) as ssock:
                    assert ssock.version() == "TLSv1.3"
        except ssl.SSLError as e:
            pytest.fail(f"TLS 1.3 not supported: {e}")

    def test_certificate_is_valid_and_not_expiring_soon(self):
        """Expired or expiring certificates cause outages and compliance gaps."""
        import ssl
        from datetime import datetime, timedelta

        cert_info = ssl.get_server_certificate(("api.example.com", 443))
        x509 = ssl.PEM_cert_to_DER_cert(cert_info)

        # Use cryptography library to parse expiry
        from cryptography import x509
        from cryptography.hazmat.backends import default_backend

        cert = x509.load_der_x509_certificate(x509, default_backend())
        expiry = cert.not_valid_after_utc

        days_until_expiry = (expiry - datetime.utcnow().replace(tzinfo=expiry.tzinfo)).days
        assert days_until_expiry > 30, \
            f"TLS certificate expires in {days_until_expiry} days — rotate before PCI audit"

# tests/pci/test_secure_development.py
import requests

class TestSecureDevelopment:

    def test_sql_injection_rejected_in_payment_search(self):
        """
        PCI DSS 6.3.3: Protect against known vulnerabilities including injection.
        """
        user_token = self._get_user_token()
        sql_payloads = [
            "' OR '1'='1",
            "1; DROP TABLE payments--",
            "' UNION SELECT card_number FROM payments--",
        ]
        for payload in sql_payloads:
            resp = requests.get(
                f"{BASE_URL}/api/payments/search",
                params={"q": payload},
                headers={"Authorization": f"Bearer {user_token}"}
            )
            # Must not return 500 (which could indicate unhandled SQL error)
            assert resp.status_code != 500, \
                f"SQL injection payload '{payload}' caused 500 error — possible injection point"
            # Must not return results that look like card data
            if resp.status_code == 200:
                body = resp.text
                assert not any(
                    len(word) == 16 and word.isdigit()
                    for word in body.split()
                ), f"Response to injection payload contains possible PAN data"

    def test_payment_form_has_csrf_protection(self):
        """
        PCI DSS 6.4.3: All payment pages must be protected against CSRF.
        """
        resp = requests.get(f"{BASE_URL}/api/checkout/init")
        assert resp.status_code == 200
        data = resp.json()

        assert "csrf_token" in data, "Payment form initialization does not provide CSRF token"

        # Attempt to submit payment without CSRF token — must be rejected
        no_csrf_resp = requests.post(
            f"{BASE_URL}/api/payments",
            json={"amount": 100},
            headers={"Authorization": f"Bearer {self._get_user_token()}"}
            # Intentionally omitting X-CSRF-Token header
        )
        assert no_csrf_resp.status_code in (400, 403), \
            "Payment submission accepted without CSRF token"

    def test_error_responses_do_not_expose_system_info(self):
        """
        PCI DSS 6.2.4: Applications must not expose technical information in errors.
        """
        resp = requests.get(
            f"{BASE_URL}/api/payments/nonexistent-id",
            headers={"Authorization": f"Bearer {self._get_user_token()}"}
        )
        error_body = resp.text.lower()

        forbidden_info = ["stack trace", "at line", "exception", "postgresql", "mysql",
                          "django", "rails", "express", "internal server"]
        for term in forbidden_info:
            assert term not in error_body, \
                f"Error response exposes technical info: '{term}'"

# tests/pci/test_access_control.py

class TestPCIAccessControl:

    def test_payment_data_access_requires_business_justification(self):
        """
        PCI DSS 7.2.1: Access to system components and CHD restricted to
        only those individuals whose job requires such access.
        """
        # Customer service role should see masked PAN only
        cs_token = self._login_as("customer_service@example.com")
        resp = requests.get(
            f"{BASE_URL}/api/admin/payments",
            headers={"Authorization": f"Bearer {cs_token}"}
        )

        if resp.status_code == 200:
            payments = resp.json().get("payments", [])
            for payment in payments:
                if "pan" in payment:
                    pan = payment["pan"]
                    # CS should see masked PAN only
                    assert "*" in pan, \
                        f"Customer service role sees unmasked PAN: {pan}"

    def test_payment_admin_requires_mfa(self):
        """
        PCI DSS 8.4.2: MFA required for all access into the CDE.
        """
        # Attempt to reach payment admin without MFA — should be challenged
        basic_auth_resp = requests.post(f"{BASE_URL}/auth/login", json={
            "email": "payment_admin@example.com",
            "password": "correctpassword"
        })
        # If the server returns a full token without MFA challenge, that's a finding
        body = basic_auth_resp.json()
        if basic_auth_resp.status_code == 200:
            assert body.get("mfa_required") is True or "mfa_token" in body, \
                "Payment admin login did not trigger MFA challenge — PCI 8.4.2 violation"

    def test_account_lockout_after_failed_attempts(self):
        """
        PCI DSS 8.3.4: Account lockout after not more than 10 failed attempts.
        """
        for i in range(11):
            resp = requests.post(f"{BASE_URL}/auth/login", json={
                "email": "lockout-test@example.com",
                "password": "wrongpassword"
            })

        # 11th attempt should indicate lockout
        assert resp.status_code in (400, 401, 423), \
            "Account not locked after 11 failed login attempts"

        body = resp.json()
        lockout_indicators = ["locked", "too many attempts", "temporarily blocked"]
        assert any(ind in str(body).lower() for ind in lockout_indicators), \
            "Account lockout response does not communicate lockout state"

# .github/workflows/pci-compliance.yml
name: PCI DSS Compliance Tests

on:
  pull_request:
    paths:
      - "src/payments/**"
      - "src/checkout/**"
      - "src/auth/**"
      - "migrations/**"

jobs:
  pci-tests:
    runs-on: ubuntu-latest
    environment: staging
    steps:
      - uses: actions/checkout@v4
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - name: Install dependencies
        run: pip install pytest requests cryptography
      - name: Run PCI compliance tests
        env:
          API_BASE_URL: ${{ vars.STAGING_API_URL }}
          TEST_USER_TOKEN: ${{ secrets.PCI_TEST_USER_TOKEN }}
          TEST_ADMIN_TOKEN: ${{ secrets.PCI_TEST_ADMIN_TOKEN }}
        run: pytest tests/pci/ -v --tb=short --junitxml=pci-results.xml
      - name: Upload PCI test report
        uses: actions/upload-artifact@v4
        if: always()
        with:
          name: pci-compliance-report
          path: pci-results.xml
          retention-days: 400  # Retain beyond 12-month audit window