Files
msa-django-auth/users/views.py
icurfer e98090f031
All checks were successful
Build And Test / build-and-push (push) Successful in 2m49s
Ansible ssh키 암복호화기능추가
2025-05-21 01:04:45 +09:00

99 lines
3.5 KiB
Python

# views.py
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
class RegisterView(APIView):
def post(self, request):
serializer = RegisterSerializer(data=request.data)
if serializer.is_valid():
user = serializer.save()
return Response({"message": "User registered successfully."}, status=status.HTTP_201_CREATED)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
class MeView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
user = request.user
serializer = RegisterSerializer(user)
return Response(serializer.data)
def put(self, request):
user = request.user
serializer = RegisterSerializer(user, data=request.data, partial=True)
if serializer.is_valid():
serializer.save()
return Response(serializer.data)
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
class CustomTokenObtainPairView(TokenObtainPairView):
serializer_class = CustomTokenObtainPairSerializer
class SSHKeyUploadView(APIView):
permission_classes = [IsAuthenticated]
def post(self, request):
private_key = request.data.get("private_key")
key_name = request.data.get("key_name")
if not private_key or not key_name:
return Response(
{"error": "private_key와 key_name 모두 필요합니다."},
status=status.HTTP_400_BAD_REQUEST
)
user = request.user
try:
# ✅ 모델 메서드로 암호화 저장 처리
user.save_private_key(private_key)
user.encrypted_private_key_name = key_name
user.save(update_fields=["encrypted_private_key", "encrypted_private_key_name"])
return Response({"message": "SSH key 저장 완료."}, status=201)
except Exception as e:
return Response({"error": f"암호화 또는 저장 실패: {str(e)}"}, status=500)
def delete(self, 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"])
return Response({"message": "SSH key deleted."}, status=200)
class SSHKeyInfoView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
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):
user = request.user
if not user.encrypted_private_key:
return Response({"error": "SSH 키가 등록되어 있지 않습니다."}, status=404)
try:
# ✅ 모델 메서드로 복호화 처리
decrypted_key = user.decrypt_private_key()
return Response({"ssh_key": decrypted_key})
except Exception as e:
return Response({"error": f"복호화 실패: {str(e)}"}, status=500)