tessl/pypi-azure-cosmos

Microsoft Azure Cosmos Client Library for Python providing access to Azure Cosmos DB SQL API operations

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

User and Permission Management

Name: tessl/pypi-azure-cosmos
Author: tessl

User and permission management for fine-grained access control in Azure Cosmos DB. This enables multi-tenant applications with role-based security and resource-level permissions.

Capabilities

User Operations

Manage users within a database for permission-based access control.

def read(self, **kwargs):
    """
    Read user properties.
    
    Parameters:
    - session_token: Session token for consistency
    
    Returns:
    User properties as dictionary
    
    Raises:
    CosmosResourceNotFoundError: If user doesn't exist
    """

Permission Management

Create and manage permissions that grant users access to specific resources.

def create_permission(self, body: dict, **kwargs):
    """
    Create a new permission for the user.
    
    Parameters:
    - body: Permission definition with 'id', 'permissionMode', 'resource'
    - session_token: Session token for consistency
    
    Returns:
    Permission properties including resource token
    
    Raises:
    CosmosResourceExistsError: If permission already exists
    """

def list_permissions(self, max_item_count: int = None, **kwargs):
    """
    List all permissions for the user.
    
    Parameters:
    - max_item_count: Maximum number of permissions to return
    - session_token: Session token for consistency
    
    Returns:
    Iterable of permission items
    """

def query_permissions(self, query: str, parameters: list = None, max_item_count: int = None, **kwargs):
    """
    Query permissions using SQL syntax.
    
    Parameters:
    - query: SQL query string
    - parameters: Query parameters
    - max_item_count: Maximum items per page
    - session_token: Session token for consistency
    
    Returns:
    Iterable of query results
    """

def get_permission(self, permission: str, **kwargs):
    """
    Get permission properties.
    
    Parameters:
    - permission: Permission ID
    - session_token: Session token for consistency
    
    Returns:
    Permission properties including resource token
    
    Raises:
    CosmosResourceNotFoundError: If permission doesn't exist
    """

def upsert_permission(self, body: dict, **kwargs):
    """
    Create or replace a permission.
    
    Parameters:
    - body: Permission definition with 'id', 'permissionMode', 'resource'
    - session_token: Session token for consistency
    
    Returns:
    Permission properties including resource token
    """

def replace_permission(self, permission: str, body: dict, **kwargs):
    """
    Replace permission properties.
    
    Parameters:
    - permission: Permission ID
    - body: Updated permission definition
    - session_token: Session token for consistency
    - etag: ETag for conditional operations
    - match_condition: Match condition for conditional operations
    
    Returns:
    Updated permission properties
    
    Raises:
    CosmosResourceNotFoundError: If permission doesn't exist
    CosmosAccessConditionFailedError: If conditional operation fails
    """

def delete_permission(self, permission: str, **kwargs):
    """
    Delete a permission.
    
    Parameters:
    - permission: Permission ID
    - session_token: Session token for consistency
    - etag: ETag for conditional operations
    - match_condition: Match condition for conditional operations
    
    Raises:
    CosmosResourceNotFoundError: If permission doesn't exist
    """

Permission Types and Resource Tokens

Permission Modes

class PermissionMode:
    Read: str  # Read-only access
    All: str   # Full access (read, write, delete)

Permission Object Structure

class Permission:
    id: str              # Permission identifier
    user_link: str       # Link to the user
    permission_mode: str # Permission mode (Read or All)
    resource_link: str   # Link to the resource (container, document, etc.)
    properties: dict     # Additional permission properties
    permission_link: str # Full permission link

Usage Examples

Basic User and Permission Setup

from azure.cosmos import PermissionMode

# Get database client
database = client.get_database_client("MultiTenantApp")

# Create a user
user_def = {"id": "tenant1_user"}
user = database.create_user(user_def)
print(f"Created user: {user.id}")

# Get user client for permission management
user_client = database.get_user_client("tenant1_user")

# Create read permission for a container
read_permission = {
    "id": "products_read",
    "permissionMode": PermissionMode.Read,
    "resource": "dbs/MultiTenantApp/colls/Products"
}

permission = user_client.create_permission(read_permission)
resource_token = permission["_token"]
print(f"Created read permission with token: {resource_token[:20]}...")

# Create full access permission for user's private container
full_permission = {
    "id": "private_data_all",
    "permissionMode": PermissionMode.All,
    "resource": "dbs/MultiTenantApp/colls/PrivateData"
}

permission = user_client.create_permission(full_permission)
full_access_token = permission["_token"]

Using Resource Tokens for Authentication

# Create a client using resource token instead of master key
resource_token_client = CosmosClient(
    url="https://myaccount.documents.azure.com:443/",
    credential=resource_token,  # Use resource token instead of master key
    consistency_level=ConsistencyLevel.Session
)

# This client can only access resources granted by the permission
try:
    # This will work - user has read permission
    container = resource_token_client.get_database_client("MultiTenantApp").get_container_client("Products")
    items = list(container.read_all_items())
    print(f"Read {len(items)} items")
    
    # This will fail - user doesn't have write permission
    container.create_item({"id": "new_item", "data": "test"})
except CosmosHttpResponseError as e:
    print(f"Access denied: {e.status_code}")

Multi-Tenant Application Pattern

def setup_tenant_permissions(database, tenant_id):
    """Set up permissions for a new tenant."""
    
    # Create user for the tenant
    user_def = {"id": f"tenant_{tenant_id}"}
    try:
        user = database.create_user(user_def)
    except CosmosResourceExistsError:
        user = database.get_user_client(f"tenant_{tenant_id}")
    
    # Create tenant-specific container if needed
    tenant_container_id = f"tenant_{tenant_id}_data"
    try:
        container = database.create_container(
            id=tenant_container_id,
            partition_key=PartitionKey(path="/tenantId"),
            offer_throughput=400
        )
    except CosmosResourceExistsError:
        pass
    
    # Grant full access to tenant's own container
    tenant_permission = {
        "id": f"tenant_{tenant_id}_full_access",
        "permissionMode": PermissionMode.All, 
        "resource": f"dbs/{database.id}/colls/{tenant_container_id}"
    }
    
    user_client = database.get_user_client(f"tenant_{tenant_id}")
    permission = user_client.upsert_permission(tenant_permission)
    
    # Grant read access to shared reference data
    shared_permission = {
        "id": f"tenant_{tenant_id}_shared_read",
        "permissionMode": PermissionMode.Read,
        "resource": f"dbs/{database.id}/colls/SharedReferenceData"
    }
    
    shared_permission_obj = user_client.upsert_permission(shared_permission)
    
    return {
        "tenant_id": tenant_id,
        "full_access_token": permission["_token"],
        "shared_read_token": shared_permission_obj["_token"]
    }

# Set up permissions for multiple tenants
tenant_tokens = {}
for tenant_id in ["acme_corp", "widgets_inc", "data_solutions"]:
    tokens = setup_tenant_permissions(database, tenant_id)
    tenant_tokens[tenant_id] = tokens
    print(f"Set up permissions for {tenant_id}")

Permission Lifecycle Management

def manage_user_permissions(database, user_id):
    """Demonstrate full permission lifecycle."""
    
    user_client = database.get_user_client(user_id)
    
    # List current permissions
    permissions = list(user_client.list_permissions())
    print(f"User {user_id} has {len(permissions)} permissions")
    
    for perm in permissions:
        print(f"  - {perm['id']}: {perm['permissionMode']} on {perm['resource']}")
    
    # Query for specific permissions
    read_perms = list(user_client.query_permissions(
        query="SELECT * FROM permissions p WHERE p.permissionMode = @mode",
        parameters=[{"name": "@mode", "value": PermissionMode.Read}]
    ))
    print(f"Found {len(read_perms)} read-only permissions")
    
    # Update a permission (upgrade read to full access)
    if read_perms:
        perm_id = read_perms[0]["id"]
        updated_permission = {
            "id": perm_id,
            "permissionMode": PermissionMode.All,
            "resource": read_perms[0]["resource"]
        }
        
        user_client.replace_permission(perm_id, updated_permission)
        print(f"Upgraded permission {perm_id} to full access")
    
    # Clean up expired or unused permissions
    permissions_to_delete = []
    for perm in permissions:
        # Example: delete permissions older than 30 days
        # In real scenario, you'd check actual timestamps
        if "temp_" in perm["id"]:
            permissions_to_delete.append(perm["id"])
    
    for perm_id in permissions_to_delete:
        user_client.delete_permission(perm_id)
        print(f"Deleted temporary permission {perm_id}")

# Manage permissions for a user
manage_user_permissions(database, "tenant1_user")

Resource Token Validation and Error Handling

def validate_resource_token_access(resource_token, database_id, container_id):
    """Validate what operations a resource token can perform."""
    
    try:
        # Create client with resource token
        token_client = CosmosClient(
            url="https://myaccount.documents.azure.com:443/",
            credential=resource_token
        )
        
        database = token_client.get_database_client(database_id)
        container = database.get_container_client(container_id)
        
        # Test read access
        try:
            items = list(container.read_all_items(max_item_count=1))
            print("✓ Read access confirmed")
            read_access = True
        except CosmosHttpResponseError:
            print("✗ No read access")
            read_access = False
        
        # Test write access
        try:
            test_item = {
                "id": f"test_{uuid.uuid4()}",
                "test": True,
                "timestamp": datetime.utcnow().isoformat()
            }
            container.create_item(test_item)
            container.delete_item(test_item["id"], partition_key=test_item.get("partitionKey", test_item["id"]))
            print("✓ Write access confirmed")
            write_access = True
        except CosmosHttpResponseError:
            print("✗ No write access")
            write_access = False
        
        return {
            "valid": True,
            "read_access": read_access,
            "write_access": write_access
        }
        
    except CosmosHttpResponseError as e:
        print(f"Token validation failed: {e.status_code} - {e.message}")
        return {
            "valid": False,
            "error": str(e)
        }

# Validate token access
import uuid
from datetime import datetime

token_info = validate_resource_token_access(
    resource_token=tenant_tokens["acme_corp"]["full_access_token"],
    database_id="MultiTenantApp",
    container_id="tenant_acme_corp_data"
)

print(f"Token validation result: {token_info}")

Install with Tessl CLI