[GCP] Skip checking for iam.serviceAccounts.actAs (#3284)

* check service account policy

* Skip the check for actAs

* add note in doc

* format

* Update docs/source/cloud-setup/cloud-permissions/gcp.rst

Co-authored-by: Zongheng Yang <zongheng.y@gmail.com>

* Address comments

* comment

---------

Co-authored-by: Zongheng Yang <zongheng.y@gmail.com>

docs/source/cloud-setup/cloud-permissions/gcp.rst

@@ -98,6 +98,10 @@ User
 
     For custom VPC users (with :code:`gcp.vpc_name` specified in :code:`~/.sky/config.yaml`, check `here <#_gcp-bring-your-vpc>`_),  :code:`compute.firewalls.create` and :code:`compute.firewalls.delete` are not necessary unless opening ports is needed via `resources.ports` in task yaml.
 
+.. note::
+
+     (Advanced) To further limit the ``iam.serviceAccounts.actAs`` permission to access SkyPilot's service account only, you can remove the permission from the list above and additionally grant your organization's users the ability to use the service account ``skypilot-v1`` created by the admin (see :ref:`Service Account <gcp-service-account-creation>`). This can be done by going to ``IAM & Admin console -> Service Accounts -> skypilot-v1 -> Permissions -> GRANT ACCESS`` and adding the users with role ``roles/iam.serviceAccountUser``. This permits the users to use the ``skypilot-v1`` service account required by SkyPilot.
+
 4. **Optional**: If the user needs to access GCS buckets, you can additionally add the following permissions:
 
 .. code-block:: text

sky/clouds/gcp.py

@@ -743,13 +743,13 @@ def check_credentials(cls) -> Tuple[bool, Optional[str]]:
 
         # This takes user's credential info from "~/.config/gcloud/application_default_credentials.json".  # pylint: disable=line-too-long
         credentials, project = google.auth.default()
-        service = googleapiclient.discovery.build('cloudresourcemanager',
-                                                  'v1',
-                                                  credentials=credentials)
+        crm = googleapiclient.discovery.build('cloudresourcemanager',
+                                              'v1',
+                                              credentials=credentials)
         gcp_minimal_permissions = gcp_utils.get_minimal_permissions()
         permissions = {'permissions': gcp_minimal_permissions}
-        request = service.projects().testIamPermissions(resource=project,
-                                                        body=permissions)
+        request = crm.projects().testIamPermissions(resource=project,
+                                                    body=permissions)
         ret_permissions = request.execute().get('permissions', [])
 
         diffs = set(gcp_minimal_permissions).difference(set(ret_permissions))

sky/provision/gcp/config.py

@@ -242,24 +242,49 @@ def _is_permission_satisfied(service_account, crm, iam, required_permissions,
         # roles, so only call setIamPolicy if needed.
         return True, policy
 
-    for binding in original_policy['bindings']:
-        if member_id in binding['members']:
-            role = binding['role']
-            try:
-                role_definition = iam.projects().roles().get(
-                    name=role).execute()
-            except TypeError as e:
-                if 'does not match the pattern' in str(e):
-                    logger.info('_configure_iam_role: fail to check permission '
-                                f'for built-in role {role}. skipped.')
-                    permissions = []
+    # TODO(zhwu): It is possible that the permission is only granted at the
+    # service-account level, not at the project level. We should check the
+    # permission at both levels.
+    # For example, `roles/iam.serviceAccountUser` can be granted at the
+    # skypilot-v1 service account level, which can be checked with
+    # service_account_policy = iam.projects().serviceAccounts().getIamPolicy(
+    #    resource=f'projects/{project_id}/serviceAcccounts/{email}').execute()
+    # We now skip the check for `iam.serviceAccounts.actAs` permission for
+    # simplicity as it can be granted at the service account level.
+    def check_permissions(policy, required_permissions):
+        for binding in policy['bindings']:
+            if member_id in binding['members']:
+                role = binding['role']
+                logger.info(f'_configure_iam_role: role {role} is attached to '
+                            f'{member_id}...')
+                try:
+                    role_definition = iam.projects().roles().get(
+                        name=role).execute()
+                except TypeError as e:
+                    if 'does not match the pattern' in str(e):
+                        logger.info(
+                            '_configure_iam_role: fail to check permission '
+                            f'for built-in role {role}. Fallback to predefined '
+                            'permission list.')
+                        # Built-in roles cannot be checked for permissions with
+                        # the current API, so we fallback to predefined list
+                        # to find the implied permissions.
+                        permissions = constants.BUILTIN_ROLE_TO_PERMISSIONS.get(
+                            role, [])
+                    else:
+                        raise
                 else:
-                    raise
-            else:
-                permissions = role_definition['includedPermissions']
-            required_permissions -= set(permissions)
-        if not required_permissions:
-            break
+                    permissions = role_definition['includedPermissions']
+                    logger.info(f'_configure_iam_role: role {role} has '
+                                f'permissions {permissions}.')
+                required_permissions -= set(permissions)
+            if not required_permissions:
+                break
+        return required_permissions
+
+    # Check the permissions
+    required_permissions = check_permissions(original_policy,
+                                             required_permissions)
     if not required_permissions:
         # All required permissions are already granted.
         return True, policy

sky/provision/gcp/constants.py

@@ -165,7 +165,10 @@
     'compute.projects.get',
     'compute.zoneOperations.get',
     'iam.roles.get',
-    'iam.serviceAccounts.actAs',
+    # We now skip the check for `iam.serviceAccounts.actAs` permission for
+    # simplicity as it can be granted at the service-account level.
+    # Check: sky.provision.gcp.config::_is_permission_satisfied
+    # 'iam.serviceAccounts.actAs',
     'iam.serviceAccounts.get',
     'serviceusage.services.enable',
     'serviceusage.services.list',
@@ -174,6 +177,20 @@
     'resourcemanager.projects.getIamPolicy',
 ]
 
+# Permissions implied by GCP built-in roles. We hardcode these here, as we
+# cannot get the permissions of built-in role from the GCP Python API.
+# The lists are not exhaustive, but should cover the permissions listed in
+# VM_MINIMAL_PERMISSIONS.
+# Check: sky.provision.gcp.config::_is_permission_satisfied
+BUILTIN_ROLE_TO_PERMISSIONS = {
+    'roles/iam.serviceAccountUser': ['iam.serviceAccounts.actAs'],
+    'roles/iam.serviceAccountViewer': [
+        'iam.serviceAccounts.get', 'iam.serviceAccounts.getIamPolicy'
+    ],
+    # TODO(zhwu): Add more built-in roles to make the permission check more
+    # robust.
+}
+
 FIREWALL_PERMISSIONS = [
     'compute.firewalls.create',
     'compute.firewalls.delete',