119 lines
4.4 KiB
Python
119 lines
4.4 KiB
Python
"""구글 ID 토큰 검증
|
|
브라우저가 받아온 토큰이 구글이 우리 앱 앞으로 발급한 것인지만 본다
|
|
|
|
코드 교환을 하지 않아 client_secret이 없다 — 검증에 필요한 것은 공개키와 client_id뿐이다.
|
|
검증을 통과하면 그 뒤로는 우리 토큰을 쓴다. 구글 토큰을 세션으로 들고 다니지 않는다.
|
|
|
|
빠지면 안 되는 검사 셋
|
|
서명 — 구글 JWKS 공개키. 없으면 아무나 만든 JSON이 통과한다
|
|
aud — 우리 client_id. 없으면 다른 앱 앞으로 발급된 진짜 구글 토큰이 통과한다
|
|
iss — accounts.google.com
|
|
"""
|
|
import asyncio
|
|
import time
|
|
|
|
import httpx
|
|
from jose import JWTError, jwt
|
|
|
|
from settings import settings
|
|
|
|
JWKS_URL = "https://www.googleapis.com/oauth2/v3/certs"
|
|
# 구글이 두 표기를 다 쓴다. 한쪽만 받으면 어느 날 전원 로그인 실패가 된다
|
|
ISSUERS = ("accounts.google.com", "https://accounts.google.com")
|
|
JWKS_TTL_SECONDS = 3600
|
|
HTTP_TIMEOUT = 5.0
|
|
|
|
_jwks: dict | None = None
|
|
_jwks_fetched_at = 0.0
|
|
# 토큰이 동시에 여러 개 들어와도 JWKS는 한 번만 받는다
|
|
_jwks_lock = asyncio.Lock()
|
|
|
|
|
|
class GoogleLoginDisabled(RuntimeError):
|
|
"""GOOGLE_CLIENT_ID 없음 — 구글 로그인만 꺼지고 서버는 뜬다"""
|
|
|
|
|
|
class GoogleTokenInvalid(RuntimeError):
|
|
"""서명·aud·iss·만료 중 하나가 어긋남"""
|
|
|
|
|
|
class GoogleAccount:
|
|
def __init__(self, sub: str, email: str, name: str, picture: str):
|
|
self.sub = sub
|
|
self.email = email
|
|
self.name = name
|
|
self.picture = picture
|
|
|
|
|
|
def is_enabled() -> bool:
|
|
return bool(settings.google_client_id)
|
|
|
|
|
|
async def fetch_jwks() -> dict:
|
|
async with httpx.AsyncClient(timeout=HTTP_TIMEOUT) as client:
|
|
response = await client.get(JWKS_URL)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
async def get_jwks(*, force: bool = False) -> dict:
|
|
global _jwks, _jwks_fetched_at
|
|
async with _jwks_lock:
|
|
fresh = _jwks is not None and (time.monotonic() - _jwks_fetched_at) < JWKS_TTL_SECONDS
|
|
if fresh and not force:
|
|
return _jwks
|
|
try:
|
|
_jwks = await fetch_jwks()
|
|
_jwks_fetched_at = time.monotonic()
|
|
except Exception as failure:
|
|
print(f"[google] JWKS 조회 실패: {failure}", flush=True)
|
|
# 낡은 캐시라도 있으면 그걸로 간다. 구글이 잠깐 안 될 때 로그인이 통째로 죽는 것보다 낫다
|
|
if _jwks is None:
|
|
raise GoogleTokenInvalid("구글 공개키를 받지 못했습니다") from failure
|
|
return _jwks
|
|
|
|
|
|
def has_key(jwks: dict, kid: str | None) -> bool:
|
|
return any(key.get("kid") == kid for key in jwks.get("keys") or [])
|
|
|
|
|
|
async def verify_id_token(credential: str) -> GoogleAccount:
|
|
if not is_enabled():
|
|
raise GoogleLoginDisabled("GOOGLE_CLIENT_ID 없음")
|
|
if not credential:
|
|
raise GoogleTokenInvalid("빈 토큰")
|
|
|
|
try:
|
|
kid = jwt.get_unverified_header(credential).get("kid")
|
|
except JWTError as failure:
|
|
raise GoogleTokenInvalid("토큰 형식이 아님") from failure
|
|
|
|
jwks = await get_jwks()
|
|
# 공개키는 주기적으로 바뀐다. 캐시에 없는 kid면 한 번만 다시 받는다
|
|
if not has_key(jwks, kid):
|
|
jwks = await get_jwks(force=True)
|
|
|
|
try:
|
|
claims = jwt.decode(
|
|
credential, jwks, algorithms=["RS256"],
|
|
audience=settings.google_client_id, issuer=ISSUERS,
|
|
# at_hash는 access_token과 짝일 때만 의미가 있다. 브라우저가 주는
|
|
# 크리덴셜에는 access_token이 없어 켜 두면 정상 토큰이 거부된다
|
|
options={"verify_at_hash": False},
|
|
)
|
|
except JWTError as failure:
|
|
# 사유는 로그에만 남긴다
|
|
print(f"[google] ID 토큰 거부: {failure}", flush=True)
|
|
raise GoogleTokenInvalid("구글 토큰이 유효하지 않습니다") from failure
|
|
|
|
sub = str(claims.get("sub") or "")
|
|
if not sub:
|
|
raise GoogleTokenInvalid("sub 없음")
|
|
# 미인증 이메일은 신원으로 쓸 수 없다
|
|
if not claims.get("email_verified"):
|
|
raise GoogleTokenInvalid("이메일이 인증되지 않은 계정입니다")
|
|
|
|
return GoogleAccount(sub=sub, email=str(claims.get("email") or ""),
|
|
name=str(claims.get("name") or ""),
|
|
picture=str(claims.get("picture") or ""))
|