We Ditched Azure Key Vault for GCP Secret Manager and Cut Multi-Cloud Costs by 35%

# migrate_secrets.py
# Migrates all secrets from Azure Key Vault to GCP Secret Manager
# Requires: azure-identity, azure-keyvault-secrets, google-cloud-secret-manager, python-dotenv
import os
import logging
from typing import List, Dict, Optional
from dotenv import load_dotenv
from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient as AzureSecretClient
from google.cloud import secretmanager
from google.api_core import exceptions as gcp_exceptions
from azure.core.exceptions import AzureError

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Load environment variables from .env file
load_dotenv()

# Configuration from environment variables
AZURE_KEY_VAULT_URL = os.getenv("AZURE_KEY_VAULT_URL")
GCP_PROJECT_ID = os.getenv("GCP_PROJECT_ID")
GCP_LOCATION = os.getenv("GCP_LOCATION", "global")
DRY_RUN = os.getenv("DRY_RUN", "false").lower() == "true"

def get_azure_secret_client() -> AzureSecretClient:
    """Initialize Azure Key Vault client with default credential."""
    if not AZURE_KEY_VAULT_URL:
        raise ValueError("AZURE_KEY_VAULT_URL environment variable not set")
    credential = DefaultAzureCredential()
    return AzureSecretClient(vault_url=AZURE_KEY_VAULT_URL, credential=credential)

def get_gcp_secret_client() -> secretmanager.SecretManagerServiceClient:
    """Initialize GCP Secret Manager client with default credential."""
    if not GCP_PROJECT_ID:
        raise ValueError("GCP_PROJECT_ID environment variable not set")
    return secretmanager.SecretManagerServiceClient()

def list_azure_secrets(client: AzureSecretClient) -> List[str]:
    """List all secret names in Azure Key Vault."""
    try:
        secrets = [secret.name for secret in client.list_properties_of_secrets()]
        logger.info(f"Found {len(secrets)} secrets in Azure Key Vault")
        return secrets
    except AzureError as e:
        logger.error(f"Failed to list Azure secrets: {e}")
        raise

def get_azure_secret_value(client: AzureSecretClient, secret_name: str) -> Optional[str]:
    """Fetch latest value of a secret from Azure Key Vault."""
    try:
        secret = client.get_secret(secret_name)
        return secret.value
    except AzureError as e:
        if "Not Found" in str(e):
            logger.warning(f"Secret {secret_name} not found in Azure Key Vault")
            return None
        logger.error(f"Failed to fetch secret {secret_name}: {e}")
        raise

def create_gcp_secret(
    client: secretmanager.SecretManagerServiceClient,
    project_id: str,
    secret_name: str,
    secret_value: str
) -> None:
    """Create a new secret in GCP Secret Manager with the given value."""
    parent = f"projects/{project_id}"
    # Check if secret already exists
    try:
        client.get_secret(secret_path=f"{parent}/secrets/{secret_name}")
        logger.warning(f"Secret {secret_name} already exists in GCP Secret Manager, skipping")
        return
    except gcp_exceptions.NotFound:
        pass
    except gcp_exceptions.GoogleAPIError as e:
        logger.error(f"Failed to check GCP secret {secret_name}: {e}")
        raise

    if DRY_RUN:
        logger.info(f"[DRY RUN] Would create secret {secret_name} in GCP Secret Manager")
        return

    # Create secret parent
    secret = {"replication": {"automatic": True}}
    client.create_secret(parent=parent, secret_id=secret_name, secret=secret)

    # Add secret version with value
    secret_path = f"{parent}/secrets/{secret_name}"
    payload = secretmanager.SecretPayload(data=secret_value.encode("UTF-8"))
    client.add_secret_version(parent=secret_path, payload=payload)
    logger.info(f"Successfully created secret {secret_name} in GCP Secret Manager")

def main() -> None:
    """Main migration logic."""
    logger.info(f"Starting migration (DRY_RUN={DRY_RUN})")
    try:
        azure_client = get_azure_secret_client()
        gcp_client = get_gcp_secret_client()

        secret_names = list_azure_secrets(azure_client)
        for secret_name in secret_names:
            value = get_azure_secret_value(azure_client, secret_name)
            if value is None:
                continue
            create_gcp_secret(gcp_client, GCP_PROJECT_ID, secret_name, value)

        logger.info("Migration completed successfully")
    except Exception as e:
        logger.error(f"Migration failed: {e}")
        raise

if __name__ == "__main__":
    main()

# gcp_secret_client.py
# Drop-in replacement for Azure Key Vault client, wraps GCP Secret Manager with caching and retry
import time
import logging
from typing import Optional, Dict
from cachetools import TTLCache, cached
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from google.cloud import secretmanager
from google.api_core import exceptions as gcp_exceptions

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class GCPSecretManagerClient:
    """Client for GCP Secret Manager with caching, retry, and unified error handling."""

    def __init__(self, project_id: str, cache_ttl: int = 300, max_retries: int = 3):
        """
        Initialize client.

        Args:
            project_id: GCP project ID
            cache_ttl: Cache TTL in seconds (default 5 minutes)
            max_retries: Maximum number of retry attempts for failed requests
        """
        self.project_id = project_id
        self.client = secretmanager.SecretManagerServiceClient()
        self.parent = f"projects/{project_id}"
        self.cache = TTLCache(maxsize=1024, ttl=cache_ttl)
        self.max_retries = max_retries

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=60),
        retry=retry_if_exception_type(
            (gcp_exceptions.ServiceUnavailable, gcp_exceptions.ResourceExhausted)
        ),
        reraise=True
    )
    def _get_secret_from_api(self, secret_name: str) -> str:
        """Fetch secret value from GCP API with retry logic."""
        secret_path = f"{self.parent}/secrets/{secret_name}/versions/latest"
        try:
            response = self.client.access_secret_version(request={"name": secret_path})
            return response.payload.data.decode("UTF-8")
        except gcp_exceptions.NotFound:
            logger.error(f"Secret {secret_name} not found")
            raise
        except gcp_exceptions.PermissionDenied:
            logger.error(f"Permission denied for secret {secret_name}")
            raise
        except gcp_exceptions.GoogleAPIError as e:
            logger.error(f"GCP API error for {secret_name}: {e}")
            raise

    @cached(cache=lambda self: self.cache, key=lambda self, secret_name: secret_name)
    def get_secret(self, secret_name: str) -> str:
        """
        Get secret value with caching and retry.

        Args:
            secret_name: Name of the secret to fetch

        Returns:
            Secret value as string

        Raises:
            ValueError: If secret not found or access denied
        """
        try:
            return self._get_secret_from_api(secret_name)
        except Exception as e:
            logger.error(f"Failed to get secret {secret_name}: {e}")
            raise ValueError(f"Could not retrieve secret {secret_name}") from e

    def set_secret(self, secret_name: str, secret_value: str) -> None:
        """
        Create or update a secret in GCP Secret Manager.

        Args:
            secret_name: Name of the secret to create/update
            secret_value: Value of the secret
        """
        # Check if secret exists
        try:
            self.client.get_secret(secret_path=f"{self.parent}/secrets/{secret_name}")
            # Add new version if secret exists
            secret_path = f"{self.parent}/secrets/{secret_name}"
            payload = secretmanager.SecretPayload(data=secret_value.encode("UTF-8"))
            self.client.add_secret_version(parent=secret_path, payload=payload)
            logger.info(f"Added new version for secret {secret_name}")
        except gcp_exceptions.NotFound:
            # Create new secret if not exists
            secret = {"replication": {"automatic": True}}
            self.client.create_secret(parent=self.parent, secret_id=secret_name, secret=secret)
            secret_path = f"{self.parent}/secrets/{secret_name}"
            payload = secretmanager.SecretPayload(data=secret_value.encode("UTF-8"))
            self.client.add_secret_version(parent=secret_path, payload=payload)
            logger.info(f"Created new secret {secret_name}")
        except gcp_exceptions.GoogleAPIError as e:
            logger.error(f"Failed to set secret {secret_name}: {e}")
            raise

    def delete_secret(self, secret_name: str) -> None:
        """Delete a secret from GCP Secret Manager."""
        secret_path = f"{self.parent}/secrets/{secret_name}"
        try:
            self.client.delete_secret(name=secret_path)
            # Invalidate cache
            if secret_name in self.cache:
                del self.cache[secret_name]
            logger.info(f"Deleted secret {secret_name}")
        except gcp_exceptions.NotFound:
            logger.warning(f"Secret {secret_name} not found, skipping deletion")
        except gcp_exceptions.GoogleAPIError as e:
            logger.error(f"Failed to delete secret {secret_name}: {e}")
            raise

# cost_analyzer.py
# Pulls billing data from Azure and GCP to calculate secret management costs
import os
import logging
from datetime import datetime, timedelta
from typing import Dict, List
from dotenv import load_dotenv
from azure.identity import DefaultAzureCredential
from azure.mgmt.costmanagement import CostManagementClient
from google.cloud import billing_v1
from google.cloud.billing_v1.services.cloud_billing import CloudBillingClient
from google.api_core import exceptions as gcp_exceptions

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
load_dotenv()

AZURE_SUBSCRIPTION_ID = os.getenv("AZURE_SUBSCRIPTION_ID")
GCP_BILLING_ACCOUNT_ID = os.getenv("GCP_BILLING_ACCOUNT_ID")
LOOKBACK_DAYS = int(os.getenv("LOOKBACK_DAYS", 30))

def get_azure_secret_costs() -> float:
    """Fetch Azure Key Vault costs for the past LOOKBACK_DAYS days."""
    if not AZURE_SUBSCRIPTION_ID:
        raise ValueError("AZURE_SUBSCRIPTION_ID not set")
    credential = DefaultAzureCredential()
    client = CostManagementClient(credential, AZURE_SUBSCRIPTION_ID)

    # Define time range
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(days=LOOKBACK_DAYS)
    time_range = f"{start_time.strftime('%Y-%m-%d')}T00:00:00Z/{end_time.strftime('%Y-%m-%d')}T00:00:00Z"

    # Query for Key Vault costs
    query = {
        "type": "ActualCost",
        "timeframe": "Custom",
        "timeperiod": {
            "from": start_time.strftime("%Y-%m-%d"),
            "to": end_time.strftime("%Y-%m-%d")
        },
        "dataset": {
            "granularity": "None",
            "filter": {
                "dimensions": {
                    "name": "ServiceName",
                    "operator": "In",
                    "values": ["Key Vault"]
                }
            },
            "aggregation": {
                "totalCost": {
                    "name": "Cost",
                    "function": "Sum"
                }
            }
        }
    }

    try:
        result = client.query.usage(resource_scope=f"/subscriptions/{AZURE_SUBSCRIPTION_ID}", parameters=query)
        total_cost = result.properties.rows[0][0] if result.properties.rows else 0.0
        logger.info(f"Azure Key Vault cost (past {LOOKBACK_DAYS} days): ${total_cost:.2f}")
        return float(total_cost)
    except Exception as e:
        logger.error(f"Failed to fetch Azure costs: {e}")
        return 0.0

def get_gcp_secret_costs() -> float:
    """Fetch GCP Secret Manager costs for the past LOOKBACK_DAYS days."""
    if not GCP_BILLING_ACCOUNT_ID:
        raise ValueError("GCP_BILLING_ACCOUNT_ID not set")
    client = CloudBillingClient()

    # Define time range
    end_time = datetime.utcnow()
    start_time = end_time - timedelta(days=LOOKBACK_DAYS)

    # Query for Secret Manager costs
    try:
        # List all billing accounts (simplified for example)
        billing_account_path = f"billingAccounts/{GCP_BILLING_ACCOUNT_ID}"
        # Note: Full GCP billing query requires BigQuery export, this is a simplified version
        # that uses the Cloud Billing API to list services and filter for Secret Manager
        services = client.list_services(parent=billing_account_path)
        secret_manager_service = None
        for service in services:
            if "Secret Manager" in service.display_name:
                secret_manager_service = service
                break
        if not secret_manager_service:
            logger.warning("Secret Manager service not found in GCP billing")
            return 0.0

        # Simplified cost calculation: assume $0.03 per 10k ops, 1M ops/month = $3
        # In production, use BigQuery billing export for accurate data
        logger.info(f"GCP Secret Manager cost (past {LOOKBACK_DAYS} days): Estimated via per-op pricing")
        return 0.0  # Replace with actual BigQuery query in production
    except gcp_exceptions.GoogleAPIError as e:
        logger.error(f"Failed to fetch GCP costs: {e}")
        return 0.0

def generate_report(azure_cost: float, gcp_cost: float) -> None:
    """Generate cost comparison report."""
    total_cost = azure_cost + gcp_cost
    savings = azure_cost - gcp_cost  # If we replaced Azure with GCP
    savings_pct = (savings / azure_cost) * 100 if azure_cost > 0 else 0

    print("\n=== Secret Management Cost Report ===")
    print(f"Time Period: Past {LOOKBACK_DAYS} days")
    print(f"Azure Key Vault Cost: ${azure_cost:.2f}")
    print(f"GCP Secret Manager Cost: ${gcp_cost:.2f}")
    print(f"Total Cost: ${total_cost:.2f}")
    if savings > 0:
        print(f"Potential Savings (Replace Azure with GCP): ${savings:.2f} ({savings_pct:.1f}%)")
    else:
        print(f"GCP is more expensive by ${-savings:.2f}")

def main() -> None:
    azure_cost = get_azure_secret_costs()
    gcp_cost = get_gcp_secret_costs()
    generate_report(azure_cost, gcp_cost)

if __name__ == "__main__":
    main()

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=60),
    retry=retry_if_exception_type((gcp_exceptions.ServiceUnavailable,))
)
def get_secret_with_retry(client, secret_name):
    return client.get_secret(secret_name)

from cachetools import TTLCache, cached

cache = TTLCache(maxsize=1024, ttl=300)  # 5 minute TTL

@cached(cache, key=lambda secret_name: secret_name)
def get_cached_secret(client, secret_name):
    return client.get_secret(secret_name)

-- BigQuery query to find unused GCP secrets
SELECT
  resource.labels.secret_id,
  MAX(timestamp) AS last_accessed
FROM
  `project.dataset.cloud_audit_logs`
WHERE
  resource.type = "secretmanager.googleapis.com/Secret"
  AND timestamp >= DATE_SUB(CURRENT_DATE(), INTERVAL 90 DAY)
GROUP BY
  resource.labels.secret_id
HAVING
  last_accessed < DATE_SUB(CURRENT_DATE(), INTERVAL 90 DAY)