Microsoft Azure Container Service Management Client Library for Python
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
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.
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
"""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
"""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
"""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
"""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]]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}")# 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"
)
]
)# 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")# 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}")# 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")# 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"
)
]
)# 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}")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}")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}")# 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"]
)
)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