0
# IAM and Security
1

2
Identity and Access Management integration for queue-level access control, including policy management and permission testing for secure task queue operations.
3

4
## Capabilities
5

6
### IAM Policy Management
7

8
Comprehensive IAM policy management for controlling access to Cloud Tasks queues and operations.
9

10
```python { .api }
11
def get_iam_policy(
12
    self,
13
    request: Union[iam_policy_pb2.GetIamPolicyRequest, dict] = None,
14
    *,
15
    resource: str = None,
16
    retry: OptionalRetry = DEFAULT,
17
    timeout: float = DEFAULT,
18
    metadata: Sequence[Tuple[str, str]] = ()
19
) -> policy_pb2.Policy:
20
    """Get the IAM policy for a queue.
21
    
22
    Args:
23
        request: The request object or dictionary.
24
        resource: Required. The resource for which the policy is being requested.
25
        retry: Designation of what errors should be retried.
26
        timeout: The timeout for this request.
27
        metadata: Strings which should be sent along with the request as metadata.
28
        
29
    Returns:
30
        The IAM policy for the resource.
31
    """
32

33
def set_iam_policy(
34
    self,
35
    request: Union[iam_policy_pb2.SetIamPolicyRequest, dict] = None,
36
    *,
37
    resource: str = None,
38
    retry: OptionalRetry = DEFAULT,
39
    timeout: float = DEFAULT,
40
    metadata: Sequence[Tuple[str, str]] = ()
41
) -> policy_pb2.Policy:
42
    """Set the IAM policy for a queue.
43
    
44
    Args:
45
        request: The request object or dictionary.
46
        resource: Required. The resource for which the policy is being specified.
47
        retry: Designation of what errors should be retried.
48
        timeout: The timeout for this request.
49
        metadata: Strings which should be sent along with the request as metadata.
50
        
51
    Returns:
52
        The updated IAM policy.
53
    """
54

55
def test_iam_permissions(
56
    self,
57
    request: Union[iam_policy_pb2.TestIamPermissionsRequest, dict] = None,
58
    *,
59
    resource: str = None,
60
    permissions: MutableSequence[str] = None,
61
    retry: OptionalRetry = DEFAULT,
62
    timeout: float = DEFAULT,
63
    metadata: Sequence[Tuple[str, str]] = ()
64
) -> iam_policy_pb2.TestIamPermissionsResponse:
65
    """Test IAM permissions on a queue.
66
    
67
    Args:
68
        request: The request object or dictionary.
69
        resource: Required. The resource for which the policy detail is being requested.
70
        permissions: Required. The set of permissions to check for the resource.
71
        retry: Designation of what errors should be retried.
72
        timeout: The timeout for this request.
73
        metadata: Strings which should be sent along with the request as metadata.
74
        
75
    Returns:
76
        The permissions that the caller has on the resource.
77
    """
78
```
79

80
### Location Management
81

82
Access location information and regional service availability.
83

84
```python { .api }
85
def get_location(
86
    self,
87
    request: Union[locations_pb2.GetLocationRequest, dict] = None,
88
    *,
89
    retry: OptionalRetry = DEFAULT,
90
    timeout: float = DEFAULT,
91
    metadata: Sequence[Tuple[str, str]] = ()
92
) -> locations_pb2.Location:
93
    """Get information about a location.
94
    
95
    Args:
96
        request: The request object or dictionary.
97
        retry: Designation of what errors should be retried.
98
        timeout: The timeout for this request.
99
        metadata: Strings which should be sent along with the request as metadata.
100
        
101
    Returns:
102
        The location information.
103
    """
104

105
def list_locations(
106
    self,
107
    request: Union[locations_pb2.ListLocationsRequest, dict] = None,
108
    *,
109
    retry: OptionalRetry = DEFAULT,
110
    timeout: float = DEFAULT,
111
    metadata: Sequence[Tuple[str, str]] = ()
112
) -> locations_pb2.ListLocationsResponse:
113
    """List information about supported locations.
114
    
115
    Args:
116
        request: The request object or dictionary.
117
        retry: Designation of what errors should be retried.
118
        timeout: The timeout for this request.
119
        metadata: Strings which should be sent along with the request as metadata.
120
        
121
    Returns:
122
        The list of locations.
123
    """
124
```
125

126
## Common IAM Roles and Permissions
127

128
### Cloud Tasks Roles
129

130
- **Cloud Tasks Admin** (`roles/cloudtasks.admin`): Full control over queues and tasks
131
- **Cloud Tasks Enqueuer** (`roles/cloudtasks.enqueuer`): Create and delete tasks
132
- **Cloud Tasks Viewer** (`roles/cloudtasks.viewer`): Read-only access to queues and tasks
133
- **Cloud Tasks Queue Admin** (`roles/cloudtasks.queueAdmin`): Manage queues and their configuration
134

135
### Key Permissions
136

137
- `cloudtasks.queues.create`: Create queues
138
- `cloudtasks.queues.get`: Get queue details
139
- `cloudtasks.queues.list`: List queues
140
- `cloudtasks.queues.update`: Update queue configuration
141
- `cloudtasks.queues.delete`: Delete queues
142
- `cloudtasks.tasks.create`: Create tasks
143
- `cloudtasks.tasks.get`: Get task details
144
- `cloudtasks.tasks.list`: List tasks
145
- `cloudtasks.tasks.delete`: Delete tasks
146
- `cloudtasks.tasks.run`: Force task execution
147

148
## Usage Examples
149

150
### Getting IAM Policy
151

152
```python
153
from google.cloud import tasks
154

155
client = tasks.CloudTasksClient()
156
queue_path = client.queue_path('my-project-id', 'us-central1', 'secure-queue')
157

158
# Get current IAM policy for a queue
159
policy = client.get_iam_policy(resource=queue_path)
160

161
print(f'IAM Policy for {queue_path}:')
162
print(f'Version: {policy.version}')
163
print(f'ETag: {policy.etag}')
164

165
for binding in policy.bindings:
166
    print(f'Role: {binding.role}')
167
    print(f'Members: {list(binding.members)}')
168
    if binding.condition:
169
        print(f'Condition: {binding.condition.expression}')
170
```
171

172
### Setting IAM Policy
173

174
```python
175
from google.cloud import tasks
176
from google.iam.v1 import policy_pb2
177

178
client = tasks.CloudTasksClient()
179
queue_path = client.queue_path('my-project-id', 'us-central1', 'secure-queue')
180

181
# Get current policy
182
current_policy = client.get_iam_policy(resource=queue_path)
183

184
# Add a new binding for task enqueuers
185
new_binding = policy_pb2.Binding(
186
    role='roles/cloudtasks.enqueuer',
187
    members=[
188
        'serviceAccount:task-creator@my-project.iam.gserviceaccount.com',
189
        'user:developer@mycompany.com'
190
    ]
191
)
192

193
# Add the binding to the policy
194
current_policy.bindings.append(new_binding)
195

196
# Set the updated policy
197
updated_policy = client.set_iam_policy(resource=queue_path, policy=current_policy)
198
print(f'Updated IAM policy with {len(updated_policy.bindings)} bindings')
199
```
200

201
### Testing IAM Permissions
202

203
```python
204
from google.cloud import tasks
205

206
client = tasks.CloudTasksClient()
207
queue_path = client.queue_path('my-project-id', 'us-central1', 'test-queue')
208

209
# Test if current credentials have specific permissions
210
permissions_to_test = [
211
    'cloudtasks.tasks.create',
212
    'cloudtasks.tasks.delete',
213
    'cloudtasks.queues.get',
214
    'cloudtasks.queues.update'
215
]
216

217
response = client.test_iam_permissions(
218
    resource=queue_path,
219
    permissions=permissions_to_test
220
)
221

222
print('Permissions test results:')
223
for permission in permissions_to_test:
224
    has_permission = permission in response.permissions
225
    print(f'- {permission}: {"✓" if has_permission else "✗"}')
226

227
# Only proceed with operations if permissions are available
228
if 'cloudtasks.tasks.create' in response.permissions:
229
    print('Can create tasks - proceeding with task creation')
230
    # Create task logic here
231
else:
232
    print('Cannot create tasks - insufficient permissions')
233
```
234

235
### Conditional IAM Policies
236

237
```python
238
from google.cloud import tasks
239
from google.iam.v1 import policy_pb2
240
from google.type import expr_pb2
241

242
client = tasks.CloudTasksClient()
243
queue_path = client.queue_path('my-project-id', 'us-central1', 'conditional-queue')
244

245
# Get current policy
246
policy = client.get_iam_policy(resource=queue_path)
247

248
# Create a conditional binding - only allow task creation during business hours
249
condition = expr_pb2.Expr(
250
    title='Business hours only',
251
    description='Allow task creation only during business hours (9 AM to 5 PM UTC)',
252
    expression='request.time.getHours() >= 9 && request.time.getHours() < 17'
253
)
254

255
conditional_binding = policy_pb2.Binding(
256
    role='roles/cloudtasks.enqueuer',
257
    members=['group:developers@mycompany.com'],
258
    condition=condition
259
)
260

261
policy.bindings.append(conditional_binding)
262

263
# Set the updated policy with conditional access
264
updated_policy = client.set_iam_policy(resource=queue_path, policy=policy)
265
print('Set conditional IAM policy for business hours access')
266
```
267

268
### Service Account Management
269

270
```python
271
from google.cloud import tasks
272

273
client = tasks.CloudTasksClient()
274
queue_path = client.queue_path('my-project-id', 'us-central1', 'service-queue')
275

276
# Grant a service account minimal permissions for task operations
277
policy = client.get_iam_policy(resource=queue_path)
278

279
# Add service account with enqueuer role
280
service_account_binding = {
281
    'role': 'roles/cloudtasks.enqueuer',
282
    'members': [
283
        'serviceAccount:task-producer@my-project.iam.gserviceaccount.com'
284
    ]
285
}
286

287
# Add admin role for queue management service account
288
admin_binding = {
289
    'role': 'roles/cloudtasks.queueAdmin',
290
    'members': [
291
        'serviceAccount:queue-manager@my-project.iam.gserviceaccount.com'
292
    ]
293
}
294

295
# Add both bindings
296
policy.bindings.extend([
297
    policy_pb2.Binding(**service_account_binding),
298
    policy_pb2.Binding(**admin_binding)
299
])
300

301
updated_policy = client.set_iam_policy(resource=queue_path, policy=policy)
302
print('Configured service account permissions')
303
```
304

305
### Location Information
306

307
```python
308
from google.cloud import tasks
309

310
client = tasks.CloudTasksClient()
311

312
# List all available locations for Cloud Tasks
313
locations_response = client.list_locations(
314
    name='projects/my-project-id'
315
)
316

317
print('Available Cloud Tasks locations:')
318
for location in locations_response.locations:
319
    print(f'- {location.location_id}: {location.display_name}')
320
    if location.metadata:
321
        print(f'  Metadata: {location.metadata}')
322

323
# Get specific location information
324
location_name = 'projects/my-project-id/locations/us-central1'
325
location_info = client.get_location(name=location_name)
326
print(f'Location details for us-central1: {location_info.display_name}')
327
```
328

329
### Security Best Practices Example
330

331
```python
332
from google.cloud import tasks
333
from google.iam.v1 import policy_pb2
334

335
def setup_secure_queue_permissions(client, queue_path):
336
    """Set up secure IAM permissions for a production queue."""
337
    
338
    # Define role-based access
339
    bindings = [
340
        # Production service accounts for task creation
341
        {
342
            'role': 'roles/cloudtasks.enqueuer',
343
            'members': [
344
                'serviceAccount:prod-app@my-project.iam.gserviceaccount.com',
345
                'serviceAccount:batch-processor@my-project.iam.gserviceaccount.com'
346
            ]
347
        },
348
        
349
        # Operations team for queue management
350
        {
351
            'role': 'roles/cloudtasks.queueAdmin',
352
            'members': [
353
                'group:ops-team@mycompany.com',
354
                'serviceAccount:automation@my-project.iam.gserviceaccount.com'
355
            ]
356
        },
357
        
358
        # Read-only access for monitoring
359
        {
360
            'role': 'roles/cloudtasks.viewer',
361
            'members': [
362
                'group:sre-team@mycompany.com',
363
                'serviceAccount:monitoring@my-project.iam.gserviceaccount.com'
364
            ]
365
        }
366
    ]
367
    
368
    # Create policy with defined bindings
369
    policy = policy_pb2.Policy(
370
        bindings=[policy_pb2.Binding(**binding) for binding in bindings]
371
    )
372
    
373
    # Set the secure policy
374
    updated_policy = client.set_iam_policy(resource=queue_path, policy=policy)
375
    
376
    # Verify permissions were set correctly
377
    test_permissions = [
378
        'cloudtasks.tasks.create',
379
        'cloudtasks.queues.get',
380
        'cloudtasks.queues.update'
381
    ]
382
    
383
    response = client.test_iam_permissions(
384
        resource=queue_path,
385
        permissions=test_permissions
386
    )
387
    
388
    print(f'Secure permissions configured for {queue_path}')
389
    print(f'Verified permissions: {response.permissions}')
390
    
391
    return updated_policy
392

393
# Usage
394
client = tasks.CloudTasksClient()
395
queue_path = client.queue_path('my-project-id', 'us-central1', 'production-queue')
396
setup_secure_queue_permissions(client, queue_path)
397
```