gcpdiag.queries.iam
Queries related to GCP Identity and Access Management.
class
Role(gcpdiag.models.Resource):
33class Role(models.Resource): 34 """Represents an IAM role""" 35 36 def __init__(self, resource_data): 37 try: 38 project_id = utils.get_project_by_res_name(resource_data['name']) 39 except ValueError: 40 project_id = None 41 42 super().__init__(project_id=project_id) 43 self._resource_data = resource_data 44 45 @property 46 def name(self) -> str: 47 return self._resource_data['name'] 48 49 @property 50 def full_path(self) -> str: 51 return self.name 52 53 @property 54 def permissions(self) -> List[str]: 55 # roles should usually include one or more permissions 56 return self._resource_data.get('includedPermissions', [])
Represents an IAM role
class
RoleNotFoundError(builtins.Exception):
Common base class for all non-exit exceptions.
class
BaseIAMPolicy(gcpdiag.models.Resource):
160class BaseIAMPolicy(models.Resource): 161 """Common class for IAM policies""" 162 163 _name: str 164 _policy_by_member: Dict[str, Any] 165 166 @property 167 def full_path(self): 168 return self._name 169 170 @abc.abstractmethod 171 def _is_resource_permission(self, permission: str) -> bool: 172 """Checks that a permission is applicable to the resource 173 174 Any role can be assigned on a resource level but only a subset of 175 permissions will be relevant to a resource 176 Irrelevant permissions are ignored in `has_role_permissions` method 177 """ 178 pass 179 180 def _expand_policy(self, resource_data: Dict[str, Any]) -> Dict[str, Any]: 181 """Groups `getIamPolicy` bindings by member 182 183 API response contains a list of bindings of a role to members: 184 { 185 "bindings": [ 186 { 187 "role": "roles/resourcemanager.organizationAdmin", 188 "members": [ 189 "user:mike@example.com", 190 "serviceAccount:my-project-id@appspot.gserviceaccount.com" 191 ] 192 }, 193 ... 194 } 195 196 This method will convert those bindings into the following structure: 197 { 198 "user:mike@example.com": { 199 "roles": { "roles/resourcemanager.organizationAdmin" }, 200 }, 201 "serviceAccount:my-project-id@appspot.gserviceaccount.com": { 202 "roles": { "roles/resourcemanager.organizationAdmin" }, 203 }, 204 } 205 """ 206 207 policy_roles = set() 208 policy_by_member: Dict[str, Any] = collections.defaultdict(dict) 209 210 # Empty lists are omitted in GCP API responses 211 for binding in resource_data.get('bindings', []): 212 if 'condition' in binding: 213 logging.warning( 214 'IAM binding contains a condition, which would be ignored: %s', 215 binding, 216 ) 217 218 # IAM binding should always have a role and at least one member 219 policy_roles.add(binding['role']) 220 for member in binding['members']: 221 member_policy = policy_by_member[member] 222 member_policy.setdefault('roles', set()).add(binding['role']) 223 224 # Populate cache for IAM roles used in the policy 225 # Unlike `has_role_permissions` this part will be executed inside 226 # `prefetch_rule` and will benefit from multi-threading execution 227 for role in policy_roles: 228 # Ignore all errors - there could be no rules involving this role 229 try: 230 _get_iam_role(role, self.project_id) 231 except (RoleNotFoundError, utils.GcpApiError) as err: 232 # Ignore roles if cannot retrieve a role 233 # For example, due to lack of permissions 234 if isinstance(err, utils.GcpApiError): 235 if hooks.is_inspection_env() and err.status == 403: 236 logging.warning( 237 "Unable to get IAM role '%s' due to lack of permissions, ignoring.", 238 role, 239 ) 240 else: 241 logging.error('API failure getting IAM roles: %s', err) 242 raise utils.GcpApiError(err) from err 243 elif isinstance(err, RoleNotFoundError): 244 logging.warning("Unable to get IAM role '%s', ignoring: %s", role, err) 245 246 # Populate cache for service accounts used in the policy 247 # Note: not implemented as a generator expression because 248 # it looks ugly without assignment expressions, available 249 # only with Python >= 3.8. 250 sa_emails = set() 251 for member in policy_by_member.keys(): 252 # Note: not matching / makes sure that we don't match for example fleet 253 # workload identities: 254 # https://cloud.google.com/anthos/multicluster-management/fleets/workload-identity 255 m = re.match(r'serviceAccount:([^/]+)$', member) 256 if m: 257 sa_emails.add(m.group(1)) 258 _batch_fetch_service_accounts(list(sa_emails), self.context) 259 260 return policy_by_member 261 262 def _expand_member_policy(self, member: str): 263 """Expands member roles into set of permissions 264 265 Permissions are using "lazy" initialization and only expanded if needed 266 """ 267 member_policy = self._policy_by_member.get(member) 268 if not member_policy or 'permissions' in member_policy: 269 return 270 271 permissions = set() 272 for role in member_policy['roles']: 273 try: 274 permissions.update(_get_iam_role(role, self.project_id).permissions) 275 except (RoleNotFoundError, utils.GcpApiError) as err: 276 if isinstance(err, utils.GcpApiError): 277 if hooks.is_inspection_env() and err.status == 403: 278 logging.warning( 279 "Unable to get IAM role '%s' due to lack of permissions, ignoring.", 280 role, 281 ) 282 else: 283 logging.error('API failure getting IAM roles: %s', err) 284 raise utils.GcpApiError(err) from err 285 elif isinstance(err, RoleNotFoundError): 286 logging.warning("Unable to find IAM role '%s', ignoring: %s", role, err) 287 member_policy['permissions'] = permissions 288 289 def _is_active_member(self, member: str) -> bool: 290 """Checks that the member isn't disabled 291 292 Currently supports only service accounts and not other account types 293 Used in `has_role_permissions` and similar methods to ensure that 294 the member isn't disabled and permissions are effectively working 295 """ 296 297 # If this is a service account, make sure that the service account is enabled. 298 # Note: not matching / makes sure that we don't match for example fleet 299 # workload identities: 300 # https://cloud.google.com/anthos/multicluster-management/fleets/workload-identity 301 m = re.match(r'serviceAccount:([^/]+)$', member) 302 if m: 303 if not is_service_account_enabled(m.group(1), self.context): 304 logging.info('service account %s is disabled', m.group(1)) 305 return False 306 307 return True 308 309 def __init__( 310 self, 311 project_id: Optional[str], 312 name: str, 313 resource_data: Dict[str, Any], 314 context: models.Context, 315 ): 316 super().__init__(project_id) 317 self._name = name 318 self.context = context 319 self._policy_by_member = self._expand_policy(resource_data) 320 321 def get_member_permissions(self, member: str) -> List[str]: 322 """Return permissions for a member (either a user or serviceAccount). 323 324 The "member" can be a user or a service account and must be specified with 325 the IAM member syntax, i.e. using the prefixes `user:` or `serviceAccount:`. 326 """ 327 328 if member not in self._policy_by_member: 329 return [] 330 331 self._expand_member_policy(member) 332 return sorted(self._policy_by_member[member]['permissions']) 333 334 def get_members(self) -> List[str]: 335 """Returns the IAM members of the project. 336 337 The "member" can be a user or a service account and is specified with 338 the IAM member syntax, i.e. using the prefixes `user:` or `serviceAccount:`. 339 """ 340 return list(self._policy_by_member.keys()) 341 342 def get_member_type(self, member) -> Optional[str]: 343 """Returns the IAM members of the project. 344 345 The "member" can be a user or a service account and is specified with 346 the IAM member syntax, i.e. using the prefixes `user:` or `serviceAccount:`. 347 """ 348 for m in self._policy_by_member.keys(): 349 parts = m.split(':') 350 if member == parts[1]: 351 return parts[0] 352 return None 353 354 def has_permission(self, member: str, permission: str) -> bool: 355 """Return true if user or service account member has this permission. 356 357 Note that any indirect bindings, for example through group membership, 358 aren't supported and only direct bindings to this member are checked. 359 """ 360 361 if member not in self._policy_by_member: 362 return False 363 364 self._expand_member_policy(member) 365 if permission not in self._policy_by_member[member]['permissions']: 366 return False 367 return self._is_active_member(member) 368 369 def has_any_permission(self, member: str, permission: set[str]) -> bool: 370 """Return true if user or service account member has any of these permission. 371 372 Note that any indirect bindings, for example through group membership, 373 aren't supported and only direct bindings to this member are checked 374 """ 375 376 if member not in self._policy_by_member: 377 return False 378 379 self._expand_member_policy(member) 380 if any(p in self._policy_by_member[member]['permissions'] for p in permission): 381 return True 382 return self._is_active_member(member) 383 384 def _has_role(self, member: str, role: str) -> bool: 385 """Checks that the member has this role 386 387 It performs exact match and doesn't expand role to list of permissions. 388 Note that this method is not public because users of this module should 389 use has_role_permissions(), i.e. verify effective permissions instead of 390 roles. 391 """ 392 393 if member not in self._policy_by_member: 394 return False 395 396 if role not in self._policy_by_member[member]['roles']: 397 return False 398 return self._is_active_member(member) 399 400 def has_role_permissions(self, member: str, role: str) -> bool: 401 """Checks that this member has all the permissions defined by this role.""" 402 403 if member not in self._policy_by_member: 404 return False 405 406 # Avoid expanding roles to permissions 407 if self._has_role(member, role): 408 # member status was already checked in `has_role` 409 return True 410 411 self._expand_member_policy(member) 412 role_permissions = { 413 p for p in _get_iam_role(role, self.project_id).permissions if self._is_resource_permission(p) 414 } 415 416 missing_roles = role_permissions - self._policy_by_member[member]['permissions'] 417 if missing_roles: 418 logging.debug( 419 "member '%s' doesn't have permissions %s", 420 member, 421 ','.join(missing_roles), 422 ) 423 return False 424 return self._is_active_member(member)
Common class for IAM policies
full_path
Returns the full path of this resource.
Example: 'projects/gcpdiag-gke-1-9b90/zones/europe-west4-a/clusters/gke1'
def
get_member_permissions(self, member: str) -> List[str]:
321 def get_member_permissions(self, member: str) -> List[str]: 322 """Return permissions for a member (either a user or serviceAccount). 323 324 The "member" can be a user or a service account and must be specified with 325 the IAM member syntax, i.e. using the prefixes `user:` or `serviceAccount:`. 326 """ 327 328 if member not in self._policy_by_member: 329 return [] 330 331 self._expand_member_policy(member) 332 return sorted(self._policy_by_member[member]['permissions'])
Return permissions for a member (either a user or serviceAccount).
The "member" can be a user or a service account and must be specified with
the IAM member syntax, i.e. using the prefixes user: or serviceAccount:.
def
get_members(self) -> List[str]:
334 def get_members(self) -> List[str]: 335 """Returns the IAM members of the project. 336 337 The "member" can be a user or a service account and is specified with 338 the IAM member syntax, i.e. using the prefixes `user:` or `serviceAccount:`. 339 """ 340 return list(self._policy_by_member.keys())
Returns the IAM members of the project.
The "member" can be a user or a service account and is specified with
the IAM member syntax, i.e. using the prefixes user: or serviceAccount:.
def
get_member_type(self, member) -> Optional[str]:
342 def get_member_type(self, member) -> Optional[str]: 343 """Returns the IAM members of the project. 344 345 The "member" can be a user or a service account and is specified with 346 the IAM member syntax, i.e. using the prefixes `user:` or `serviceAccount:`. 347 """ 348 for m in self._policy_by_member.keys(): 349 parts = m.split(':') 350 if member == parts[1]: 351 return parts[0] 352 return None
Returns the IAM members of the project.
The "member" can be a user or a service account and is specified with
the IAM member syntax, i.e. using the prefixes user: or serviceAccount:.
def
has_permission(self, member: str, permission: str) -> bool:
354 def has_permission(self, member: str, permission: str) -> bool: 355 """Return true if user or service account member has this permission. 356 357 Note that any indirect bindings, for example through group membership, 358 aren't supported and only direct bindings to this member are checked. 359 """ 360 361 if member not in self._policy_by_member: 362 return False 363 364 self._expand_member_policy(member) 365 if permission not in self._policy_by_member[member]['permissions']: 366 return False 367 return self._is_active_member(member)
Return true if user or service account member has this permission.
Note that any indirect bindings, for example through group membership, aren't supported and only direct bindings to this member are checked.
def
has_any_permission(self, member: str, permission: set[str]) -> bool:
369 def has_any_permission(self, member: str, permission: set[str]) -> bool: 370 """Return true if user or service account member has any of these permission. 371 372 Note that any indirect bindings, for example through group membership, 373 aren't supported and only direct bindings to this member are checked 374 """ 375 376 if member not in self._policy_by_member: 377 return False 378 379 self._expand_member_policy(member) 380 if any(p in self._policy_by_member[member]['permissions'] for p in permission): 381 return True 382 return self._is_active_member(member)
Return true if user or service account member has any of these permission.
Note that any indirect bindings, for example through group membership, aren't supported and only direct bindings to this member are checked
def
has_role_permissions(self, member: str, role: str) -> bool:
400 def has_role_permissions(self, member: str, role: str) -> bool: 401 """Checks that this member has all the permissions defined by this role.""" 402 403 if member not in self._policy_by_member: 404 return False 405 406 # Avoid expanding roles to permissions 407 if self._has_role(member, role): 408 # member status was already checked in `has_role` 409 return True 410 411 self._expand_member_policy(member) 412 role_permissions = { 413 p for p in _get_iam_role(role, self.project_id).permissions if self._is_resource_permission(p) 414 } 415 416 missing_roles = role_permissions - self._policy_by_member[member]['permissions'] 417 if missing_roles: 418 logging.debug( 419 "member '%s' doesn't have permissions %s", 420 member, 421 ','.join(missing_roles), 422 ) 423 return False 424 return self._is_active_member(member)
Checks that this member has all the permissions defined by this role.
def
fetch_iam_policy( request, resource_class: Type[BaseIAMPolicy], project_id: Optional[str], name: str, context: gcpdiag.models.Context, raise_error_if_fails=True):
427def fetch_iam_policy( 428 request, 429 resource_class: Type[BaseIAMPolicy], 430 project_id: Optional[str], 431 name: str, 432 context: models.Context, 433 raise_error_if_fails=True, 434): 435 """Executes `getIamPolicy` request and converts into a resource class 436 437 Supposed to be used by `get_*_policy` functions in gcpdiag.queries.* and 438 requires an API request, which can be executed, to be passed in parameters 439 440 An abstract policy request should look like: 441 class ResourcePolicy(BaseIAMPolicy): 442 pass 443 444 def get_resource_policy(name): 445 api_request = get_api(..).resources().get(name=name) 446 ... 447 return fetch_iam_policy(api_request, ResourcePolicy, project_id, name) 448 449 Note: API calls aren't cached and it should be done externally 450 """ 451 452 logging.info("fetching IAM policy of '%s'", name) 453 try: 454 response = request.execute(num_retries=config.API_RETRIES) 455 except googleapiclient.errors.HttpError as err: 456 if raise_error_if_fails: 457 raise utils.GcpApiError(err) from err 458 else: 459 return 460 return resource_class(project_id, name, response, context)
Executes getIamPolicy request and converts into a resource class
Supposed to be used by get_*_policy functions in gcpdiag.queries.* and
requires an API request, which can be executed, to be passed in parameters
An abstract policy request should look like:
class ResourcePolicy(BaseIAMPolicy): pass
def get_resource_policy(name): api_request = get_api(..).resources().get(name=name) ... return fetch_iam_policy(api_request, ResourcePolicy, project_id, name)
Note: API calls aren't cached and it should be done externally
463class ProjectPolicy(BaseIAMPolicy): 464 """Represents the IAM policy of a single project. 465 466 Note that you should use the get_project_policy() method so that the 467 objects are cached and you don't re-fetch the project policy. 468 469 See also the API documentation: 470 https://cloud.google.com/resource-manager/reference/rest/v1/projects/getIamPolicy 471 """ 472 473 def _is_resource_permission(self, permission: str) -> bool: 474 # Filter out permissions that can be granted only on organization or folders 475 # It also excludes some permissions that aren't supported in custom roles 476 # 477 # https://cloud.google.com/resource-manager/docs/access-control-proj#permissions 478 # https://cloud.google.com/monitoring/access-control#custom_roles 479 if permission.startswith('resourcemanager.projects.') or permission.startswith( 480 'stackdriver.projects.' 481 ): 482 return False 483 return True
Represents the IAM policy of a single project.
Note that you should use the get_project_policy() method so that the objects are cached and you don't re-fetch the project policy.
See also the API documentation: https://cloud.google.com/resource-manager/reference/rest/v1/projects/getIamPolicy
@caching.cached_api_call(in_memory=True)
def
get_project_policy( context: gcpdiag.models.Context, raise_error_if_fails=True) -> ProjectPolicy:
486@caching.cached_api_call(in_memory=True) 487def get_project_policy(context: models.Context, raise_error_if_fails=True) -> ProjectPolicy: 488 """Return the ProjectPolicy object for a project, caching the result.""" 489 project_id = context.project_id 490 resource_name = f'projects/{project_id}' 491 492 crm_api = apis.get_api('cloudresourcemanager', 'v3', project_id) 493 request = crm_api.projects().getIamPolicy(resource='projects/' + project_id) 494 return fetch_iam_policy( 495 request, 496 ProjectPolicy, 497 project_id, 498 resource_name, 499 context, 500 raise_error_if_fails, 501 )
Return the ProjectPolicy object for a project, caching the result.
504class OrganizationPolicy(BaseIAMPolicy): 505 """Represents the IAM policy of a single organization using v1 API. 506 507 See also the API documentation: 508 https://cloud.google.com/resource-manager/reference/rest/v1/organizations/getIamPolicy 509 """ 510 511 def _is_resource_permission(self, permission: str) -> bool: 512 # Filter out permissions that can be granted only on projects or folders 513 if permission.startswith('resourcemanager.projects.') or permission.startswith( 514 'resourcemanager.folders.' 515 ): 516 return False 517 return True
Represents the IAM policy of a single organization using v1 API.
See also the API documentation: https://cloud.google.com/resource-manager/reference/rest/v1/organizations/getIamPolicy
@caching.cached_api_call(in_memory=True)
def
get_organization_policy( context: gcpdiag.models.Context, organization_id: str, raise_error_if_fails=True) -> OrganizationPolicy:
520@caching.cached_api_call(in_memory=True) 521def get_organization_policy( 522 context: models.Context, organization_id: str, raise_error_if_fails=True 523) -> OrganizationPolicy: 524 """Return the OrganizationPolicy object for an organization, caching the result.""" 525 526 resource_name = f'organizations/{organization_id}' 527 528 crm_api = apis.get_api('cloudresourcemanager', 'v1') 529 request = crm_api.organizations().getIamPolicy(resource=resource_name) 530 return fetch_iam_policy( 531 request, 532 OrganizationPolicy, 533 None, 534 resource_name, 535 context, 536 raise_error_if_fails, 537 )
Return the OrganizationPolicy object for an organization, caching the result.
class
ServiceAccount(gcpdiag.models.Resource):
540class ServiceAccount(models.Resource): 541 """Class represents the service account. 542 543 Add more fields as needed from the declaration: 544 https://cloud.google.com/iam/docs/reference/rest/v1/projects.serviceAccounts#ServiceAccount 545 """ 546 547 _resource_data: dict 548 549 def __init__(self, project_id, resource_data): 550 super().__init__(project_id=project_id) 551 self._resource_data = resource_data 552 553 @property 554 def name(self) -> str: 555 return self._resource_data['name'] 556 557 @property 558 def email(self) -> str: 559 return self._resource_data['email'] 560 561 @property 562 def unique_id(self) -> str: 563 return self._resource_data['uniqueId'] 564 565 @property 566 def disabled(self) -> bool: 567 return self._resource_data.get('disabled', False) 568 569 @property 570 def full_path(self) -> str: 571 # example: "name": 572 # "projects/skanzhelev-gke-dev/serviceAccounts/test-service-account-1 573 # @skanzhelev-gke-dev.iam.gserviceaccount.com" 574 return self.name 575 576 @property 577 def short_path(self) -> str: 578 path = self.full_path 579 path = re.sub(r'^projects/', '', path) 580 path = re.sub(r'/serviceAccounts/', '/', path) 581 return path
Class represents the service account.
Add more fields as needed from the declaration: https://cloud.google.com/iam/docs/reference/rest/v1/projects.serviceAccounts#ServiceAccount
full_path: str
569 @property 570 def full_path(self) -> str: 571 # example: "name": 572 # "projects/skanzhelev-gke-dev/serviceAccounts/test-service-account-1 573 # @skanzhelev-gke-dev.iam.gserviceaccount.com" 574 return self.name
Returns the full path of this resource.
Example: 'projects/gcpdiag-gke-1-9b90/zones/europe-west4-a/clusters/gke1'
short_path: str
576 @property 577 def short_path(self) -> str: 578 path = self.full_path 579 path = re.sub(r'^projects/', '', path) 580 path = re.sub(r'/serviceAccounts/', '/', path) 581 return path
Returns the short name for this resource.
Note that it isn't clear from this name what kind of resource it is.
Example: 'gke1'
SERVICE_AGENT_DOMAINS =
('cloudservices.gserviceaccount.com', 'cloudbuild.gserviceaccount.com', 'cloudcomposer-accounts.iam.gserviceaccount.com', 'cloud-filer.iam.gserviceaccount.com', 'cloud-memcache-sa.iam.gserviceaccount.com', 'cloud-ml.google.com.iam.gserviceaccount.com', 'cloud-redis.iam.gserviceaccount.com', 'cloud-tpu.iam.gserviceaccount.com', 'compute-system.iam.gserviceaccount.com', 'container-analysis.iam.gserviceaccount.com', 'container-engine-robot.iam.gserviceaccount.com', 'containerregistry.iam.gserviceaccount.com', 'dataflow-service-producer-prod.iam.gserviceaccount.com', 'dataproc-accounts.iam.gserviceaccount.com', 'dlp-api.iam.gserviceaccount.com', 'endpoints-portal.iam.gserviceaccount.com', 'firebase-rules.iam.gserviceaccount.com', 'gae-api-prod.google.com.iam.gserviceaccount.com', 'gcf-admin-robot.iam.gserviceaccount.com', 'gcp-gae-service.iam.gserviceaccount.com', 'genomics-api.google.com.iam.gserviceaccount.com', 'remotebuildexecution.iam.gserviceaccount.com', 'serverless-robot-prod.iam.gserviceaccount.com', 'service-consumer-management.iam.gserviceaccount.com', 'service-networking.iam.gserviceaccount.com', 'sourcerepo-service-accounts.iam.gserviceaccount.com', 'appspot.gserviceaccount.com', 'cloudservices.gserviceaccount.com', 'crashlytics-bigquery-prod.iam.gserviceaccount.com', 'fcm-bq-export-prod.iam.gserviceaccount.com', 'firebase-sa-management.iam.gserviceaccount.com', 'performance-bq-export-prod.iam.gserviceaccount.com', 'predictions-bq-export-prod.iam.gserviceaccount.com', 'system.gserviceaccount.com')
DEFAULT_SERVICE_ACCOUNT_DOMAINS =
('appspot.gserviceaccount.com', 'developer.gserviceaccount.com')
def
is_service_account_existing(email: str, context: gcpdiag.models.Context) -> bool:
784def is_service_account_existing(email: str, context: models.Context) -> bool: 785 """Verify that a service account exists. 786 787 If we get a non-404 API error when retrieving the service account, we will 788 assume 789 that the service account exists, not to throw false positives (but 790 a warning will be printed out). 791 """ 792 # Make sure that the service account is fetched (this is also 793 # called by get_project_policy). 794 _batch_fetch_service_accounts([email], context) 795 return email not in _service_account_cache_is_not_found
Verify that a service account exists.
If we get a non-404 API error when retrieving the service account, we will assume that the service account exists, not to throw false positives (but a warning will be printed out).
def
is_service_account_enabled(email: str, context: gcpdiag.models.Context) -> bool:
798def is_service_account_enabled(email: str, context: models.Context) -> bool: 799 """Verify that a service account exists and is enabled. 800 801 If we get an API error when retrieving the service account, we will assume 802 that the service account is enabled, not to throw false positives (but 803 a warning will be printed out). 804 """ 805 _batch_fetch_service_accounts([email], context) 806 return (email not in _service_account_cache_is_not_found) and not ( 807 email in _service_account_cache and _service_account_cache[email].disabled 808 )
Verify that a service account exists and is enabled.
If we get an API error when retrieving the service account, we will assume that the service account is enabled, not to throw false positives (but a warning will be printed out).
811class ServiceAccountIAMPolicy(BaseIAMPolicy): 812 def _is_resource_permission(self, permission): 813 return True
Common class for IAM policies
@caching.cached_api_call(in_memory=True)
def
get_service_account_iam_policy( context: gcpdiag.models.Context, service_account: str) -> ServiceAccountIAMPolicy:
816@caching.cached_api_call(in_memory=True) 817def get_service_account_iam_policy( 818 context: models.Context, service_account: str 819) -> ServiceAccountIAMPolicy: 820 """Returns an IAM policy for a service account""" 821 project_id = context.project_id 822 resource_name = f'projects/{project_id}/serviceAccounts/{service_account}' 823 824 iam_api = apis.get_api('iam', 'v1', project_id) 825 request = iam_api.projects().serviceAccounts().getIamPolicy(resource=resource_name) 826 return fetch_iam_policy(request, ServiceAccountIAMPolicy, project_id, resource_name, context)
Returns an IAM policy for a service account
@caching.cached_api_call(in_memory=True)
def
get_service_account_list(project_id: str) -> List[ServiceAccount]:
829@caching.cached_api_call(in_memory=True) 830def get_service_account_list(project_id: str) -> List[ServiceAccount]: 831 """Returns list of service accounts""" 832 833 iam_api = apis.get_api('iam', 'v1', project_id) 834 project_name = f'projects/{project_id}' 835 request = iam_api.projects().serviceAccounts().list(name=project_name, pageSize=100) 836 try: 837 response = request.execute(num_retries=config.API_RETRIES) 838 except googleapiclient.errors.HttpError as err: 839 raise utils.GcpApiError(err) from err 840 return [ 841 ServiceAccount(project_id, service_account) for service_account in response.get('accounts', []) 842 ]
Returns list of service accounts
