1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

sequenceDiagram
    participant Client
    participant STS as Security Token Service
    participant ServiceA as Service A (Audience)
    participant ServiceB as Service B (Target)

    Client->>STS: POST /token (grant_type=urn:ietf:params:oauth:grant-type:token-exchange)
    Note over Client,STS: subject_token=access_token, audience=ServiceB
    STS->>STS: Validate subject_token
    STS-->>Client: New access_token for ServiceB
    Client->>ServiceB: GET /api/resource (Authorization: Bearer new_token)
    ServiceB-->>Client: 200 OK

 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

import requests
import json
import base64
from datetime import datetime, timedelta

# STS 엔드포인트 (예: Keycloak, Okta, Auth0)
sts_url = "https://auth.example.com/oauth/token"

# 클라이언트 자격 증명 (기밀 클라이언트)
client_id = "my-client"
client_secret = "my-secret"

# 기존 액세스 토큰 (subject_token)
subject_token = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9..."

def exchange_token(subject_token, target_audience, target_scope):
    """
    OAuth 2.0 Token Exchange 요청을 보내는 함수
    """
    # 요청 바디 구성
    payload = {
        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
        "subject_token": subject_token,
        "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
        "audience": target_audience,
        "scope": target_scope,
        "requested_token_type": "urn:ietf:params:oauth:token-type:access_token"
    }
    
    # 클라이언트 인증 (Basic Auth)
    auth_header = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
    headers = {
        "Content-Type": "application/x-www-form-urlencoded",
        "Authorization": f"Basic {auth_header}"
    }
    
    try:
        response = requests.post(sts_url, data=payload, headers=headers, timeout=10)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"Token Exchange 실패: {e}")
        return None

# 사용 예시
new_token = exchange_token(
    subject_token=subject_token,
    target_audience="https://api.payment-service.com",
    target_scope="payment:read payment:write"
)

if new_token:
    print(f"새로운 액세스 토큰: {new_token.get('access_token')[:50]}...")
    print(f"토큰 타입: {new_token.get('token_type')}")
    print(f"만료 시간: {new_token.get('expires_in')}초")
else:
    print("토큰 교환 실패")

 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

from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
from jwt import PyJWKClient

app = FastAPI()
security = HTTPBearer()

# STS의 JWKS 엔드포인트
jwks_url = "https://auth.example.com/oauth/jwks"
jwks_client = PyJWKClient(jwks_url)

def verify_token(credentials: HTTPAuthorizationCredentials = Security(security)):
    token = credentials.credentials
    try:
        # JWKS를 사용하여 토큰 서명 검증
        signing_key = jwks_client.get_signing_key_from_jwt(token)
        payload = jwt.decode(
            token,
            signing_key.key,
            algorithms=["RS256"],
            audience="https://api.payment-service.com",  # 대상 서비스 확인
            issuer="https://auth.example.com"  # 발급자 확인
        )
        return payload
    except jwt.PyJWTError as e:
        raise HTTPException(status_code=401, detail=f"토큰 검증 실패: {e}")

@app.get("/api/payment")
async def get_payment(payload: dict = Depends(verify_token)):
    return {"message": "Payment data", "user": payload.get("sub")}