change RS256 algorithm
This commit is contained in:
3
.gitignore
vendored
3
.gitignore
vendored
@ -163,3 +163,6 @@ cython_debug/
|
||||
#.idea/
|
||||
_media/cluster/
|
||||
wheelhouse
|
||||
|
||||
# RS256 을 위한 적용 keys 폴더
|
||||
keys
|
||||
15
README.md
15
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
|
||||
|
||||
```bash
|
||||
gunicorn auth_prj.wsgi:application --bind 0.0.0.0:8800 --workers 3
|
||||
```
|
||||
## 2025-09-28 RS256변경 적용 ( v0.0.12 )
|
||||
* 비대칭키 방식 → Private Key로 서명, Public Key로 검증.
|
||||
* 토큰 발급 서버는 Private Key만 보관.
|
||||
* 검증 서버들은 Public Key만 있으면 됨 → 여러 서비스/마이크로서비스 환경에 적합.
|
||||
* Istio, Keycloak, Auth0 등 대부분의 IDP/게이트웨이가 RS256 + JWKS(JSON Web Key Set) 방식 권장.
|
||||
@ -169,12 +169,36 @@ TEMPLATES = [
|
||||
|
||||
WSGI_APPLICATION = 'auth_prj.wsgi.application'
|
||||
|
||||
SIMPLE_JWT = {
|
||||
"ACCESS_TOKEN_LIFETIME": timedelta(minutes=5),
|
||||
"REFRESH_TOKEN_LIFETIME": timedelta(days=1),
|
||||
"ROTATE_REFRESH_TOKENS": True,
|
||||
"BLACKLIST_AFTER_ROTATION": True, # 사용한 토큰은 갱신하면 블랙리스트처리됨
|
||||
}
|
||||
# JWT 설정
|
||||
# https://django-rest-framework-simplejwt.readthedocs.io/en/latest/settings.html
|
||||
# istio jwt token check
|
||||
ISTIO_JWT = os.environ.get("ISTIO_JWT", "0") == "1"
|
||||
|
||||
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": {
|
||||
|
||||
@ -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'),
|
||||
]
|
||||
@ -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)
|
||||
@ -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
|
||||
]
|
||||
|
||||
31
users/views_jwks.py
Normal file
31
users/views_jwks.py
Normal file
@ -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]})
|
||||
Reference in New Issue
Block a user