150 lines
4.8 KiB
Python
150 lines
4.8 KiB
Python
"""
|
|
Azure Authentication Module
|
|
|
|
Handles authentication to Azure services for Key Vault management.
|
|
"""
|
|
|
|
from azure.identity import InteractiveBrowserCredential
|
|
from azure.mgmt.resource import SubscriptionClient
|
|
from typing import Optional, List, Dict
|
|
|
|
|
|
class AzureAuthenticator:
|
|
"""Handles Azure authentication for Key Vault and resource management."""
|
|
|
|
def __init__(self):
|
|
"""
|
|
Initialize the Azure authenticator.
|
|
"""
|
|
self.tenant_id: Optional[str] = None
|
|
self.subscription_id: Optional[str] = None
|
|
self.credential: Optional[InteractiveBrowserCredential] = None
|
|
self.subscriptions: List[Dict[str, str]] = []
|
|
|
|
def authenticate(self, credential: InteractiveBrowserCredential = None) -> bool:
|
|
"""
|
|
Authenticate to Azure. Can reuse credential from Graph authenticator.
|
|
|
|
Args:
|
|
credential: Optional credential to reuse (from GraphAuthenticator)
|
|
|
|
Returns:
|
|
bool: True if authentication succeeded, False otherwise
|
|
|
|
Raises:
|
|
Exception: If authentication fails
|
|
"""
|
|
try:
|
|
if credential:
|
|
# Reuse credential from Graph authentication
|
|
self.credential = credential
|
|
else:
|
|
# Create new interactive browser credential with "organizations" tenant
|
|
# This allows the user to login with any organizational account
|
|
# additionally_allowed_tenants="*" allows acquiring tokens for any tenant
|
|
self.credential = InteractiveBrowserCredential(
|
|
tenant_id="organizations",
|
|
additionally_allowed_tenants=["*"]
|
|
)
|
|
|
|
# List all subscriptions (this will use the cached token from Graph auth)
|
|
sub_client = SubscriptionClient(self.credential)
|
|
subscriptions_list = list(sub_client.subscriptions.list())
|
|
|
|
# Extract tenant ID from the first subscription
|
|
if subscriptions_list:
|
|
# Get tenant ID from subscription (format: /subscriptions/{sub-id})
|
|
# The tenant info is in the subscription object
|
|
first_sub = subscriptions_list[0]
|
|
self.tenant_id = first_sub.tenant_id if hasattr(first_sub, 'tenant_id') else None
|
|
if self.tenant_id:
|
|
print(f"Detected Tenant ID: {self.tenant_id}")
|
|
|
|
if subscriptions_list:
|
|
print(f"Successfully authenticated to Azure. Found {len(subscriptions_list)} subscription(s).")
|
|
|
|
# Store subscriptions for later selection
|
|
self.subscriptions = [
|
|
{
|
|
'id': sub.subscription_id,
|
|
'name': sub.display_name
|
|
}
|
|
for sub in subscriptions_list
|
|
]
|
|
|
|
return True
|
|
|
|
return False
|
|
|
|
except Exception as e:
|
|
raise Exception(f"Azure authentication failed: {str(e)}")
|
|
|
|
def get_credential(self) -> InteractiveBrowserCredential:
|
|
"""
|
|
Get the credential object.
|
|
|
|
Returns:
|
|
InteractiveBrowserCredential: The credential object
|
|
|
|
Raises:
|
|
Exception: If not authenticated
|
|
"""
|
|
if not self.credential:
|
|
raise Exception("Not authenticated. Call authenticate() first.")
|
|
return self.credential
|
|
|
|
def get_subscriptions(self) -> List[Dict[str, str]]:
|
|
"""
|
|
Get the list of available subscriptions.
|
|
|
|
Returns:
|
|
List[Dict]: List of subscriptions with 'id' and 'name'
|
|
"""
|
|
return self.subscriptions
|
|
|
|
def set_subscription(self, subscription_id: str):
|
|
"""
|
|
Set the active subscription ID.
|
|
|
|
Args:
|
|
subscription_id: The subscription ID to use
|
|
"""
|
|
self.subscription_id = subscription_id
|
|
|
|
def get_subscription_id(self) -> str:
|
|
"""
|
|
Get the subscription ID.
|
|
|
|
Returns:
|
|
str: The subscription ID
|
|
|
|
Raises:
|
|
Exception: If subscription not set
|
|
"""
|
|
if not self.subscription_id:
|
|
raise Exception("Subscription not set. Call set_subscription() first.")
|
|
return self.subscription_id
|
|
|
|
def get_tenant_id(self) -> str:
|
|
"""
|
|
Get the tenant ID.
|
|
|
|
Returns:
|
|
str: The tenant ID
|
|
|
|
Raises:
|
|
Exception: If not authenticated
|
|
"""
|
|
if not self.tenant_id:
|
|
raise Exception("Tenant ID not available. Call authenticate() first.")
|
|
return self.tenant_id
|
|
|
|
def is_authenticated(self) -> bool:
|
|
"""
|
|
Check if currently authenticated.
|
|
|
|
Returns:
|
|
bool: True if authenticated, False otherwise
|
|
"""
|
|
return self.credential is not None
|