Azure Key Vault MCP

import sys
import os
from typing import List, Dict, Optional
from azure.keyvault.secrets import SecretClient
from azure.keyvault.certificates import CertificateClient, CertificatePolicy
from azure.core.exceptions import ResourceNotFoundError

# SSL 인증서 검증 제어 (프록시 환경 대응)
if os.environ.get("AZURE_KEYVAULT_DISABLE_SSL_VERIFY", "").lower() in ("1", "true", "yes"):
    import ssl
    import urllib3
    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    os.environ['PYTHONHTTPSVERIFY'] = '0'
    os.environ['CURL_CA_BUNDLE'] = ''
    os.environ['REQUESTS_CA_BUNDLE'] = ''

class KeyVaultManager:
    """Key Vault Secret 및 Certificate 관리"""
    
    def __init__(self, vault_url: str, credential):
        self.vault_url = vault_url
        self.credential = credential
        
        # ⭐ SSL 검증 완전히 비활성화 (KT 프록시 환경 대응)
        print("🔧 SSL 검증 비활성화 모드로 Key Vault 연결 중...", file=sys.stderr)
        
        import ssl
        import urllib3
        import requests
        from requests.adapters import HTTPAdapter
        from urllib3.util.ssl_ import create_urllib3_context
        from azure.core.pipeline.transport import RequestsTransport
        
        # SSL 경고 비활성화
        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
        
        # 전역적으로 SSL 검증 비활성화 (requests 라이브러리)
        requests.packages.urllib3.disable_warnings()
        
        # 커스텀 SSL 컨텍스트 생성 (검증 완전히 비활성화)
        class NoVerifyHTTPAdapter(HTTPAdapter):
            def init_poolmanager(self, *args, **kwargs):
                ctx = create_urllib3_context()
                ctx.check_hostname = False
                ctx.verify_mode = ssl.CERT_NONE
                kwargs['ssl_context'] = ctx
                return super().init_poolmanager(*args, **kwargs)
        
        # SSL 검증 비활성화된 세션 생성
        session = requests.Session()
        session.verify = False
        session.mount('https://', NoVerifyHTTPAdapter())
        
        # RequestsTransport를 상속받아 session을 강제로 설정
        class CustomRequestsTransport(RequestsTransport):
            def __init__(self, session=None, *args, **kwargs):
                # session을 먼저 설정하지 않고 부모 초기화
                super().__init__(*args, **kwargs)
                if session:
                    # 내부 session 속성 강제 설정
                    self.session = session
                    if hasattr(self, '_session'):
                        self._session = session
                    # _client도 설정 (일부 버전에서 필요)
                    if hasattr(self, '_client'):
                        self._client.session = session
            
            def send(self, request, **kwargs):
                # Azure SDK는 connection_verify를 사용하므로 이를 False로 설정
                kwargs['connection_verify'] = False
                # verify도 False로 설정 (혹시 모를 경우 대비)
                if 'verify' in kwargs:
                    kwargs.pop('verify')
                # 부모 클래스의 send 메서드 호출
                return super().send(request, **kwargs)
        
        transport = CustomRequestsTransport(session=session)
        
        # 클라이언트 생성
        self.secret_client = SecretClient(
            vault_url=vault_url, 
            credential=credential,
            transport=transport
        )
        self.cert_client = CertificateClient(
            vault_url=vault_url, 
            credential=credential,
            transport=transport
        )
        
        print("✅ SSL 검증 비활성화 완료", file=sys.stderr)
        self._test_connection()
    
    def _test_connection(self):
        """Key Vault 접근 테스트"""
        try:
            list(self.secret_client.list_properties_of_secrets(max_page_size=1))
            print(f"✅ Key Vault 연결 성공: {self.vault_url}", file=sys.stderr)
        except Exception as e:
            error_msg = str(e)
            
            # SSL 오류 감지
            is_ssl_error = (
                "CERTIFICATE_VERIFY_FAILED" in error_msg or 
                "certificate verify failed" in error_msg.lower() or
                "ssl" in error_msg.lower()
            )
            
            if is_ssl_error:
                print(f"❌ SSL 인증서 검증 오류 (여전히 발생):", file=sys.stderr)
                print(f"   {error_msg}", file=sys.stderr)
                print(f"", file=sys.stderr)
                print(f"💡 해결 방법:", file=sys.stderr)
                print(f"   1. MCP 설정에서 환경 변수 추가:", file=sys.stderr)
                print(f"      AZURE_KEYVAULT_DISABLE_SSL_VERIFY=1", file=sys.stderr)
                print(f"   2. Claude Desktop 재시작", file=sys.stderr)
                raise ConnectionError(f"SSL 인증서 검증 실패: {error_msg}")
            
            # 권한 오류
            print(f"❌ Key Vault 접근 실패: {e}", file=sys.stderr)
            vault_name = self.vault_url.split("//")[1].split(".")[0]
            print(f"💡 권한 부여가 필요할 수 있습니다:", file=sys.stderr)
            print(f"az role assignment create \\", file=sys.stderr)
            print(f"  --role 'Key Vault Secrets Officer' \\", file=sys.stderr)
            print(f"  --assignee $(az ad signed-in-user show --query id -o tsv) \\", file=sys.stderr)
            print(f"  --scope $(az keyvault show --name {vault_name} --query id -o tsv)", file=sys.stderr)
            raise ConnectionError(f"Key Vault 접근 실패: {e}")
    
    # ===== SECRET 관리 =====
    
    def set_secret(self, name: str, value: str) -> Dict:
        """Secret 생성/업데이트"""
        try:
            secret = self.secret_client.set_secret(name, value)
            return {
                "success": True,
                "name": secret.name,
                "version": secret.properties.version,
                "created": str(secret.properties.created_on)
            }
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    def get_secret(self, name: str) -> Dict:
        """Secret 조회"""
        try:
            secret = self.secret_client.get_secret(name)
            return {
                "success": True,
                "name": secret.name,
                "value": secret.value,
                "version": secret.properties.version,
                "updated": str(secret.properties.updated_on)
            }
        except ResourceNotFoundError:
            return {"success": False, "error": f"Secret '{name}'을 찾을 수 없습니다."}
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    def list_secrets(self) -> List[Dict]:
        """모든 Secret 목록 조회"""
        try:
            secrets = []
            for secret_props in self.secret_client.list_properties_of_secrets():
                secrets.append({
                    "name": secret_props.name,
                    "enabled": secret_props.enabled,
                    "created": str(secret_props.created_on),
                    "updated": str(secret_props.updated_on)
                })
            return secrets
        except Exception as e:
            print(f"❌ Secret 목록 조회 실패: {e}", file=sys.stderr)
            return []
    
    def delete_secret(self, name: str) -> Dict:
        """Secret 삭제 (soft delete)"""
        try:
            poller = self.secret_client.begin_delete_secret(name)
            deleted_secret = poller.result()
            return {
                "success": True,
                "name": deleted_secret.name,
                "deleted_on": str(deleted_secret.deleted_date)
            }
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    # ===== CERTIFICATE 관리 =====
    
    def import_certificate(self, name: str, pfx_bytes: bytes, password: Optional[str] = None) -> Dict:
        """PFX 인증서 import"""
        try:
            has_existing = False
            try:
                self.cert_client.get_certificate(name)
                has_existing = True
            except ResourceNotFoundError:
                has_existing = False
            
            import_kwargs = {
                "certificate_name": name,
                "certificate_bytes": pfx_bytes,
                "password": password
            }
            
            # 기존 인증서가 없을 때만 정책 설정
            # Azure Key Vault import 시 content_type을 명시적으로 설정
            if not has_existing:
                # PFX 형식 인증서를 위한 정책 생성
                policy = CertificatePolicy(
                    content_type="application/x-pkcs12"
                )
                import_kwargs["policy"] = policy
            
            cert = self.cert_client.import_certificate(**import_kwargs)
            
            return {
                "success": True,
                "name": cert.name,
                "id": cert.id,
                "thumbprint": cert.properties.x509_thumbprint.hex() if cert.properties.x509_thumbprint else None,
                "is_new": not has_existing  # 신규 추가인지 갱신인지 구분
            }
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    def get_certificate(self, name: str) -> Dict:
        """인증서 조회"""
        try:
            cert = self.cert_client.get_certificate(name)
            return {
                "success": True,
                "name": cert.name,
                "id": cert.id,
                "enabled": cert.properties.enabled,
                "created": str(cert.properties.created_on),
                "expires": str(cert.properties.expires_on),
                "thumbprint": cert.properties.x509_thumbprint.hex() if cert.properties.x509_thumbprint else None
            }
        except ResourceNotFoundError:
            return {"success": False, "error": f"Certificate '{name}'을 찾을 수 없습니다."}
        except Exception as e:
            return {"success": False, "error": str(e)}
    
    def list_certificates(self) -> List[Dict]:
        """모든 인증서 목록 조회"""
        try:
            certs = []
            for cert_props in self.cert_client.list_properties_of_certificates():
                certs.append({
                    "name": cert_props.name,
                    "enabled": cert_props.enabled,
                    "created": str(cert_props.created_on),
                    "expires": str(cert_props.expires_on) if cert_props.expires_on else None,
                    "thumbprint": cert_props.x509_thumbprint.hex() if cert_props.x509_thumbprint else None
                })
            return certs
        except Exception as e:
            print(f"❌ 인증서 목록 조회 실패: {e}", file=sys.stderr)
            return []
    
    def delete_certificate(self, name: str) -> Dict:
        """인증서 삭제"""
        try:
            poller = self.cert_client.begin_delete_certificate(name)
            deleted_cert = poller.result()
            # deleted_date 속성 확인 (속성 이름이 다를 수 있음)
            deleted_on = None
            if hasattr(deleted_cert, 'deleted_date') and deleted_cert.deleted_date:
                deleted_on = str(deleted_cert.deleted_date)
            elif hasattr(deleted_cert, 'deleted_on') and deleted_cert.deleted_on:
                deleted_on = str(deleted_cert.deleted_on)
            elif hasattr(deleted_cert.properties, 'deleted_on') and deleted_cert.properties.deleted_on:
                deleted_on = str(deleted_cert.properties.deleted_on)
            
            return {
                "success": True,
                "name": deleted_cert.name,
                "deleted_on": deleted_on
            }
        except Exception as e:
            return {"success": False, "error": str(e)}