SecurePass ZT — Zero-Trust Identity & Access Management Platform

import pytest
import requests
from itertools import combinations

class RBACValidator:
    def __init__(self, base_url: str, admin_token: str):
        self.base_url = base_url
        self.admin_token = admin_token
        self.roles = ['admin', 'manager', 'developer', 'analyst', 'viewer']
        self.resources = ['projects', 'users', 'settings', 'audit_logs', 'billing']
        self.actions = ['read', 'write', 'delete', 'approve']
    
    def get_role_permissions(self, role: str) -> dict:
        """Fetch permissions for a specific role."""
        response = requests.get(
            f"{self.base_url}/api/v1/roles/{role}/permissions",
            headers={'Authorization': f'Bearer {self.admin_token}'}
        )
        response.raise_for_status()
        return response.json()['permissions']
    
    def test_permission_boundary(self, role: str, resource: str, action: str, should_succeed: bool) -> bool:
        """Test if user with role can perform action on resource."""
        # Create test user with specific role
        user_token = self._create_test_user(role)
        
        # Attempt action on resource
        response = requests.post(
            f"{self.base_url}/api/v1/{resource}/{action}",
            headers={'Authorization': f'Bearer {user_token}'},
            json={'resource_id': 'TEST-123'}
        )
        
        # Verify response matches expectation
        if should_succeed:
            assert response.status_code == 200, f"Expected success for {role} → {resource}:{action}"
        else:
            assert response.status_code == 403, f"Expected 403 for {role} → {resource}:{action}, got {response.status_code}"
        
        self._delete_test_user(user_token)
        return True
    
    def validate_privilege_escalation_attempts(self, initial_role: str) -> dict:
        """Test if user can escalate privileges through API calls."""
        user_token = self._create_test_user(initial_role)
        escalation_attempts = []
        
        for target_role in self.roles:
            if target_role == initial_role:
                continue
            
            # Attempt to escalate role via PATCH endpoint
            response = requests.patch(
                f"{self.base_url}/api/v1/users/me/role",
                headers={'Authorization': f'Bearer {user_token}'},
                json={'new_role': target_role}
            )
            
            escalation_attempts.append({
                'initial_role': initial_role,
                'target_role': target_role,
                'status_code': response.status_code,
                'prevented': response.status_code == 403
            })
        
        self._delete_test_user(user_token)
        return escalation_attempts

@pytest.mark.parametrize('role,resource,action,should_succeed', [
    ('admin', 'projects', 'write', True),
    ('admin', 'billing', 'delete', True),
    ('admin', 'audit_logs', 'read', True),
    ('manager', 'projects', 'write', True),
    ('manager', 'settings', 'write', False),
    ('manager', 'billing', 'read', False),
    ('developer', 'projects', 'write', True),
    ('developer', 'users', 'write', False),
    ('developer', 'billing', 'read', False),
    ('analyst', 'projects', 'read', True),
    ('analyst', 'projects', 'write', False),
    ('viewer', 'projects', 'read', True),
    ('viewer', 'projects', 'write', False),
])
def test_rbac_permissions(role, resource, action, should_succeed):
    """Test RBAC permission enforcement for all role-resource-action combinations."""
    validator = RBACValidator('https://iam-staging.example.com', os.getenv('ADMIN_TOKEN'))
    assert validator.test_permission_boundary(role, resource, action, should_succeed)

@pytest.mark.parametrize('role', ['viewer', 'analyst', 'developer', 'manager'])
def test_privilege_escalation_prevention(role):
    """Verify users cannot escalate privileges through any vector."""
    validator = RBACValidator('https://iam-staging.example.com', os.getenv('ADMIN_TOKEN'))
    attempts = validator.validate_privilege_escalation_attempts(role)
    
    # All escalation attempts should be prevented (403)
    prevented_count = sum(1 for a in attempts if a['prevented'])
    assert prevented_count == len(attempts), f"Found {len(attempts) - prevented_count} successful escalation attempts for role {role}"

import pytest
import requests
import jwt
import time
from urllib.parse import urlencode

class OAuth2Validator:
    def __init__(self, oauth_endpoint: str, client_id: str, client_secret: str):
        self.oauth_endpoint = oauth_endpoint
        self.client_id = client_id
        self.client_secret = client_secret
    
    def test_redirect_uri_validation(self) -> dict:
        """Verify redirect URI is strictly validated to prevent open redirect."""
        valid_redirect = 'https://app.example.com/callback'
        attack_redirect = 'https://attacker.com/steal-token'
        
        # Legitimate request
        response_valid = requests.get(
            f"{self.oauth_endpoint}/authorize",
            params={
                'client_id': self.client_id,
                'redirect_uri': valid_redirect,
                'response_type': 'code',
                'scope': 'openid profile email'
            }
        )
        assert response_valid.status_code == 200
        
        # Attack with unregistered redirect
        response_attack = requests.get(
            f"{self.oauth_endpoint}/authorize",
            params={
                'client_id': self.client_id,
                'redirect_uri': attack_redirect,
                'response_type': 'code',
                'scope': 'openid profile email'
            }
        )
        assert response_attack.status_code == 400, "Unregistered redirect should be rejected"
        return {'valid': True, 'attack_prevented': True}
    
    def test_code_reuse_prevention(self, auth_code: str) -> dict:
        """Verify authorization codes cannot be reused after single exchange."""
        # First exchange (should succeed)
        response1 = requests.post(
            f"{self.oauth_endpoint}/token",
            data={
                'grant_type': 'authorization_code',
                'code': auth_code,
                'client_id': self.client_id,
                'client_secret': self.client_secret,
                'redirect_uri': 'https://app.example.com/callback'
            }
        )
        assert response1.status_code == 200
        token1 = response1.json()['access_token']
        
        # Second exchange attempt with same code (should fail)
        response2 = requests.post(
            f"{self.oauth_endpoint}/token",
            data={
                'grant_type': 'authorization_code',
                'code': auth_code,
                'client_id': self.client_id,
                'client_secret': self.client_secret,
                'redirect_uri': 'https://app.example.com/callback'
            }
        )
        assert response2.status_code == 400, "Code reuse should be prevented"
        return {'first_exchange': True, 'reuse_prevented': True}
    
    def test_state_parameter_validation(self, state: str) -> dict:
        """Verify state parameter is required and validated to prevent CSRF."""
        # Missing state parameter
        response_no_state = requests.get(
            f"{self.oauth_endpoint}/authorize",
            params={
                'client_id': self.client_id,
                'redirect_uri': 'https://app.example.com/callback',
                'response_type': 'code'
            }
        )
        # Should reject or warn about missing state
        assert response_no_state.status_code in [400, 302]
        
        # With state parameter
        response_with_state = requests.get(
            f"{self.oauth_endpoint}/authorize",
            params={
                'client_id': self.client_id,
                'redirect_uri': 'https://app.example.com/callback',
                'response_type': 'code',
                'state': state
            }
        )
        assert response_with_state.status_code == 302
        assert state in response_with_state.headers.get('Location', '')
        return {'state_validation': True, 'csrf_prevented': True}
    
    def test_jwt_signature_validation(self, id_token: str) -> dict:
        """Verify ID token JWT signature is validated and none algorithm is rejected."""
        # Valid signature (should decode)
        try:
            decoded = jwt.decode(
                id_token,
                options={"verify_signature": False},  # Get payload to inspect
                algorithms=['RS256']
            )
            assert 'sub' in decoded and 'aud' in decoded
        except:
            pass
        
        # Tampered signature (should fail verification)
        parts = id_token.split('.')
        tampered = parts[0] + '.' + parts[1] + '.' + 'INVALID_SIGNATURE'
        
        with pytest.raises(jwt.InvalidSignatureError):
            jwt.decode(
                tampered,
                'RS256_PUBLIC_KEY',
                algorithms=['RS256']
            )
        
        # None algorithm attack (should be rejected)
        none_token = jwt.encode({'sub': 'attacker'}, '', algorithm='none')
        with pytest.raises(jwt.InvalidAlgorithmError):
            jwt.decode(
                none_token,
                'RS256_PUBLIC_KEY',
                algorithms=['RS256']  # Only RS256 allowed
            )
        
        return {'signature_validated': True, 'none_algorithm_blocked': True}

@pytest.mark.parametrize('client_id,client_secret', [
    ('client-prod', os.getenv('PROD_CLIENT_SECRET')),
])
def test_oauth_flows(client_id, client_secret):
    """Test complete OAuth 2.0 and OIDC flow with all security validations."""
    validator = OAuth2Validator(
        'https://oauth.example.com',
        client_id,
        client_secret
    )
    
    assert validator.test_redirect_uri_validation()['attack_prevented']
    assert validator.test_state_parameter_validation('random-state-value')['csrf_prevented']
    assert validator.test_jwt_signature_validation(os.getenv('SAMPLE_ID_TOKEN'))['none_algorithm_blocked']