Private Networking

def begin_create_or_update(resource_group_name: str, registry_name: str, private_endpoint_connection_name: str, private_endpoint_connection: PrivateEndpointConnection, **kwargs) -> LROPoller[PrivateEndpointConnection]:
    """
    Create or update a private endpoint connection for a container registry.
    
    Parameters:
    - resource_group_name: str - Name of the resource group
    - registry_name: str - Name of the registry
    - private_endpoint_connection_name: str - Name of the private endpoint connection
    - private_endpoint_connection: PrivateEndpointConnection - Connection configuration
    
    Returns:
    LROPoller[PrivateEndpointConnection] - Long-running operation poller for the connection
    """

def begin_delete(resource_group_name: str, registry_name: str, private_endpoint_connection_name: str, **kwargs) -> LROPoller[None]:
    """
    Delete a private endpoint connection from a container registry.
    
    Parameters:
    - resource_group_name: str - Name of the resource group
    - registry_name: str - Name of the registry
    - private_endpoint_connection_name: str - Name of the connection to delete
    
    Returns:
    LROPoller[None] - Long-running operation poller
    """

def get(resource_group_name: str, registry_name: str, private_endpoint_connection_name: str, **kwargs) -> PrivateEndpointConnection:
    """
    Get properties of a private endpoint connection.
    
    Parameters:
    - resource_group_name: str - Name of the resource group
    - registry_name: str - Name of the registry
    - private_endpoint_connection_name: str - Name of the private endpoint connection
    
    Returns:
    PrivateEndpointConnection - Private endpoint connection with complete configuration
    """

def list(resource_group_name: str, registry_name: str, **kwargs) -> ItemPaged[PrivateEndpointConnection]:
    """
    List all private endpoint connections for a container registry.
    
    Parameters:
    - resource_group_name: str - Name of the resource group
    - registry_name: str - Name of the registry
    
    Returns:
    ItemPaged[PrivateEndpointConnection] - Paginated list of all private endpoint connections
    """

class PrivateEndpointConnection:
    """
    An object that represents a private endpoint connection for a container registry.
    
    Attributes:
    - id: str - Resource ID
    - name: str - Resource name
    - type: str - Resource type
    - private_endpoint: PrivateEndpoint - Private endpoint details
    - private_link_service_connection_state: PrivateLinkServiceConnectionState - Connection state
    - provisioning_state: ProvisioningState - Current provisioning state
    """

class PrivateEndpoint:
    """
    Private endpoint details.
    
    Attributes:
    - id: str - Resource ID of the private endpoint
    """

class PrivateLinkServiceConnectionState:
    """
    Connection state for private link service.
    
    Attributes:
    - status: ConnectionStatus - Connection status (Approved, Pending, Rejected, Disconnected)
    - description: str - Description of the connection state
    - actions_required: ActionsRequired - Actions required to resolve connection issues
    """

class PrivateLinkResource:
    """
    A private link resource for a container registry.
    
    Attributes:
    - id: str - Resource ID
    - name: str - Resource name
    - type: str - Resource type
    - group_id: str - Group ID of the private link resource
    - required_members: List[str] - Required members for the private link resource
    - required_zone_names: List[str] - Required DNS zone names
    """

class PrivateLinkResourceListResult:
    """
    Result of listing private link resources.
    
    Attributes:
    - value: List[PrivateLinkResource] - List of private link resources
    """

class ConnectionStatus(str, Enum):
    """Private endpoint connection status."""
    APPROVED = "Approved"
    PENDING = "Pending"
    REJECTED = "Rejected"
    DISCONNECTED = "Disconnected"

class ActionsRequired(str, Enum):
    """Actions required for private endpoint connection."""
    NONE = "None"
    RECREATE = "Recreate"

class ProvisioningState(str, Enum):
    """Provisioning state of a resource."""
    CREATING = "Creating"
    UPDATING = "Updating"
    DELETING = "Deleting"
    SUCCEEDED = "Succeeded"
    FAILED = "Failed"
    CANCELED = "Canceled"

from azure.mgmt.containerregistry import ContainerRegistryManagementClient
from azure.mgmt.containerregistry.models import (
    PrivateEndpointConnection, PrivateEndpoint, PrivateLinkServiceConnectionState,
    ConnectionStatus, ActionsRequired
)
from azure.identity import DefaultAzureCredential

client = ContainerRegistryManagementClient(
    DefaultAzureCredential(), 
    "subscription-id"
)

# Create or approve a private endpoint connection
private_endpoint_connection = PrivateEndpointConnection(
    private_endpoint=PrivateEndpoint(
        id="/subscriptions/subscription-id/resourceGroups/my-rg/providers/Microsoft.Network/privateEndpoints/my-private-endpoint"
    ),
    private_link_service_connection_state=PrivateLinkServiceConnectionState(
        status=ConnectionStatus.APPROVED,
        description="Connection approved for secure access",
        actions_required=ActionsRequired.NONE
    )
)

# Create the connection
connection_poller = client.private_endpoint_connections.begin_create_or_update(
    "my-resource-group",
    "my-registry",
    "my-private-endpoint-connection",
    private_endpoint_connection
)

connection = connection_poller.result()
print(f"Created private endpoint connection: {connection.name}")
print(f"Status: {connection.private_link_service_connection_state.status}")

# List all private endpoint connections
connections = client.private_endpoint_connections.list(
    "my-resource-group",
    "my-registry"
)

print("Private Endpoint Connections:")
print("-" * 50)
for connection in connections:
    print(f"Connection: {connection.name}")
    print(f"  Status: {connection.private_link_service_connection_state.status}")
    print(f"  Provisioning State: {connection.provisioning_state}")
    print(f"  Private Endpoint ID: {connection.private_endpoint.id}")
    print(f"  Description: {connection.private_link_service_connection_state.description}")
    if connection.private_link_service_connection_state.actions_required != ActionsRequired.NONE:
        print(f"  Actions Required: {connection.private_link_service_connection_state.actions_required}")
    print()

# Get a specific connection that might be pending
pending_connection = client.private_endpoint_connections.get(
    "my-resource-group",
    "my-registry",
    "pending-connection"
)

if pending_connection.private_link_service_connection_state.status == ConnectionStatus.PENDING:
    # Approve the connection
    approved_connection = PrivateEndpointConnection(
        private_endpoint=pending_connection.private_endpoint,
        private_link_service_connection_state=PrivateLinkServiceConnectionState(
            status=ConnectionStatus.APPROVED,
            description="Connection approved by administrator",
            actions_required=ActionsRequired.NONE
        )
    )
    
    approval_poller = client.private_endpoint_connections.begin_create_or_update(
        "my-resource-group",
        "my-registry",
        "pending-connection",
        approved_connection
    )
    
    approved = approval_poller.result()
    print(f"Approved connection: {approved.name}")
    print(f"New status: {approved.private_link_service_connection_state.status}")

# Reject a private endpoint connection
rejected_connection = PrivateEndpointConnection(
    private_endpoint=PrivateEndpoint(
        id="/subscriptions/subscription-id/resourceGroups/untrusted-rg/providers/Microsoft.Network/privateEndpoints/untrusted-endpoint"
    ),
    private_link_service_connection_state=PrivateLinkServiceConnectionState(
        status=ConnectionStatus.REJECTED,
        description="Connection rejected due to security policy",
        actions_required=ActionsRequired.NONE
    )
)

rejection_poller = client.private_endpoint_connections.begin_create_or_update(
    "my-resource-group",
    "my-registry",
    "untrusted-connection",
    rejected_connection
)

rejected = rejection_poller.result()
print(f"Rejected connection: {rejected.name}")

from azure.mgmt.containerregistry.models import (
    RegistryUpdateParameters, PublicNetworkAccess, NetworkRuleSet, DefaultAction
)

# Update registry to disable public access and require private endpoints
registry_update_params = RegistryUpdateParameters(
    public_network_access=PublicNetworkAccess.DISABLED,
    network_rule_set=NetworkRuleSet(
        default_action=DefaultAction.DENY,
        ip_rules=[]  # No IP rules since we're using private endpoints only
    )
)

updated_registry = client.registries.begin_update(
    "my-resource-group",
    "my-registry",
    registry_update_params
).result()

print(f"Updated registry for private access only")
print(f"Public network access: {updated_registry.public_network_access}")

# List available private link resources for the registry
private_link_resources = client.registries.list_private_link_resources(
    "my-resource-group",
    "my-registry"
)

print("Available Private Link Resources:")
print("-" * 40)
for resource in private_link_resources.value:
    print(f"Resource: {resource.name}")
    print(f"  Group ID: {resource.group_id}")
    print(f"  Required Members: {resource.required_members}")
    print(f"  Required Zone Names: {resource.required_zone_names}")
    print()

# Get specific private link resource
if private_link_resources.value:
    specific_resource = client.registries.get_private_link_resource(
        "my-resource-group",
        "my-registry",
        private_link_resources.value[0].group_id
    )
    print(f"Specific resource details: {specific_resource.name}")

# Complete setup for private networking including DNS configuration
import json

def setup_private_networking(client, resource_group, registry_name, vnet_config):
    """
    Complete private networking setup for a container registry.
    """
    
    # Step 1: Configure registry for private access
    print("Configuring registry for private access...")
    registry_params = RegistryUpdateParameters(
        public_network_access=PublicNetworkAccess.DISABLED,
        network_rule_set=NetworkRuleSet(default_action=DefaultAction.DENY)
    )
    
    client.registries.begin_update(
        resource_group, 
        registry_name,
        registry_params
    ).result()
    
    # Step 2: Get private link resources info for DNS setup
    print("Getting private link resource information...")
    private_link_resources = client.registries.list_private_link_resources(
        resource_group,
        registry_name
    )
    
    dns_config = {}
    for resource in private_link_resources.value:
        dns_config[resource.group_id] = {
            "required_zone_names": resource.required_zone_names,
            "required_members": resource.required_members
        }
    
    # Step 3: List existing private endpoint connections
    print("Checking existing private endpoint connections...")
    connections = list(client.private_endpoint_connections.list(
        resource_group,
        registry_name
    ))
    
    # Step 4: Auto-approve connections from trusted subnets
    trusted_subnets = vnet_config.get("trusted_subnets", [])
    
    for connection in connections:
        if connection.private_link_service_connection_state.status == ConnectionStatus.PENDING:
            # Check if private endpoint is from trusted subnet
            pe_id = connection.private_endpoint.id
            should_approve = any(subnet in pe_id for subnet in trusted_subnets)
            
            if should_approve:
                approved_connection = PrivateEndpointConnection(
                    private_endpoint=connection.private_endpoint,
                    private_link_service_connection_state=PrivateLinkServiceConnectionState(
                        status=ConnectionStatus.APPROVED,
                        description="Auto-approved from trusted subnet",
                        actions_required=ActionsRequired.NONE
                    )
                )
                
                client.private_endpoint_connections.begin_create_or_update(
                    resource_group,
                    registry_name,
                    connection.name,
                    approved_connection
                ).result()
                
                print(f"Auto-approved connection: {connection.name}")
    
    return {
        "registry_status": "private_access_configured",
        "dns_configuration": dns_config,
        "approved_connections": len([c for c in connections if c.private_link_service_connection_state.status == ConnectionStatus.APPROVED])
    }

# Example usage
vnet_configuration = {
    "trusted_subnets": [
        "/subscriptions/sub-id/resourceGroups/rg/providers/Microsoft.Network/virtualNetworks/vnet/subnets/trusted-subnet",
        "/subscriptions/sub-id/resourceGroups/rg/providers/Microsoft.Network/virtualNetworks/vnet/subnets/app-subnet"
    ]
}

result = setup_private_networking(
    client, 
    "my-resource-group", 
    "my-registry",
    vnet_configuration
)

print("Private networking setup completed:")
print(json.dumps(result, indent=2))

def monitor_private_endpoints(client, resource_group, registry_name):
    """
    Monitor the health and status of private endpoint connections.
    """
    connections = list(client.private_endpoint_connections.list(
        resource_group,
        registry_name
    ))
    
    health_report = {
        "total_connections": len(connections),
        "approved": 0,
        "pending": 0,
        "rejected": 0, 
        "disconnected": 0,
        "failed": 0,
        "needs_attention": []
    }
    
    for connection in connections:
        status = connection.private_link_service_connection_state.status
        
        if status == ConnectionStatus.APPROVED:
            health_report["approved"] += 1
        elif status == ConnectionStatus.PENDING:
            health_report["pending"] += 1
            health_report["needs_attention"].append({
                "connection": connection.name,
                "issue": "Pending approval",
                "action": "Review and approve/reject"
            })
        elif status == ConnectionStatus.REJECTED:
            health_report["rejected"] += 1
        elif status == ConnectionStatus.DISCONNECTED:
            health_report["disconnected"] += 1
            health_report["needs_attention"].append({
                "connection": connection.name,
                "issue": "Disconnected",
                "action": "Check private endpoint configuration"
            })
        
        # Check provisioning state
        if connection.provisioning_state == ProvisioningState.FAILED:
            health_report["failed"] += 1
            health_report["needs_attention"].append({
                "connection": connection.name,
                "issue": f"Provisioning failed",
                "action": "Review error details and retry"
            })
        
        # Check if actions are required
        if connection.private_link_service_connection_state.actions_required != ActionsRequired.NONE:
            health_report["needs_attention"].append({
                "connection": connection.name,
                "issue": f"Actions required: {connection.private_link_service_connection_state.actions_required}",
                "action": "Take required action"
            })
    
    return health_report

# Monitor private endpoint health
health_status = monitor_private_endpoints(
    client,
    "my-resource-group",
    "my-registry"
)

print("Private Endpoint Health Report:")
print("=" * 40)
print(f"Total Connections: {health_status['total_connections']}")
print(f"Approved: {health_status['approved']}")
print(f"Pending: {health_status['pending']}")
print(f"Rejected: {health_status['rejected']}")
print(f"Disconnected: {health_status['disconnected']}")
print(f"Failed: {health_status['failed']}")

if health_status["needs_attention"]:
    print("\nIssues Requiring Attention:")
    print("-" * 30)
    for issue in health_status["needs_attention"]:
        print(f"Connection: {issue['connection']}")
        print(f"  Issue: {issue['issue']}")
        print(f"  Action: {issue['action']}")
        print()