CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-azure-mgmt-containerservice

Microsoft Azure Container Service Management Client Library for Python

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview
Eval results
Files

private-endpoints.mddocs/

Private Endpoints

Private networking configuration for AKS clusters including private endpoint connections, private link resources, network security management, and secure communication patterns. This enables AKS clusters to operate within private networks without exposure to public internet, enhancing security and compliance for enterprise workloads.

Capabilities

Private Endpoint Connection Management

Manage private endpoint connections for secure, private communication with AKS clusters over Azure's backbone network.

def update(
    resource_group_name: str,
    resource_name: str,
    private_endpoint_connection_name: str,
    parameters: PrivateEndpointConnection
) -> PrivateEndpointConnection:
    """
    Updates a private endpoint connection.
    
    Args:
        resource_group_name: Azure resource group name
        resource_name: Cluster name
        private_endpoint_connection_name: Private endpoint connection name
        parameters: Updated connection configuration
        
    Returns:
        PrivateEndpointConnection: Updated connection details
    """

def begin_delete(
    resource_group_name: str,
    resource_name: str,
    private_endpoint_connection_name: str
) -> LROPoller[None]:
    """
    Deletes a private endpoint connection.
    
    Args:
        resource_group_name: Azure resource group name
        resource_name: Cluster name
        private_endpoint_connection_name: Private endpoint connection name
        
    Returns:
        LROPoller[None]: Long-running operation poller
    """

Private Endpoint Connection Discovery

List and retrieve private endpoint connections to understand current network connectivity.

def list(resource_group_name: str, resource_name: str) -> ItemPaged[PrivateEndpointConnection]:
    """
    Lists private endpoint connections for the cluster.
    
    Args:
        resource_group_name: Azure resource group name
        resource_name: Cluster name
        
    Returns:
        ItemPaged[PrivateEndpointConnection]: Paginated connection list
    """

def get(
    resource_group_name: str,
    resource_name: str,
    private_endpoint_connection_name: str
) -> PrivateEndpointConnection:
    """
    Gets a specific private endpoint connection.
    
    Args:
        resource_group_name: Azure resource group name
        resource_name: Cluster name
        private_endpoint_connection_name: Private endpoint connection name
        
    Returns:
        PrivateEndpointConnection: Connection configuration and status
    """

Private Link Resource Management

Manage private link resources that define the linkable resources within the AKS cluster.

def list_private_link_resources(
    resource_group_name: str,
    resource_name: str
) -> PrivateLinkResourcesListResult:
    """
    Lists private link resources for the cluster.
    
    Args:
        resource_group_name: Azure resource group name
        resource_name: Cluster name
        
    Returns:
        PrivateLinkResourcesListResult: Available private link resources
    """

Private Link Service ID Resolution

Resolve private link service IDs for network configuration and connectivity planning.

def resolve_private_link_service_id(
    resource_group_name: str,
    resource_name: str,
    parameters: PrivateLinkResource
) -> PrivateLinkResource:
    """
    Resolves private link service ID for the cluster.
    
    Args:
        resource_group_name: Azure resource group name
        resource_name: Cluster name
        parameters: Private link resource configuration
        
    Returns:
        PrivateLinkResource: Resolved private link service details
    """

Core Types

class PrivateEndpointConnection:
    """
    Private endpoint connection configuration and status.
    
    Attributes:
        id: Connection resource ID (read-only)
        name: Connection name (read-only)
        type: Resource type (read-only)
        private_endpoint: Private endpoint details
        private_link_service_connection_state: Connection state
        provisioning_state: Current provisioning state (read-only)
    """
    id: Optional[str]
    name: Optional[str]
    type: Optional[str]
    private_endpoint: Optional[PrivateEndpoint]
    private_link_service_connection_state: Optional[PrivateLinkServiceConnectionState]
    provisioning_state: Optional[str]

class PrivateEndpoint:
    """
    Private endpoint details for network connectivity.
    
    Attributes:
        id: Private endpoint resource ID
    """
    id: Optional[str]

class PrivateLinkServiceConnectionState:
    """
    Private link service connection state and configuration.
    
    Attributes:
        status: Connection status ('Pending', 'Approved', 'Rejected', 'Disconnected')
        description: Connection state description
        actions_required: Required actions for connection
    """
    status: Optional[str]
    description: Optional[str]
    actions_required: Optional[str]

class PrivateLinkResource:
    """
    Private link resource definition.
    
    Attributes:
        id: Resource ID (read-only)
        name: Resource name (read-only)
        type: Resource type (read-only)
        group_id: Private link resource group identifier
        required_members: Required member names for private link
        private_link_service_id: Private link service ID
    """
    id: Optional[str]
    name: Optional[str] 
    type: Optional[str]
    group_id: Optional[str]
    required_members: Optional[List[str]]
    private_link_service_id: Optional[str]

class PrivateLinkResourcesListResult:
    """
    List of private link resources.
    
    Attributes:
        value: List of private link resources
    """
    value: Optional[List[PrivateLinkResource]]

Usage Examples

Enable Private Cluster

from azure.identity import DefaultAzureCredential
from azure.mgmt.containerservice import ContainerServiceClient
from azure.mgmt.containerservice.models import (
    ManagedCluster,
    ManagedClusterAPIServerAccessProfile,
    ManagedClusterAgentPoolProfile
)

credential = DefaultAzureCredential()
client = ContainerServiceClient(credential, subscription_id)

# Create private AKS cluster
private_cluster = ManagedCluster(
    location="eastus",
    api_server_access_profile=ManagedClusterAPIServerAccessProfile(
        enable_private_cluster=True,
        private_dns_zone="system",  # Uses system-managed private DNS zone
        enable_private_cluster_public_fqdn=False  # No public FQDN
    ),
    agent_pool_profiles=[
        ManagedClusterAgentPoolProfile(
            name="nodepool1",
            count=3,
            vm_size="Standard_DS2_v2",
            os_type="Linux",
            mode="System"
        )
    ]
)

# Create cluster
operation = client.managed_clusters.begin_create_or_update(
    "myResourceGroup",
    "myPrivateCluster",
    private_cluster
)
result = operation.result()
print(f"Private cluster created: {result.name}")

Custom Private DNS Zone

# Private cluster with custom DNS zone
custom_dns_cluster = ManagedCluster(
    location="eastus",
    api_server_access_profile=ManagedClusterAPIServerAccessProfile(
        enable_private_cluster=True,
        private_dns_zone="/subscriptions/sub-id/resourceGroups/dns-rg/providers/Microsoft.Network/privateDnsZones/privatelink.eastus.azmk8s.io",
        authorized_ip_ranges=[
            "10.0.0.0/8",    # Corporate network
            "172.16.0.0/12"  # VPN range
        ]
    ),
    agent_pool_profiles=[
        ManagedClusterAgentPoolProfile(
            name="nodepool1",
            count=3,
            vm_size="Standard_DS2_v2",
            os_type="Linux",
            mode="System",
            vnet_subnet_id="/subscriptions/sub-id/resourceGroups/network-rg/providers/Microsoft.Network/virtualNetworks/myVNet/subnets/aks-subnet"
        )
    ]
)

Manage Private Endpoint Connections

# List all private endpoint connections
connections = client.private_endpoint_connections.list(
    "myResourceGroup",
    "myPrivateCluster"
)

for conn in connections:
    print(f"Connection: {conn.name}")
    if conn.private_link_service_connection_state:
        state = conn.private_link_service_connection_state
        print(f"  Status: {state.status}")
        print(f"  Description: {state.description}")

# Get specific connection
connection = client.private_endpoint_connections.get(
    "myResourceGroup", 
    "myPrivateCluster",
    "connection-name"
)

# Approve pending connection
if connection.private_link_service_connection_state.status == "Pending":
    from azure.mgmt.containerservice.models import (
        PrivateEndpointConnection,
        PrivateLinkServiceConnectionState
    )
    
    connection.private_link_service_connection_state = PrivateLinkServiceConnectionState(
        status="Approved",
        description="Approved for production use"
    )
    
    updated_connection = client.private_endpoint_connections.update(
        "myResourceGroup",
        "myPrivateCluster", 
        "connection-name",
        connection
    )
    print("Connection approved")

Private Link Resource Discovery

# Discover available private link resources
private_resources = client.private_link_resources.list(
    "myResourceGroup",
    "myPrivateCluster"  
)

for resource in private_resources.value:
    print(f"Private link resource: {resource.name}")
    print(f"  Group ID: {resource.group_id}")
    print(f"  Required members: {resource.required_members}")
    if resource.private_link_service_id:
        print(f"  Service ID: {resource.private_link_service_id}")

# Resolve private link service ID
from azure.mgmt.containerservice.models import PrivateLinkResource

resolve_request = PrivateLinkResource(
    group_id="management"
)

resolved = client.resolve_private_link_service_id.post(
    "myResourceGroup",
    "myPrivateCluster",
    resolve_request
)
print(f"Resolved service ID: {resolved.private_link_service_id}")

Network Security Configuration

# Configure authorized IP ranges for private cluster
authorized_ranges = [
    "10.0.0.0/8",        # Corporate network
    "192.168.0.0/16",    # Office networks  
    "172.16.0.0/12",     # VPN ranges
    "20.0.0.0/24"        # Management subnet
]

# Update cluster with IP restrictions
cluster = client.managed_clusters.get("myRG", "myPrivateCluster")
cluster.api_server_access_profile.authorized_ip_ranges = authorized_ranges

update_op = client.managed_clusters.begin_create_or_update(
    "myRG", "myPrivateCluster", cluster
)
update_op.wait()
print("Authorized IP ranges updated")

Hybrid Connectivity Setup

# Private cluster with public FQDN for hybrid scenarios
hybrid_cluster = ManagedCluster(
    location="eastus",
    api_server_access_profile=ManagedClusterAPIServerAccessProfile(
        enable_private_cluster=True,
        enable_private_cluster_public_fqdn=True,  # Keep public FQDN
        private_dns_zone="system",
        authorized_ip_ranges=[
            "203.0.113.0/24"  # Office public IP range
        ]
    ),
    agent_pool_profiles=[
        ManagedClusterAgentPoolProfile(
            name="nodepool1", 
            count=3,
            vm_size="Standard_DS2_v2",
            os_type="Linux",
            mode="System"
        )
    ]
)

Multi-Region Private Connectivity

# Setup private connectivity across regions
regions = ["eastus", "westus", "northeurope"]
dns_zones = {
    "eastus": "/subscriptions/sub/resourceGroups/dns/providers/Microsoft.Network/privateDnsZones/privatelink.eastus.azmk8s.io",
    "westus": "/subscriptions/sub/resourceGroups/dns/providers/Microsoft.Network/privateDnsZones/privatelink.westus.azmk8s.io", 
    "northeurope": "/subscriptions/sub/resourceGroups/dns/providers/Microsoft.Network/privateDnsZones/privatelink.northeurope.azmk8s.io"
}

for region in regions:
    cluster = ManagedCluster(
        location=region,
        api_server_access_profile=ManagedClusterAPIServerAccessProfile(
            enable_private_cluster=True,
            private_dns_zone=dns_zones[region],
            enable_private_cluster_public_fqdn=False
        ),
        agent_pool_profiles=[
            ManagedClusterAgentPoolProfile(
                name="nodepool1",
                count=3, 
                vm_size="Standard_DS2_v2",
                os_type="Linux",
                mode="System",
                vnet_subnet_id=f"/subscriptions/sub/resourceGroups/{region}-network/providers/Microsoft.Network/virtualNetworks/{region}-vnet/subnets/aks-subnet"
            )
        ]
    )
    
    operation = client.managed_clusters.begin_create_or_update(
        f"{region}-rg",
        f"private-cluster-{region}",
        cluster
    )
    print(f"Creating private cluster in {region}")

Connection State Management

def manage_private_endpoint_connection(client, rg, cluster_name, connection_name, action):
    """Manage private endpoint connection state."""
    
    connection = client.private_endpoint_connections.get(rg, cluster_name, connection_name)
    
    from azure.mgmt.containerservice.models import PrivateLinkServiceConnectionState
    
    if action == "approve":
        connection.private_link_service_connection_state = PrivateLinkServiceConnectionState(
            status="Approved",
            description="Approved by administrator"
        )
    elif action == "reject":
        connection.private_link_service_connection_state = PrivateLinkServiceConnectionState(
            status="Rejected", 
            description="Rejected due to policy violation"
        )
    elif action == "disconnect":
        # Delete the connection
        delete_op = client.private_endpoint_connections.begin_delete(
            rg, cluster_name, connection_name
        )
        delete_op.wait()
        return "Disconnected"
    
    if action in ["approve", "reject"]:
        updated = client.private_endpoint_connections.update(
            rg, cluster_name, connection_name, connection
        )
        return updated.private_link_service_connection_state.status

# Usage
status = manage_private_endpoint_connection(
    client, "myRG", "myCluster", "connection1", "approve"
)
print(f"Connection status: {status}")

Error Handling

from azure.core.exceptions import HttpResponseError, ResourceNotFoundError

try:
    connection = client.private_endpoint_connections.get(
        "myRG", "myCluster", "nonexistent-connection"
    )
except ResourceNotFoundError:
    print("Private endpoint connection not found")
except HttpResponseError as e:
    if e.status_code == 403:
        print("Insufficient permissions to access private endpoint")
    else:
        print(f"HTTP error: {e.status_code}")

# Handle private cluster configuration errors
try:
    operation = client.managed_clusters.begin_create_or_update(
        "myRG", "myCluster", private_cluster
    )
    result = operation.result()
except HttpResponseError as e:
    if "PrivateClusterNotSupported" in str(e):
        print("Private clusters not supported in this region")
    elif "InvalidPrivateDNSZone" in str(e):
        print("Invalid private DNS zone configuration")
    else:
        print(f"Cluster creation failed: {e}")

Best Practices

Network Security

# Implement defense in depth for private clusters
security_config = {
    # Network segmentation
    "subnet_nsg": "/subscriptions/sub/resourceGroups/security/providers/Microsoft.Network/networkSecurityGroups/aks-nsg",
    
    # Authorized IP ranges (least privilege)
    "authorized_ranges": [
        "10.1.0.0/16",      # Management subnet only
        "10.2.100.0/24"     # Jump box subnet
    ],
    
    # Private DNS configuration
    "private_dns_zone": "system",  # Use system-managed zone
    
    # Disable public FQDN
    "enable_public_fqdn": False
}

# Apply security configuration
secure_cluster = ManagedCluster(
    location="eastus",
    api_server_access_profile=ManagedClusterAPIServerAccessProfile(
        enable_private_cluster=True,
        private_dns_zone=security_config["private_dns_zone"],
        enable_private_cluster_public_fqdn=security_config["enable_public_fqdn"],
        authorized_ip_ranges=security_config["authorized_ranges"]
    )
)

Connection Monitoring

def monitor_private_connections(client, rg, cluster_name):
    """Monitor private endpoint connection health."""
    
    connections = client.private_endpoint_connections.list(rg, cluster_name)
    
    health_report = {
        "total_connections": 0,
        "approved": 0,
        "pending": 0, 
        "rejected": 0,
        "disconnected": 0
    }
    
    for conn in connections:
        health_report["total_connections"] += 1
        
        if conn.private_link_service_connection_state:
            status = conn.private_link_service_connection_state.status.lower()
            if status in health_report:
                health_report[status] += 1
    
    return health_report

# Usage
health = monitor_private_connections(client, "myRG", "myCluster")
print(f"Connection health: {health}")

Install with Tessl CLI

npx tessl i tessl/pypi-azure-mgmt-containerservice

docs

agent-pools.md

index.md

machines.md

maintenance.md

managed-clusters.md

models.md

private-endpoints.md

snapshots.md

trusted-access.md

tile.json