spakky-oauth

spakky-oauth는 application이 사용자를 대신해 외부 API를 부를 때 쓰는 OAuth 2.0 public client 플러그인입니다. Authorization code + PKCE로 token을 받고, 만료 전에 갱신하고, 로그아웃 시 폐기합니다.

spakky-oidc와 방향이 반대입니다. spakky-oidc는 들어오는 bearer token을 검증하고, spakky-oauth는 나가는 호출에 쓸 token을 얻어 보관합니다.

제공 기능

• Authorization code flow 시작: state, PKCE S256 challenge, openid scope가 있으면 nonce 생성
• Callback code 교환과 token 저장
• 만료 refresh_leeway_seconds 전 갱신, refresh token 회전 반영
• 같은 주체의 동시 갱신을 하나로 묶는 process-local 잠금
• 다른 instance가 먼저 회전해 저장한 token 재사용 (best-effort; 분산 잠금은 아님)
• Revocation endpoint 호출과 local token 삭제

포함하지 않는 것

• 로그인·callback HTTP route. Application의 inbound adapter(FastAPI 등)가 OAuthClient를 호출합니다.
• ID token 서명·claim 검증. OAuthAuthorizationResult가 tokens.id_token과 nonce를 돌려주므로 application이 검증합니다.
• Confidential client 인증(client_secret), device code, client credentials grant.
• 여러 process 사이의 분산 잠금. 저장소 구현이 필요하면 직접 제공합니다.

설치

pip install spakky-oauth

구성

OAuthConfig는 SPAKKY_OAUTH__ prefix의 환경 변수에서 읽습니다.

export SPAKKY_OAUTH__CLIENTS='{
  "example": {
    "authorization_endpoint": "https://auth.example.test/authorize",
    "token_endpoint": "https://auth.example.test/token",
    "revocation_endpoint": "https://auth.example.test/revoke",
    "client_id": "my-app",
    "scopes": ["openid", "offline_access"],
    "token_parameters": {"resource": "https://api.example.test"}
  }
}'

OAuthConfig

OAuthClientRegistration

Endpoint는 https여야 하며 loopback host에 한해 http를 허용합니다. 추가 parameter는 client_id, state, code_verifier 같은 protocol field를 덮어쓸 수 없습니다.

저장소 등록

Plugin은 OAuthConfig와 OAuthClient만 등록합니다. Token과 authorization state 저장소는 application이 Pod로 등록합니다. 개발과 테스트에는 in-memory 구현을 쓸 수 있습니다.

from spakky.core.pod.annotations.pod import Pod
from spakky.plugins.oauth.store import (
    InMemoryOAuthAuthorizationStateStore,
    InMemoryOAuthTokenStore,
)


@Pod()
class DevTokenStore(InMemoryOAuthTokenStore): ...


@Pod()
class DevStateStore(InMemoryOAuthAuthorizationStateStore): ...

운영에서는 IOAuthTokenStore와 IOAuthAuthorizationStateStore를 database나 cache로 구현합니다. In-memory 구현은 만료된 authorization을 지우지 않으므로 운영에 쓰지 않습니다.

OAuthTokenSet과 OAuthPendingAuthorization의 token과 code_verifier는 SecretStr입니다. model_dump_json()은 이 값을 **********로 바꾸므로 그대로 저장하면 token을 잃습니다. 저장소 구현은 get_secret_value()로 값을 꺼내 직접 직렬화하고, 암호화도 책임집니다.

사용법

from spakky.plugins.oauth.client import OAuthClient
from spakky.plugins.oauth.error import OAuthAuthorizationRequiredError


async def login(oauth: OAuthClient, user_id: str) -> str:
    request = await oauth.begin_authorization(
        "example",
        user_id,
        "https://app.example.test/auth/callback",
    )
    return request.url  # 사용자를 이 URL로 redirect


async def callback(oauth: OAuthClient, state: str, code: str) -> None:
    # 먼저 state가 이 브라우저 session에서 시작한 값인지 대조합니다.
    result = await oauth.complete_authorization(state, code)
    # result.tokens.id_token과 result.nonce로 ID token을 검증합니다.


async def call_api(oauth: OAuthClient, user_id: str) -> str:
    try:
        return await oauth.access_token("example", user_id)
    except OAuthAuthorizationRequiredError:
        # 다시 로그인이 필요합니다.
        raise

Callback은 반드시 로그인한 session에 묶습니다. begin_authorization()이 돌려준 state를 사용자의 브라우저 session에 저장하고, callback에서 받은 state가 그 값과 같은지 확인한 뒤 complete_authorization()을 호출합니다. 이 확인이 없으면 공격자가 자기 주체로 시작한 URL을 피해자에게 열게 해 피해자의 token을 공격자 주체 아래 저장할 수 있습니다.

Resource server가 token을 401로 거절하면 그 token을 넘겨 한 번 갱신합니다. 저장된 token이 이미 다른 token으로 바뀌었다면 갱신하지 않고 새 token을 돌려줍니다.

token = await oauth.access_token("example", user_id, rejected_token=token)

Sign-in마다 client id를 발급하는 server는 begin_authorization(..., client_id=...)와 complete_authorization(..., client_id=...)로 실제 client id를 전달합니다. 요청별 추가 parameter는 begin_authorization(..., parameters={...})로 넘깁니다.

에러

Token endpoint가 error="invalid_grant"로 답한 경우에만 OAuthAuthorizationRequiredError를 올립니다. 그 밖의 4xx/5xx는 OAuthTokenEndpointError입니다. 두 경우 모두 access_token()은 저장된 token을 지우지 않습니다. 다른 instance가 방금 회전해 저장하려는 token을 지우지 않기 위해서입니다. 다시 로그인시키기 전에 저장분을 비우려면 revoke()를 호출합니다.

라이선스

MIT