# # Licensed to the Apache Software Foundation (ASF) under one or more # contributor license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright ownership. # The ASF licenses this file to You under the Apache License, Version 2.0 # (the "License"); you may not use this file except in compliance with # the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # # This script is used to export the IAM policy of a Google Cloud project to a YAML format. # It retrieves the IAM policy bindings, parses the members, and formats the output in a structured # YAML format, excluding service accounts and groups. The output includes usernames, emails, and # their associated permissions, with optional conditions for roles that have conditions attached. # You need to have the Google Cloud SDK installed and authenticated to run this script. import argparse import os import sys import yaml import roles.generate_roles as generate_roles from generate import export_project_iam, to_yaml_file from google.cloud.iam_admin_v1 import GetRoleRequest, IAMClient def migrate_permissions(data: list) -> list: """ Migrates permissions from the permissions to the new roles defined on beam_roles/ directory. The rules are: - If the user has owner role, leave it as is, remove any other role as it is redundant. - If the user has any admin or secret related role, it will be migrated to the beam_admin role. - If the user has an editor role or any user role but not an admin or secret related role, it will be migrated to the beam_infra_manager role. - If the user has a role that is not only viewer, it will be migrated to the beam_committer role. - The users with just viewer roles will be migrated to the beam_viewer role. The rules are in a hierarchical order, meaning that if a user has a high role, it will also have the lower roles. Args: data: A list of dictionaries containing user permissions and details. Returns: A list of dictionaries with migrated permissions. """ migrated_data = [] for item in data: username = item["username"] email = item["email"] permissions = item["permissions"] # Initialize the new roles new_roles = { "beam_owner": False, "beam_admin": False, "beam_infra_manager": False, "beam_committer": False, "beam_viewer": False } for permission in permissions: role = permission["role"] # If the role is 'roles/owner', it is considered an owner role. if role == "roles/owner": new_roles["beam_owner"] = True # If it ends with 'admin' or containes 'secretmanager' in the role, it is considered an admin role. Case insensitive. elif 'admin' in role.lower() or 'secretmanager' in role.lower(): new_roles["beam_admin"] = True new_roles["beam_infra_manager"] = True new_roles["beam_committer"] = True new_roles["beam_viewer"] = True # If it is an editor role, it will be migrated to the beam_infra_manager. elif role == "roles/editor": new_roles["beam_infra_manager"] = True new_roles["beam_committer"] = True new_roles["beam_viewer"] = True elif role != "roles/viewer": # If it is a role that is not only viewer, it will be migrated to the beam_committer role. new_roles["beam_committer"] = True new_roles["beam_viewer"] = True # If it is a viewer role, it will be migrated to the beam_viewer role. else: new_roles["beam_viewer"] = True # Create the migrated entry migrated_entry = { "username": username, "email": email, "permissions": [] } if new_roles["beam_owner"]: migrated_entry["permissions"].append({"role": "roles/owner"}) else: if new_roles["beam_admin"]: migrated_entry["permissions"].append({"role": "projects/PROJECT-ID/roles/beam_admin"}) if new_roles["beam_infra_manager"]: migrated_entry["permissions"].append({"role": "projects/PROJECT-ID/roles/beam_infra_manager"}) if new_roles["beam_committer"]: migrated_entry["permissions"].append({"role": "projects/PROJECT-ID/roles/beam_committer"}) if new_roles["beam_viewer"]: migrated_entry["permissions"].append({"role": "projects/PROJECT-ID/roles/beam_viewer"}) migrated_data.append(migrated_entry) return migrated_data def get_gcp_role_permissions(role_id: str) -> list: """ Retrieves the permissions associated to a google cloud role. Args: project_id: The ID of the Google Cloud project. role_id: The name of the role to retrieve permissions for. Returns: A list of permissions associated with the specified role. """ client = IAMClient() request = GetRoleRequest(name=role_id) role = client.get_role(request=request) return list(role.included_permissions) def get_roles_from_file(file_path: str) -> list: """ Reads a YAML file containing roles and returns a list of dictionaries with user data. Args: file_path: The path to the YAML file containing roles. Returns: A list of dictionaries with user data. """ with open(file_path, 'r') as file: data = yaml.safe_load(file) roles = [] for role in data: email = role.get("email") username = role.get("username") permissions = role.get("permissions", []) roles.append({ "email": email, "username": username, "permissions": permissions }) return roles def permission_differences(project_id: str, user_email: str) -> list: """ Generates a list of differences between the original and migrated permissions for a user. It gets the permission from the generated files, so it is expected that the files are already generated and up to date. Args: project_id: The ID of the Google Cloud project. user_email: The email of the user to compare permissions for. Returns: A list of dictionaries containing the differences in permissions for the specified user. """ cache = {} user_differences = {} original = get_roles_from_file(f"{project_id}.original-roles.yaml") migrated = get_roles_from_file(f"{project_id}.migrated-roles.yaml") # Get the permissions on the beam_roles beam_roles = generate_roles.get_roles() for role_name, role_data in beam_roles.items(): permissions = role_data["permissions"] cache[role_name] = permissions # Get the permissions for the original roles for user in original: username = user["username"] email = user["email"] # Skip if the user email does not match the specified user_email if user_email and email != user_email: continue original_roles = user["permissions"] original_permissions = [] for role in original_roles: if '_withcond_' in role['role']: # Skip roles with conditions, as they are not supported in the new roles continue if 'organizations/' in role['role']: # Skip organization roles, as they are not supported in the new roles continue if role['role'] not in cache: permissions = get_gcp_role_permissions(role["role"]) cache[role['role']] = sorted(permissions) original_permissions.extend(cache[role['role']]) # Initialize the user differences entry user_differences[username] = { "email": email, "original_roles": original_roles, "original_permissions": sorted(original_permissions), "migrated_roles": [], "migrated_permissions": [], "differences": [] } # Get the permissions for the migrated roles for user in migrated: username = user["username"] email = user["email"] # Skip if the user email does not match the specified user_email if user_email and email != user_email: continue migrated_roles = user["permissions"] migrated_permissions = [] for role in migrated_roles: full_role_name = role["role"] # Owner is a special case, it should not be migrated to any other role. if "roles/owner" in full_role_name: migrated_permissions.extend(get_gcp_role_permissions(full_role_name)) else: role_name = full_role_name.split('roles/')[1] migrated_permissions.extend(cache[role_name]) user_differences[username]["migrated_roles"] = migrated_roles user_differences[username]["migrated_permissions"] = sorted(migrated_permissions) # Compare original and migrated permissions differences_list = [] for username, user_data in user_differences.items(): original_permissions = user_data["original_permissions"] migrated_permissions = user_data["migrated_permissions"] # Find differences in permissions original_set = set(original_permissions) migrated_set = set(migrated_permissions) added_permissions = migrated_set.difference(original_set) removed_permissions = original_set.difference(migrated_set) if added_permissions or removed_permissions: differences = { "username": username, "email": user_data["email"], "added_permissions": sorted(list(added_permissions)), "removed_permissions": sorted(list(removed_permissions)) } differences_list.append(differences) return differences_list def main(): """ Main function to run the script. This function parses command-line arguments to either export IAM policies or generate permission differences for a specified GCP project. """ parser = argparse.ArgumentParser( description="Export IAM policies or generate permission differences for a GCP project." ) parser.add_argument( "project_id", help="The Google Cloud project ID." ) parser.add_argument( "--difference", dest="user_email", metavar="USER_EMAIL", help="Generate permission differences for the specified user email." ) args = parser.parse_args() project_id = args.project_id user_email = args.user_email if user_email: # If the iam policy has not been generated yet, it will generate the original IAM policy first. if not os.path.exists(f"{project_id}.original-roles.yaml") or not os.path.exists(f"{project_id}.migrated-roles.yaml"): print(f"Original IAM policy for project {project_id} not found. Generating original and migrated roles first.") print(f"Exporting IAM policy for project {project_id}...") iam_data = export_project_iam(project_id) original_filename = f"{project_id}.original-roles.yaml" original_header = f"Exported original IAM policy for project {project_id}" to_yaml_file(iam_data, original_filename, header_info=original_header) print("Migrating permissions to new roles...") migrated_data = migrate_permissions(iam_data) migrated_filename = f"{project_id}.migrated-roles.yaml" migrated_header = f"Migrated IAM policy for project {project_id} to new beam_roles" to_yaml_file(migrated_data, migrated_filename, header_info=migrated_header) print(f"Generated {original_filename} and {migrated_filename}") print(f"Generating permission differences for {user_email} in project {project_id}...") differences = permission_differences(project_id, user_email) if differences: output_filename = f"{project_id}.permission-differences.yaml" header = f"Permission differences for user {user_email} in project {project_id}" to_yaml_file(differences, output_filename, header_info=header) print(f"Generated {output_filename}") else: print(f"No permission differences found for user {user_email} in project {project_id}.") else: print(f"Exporting IAM policy for project {project_id}...") iam_data = export_project_iam(project_id) original_filename = f"{project_id}.original-roles.yaml" original_header = f"Exported original IAM policy for project {project_id}" to_yaml_file(iam_data, original_filename, header_info=original_header) print("Migrating permissions to new roles...") migrated_data = migrate_permissions(iam_data) migrated_filename = f"{project_id}.migrated-roles.yaml" migrated_header = f"Migrated IAM policy for project {project_id} to new beam_roles" to_yaml_file(migrated_data, migrated_filename, header_info=migrated_header) print(f"Generated {original_filename} and {migrated_filename}") print(f"To generate permission differences, run: python {sys.argv[0]} {project_id} --difference ") if __name__ == "__main__": main()