o2o-castad-backend/access_plan.md

23 KiB

Access Token 인증 설계 문서

작성일: 2026-01-28 목적: 모든 API 요청에서 액세스 토큰을 검증하여 로그인 사용자를 인증하는 최적의 설계 제안


1. 현재 시스템 분석

1.1 기존 인증 구조

인증 의존성 (app/user/dependencies/auth.py)

현재 3가지 인증 의존성이 구현되어 있습니다:

의존성 용도 토큰 없음 시
get_current_user() 필수 인증 예외 발생 (401)
get_current_user_optional() 선택적 인증 None 반환
get_current_admin() 관리자 전용 예외 발생 (403)

JWT 토큰 구조 (app/user/services/jwt.py)

# Access Token Payload
{
    "sub": user_uuid,      # 사용자 고유 식별자 (UUID7)
    "exp": datetime,        # 만료 시간
    "type": "access"        # 토큰 타입
}
  • Access Token 유효기간: JWT_ACCESS_TOKEN_EXPIRE_MINUTES (설정 파일)
  • Refresh Token 유효기간: JWT_REFRESH_TOKEN_EXPIRE_DAYS (설정 파일)
  • 알고리즘: HS256

커스텀 예외 (app/user/exceptions.py)

예외 HTTP 상태 코드
MissingTokenError 401 MISSING_TOKEN
InvalidTokenError 401 INVALID_TOKEN
TokenExpiredError 401 TOKEN_EXPIRED
TokenRevokedError 401 TOKEN_REVOKED
UserNotFoundError 404 USER_NOT_FOUND
UserInactiveError 403 USER_INACTIVE
AdminRequiredError 403 ADMIN_REQUIRED

1.2 현재 엔드포인트 인증 현황

인증이 적용된 엔드포인트 (3개)

엔드포인트 메서드 의존성
/auth/me GET get_current_user
/auth/logout POST get_current_user
/auth/logout/all POST get_current_user

인증이 없는 엔드포인트 (13개)

모듈 엔드포인트 메서드 설명
Home /crawling POST 네이버 지도 크롤링
Home /autocomplete POST 자동완성 크롤링
Home /image/upload/server POST 이미지 업로드 (로컬)
Home /image/upload/blob POST 이미지 업로드 (Azure Blob)
Lyric /lyric/generate POST 가사 생성
Lyric /lyric/status/{task_id} GET 가사 상태 조회
Lyric /lyrics/ GET 가사 목록 조회
Lyric /lyric/{task_id} GET 가사 상세 조회
Song /song/generate/{task_id} POST 노래 생성
Song /song/status/{song_id} GET 노래 상태 조회
Video /video/generate/{task_id} GET 영상 생성
Video /video/status/{creatomate_render_id} GET 영상 상태 조회
Video /video/download/{task_id} GET 영상 다운로드
Video /videos/ GET 영상 목록 조회
Archive /archive/videos/ GET 완료된 영상 목록 조회 (아카이브)
Archive /archive/videos/{task_id} DELETE 아카이브 영상 삭제 (CASCADE)

1.3 모델의 user_uuid 외래키 현황

user_uuid 외래키는 Project 테이블에만 존재합니다. 하위 리소스(Lyric, Song, Video, Image)의 소유권은 Project를 통해 간접적으로 확인합니다.

모델 user_uuid 필드 nullable 비고
Project user_uuid → User.user_uuid True 소유권 기준
Image 없음 - task_id로 Project 연결
Lyric 없음 - project_id로 Project 연결
Song 없음 - project_id로 Project 연결
Video 없음 - project_id로 Project 연결

소유권 확인 흐름:

Lyric/Song/Video → project_id → Project → user_uuid → User
Image → task_id → Project (같은 task_id) → user_uuid → User

2. 설계 방안 비교

2.1 방안 A: 의존성 주입 방식 (Dependency Injection)

각 엔드포인트에 개별적으로 인증 의존성을 추가하는 방식입니다.

# 예시: 필수 인증
@router.post("/lyric/generate")
async def generate_lyric(
    request_body: GenerateLyricRequest,
    current_user: User = Depends(get_current_user),  # 인증 추가
    session: AsyncSession = Depends(get_session),
):
    # current_user.user_uuid 사용 가능
    ...

# 예시: 선택적 인증
@router.get("/lyrics/")
async def list_lyrics(
    current_user: User | None = Depends(get_current_user_optional),  # 선택적 인증
    session: AsyncSession = Depends(get_session),
):
    if current_user:
        # 로그인 사용자: 자신의 가사만 조회
        ...
    else:
        # 비로그인 사용자: 공개 가사 조회
        ...

장점:

  • FastAPI 표준 패턴 (공식 문서 권장)
  • 엔드포인트별 세밀한 제어 가능
  • 테스트 작성이 용이 (의존성 오버라이드)
  • 기존 코드와의 호환성 우수

단점:

  • 각 엔드포인트에 개별 적용 필요
  • 실수로 인증 누락 가능

2.2 방안 B: 미들웨어 방식 (Middleware)

모든 요청에 미들웨어가 자동으로 토큰을 검증하는 방식입니다.

# app/middleware/auth.py
class AuthMiddleware(BaseHTTPMiddleware):
    # 인증 예외 경로
    EXEMPT_PATHS = [
        "/docs", "/openapi.json", "/health",
        "/auth/kakao/login", "/auth/kakao/callback", "/auth/kakao/verify",
        "/auth/refresh",
    ]

    async def dispatch(self, request: Request, call_next):
        # 예외 경로는 통과
        if any(request.url.path.startswith(p) for p in self.EXEMPT_PATHS):
            return await call_next(request)

        # 토큰 검증
        auth_header = request.headers.get("Authorization")
        if not auth_header or not auth_header.startswith("Bearer "):
            return JSONResponse(status_code=401, content={"code": "MISSING_TOKEN"})

        token = auth_header.split(" ")[1]
        payload = decode_token(token)
        if not payload:
            return JSONResponse(status_code=401, content={"code": "INVALID_TOKEN"})

        # request.state에 사용자 정보 저장
        request.state.user_uuid = payload.get("sub")
        return await call_next(request)

장점:

  • 모든 엔드포인트에 자동 적용
  • 중앙 집중식 관리
  • 인증 누락 방지

단점:

  • 예외 경로 관리가 복잡해질 수 있음
  • DB 조회를 미들웨어에서 처리하면 성능 이슈
  • 선택적 인증 구현이 어려움
  • FastAPI의 의존성 주입 패턴과 맞지 않음

2.3 방안 C: 라우터 레벨 의존성 (Router-Level Dependencies) 권장

라우터 전체에 기본 인증을 적용하고, 개별 엔드포인트에서 오버라이드하는 방식입니다.

# 인증이 필요한 라우터
lyric_router = APIRouter(
    prefix="/lyric",
    tags=["Lyric"],
    dependencies=[Depends(get_current_user_optional)],  # 라우터 전체에 적용
)

# 필수 인증이 필요한 엔드포인트
@lyric_router.post(
    "/generate",
    dependencies=[Depends(get_current_user)],  # 오버라이드: 필수 인증
)
async def generate_lyric(...):
    ...

# 선택적 인증 (라우터 기본값 사용)
@lyric_router.get("/lyrics/")
async def list_lyrics(...):
    ...

장점:

  • 모듈별 일관된 인증 정책 적용
  • 엔드포인트별 세밀한 제어 가능
  • FastAPI 표준 패턴 준수
  • 인증 누락 가능성 감소

단점:

  • 라우터 수정 필요
  • 새 엔드포인트 추가 시 주의 필요

3. 권장 설계안

3.1 최적 설계: 의존성 주입 방식 (방안 A) + 라우터 레벨 기본값 (방안 C)

핵심 원칙

  1. 데이터 생성 엔드포인트: 필수 인증 (get_current_user)
  2. 데이터 조회 엔드포인트: 선택적 인증 (get_current_user_optional)
  3. 공개 엔드포인트: 인증 없음 (로그인, 콜백 등)

엔드포인트별 인증 정책

엔드포인트 인증 타입 이유
/auth/kakao/login 없음 로그인 진입점
/auth/kakao/callback 없음 OAuth 콜백
/auth/kakao/verify 없음 토큰 발급
/auth/refresh 없음 토큰 갱신
/auth/me 필수 내 정보 조회
/auth/logout 필수 로그아웃
/auth/logout/all 필수 전체 로그아웃
/crawling 선택적 비로그인도 테스트 가능
/autocomplete 선택적 비로그인도 테스트 가능
/image/upload/blob 필수 리소스 생성
/lyric/generate 필수 리소스 생성
/lyric/status/{task_id} 선택적 상태 조회
/lyric/{task_id} 선택적 상세 조회
/lyrics/ 선택적 목록 조회
/song/generate/{task_id} 필수 리소스 생성
/song/status/{song_id} 선택적 상태 조회
/video/generate/{task_id} 필수 리소스 생성
/video/status/{...} 선택적 상태 조회
/video/download/{task_id} 선택적 다운로드
/videos/ 선택적 목록 조회
/archive/videos/ 필수 완료된 영상 목록 조회 (아카이브)
/archive/videos/{task_id} 필수 아카이브 영상 삭제 + 소유권 검증

3.2 구현 코드 예시

3.2.1 리소스 생성 엔드포인트 (필수 인증)

# app/lyric/api/routers/v1/lyric.py

from app.user.dependencies import get_current_user
from app.user.models import User

@router.post("/generate")
async def generate_lyric(
    request_body: GenerateLyricRequest,
    background_tasks: BackgroundTasks,
    current_user: User = Depends(get_current_user),  # ✅ 필수 인증
    session: AsyncSession = Depends(get_session),
) -> GenerateLyricResponse:
    """고객 정보를 기반으로 가사를 생성합니다. (백그라운드 처리)"""

    # Project 생성 시 user_uuid 연결 (소유권의 기준점)
    project = Project(
        store_name=request_body.customer_name,
        region=request_body.region,
        task_id=task_id,
        user_uuid=current_user.user_uuid,  # ✅ 사용자 연결
        ...
    )

    # Lyric은 project_id를 통해 소유권 확인 (user_uuid 없음)
    lyric = Lyric(
        project_id=project.id,  # ✅ Project 연결 → 소유권 간접 확인
        task_id=task_id,
        ...
    )

3.2.2 데이터 조회 엔드포인트 (선택적 인증)

# app/lyric/api/routers/v1/lyric.py

from app.user.dependencies import get_current_user_optional
from app.user.models import User

@router.get("/lyrics/")
async def list_lyrics(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    current_user: User | None = Depends(get_current_user_optional),  # ✅ 선택적 인증
    session: AsyncSession = Depends(get_session),
) -> PaginatedResponse[LyricListItem]:
    """페이지네이션으로 완료된 가사 목록을 조회합니다."""

    base_query = select(Lyric).where(Lyric.status == "completed")

    # 로그인 사용자: Project.user_uuid를 통해 자신의 가사만 조회
    if current_user:
        base_query = (
            base_query
            .join(Project, Lyric.project_id == Project.id)
            .where(Project.user_uuid == current_user.user_uuid)  # ✅ Project를 통한 소유자 필터
        )
    else:
        # 비로그인 사용자: 공개 데이터만 조회 (Project.user_uuid가 NULL)
        base_query = (
            base_query
            .join(Project, Lyric.project_id == Project.id)
            .where(Project.user_uuid.is_(None))
        )

    # 페이지네이션 처리
    ...

3.2.3 이미지 업로드 (필수 인증 + user_uuid 폴더 구조)

# app/home/api/routers/v1/home.py

from app.user.dependencies import get_current_user
from app.user.models import User
from app.utils.upload_blob_as_request import AzureBlobUploader

@router.post("/image/upload/blob")
async def upload_images_blob(
    images_json: Optional[str] = Form(default=None),
    files: Optional[list[UploadFile]] = File(default=None),
    current_user: User = Depends(get_current_user),  # ✅ 필수 인증
) -> ImageUploadResponse:
    """이미지 업로드 (URL + Azure Blob Storage)"""

    task_id = await generate_task_id()

    # ✅ user_uuid를 포함한 Blob 업로더 생성
    uploader = AzureBlobUploader(
        user_uuid=current_user.user_uuid,
        task_id=task_id,
    )
    # Blob 경로: {BASE_URL}/{user_uuid}/{task_id}/image/{file_name}

    # Image 저장 (user_uuid 없음 - task_id로 Project와 연결)
    image = Image(
        task_id=task_id,  # ✅ task_id를 통해 Project와 연결 → 소유권 확인
        img_name=img_name,
        img_url=blob_url,
        ...
    )

Note: Image 테이블에는 user_uuid 필드가 없습니다. 소유권은 task_id를 공유하는 Project를 통해 확인합니다.


3.3 소유권 검증 유틸리티

데이터 조회/수정 시 Project를 통한 소유권 검증을 위한 유틸리티 함수를 추가합니다.

# app/utils/ownership.py

from fastapi import HTTPException, status
from typing import Optional
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.user.models import User
from app.home.models import Project


def verify_ownership(
    project_user_uuid: Optional[str],
    current_user: Optional[User],
    raise_on_mismatch: bool = True,
) -> bool:
    """
    Project 기반 리소스 소유권 검증

    Args:
        project_user_uuid: Project의 user_uuid
        current_user: 현재 로그인 사용자 (None이면 비로그인)
        raise_on_mismatch: True면 불일치 시 예외 발생

    Returns:
        bool: 소유권 일치 여부

    Raises:
        HTTPException: 소유권 불일치 시 (raise_on_mismatch=True)
    """
    # Project에 소유자가 없으면 모두 접근 가능 (공개 리소스)
    if project_user_uuid is None:
        return True

    # 비로그인 사용자가 소유자가 있는 리소스에 접근
    if current_user is None:
        if raise_on_mismatch:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail={"code": "AUTH_REQUIRED", "message": "로그인이 필요합니다."},
            )
        return False

    # 소유자 불일치
    if project_user_uuid != current_user.user_uuid:
        if raise_on_mismatch:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail={"code": "FORBIDDEN", "message": "접근 권한이 없습니다."},
            )
        return False

    return True


async def get_project_owner_uuid(
    session: AsyncSession,
    project_id: int,
) -> Optional[str]:
    """project_id로 소유자의 user_uuid 조회"""
    result = await session.execute(
        select(Project.user_uuid).where(Project.id == project_id)
    )
    return result.scalar_one_or_none()

사용 예시:

@router.get("/{task_id}")
async def get_lyric_detail(
    task_id: str,
    current_user: User | None = Depends(get_current_user_optional),
    session: AsyncSession = Depends(get_session),
) -> LyricDetailResponse:
    """task_id로 생성된 가사를 조회합니다."""

    lyric = await get_lyric_by_task_id(session, task_id)

    # ✅ Project를 통한 소유권 검증
    project_user_uuid = await get_project_owner_uuid(session, lyric.project_id)
    verify_ownership(project_user_uuid, current_user)

    return LyricDetailResponse(...)

4. 구현 체크리스트

4.1 Phase 1: 인증 인프라 확장

  • app/utils/ownership.py 생성 (소유권 검증 유틸리티)
  • AzureBlobUploader 클래스에 user_uuid 파라미터 활성화
  • 기존 get_current_user, get_current_user_optional 테스트

4.2 Phase 2: 리소스 생성 엔드포인트 인증 적용

  • POST /image/upload/blob - 필수 인증 (Image에는 user_uuid 없음, task_id로 연결)
  • POST /lyric/generate - 필수 인증 + Project.user_uuid 저장
  • POST /song/generate/{task_id} - 필수 인증 (Project 통해 소유권 확인)
  • GET /video/generate/{task_id} - 필수 인증 (Project 통해 소유권 확인)

4.3 Phase 3: 조회 엔드포인트 인증 적용

  • GET /lyric/status/{task_id} - 선택적 인증 + Project 통해 소유권 검증
  • GET /lyric/{task_id} - 선택적 인증 + Project 통해 소유권 검증
  • GET /lyrics/ - 선택적 인증 + Project.user_uuid 기반 필터
  • GET /song/status/{song_id} - 선택적 인증 + Project 통해 소유권 검증
  • GET /video/status/{...} - 선택적 인증 + Project 통해 소유권 검증
  • GET /video/download/{task_id} - 선택적 인증 + Project 통해 소유권 검증
  • GET /videos/ - 선택적 인증 + Project.user_uuid 기반 필터
  • GET /archive/videos/ - 필수 인증 + Project.user_uuid 기반 필터
  • DELETE /archive/videos/{task_id} - 필수 인증 + 소유권 검증 + CASCADE 삭제

4.4 Phase 4: 크롤링 엔드포인트

  • POST /crawling - 선택적 인증 (비로그인도 허용)
  • POST /autocomplete - 선택적 인증 (비로그인도 허용)

4.5 Phase 5: OpenAPI 문서 업데이트

  • main.pycustom_openapi() 수정하여 인증이 필요한 엔드포인트에 security 표시

5. OpenAPI Security 스키마 업데이트

main.pycustom_openapi() 함수를 수정하여 인증이 필요한 엔드포인트를 표시합니다.

def custom_openapi():
    """커스텀 OpenAPI 스키마 생성 (Bearer 인증 추가)"""
    if app.openapi_schema:
        return app.openapi_schema

    openapi_schema = get_openapi(
        title=app.title,
        version=app.version,
        description=app.description,
        routes=app.routes,
        tags=tags_metadata,
    )

    # Bearer 토큰 인증 스키마 추가
    openapi_schema["components"]["securitySchemes"] = {
        "BearerAuth": {
            "type": "http",
            "scheme": "bearer",
            "bearerFormat": "JWT",
            "description": "JWT 액세스 토큰을 입력하세요.",
        }
    }

    # 인증이 필요한 경로 패턴
    AUTH_REQUIRED_PATHS = [
        "/auth/me", "/auth/logout",
        "/image/upload/blob",
        "/lyric/generate",
        "/song/generate",
        "/video/generate",
        "/archive/videos",  # GET (목록조회), DELETE (삭제)
    ]

    AUTH_OPTIONAL_PATHS = [
        "/crawling", "/autocomplete",
        "/lyric/status", "/lyric/",
        "/lyrics/",
        "/song/status",
        "/video/status", "/video/download",
        "/videos/",
    ]

    for path, path_item in openapi_schema["paths"].items():
        for method, operation in path_item.items():
            if method in ["get", "post", "put", "patch", "delete"]:
                # 필수 인증
                if any(auth_path in path for auth_path in AUTH_REQUIRED_PATHS):
                    operation["security"] = [{"BearerAuth": []}]
                # 선택적 인증 (문서에는 표시하지만 필수 아님)
                elif any(auth_path in path for auth_path in AUTH_OPTIONAL_PATHS):
                    operation["security"] = [{"BearerAuth": []}, {}]  # 빈 객체 = 인증 없이도 가능

    app.openapi_schema = openapi_schema
    return app.openapi_schema

6. 테스트 전략

6.1 단위 테스트

# tests/test_auth_dependencies.py

import pytest
from fastapi import HTTPException
from app.user.dependencies import get_current_user, get_current_user_optional


@pytest.mark.asyncio
async def test_get_current_user_missing_token():
    """토큰 없이 접근 시 MissingTokenError 발생"""
    with pytest.raises(HTTPException) as exc_info:
        await get_current_user(credentials=None, session=mock_session)
    assert exc_info.value.status_code == 401
    assert exc_info.value.detail["code"] == "MISSING_TOKEN"


@pytest.mark.asyncio
async def test_get_current_user_optional_missing_token():
    """선택적 인증에서 토큰 없으면 None 반환"""
    result = await get_current_user_optional(credentials=None, session=mock_session)
    assert result is None

6.2 통합 테스트

# tests/test_lyric_auth.py

import pytest
from httpx import AsyncClient


@pytest.mark.asyncio
async def test_generate_lyric_without_auth(client: AsyncClient):
    """인증 없이 가사 생성 시 401 반환"""
    response = await client.post("/lyric/generate", json={...})
    assert response.status_code == 401
    assert response.json()["detail"]["code"] == "MISSING_TOKEN"


@pytest.mark.asyncio
async def test_generate_lyric_with_auth(client: AsyncClient, auth_headers: dict):
    """인증된 사용자의 가사 생성 성공"""
    response = await client.post(
        "/lyric/generate",
        json={...},
        headers=auth_headers,
    )
    assert response.status_code == 200

7. 마이그레이션 고려사항

7.1 기존 데이터 처리

현재 Project 테이블에서 user_uuidNULL인 레코드들에 대한 처리 방안:

  1. 기존 데이터 유지: Project.user_uuidNULL이면 해당 Project 및 하위 리소스(Lyric, Song, Video, Image)를 공개 데이터로 취급
  2. 관리자 전용: Project.user_uuidNULL인 데이터는 관리자만 접근 가능
  3. 마이그레이션: 특정 사용자에게 기존 Project 할당 (비권장)

권장: 옵션 1 (기존 데이터는 공개 데이터로 유지)

Note: user_uuid는 Project 테이블에만 존재합니다. Lyric, Song, Video, Image 테이블에는 user_uuid 필드가 없으며, 소유권은 Project를 통해 간접적으로 확인합니다.

7.2 하위 호환성

  • 기존 API 클라이언트가 인증 없이 호출하는 경우 401 응답
  • 프론트엔드 업데이트 필요: 모든 API 호출에 Authorization 헤더 추가
  • 점진적 적용: Phase별로 적용하여 영향 범위 최소화

8. 결론

권장 구현 순서

  1. 의존성 주입 방식 채택 (FastAPI 표준 패턴)
  2. 필수/선택적 인증 엔드포인트별 적용
  3. Project 기반 소유권 검증 유틸리티 활용
  4. Project.user_uuid 연결 리소스 생성 시 적용

데이터 모델 구조

User (user_uuid)
  └── Project (user_uuid → User)  ← 소유권의 기준점
        ├── Lyric (project_id → Project)
        │     └── Song (lyric_id → Lyric)
        │           └── Video (song_id → Song)
        └── Image (task_id = Project.task_id)

이 설계를 통해:

  • Project 테이블만 user_uuid를 가지며, 모든 소유권 확인의 기준점 역할
  • 하위 리소스(Lyric, Song, Video, Image)는 Project를 통해 간접적으로 소유권 확인
  • 사용자별 데이터 격리가 가능합니다
  • 비로그인 사용자도 제한된 기능을 사용할 수 있습니다
  • Azure Blob Storage 폴더 구조에 user_uuid가 포함됩니다