infra/enforcement/iam.py - beam - Git at Google

 # 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.

 import argparse
 import datetime
 import logging
 import os
 import sys
 import yaml
 from google.api_core import exceptions
 from google.cloud import resourcemanager_v3
 from typing import Optional, List, Dict, Tuple
 from sending import SendingClient

 CONFIG_FILE = "config.yml"

 class IAMPolicyComplianceChecker:

     def is_project_service_account_email(self, email: Optional[str]) -> bool:
         """
         Returns True if the email is not a service account, or if it is a service account and the email contains the project_id.
         """
         if email and email.endswith('.gserviceaccount.com'):
             return self.project_id in email
         return True

     def __init__(self, project_id: str, users_file: str, logger: logging.Logger, sending_client: Optional[SendingClient] = None):
         self.project_id = project_id
         self.users_file = users_file
         self.client = resourcemanager_v3.ProjectsClient()
         self.logger = logger
         self.sending_client = sending_client

     def _parse_member(self, member: str) -> tuple[str, Optional[str], str]:
         """Parses an IAM member string to extract type, email, and a derived username.

         Args:
             member: The IAM member string
         Returns:
             A tuple containing:
                 - username: The derived username from the member string.
                 - email: The email address if available, otherwise None.
                 - member_type: The type of the member (e.g., user, serviceAccount, group).
         """
         email = None
         username = member

         # Split the member string to determine type and identifier
         parts = member.split(':', 1)
         member_type = parts[0] if len(parts) > 1 else "unknown"
         identifier = parts[1] if len(parts) > 1 else member

         if member_type in ["user", "serviceAccount", "group"]:
             email = identifier
             if '@' in identifier:
                 username = identifier.split('@')[0]
             else:
                 username = identifier
         else:
             username = identifier
             member_type = "unknown"
             email = None

         return username, email, member_type

     def _export_project_iam(self) -> List[Dict]:
         """Exports the IAM policy for a given project to YAML format.

         Returns:
             A list of dictionaries containing the IAM policy details.
         """

         try:
             policy = self.client.get_iam_policy(resource=f"projects/{self.project_id}")
             self.logger.debug(f"Retrieved IAM policy for project {self.project_id}")
         except exceptions.NotFound as e:
             self.logger.error(f"Project {self.project_id} not found: {e}")
             raise
         except exceptions.PermissionDenied as e:
             self.logger.error(f"Permission denied for project {self.project_id}: {e}")
             raise
         except Exception as e:
             self.logger.error(f"An error occurred while retrieving IAM policy for project {self.project_id}: {e}")
             raise

         members_data = {}

         for binding in policy.bindings:
             role = binding.role

             for member_str in binding.members:
                 if member_str not in members_data:
                     username, email_address, member_type = self._parse_member(member_str)
                     # Skip service accounts not matching the project_id
                     if member_type == "serviceAccount" and not self.is_project_service_account_email(email_address):
                         self.logger.debug(f"Skipping service account not matching project_id ({self.project_id}): {email_address}")
                         continue
                     if member_type == "unknown":
                         self.logger.warning(f"Skipping member {member_str} with no email address")
                         continue  # Skip if no email address is found, probably a malformed member
                     members_data[member_str] = {
                         "username": username,
                         "email": email_address,
                         "member_type": member_type,
                         "permissions": []
                     }

                 # Skip permissions that have a condition
                 if "withcond" in role:
                     continue

                 permission_entry = {}
                 permission_entry["role"] = role

                 members_data[member_str]["permissions"].append(permission_entry)

         output_list = []
         for data in members_data.values():
             data["permissions"] = sorted(data["permissions"], key=lambda p: p["role"])
             output_list.append({
                 "username": data["username"],
                 "email": data["email"],
                 "member_type": data["member_type"],
                 "permissions": data["permissions"]
             })

         output_list.sort(key=lambda x: x["username"])
         return output_list

     def _read_project_iam_file(self) -> List[Dict]:
         """Reads the IAM policy from a YAML file.

         Returns:
             A list of dictionaries containing the IAM policy details.
         """
         try:
             with open(self.users_file, "r") as file:
                 iam_policy = yaml.safe_load(file)


                 self.logger.debug(f"Retrieved IAM policy from file for project {self.project_id}")
                 return iam_policy
         except FileNotFoundError:
             self.logger.error(f"IAM policy file not found for project {self.project_id}")
             return []
         except Exception as e:
             self.logger.error(f"An error occurred while reading IAM policy file for project {self.project_id}: {e}")
             return []

     def _to_yaml_file(self, data: List[Dict], output_file: str, header_info: str = "") -> None:
         """
         Writes a list of dictionaries to a YAML file.
         Include the apache license header on the files

         Args:
             data: A list of dictionaries containing user permissions and details.
             output_file: The file path where the YAML output will be written.
             header_info: A string containing the header information to be included in the YAML file.
         """

         apache_license_header = """# 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.
     """

         # Prepare the header with the Apache license
         header = f"{apache_license_header}\n# {header_info}\n# Generated on {datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%d %H:%M:%S')} UTC\n\n"

         try:
             with open(output_file, "w") as file:
                 file.write(header)
                 yaml.dump(data, file, sort_keys=False, default_flow_style=False, indent=2)
             self.logger.info(f"Successfully wrote IAM policy data to {output_file}")
         except IOError as e:
             self.logger.error(f"Failed to write to {output_file}: {e}")
             raise

     def check_compliance(self) -> List[str]:
         """
         Checks the compliance of the IAM policy against the defined policies.

         Returns:
             A list of strings describing any compliance issues found.
         """

         current_users = {user['email']: user for user in self._export_project_iam() if self.is_project_service_account_email(user.get('email'))}
         existing_users = {user['email']: user for user in self._read_project_iam_file() if self.is_project_service_account_email(user.get('email'))}

         if not existing_users:
             error_msg = f"No IAM policy found in the {self.users_file}."
             self.logger.info(error_msg)
             raise RuntimeError(error_msg)

         differences = []

         all_emails = set(current_users.keys()) | set(existing_users.keys())

         for email in sorted(list(all_emails)):
             current_user = current_users.get(email)
             existing_user = existing_users.get(email)

             if current_user and not existing_user:
                 differences.append(f"User {email} not found in existing policy.")
             elif not current_user and existing_user:
                 differences.append(f"User {email} found in policy file but not in GCP.")
             elif current_user and existing_user:
                 if current_user.get("member_type") != existing_user.get("member_type"):
                     differences.append(f"User {email} has different member type. In GCP: {current_user.get('member_type')}, in file: {existing_user.get('member_type')}")
                 if current_user["permissions"] != existing_user["permissions"]:
                     msg = f"\nPermissions for user {email} differ."
                     msg += f"\nIn GCP: {current_user['permissions']}"
                     msg += f"\nIn {self.users_file}: {existing_user['permissions']}"
                     self.logger.info(msg)
                     differences.append(msg)

         return differences

     def create_announcement(self, recipient: str) -> None:
         """
         Creates an announcement about compliance issues using the SendingClient.

         Args:
             recipient (str): The email address of the announcement recipient.
         """
         if not self.sending_client:
             raise ValueError("SendingClient is required for creating announcements")

         diff = self.check_compliance()

         if not diff:
             self.logger.info("No compliance issues found, no announcement will be created.")
             return

         title = f"IAM Policy Non-Compliance Detected"
         body = f"IAM policy for project {self.project_id} is not compliant with the defined policies on {self.users_file}\n\n"
         for issue in diff:
             body += f"- {issue}\n"

         announcement = f"Dear team,\n\nThis is an automated notification about compliance issues detected in the IAM policy for project {self.project_id}.\n\n"
         announcement += f"We found {len(diff)} compliance issue(s) that need your attention.\n"
         announcement += f"\nPlease check the GitHub issue for detailed information and take appropriate action to resolve these compliance violations."

         self.sending_client.create_announcement(title, body, recipient, announcement)

     def print_announcement(self, recipient: str) -> None:
         """
         Prints announcement details instead of sending them (for testing purposes).

         Args:
             recipient (str): The email address of the announcement recipient.
         """
         if not self.sending_client:
             raise ValueError("SendingClient is required for printing announcements")

         diff = self.check_compliance()

         if not diff:
             self.logger.info("No compliance issues found, no announcement will be printed.")
             return

         title = f"IAM Policy Non-Compliance Detected"
         body = f"IAM policy for project {self.project_id} is not compliant with the defined policies on {self.users_file}\n\n"
         for issue in diff:
             body += f"- {issue}\n"

         announcement = f"Dear team,\n\nThis is an automated notification about compliance issues detected in the IAM policy for project {self.project_id}.\n\n"
         announcement += f"We found {len(diff)} compliance issue(s) that need your attention.\n"
         announcement += f"\nPlease check the GitHub issue for detailed information and take appropriate action to resolve these compliance violations."

         self.sending_client.print_announcement(title, body, recipient, announcement)

     def generate_compliance(self) -> None:
         """
         Modifies the users file to match the current IAM policy.
         If no changes are needed, no file will be written.
         """

         try:
             diff = self.check_compliance()
         except RuntimeError:
             self.logger.info("No existing IAM policy found.")
             diff = ["No existing policy found"]

         if not diff or (len(diff) == 1 and "No existing policy found" not in diff[0]):
             self.logger.info("No compliance issues found, no changes will be made.")
             return

         current_policy = self._export_project_iam()
         header_info = f"IAM policy for project {self.project_id}"

         self._to_yaml_file(current_policy, self.users_file, header_info)
         self.logger.info(f"Generated new compliance file: {self.users_file}")

 def config_process() -> Dict[str, str]:
     with open(CONFIG_FILE, "r") as file:
         config = yaml.safe_load(file)

     if not config:
         raise ValueError("Configuration file is empty or invalid.")

     config_res = dict()

     config_res["project_id"] = config.get("project_id", "apache-beam-testing")
     config_res["logging_level"] = config.get("logging", {}).get("level", "INFO")
     config_res["logging_format"] = config.get("logging", {}).get("format", "[%(asctime)s] %(levelname)s: %(message)s")
     config_res["users_file"] = config.get("users_file", "../iam/users.yml")
     config_res["action"] = config.get("action", "check")

     # SendingClient configuration
     config_res["github_token"] = os.getenv("GITHUB_TOKEN", "")
     config_res["github_repo"] = os.getenv("GITHUB_REPOSITORY", "apache/beam")
     config_res["smtp_server"] = os.getenv("SMTP_SERVER", "")
     config_res["smtp_port"] = os.getenv("SMTP_PORT", 587)
     config_res["email"] = os.getenv("EMAIL_ADDRESS", "")
     config_res["password"] = os.getenv("EMAIL_PASSWORD", "")
     config_res["recipient"] = os.getenv("EMAIL_RECIPIENT", "")

     return config_res

 def main():
     # Parse command line arguments
     parser = argparse.ArgumentParser(description="IAM Policy Compliance Checker")
     parser.add_argument("--action", choices=["check", "announce", "print", "generate"],
                        help="Action to perform: check compliance, create announcement, print announcement, or generate new compliance")
     args = parser.parse_args()

     config = config_process()

     # Command line argument takes precedence over config file
     action = args.action if args.action else config.get("action", "check")

     logging.basicConfig(level=getattr(logging, config["logging_level"].upper(), logging.INFO),
                         format=config["logging_format"])
     logger = logging.getLogger("IAMPolicyComplianceChecker")

     # Create SendingClient if needed for announcement actions
     sending_client = None
     if action in ["announce", "print"]:
         try:
             # Provide default values for testing, especially for print action
             github_token = config["github_token"] or "dummy-token"
             github_repo = config["github_repo"] or "dummy/repo"
             smtp_server = config["smtp_server"] or "dummy-server"
             smtp_port = int(config["smtp_port"]) if config["smtp_port"] else 587
             email = config["email"] or "dummy@example.com"
             password = config["password"] or "dummy-password"

             sending_client = SendingClient(
                 logger=logger,
                 github_token=github_token,
                 github_repo=github_repo,
                 smtp_server=smtp_server,
                 smtp_port=smtp_port,
                 email=email,
                 password=password
             )
         except Exception as e:
             logger.error(f"Failed to initialize SendingClient: {e}")
             return 1

     logger.info(f"Starting IAM policy compliance check with action: {action}")
     iam_checker = IAMPolicyComplianceChecker(config["project_id"], config["users_file"], logger, sending_client)

     try:
         if action == "check":
             compliance_issues = iam_checker.check_compliance()
             if compliance_issues:
                 logger.warning("IAM policy compliance issues found:")
                 for issue in compliance_issues:
                     logger.warning(issue)
             else:
                 logger.info("IAM policy is compliant.")
         elif action == "announce":
             logger.info("Creating announcement for compliance violations...")
             recipient = config["recipient"] or "admin@example.com"
             iam_checker.create_announcement(recipient)
         elif action == "print":
             logger.info("Printing announcement for compliance violations...")
             recipient = config["recipient"] or "admin@example.com"
             iam_checker.print_announcement(recipient)
         elif action == "generate":
             logger.info("Generating new compliance based on current IAM policy...")
             iam_checker.generate_compliance()
         else:
             logger.error(f"Unknown action: {action}")
             return 1
     except Exception as e:
         logger.error(f"Error executing action '{action}': {e}")
         return 1

     return 0

 if __name__ == "__main__":

     sys.exit(main())
	# 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.

	import argparse
	import datetime
	import logging
	import os
	import sys
	import yaml
	from google.api_core import exceptions
	from google.cloud import resourcemanager_v3
	from typing import Optional, List, Dict, Tuple
	from sending import SendingClient

	CONFIG_FILE = "config.yml"

	class IAMPolicyComplianceChecker:

	def is_project_service_account_email(self, email: Optional[str]) -> bool:
	"""
	Returns True if the email is not a service account, or if it is a service account and the email contains the project_id.
	"""
	if email and email.endswith('.gserviceaccount.com'):
	return self.project_id in email
	return True

	def __init__(self, project_id: str, users_file: str, logger: logging.Logger, sending_client: Optional[SendingClient] = None):
	self.project_id = project_id
	self.users_file = users_file
	self.client = resourcemanager_v3.ProjectsClient()
	self.logger = logger
	self.sending_client = sending_client

	def _parse_member(self, member: str) -> tuple[str, Optional[str], str]:
	"""Parses an IAM member string to extract type, email, and a derived username.

	Args:
	member: The IAM member string
	Returns:
	A tuple containing:
	- username: The derived username from the member string.
	- email: The email address if available, otherwise None.
	- member_type: The type of the member (e.g., user, serviceAccount, group).
	"""
	email = None
	username = member

	# Split the member string to determine type and identifier
	parts = member.split(':', 1)
	member_type = parts[0] if len(parts) > 1 else "unknown"
	identifier = parts[1] if len(parts) > 1 else member

	if member_type in ["user", "serviceAccount", "group"]:
	email = identifier
	if '@' in identifier:
	username = identifier.split('@')[0]
	else:
	username = identifier
	else:
	username = identifier
	member_type = "unknown"
	email = None

	return username, email, member_type

	def _export_project_iam(self) -> List[Dict]:
	"""Exports the IAM policy for a given project to YAML format.

	Returns:
	A list of dictionaries containing the IAM policy details.
	"""

	try:
	policy = self.client.get_iam_policy(resource=f"projects/{self.project_id}")
	self.logger.debug(f"Retrieved IAM policy for project {self.project_id}")
	except exceptions.NotFound as e:
	self.logger.error(f"Project {self.project_id} not found: {e}")
	raise
	except exceptions.PermissionDenied as e:
	self.logger.error(f"Permission denied for project {self.project_id}: {e}")
	raise
	except Exception as e:
	self.logger.error(f"An error occurred while retrieving IAM policy for project {self.project_id}: {e}")
	raise

	members_data = {}

	for binding in policy.bindings:
	role = binding.role

	for member_str in binding.members:
	if member_str not in members_data:
	username, email_address, member_type = self._parse_member(member_str)
	# Skip service accounts not matching the project_id
	if member_type == "serviceAccount" and not self.is_project_service_account_email(email_address):
	self.logger.debug(f"Skipping service account not matching project_id ({self.project_id}): {email_address}")
	continue
	if member_type == "unknown":
	self.logger.warning(f"Skipping member {member_str} with no email address")
	continue # Skip if no email address is found, probably a malformed member
	members_data[member_str] = {
	"username": username,
	"email": email_address,
	"member_type": member_type,
	"permissions": []
	}

	# Skip permissions that have a condition
	if "withcond" in role:
	continue

	permission_entry = {}
	permission_entry["role"] = role

	members_data[member_str]["permissions"].append(permission_entry)

	output_list = []
	for data in members_data.values():
	data["permissions"] = sorted(data["permissions"], key=lambda p: p["role"])
	output_list.append({
	"username": data["username"],
	"email": data["email"],
	"member_type": data["member_type"],
	"permissions": data["permissions"]
	})

	output_list.sort(key=lambda x: x["username"])
	return output_list

	def _read_project_iam_file(self) -> List[Dict]:
	"""Reads the IAM policy from a YAML file.

	Returns:
	A list of dictionaries containing the IAM policy details.
	"""
	try:
	with open(self.users_file, "r") as file:
	iam_policy = yaml.safe_load(file)


	self.logger.debug(f"Retrieved IAM policy from file for project {self.project_id}")
	return iam_policy
	except FileNotFoundError:
	self.logger.error(f"IAM policy file not found for project {self.project_id}")
	return []
	except Exception as e:
	self.logger.error(f"An error occurred while reading IAM policy file for project {self.project_id}: {e}")
	return []

	def _to_yaml_file(self, data: List[Dict], output_file: str, header_info: str = "") -> None:
	"""
	Writes a list of dictionaries to a YAML file.
	Include the apache license header on the files

	Args:
	data: A list of dictionaries containing user permissions and details.
	output_file: The file path where the YAML output will be written.
	header_info: A string containing the header information to be included in the YAML file.
	"""

	apache_license_header = """# 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.
	"""

	# Prepare the header with the Apache license
	header = f"{apache_license_header}\n# {header_info}\n# Generated on {datetime.datetime.now(datetime.timezone.utc).strftime('%Y-%m-%d %H:%M:%S')} UTC\n\n"

	try:
	with open(output_file, "w") as file:
	file.write(header)
	yaml.dump(data, file, sort_keys=False, default_flow_style=False, indent=2)
	self.logger.info(f"Successfully wrote IAM policy data to {output_file}")
	except IOError as e:
	self.logger.error(f"Failed to write to {output_file}: {e}")
	raise

	def check_compliance(self) -> List[str]:
	"""
	Checks the compliance of the IAM policy against the defined policies.

	Returns:
	A list of strings describing any compliance issues found.
	"""

	current_users = {user['email']: user for user in self._export_project_iam() if self.is_project_service_account_email(user.get('email'))}
	existing_users = {user['email']: user for user in self._read_project_iam_file() if self.is_project_service_account_email(user.get('email'))}

	if not existing_users:
	error_msg = f"No IAM policy found in the {self.users_file}."
	self.logger.info(error_msg)
	raise RuntimeError(error_msg)

	differences = []

	all_emails = set(current_users.keys()) \| set(existing_users.keys())

	for email in sorted(list(all_emails)):
	current_user = current_users.get(email)
	existing_user = existing_users.get(email)

	if current_user and not existing_user:
	differences.append(f"User {email} not found in existing policy.")
	elif not current_user and existing_user:
	differences.append(f"User {email} found in policy file but not in GCP.")
	elif current_user and existing_user:
	if current_user.get("member_type") != existing_user.get("member_type"):
	differences.append(f"User {email} has different member type. In GCP: {current_user.get('member_type')}, in file: {existing_user.get('member_type')}")
	if current_user["permissions"] != existing_user["permissions"]:
	msg = f"\nPermissions for user {email} differ."
	msg += f"\nIn GCP: {current_user['permissions']}"
	msg += f"\nIn {self.users_file}: {existing_user['permissions']}"
	self.logger.info(msg)
	differences.append(msg)

	return differences

	def create_announcement(self, recipient: str) -> None:
	"""
	Creates an announcement about compliance issues using the SendingClient.

	Args:
	recipient (str): The email address of the announcement recipient.
	"""
	if not self.sending_client:
	raise ValueError("SendingClient is required for creating announcements")

	diff = self.check_compliance()

	if not diff:
	self.logger.info("No compliance issues found, no announcement will be created.")
	return

	title = f"IAM Policy Non-Compliance Detected"
	body = f"IAM policy for project {self.project_id} is not compliant with the defined policies on {self.users_file}\n\n"
	for issue in diff:
	body += f"- {issue}\n"

	announcement = f"Dear team,\n\nThis is an automated notification about compliance issues detected in the IAM policy for project {self.project_id}.\n\n"
	announcement += f"We found {len(diff)} compliance issue(s) that need your attention.\n"
	announcement += f"\nPlease check the GitHub issue for detailed information and take appropriate action to resolve these compliance violations."

	self.sending_client.create_announcement(title, body, recipient, announcement)

	def print_announcement(self, recipient: str) -> None:
	"""
	Prints announcement details instead of sending them (for testing purposes).

	Args:
	recipient (str): The email address of the announcement recipient.
	"""
	if not self.sending_client:
	raise ValueError("SendingClient is required for printing announcements")

	diff = self.check_compliance()

	if not diff:
	self.logger.info("No compliance issues found, no announcement will be printed.")
	return

	title = f"IAM Policy Non-Compliance Detected"
	body = f"IAM policy for project {self.project_id} is not compliant with the defined policies on {self.users_file}\n\n"
	for issue in diff:
	body += f"- {issue}\n"

	announcement = f"Dear team,\n\nThis is an automated notification about compliance issues detected in the IAM policy for project {self.project_id}.\n\n"
	announcement += f"We found {len(diff)} compliance issue(s) that need your attention.\n"
	announcement += f"\nPlease check the GitHub issue for detailed information and take appropriate action to resolve these compliance violations."

	self.sending_client.print_announcement(title, body, recipient, announcement)

	def generate_compliance(self) -> None:
	"""
	Modifies the users file to match the current IAM policy.
	If no changes are needed, no file will be written.
	"""

	try:
	diff = self.check_compliance()
	except RuntimeError:
	self.logger.info("No existing IAM policy found.")
	diff = ["No existing policy found"]

	if not diff or (len(diff) == 1 and "No existing policy found" not in diff[0]):
	self.logger.info("No compliance issues found, no changes will be made.")
	return

	current_policy = self._export_project_iam()
	header_info = f"IAM policy for project {self.project_id}"

	self._to_yaml_file(current_policy, self.users_file, header_info)
	self.logger.info(f"Generated new compliance file: {self.users_file}")

	def config_process() -> Dict[str, str]:
	with open(CONFIG_FILE, "r") as file:
	config = yaml.safe_load(file)

	if not config:
	raise ValueError("Configuration file is empty or invalid.")

	config_res = dict()

	config_res["project_id"] = config.get("project_id", "apache-beam-testing")
	config_res["logging_level"] = config.get("logging", {}).get("level", "INFO")
	config_res["logging_format"] = config.get("logging", {}).get("format", "[%(asctime)s] %(levelname)s: %(message)s")
	config_res["users_file"] = config.get("users_file", "../iam/users.yml")
	config_res["action"] = config.get("action", "check")

	# SendingClient configuration
	config_res["github_token"] = os.getenv("GITHUB_TOKEN", "")
	config_res["github_repo"] = os.getenv("GITHUB_REPOSITORY", "apache/beam")
	config_res["smtp_server"] = os.getenv("SMTP_SERVER", "")
	config_res["smtp_port"] = os.getenv("SMTP_PORT", 587)
	config_res["email"] = os.getenv("EMAIL_ADDRESS", "")
	config_res["password"] = os.getenv("EMAIL_PASSWORD", "")
	config_res["recipient"] = os.getenv("EMAIL_RECIPIENT", "")

	return config_res

	def main():
	# Parse command line arguments
	parser = argparse.ArgumentParser(description="IAM Policy Compliance Checker")
	parser.add_argument("--action", choices=["check", "announce", "print", "generate"],
	help="Action to perform: check compliance, create announcement, print announcement, or generate new compliance")
	args = parser.parse_args()

	config = config_process()

	# Command line argument takes precedence over config file
	action = args.action if args.action else config.get("action", "check")

	logging.basicConfig(level=getattr(logging, config["logging_level"].upper(), logging.INFO),
	format=config["logging_format"])
	logger = logging.getLogger("IAMPolicyComplianceChecker")

	# Create SendingClient if needed for announcement actions
	sending_client = None
	if action in ["announce", "print"]:
	try:
	# Provide default values for testing, especially for print action
	github_token = config["github_token"] or "dummy-token"
	github_repo = config["github_repo"] or "dummy/repo"
	smtp_server = config["smtp_server"] or "dummy-server"
	smtp_port = int(config["smtp_port"]) if config["smtp_port"] else 587
	email = config["email"] or "dummy@example.com"
	password = config["password"] or "dummy-password"

	sending_client = SendingClient(
	logger=logger,
	github_token=github_token,
	github_repo=github_repo,
	smtp_server=smtp_server,
	smtp_port=smtp_port,
	email=email,
	password=password
	)
	except Exception as e:
	logger.error(f"Failed to initialize SendingClient: {e}")
	return 1

	logger.info(f"Starting IAM policy compliance check with action: {action}")
	iam_checker = IAMPolicyComplianceChecker(config["project_id"], config["users_file"], logger, sending_client)

	try:
	if action == "check":
	compliance_issues = iam_checker.check_compliance()
	if compliance_issues:
	logger.warning("IAM policy compliance issues found:")
	for issue in compliance_issues:
	logger.warning(issue)
	else:
	logger.info("IAM policy is compliant.")
	elif action == "announce":
	logger.info("Creating announcement for compliance violations...")
	recipient = config["recipient"] or "admin@example.com"
	iam_checker.create_announcement(recipient)
	elif action == "print":
	logger.info("Printing announcement for compliance violations...")
	recipient = config["recipient"] or "admin@example.com"
	iam_checker.print_announcement(recipient)
	elif action == "generate":
	logger.info("Generating new compliance based on current IAM policy...")
	iam_checker.generate_compliance()
	else:
	logger.error(f"Unknown action: {action}")
	return 1
	except Exception as e:
	logger.error(f"Error executing action '{action}': {e}")
	return 1

	return 0

	if __name__ == "__main__":

	sys.exit(main())