CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-azure-mgmt-eventhub

Microsoft Azure Event Hub 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

security-access-control.mddocs/

Security and Access Control

Comprehensive security management for Azure Event Hub including authorization rules, access keys, network security, private endpoints, and managed identity configuration. These capabilities ensure secure, controlled access to Event Hub resources in enterprise environments.

Capabilities

Authorization Rules Management

Authorization rules control access permissions to namespaces and Event Hubs, defining what operations (Manage, Send, Listen) clients can perform with associated connection strings and access keys.

class NamespacesOperations:
    def list_authorization_rules(
        self,
        resource_group_name: str,
        namespace_name: str,
        **kwargs
    ) -> Iterable[AuthorizationRule]:
        """List namespace authorization rules."""
    
    def create_or_update_authorization_rule(
        self,
        resource_group_name: str,
        namespace_name: str,
        authorization_rule_name: str,
        parameters: Union[AuthorizationRule, IO[bytes]],
        **kwargs
    ) -> AuthorizationRule:
        """Create or update namespace authorization rule."""
    
    def delete_authorization_rule(
        self,
        resource_group_name: str,
        namespace_name: str,
        authorization_rule_name: str,
        **kwargs
    ) -> None:
        """Delete namespace authorization rule."""
    
    def get_authorization_rule(
        self,
        resource_group_name: str,
        namespace_name: str,
        authorization_rule_name: str,
        **kwargs
    ) -> AuthorizationRule:
        """Get namespace authorization rule."""
    
    def list_keys(
        self,
        resource_group_name: str,
        namespace_name: str,
        authorization_rule_name: str,
        **kwargs
    ) -> AccessKeys:
        """Get namespace authorization rule access keys."""
    
    def regenerate_keys(
        self,
        resource_group_name: str,
        namespace_name: str,
        authorization_rule_name: str,
        parameters: Union[RegenerateAccessKeyParameters, IO[bytes]],
        **kwargs
    ) -> AccessKeys:
        """Regenerate namespace authorization rule keys."""

class EventHubsOperations:
    def list_authorization_rules(
        self,
        resource_group_name: str,
        namespace_name: str,
        event_hub_name: str,
        **kwargs
    ) -> Iterable[AuthorizationRule]:
        """List Event Hub authorization rules."""
    
    def create_or_update_authorization_rule(
        self,
        resource_group_name: str,
        namespace_name: str,
        event_hub_name: str,
        authorization_rule_name: str,
        parameters: Union[AuthorizationRule, IO[bytes]],
        **kwargs
    ) -> AuthorizationRule:
        """Create or update Event Hub authorization rule."""
    
    def delete_authorization_rule(
        self,
        resource_group_name: str,
        namespace_name: str,
        event_hub_name: str,
        authorization_rule_name: str,
        **kwargs
    ) -> None:
        """Delete Event Hub authorization rule."""
    
    def get_authorization_rule(
        self,
        resource_group_name: str,
        namespace_name: str,
        event_hub_name: str,
        authorization_rule_name: str,
        **kwargs
    ) -> AuthorizationRule:
        """Get Event Hub authorization rule."""
    
    def list_keys(
        self,
        resource_group_name: str,
        namespace_name: str,
        event_hub_name: str,
        authorization_rule_name: str,
        **kwargs
    ) -> AccessKeys:
        """Get Event Hub authorization rule access keys."""
    
    def regenerate_keys(
        self,
        resource_group_name: str,
        namespace_name: str,
        event_hub_name: str,
        authorization_rule_name: str,
        parameters: Union[RegenerateAccessKeyParameters, IO[bytes]],
        **kwargs
    ) -> AccessKeys:
        """Regenerate Event Hub authorization rule keys."""

Usage Example

from azure.mgmt.eventhub.models import AuthorizationRule, AccessRights, RegenerateAccessKeyParameters, KeyType

# Create namespace-level authorization rule for full management access
mgmt_rule_params = AuthorizationRule(
    rights=[AccessRights.MANAGE, AccessRights.SEND, AccessRights.LISTEN]
)

mgmt_rule = client.namespaces.create_or_update_authorization_rule(
    resource_group_name="my-resource-group",
    namespace_name="my-eventhub-namespace",
    authorization_rule_name="FullAccessRule",
    parameters=mgmt_rule_params
)

# Create Event Hub-level authorization rule for send-only access
send_rule_params = AuthorizationRule(
    rights=[AccessRights.SEND]
)

send_rule = client.event_hubs.create_or_update_authorization_rule(
    resource_group_name="my-resource-group",
    namespace_name="my-eventhub-namespace",
    event_hub_name="telemetry-events",
    authorization_rule_name="SendOnlyRule",
    parameters=send_rule_params
)

# Get access keys for the authorization rule
access_keys = client.namespaces.list_keys(
    resource_group_name="my-resource-group",
    namespace_name="my-eventhub-namespace",
    authorization_rule_name="FullAccessRule"
)

print(f"Primary Connection String: {access_keys.primary_connection_string}")
print(f"Secondary Key: {access_keys.secondary_key}")

# Regenerate primary key for security rotation
regenerate_params = RegenerateAccessKeyParameters(key_type=KeyType.PRIMARY_KEY)

new_keys = client.namespaces.regenerate_keys(
    resource_group_name="my-resource-group",
    namespace_name="my-eventhub-namespace", 
    authorization_rule_name="FullAccessRule",
    parameters=regenerate_params
)

Network Security Management

Network security features control access to Event Hub namespaces through IP filtering, virtual network rules, and default access policies to secure network-level access.

class NamespacesOperations:
    def create_or_update_network_rule_set(
        self,
        resource_group_name: str,
        namespace_name: str,
        parameters: Union[NetworkRuleSet, IO[bytes]],
        **kwargs
    ) -> NetworkRuleSet:
        """Create or update network rule set."""
    
    def get_network_rule_set(
        self,
        resource_group_name: str,
        namespace_name: str,
        **kwargs
    ) -> NetworkRuleSet:
        """Get network rule set."""
    
    def list_network_rule_set(
        self,
        resource_group_name: str,
        namespace_name: str,
        **kwargs
    ) -> Iterable[NetworkRuleSet]:
        """List network rule sets."""

Usage Example

from azure.mgmt.eventhub.models import (
    NetworkRuleSet, DefaultAction, NWRuleSetIpRules, 
    NWRuleSetVirtualNetworkRules, NetworkRuleIPAction
)

# Configure network access rules
ip_rules = [
    NWRuleSetIpRules(
        ip_mask="192.168.1.0/24",
        action=NetworkRuleIPAction.ALLOW
    ),
    NWRuleSetIpRules(
        ip_mask="10.0.0.0/16", 
        action=NetworkRuleIPAction.ALLOW
    )
]

vnet_rules = [
    NWRuleSetVirtualNetworkRules(
        subnet={"id": "/subscriptions/{sub}/resourceGroups/{rg}/providers/Microsoft.Network/virtualNetworks/{vnet}/subnets/{subnet}"},
        ignore_missing_vnet_service_endpoint=False
    )
]

network_rule_set = NetworkRuleSet(
    default_action=DefaultAction.DENY,  # Deny by default, allow only specified IPs/VNets
    virtual_network_rules=vnet_rules,
    ip_rules=ip_rules,
    trusted_service_access_enabled=True,  # Allow trusted Azure services
    public_network_access=PublicNetworkAccess.ENABLED
)

network_rules = client.namespaces.create_or_update_network_rule_set(
    resource_group_name="my-resource-group",
    namespace_name="my-eventhub-namespace",
    parameters=network_rule_set
)

print(f"Network rules configured with default action: {network_rules.default_action}")

Private Endpoint Management

Private endpoints enable secure, private connectivity to Event Hub namespaces through Azure Private Link, ensuring traffic stays within the Azure backbone network.

class PrivateEndpointConnectionsOperations:
    def list(
        self,
        resource_group_name: str,
        namespace_name: str,
        **kwargs
    ) -> Iterable[PrivateEndpointConnection]:
        """List private endpoint connections."""
    
    def create_or_update(
        self,
        resource_group_name: str,
        namespace_name: str,
        private_endpoint_connection_name: str,
        parameters: Union[PrivateEndpointConnection, IO[bytes]],
        **kwargs
    ) -> PrivateEndpointConnection:
        """Create or update private endpoint connection."""
    
    def begin_delete(
        self,
        resource_group_name: str,
        namespace_name: str,
        private_endpoint_connection_name: str,
        **kwargs
    ) -> LROPoller[None]:
        """Delete private endpoint connection (long-running operation)."""
    
    def get(
        self,
        resource_group_name: str,
        namespace_name: str,
        private_endpoint_connection_name: str,
        **kwargs
    ) -> PrivateEndpointConnection:
        """Get private endpoint connection."""

class PrivateLinkResourcesOperations:
    def get(
        self,
        resource_group_name: str,
        namespace_name: str,
        **kwargs
    ) -> PrivateLinkResourcesListResult:
        """Get private link resources."""

Usage Example

from azure.mgmt.eventhub.models import (
    PrivateEndpointConnection, ConnectionState, 
    PrivateLinkConnectionStatus
)

# List existing private endpoint connections
private_connections = client.private_endpoint_connections.list(
    resource_group_name="my-resource-group",
    namespace_name="my-eventhub-namespace"
)

for connection in private_connections:
    print(f"Private Endpoint: {connection.name}")
    print(f"Connection State: {connection.connection_state.status}")

# Approve a private endpoint connection
connection_state = ConnectionState(
    status=PrivateLinkConnectionStatus.APPROVED,
    description="Approved for production use"
)

private_endpoint_connection = PrivateEndpointConnection(
    connection_state=connection_state
)

approved_connection = client.private_endpoint_connections.create_or_update(
    resource_group_name="my-resource-group",
    namespace_name="my-eventhub-namespace",
    private_endpoint_connection_name="my-private-endpoint-connection",
    parameters=private_endpoint_connection
)

# Get available private link resources
private_link_resources = client.private_link_resources.get(
    resource_group_name="my-resource-group",
    namespace_name="my-eventhub-namespace"
)

for resource in private_link_resources.value:
    print(f"Private Link Resource: {resource.name}")
    print(f"Group ID: {resource.group_id}")
    print(f"Required Members: {resource.required_members}")

Network Security Perimeter Configuration

Advanced network security perimeter management for controlling access boundaries and resource associations in enterprise environments.

class NetworkSecurityPerimeterConfigurationOperations:
    def list(
        self,
        resource_group_name: str,
        namespace_name: str,
        **kwargs
    ) -> Iterable[NetworkSecurityPerimeterConfiguration]:
        """List network security perimeter configurations."""

class NetworkSecurityPerimeterConfigurationsOperations:
    def begin_create_or_update(
        self,
        resource_group_name: str,
        namespace_name: str,
        resource_association_name: str,
        **kwargs
    ) -> LROPoller[None]:
        """Create or update network security perimeter configuration (long-running operation)."""

Usage Example

# List network security perimeter configurations
perimeter_configs = client.network_security_perimeter_configuration.list(
    resource_group_name="my-resource-group",
    namespace_name="my-eventhub-namespace"
)

for config in perimeter_configs:
    print(f"Configuration ID: {config.id}")
    print(f"Perimeter: {config.network_security_perimeter}")

# Create or update network security perimeter configuration
perimeter_operation = client.network_security_perimeter_configurations.begin_create_or_update(
    resource_group_name="my-resource-group", 
    namespace_name="my-eventhub-namespace",
    resource_association_name="my-perimeter-association"
)

# Wait for completion
perimeter_operation.result()
print("Network security perimeter configuration updated")

Types

class AuthorizationRule(ProxyResource):
    def __init__(
        self,
        rights: Optional[List[Union[str, AccessRights]]] = None,
        **kwargs: Any
    ): ...
    
    rights: Optional[List[AccessRights]]

class AccessKeys:
    def __init__(
        self,
        primary_connection_string: Optional[str] = None,
        secondary_connection_string: Optional[str] = None,
        alias_primary_connection_string: Optional[str] = None,
        alias_secondary_connection_string: Optional[str] = None,
        primary_key: Optional[str] = None,
        secondary_key: Optional[str] = None,
        key_name: Optional[str] = None,
        **kwargs: Any
    ): ...

class NetworkRuleSet(ProxyResource):
    def __init__(
        self,
        trusted_service_access_enabled: Optional[bool] = None,
        default_action: Optional[Union[str, DefaultAction]] = None,
        virtual_network_rules: Optional[List[NWRuleSetVirtualNetworkRules]] = None,
        ip_rules: Optional[List[NWRuleSetIpRules]] = None,
        public_network_access: Optional[Union[str, PublicNetworkAccess]] = None,
        **kwargs: Any
    ): ...

class NWRuleSetIpRules:
    def __init__(
        self,
        ip_mask: Optional[str] = None,
        action: Optional[Union[str, NetworkRuleIPAction]] = None,
        **kwargs: Any
    ): ...

class NWRuleSetVirtualNetworkRules:
    def __init__(
        self,
        subnet: Optional[Subnet] = None,
        ignore_missing_vnet_service_endpoint: Optional[bool] = None,
        **kwargs: Any
    ): ...

class PrivateEndpointConnection(ProxyResource):
    def __init__(
        self,
        private_endpoint: Optional[PrivateEndpoint] = None,
        connection_state: Optional[ConnectionState] = None,
        provisioning_state: Optional[Union[str, EndPointProvisioningState]] = None,
        **kwargs: Any
    ): ...

class ConnectionState:
    def __init__(
        self,
        status: Optional[Union[str, PrivateLinkConnectionStatus]] = None,
        description: Optional[str] = None,
        actions_required: Optional[str] = None,
        **kwargs: Any
    ): ...

class RegenerateAccessKeyParameters:
    def __init__(
        self,
        key_type: Union[str, KeyType],
        key: Optional[str] = None,
        **kwargs: Any
    ): ...

# Enums
class AccessRights(str, Enum):
    MANAGE = "Manage"
    SEND = "Send" 
    LISTEN = "Listen"

class DefaultAction(str, Enum):
    ALLOW = "Allow"
    DENY = "Deny"

class NetworkRuleIPAction(str, Enum):
    ALLOW = "Allow"

class PrivateLinkConnectionStatus(str, Enum):
    PENDING = "Pending"
    APPROVED = "Approved"
    REJECTED = "Rejected"
    DISCONNECTED = "Disconnected"

class KeyType(str, Enum):
    PRIMARY_KEY = "PrimaryKey"
    SECONDARY_KEY = "SecondaryKey"

class PublicNetworkAccess(str, Enum):
    ENABLED = "Enabled"
    DISABLED = "Disabled"
    SECURED_BY_PERIMETER = "SecuredByPerimeter"

Install with Tessl CLI

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

docs

core-management.md

enterprise-features.md

index.md

security-access-control.md

tile.json