1
# 步骤 1：重定向到授权端点
2
GET /authorize?
3
    response_type=code&
4
    client_id=YOUR_CLIENT_ID&
5
    redirect_uri=https://example.com/callback&
6
    scope=openid profile email&
7
    state=RANDOM_STATE
8

9
# 步骤 2：用户授权后，回调
10
GET /callback?code=AUTH_CODE&state=RANDOM_STATE
11

12
# 步骤 3：用 code 换 token（后端→授权服务器，不经过浏览器）
13
POST /token
14
    grant_type=authorization_code&
15
    code=AUTH_CODE&
16
    redirect_uri=https://example.com/callback&
17
    client_id=YOUR_CLIENT_ID&
18
    client_secret=YOUR_CLIENT_SECRET

1
# Flask 实现 OAuth2 授权码流程
2
from flask import Flask, request, redirect, session
3
import requests, secrets
4

5
app = Flask(__name__)
6
app.secret_key = "your-secret-key"
7
CLIENT_ID, CLIENT_SECRET = "your-client-id", "your-client-secret"
8
REDIRECT_URI = "https://example.com/callback"
9
AUTH_SERVER = "https://accounts.google.com"
10

11
@app.route("/login")
12
def login():
13
    state = secrets.token_urlsafe(32)
14
    session["oauth_state"] = state
15
    return redirect(
16
        f"{AUTH_SERVER}/o/oauth2/v2/auth?response_type=code&client_id={CLIENT_ID}"
17
        f"&redirect_uri={REDIRECT_URI}&scope=openid profile email&state={state}"
18
    )
19

20
@app.route("/callback")
21
def callback():
22
    if request.args.get("state") != session.pop("oauth_state", None):
23
        return "State 验证失败", 400
24
    code = request.args.get("code")
25
    if not code: return "授权失败", 400
26
    token_resp = requests.post(f"{AUTH_SERVER}/o/oauth2/token", data={
27
        "grant_type": "authorization_code", "code": code,
28
        "redirect_uri": REDIRECT_URI,
29
        "client_id": CLIENT_ID, "client_secret": CLIENT_SECRET,
30
    })
31
    return "登录成功"

1
# PKCE 实现
2
import hashlib, base64, os
3

4
# 生成 code_verifier（43-128 字符）和 code_challenge（SHA-256 哈希）
5
code_verifier = base64.urlsafe_b64encode(os.urandom(32)).rstrip(b'=').decode()
6
code_challenge = base64.urlsafe_b64encode(
7
    hashlib.sha256(code_verifier.encode()).digest()
8
).rstrip(b'=').decode()
9

10
# 授权请求携带 code_challenge；Token 请求携带 code_verifier（原始值，非哈希）
11
auth_url = (
12
    f"{AUTH_SERVER}/authorize?response_type=code&client_id={CLIENT_ID}&"
13
    f"redirect_uri={REDIRECT_URI}&scope=openid profile&"
14
    f"code_challenge={code_challenge}&code_challenge_method=S256"
15
)
16
token_data = {"grant_type": "authorization_code", "code": auth_code,
17
              "client_id": CLIENT_ID, "code_verifier": code_verifier}

1
POST /token
2
    grant_type=client_credentials&
3
    client_id=SERVICE_ID&
4
    client_secret=SERVICE_SECRET&
5
    scope=api.read

1
# Python 服务间调用示例
2
import requests
3

4
def get_service_token():
5
    resp = requests.post("https://auth.example.com/oauth2/token", data={
6
        "grant_type": "client_credentials",
7
        "client_id": "my-service-id",
8
        "client_secret": "my-service-secret",
9
        "scope": "api.read api.write",
10
    })
11
    return resp.json()["access_token"]
12

13
def call_internal_api():
14
    token = get_service_token()
15
    resp = requests.get(
16
        "https://api.internal.example.com/users",
17
        headers={"Authorization": f"Bearer {token}"},
18
    )
19
    return resp.json()

1
import hashlib, base64, secrets
2

3
def generate_pkce_pair():
4
    """生成 PKCE 的 code_verifier 和 code_challenge 对"""
5
    code_verifier = base64.urlsafe_b64encode(
6
        secrets.token_bytes(32)  # 32 字节 → 43 字符 Base64
7
    ).rstrip(b'=').decode('ascii')
8

9
    digest = hashlib.sha256(code_verifier.encode('ascii')).digest()
10
    code_challenge = base64.urlsafe_b64encode(digest).rstrip(b'=').decode('ascii')
11
    return code_verifier, code_challenge
12

13
verifier, challenge = generate_pkce_pair()
14
print(f"code_verifier:  {verifier}")   # 发送给 /token 端点
15
print(f"code_challenge: {challenge}")  # 发送给 /authorize 端点

1
{
2
  "iss": "https://accounts.google.com",
3
  "sub": "1234567890",
4
  "aud": "your-client-id",
5
  "exp": 1699999999,
6
  "iat": 1699999000,
7
  "auth_time": 1699998900,
8
  "nonce": "n-0S6_WzA2Mj",
9
  "name": "Zhang San",
10
  "email": "zhangsan@example.com",
11
  "email_verified": true
12
}

1
GET /userinfo
2
Authorization: Bearer <access_token>

1
{
2
  "sub": "1234567890",
3
  "name": "Zhang San",
4
  "given_name": "San",
5
  "family_name": "Zhang",
6
  "email": "zhangsan@example.com",
7
  "phone_number": "+86 13800138000",
8
  "address": { "formatted": "北京市朝阳区xxx", "country": "CN" }
9
}

1
{
2
  "issuer": "https://accounts.google.com",
3
  "authorization_endpoint": "https://accounts.google.com/o/oauth2/v2/auth",
4
  "token_endpoint": "https://oauth2.googleapis.com/token",
5
  "userinfo_endpoint": "https://openidconnect.googleapis.com/v1/userinfo",
6
  "jwks_uri": "https://www.googleapis.com/oauth2/v3/certs",
7
  "response_types_supported": ["code", "token", "id_token"],
8
  "id_token_signing_alg_values_supported": ["RS256"],
9
  "scopes_supported": ["openid", "email", "profile"]
10
}

1
# Python 动态发现 OIDC 配置
2
import requests
3

4
def discover_oidc_config(issuer):
5
    resp = requests.get(f"{issuer}/.well-known/openid-configuration")
6
    resp.raise_for_status()
7
    return resp.json()
8

9
config = discover_oidc_config("https://accounts.google.com")
10
print(f"授权端点: {config['authorization_endpoint']}")
11
print(f"Token 端点: {config['token_endpoint']}")
12
print(f"JWKS 端点: {config['jwks_uri']}")

1
# JWT 格式的 Access Token 解码示例
2
import jwt  # PyJWT 库
3

4
token = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.signature"
5
decoded = jwt.decode(token, options={"verify_signature": False})
6
print(decoded)
7
# {'sub': '1234567890', 'name': 'John Doe', 'aud': 'client-id', 'exp': 1699999999}

1
package main
2

3
import (
4
    "crypto/sha256"
5
    "encoding/base64"
6
    "fmt"
7
    "net/http"
8
    "sync"
9
    "time"
10
    "github.com/golang-jwt/jwt/v5"
11
)
12

13
type AuthServer struct {
14
    clients   map[string]*Client
15
    authCodes map[string]*AuthCode
16
    jwtSecret []byte
17
    mu        sync.RWMutex
18
}
19
type Client struct{ ID, Secret string; RedirectURIs []string }
20
type AuthCode struct {
21
    Code, ClientID, RedirectURI, CodeChallenge string
22
    ExpiresAt time.Time; Used bool
23
}
24

25
// Authorize: 验证 client_id + redirect_uri，生成授权码
26
func (s *AuthServer) Authorize(w http.ResponseWriter, r *http.Request) {
27
    clientID := r.URL.Query().Get("client_id")
28
    redirectURI := r.URL.Query().Get("redirect_uri")
29
    codeChallenge := r.URL.Query().Get("code_challenge")
30
    state := r.URL.Query().Get("state")
31
    client, ok := s.clients[clientID]
32
    if !ok { http.Error(w, "invalid client_id", 400); return }
33
    validURI := false
34
    for _, uri := range client.RedirectURIs {
35
        if uri == redirectURI { validURI = true; break }
36
    }
37
    if !validURI { http.Error(w, "invalid redirect_uri", 400); return }
38
    code := randomString(32)
39
    s.mu.Lock()
40
    s.authCodes[code] = &AuthCode{
41
        Code: code, ClientID: clientID, RedirectURI: redirectURI,
42
        CodeChallenge: codeChallenge, ExpiresAt: time.Now().Add(10 * time.Minute),
43
    }
44
    s.mu.Unlock()
45
    http.Redirect(w, r, fmt.Sprintf("%s?code=%s&state=%s", redirectURI, code, state), 302)
46
}
47

48
// Token: 验证授权码 + PKCE，签发 JWT
49
func (s *AuthServer) Token(w http.ResponseWriter, r *http.Request) {
50
    code := r.FormValue("code")
51
    codeVerifier := r.FormValue("code_verifier")
52
    s.mu.Lock()
53
    ac, ok := s.authCodes[code]
54
    if !ok || ac.Used || time.Now().After(ac.ExpiresAt) {
55
        http.Error(w, "invalid code", 400); s.mu.Unlock(); return
56
    }
57
    if ac.CodeChallenge != "" {
58
        h := sha256.Sum256([]byte(codeVerifier))
59
        if base64.RawURLEncoding.EncodeToString(h[:]) != ac.CodeChallenge {
60
            http.Error(w, "PKCE failed", 400); s.mu.Unlock(); return
61
        }
62
    }
63
    ac.Used = true; s.mu.Unlock()
64
    token := jwt.NewWithClaims(jwt.SigningMethodHS256, jwt.MapClaims{
65
        "sub": "user-123", "aud": ac.ClientID, "scope": "openid profile email",
66
        "exp": time.Now().Add(time.Hour).Unix(),
67
    })
68
    accessToken, _ := token.SignedString(s.jwtSecret)
69
    w.Header().Set("Content-Type", "application/json")
70
    fmt.Fprintf(w, `{"access_token":"%s","token_type":"Bearer","expires_in":3600}`, accessToken)
71
}
72

73
func main() {
74
    s := &AuthServer{
75
        clients: map[string]*Client{
76
            "test-client": {ID: "test-client", Secret: "test-secret",
77
                RedirectURIs: []string{"http://localhost:8080/callback"}},
78
        },
79
        authCodes: make(map[string]*AuthCode),
80
        jwtSecret: []byte("your-256-bit-secret"),
81
    }
82
    http.HandleFunc("/authorize", s.Authorize)
83
    http.HandleFunc("/token", s.Token)
84
    http.ListenAndServe(":9090", nil)
85
}

1
# 1. 用 curl 模拟授权码换 token
2
curl -X POST https://auth.example.com/token \
3
  -d "grant_type=authorization_code" \
4
  -d "code=YOUR_AUTH_CODE" \
5
  -d "redirect_uri=http://localhost:8080/callback" \
6
  -d "client_id=test" -d "client_secret=secret"
7

8
# 2. 解码 JWT Token（不验证签名）
9
echo "YOUR_JWT_TOKEN" | cut -d'.' -f2 | base64 -d 2>/dev/null | python3 -m json.tool
10

11
# 3. 查看 OIDC 发现文档
12
curl https://auth.example.com/.well-known/openid-configuration | python3 -m json.tool
13

14
# 4. 获取 JWKS（公钥，用于验证 JWT 签名）
15
curl https://auth.example.com/.well-known/jwks.json | python3 -m json.tool

1
import hashlib, base64, secrets
2

3
# Step 1: 生成 code_verifier（高熵随机字符串）
4
code_verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).rstrip(b'=').decode()
5

6
# Step 2: 计算 code_challenge = BASE64URL(SHA256(code_verifier))
7
code_challenge = base64.urlsafe_b64encode(
8
    hashlib.sha256(code_verifier.encode()).digest()
9
).rstrip(b'=').decode()
10

11
print(f"code_verifier:  {code_verifier}")
12
print(f"  (长度: {len(code_verifier)} chars, 32 bytes 原始数据)")
13
print(f"code_challenge: {code_challenge}")
14
print(f"  = BASE64URL(SHA256(code_verifier))")
15

16
# 验证：重新计算 challenge 应与原始匹配
17
recomputed = base64.urlsafe_b64encode(
18
    hashlib.sha256(code_verifier.encode()).digest()
19
).rstrip(b'=').decode()
20
print(f"重新计算匹配: {recomputed == code_challenge}")

1
code_verifier:  U7j1QxRl9HRvlszG63RFQqVcMzW1Jr181yczQAjqLyk
2
  (长度: 43 chars, 32 bytes 原始数据)
3
code_challenge: hoZLEUI4_x5OuPB068oKDKNtCdjGXRS_EcaXcKRb4-s
4
  = BASE64URL(SHA256(code_verifier))
5
重新计算匹配: True

1
import hashlib, base64, secrets, json
2

3
# Step 1: 客户端准备（生成 state 和 PKCE 参数）
4
state = secrets.token_urlsafe(16)
5
code_verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).rstrip(b'=').decode()
6
code_challenge = base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode()).digest()).rstrip(b'=').decode()
7

8
print("Step 1: 客户端准备")
9
print(f"  state:          {state}")
10
print(f"  code_verifier:  {code_verifier}")
11
print(f"  code_challenge: {code_challenge}")
12

13
# Step 2: 构造授权 URL（用户在浏览器中访问）
14
auth_url = (
15
    f"https://auth.example.com/authorize?"
16
    f"response_type=code&"
17
    f"client_id=my-app&"
18
    f"redirect_uri=http://localhost:8080/callback&"
19
    f"scope=openid profile email&"
20
    f"state={state}&"
21
    f"code_challenge={code_challenge}&"
22
    f"code_challenge_method=S256"
23
)
24
print(f"\nStep 2: 授权 URL")
25
print(f"  {auth_url[:80]}...")
26

27
# Step 3: 授权服务器返回授权码
28
auth_code = secrets.token_urlsafe(16)
29
print(f"\nStep 3: 授权服务器返回")
30
print(f"  authorization_code: {auth_code}")
31
print(f"  state: {state} (客户端验证与 Step 1 的 state 匹配)")
32

33
# Step 4: 客户端用授权码 + code_verifier 交换 token
34
print(f"\nStep 4: Token 交换请求")
35
print(f"  POST /token")
36
print(f"  grant_type=authorization_code")
37
print(f"  code={auth_code}")
38
print(f"  code_verifier={code_verifier}")
39
print(f"  → 服务器验证: SHA256(code_verifier) == code_challenge ")

1
Step 1: 客户端准备
2
  state:          Z6cKscUxwFpGp_NNM6NEig
3
  code_verifier:  wfqsnk7mBiPGMurej-VqEQ7dwOsaR32ePJCgsPHdUN0
4
  code_challenge: _57HreRsePbg2wTzNUBBOLXX08Q5qEgQXs7eF-eI1ZU
5

6
Step 2: 授权 URL
7
  https://auth.example.com/authorize?response_type=code&client_id=my-app&redirect_...
8

9
Step 3: 授权服务器返回
10
  authorization_code: yuRzyVu88TWb1drDlnSKKw
11
  state: Z6cKscUxwFpGp_NNM6NEig (客户端验证与 Step 1 的 state 匹配)
12

13
Step 4: Token 交换请求
14
  POST /token
15
  grant_type=authorization_code
16
  code=yuRzyVu88TWb1drDlnSKKw
17
  code_verifier=wfqsnk7mBiPGMurej-VqEQ7dwOsaR32ePJCgsPHdUN0
18
  → 服务器验证: SHA256(code_verifier) == code_challenge

1
import jwt, json, base64
2

3
# 模拟 Access Token
4
secret = "jwt-secret-key"
5
token = jwt.encode({
6
    "iss": "https://auth.example.com",
7
    "sub": "user-12345",
8
    "aud": "my-app",
9
    "exp": 1778167624,
10
    "iat": 1778164024,
11
    "scope": "openid profile email",
12
    "client_id": "my-app"
13
}, secret, algorithm="HS256")
14

15
# 解码 JWT（不验证签名——仅用于调试）
16
header = json.loads(base64.urlsafe_b64decode(token.split('.')[0] + '=='))
17
payload = json.loads(base64.urlsafe_b64decode(token.split('.')[1] + '=='))
18

19
print(f"JWT Header:")
20
print(json.dumps(header, indent=2))
21
print(f"\nJWT Payload:")
22
print(json.dumps(payload, indent=2))
23
print(f"\n  JWT 是 Base64 编码而非加密——任何人都可以解码查看内容")

1
JWT Header:
2
{
3
  "alg": "HS256",
4
  "typ": "JWT"
5
}
6

7
JWT Payload:
8
{
9
  "iss": "https://auth.example.com",
10
  "sub": "user-12345",
11
  "aud": "my-app",
12
  "exp": 1778167624,
13
  "iat": 1778164024,
14
  "scope": "openid profile email",
15
  "client_id": "my-app"
16
}
17

18
  JWT 是 Base64 编码而非加密——任何人都可以解码查看内容

1
# 解码 JWT Payload（不验证签名）
2
echo 'PAYLOAD_PART' | base64 -d 2>/dev/null | python3 -m json.tool
3

4
# 查看 OIDC 发现文档
5
curl https://auth.example.com/.well-known/openid-configuration | python3 -m json.tool
6

7
# 获取 JWKS（公钥，用于验证 JWT 签名）
8
curl https://auth.example.com/.well-known/jwks.json | python3 -m json.tool

1
import secrets
2

3
# 正常流程：state 匹配
4
original_state = secrets.token_urlsafe(16)
5
returned_state = original_state
6
print(f"正常流程:")
7
print(f"  客户端发送 state: {original_state}")
8
print(f"  服务器返回 state: {returned_state}")
9
print(f"  匹配: {original_state == returned_state} ")
10

11
# CSRF 攻击：攻击者伪造 state
12
attacker_state = secrets.token_urlsafe(16)
13
print(f"\nCSRF 攻击:")
14
print(f"  客户端发送 state: {original_state}")
15
print(f"  攻击者构造 state: {attacker_state}")
16
print(f"  匹配: {original_state == attacker_state}  → 拒绝请求")

1
正常流程:
2
  客户端发送 state: njmI0TseDc1z_soOmcPxbw
3
  服务器返回 state: njmI0TseDc1z_soOmcPxbw
4
  匹配: True
5

6
CSRF 攻击:
7
  客户端发送 state: njmI0TseDc1z_soOmcPxbw
8
  攻击者构造 state: lyD_P0zaVZasB9-EJVIqdg
9
  匹配: False  → 拒绝请求