headscale-tailscale-replacement/farmq-admin/utils/proxmox_client.py

#!/usr/bin/env python3
"""
Proxmox VE API 클라이언트
"""

import requests
import json
import urllib3
from urllib.parse import quote_plus
from typing import Dict, List, Optional, Tuple

# SSL 경고 무시
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class ProxmoxClient:
    def __init__(self, host: str, username: str = "root@pam", password: str = "", api_token: str = ""):
        self.host = host
        self.username = username
        self.password = password
        self.api_token = api_token
        self.base_url = f"https://{host}:443/api2/json"
        self.session = requests.Session()
        self.session.verify = False
        self.ticket = None
        self.csrf_token = None
        
    def login(self) -> bool:
        """세션 쿠키 방식으로 로그인"""
        if self.api_token:
            # API Token 방식
            self.session.headers.update({
                'Authorization': f'PVEAPIToken={self.api_token}'
            })
            return self._test_connection()
        else:
            # 패스워드 방식
            login_data = {
                'username': self.username,
                'password': self.password
            }
            
            try:
                response = self.session.post(
                    f"{self.base_url}/access/ticket",
                    data=login_data,
                    timeout=10
                )
                
                if response.status_code == 200:
                    data = response.json()['data']
                    self.ticket = data['ticket']
                    self.csrf_token = data['CSRFPreventionToken']
                    
                    # 쿠키와 헤더 설정
                    self.session.cookies.update({'PVEAuthCookie': self.ticket})
                    self.session.headers.update({'CSRFPreventionToken': self.csrf_token})
                    
                    return True
                
            except Exception as e:
                print(f"로그인 실패: {e}")
                return False
                
        return False
    
    def _test_connection(self) -> bool:
        """연결 테스트"""
        try:
            response = self.session.get(f"{self.base_url}/version", timeout=10)
            return response.status_code == 200
        except:
            return False
    
    def get_vm_list(self) -> List[Dict]:
        """VM 목록 조회"""
        try:
            response = self.session.get(
                f"{self.base_url}/cluster/resources?type=vm",
                timeout=10
            )
            
            if response.status_code == 200:
                return response.json()['data']
            
        except Exception as e:
            print(f"VM 목록 조회 실패: {e}")
            
        return []
    
    def get_vm_status(self, node: str, vmid: int) -> Dict:
        """특정 VM 상태 확인"""
        try:
            response = self.session.get(
                f"{self.base_url}/nodes/{node}/qemu/{vmid}/status/current",
                timeout=10
            )
            
            if response.status_code == 200:
                return response.json()['data']
                
        except Exception as e:
            print(f"VM {vmid} 상태 조회 실패: {e}")
            
        return {}
    
    def get_vnc_ticket(self, node: str, vmid: int) -> Optional[Dict]:
        """VNC 접속 티켓 생성"""
        try:
            data = {
                'websocket': '1',
                'generate-password': '1'  # 패스워드 생성 활성화
            }
            response = self.session.post(
                f"{self.base_url}/nodes/{node}/qemu/{vmid}/vncproxy",
                data=data,
                timeout=10
            )
            
            if response.status_code == 200:
                vnc_data = response.json()['data']
                print(f"✅ VNC 티켓 생성 성공: {vnc_data}")
                
                # WebSocket URL 생성
                encoded_ticket = quote_plus(vnc_data['ticket'])
                vnc_data['websocket_url'] = f"wss://{self.host}:443/api2/json/nodes/{node}/qemu/{vmid}/vncwebsocket?port={vnc_data['port']}&vncticket={encoded_ticket}"
                
                # 디버깅 정보 추가
                print(f"🔗 WebSocket URL: {vnc_data['websocket_url']}")
                return vnc_data
            else:
                print(f"❌ VNC 티켓 생성 HTTP 오류: {response.status_code}")
                print(f"Response: {response.text}")
                
        except Exception as e:
            print(f"❌ VNC 티켓 생성 실패: {e}")
            
        return None
    
    def start_vm(self, node: str, vmid: int) -> bool:
        """VM 시작"""
        try:
            response = self.session.post(
                f"{self.base_url}/nodes/{node}/qemu/{vmid}/status/start",
                timeout=30
            )
            return response.status_code == 200
            
        except Exception as e:
            print(f"VM {vmid} 시작 실패: {e}")
            return False
    
    def stop_vm(self, node: str, vmid: int) -> bool:
        """VM 정지"""
        try:
            response = self.session.post(
                f"{self.base_url}/nodes/{node}/qemu/{vmid}/status/stop",
                timeout=30
            )
            return response.status_code == 200
            
        except Exception as e:
            print(f"VM {vmid} 정지 실패: {e}")
            return False
    
    def get_nodes(self) -> List[Dict]:
        """노드 목록 조회"""
        try:
            response = self.session.get(f"{self.base_url}/nodes", timeout=10)
            if response.status_code == 200:
                return response.json()['data']
        except Exception as e:
            print(f"노드 목록 조회 실패: {e}")
            
        return []