OAuth2 with PKCE — 현대 canonical flow

# Client 쪽 — PKCE verifier 와 challenge 생성
import secrets, hashlib, base64

# Step 1: verifier 생성 (random URL-safe string, 43-128 글자)
code_verifier = secrets.token_urlsafe(64)

# Step 2: challenge = base64url(sha256(verifier)) 유도
challenge_bytes = hashlib.sha256(code_verifier.encode('ascii')).digest()
code_challenge  = base64.urlsafe_b64encode(challenge_bytes).rstrip(b'=').decode('ascii')

# Step 3: authorization URL 구성
from urllib.parse import urlencode
auth_url = 'https://auth.example.com/authorize?' + urlencode({
    'response_type':         'code',
    'client_id':             'my-app-id',
    'redirect_uri':          'https://my-app.example.com/callback',
    'scope':                 'read:user write:repo',
    'state':                 secrets.token_urlsafe(16),  # CSRF 보호
    'code_challenge':        code_challenge,
    'code_challenge_method': 'S256',
})
print('브라우저에서 이 URL 열어:', auth_url)
# code_verifier 저장 — 교환 step 에 필요

# Client 쪽 — authorization code 를 token 으로 교환
import httpx

# Auth server 가 redirect_uri?code=...&state=... 로 redirect 후
# (보낸 state 일치 검증!)

resp = httpx.post(
    'https://auth.example.com/token',
    data={
        'grant_type':   'authorization_code',
        'client_id':    'my-app-id',
        'redirect_uri': 'https://my-app.example.com/callback',
        'code':         'redirect 에서 받은 code',
        'code_verifier': code_verifier,    # <-- 이 flow 시작했다는 증명
        # confidential client 는 client_secret 도 여기 보냄; public client 안 보냄
    },
)
tokens = resp.json()
print(tokens)
# {
#   "access_token":  "eyJhbGciOi...",
#   "token_type":    "Bearer",
#   "expires_in":    3600,
#   "refresh_token": "def50200...",
#   "scope":         "read:user write:repo"
# }

# Server 쪽 — auth server 가 token 교환 동안 PKCE validate
from fastapi import FastAPI, Form, HTTPException
import hashlib, base64, secrets, time

app = FastAPI()
_codes: dict[str, dict] = {}  # code -> {client_id, code_challenge, redirect_uri, user_id, exp}

@app.post('/token')
async def token(
    grant_type:   str = Form(...),
    client_id:    str = Form(...),
    redirect_uri: str = Form(...),
    code:         str = Form(...),
    code_verifier: str = Form(...),
):
    if grant_type != 'authorization_code':
        raise HTTPException(400, detail='미지원 grant_type')
    record = _codes.pop(code, None)
    if not record or record['exp'] < time.time():
        raise HTTPException(400, detail='무효 혹은 만료된 code')
    if record['client_id'] != client_id or record['redirect_uri'] != redirect_uri:
        raise HTTPException(400, detail='client_id 혹은 redirect_uri 불일치')

    # 핵심 체크: client 가 verifier 알았어?
    derived_challenge = base64.urlsafe_b64encode(
        hashlib.sha256(code_verifier.encode('ascii')).digest()
    ).rstrip(b'=').decode('ascii')
    if derived_challenge != record['code_challenge']:
        raise HTTPException(400, detail='PKCE verifier 불일치')

    # 다 좋음 — access token 발급 (JWT 나 opaque)
    return {
        'access_token':  issue_access_token(record['user_id']),
        'token_type':    'Bearer',
        'expires_in':    3600,
        'refresh_token': issue_refresh_token(record['user_id']),
    }