AKS connection using azure sdk

aks.py

# Here's a Python program that uses the Azure SDK to connect to Azure Kubernetes Service (AKS) and checks for the presence of a namespace, configmap, and secret with specific key-value pairs:

# ```python
from azure.identity import DefaultAzureCredential
from azure.mgmt.containerservice import ContainerServiceClient
from kubernetes import client, config
from kubernetes.client.rest import ApiException

def check_aks_resources(resource_group, cluster_name, subscription_id):
    # Authenticate with Azure
    credential = DefaultAzureCredential()
    
    # Create a ContainerServiceClient
    aks_client = ContainerServiceClient(credential, subscription_id)
    
    # Get AKS cluster credentials
    credentials = aks_client.managed_clusters.list_cluster_admin_credentials(resource_group, cluster_name)
    kubeconfig = credentials.kubeconfigs[0].value
    
    # Load kubeconfig
    config.load_kube_config_from_dict(kubeconfig)
    
    # Create Kubernetes API client
    v1 = client.CoreV1Api()
    
    # Check namespace
    try:
        v1.read_namespace("dev")
        print("Namespace 'dev' exists.")
    except ApiException as e:
        if e.status == 404:
            print("Namespace 'dev' does not exist.")
        else:
            print(f"Error checking namespace: {e}")
    
    # Check configmap
    try:
        configmap = v1.read_namespaced_config_map("dev-configmap", "dev")
        if configmap.data.get("env") == "development":
            print("ConfigMap with key-value pair 'env=development' exists.")
        else:
            print("ConfigMap exists but doesn't have the correct key-value pair.")
    except ApiException as e:
        if e.status == 404:
            print("ConfigMap 'dev-configmap' does not exist in 'dev' namespace.")
        else:
            print(f"Error checking configmap: {e}")
    
    # Check secret
    try:
        secret = v1.read_namespaced_secret("dev-secret", "dev")
        if "dev_user" in secret.data and "dev_pass" in secret.data:
            print("Secret with 'dev_user' and 'dev_pass' exists.")
        else:
            print("Secret exists but doesn't have the required keys.")
    except ApiException as e:
        if e.status == 404:
            print("Secret 'dev-secret' does not exist in 'dev' namespace.")
        else:
            print(f"Error checking secret: {e}")

"""
# Replace these with your actual values
resource_group = "your-resource-group"
cluster_name = "your-aks-cluster-name"
subscription_id = "your-subscription-id"

check_aks_resources(resource_group, cluster_name, subscription_id)
```

This program does the following:

1. It uses the Azure Identity library to authenticate with Azure.
2. It creates a ContainerServiceClient to interact with AKS.
3. It retrieves the AKS cluster credentials.
4. It loads the Kubernetes configuration from the retrieved credentials.
5. It creates a Kubernetes API client.
6. It checks for the existence of the "dev" namespace.
7. It checks for a ConfigMap named "dev-configmap" in the "dev" namespace with the key-value pair "env=development".
8. It checks for a Secret named "dev-secret" in the "dev" namespace with keys "dev_user" and "dev_pass".

To use this program, you'll need to install the required libraries:

```
pip install azure-identity azure-mgmt-containerservice kubernetes
```

Also, make sure to replace the placeholder values for `resource_group`, `cluster_name`, and `subscription_id` with your actual Azure resource information.
"""

createKubeConfig.py

import yaml
from azure.identity import ClientSecretCredential
from azure.mgmt.containerservice import ContainerServiceClient
import base64

def create_azure_kubeconfig(credential, subscription_id, resource_group, cluster_name, output_file='kubeconfig'):
    # Create a ContainerServiceClient
    client = ContainerServiceClient(credential, subscription_id)

    # Get the cluster credentials
    credentials = client.managed_clusters.list_cluster_user_credentials(resource_group, cluster_name)
    kubeconfig = yaml.safe_load(credentials.kubeconfigs[0].value.decode('utf-8'))

    # Update the kubeconfig to use Azure AD authentication
    for user in kubeconfig['users']:
        user['user'] = {
            'exec': {
                'apiVersion': 'client.authentication.k8s.io/v1beta1',
                'command': 'kubelogin',
                'args': [
                    'get-token',
                    '--environment',
                    'AzurePublicCloud',
                    '--server-id',
                    '6dae42f8-4368-4678-94ff-3960e28e3630',  # Azure Kubernetes Service AAD Server
                    '--client-id',
                    credential.client_id,
                    '--client-secret',
                    credential.client_secret,
                    '--tenant-id',
                    credential.tenant_id
                ]
            }
        }

    # Write the updated kubeconfig to a file
    with open(output_file, 'w') as f:
        yaml.dump(kubeconfig, f)

    print(f"Kubeconfig file created: {output_file}")

# Example usage
subscription_id = 'your-subscription-id'
resource_group = 'your-resource-group'
cluster_name = 'your-aks-cluster-name'

# Assuming you already have the credential object
# credential = ClientSecretCredential(tenant_id, client_id, client_secret)

create_azure_kubeconfig(credential, subscription_id, resource_group, cluster_name)

inMemoryKubeConfig.py

from azure.identity import ClientSecretCredential
from azure.mgmt.containerservice import ContainerServiceClient
from kubernetes import client, config
import base64
import yaml

# Define your variables
subscription_id = 'your_subscription_id'
resource_group_name = 'your_resource_group_name'
aks_cluster_name = 'your_aks_cluster_name'

# Set up SPN authentication using ClientSecretCredential
credential = ClientSecretCredential(
    tenant_id='your_tenant_id',
    client_id='your_client_id',
    client_secret='your_client_secret'
)

# Create a ContainerServiceClient instance
container_client = ContainerServiceClient(credential, subscription_id)

# Get the AKS cluster's access credentials
aks_credentials = container_client.managed_clusters.list_cluster_admin_credentials(
    resource_group_name, aks_cluster_name
)

# Extract the kubeconfig content from the response
kubeconfig_base64 = aks_credentials.kubeconfigs[0].value
kubeconfig_content = base64.b64decode(kubeconfig_base64).decode()

# Parse the kubeconfig YAML and modify it to use the SPN credentials
kubeconfig_dict = yaml.safe_load(kubeconfig_content)

# Ensure that the user section corresponds to the SPN credentials
for user in kubeconfig_dict['users']:
    # This sets the context to the client_id and client_secret for the SPN
    user['user']['exec'] = {
        'apiVersion': 'client.authentication.k8s.io/v1beta1',
        'command': 'kubectl',
        'args': ['--token'],
        'env': [
            {'name': 'AZURE_CLIENT_ID', 'value': 'your_client_id'},
            {'name': 'AZURE_CLIENT_SECRET', 'value': 'your_client_secret'},
            {'name': 'AZURE_TENANT_ID', 'value': 'your_tenant_id'}
        ]
    }

# Load the modified kubeconfig directly into the Kubernetes client
config.load_kube_config_from_dict(kubeconfig_dict)

# Now you can use the Kubernetes Python client to interact with the AKS cluster
v1 = client.CoreV1Api()
print("Listing pods in default namespace:")
pods = v1.list_namespaced_pod(namespace='default')
for pod in pods.items:
    print(f"Pod name: {pod.metadata.name}")

loadKubeConfigFromClientCreds.py

from kubernetes import client, config
from kubernetes.client import Configuration
from azure.identity import ClientSecretCredential
from azure.mgmt.containerservice import ContainerServiceClient
import yaml

class AzureKubeConfig(Configuration):
    def __init__(self, credential, subscription_id, resource_group, cluster_name):
        Configuration.__init__(self)
        
        # Get the cluster info
        aks_client = ContainerServiceClient(credential, subscription_id)
        cluster_info = aks_client.managed_clusters.get(resource_group, cluster_name)
        
        # Set the host
        self.host = cluster_info.fqdn
        
        # Set up the API key authorization using the credential
        self.api_key = {
            'authorization': 'Bearer ' + credential.get_token("https://management.azure.com/.default").token
        }
        self.api_key_prefix = {'authorization': 'Bearer'}

def load_azure_kube_config(credential, subscription_id, resource_group, cluster_name):
    # Create the custom configuration
    azure_config = AzureKubeConfig(credential, subscription_id, resource_group, cluster_name)
    
    # Set this as the default configuration
    Configuration.set_default(azure_config)

# Example usage
subscription_id = 'your-subscription-id'
resource_group = 'your-resource-group'
cluster_name = 'your-aks-cluster-name'

tenant_id = 'your-tenant-id'
client_id = 'your-client-id'
client_secret = 'your-client-secret'

credential = ClientSecretCredential(tenant_id, client_id, client_secret)

# Load the configuration
load_azure_kube_config(credential, subscription_id, resource_group, cluster_name)

# Now you can use the client to interact with your Kubernetes cluster
v1 = client.CoreV1Api()
print("Listing pods with their IPs:")
ret = v1.list_pod_for_all_namespaces(watch=False)
for i in ret.items:
    print(f"{i.status.pod_ip}\t{i.metadata.namespace}\t{i.metadata.name}")