ISMS-P로 읽는 JWT 인증/인가

1eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJhYmMxMjMiLCJleHAiOjE3ODk5OTk5OTl9.4f8a...
2 
3└──── Header ────┘ └──────────── Payload ────────────┘ └ Signature ┘

1app/auth/
2├── jwt.py            # JWT 발급/검증
3├── router.py         # 로그인, OAuth, /me, 로그아웃
4├── service.py         # OAuth 인증 서비스
5├── dependencies.py    # get_current_user 의존성
6├── schemas.py          # 요청/응답 스키마
7└── providers/
8    ├── google.py
9    └── kakao.py

1from datetime import datetime, timedelta, timezone
2 
3from fastapi import HTTPException, status
4from jose import JWTError, jwt
5 
6from app.core.config import settings
7 
8ALGORITHM = "HS256"
9 
10 
11def create_access_token(user_id: str, extra_claims: dict | None = None) -> str:
12    expire = datetime.now(timezone.utc) + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
13    payload = {"sub": user_id, "exp": expire}
14    if extra_claims:
15        payload.update(extra_claims)
16    return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm=ALGORITHM)
17 
18 
19def decode_access_token(token: str) -> str:
20    """sub 값만 반환 (get_current_user 의존성용)"""
21    try:
22        payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[ALGORITHM])
23        user_id: str = payload.get("sub")
24        if not user_id:
25            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
26        return user_id
27    except JWTError:
28        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")

1class Settings(BaseSettings):
2    JWT_SECRET_KEY: str
3    ACCESS_TOKEN_EXPIRE_MINUTES: int = 1440   # 24시간

1# app/auth/schemas.py
2class TokenResponse(BaseModel):
3    access_token: str
4    token_type: str = "bearer"
5    is_new_user: bool = False
6    profile_hint: Optional[ProfileHint] = None

1_MOCK_USERS: dict[str, dict] = {
2    "test@mfds.com": {"password": "test1234", "name": "김건강"},
3    "user@example.com": {"password": "password", "name": "홍길동"},
4}
5 
6 
7@router.post("/login", tags=["auth"])
8async def mock_login(body: LoginIn):
9    mock = _MOCK_USERS.get(body.email)
10    if not mock or mock["password"] != body.password:
11        raise HTTPException(status_code=401, detail="이메일 또는 비밀번호가 올바르지 않습니다.")
12 
13    # DB 저장 없이 이메일을 sub으로 토큰 발급
14    token = create_access_token(body.email)
15    return {
16        "access_token": token,
17        "token_type": "bearer",
18        "user": {"email": body.email, "name": mock["name"]},
19    }

1@router.get("/google/callback")
2async def google_callback(code: str, request: Request, redirect_uri: str | None = None, db: AsyncSession = Depends(get_db)):
3    try:
4        redirect_uri = redirect_uri or str(request.url_for("google_callback"))
5        user_info = await _google.fetch_user_info(code, redirect_uri=redirect_uri)
6        return await _service.authenticate(db, user_info)
7    except Exception as e:
8        logger.error("Google OAuth error: %s", e)
9        raise HTTPException(status_code=400, detail="oauth_failed")

1# app/auth/service.py
2class OAuthService:
3    async def authenticate(self, db, user_info: OAuthUserInfo) -> TokenResponse:
4        user, is_new = await get_or_create_oauth_user(
5            db,
6            provider=user_info.provider,
7            provider_id=user_info.provider_id,
8            name=user_info.name,
9            email=user_info.email,
10            profile_image=user_info.profile_image,
11        )
12        token = create_access_token(str(user.id))
13        hint = ProfileHint(
14            name=user_info.name, gender=user_info.gender,
15            birth_year=user_info.birth_year, birthday=user_info.birthday,
16        ) if is_new else None
17        return TokenResponse(access_token=token, is_new_user=is_new, profile_hint=hint)

1from fastapi import Depends
2from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
3 
4from .jwt import decode_access_token
5 
6_bearer = HTTPBearer()
7 
8 
9async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(_bearer)) -> str:
10    return decode_access_token(credentials.credentials)

1@router.get("/me", response_model=UserResponse)
2async def get_me(
3    user_id: str = Depends(get_current_user),
4    db: AsyncSession = Depends(get_db),
5) -> UserResponse:
6    ...

1def create_refresh_token(user_id: str) -> str:
2    expire = datetime.now(timezone.utc) + timedelta(days=14)
3    payload = {"sub": user_id, "exp": expire, "type": "refresh"}
4    return jwt.encode(payload, settings.JWT_REFRESH_SECRET_KEY, algorithm=ALGORITHM)
5 
6# 로그인 시: refresh_token을 DB(users 테이블 또는 별도 sessions 테이블)에 저장
7# /auth/refresh: DB에 저장된 값과 대조 후에만 새 access_token 발급
8# /auth/logout: DB에서 refresh_token 레코드를 삭제 → 이후 재발급 불가

1async def get_current_user_owned_journal(
2    journal_id: uuid.UUID,
3    user_id: str = Depends(get_current_user),
4    db: AsyncSession = Depends(get_db),
5):
6    journal = await get_journal_by_id(db, journal_id)
7    if journal.user_id != uuid.UUID(user_id):
8        raise HTTPException(status_code=403, detail="접근 권한이 없습니다.")
9    return journal