백엔드가 선녀였지... 지금은 CI/CD 파이프라인을 구성 중이지만, 며칠 전 백엔드 개발을 하면서 프로젝트 구조를 어떤 식으로 짰는지, 뭘 고민했는지 기록으로 남겨두려고 시간을 내본다... 후 크롤러에서 진을 다 빼서 백엔드 파트는 선녀같았다.


🗂️ 디렉토리 구조도

마이크로서비스 아키텍처를 기반으로 한 웹 애플리케이션 개발이 목표

그래서 백엔드 서비스를 기능별로 나누어서 개발을 진행했고, 나는 유저 관련 서비스/자기소개서 관련 서비스를 맡았다. 구조는 비슷하기 때문에 유저 서비스를 대표로 이야기를 할 것이다.

GAENCHWIS/
├── server/
│   ├── user_service/                         # 사용자 관련 마이크로서비스
│   │   ├── app/                             # 핵심 애플리케이션 코드
│   │   │   ├── core/                        # 핵심 기능 및 설정
│   │   │   │   ├── security/                # 인증/인가 관련
│   │   │   │   │   └── token_validator.py   # JWT 토큰 검증 
│   │   │   │   ├── aws_client.py           # AWS 서비스 클라이언트 
│   │   │   │   ├── config.py               # 환경설정 관리
│   │   │   │   ├── constants.py            # 상수 정의
│   │   │   │   └── enums.py                # 열거형 정의
│   │   │   ├── repositories/                # 데이터 접근 계층
│   │   │   │   └── user_repository.py      # DynamoDB CRUD 로직
│   │   │   ├── routes/                      # API 라우팅 
│   │   │   │   └── user_routes.py          # 사용자 관련 엔드포인트
│   │   │   └── schemas/                     # 데이터 모델
│   │   │       └── user_schema.py          # Pydantic 모델 정의
│   │   ├── main.py                          # FastAPI 앱 진입점
│   │   ├── Dockerfile                       # 컨테이너 정의
│   │   ├── requirements.txt                 # Python 의존성
│   │   ├── buildspec.yml                    # AWS CodeBuild 설정   
│   │   └── task-definition.json             # ECS 작업 정의

디렉토리 구조도는 위와 같다. AWS 기반의 마이크로서비스 아키텍처를 구현했다.

📂 계층형 아키텍처 (Layered Architecture)

프로젝트 구조를 할 때 정말 많은 고민이 있었다... 개발은 빠르게 해야 하긴 하겠고...
그렇다고 아무렇게나 만들면 나중에 유지보수하기가 어렵잖아... 기획이나 디자인이 틀은 잡았으나 계속 변동이 있어서 수정이 많이 들어갈 것을 고려해야 했다. 그리고 개발이 처음인 조원을 고려해서 너무 복잡하고 어렵게 나누면 안되었다. 이 중간을 고려해야 했다.

  • core/ - 핵심 기능 및 설정 관리
  • repositories/ - 데이터 접근 계층
  • routes/ - API 엔드포인트 정의
  • schemas/ - 데이터 모델 정의

이렇게 크게 네 부분으로 나누어서 설계했다. 가장 많이 고려한 것은 관심사를 분리하는 것이었다. 코드가 변경이 된다 하더라도 영향이 가는 범위를 최소화하고 싶었다. (유지보수를 쉽게...) 이렇게 하길 잘 했다 생각했던 이유는, DynamoDB의 스키마가 자주 바뀌곤 했는데 repository 계층만 수정하면 되었다. 또한 이럴 때 API 응답 형식도 바뀌곤 했는데 schemas 디렉토리만 수정하면 되었기 때문에 기획이 부족했던 우리 조의 프로젝트에 딱이었다 ㅎㅎ... 거창한 프로젝트도 아니고 이렇게까지 나눌 이유가 있나 싶기도 했는데 아무튼 괜찮은 것 같았음.

그리고 나는 극한의 효율충... 코드 중복이 제일 싫은 사람... 때문에 각 계층마다 책임을 명확하게 구분하려고 노력했고, 이를 통해서 중복 코드를 최소화하는 것이 목표였다.

⚡ FastAPI

학원에서 Flask와 FastAPI 둘 다 배우긴 했지만 Flask를 더 중점적으로(?) 배웠기 때문에 원래는 Flask를 사용할 생각이었다. 하지만... 우리는 API 문서를 만들 시간이 없었다. 흑흑... 생각보다 시간이 없었답니다... 개발 관련 규칙을 정해놨지만 그걸 지킬 수도 없이 정신없이 흘러가던 프로젝트~ API를 자동으로 문서화 해준다는 장점을 떠올리며 개발 직전 FastAPI로 급선회했다. 결국엔 잘 선택한 것 같음!

또한 Pydantic을 통해서 타입 검증도 쉽게 할 수 있었고, 타입 힌팅도 지원해서 개발이 편했다. 뭐... 알고 고른 건 아니었는데 암튼 현대적인 Python 웹 프레임워크라고 함. 써볼만 했음.

FastAPI를 채택하며 알게된 거지만 비동기 처리를 하기 때문에 성능이 좋다고 한다.

🦄 Uvicron 서버

ASGI(Asynchronous Server Gateway Interface) 서버 구현체

FastAPI를 사용하면서 사용해봤다. 비동기 HTTP 요청을 처리하는데 사용되고, 자동 리로딩이 가능하다. DynamoDB 작업의 비동기 처리를 하기 위해서는 Uvicorn을 사용하는 것이 적합했던 것 같다. 뭔가... 유니콘 같음...

import os
from pathlib import Path
from dotenv import load_dotenv
from fastapi import FastAPI, Request, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse

env_path = Path(__file__).parent.parent / '.env'
load_dotenv(dotenv_path=env_path)

app = FastAPI(
    title="User Service",
    description="유저 관리 서비스"
)

app.include_router(router, prefix="/api/v1", tags=["user"])

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("main:app",  host="0.0.0.0", port=8000, reload=True, log_level="info")

기본적으로 이렇게 하면 서버가 구축된다! 서버 실행도 매우 편하다.

uvicorn main:app --port 8000 --reload

이 명령어면 바로 열림... 개굿. port 옵션은 내가 원하는 포트를 열 수 있어서 뒤에 8003을 하면 8003번 포트로 열린다. reload 옵션을 주면 수정 사항이 바로바로 반영되어 리로드 된다.

후... 벨로그엔 좀 지우고 올린거지만, 환경 변수를 갑자기 자꾸 못 가져와서... 로그 찍어둔 부분 개많음... 로그 언제 지우냐...

🔐 보안 및 인증

🔒 AWS Cognito 통합

우리 서비스는 AWS Cognito를 사용해서 SNS 로그인을 간편하게 할 수 있게 했다! 와... 세상 참 좋아졌다... 예전에 구글 로그인, 카카오 로그인, 네이버 로그인 개발문서 뒤져가면서 힘들게 공부하고 낑낑대며 구현했던 게 새록새록 떠올랐는데... 그걸 얘가 알아서 다 해준다고? 너무 좋았다... 하지만 그래도 인증이 필요하기 때문에! 개발 말미에 인증 처리 부분을 만들었다.

core/ 계층에 보안을 담당하는 security/ 계층을 만들어서 글로벌 인스턴스를 만들었다.

AWS Cognito를 사용한 로그인 인증 과정

  1. 초기 로그인 요청
사용자 → 앱/웹 프론트엔드 → Cognito
- username/password 입력
- Cognito의 인증 엔드포인트로 전송
  1. 인증 성공 시 토큰 발급
Cognito → 프론트엔드
- ID 토큰 (사용자 정보 포함)
- Access 토큰 (API 접근용)
- Refresh 토큰 (토큰 갱신용)
  1. API 요청 시 토큰 검증
1. 프론트엔드: Authorization 헤더에 Access 토큰 포함
2. 백엔드: TokenValidator가 토큰 검증
   - JWKS에서 public key 조회
   - 토큰 서명 검증
   - 만료 시간 확인
   - audience 확인
  1. 토큰 갱신
Access 토큰 만료 시:
1. Refresh 토큰으로 Cognito 토큰 엔드포인트 호출
2. 새로운 Access 토큰과 ID 토큰 발급

코그니토를 사용한 로그인 인증 과정은 이렇다. 처음엔 다 해준다기에 백엔드가 필요 없는 줄 알았지... 하지만 역할이 필요하긴 했다! (당연함. 사용자 인증은 코드에서 하셔야지요.) 받은 토큰을 검증해야 하는 과정을 구현해야 했다.

토큰 검증 코드

토큰 검증 순서는 아래와 같다.

  1. 토큰 헤더에서 kid(Key ID) 추출
  2. kid로 해당하는 public key 조회
  3. JWT 토큰 디코딩 및 검증
  4. audience 검증 (APP_CLIENT_ID와 일치 확인)
from fastapi import HTTPException
from typing import Dict
import jwt
import jwt.algorithms
from jwt.exceptions import ExpiredSignatureError, InvalidTokenError
import requests
import json
import os
from dotenv import load_dotenv
import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

load_dotenv()

REGION = os.getenv('AWS_REGION')
USER_POOL_ID = os.getenv('COGNITO_USER_POOL_ID')
APP_CLIENT_ID = os.getenv('COGNITO_APP_CLIENT_ID')

class TokenValidator:
    _instance = None
    jwks = None
    jwks_url = None
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(TokenValidator, cls).__new__(cls)
            cls._instance.__init__()
        return cls._instance

    def __init__(self):
        if self.jwks_url is None:
            self.jwks_url = f"https://cognito-idp.{REGION}.amazonaws.com/{USER_POOL_ID}/.well-known/jwks.json"
            self._load_jwks()

    def _load_jwks(self):
        try:
            if not self.jwks:
                response = requests.get(self.jwks_url)
                self.jwks = response.json()
        except Exception as e:
            logger.error(f"Failed to load JWKS: {str(e)}")
            raise HTTPException(
                status_code=500,
                detail=f"Failed to load JWKS: {str(e)}"
            )
    
    def get_public_key(self, kid):
        try:
            if not self.jwks:
                self._load_jwks()
                
            key = next((k for k in self.jwks['keys'] if k['kid'] == kid), None)
            if not key:
                logger.error(f"Key ID not found: {kid}")
                raise HTTPException(
                    status_code=401, 
                    detail="Invalid token: Key ID not found"
                )
            return key
        except Exception as e:
            logger.error(f"Error getting public key: {str(e)}")
            raise HTTPException(
                status_code=401,
                detail=f"Error getting public key: {str(e)}"
            )
    
    def decode_and_validate_token(self, token: str) -> Dict:
        try:
            headers = jwt.get_unverified_header(token)
            kid = headers.get('kid')
            
            if not kid:
                logger.error("No kid found in token header")
                raise HTTPException(status_code=401, detail="No kid found in token")
                
            public_key = self.get_public_key(kid)
            
            decoded_token = jwt.decode(
                token,
                key=jwt.algorithms.RSAAlgorithm.from_jwk(json.dumps(public_key)),
                algorithms=['RS256'],
                audience=APP_CLIENT_ID
            )
            
            return decoded_token
                
        except jwt.exceptions.InvalidTokenError as e:
            logger.error(f"Invalid token error: {str(e)}")
            raise HTTPException(status_code=401, detail=f"Invalid token: {str(e)}")
        except Exception as e:
            logger.error(f"Unexpected error: {str(e)}")
            raise HTTPException(status_code=401, detail=str(e))

    def verify_access_token(self, access_token: str) -> bool:
        try:
            self.decode_and_validate_token(access_token)
            return True
        except Exception:
            return False
            
    def verify_id_token(self, id_token: str) -> Dict:
        return self.decode_and_validate_token(id_token)

token_validator = TokenValidator()

싱글톤 패턴을 구현해 단일 인스턴스만 생성하게 했고, JWKS 정보를 메모리에 한 번만 로드해서 재사용하게 했다. (_load_jwks 메서드를 통해 JWKS를 초기에 로드하고 캐시한다.)

어차피 JWKS는 자주 변경되는 정보는 아니다. 그렇기 때문에 여러 번 로드할 필요가 없이 한 번만 메모리에 저장하게 한다. 또 매 요청마다 JWKS를 새로 로드하게 되면 불필요하게 HTTP 요청이 발생하게 된다. 캐시된 JWKS를 재사용하면 검증 속도를 높일 수 있다.

검증 메서드

  1. verify_access_token → 액세스 토큰 검증 (성공/실패 여부만 반환)
  2. verify_id_token → ID 토큰 검증 (사용자 정보 반환)

에러 처리

  • 토큰 무효
  • kid 누락
  • JWKS 로드 실패
  • 예상치 못한 오류

🪪 JWT 기반 토큰 검증

위에서 우리는 JWKS를 언급했다! 나도 이번에 처음 알게 됨... 이것은 JWT 기반 인증 시스템에서 일반적으로 사용되는 표준이라고 한다. AWS Cognito 역시 JWKS를 사용한다.

JWKS(JSON Web Key Set)는 JWT 토큰 검증에서 사용되는 공개키들의 집합

{
  "keys": [
    {
      "kid": "키 식별자",
      "kty": "키 타입(RSA 등)",
      "n": "모듈러스",
      "e": "지수",
      "alg": "알고리즘(RS256 등)"
    }
  ]
}

Cognito가 private key로 토큰 서명을 하면 → 클라이언트는 JWKS URL에서 public key 목록을 조회하고 → 토큰의 kid와 일치하는 public key로 서명을 검증한다.


느낀점

뭐... 이런저런 고민들을 하며 설계를 했었다. 최대한 개발에 익숙하지 않은 사람에게도 쉽게 받아들여질 수 있게... 노력해서 만들긴 했는데 그 노력이 통했는지 모르겠다. 후 그래도 처음으로 주도적으로 백엔드 개발을 해본 거라 신기하기도 했고 재밌었다. 그리고 또... AI가 참... 눈부신 발전을 이루었구나 ^^ AI만 있다고 만사 장땡은 아니지만 정말 정말 빠른 속도로 개발을 끝낼 수 있었다. 물론 수정할 부분이 많긴 하지만... 약간 현타가 왔달까~ 그래도 재밌었다. 날... 날 대체하면 안될텐데... 암튼 프로젝트 재밌네~


본 포스팅은 글로벌소프트웨어캠퍼스와 교보DTS가 함께 진행하는 챌린지입니다.

profile
영차영차 😎

0개의 댓글

관련 채용 정보