v0.0.25 | KVM 서버 관리 API 추가
All checks were successful
Build And Test / build-and-push (push) Successful in 2m7s

- KVMServer 모델 추가 (멀티 서버 지원)
- 서버별 SSH 키 암호화 저장
- msa-django-libvirt 연동용 SSH 정보 조회 API

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-01-16 11:14:56 +09:00
parent 85f5688a0b
commit 2c050829ff
6 changed files with 494 additions and 4 deletions

View File

@ -198,3 +198,97 @@ class NHNCloudProject(models.Model):
"""자격증명 저장 (비밀번호 암호화)"""
self.encrypted_password = self.encrypt_password(password)
self.save()
# ============================================
# KVM 서버 관리 (멀티 서버 지원)
# ============================================
class KVMServer(models.Model):
"""
사용자별 KVM 서버 관리 (멀티 서버 지원)
msa-django-libvirt에서 SSH 접속 정보를 요청할 때 사용
"""
user = models.ForeignKey(
CustomUser,
on_delete=models.CASCADE,
related_name='kvm_servers',
verbose_name="사용자"
)
name = models.CharField(max_length=100, verbose_name="서버 별칭")
host = models.CharField(max_length=255, verbose_name="호스트 (IP 또는 도메인)")
port = models.IntegerField(default=22, verbose_name="SSH 포트")
username = models.CharField(max_length=100, verbose_name="SSH 사용자명")
encrypted_private_key_name = models.CharField(
max_length=100, blank=True, null=True, verbose_name="SSH 키 이름"
)
encrypted_private_key = models.BinaryField(
blank=True, null=True, verbose_name="SSH 개인키 (암호화)"
)
libvirt_uri = models.CharField(
max_length=255, blank=True, null=True,
verbose_name="Libvirt URI",
help_text="예: qemu+ssh://user@host/system"
)
description = models.TextField(blank=True, null=True, verbose_name="설명")
tags = models.CharField(
max_length=500, blank=True, null=True,
verbose_name="태그",
help_text="쉼표로 구분된 태그 목록"
)
is_active = models.BooleanField(default=True, verbose_name="활성화 상태")
last_used_at = models.DateTimeField(blank=True, null=True, verbose_name="마지막 사용 시각")
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
verbose_name = "KVM 서버"
verbose_name_plural = "KVM 서버"
unique_together = ['user', 'host', 'port'] # 동일 사용자가 같은 호스트:포트 중복 등록 방지
ordering = ['-is_active', '-created_at']
def __str__(self):
return f"{self.name} ({self.host}:{self.port})"
def get_encryption_key(self) -> bytes:
"""SECRET_KEY 기반 Fernet 키 생성"""
hashed = hashlib.sha256(settings.SECRET_KEY.encode()).digest()
return base64.urlsafe_b64encode(hashed[:32])
def encrypt_private_key(self, private_key: str) -> bytes:
"""SSH 개인키 암호화"""
cipher = Fernet(self.get_encryption_key())
return cipher.encrypt(private_key.encode())
def decrypt_private_key(self) -> str:
"""SSH 개인키 복호화"""
if self.encrypted_private_key:
cipher = Fernet(self.get_encryption_key())
decrypted = cipher.decrypt(self.encrypted_private_key).decode()
self.last_used_at = timezone.now()
self.save(update_fields=['last_used_at'])
return decrypted
return ""
def save_ssh_key(self, private_key: str, key_name: str = None):
"""SSH 키 저장 (암호화)"""
self.encrypted_private_key = self.encrypt_private_key(private_key)
if key_name:
self.encrypted_private_key_name = key_name
self.save()
def get_tags_list(self) -> list:
"""태그 문자열을 리스트로 반환"""
if self.tags:
return [tag.strip() for tag in self.tags.split(',') if tag.strip()]
return []
def set_tags_list(self, tags_list: list):
"""태그 리스트를 문자열로 저장"""
self.tags = ', '.join(tags_list)
def get_libvirt_uri(self) -> str:
"""Libvirt URI 반환 (없으면 기본값 생성)"""
if self.libvirt_uri:
return self.libvirt_uri
return f"qemu+ssh://{self.username}@{self.host}:{self.port}/system"