0
# gRPC Services
1
2
Client and server classes for IAM policy service operations. This module provides both client stubs for making remote calls to IAM services and server-side servicer base classes for implementing IAM services.
3
4
## Capabilities
5
6
### Client Service Stub
7
8
Client stub for making gRPC calls to IAM policy services. Provides methods for setting, getting, and testing permissions on Google Cloud resources.
9
10
```python { .api }
11
class IAMPolicyStub:
12
"""
13
Client stub for IAM Policy service.
14
15
Args:
16
channel: gRPC channel to the service endpoint
17
"""
18
def __init__(self, channel): ...
19
20
def SetIamPolicy(self, request: SetIamPolicyRequest, timeout=None, metadata=None, credentials=None) -> Policy:
21
"""
22
Sets the access control policy on a resource.
23
24
Args:
25
request (SetIamPolicyRequest): The request containing resource and policy
26
timeout (float, optional): Request timeout in seconds
27
metadata (List[Tuple[str, str]], optional): gRPC metadata
28
credentials (grpc.CallCredentials, optional): Call credentials
29
30
Returns:
31
Policy: The updated policy with etag
32
33
Raises:
34
grpc.RpcError: If the RPC fails
35
"""
36
37
def GetIamPolicy(self, request: GetIamPolicyRequest, timeout=None, metadata=None, credentials=None) -> Policy:
38
"""
39
Gets the access control policy for a resource.
40
41
Args:
42
request (GetIamPolicyRequest): The request containing resource name
43
timeout (float, optional): Request timeout in seconds
44
metadata (List[Tuple[str, str]], optional): gRPC metadata
45
credentials (grpc.CallCredentials, optional): Call credentials
46
47
Returns:
48
Policy: The current IAM policy for the resource
49
50
Raises:
51
grpc.RpcError: If the RPC fails
52
"""
53
54
def TestIamPermissions(self, request: TestIamPermissionsRequest, timeout=None, metadata=None, credentials=None) -> TestIamPermissionsResponse:
55
"""
56
Tests the specified permissions against the IAM access control policy.
57
58
Args:
59
request (TestIamPermissionsRequest): The request containing resource and permissions
60
timeout (float, optional): Request timeout in seconds
61
metadata (List[Tuple[str, str]], optional): gRPC metadata
62
credentials (grpc.CallCredentials, optional): Call credentials
63
64
Returns:
65
TestIamPermissionsResponse: Response containing permissions the caller has
66
67
Raises:
68
grpc.RpcError: If the RPC fails
69
"""
70
```
71
72
### Server Service Implementation
73
74
Base class for implementing IAM policy services on the server side.
75
76
```python { .api }
77
class IAMPolicyServicer:
78
"""
79
Base class for implementing IAM Policy service.
80
81
Subclass this to provide actual implementations of the service methods.
82
"""
83
84
def SetIamPolicy(self, request: SetIamPolicyRequest, context: grpc.ServicerContext) -> Policy:
85
"""
86
Sets the access control policy on a resource.
87
88
Args:
89
request (SetIamPolicyRequest): The request containing resource and policy
90
context (grpc.ServicerContext): gRPC service context
91
92
Returns:
93
Policy: The updated policy
94
95
Raises:
96
NotImplementedError: Must be implemented by subclass
97
"""
98
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
99
context.set_details('Method not implemented!')
100
raise NotImplementedError('Method not implemented!')
101
102
def GetIamPolicy(self, request: GetIamPolicyRequest, context: grpc.ServicerContext) -> Policy:
103
"""
104
Gets the access control policy for a resource.
105
106
Args:
107
request (GetIamPolicyRequest): The request containing resource name
108
context (grpc.ServicerContext): gRPC service context
109
110
Returns:
111
Policy: The current IAM policy
112
113
Raises:
114
NotImplementedError: Must be implemented by subclass
115
"""
116
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
117
context.set_details('Method not implemented!')
118
raise NotImplementedError('Method not implemented!')
119
120
def TestIamPermissions(self, request: TestIamPermissionsRequest, context: grpc.ServicerContext) -> TestIamPermissionsResponse:
121
"""
122
Tests permissions against the IAM access control policy.
123
124
Args:
125
request (TestIamPermissionsRequest): The request containing resource and permissions
126
context (grpc.ServicerContext): gRPC service context
127
128
Returns:
129
TestIamPermissionsResponse: Response with permissions the caller has
130
131
Raises:
132
NotImplementedError: Must be implemented by subclass
133
"""
134
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
135
context.set_details('Method not implemented!')
136
raise NotImplementedError('Method not implemented!')
137
```
138
139
### Service Registration
140
141
Utility function for registering an IAM policy servicer with a gRPC server.
142
143
```python { .api }
144
def add_IAMPolicyServicer_to_server(servicer: IAMPolicyServicer, server: grpc.Server) -> None:
145
"""
146
Registers an IAM Policy servicer with a gRPC server.
147
148
Args:
149
servicer (IAMPolicyServicer): The servicer implementation
150
server (grpc.Server): The gRPC server to register with
151
"""
152
```
153
154
### Request/Response Messages
155
156
Core message types used in gRPC service operations.
157
158
```python { .api }
159
class SetIamPolicyRequest:
160
"""
161
Request to set IAM policy on a resource.
162
163
Attributes:
164
resource (str): Resource name (e.g., "projects/my-project/topics/my-topic")
165
policy (Policy): The IAM policy to set
166
update_mask (google.protobuf.FieldMask): Optional field mask for partial updates
167
"""
168
resource: str
169
policy: Policy
170
update_mask: google.protobuf.FieldMask
171
172
class GetIamPolicyRequest:
173
"""
174
Request to get IAM policy from a resource.
175
176
Attributes:
177
resource (str): Resource name
178
options (GetPolicyOptions): Optional policy retrieval options
179
"""
180
resource: str
181
options: GetPolicyOptions
182
183
class TestIamPermissionsRequest:
184
"""
185
Request to test permissions on a resource.
186
187
Attributes:
188
resource (str): Resource name
189
permissions (List[str]): Permissions to test (e.g., ["storage.objects.get", "storage.objects.create"])
190
"""
191
resource: str
192
permissions: List[str]
193
194
class TestIamPermissionsResponse:
195
"""
196
Response containing permissions the caller has.
197
198
Attributes:
199
permissions (List[str]): Subset of requested permissions that the caller has
200
"""
201
permissions: List[str]
202
```
203
204
## Usage Examples
205
206
### Client Usage
207
208
```python
209
import grpc
210
from google.iam.v1 import iam_policy_pb2_grpc
211
from google.iam.v1 import iam_policy_pb2
212
from google.iam.v1 import policy_pb2
213
214
# Create secure channel to IAM service
215
credentials = grpc.ssl_channel_credentials()
216
channel = grpc.secure_channel('iam.googleapis.com:443', credentials)
217
218
# Create client stub
219
client = iam_policy_pb2_grpc.IAMPolicyStub(channel)
220
221
# Set IAM policy
222
policy = policy_pb2.Policy()
223
binding = policy_pb2.Binding()
224
binding.role = "roles/viewer"
225
binding.members.append("user:alice@example.com")
226
policy.bindings.append(binding)
227
228
set_request = iam_policy_pb2.SetIamPolicyRequest()
229
set_request.resource = "projects/my-project/topics/my-topic"
230
set_request.policy.CopyFrom(policy)
231
232
try:
233
response = client.SetIamPolicy(set_request, timeout=30.0)
234
print(f"Policy set successfully, etag: {response.etag}")
235
except grpc.RpcError as e:
236
print(f"Failed to set policy: {e.code()}: {e.details()}")
237
238
# Get IAM policy
239
get_request = iam_policy_pb2.GetIamPolicyRequest()
240
get_request.resource = "projects/my-project/topics/my-topic"
241
242
try:
243
policy_response = client.GetIamPolicy(get_request)
244
print(f"Retrieved policy with {len(policy_response.bindings)} bindings")
245
for binding in policy_response.bindings:
246
print(f" Role: {binding.role}, Members: {list(binding.members)}")
247
except grpc.RpcError as e:
248
print(f"Failed to get policy: {e.code()}: {e.details()}")
249
250
# Test permissions
251
test_request = iam_policy_pb2.TestIamPermissionsRequest()
252
test_request.resource = "projects/my-project/topics/my-topic"
253
test_request.permissions.extend([
254
"pubsub.topics.get",
255
"pubsub.topics.publish",
256
"pubsub.topics.delete"
257
])
258
259
try:
260
permissions_response = client.TestIamPermissions(test_request)
261
print(f"User has permissions: {list(permissions_response.permissions)}")
262
except grpc.RpcError as e:
263
print(f"Failed to test permissions: {e.code()}: {e.details()}")
264
265
# Clean up
266
channel.close()
267
```
268
269
### Server Implementation
270
271
```python
272
import grpc
273
from concurrent import futures
274
from google.iam.v1 import iam_policy_pb2_grpc
275
from google.iam.v1 import iam_policy_pb2
276
from google.iam.v1 import policy_pb2
277
278
class MyIAMPolicyService(iam_policy_pb2_grpc.IAMPolicyServicer):
279
"""Custom IAM policy service implementation."""
280
281
def __init__(self):
282
# In practice, you'd have a database or other storage
283
self.policies = {}
284
285
def SetIamPolicy(self, request, context):
286
"""Implement policy setting logic."""
287
try:
288
# Validate resource name
289
if not request.resource:
290
context.abort(grpc.StatusCode.INVALID_ARGUMENT, "Resource name is required")
291
292
# Store policy (in practice, save to database)
293
self.policies[request.resource] = request.policy
294
295
# Return the policy with etag for optimistic concurrency
296
response_policy = policy_pb2.Policy()
297
response_policy.CopyFrom(request.policy)
298
response_policy.etag = b"etag-" + request.resource.encode()
299
300
return response_policy
301
302
except Exception as e:
303
context.abort(grpc.StatusCode.INTERNAL, f"Internal error: {str(e)}")
304
305
def GetIamPolicy(self, request, context):
306
"""Implement policy retrieval logic."""
307
try:
308
if request.resource not in self.policies:
309
context.abort(grpc.StatusCode.NOT_FOUND, "Policy not found")
310
311
return self.policies[request.resource]
312
313
except Exception as e:
314
context.abort(grpc.StatusCode.INTERNAL, f"Internal error: {str(e)}")
315
316
def TestIamPermissions(self, request, context):
317
"""Implement permission testing logic."""
318
try:
319
# In practice, you'd evaluate permissions against stored policies
320
# This is a simplified example
321
granted_permissions = []
322
323
if request.resource in self.policies:
324
policy = self.policies[request.resource]
325
# Simple check - in practice would be more complex
326
for binding in policy.bindings:
327
if "user:current-user@example.com" in binding.members:
328
# Grant all requested permissions for demo
329
granted_permissions = list(request.permissions)
330
break
331
332
response = iam_policy_pb2.TestIamPermissionsResponse()
333
response.permissions.extend(granted_permissions)
334
return response
335
336
except Exception as e:
337
context.abort(grpc.StatusCode.INTERNAL, f"Internal error: {str(e)}")
338
339
# Create and start server
340
def serve():
341
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
342
343
# Register service
344
iam_policy_pb2_grpc.add_IAMPolicyServicer_to_server(
345
MyIAMPolicyService(), server
346
)
347
348
# Listen on port
349
listen_addr = '[::]:50051'
350
server.add_insecure_port(listen_addr)
351
352
print(f"Starting server on {listen_addr}")
353
server.start()
354
server.wait_for_termination()
355
356
if __name__ == '__main__':
357
serve()
358
```