Building a Compliance Test Suite for Fintech Applications

class TestPasswordPolicy:
    """PCI-DSS Req 8.3, SOC 2 CC6.1"""
    
    def test_minimum_length_enforced(self, api_client):
        response = api_client.post('/auth/register', {
            'email': 'test@example.com',
            'password': 'Short1!'  # 7 characters, below 12-char minimum
        })
        assert response.status_code == 422
        assert 'password' in response.json()['errors']
    
    def test_complexity_required(self, api_client):
        weak_passwords = [
            'alllowercase123!',  # No uppercase
            'ALLUPPERCASE123!',  # No lowercase
            'NoNumbers!!!!!',    # No digits
            'NoSpecial123abc',   # No special chars
        ]
        for password in weak_passwords:
            response = api_client.post('/auth/register', {
                'email': 'test@example.com',
                'password': password
            })
            assert response.status_code == 422, f"Weak password accepted: {password}"
    
    def test_password_history_enforced(self, authenticated_client, test_user):
        """Cannot reuse last 4 passwords — PCI-DSS Req 8.3.6"""
        # Cycle through passwords
        passwords = ['NewPass1!', 'NewPass2!', 'NewPass3!', 'NewPass4!']
        for pwd in passwords:
            authenticated_client.post('/auth/change-password', {
                'current_password': test_user.current_password,
                'new_password': pwd
            })
            test_user.current_password = pwd
        
        # Try to reuse the first password
        response = authenticated_client.post('/auth/change-password', {
            'current_password': test_user.current_password,
            'new_password': passwords[0]
        })
        assert response.status_code == 422
        assert 'previously used' in response.json()['message'].lower()

class TestMFA:
    """PCI-DSS Req 8.4, SOC 2 CC6.2"""
    
    def test_admin_requires_mfa(self, api_client):
        """Admin access must require MFA — no exceptions"""
        # Valid password, no MFA token
        response = api_client.post('/auth/login', {
            'email': 'admin@example.com',
            'password': VALID_ADMIN_PASSWORD,
        })
        # Should not return full session token
        assert response.status_code == 200
        data = response.json()
        assert 'mfa_required' in data
        assert 'session_token' not in data  # No full auth without MFA
    
    def test_mfa_lockout_after_failures(self, api_client, mfa_user):
        """Account locks after repeated MFA failures"""
        for _ in range(10):
            api_client.post('/auth/mfa/verify', {
                'user_id': mfa_user.id,
                'token': '000000'  # Wrong token
            })
        
        # 11th attempt should be locked
        response = api_client.post('/auth/mfa/verify', {
            'user_id': mfa_user.id,
            'token': mfa_user.valid_totp_token()
        })
        assert response.status_code == 423  # Locked
    
    def test_session_timeout(self, authenticated_client):
        """Idle sessions expire — PCI-DSS Req 8.2.8 (15 min max)"""
        # Simulate session idle for 16 minutes
        token = authenticated_client.auth_token
        advance_time(minutes=16)
        
        response = authenticated_client.get('/api/account')
        assert response.status_code == 401

class TestRoleBasedAccess:
    """PCI-DSS Req 7, SOC 2 CC6.3"""
    
    @pytest.mark.parametrize('role,endpoint,expected_status', [
        ('viewer', '/admin/users', 403),
        ('viewer', '/admin/config', 403),
        ('viewer', '/admin/transactions', 403),
        ('support', '/admin/users', 200),
        ('support', '/admin/config', 403),
        ('admin', '/admin/config', 200),
    ])
    def test_endpoint_access_by_role(self, api_client, role, endpoint, expected_status):
        client = api_client.as_role(role)
        response = client.get(endpoint)
        assert response.status_code == expected_status

class TestTLSConfiguration:
    """PCI-DSS Req 4.2.1"""
    
    def test_tls_version_minimum(self):
        """TLS 1.2 minimum — TLS 1.0 and 1.1 must be rejected"""
        import ssl
        
        for tls_version in [ssl.TLSVersion.TLSv1, ssl.TLSVersion.TLSv1_1]:
            context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
            context.maximum_version = tls_version
            context.check_hostname = False
            context.verify_mode = ssl.CERT_NONE
            
            with pytest.raises((ssl.SSLError, ConnectionResetError)):
                with socket.create_connection((TARGET_HOST, 443)) as sock:
                    with context.wrap_socket(sock, server_hostname=TARGET_HOST):
                        pass
    
    def test_weak_cipher_suites_rejected(self):
        """Export-grade and NULL cipher suites must be disabled"""
        weak_ciphers = ['NULL-SHA', 'EXP-RC4-MD5', 'DES-CBC-SHA']
        result = subprocess.run([
            'nmap', '--script', 'ssl-enum-ciphers',
            '-p', '443', TARGET_HOST
        ], capture_output=True, text=True)
        
        for cipher in weak_ciphers:
            assert cipher not in result.stdout, f"Weak cipher enabled: {cipher}"
    
    def test_hsts_header_present(self, api_client):
        """HSTS enforces HTTPS — defense in depth"""
        response = api_client.get('/')
        assert 'Strict-Transport-Security' in response.headers
        hsts = response.headers['Strict-Transport-Security']
        assert 'max-age=31536000' in hsts  # 1 year minimum

class TestEncryptionAtRest:
    """PCI-DSS Req 3.5"""
    
    def test_pan_never_stored_in_plaintext(self, db_connection):
        """Primary Account Number must be masked or encrypted in storage"""
        # Pull raw DB record for a transaction
        result = db_connection.execute(
            "SELECT card_number FROM transactions LIMIT 100"
        ).fetchall()
        
        pan_pattern = re.compile(r'\b[3-6]\d{12,18}\b')
        for row in result:
            assert not pan_pattern.match(str(row['card_number'])), \
                "Plaintext PAN found in transactions table"
    
    def test_encryption_key_not_in_application_config(self):
        """Encryption keys must not be stored with encrypted data"""
        config_files = glob.glob('**/*.yaml', recursive=True) + \
                      glob.glob('**/*.json', recursive=True) + \
                      glob.glob('**/*.env', recursive=True)
        
        key_pattern = re.compile(r'(encryption_key|secret_key|aes_key)\s*[:=]\s*\S+')
        for filepath in config_files:
            with open(filepath) as f:
                content = f.read()
            assert not key_pattern.search(content), \
                f"Potential hardcoded key in {filepath}"

class TestAuditLogging:
    """PCI-DSS Req 10.2, SOC 2 CC7.2"""
    
    REQUIRED_EVENTS = [
        ('user_login', {'user_id', 'ip_address', 'timestamp', 'success'}),
        ('user_logout', {'user_id', 'session_id', 'timestamp'}),
        ('failed_login', {'attempted_email', 'ip_address', 'timestamp'}),
        ('password_change', {'user_id', 'changed_by', 'timestamp'}),
        ('privilege_escalation', {'user_id', 'from_role', 'to_role', 'authorized_by', 'timestamp'}),
        ('data_access', {'user_id', 'resource_type', 'resource_id', 'action', 'timestamp'}),
        ('config_change', {'user_id', 'setting', 'old_value', 'new_value', 'timestamp'}),
    ]
    
    @pytest.mark.parametrize('event_type,required_fields', REQUIRED_EVENTS)
    def test_event_captured_with_required_fields(self, event_type, required_fields, api_client):
        trigger_event(event_type, api_client)
        
        events = get_audit_events(event_type=event_type, limit=1)
        assert len(events) == 1
        
        event = events[0]
        missing = required_fields - set(event.keys())
        assert not missing, f"Audit event {event_type} missing fields: {missing}"
    
    def test_audit_log_is_append_only(self, db_connection, test_user):
        """Audit logs cannot be modified or deleted — PCI-DSS Req 10.3.2"""
        # Create an audit event
        login_event = trigger_login(test_user)
        event_id = login_event['id']
        
        # Attempt to update via application API — should fail
        response = self.api_client.patch(f'/internal/audit/{event_id}', {
            'user_id': 'manipulated'
        })
        assert response.status_code in [403, 404, 405]
        
        # Attempt direct DB update — should fail due to DB-level constraints
        with pytest.raises(Exception):  # Specific to your DB; psycopg2.errors.InsufficientPrivilege etc.
            db_connection.execute(
                f"UPDATE audit_log SET user_id = 'manipulated' WHERE id = '{event_id}'"
            )
    
    def test_audit_log_timestamp_not_manipulable(self, api_client, test_user):
        """Timestamps must come from a trusted source, not client"""
        before = datetime.utcnow()
        
        # Login with a manipulated client timestamp in the request
        response = api_client.post('/auth/login', 
            data={'email': test_user.email, 'password': test_user.password},
            headers={'X-Client-Time': '2020-01-01T00:00:00Z'}  # Past date
        )
        
        event = get_audit_events(event_type='user_login', limit=1)[0]
        event_time = datetime.fromisoformat(event['timestamp'])
        
        # Server-side timestamp should be near current time, not the spoofed client time
        assert event_time > before
        assert event_time < datetime.utcnow() + timedelta(seconds=5)

class TestDataRetention:
    """PCI-DSS Req 3.2, GDPR Art. 17"""
    
    def test_sensitive_auth_data_not_retained_post_authorization(self, db_connection):
        """CVV2, full track data, PIN must not be stored after auth — PCI-DSS Req 3.2.1"""
        columns_to_check = [
            ('transactions', 'cvv'),
            ('transactions', 'track_data'),
            ('transactions', 'pin_block'),
            ('payment_attempts', 'cvv'),
        ]
        for table, column in columns_to_check:
            # Check if column even exists — it shouldn't
            result = db_connection.execute(f"""
                SELECT column_name 
                FROM information_schema.columns 
                WHERE table_name = '{table}' AND column_name = '{column}'
            """).fetchone()
            assert result is None, f"Sensitive data column {table}.{column} exists in schema"
    
    def test_account_deletion_removes_pii(self, api_client, authenticated_client, test_user):
        """GDPR right to erasure — deleting account removes personal data"""
        user_id = test_user.id
        
        authenticated_client.delete(f'/account')
        
        # Verify PII removed from user record
        user_record = get_user_record(user_id)
        assert user_record['email'] == f'deleted-{user_id}@anonymized.internal'
        assert user_record['name'] is None
        assert user_record['phone'] is None
        
        # Financial records retained for regulatory compliance but anonymized
        transactions = get_transactions(user_id=user_id)
        for tx in transactions:
            assert tx['user_email'] != test_user.email  # Anonymized
    
    def test_data_retention_policy_enforced_automatically(self, db_connection):
        """Records past retention period purged by scheduled job"""
        # Insert record with date beyond retention window
        old_record = insert_test_record(
            created_at=datetime.utcnow() - timedelta(days=2557)  # 7 years + 1 day
        )
        
        # Trigger retention enforcement
        run_retention_job()
        
        # Verify old record is gone
        result = db_connection.execute(
            "SELECT id FROM audit_log WHERE id = %s", (old_record['id'],)
        ).fetchone()
        assert result is None

tests/
  compliance/
    pci_dss/
      req_6_secure_dev/
        test_dependency_vulnerabilities.py
        test_code_review_gates.py
      req_7_access_control/
        test_rbac.py
        test_least_privilege.py
      req_8_authentication/
        test_password_policy.py
        test_mfa.py
        test_session_management.py
      req_10_logging/
        test_audit_events.py
        test_log_integrity.py
      req_11_testing/
        test_vulnerability_scanning.py
    soc2/
      cc6_logical_access/
        test_authentication.py
        test_authorization.py
      cc7_monitoring/
        test_alerting.py
        test_incident_detection.py
      cc8_change_management/
        test_deployment_controls.py
    shared/
      test_encryption.py
      test_data_retention.py

@pytest.mark.compliance
@pytest.mark.pci_dss_req_8
class TestMFA:
    ...

# Run all compliance tests
pytest tests/compliance/ -v

# Run only PCI-DSS tests
pytest tests/compliance/pci_dss/ -v

# Run specific requirement
pytest -m pci_dss_req_8 -v

# CI configuration excerpt
compliance-gates:
  stage: compliance
  script:
    - pytest tests/compliance/ -v 
        --junitxml=compliance-results.xml
        -m "not slow"
  artifacts:
    reports:
      junit: compliance-results.xml
    paths:
      - compliance-results.xml
    expire_in: 13 months  # Retain beyond annual audit window
  rules:
    - if: $CI_PIPELINE_SOURCE == "push"
      when: always

compliance-full:
  stage: compliance
  script:
    - pytest tests/compliance/ -v --junitxml=compliance-full.xml
  artifacts:
    paths:
      - compliance-full.xml
    expire_in: 13 months
  rules:
    - if: $CI_PIPELINE_SOURCE == "schedule"
      when: always

# scripts/compliance_report.py
import xml.etree.ElementTree as ET

CONTROL_MAPPING = {
    'TestPasswordPolicy': ['PCI-DSS 8.3', 'SOC2 CC6.1'],
    'TestMFA': ['PCI-DSS 8.4', 'SOC2 CC6.2'],
    'TestAuditLogging': ['PCI-DSS 10.2', 'SOC2 CC7.2'],
    'TestEncryptionAtRest': ['PCI-DSS 3.5', 'SOC2 CC6.1'],
    'TestDataRetention': ['PCI-DSS 3.2', 'GDPR Art. 17'],
}

def generate_compliance_report(junit_xml_path):
    tree = ET.parse(junit_xml_path)
    root = tree.getroot()
    
    report = {}
    for testcase in root.iter('testcase'):
        classname = testcase.get('classname', '').split('.')[-1]
        controls = CONTROL_MAPPING.get(classname, ['Unmapped'])
        
        status = 'PASS' if testcase.find('failure') is None else 'FAIL'
        
        for control in controls:
            if control not in report:
                report[control] = {'pass': 0, 'fail': 0, 'tests': []}
            report[control][status.lower()] += 1
            report[control]['tests'].append({
                'name': testcase.get('name'),
                'status': status
            })
    
    return report