0
# User and Permission Management
1

2
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.
3

4
## Capabilities
5

6
### User Operations
7

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

10
```python { .api }
11
def read(self, **kwargs):
12
    """
13
    Read user properties.
14
    
15
    Parameters:
16
    - session_token: Session token for consistency
17
    
18
    Returns:
19
    User properties as dictionary
20
    
21
    Raises:
22
    CosmosResourceNotFoundError: If user doesn't exist
23
    """
24
```
25

26
### Permission Management
27

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

30
```python { .api }
31
def create_permission(self, body: dict, **kwargs):
32
    """
33
    Create a new permission for the user.
34
    
35
    Parameters:
36
    - body: Permission definition with 'id', 'permissionMode', 'resource'
37
    - session_token: Session token for consistency
38
    
39
    Returns:
40
    Permission properties including resource token
41
    
42
    Raises:
43
    CosmosResourceExistsError: If permission already exists
44
    """
45

46
def list_permissions(self, max_item_count: int = None, **kwargs):
47
    """
48
    List all permissions for the user.
49
    
50
    Parameters:
51
    - max_item_count: Maximum number of permissions to return
52
    - session_token: Session token for consistency
53
    
54
    Returns:
55
    Iterable of permission items
56
    """
57

58
def query_permissions(self, query: str, parameters: list = None, max_item_count: int = None, **kwargs):
59
    """
60
    Query permissions using SQL syntax.
61
    
62
    Parameters:
63
    - query: SQL query string
64
    - parameters: Query parameters
65
    - max_item_count: Maximum items per page
66
    - session_token: Session token for consistency
67
    
68
    Returns:
69
    Iterable of query results
70
    """
71

72
def get_permission(self, permission: str, **kwargs):
73
    """
74
    Get permission properties.
75
    
76
    Parameters:
77
    - permission: Permission ID
78
    - session_token: Session token for consistency
79
    
80
    Returns:
81
    Permission properties including resource token
82
    
83
    Raises:
84
    CosmosResourceNotFoundError: If permission doesn't exist
85
    """
86

87
def upsert_permission(self, body: dict, **kwargs):
88
    """
89
    Create or replace a permission.
90
    
91
    Parameters:
92
    - body: Permission definition with 'id', 'permissionMode', 'resource'
93
    - session_token: Session token for consistency
94
    
95
    Returns:
96
    Permission properties including resource token
97
    """
98

99
def replace_permission(self, permission: str, body: dict, **kwargs):
100
    """
101
    Replace permission properties.
102
    
103
    Parameters:
104
    - permission: Permission ID
105
    - body: Updated permission definition
106
    - session_token: Session token for consistency
107
    - etag: ETag for conditional operations
108
    - match_condition: Match condition for conditional operations
109
    
110
    Returns:
111
    Updated permission properties
112
    
113
    Raises:
114
    CosmosResourceNotFoundError: If permission doesn't exist
115
    CosmosAccessConditionFailedError: If conditional operation fails
116
    """
117

118
def delete_permission(self, permission: str, **kwargs):
119
    """
120
    Delete a permission.
121
    
122
    Parameters:
123
    - permission: Permission ID
124
    - session_token: Session token for consistency
125
    - etag: ETag for conditional operations
126
    - match_condition: Match condition for conditional operations
127
    
128
    Raises:
129
    CosmosResourceNotFoundError: If permission doesn't exist
130
    """
131
```
132

133
## Permission Types and Resource Tokens
134

135
### Permission Modes
136

137
```python { .api }
138
class PermissionMode:
139
    Read: str  # Read-only access
140
    All: str   # Full access (read, write, delete)
141
```
142

143
### Permission Object Structure
144

145
```python { .api }
146
class Permission:
147
    id: str              # Permission identifier
148
    user_link: str       # Link to the user
149
    permission_mode: str # Permission mode (Read or All)
150
    resource_link: str   # Link to the resource (container, document, etc.)
151
    properties: dict     # Additional permission properties
152
    permission_link: str # Full permission link
153
```
154

155
## Usage Examples
156

157
### Basic User and Permission Setup
158

159
```python
160
from azure.cosmos import PermissionMode
161

162
# Get database client
163
database = client.get_database_client("MultiTenantApp")
164

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

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

173
# Create read permission for a container
174
read_permission = {
175
    "id": "products_read",
176
    "permissionMode": PermissionMode.Read,
177
    "resource": "dbs/MultiTenantApp/colls/Products"
178
}
179

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

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

191
permission = user_client.create_permission(full_permission)
192
full_access_token = permission["_token"]
193
```
194

195
### Using Resource Tokens for Authentication
196

197
```python
198
# Create a client using resource token instead of master key
199
resource_token_client = CosmosClient(
200
    url="https://myaccount.documents.azure.com:443/",
201
    credential=resource_token,  # Use resource token instead of master key
202
    consistency_level=ConsistencyLevel.Session
203
)
204

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

218
### Multi-Tenant Application Pattern
219

220
```python
221
def setup_tenant_permissions(database, tenant_id):
222
    """Set up permissions for a new tenant."""
223
    
224
    # Create user for the tenant
225
    user_def = {"id": f"tenant_{tenant_id}"}
226
    try:
227
        user = database.create_user(user_def)
228
    except CosmosResourceExistsError:
229
        user = database.get_user_client(f"tenant_{tenant_id}")
230
    
231
    # Create tenant-specific container if needed
232
    tenant_container_id = f"tenant_{tenant_id}_data"
233
    try:
234
        container = database.create_container(
235
            id=tenant_container_id,
236
            partition_key=PartitionKey(path="/tenantId"),
237
            offer_throughput=400
238
        )
239
    except CosmosResourceExistsError:
240
        pass
241
    
242
    # Grant full access to tenant's own container
243
    tenant_permission = {
244
        "id": f"tenant_{tenant_id}_full_access",
245
        "permissionMode": PermissionMode.All, 
246
        "resource": f"dbs/{database.id}/colls/{tenant_container_id}"
247
    }
248
    
249
    user_client = database.get_user_client(f"tenant_{tenant_id}")
250
    permission = user_client.upsert_permission(tenant_permission)
251
    
252
    # Grant read access to shared reference data
253
    shared_permission = {
254
        "id": f"tenant_{tenant_id}_shared_read",
255
        "permissionMode": PermissionMode.Read,
256
        "resource": f"dbs/{database.id}/colls/SharedReferenceData"
257
    }
258
    
259
    shared_permission_obj = user_client.upsert_permission(shared_permission)
260
    
261
    return {
262
        "tenant_id": tenant_id,
263
        "full_access_token": permission["_token"],
264
        "shared_read_token": shared_permission_obj["_token"]
265
    }
266

267
# Set up permissions for multiple tenants
268
tenant_tokens = {}
269
for tenant_id in ["acme_corp", "widgets_inc", "data_solutions"]:
270
    tokens = setup_tenant_permissions(database, tenant_id)
271
    tenant_tokens[tenant_id] = tokens
272
    print(f"Set up permissions for {tenant_id}")
273
```
274

275
### Permission Lifecycle Management
276

277
```python
278
def manage_user_permissions(database, user_id):
279
    """Demonstrate full permission lifecycle."""
280
    
281
    user_client = database.get_user_client(user_id)
282
    
283
    # List current permissions
284
    permissions = list(user_client.list_permissions())
285
    print(f"User {user_id} has {len(permissions)} permissions")
286
    
287
    for perm in permissions:
288
        print(f"  - {perm['id']}: {perm['permissionMode']} on {perm['resource']}")
289
    
290
    # Query for specific permissions
291
    read_perms = list(user_client.query_permissions(
292
        query="SELECT * FROM permissions p WHERE p.permissionMode = @mode",
293
        parameters=[{"name": "@mode", "value": PermissionMode.Read}]
294
    ))
295
    print(f"Found {len(read_perms)} read-only permissions")
296
    
297
    # Update a permission (upgrade read to full access)
298
    if read_perms:
299
        perm_id = read_perms[0]["id"]
300
        updated_permission = {
301
            "id": perm_id,
302
            "permissionMode": PermissionMode.All,
303
            "resource": read_perms[0]["resource"]
304
        }
305
        
306
        user_client.replace_permission(perm_id, updated_permission)
307
        print(f"Upgraded permission {perm_id} to full access")
308
    
309
    # Clean up expired or unused permissions
310
    permissions_to_delete = []
311
    for perm in permissions:
312
        # Example: delete permissions older than 30 days
313
        # In real scenario, you'd check actual timestamps
314
        if "temp_" in perm["id"]:
315
            permissions_to_delete.append(perm["id"])
316
    
317
    for perm_id in permissions_to_delete:
318
        user_client.delete_permission(perm_id)
319
        print(f"Deleted temporary permission {perm_id}")
320

321
# Manage permissions for a user
322
manage_user_permissions(database, "tenant1_user")
323
```
324

325
### Resource Token Validation and Error Handling
326

327
```python
328
def validate_resource_token_access(resource_token, database_id, container_id):
329
    """Validate what operations a resource token can perform."""
330
    
331
    try:
332
        # Create client with resource token
333
        token_client = CosmosClient(
334
            url="https://myaccount.documents.azure.com:443/",
335
            credential=resource_token
336
        )
337
        
338
        database = token_client.get_database_client(database_id)
339
        container = database.get_container_client(container_id)
340
        
341
        # Test read access
342
        try:
343
            items = list(container.read_all_items(max_item_count=1))
344
            print("✓ Read access confirmed")
345
            read_access = True
346
        except CosmosHttpResponseError:
347
            print("✗ No read access")
348
            read_access = False
349
        
350
        # Test write access
351
        try:
352
            test_item = {
353
                "id": f"test_{uuid.uuid4()}",
354
                "test": True,
355
                "timestamp": datetime.utcnow().isoformat()
356
            }
357
            container.create_item(test_item)
358
            container.delete_item(test_item["id"], partition_key=test_item.get("partitionKey", test_item["id"]))
359
            print("✓ Write access confirmed")
360
            write_access = True
361
        except CosmosHttpResponseError:
362
            print("✗ No write access")
363
            write_access = False
364
        
365
        return {
366
            "valid": True,
367
            "read_access": read_access,
368
            "write_access": write_access
369
        }
370
        
371
    except CosmosHttpResponseError as e:
372
        print(f"Token validation failed: {e.status_code} - {e.message}")
373
        return {
374
            "valid": False,
375
            "error": str(e)
376
        }
377

378
# Validate token access
379
import uuid
380
from datetime import datetime
381

382
token_info = validate_resource_token_access(
383
    resource_token=tenant_tokens["acme_corp"]["full_access_token"],
384
    database_id="MultiTenantApp",
385
    container_id="tenant_acme_corp_data"
386
)
387

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