API Security Testing: Complete Guide to OAuth, JWT, and API Keys

# Test OAuth authorization code flow
import requests
from urllib.parse import urlencode, parse_qs

class OAuthFlowTester:
    def __init__(self, client_id, client_secret, auth_url, token_url):
        self.client_id = client_id
        self.client_secret = client_secret
        self.auth_url = auth_url
        self.token_url = token_url

    def test_authorization_flow(self):
        """Test complete OAuth flow"""

        # Step 1: Get authorization code
        params = {
            'response_type': 'code',
            'client_id': self.client_id,
            'redirect_uri': 'http://localhost:3000/callback',
            'scope': 'read write',
            'state': 'random_state_string'
        }

        auth_request = f"{self.auth_url}?{urlencode(params)}"
        print(f"Authorization URL: {auth_request}")

        # Simulate user authorization (manual step in real testing)
        auth_code = input("Enter authorization code: ")

        # Step 2: Exchange code for token
        token_response = requests.post(self.token_url, data={
            'grant_type': 'authorization_code',
            'code': auth_code,
            'redirect_uri': 'http://localhost:3000/callback',
            'client_id': self.client_id,
            'client_secret': self.client_secret
        })

        assert token_response.status_code == 200, "Token exchange failed"

        tokens = token_response.json()
        assert 'access_token' in tokens
        assert 'refresh_token' in tokens

        print("✓ OAuth flow successful")
        return tokens

    def test_token_refresh(self, refresh_token):
        """Test refresh token flow"""
        response = requests.post(self.token_url, data={
            'grant_type': 'refresh_token',
            'refresh_token': refresh_token,
            'client_id': self.client_id,
            'client_secret': self.client_secret
        })

        assert response.status_code == 200
        new_tokens = response.json()
        assert 'access_token' in new_tokens
        print("✓ Token refresh successful")

    def test_invalid_client_credentials(self):
        """Test with invalid credentials"""
        response = requests.post(self.token_url, data={
            'grant_type': 'client_credentials',
            'client_id': 'invalid_client',
            'client_secret': 'invalid_secret'
        })

        assert response.status_code == 401, "Should reject invalid credentials"
        print("✓ Invalid credentials properly rejected")

def test_oauth_security():
    """Test OAuth security vulnerabilities"""

    # Test 1: State parameter (CSRF protection)
    def test_state_parameter():
        # Request without state
        response = requests.get(auth_url, params={
            'response_type': 'code',
            'client_id': client_id,
            'redirect_uri': redirect_uri
            # Missing 'state' parameter
        })
        # Should enforce state parameter
        assert 'error' in response.url or response.status_code == 400

    # Test 2: Redirect URI validation
    def test_redirect_uri_validation():
        # Attempt redirect URI manipulation
        malicious_redirect = 'http://evil.com/callback'

        response = requests.get(auth_url, params={
            'response_type': 'code',
            'client_id': client_id,
            'redirect_uri': malicious_redirect,
            'state': 'test'
        })

        # Should reject unregistered redirect URIs
        assert response.status_code != 302 or 'evil.com' not in response.headers.get('Location', '')

    # Test 3: Token expiration
    def test_token_expiration():
        # Use expired token
        expired_token = 'expired_token_here'

        response = requests.get(
            'https://api.example.com/protected',
            headers={'Authorization': f'Bearer {expired_token}'}
        )

        assert response.status_code == 401
        assert 'token expired' in response.json().get('error', '').lower()

    test_state_parameter()
    test_redirect_uri_validation()
    test_token_expiration()
    print("✓ OAuth security tests passed")

// JWT testing utilities
const jwt = require('jsonwebtoken');
const crypto = require('crypto');

class JWTTester {
    constructor(secret) {
        this.secret = secret;
    }

    // Test 1: Valid JWT
    testValidJWT() {
        const payload = {
            userId: 123,
            username: 'testuser',
            role: 'user'
        };

        const token = jwt.sign(payload, this.secret, { expiresIn: '1h' });

        try {
            const decoded = jwt.verify(token, this.secret);
            console.log('✓ Valid JWT verified successfully');
            return decoded;
        } catch (err) {
            console.error('✗ JWT verification failed:', err.message);
        }
    }

    // Test 2: Expired JWT
    testExpiredJWT() {
        const token = jwt.sign({ userId: 123 }, this.secret, { expiresIn: '-1s' });

        try {
            jwt.verify(token, this.secret);
            console.error('✗ Expired JWT was accepted!');
        } catch (err) {
            if (err.name === 'TokenExpiredError') {
                console.log('✓ Expired JWT properly rejected');
            }
        }
    }

    // Test 3: Algorithm confusion attack
    testAlgorithmConfusion() {
        // Generate token with 'none' algorithm
        const header = Buffer.from(JSON.stringify({ alg: 'none', typ: 'JWT' })).toString('base64');
        const payload = Buffer.from(JSON.stringify({ userId: 123, role: 'admin' })).toString('base64');
        const maliciousToken = `${header}.${payload}.`;

        try {
            jwt.verify(maliciousToken, this.secret);
            console.error('✗ Algorithm confusion attack succeeded!');
        } catch (err) {
            console.log('✓ Algorithm confusion attack prevented');
        }
    }

    // Test 4: Weak secret
    testWeakSecret() {
        const weakSecrets = ['secret', '123456', 'password', 'jwt_secret'];

        const token = jwt.sign({ userId: 123 }, 'weak_secret');

        for (const secret of weakSecrets) {
            try {
                jwt.verify(token, secret);
                console.error(`✗ JWT cracked with weak secret: ${secret}`);
                return;
            } catch (err) {
                // Continue trying
            }
        }

        console.log('✓ JWT not vulnerable to weak secret');
    }

    // Test 5: Token tampering
    testTokenTampering() {
        const token = jwt.sign({ userId: 123, role: 'user' }, this.secret);

        // Attempt to modify payload
        const parts = token.split('.');
        const payload = JSON.parse(Buffer.from(parts[1], 'base64').toString());
        payload.role = 'admin';  // Escalate privileges

        const tamperedPayload = Buffer.from(JSON.stringify(payload)).toString('base64');
        const tamperedToken = `${parts[0]}.${tamperedPayload}.${parts[2]}`;

        try {
            jwt.verify(tamperedToken, this.secret);
            console.error('✗ Tampered JWT was accepted!');
        } catch (err) {
            console.log('✓ Tampered JWT properly rejected');
        }
    }
}

// Run tests
const tester = new JWTTester('your-secret-key');
tester.testValidJWT();
tester.testExpiredJWT();
tester.testAlgorithmConfusion();
tester.testWeakSecret();
tester.testTokenTampering();

# API key security tests
import requests
import hashlib

class APIKeyTester:
    def __init__(self, base_url):
        self.base_url = base_url

    def test_api_key_in_header(self, api_key):
        """Test API key in header (secure)"""
        response = requests.get(
            f"{self.base_url}/api/data",
            headers={'X-API-Key': api_key}
        )

        assert response.status_code == 200
        print("✓ API key in header works")

    def test_api_key_in_url(self, api_key):
        """Test API key in URL (insecure - should be rejected)"""
        response = requests.get(
            f"{self.base_url}/api/data?api_key={api_key}"
        )

        # Should reject API keys in URL for security
 (as discussed in [Penetration Testing Basics for QA Testers](/blog/penetration-testing-basics))        if response.status_code == 200:
            print("✗ WARNING: API key accepted in URL (security risk)")
        else:
            print("✓ API key in URL properly rejected")

    def test_api_key_rotation(self, old_key, new_key):
        """Test API key rotation"""
        # Old key should be revoked
        response = requests.get(
            f"{self.base_url}/api/data",
            headers={'X-API-Key': old_key}
        )
        assert response.status_code == 401

        # New key should work
        response = requests.get(
            f"{self.base_url}/api/data",
            headers={'X-API-Key': new_key}
        )
        assert response.status_code == 200
        print("✓ API key rotation successful")

    def test_rate_limiting(self, api_key):
        """Test rate limiting per API key"""
        rate_limit = 100  # requests per minute
        responses = []

        for i in range(rate_limit + 10):
            response = requests.get(
                f"{self.base_url}/api/data",
                headers={'X-API-Key': api_key}
            )
            responses.append(response.status_code)

        # Should hit rate limit
        assert 429 in responses, "Rate limiting not enforced"
        print("✓ Rate limiting enforced")

# SQL injection tests
def test_sql_injection():
    payloads = [
        "' OR '1'='1",
        "'; DROP TABLE users--",
        "' UNION SELECT * FROM users--",
        "1' AND '1'='1",
        "admin'--"
    ]

    for payload in payloads:
        response = requests.post(
            'https://api.example.com/login',
            json={'username': payload, 'password': 'test'}
        )

        # Should not return successful login
        assert response.status_code != 200 or 'token' not in response.json()

        # Should not expose database errors
        assert 'sql' not in response.text.lower()
        assert 'syntax error' not in response.text.lower()

    print("✓ SQL injection tests passed")

# MongoDB injection tests
def test_nosql_injection():
    payloads = [
        {"$ne": None},
        {"$gt": ""},
        {"$regex": ".*"},
        {"$where": "1==1"}
    ]

    for payload in payloads:
        response = requests.post(
            'https://api.example.com/search',
            json={'username': payload}
        )

        # Should not return unauthorized data
        assert response.status_code in [400, 422]  # Bad request

    print("✓ NoSQL injection tests passed")

# Test Broken Object Level Authorization
def test_bola():
    """Test IDOR vulnerability"""

    # User A's token
    token_a = 'user_a_token'

    # User B's resource
    user_b_resource_id = 456

    # Attempt to access User B's resource with User A's token
    response = requests.get(
        f'https://api.example.com/users/{user_b_resource_id}/profile',
        headers={'Authorization': f'Bearer {token_a}'}
    )

    # Should return 403 Forbidden
    assert response.status_code == 403, "IDOR vulnerability detected!"
    print("✓ BOLA/IDOR protection working")

    # Test with User A's own resource
    user_a_resource_id = 123
    response = requests.get(
        f'https://api.example.com/users/{user_a_resource_id}/profile',
        headers={'Authorization': f'Bearer {token_a}'}
    )

    assert response.status_code == 200
    print("✓ Authorized access working")

# Using OWASP ZAP API scan
docker run -v $(pwd):/zap/wrk:rw owasp/zap2docker-stable \
  zap-api-scan.py \
  -t https://api.example.com \
  -f openapi \
  -d /zap/wrk/openapi.json \
  -r /zap/wrk/api-security-report.html

# Using Postman Newman with security tests
newman run api-security-tests.json \
  --environment production.json \
  --reporters cli,html \
  --reporter-html-export security-report.html