dev/msa-django-ansible
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Ansible 기능 추가

Browse Source
This commit is contained in:
icurfer
2025-05-21 01:08:10 +09:00
parent 69aca7ae85
commit ed5c62b1e5
10 changed files with 161 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
ansible/migrations/0002_ansibletask_author_email.py Normal file
View File

@ -0,0 +1,19 @@
# Generated by Django 4.2.14 on 2025-05-20 15:56
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('ansible', '0001_initial'),
]
operations = [
migrations.AddField(
model_name='ansibletask',
name='author_email',
field=models.CharField(default='icurfer@gmail.com', max_length=150),
preserve_default=False,
),
]

14
ansible/models.py
View File

@ -1,12 +1,16 @@
# ansible/models.py
from django.db import models
class AnsibleTask(models.Model):
name = models.CharField(max_length=200)
playbook_content = models.TextField() # ✅ YAML 문자열
inventory_content = models.TextField() # ✅ 인벤토리 형식 문자열
status = models.CharField(max_length=50, default='pending') # 'pending', 'running', 'success', 'failed', 'error'
output = models.TextField(blank=True) # ✅ 실행 결과 로그
author_email = models.CharField(max_length=150)
playbook_content = models.TextField()
inventory_content = models.TextField()
status = models.CharField(max_length=50, default='pending') # pending, running, success, failed, error
output = models.TextField(blank=True)
created_at = models.DateTimeField(auto_now_add=True)
def __str__(self):
return f"{self.name} ({self.status})"
return f"{self.name} ({self.status})"

3
ansible/serializers.py
View File

@ -1,8 +1,8 @@
# msa-django-ansible/serializers.py
from rest_framework import serializers
from .models import AnsibleTask
# ✅ 기본 Serializer: 목록 / 생성용
class AnsibleTaskSerializer(serializers.ModelSerializer):
class Meta:
model = AnsibleTask
@ -18,7 +18,6 @@ class AnsibleTaskSerializer(serializers.ModelSerializer):
read_only_fields = ("id", "status", "output", "created_at")
# ✅ 상세용 Serializer: 실행 결과만 확인
class AnsibleTaskDetailSerializer(serializers.ModelSerializer):
class Meta:
model = AnsibleTask

23
ansible/services.py
View File

@ -1,9 +1,20 @@
# services.py
# msa-django-ansible/services.py
import os
import requests
import tempfile
import subprocess
from .models import AnsibleTask
from django.conf import settings
from .models import AnsibleTask
def get_ssh_key_from_auth_server(access_token: str) -> str:
url = settings.AUTH_VERIFY_URL + "/api/auth/ssh-key/view/"
print(url)
headers = {"Authorization": f"Bearer {access_token}"}
response = requests.get(url, headers=headers)
if response.status_code != 200:
raise Exception("🔐 Auth 서버에서 SSH 키 조회 실패")
return response.json().get("ssh_key")
def run_ansible_job(task: AnsibleTask, ssh_key: str):
@ -16,12 +27,11 @@ def run_ansible_job(task: AnsibleTask, ssh_key: str):
tempfile.NamedTemporaryFile(delete=False, mode="w") as private_key_file:
playbook_file.write(task.playbook_content.strip())
playbook_file.close()
inventory_file.write(task.inventory_content.strip())
inventory_file.close()
private_key_file.write(ssh_key.strip() + "\n")
playbook_file.close()
inventory_file.close()
private_key_file.close()
os.chmod(private_key_file.name, 0o600)
@ -45,5 +55,4 @@ def run_ansible_job(task: AnsibleTask, ssh_key: str):
for f in [playbook_file.name, inventory_file.name, private_key_file.name]:
if os.path.exists(f):
os.remove(f)
task.save()

3
ansible/urls.py
View File

@ -1,3 +1,4 @@
# ansible/urls.py
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from .views import AnsibleTaskViewSet
@ -7,4 +8,4 @@ router.register(r'tasks', AnsibleTaskViewSet, basename='ansibletask')
urlpatterns = [
path('', include(router.urls)),
]
]

124
ansible/views.py
View File

@ -1,101 +1,47 @@
from rest_framework.views import APIView
from rest_framework import viewsets, status
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
from cryptography.fernet import Fernet
from django.conf import settings
import base64
import hashlib
# Fernet 키를 settings.SECRET_KEY에서 파생
hashed = hashlib.sha256(settings.SECRET_KEY.encode()).digest()
fernet_key = base64.urlsafe_b64encode(hashed[:32])
fernet = Fernet(fernet_key)
from rest_framework.decorators import action
from .models import AnsibleTask
from .serializers import AnsibleTaskSerializer, AnsibleTaskDetailSerializer
from .services import run_ansible_job, get_ssh_key_from_auth_server
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):
class AnsibleTaskViewSet(viewsets.ModelViewSet):
serializer_class = AnsibleTaskSerializer
permission_classes = [IsAuthenticated]
def get(self, request):
user = request.user
serializer = RegisterSerializer(user)
return Response(serializer.data)
def get_queryset(self):
# ✅ 현재 로그인한 사용자의 email로 필터
return AnsibleTask.objects.filter(author_email=self.request.user.email).order_by("-created_at")
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)
def get_serializer_class(self):
if self.action == "retrieve":
return AnsibleTaskDetailSerializer
return AnsibleTaskSerializer
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
class CustomTokenObtainPairView(TokenObtainPairView):
serializer_class = CustomTokenObtainPairSerializer
# ✅ author_email 저장
task = serializer.save(author_email=request.user.email)
return Response(self.get_serializer(task).data, status=status.HTTP_201_CREATED)
# try:
# token = request.headers.get("Authorization", "").replace("Bearer ", "")
# ssh_key = get_ssh_key_from_auth_server(token)
# run_ansible_job(task, ssh_key)
# return Response(self.get_serializer(task).data, status=status.HTTP_201_CREATED)
# except Exception as e:
# return Response({"error": f"작업 실행 실패: {str(e)}"}, status=500)
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
@action(detail=True, methods=["post"], url_path="run")
def run_task(self, request, pk=None):
task = self.get_object()
try:
encrypted_key = fernet.encrypt(private_key.encode()).decode() # ✅ decode 추가
user.encrypted_private_key = encrypted_key
user.encrypted_private_key_name = key_name
user.save(update_fields=["encrypted_private_key", "encrypted_private_key_name"])
return Response({"message": "SSH key 저장 완료."})
token = request.headers.get("Authorization", "").replace("Bearer ", "")
ssh_key = get_ssh_key_from_auth_server(token)
run_ansible_job(task, ssh_key)
return Response(self.get_serializer(task).data)
except Exception as e:
return Response({"error": 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
})
# ✅ 실제 암호화된 키를 반환하는 API
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)
return Response({"ssh_key": user.encrypted_private_key})
return Response({"error": f"작업 실행 실패: {str(e)}"}, status=500)