tessl/pypi-azure-mgmt-servicebus

Microsoft Azure Service Bus Management Client Library for programmatic control of Service Bus namespaces, queues, topics, subscriptions, and authorization rules

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Authorization and Security

Name: tessl/pypi-azure-mgmt-servicebus
Author: tessl

Access control through authorization rules, access key management, private endpoint configuration, and network security rules. Service Bus provides comprehensive security features including Shared Access Signature (SAS) authentication, Azure Active Directory integration, network isolation, and encryption.

Capabilities

Shared Access Signature (SAS) Authorization

Manage authorization rules with granular access rights for different scenarios and applications.

def list_authorization_rules(
    self,
    resource_group_name: str,
    namespace_name: str
) -> ItemPaged[SBAuthorizationRule]:
    """List authorization rules for a namespace.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
    
    Returns:
        ItemPaged[SBAuthorizationRule]: Iterable of authorization rules.
    """

def create_or_update_authorization_rule(
    self,
    resource_group_name: str,
    namespace_name: str,
    authorization_rule_name: str,
    parameters: SBAuthorizationRule
) -> SBAuthorizationRule:
    """Create or update an authorization rule.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
        authorization_rule_name (str): Name of the authorization rule.
        parameters (SBAuthorizationRule): Authorization rule configuration.
    
    Returns:
        SBAuthorizationRule: The created or updated authorization rule.
    """

def get_authorization_rule(
    self,
    resource_group_name: str,
    namespace_name: str,
    authorization_rule_name: str
) -> SBAuthorizationRule:
    """Get an authorization rule.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
        authorization_rule_name (str): Name of the authorization rule.
    
    Returns:
        SBAuthorizationRule: The authorization rule.
    """

def delete_authorization_rule(
    self,
    resource_group_name: str,
    namespace_name: str,
    authorization_rule_name: str
) -> None:
    """Delete an authorization rule.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
        authorization_rule_name (str): Name of the authorization rule.
    """

Access Key Management

Generate, retrieve, and regenerate access keys for secure authentication to Service Bus resources.

def list_keys(
    self,
    resource_group_name: str,
    namespace_name: str,
    authorization_rule_name: str
) -> AccessKeys:
    """Get access keys for an authorization rule.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
        authorization_rule_name (str): Name of the authorization rule.
    
    Returns:
        AccessKeys: Primary and secondary keys with connection strings.
    """

def regenerate_keys(
    self,
    resource_group_name: str,
    namespace_name: str,
    authorization_rule_name: str,
    parameters: RegenerateAccessKeyParameters
) -> AccessKeys:
    """Regenerate access keys for an authorization rule.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
        authorization_rule_name (str): Name of the authorization rule.
        parameters (RegenerateAccessKeyParameters): Key regeneration parameters.
    
    Returns:
        AccessKeys: New access keys with connection strings.
    """

Private Endpoint Configuration

Manage private endpoint connections for secure network access to Service Bus namespaces.

def list(
    self,
    resource_group_name: str,
    namespace_name: str
) -> ItemPaged[PrivateEndpointConnection]:
    """List private endpoint connections for a namespace.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
    
    Returns:
        ItemPaged[PrivateEndpointConnection]: Iterable of private endpoint connections.
    """

def create_or_update(
    self,
    resource_group_name: str,
    namespace_name: str,
    private_endpoint_connection_name: str,
    parameters: PrivateEndpointConnection
) -> PrivateEndpointConnection:
    """Create or update a private endpoint connection.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
        private_endpoint_connection_name (str): Name of the private endpoint connection.
        parameters (PrivateEndpointConnection): Connection configuration.
    
    Returns:
        PrivateEndpointConnection: The private endpoint connection.
    """

def get(
    self,
    resource_group_name: str,
    namespace_name: str,
    private_endpoint_connection_name: str
) -> PrivateEndpointConnection:
    """Get a private endpoint connection.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
        private_endpoint_connection_name (str): Name of the private endpoint connection.
    
    Returns:
        PrivateEndpointConnection: The private endpoint connection.
    """

def delete(
    self,
    resource_group_name: str,
    namespace_name: str,
    private_endpoint_connection_name: str
) -> None:
    """Delete a private endpoint connection.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
        private_endpoint_connection_name (str): Name of the private endpoint connection.
    """

Private Link Resources

Discover available private link resources for network integration.

def get(
    self,
    resource_group_name: str,
    namespace_name: str
) -> PrivateLinkResourcesListResult:
    """Get private link resources for a namespace.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
    
    Returns:
        PrivateLinkResourcesListResult: Available private link resources.
    """

Network Rule Set Management

Configure IP filtering and virtual network access rules for namespace-level security.

def create_or_update_network_rule_set(
    self,
    resource_group_name: str,
    namespace_name: str,
    parameters: NetworkRuleSet
) -> NetworkRuleSet:
    """Create or update network rule set for a namespace.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
        parameters (NetworkRuleSet): Network rule configuration.
    
    Returns:
        NetworkRuleSet: The network rule set.
    """

def get_network_rule_set(
    self,
    resource_group_name: str,
    namespace_name: str
) -> NetworkRuleSet:
    """Get network rule set for a namespace.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
    
    Returns:
        NetworkRuleSet: Current network rule set.
    """

Usage Examples

Creating Authorization Rules with Different Access Rights

from azure.mgmt.servicebus import ServiceBusManagementClient
from azure.mgmt.servicebus.models import SBAuthorizationRule, AccessRights
from azure.identity import DefaultAzureCredential

client = ServiceBusManagementClient(DefaultAzureCredential(), subscription_id)

# Create a send-only authorization rule for message producers
send_only_rule = SBAuthorizationRule(
    rights=[AccessRights.SEND]
)

producer_auth = client.namespaces.create_or_update_authorization_rule(
    resource_group_name="my-resource-group",
    namespace_name="my-servicebus-namespace",
    authorization_rule_name="ProducerPolicy",
    parameters=send_only_rule
)

# Create a listen-only authorization rule for message consumers
listen_only_rule = SBAuthorizationRule(
    rights=[AccessRights.LISTEN]
)

consumer_auth = client.namespaces.create_or_update_authorization_rule(
    resource_group_name="my-resource-group",
    namespace_name="my-servicebus-namespace",
    authorization_rule_name="ConsumerPolicy",
    parameters=listen_only_rule
)

# Create a full-access authorization rule for administrative tasks
admin_rule = SBAuthorizationRule(
    rights=[AccessRights.MANAGE, AccessRights.SEND, AccessRights.LISTEN]
)

admin_auth = client.namespaces.create_or_update_authorization_rule(
    resource_group_name="my-resource-group",
    namespace_name="my-servicebus-namespace",
    authorization_rule_name="AdminPolicy",
    parameters=admin_rule
)

print("Authorization rules created successfully")

Managing Access Keys and Connection Strings

from azure.mgmt.servicebus.models import RegenerateAccessKeyParameters, KeyType

# Get access keys for an authorization rule
keys = client.namespaces.list_keys(
    resource_group_name="my-resource-group",
    namespace_name="my-servicebus-namespace",
    authorization_rule_name="ProducerPolicy"
)

print(f"Primary connection string: {keys.primary_connection_string}")
print(f"Secondary connection string: {keys.secondary_connection_string}")
print(f"Primary key: {keys.primary_key}")
print(f"Secondary key: {keys.secondary_key}")

# Regenerate the primary key
regenerate_params = RegenerateAccessKeyParameters(
    key_type=KeyType.PRIMARY_KEY
)

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

print(f"New primary key: {new_keys.primary_key}")

# Regenerate with custom key value
custom_regenerate_params = RegenerateAccessKeyParameters(
    key_type=KeyType.SECONDARY_KEY,
    key="my-custom-key-value-here"  # Optional custom key
)

custom_keys = client.namespaces.regenerate_keys(
    resource_group_name="my-resource-group",
    namespace_name="my-servicebus-namespace", 
    authorization_rule_name="ProducerPolicy",
    parameters=custom_regenerate_params
)

Configuring Network Security Rules

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

# Configure IP-based access control
ip_rules = [
    NWRuleSetIpRules(
        ip_mask="203.0.113.0/24",  # Allow entire subnet
        action=NetworkRuleIPAction.ALLOW
    ),
    NWRuleSetIpRules(
        ip_mask="198.51.100.42",  # Allow specific IP
        action=NetworkRuleIPAction.ALLOW
    )
]

# Configure virtual network rules
vnet_rules = [
    NWRuleSetVirtualNetworkRules(
        subnet=Subnet(
            id="/subscriptions/{subscription-id}/resourceGroups/{rg}/providers/Microsoft.Network/virtualNetworks/{vnet}/subnets/{subnet}"
        ),
        ignore_missing_vnet_service_endpoint=False
    )
]

# Create network rule set
network_rules = NetworkRuleSet(
    trusted_service_access_enabled=True,  # Allow trusted Azure services
    default_action=DefaultAction.DENY,    # Deny by default, allow only specified IPs/VNets
    ip_rules=ip_rules,
    virtual_network_rules=vnet_rules
)

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

print("Network rules configured successfully")

Managing Private Endpoint Connections

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

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

for connection in connections:
    print(f"Connection: {connection.name}")
    print(f"Status: {connection.connection_state.status}")
    print(f"Description: {connection.connection_state.description}")

# Approve a private endpoint connection
connection_state = ConnectionState(
    status=PrivateLinkConnectionStatus.APPROVED,
    description="Approved by administrator"
)

updated_connection = client.private_endpoint_connections.create_or_update(
    resource_group_name="my-resource-group",
    namespace_name="my-servicebus-namespace",
    private_endpoint_connection_name="connection-name",
    parameters=PrivateEndpointConnection(
        connection_state=connection_state
    )
)

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

for resource in link_resources.value:
    print(f"Resource: {resource.name}")
    print(f"Group ID: {resource.group_id}")
    print(f"Required members: {resource.required_members}")

Implementing Role-Based Access Control

# Create different authorization rules for different application roles

# Application service that only sends messages
app_sender_rule = SBAuthorizationRule(
    rights=[AccessRights.SEND]
)

client.namespaces.create_or_update_authorization_rule(
    resource_group_name="my-resource-group",
    namespace_name="my-servicebus-namespace",
    authorization_rule_name="AppSenderPolicy",
    parameters=app_sender_rule
)

# Background service that only processes messages
processor_rule = SBAuthorizationRule(
    rights=[AccessRights.LISTEN]
)

client.namespaces.create_or_update_authorization_rule(
    resource_group_name="my-resource-group",
    namespace_name="my-servicebus-namespace",
    authorization_rule_name="ProcessorPolicy", 
    parameters=processor_rule
)

# Monitoring service that needs to peek at queues
monitor_rule = SBAuthorizationRule(
    rights=[AccessRights.LISTEN]  # Listen includes peek capabilities
)

client.namespaces.create_or_update_authorization_rule(
    resource_group_name="my-resource-group",
    namespace_name="my-servicebus-namespace",
    authorization_rule_name="MonitorPolicy",
    parameters=monitor_rule
)

# Administrative service with full access
admin_rule = SBAuthorizationRule(
    rights=[AccessRights.MANAGE, AccessRights.SEND, AccessRights.LISTEN]
)

client.namespaces.create_or_update_authorization_rule(
    resource_group_name="my-resource-group",
    namespace_name="my-servicebus-namespace",
    authorization_rule_name="AdminPolicy",
    parameters=admin_rule
)

# Get connection strings for each role
app_keys = client.namespaces.list_keys(
    resource_group_name="my-resource-group",
    namespace_name="my-servicebus-namespace",
    authorization_rule_name="AppSenderPolicy"
)

processor_keys = client.namespaces.list_keys(
    resource_group_name="my-resource-group",
    namespace_name="my-servicebus-namespace",
    authorization_rule_name="ProcessorPolicy"
)

print("Role-based access control configured:")
print(f"App sender connection: {app_keys.primary_connection_string}")
print(f"Processor connection: {processor_keys.primary_connection_string}")

Types

class SBAuthorizationRule:
    def __init__(self, **kwargs): ...
    
    # Authorization rule properties
    id: Optional[str]
    name: Optional[str]
    type: Optional[str]
    rights: List[Union[str, AccessRights]]

class AccessKeys:
    def __init__(self, **kwargs): ...
    
    # Connection strings for different scenarios
    primary_connection_string: Optional[str]
    secondary_connection_string: Optional[str]
    alias_primary_connection_string: Optional[str]    # For GEO DR
    alias_secondary_connection_string: Optional[str]  # For GEO DR
    
    # Raw access keys
    primary_key: Optional[str]
    secondary_key: Optional[str]
    key_name: Optional[str]

class RegenerateAccessKeyParameters:
    def __init__(self, **kwargs): ...
    
    key_type: Union[str, KeyType]  # "PrimaryKey" or "SecondaryKey"
    key: Optional[str]  # Optional custom key value

class NetworkRuleSet:
    def __init__(self, **kwargs): ...
    
    # Network access configuration
    trusted_service_access_enabled: Optional[bool]  # Allow trusted Azure services
    default_action: Optional[Union[str, DefaultAction]]  # "Allow" or "Deny"
    
    # Access rules
    virtual_network_rules: Optional[List[NWRuleSetVirtualNetworkRules]]
    ip_rules: Optional[List[NWRuleSetIpRules]]

class NWRuleSetIpRules:
    def __init__(self, **kwargs): ...
    
    ip_mask: Optional[str]  # IP address or CIDR range
    action: Optional[Union[str, NetworkRuleIPAction]]  # "Allow"

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

class Subnet:
    def __init__(self, **kwargs): ...
    
    id: Optional[str]  # Azure resource ID of the subnet

class PrivateEndpointConnection:
    def __init__(self, **kwargs): ...
    
    # Connection properties
    id: Optional[str]
    name: Optional[str]
    type: Optional[str]
    
    # Private endpoint details
    private_endpoint: Optional[PrivateEndpoint]
    connection_state: Optional[ConnectionState]
    provisioning_state: Optional[Union[str, EndPointProvisioningState]]

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

class PrivateEndpoint:
    def __init__(self, **kwargs): ...
    
    id: Optional[str]  # Azure resource ID of the private endpoint

from enum import Enum

class AccessRights(str, Enum):
    MANAGE = "Manage"  # Full control including create/delete entities
    SEND = "Send"      # Send messages to entities
    LISTEN = "Listen"  # Receive/peek messages from entities

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

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 EndPointProvisioningState(str, Enum):
    CREATING = "Creating"
    UPDATING = "Updating"
    DELETING = "Deleting"
    SUCCEEDED = "Succeeded"
    CANCELED = "Canceled"
    FAILED = "Failed"

Security Best Practices

Key Management

Use separate authorization rules for different application roles
Regenerate keys regularly and implement key rotation strategies
Use secondary keys for zero-downtime key rotation
Store connection strings securely using Azure Key Vault

Network Security

Enable network rules to restrict access to trusted networks
Use private endpoints for secure connectivity from Azure VNets
Configure IP allowlists for on-premises or specific external access
Enable trusted service access for Azure platform services

Access Control

Follow principle of least privilege when assigning rights
Use Manage rights only for administrative operations
Separate send and listen operations with different authorization rules
Regularly audit and review authorization rules

Install with Tessl CLI

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

docs

authorization-security.md

disaster-recovery.md

index.md

message-filtering-rules.md

migration-monitoring.md

namespace-management.md

queue-operations.md

topic-subscription-management.md

tile.json

tessl/pypi-azure-mgmt-servicebus

authorization-security.md.css-3qkkll{font-size:var(--chakra-font-sizes-sm);font-weight:var(--chakra-font-weights-normal);color:var(--chakra-colors-gray-300);}docs/

Authorization and Security

Capabilities

Shared Access Signature (SAS) Authorization

Access Key Management

Private Endpoint Configuration

Private Link Resources

Network Rule Set Management

Usage Examples

Creating Authorization Rules with Different Access Rights

Managing Access Keys and Connection Strings

Configuring Network Security Rules

Managing Private Endpoint Connections

Implementing Role-Based Access Control

Types

Security Best Practices

Key Management

Network Security

Access Control

authorization-security.mddocs/