change RS256 algorithm

.gitignore

@@ -163,3 +163,6 @@ cython_debug/
 #.idea/
 _media/cluster/
 wheelhouse
+
+# RS256 을 위한 적용 keys 폴더
+keys 

README.md

@@ -1,14 +1,19 @@
 # msa-django-auth
 
 ## dev env
+개발환경 테스트 실행
+```bash
+python3 manage.py runserver 0.0.0.0:8000
+```
+
 ### auth
 
 ```bash
 gunicorn auth_prj.wsgi:application --bind 0.0.0.0:8000 --workers 3
 ```
 
-### blog
+## 2025-09-28 RS256변경 적용 ( v0.0.12 )
-
+* 비대칭키 방식 → Private Key로 서명, Public Key로 검증.
-```bash
+* 토큰 발급 서버는 Private Key만 보관.
-gunicorn auth_prj.wsgi:application --bind 0.0.0.0:8800 --workers 3
+* 검증 서버들은 Public Key만 있으면 됨 → 여러 서비스/마이크로서비스 환경에 적합.
-```
+* Istio, Keycloak, Auth0 등 대부분의 IDP/게이트웨이가 RS256 + JWKS(JSON Web Key Set) 방식 권장.

auth_prj/settings.py

@@ -169,12 +169,36 @@ TEMPLATES = [
 
 WSGI_APPLICATION = 'auth_prj.wsgi.application'
 
-SIMPLE_JWT = {
+# JWT 설정
-    "ACCESS_TOKEN_LIFETIME": timedelta(minutes=5),
+# https://django-rest-framework-simplejwt.readthedocs.io/en/latest/settings.html
-    "REFRESH_TOKEN_LIFETIME": timedelta(days=1),
+# istio jwt token check 
-    "ROTATE_REFRESH_TOKENS": True,
+ISTIO_JWT = os.environ.get("ISTIO_JWT", "0") == "1"
-    "BLACKLIST_AFTER_ROTATION": True, # 사용한 토큰은 갱신하면 블랙리스트처리됨
+
-}
+if ISTIO_JWT:
+    # RS256 모드 
+    # 운영환경에서 key파일은 POD mount로 적용하는게 안전
+    with open(BASE_DIR / "keys/private.pem", "r") as f:
+        PRIVATE_KEY = f.read()
+    with open(BASE_DIR / "keys/public.pem", "r") as f:
+        PUBLIC_KEY = f.read()
+
+    SIMPLE_JWT = {
+        "ALGORITHM": "RS256",
+        "SIGNING_KEY": PRIVATE_KEY,
+        "VERIFYING_KEY": PUBLIC_KEY,
+        "ISSUER": "msa-user",
+        "ACCESS_TOKEN_LIFETIME": timedelta(minutes=30),  
+        "REFRESH_TOKEN_LIFETIME": timedelta(days=1),
+        "ROTATE_REFRESH_TOKENS": True,
+        "BLACKLIST_AFTER_ROTATION": True, # 사용한 토큰은 갱신하면 블랙리스트처리됨
+    }
+else:
+    SIMPLE_JWT = {
+        "ACCESS_TOKEN_LIFETIME": timedelta(minutes=5),
+        "REFRESH_TOKEN_LIFETIME": timedelta(days=1),
+        "ROTATE_REFRESH_TOKENS": True,
+        "BLACKLIST_AFTER_ROTATION": True, # 사용한 토큰은 갱신하면 블랙리스트처리됨
+    }
 
 DATABASES = {
     "default": {

auth_prj/urls copy.py

@@ -1,23 +0,0 @@
-from django.urls import path, include, re_path
-from rest_framework import permissions
-from drf_yasg.views import get_schema_view
-from drf_yasg import openapi
-
-schema_view = get_schema_view(
-   openapi.Info(
-      title="msa-django-auth API",
-      default_version='v1',
-      description="인증 서비스용 JWT API 문서",
-   ),
-   public=True,
-   permission_classes=(permissions.AllowAny,),
-)
-
-urlpatterns = [
-    path('admin/', admin.site.urls),
-    path('api/auth/', include('users.urls')),
-
-    re_path(r'^swagger(?P<format>\.json|\.yaml)$', schema_view.without_ui(cache_timeout=0), name='schema-json'),
-    path('swagger/', schema_view.with_ui('swagger', cache_timeout=0), name='schema-swagger-ui'),
-    path('redoc/', schema_view.with_ui('redoc', cache_timeout=0), name='schema-redoc'),
-]

users/_unused_views.py

@@ -1,137 +0,0 @@
-# views.py
-import logging
-from opentelemetry import trace # OpenTelemetry 임포트
-from rest_framework.views import APIView
-from rest_framework.response import Response
-from rest_framework import status
-from rest_framework.permissions import IsAuthenticated
-from rest_framework_simplejwt.views import TokenObtainPairView
-from .serializers import RegisterSerializer, CustomTokenObtainPairSerializer
-
-logger = logging.getLogger(__name__)
-tracer = trace.get_tracer(__name__)  # 트레이서 가져오기
-
-def get_request_info(request):
-    ip = request.META.get("REMOTE_ADDR", "unknown")
-    ua = request.META.get("HTTP_USER_AGENT", "unknown")
-    email = getattr(request.user, "email", "anonymous")
-    return email, ip, ua
-
-class RegisterView(APIView):
-    def post(self, request):
-        email, ip, ua = get_request_info(request)
-        serializer = RegisterSerializer(data=request.data)
-        if serializer.is_valid():
-            user = serializer.save()
-            logger.info(f"[REGISTER] user={user.email} | status=success | IP={ip} | UA={ua}")
-            return Response({"message": "User registered successfully."}, status=status.HTTP_201_CREATED)
-        logger.warning(f"[REGISTER] user={email} | status=fail | IP={ip} | UA={ua} | detail={serializer.errors}")
-        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
-
-
-class MeView(APIView):
-    permission_classes = [IsAuthenticated]
-
-    def get(self, request):
-        email, ip, ua = get_request_info(request)
-        logger.debug(f"[ME GET] user={email} | IP={ip} | UA={ua}")
-        serializer = RegisterSerializer(request.user)
-        return Response(serializer.data)
-
-    def put(self, request):
-        email, ip, ua = get_request_info(request)
-        serializer = RegisterSerializer(request.user, data=request.data, partial=True)
-        if serializer.is_valid():
-            serializer.save()
-            logger.info(f"[ME UPDATE] user={email} | status=success | IP={ip} | UA={ua}")
-            return Response(serializer.data)
-        logger.warning(f"[ME UPDATE] user={email} | status=fail | IP={ip} | UA={ua} | detail={serializer.errors}")
-        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
-
-
-class CustomTokenObtainPairView(TokenObtainPairView):
-    serializer_class = CustomTokenObtainPairSerializer
-
-    def post(self, request, *args, **kwargs):
-        ip = request.META.get("REMOTE_ADDR", "unknown")
-        ua = request.META.get("HTTP_USER_AGENT", "unknown")
-        email = request.data.get("email", "unknown")
-        logger.info(f"[LOGIN] user={email} | status=attempt | IP={ip} | UA={ua}")
-        response = super().post(request, *args, **kwargs)
-
-        # 성공 실패 판단
-        if response.status_code == 200:
-            logger.info(f"[LOGIN] user={email} | status=success | IP={ip} | UA={ua}")
-        else:
-            logger.warning(f"[LOGIN] user={email} | status=fail | IP={ip} | UA={ua} | detail={response.data}")
-        return response
-
-
-class SSHKeyUploadView(APIView):
-    permission_classes = [IsAuthenticated]
-
-    def post(self, request):
-        email, ip, ua = get_request_info(request)
-        private_key = request.data.get("private_key")
-        key_name = request.data.get("key_name")
-
-        if not private_key or not key_name:
-            logger.warning(f"[SSH UPLOAD] user={email} | status=fail | reason=missing_key_or_name | IP={ip} | UA={ua}")
-            return Response(
-                {"error": "private_key와 key_name 모두 필요합니다."},
-                status=status.HTTP_400_BAD_REQUEST
-            )
-
-        try:
-            user = request.user
-            user.save_private_key(private_key)
-            user.encrypted_private_key_name = key_name
-            user.save(update_fields=["encrypted_private_key", "encrypted_private_key_name"])
-            logger.info(f"[SSH UPLOAD] user={email} | status=success | key_name={key_name} | IP={ip} | UA={ua}")
-            return Response({"message": "SSH key 저장 완료."}, status=201)
-        except Exception as e:
-            logger.exception(f"[SSH UPLOAD] user={email} | status=fail | reason=exception | IP={ip} | UA={ua}")
-            return Response({"error": f"암호화 또는 저장 실패: {str(e)}"}, status=500)
-
-    def delete(self, request):
-        email, ip, ua = get_request_info(request)
-        user = request.user
-        user.encrypted_private_key = None
-        user.encrypted_private_key_name = None
-        user.last_used_at = None
-        user.save(update_fields=["encrypted_private_key", "encrypted_private_key_name", "last_used_at"])
-        logger.info(f"[SSH DELETE] user={email} | status=success | IP={ip} | UA={ua}")
-        return Response({"message": "SSH key deleted."}, status=200)
-
-
-class SSHKeyInfoView(APIView):
-    permission_classes = [IsAuthenticated]
-
-    def get(self, request):
-        email, ip, ua = get_request_info(request)
-        logger.debug(f"[SSH INFO] user={email} | IP={ip} | UA={ua}")
-        user = request.user
-        return Response({
-            "has_key": bool(user.encrypted_private_key),
-            "encrypted_private_key_name": user.encrypted_private_key_name,
-            "last_used_at": user.last_used_at
-        })
-
-
-class SSHKeyRetrieveView(APIView):
-    permission_classes = [IsAuthenticated]
-
-    def get(self, request):
-        email, ip, ua = get_request_info(request)
-        user = request.user
-        if not user.encrypted_private_key:
-            logger.warning(f"[SSH RETRIEVE] user={email} | status=fail | reason=not_found | IP={ip} | UA={ua}")
-            return Response({"error": "SSH 키가 등록되어 있지 않습니다."}, status=404)
-
-        try:
-            decrypted_key = user.decrypt_private_key()
-            logger.info(f"[SSH RETRIEVE] user={email} | status=success | IP={ip} | UA={ua}")
-            return Response({"ssh_key": decrypted_key})
-        except Exception as e:
-            logger.exception(f"[SSH RETRIEVE] user={email} | status=fail | reason=exception | IP={ip} | UA={ua}")
-            return Response({"error": f"복호화 실패: {str(e)}"}, status=500)

users/urls.py

@@ -1,6 +1,7 @@
 from django.urls import path
 from .views import RegisterView, MeView, CustomTokenObtainPairView, SSHKeyUploadView, SSHKeyInfoView, SSHKeyRetrieveView
 from rest_framework_simplejwt.views import TokenObtainPairView, TokenRefreshView, TokenVerifyView
+from .views_jwks import jwks_view # django-jwks
 
 urlpatterns = [
     path('register/', RegisterView.as_view(), name='register'),
@@ -12,4 +13,5 @@ urlpatterns = [
     path("ssh-key/", SSHKeyUploadView.as_view(), name="ssh_key_upload"),
     path("ssh-key/info/", SSHKeyInfoView.as_view(), name="ssh_key_info"),
     path("ssh-key/view/", SSHKeyRetrieveView.as_view(), name="ssh_key_retrieve"),
+    path(".well-known/jwks.json", jwks_view, name="jwks"), # django-jwks
 ]

users/views_jwks.py

@@ -0,0 +1,31 @@
+# users/views_jwks.py
+from django.http import JsonResponse, HttpResponseNotFound
+from django.conf import settings
+import base64
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.backends import default_backend
+
+def jwks_view(request):
+    if settings.SIMPLE_JWT["ALGORITHM"] != "RS256":
+        return HttpResponseNotFound("JWKS is only available in RS256 mode")
+
+    public_key = settings.SIMPLE_JWT["VERIFYING_KEY"]
+
+    key = serialization.load_pem_public_key(
+        public_key.encode(), backend=default_backend()
+    )
+    numbers = key.public_numbers()
+
+    e = numbers.e.to_bytes((numbers.e.bit_length() + 7) // 8, "big")
+    n = numbers.n.to_bytes((numbers.n.bit_length() + 7) // 8, "big")
+
+    jwk = {
+        "kty": "RSA",
+        "use": "sig",
+        "alg": "RS256",
+        "kid": "msa-user-key",
+        "n": base64.urlsafe_b64encode(n).decode().rstrip("="),
+        "e": base64.urlsafe_b64encode(e).decode().rstrip("="),
+    }
+
+    return JsonResponse({"keys": [jwk]})

version

@@ -1 +1 @@
-0.0.11_rc8
+v0.0.12