infra/enforcement/iam.py

# 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
from sending import SendingClient
from datetime import datetime, timezone

CONFIG_FILE = "config.yml"
IAC_DRIFT_IAM_USER = "IAC_DRIFT_IAM_USER"

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"IAC_DRIFT_IAM_USER: Unauthorized user '{email}' detected in GCP but 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

        iam_drift_issues = [issue for issue in diff if IAC_DRIFT_IAM_USER in issue]
        general_issues = [issue for issue in diff if IAC_DRIFT_IAM_USER not in issue]

        if general_issues:
            self.logger.info(f"Found {len(general_issues)} general IAM compliance issues. Triggering announcement...")
            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 general_issues:
                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(general_issues)} 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)

        if iam_drift_issues:
            self.logger.info(f"Found {len(iam_drift_issues)} critical IAM security alerts. Dispatching to GitHub security issue...")
            title = f"[{IAC_DRIFT_IAM_USER}] Action Required: Unauthorized IAM Users Detected"
            body = f"Critical security violations detected in IAM policies for project {self.project_id}:\n\n"
            for issue in iam_drift_issues:
                body += f"- {issue}\n"

            announcement = f"URGENT: Dear team,\n\nThis is an automated security alert regarding unauthorized IAM access in project {self.project_id}.\n\n"
            announcement += f"We found {len(iam_drift_issues)} critical security alert(s) that require IMMEDIATE attention.\n"
            announcement += f"\nPlease check the GitHub issue for detailed information and revoke unauthorized access immediately."

            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

        iam_drift_issues = [issue for issue in diff if IAC_DRIFT_IAM_USER in issue]
        general_issues = [issue for issue in diff if IAC_DRIFT_IAM_USER not in issue]

        if general_issues:
            self.logger.info(f"Found {len(general_issues)} general IAM compliance issues. Printing announcement...")
            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 general_issues:
                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(general_issues)} 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)

        if iam_drift_issues:
            self.logger.info("Printing security dashboard update for IAM vulnerabilities...")
            title = f"[{IAC_DRIFT_IAM_USER}] Action Required: Unauthorized IAM Users Detected"
            body = f"Critical security violations detected in IAM policies for project {self.project_id}:\n\n"
            for issue in iam_drift_issues:
                body += f"- {issue}\n"

            announcement = f"URGENT: Dear team,\n\nThis is an automated security alert regarding unauthorized IAM access in project {self.project_id}.\n\n"
            announcement += f"We found {len(iam_drift_issues)} critical security alert(s) that require IMMEDIATE attention.\n"
            announcement += f"\nPlease check the GitHub issue for detailed information and revoke unauthorized access immediately."

            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())