feat(backend): Wave 2a — JWT authentication system

Author: hyeonsang

.gitignore

@@ -149,3 +149,4 @@ local.properties
 .recommenders/
 
 .github/agents/ai-strategy-consultant.md
+.history/

backend/app/config.py

@@ -1,5 +1,8 @@
 """Application configuration via pydantic-settings."""
 
+import secrets
+
+from pydantic import model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -13,6 +16,13 @@ class Settings(BaseSettings):
     MY_PW: str = "admin"
     SECRET_KEY: str = ""
 
+    @model_validator(mode="after")
+    def _generate_secret_key(self) -> "Settings":
+        """Auto-generate SECRET_KEY if not provided."""
+        if not self.SECRET_KEY:
+            self.SECRET_KEY = secrets.token_urlsafe(32)
+        return self
+
     # 서버
     APP_PORT: int = 8000
     APP_HOST: str = "0.0.0.0"

backend/app/dependencies.py

@@ -0,0 +1,36 @@
+"""Shared FastAPI dependencies."""
+
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from jose import JWTError
+
+from app.services.auth_service import decode_token
+
+security = HTTPBearer()
+
+
+async def get_current_user(
+    credentials: HTTPAuthorizationCredentials = Depends(security),
+) -> str:
+    """Validate JWT access token and return the username."""
+    token = credentials.credentials
+    try:
+        payload = decode_token(token)
+        if payload.get("type") != "access":
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Invalid token type",
+            )
+        username: str = payload.get("sub")
+        if username is None:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Invalid token payload",
+            )
+        return username
+    except JWTError:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid or expired token",
+            headers={"WWW-Authenticate": "Bearer"},
+        )

backend/app/main.py

@@ -9,7 +9,7 @@
 from app.config import settings
 from app.database import init_db
 from app.models.download import Download  # noqa: F401 — Base.metadata 등록용
-from app.routers import health
+from app.routers import auth, health
 
 
 @asynccontextmanager
@@ -34,3 +34,4 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
 )
 
 app.include_router(health.router)
+app.include_router(auth.router)

backend/app/routers/auth.py

@@ -0,0 +1,50 @@
+"""Authentication endpoints."""
+
+from fastapi import APIRouter, HTTPException, status
+from jose import JWTError
+
+from app.schemas.auth import LoginRequest, RefreshRequest, TokenResponse
+from app.services.auth_service import (
+    create_access_token,
+    create_refresh_token,
+    decode_token,
+    verify_credentials,
+)
+
+router = APIRouter(prefix="/api/auth", tags=["auth"])
+
+
+@router.post("/login", response_model=TokenResponse)
+async def login(request: LoginRequest) -> TokenResponse:
+    """Authenticate with ID/PW and receive JWT tokens."""
+    if not verify_credentials(request.id, request.pw):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid ID or password",
+        )
+    return TokenResponse(
+        access_token=create_access_token(request.id),
+        refresh_token=create_refresh_token(request.id),
+    )
+
+
+@router.post("/refresh", response_model=TokenResponse)
+async def refresh(request: RefreshRequest) -> TokenResponse:
+    """Exchange a refresh token for a new access token."""
+    try:
+        payload = decode_token(request.refresh_token)
+        if payload.get("type") != "refresh":
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Invalid token type",
+            )
+        username = payload.get("sub")
+        return TokenResponse(
+            access_token=create_access_token(username),
+            refresh_token=create_refresh_token(username),
+        )
+    except JWTError:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid or expired refresh token",
+        )

backend/app/schemas/auth.py

@@ -0,0 +1,24 @@
+"""Authentication request/response schemas."""
+
+from pydantic import BaseModel
+
+
+class LoginRequest(BaseModel):
+    """Login request body."""
+
+    id: str
+    pw: str
+
+
+class TokenResponse(BaseModel):
+    """JWT token response."""
+
+    access_token: str
+    refresh_token: str
+    token_type: str = "bearer"
+
+
+class RefreshRequest(BaseModel):
+    """Token refresh request body."""
+
+    refresh_token: str

backend/app/services/auth_service.py

@@ -0,0 +1,45 @@
+"""JWT token creation and verification."""
+
+from datetime import datetime, timedelta, timezone
+
+from jose import jwt
+
+from app.config import settings
+
+ACCESS_TOKEN_EXPIRE_MINUTES = 30
+REFRESH_TOKEN_EXPIRE_DAYS = 7
+ALGORITHM = "HS256"
+
+
+def verify_credentials(user_id: str, password: str) -> bool:
+    """Check credentials against environment variables."""
+    return user_id == settings.MY_ID and password == settings.MY_PW
+
+
+def create_access_token(subject: str) -> str:
+    """Create a JWT access token."""
+    now = datetime.now(timezone.utc)
+    payload = {
+        "sub": subject,
+        "type": "access",
+        "exp": now + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
+        "iat": now,
+    }
+    return jwt.encode(payload, settings.SECRET_KEY, algorithm=ALGORITHM)
+
+
+def create_refresh_token(subject: str) -> str:
+    """Create a JWT refresh token."""
+    now = datetime.now(timezone.utc)
+    payload = {
+        "sub": subject,
+        "type": "refresh",
+        "exp": now + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS),
+        "iat": now,
+    }
+    return jwt.encode(payload, settings.SECRET_KEY, algorithm=ALGORITHM)
+
+
+def decode_token(token: str) -> dict:
+    """Decode and verify a JWT token. Raises JWTError on failure."""
+    return jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])

backend/tests/conftest.py

@@ -0,0 +1,11 @@
+import pytest
+
+from app.config import settings
+
+
+@pytest.fixture(autouse=True)
+def override_settings(monkeypatch):
+    """Ensure tests always use fixed credentials regardless of env vars."""
+    monkeypatch.setattr(settings, "MY_ID", "admin")
+    monkeypatch.setattr(settings, "MY_PW", "admin")
+    monkeypatch.setattr(settings, "SECRET_KEY", "test-secret-key-for-testing")

backend/tests/test_auth.py

@@ -0,0 +1,94 @@
+"""Tests for authentication endpoints."""
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+
+from app.main import app
+
+
+@pytest.mark.asyncio
+async def test_login_success() -> None:
+    """Valid credentials should return access + refresh tokens."""
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.post(
+            "/api/auth/login",
+            json={"id": "admin", "pw": "admin"},
+        )
+    assert response.status_code == 200
+    data = response.json()
+    assert "access_token" in data
+    assert "refresh_token" in data
+    assert data["token_type"] == "bearer"
+
+
+@pytest.mark.asyncio
+async def test_login_invalid_password() -> None:
+    """Wrong password should return 401."""
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.post(
+            "/api/auth/login",
+            json={"id": "admin", "pw": "wrongpassword"},
+        )
+    assert response.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_login_invalid_id() -> None:
+    """Wrong ID should return 401."""
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.post(
+            "/api/auth/login",
+            json={"id": "wronguser", "pw": "admin"},
+        )
+    assert response.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_refresh_token() -> None:
+    """Valid refresh token should return new access token."""
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        login_resp = await client.post(
+            "/api/auth/login",
+            json={"id": "admin", "pw": "admin"},
+        )
+        refresh_token = login_resp.json()["refresh_token"]
+
+        refresh_resp = await client.post(
+            "/api/auth/refresh",
+            json={"refresh_token": refresh_token},
+        )
+    assert refresh_resp.status_code == 200
+    data = refresh_resp.json()
+    assert "access_token" in data
+    assert "refresh_token" in data
+
+
+@pytest.mark.asyncio
+async def test_refresh_with_access_token_fails() -> None:
+    """Using access token as refresh token should fail."""
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        login_resp = await client.post(
+            "/api/auth/login",
+            json={"id": "admin", "pw": "admin"},
+        )
+        access_token = login_resp.json()["access_token"]
+
+        refresh_resp = await client.post(
+            "/api/auth/refresh",
+            json={"refresh_token": access_token},
+        )
+    assert refresh_resp.status_code == 401
+
+
+@pytest.mark.asyncio
+async def test_health_endpoint_without_token() -> None:
+    """Health check should work without authentication."""
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as client:
+        response = await client.get("/api/health")
+    assert response.status_code == 200