#!/usr/bin/env python3
"""AWS IAM permission boundary management agent using boto3."""

import argparse
import json
import logging
import os
import sys
from datetime import datetime
from typing import List

try:
    import boto3
    from botocore.exceptions import ClientError
except ImportError:
    sys.exit("boto3 required: pip install boto3")

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)


def get_iam_client(profile: str = "", region: str = "us-east-1"):
    """Create IAM client with optional profile."""
    session = boto3.Session(profile_name=profile) if profile else boto3.Session()
    return session.client("iam", region_name=region)


def create_permission_boundary(client, policy_name: str, allowed_services: List[str],
                                allowed_regions: List[str] = None) -> dict:
    """Create a permission boundary policy restricting services and regions."""
    statements = [{
        "Sid": "AllowedServices",
        "Effect": "Allow",
        "Action": [f"{svc}:*" for svc in allowed_services],
        "Resource": "*",
    }]
    if allowed_regions:
        statements[0]["Condition"] = {
            "StringEquals": {"aws:RequestedRegion": allowed_regions}
        }
    statements.append({
        "Sid": "DenyBoundaryChanges",
        "Effect": "Deny",
        "Action": ["iam:DeleteRolePermissionsBoundary", "iam:PutRolePermissionsBoundary"],
        "Resource": "*",
    })
    policy_doc = {"Version": "2012-10-17", "Statement": statements}
    try:
        resp = client.create_policy(
            PolicyName=policy_name,
            PolicyDocument=json.dumps(policy_doc),
            Description=f"Permission boundary: {', '.join(allowed_services)}",
        )
        arn = resp["Policy"]["Arn"]
        logger.info("Created boundary policy: %s", arn)
        return {"policy_arn": arn, "policy_document": policy_doc}
    except ClientError as exc:
        return {"error": str(exc)}


def attach_boundary_to_role(client, role_name: str, boundary_arn: str) -> dict:
    """Attach permission boundary to an IAM role."""
    try:
        client.put_role_permissions_boundary(
            RoleName=role_name, PermissionsBoundary=boundary_arn)
        logger.info("Attached boundary %s to role %s", boundary_arn, role_name)
        return {"role": role_name, "boundary_arn": boundary_arn, "attached": True}
    except ClientError as exc:
        return {"role": role_name, "error": str(exc)}


def audit_roles_without_boundary(client) -> List[dict]:
    """Find IAM roles that lack a permission boundary."""
    paginator = client.get_paginator("list_roles")
    unbounded = []
    for page in paginator.paginate():
        for role in page["Roles"]:
            if "PermissionsBoundary" not in role:
                unbounded.append({
                    "role_name": role["RoleName"],
                    "arn": role["Arn"],
                    "created": role["CreateDate"].isoformat(),
                })
    logger.info("Found %d roles without permission boundary", len(unbounded))
    return unbounded


def audit_boundary_effectiveness(client, role_name: str) -> dict:
    """Audit effective permissions for a role with boundary."""
    try:
        role = client.get_role(RoleName=role_name)["Role"]
        boundary = role.get("PermissionsBoundary", {})
        policies_resp = client.list_attached_role_policies(RoleName=role_name)
        inline_resp = client.list_role_policies(RoleName=role_name)
        return {
            "role": role_name,
            "boundary_arn": boundary.get("PermissionsBoundaryArn", "NONE"),
            "attached_policies": [p["PolicyName"] for p in policies_resp["AttachedPolicies"]],
            "inline_policies": inline_resp["PolicyNames"],
        }
    except ClientError as exc:
        return {"role": role_name, "error": str(exc)}


def generate_report(client) -> dict:
    """Generate permission boundary compliance report."""
    unbounded = audit_roles_without_boundary(client)
    report = {
        "analysis_date": datetime.utcnow().isoformat(),
        "roles_without_boundary": unbounded,
        "roles_without_boundary_count": len(unbounded),
        "recommendations": [],
    }
    if unbounded:
        report["recommendations"].append(
            f"Attach permission boundaries to {len(unbounded)} roles lacking boundaries")
    return report


def main():
    parser = argparse.ArgumentParser(description="AWS IAM Permission Boundary Agent")
    parser.add_argument("--profile", default="", help="AWS CLI profile name")
    parser.add_argument("--region", default="us-east-1")
    parser.add_argument("--audit", action="store_true", help="Audit roles without boundaries")
    parser.add_argument("--output-dir", default=".")
    parser.add_argument("--output", default="iam_boundary_report.json")
    args = parser.parse_args()

    os.makedirs(args.output_dir, exist_ok=True)
    client = get_iam_client(args.profile, args.region)
    report = generate_report(client)
    out_path = os.path.join(args.output_dir, args.output)
    with open(out_path, "w") as f:
        json.dump(report, f, indent=2, default=str)
    logger.info("Report saved to %s", out_path)
    print(json.dumps(report, indent=2, default=str))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
AWS IAM Permission Boundary Management Tool

Audits IAM roles for permission boundary compliance, identifies roles
without boundaries, and generates boundary policies based on actual
usage patterns from CloudTrail.

Requirements:
    pip install boto3 pandas
"""

import json
import sys
from datetime import datetime, timezone

try:
    import boto3
    from botocore.exceptions import ClientError
except ImportError:
    print("[ERROR] boto3 is required: pip install boto3")
    sys.exit(1)


class PermissionBoundaryAuditor:
    """Audit IAM roles for permission boundary compliance."""

    def __init__(self, profile_name=None, region="us-east-1"):
        session = boto3.Session(profile_name=profile_name, region_name=region)
        self.iam = session.client("iam")
        self.account_id = session.client("sts").get_caller_identity()["Account"]

    def list_roles_without_boundary(self):
        """Find all IAM roles that do not have a permission boundary attached."""
        roles_without_boundary = []
        paginator = self.iam.get_paginator("list_roles")

        for page in paginator.paginate():
            for role in page["Roles"]:
                role_name = role["RoleName"]
                boundary = role.get("PermissionsBoundary")

                if boundary is None:
                    # Skip AWS service-linked roles
                    if role["Path"].startswith("/aws-service-role/"):
                        continue

                    roles_without_boundary.append({
                        "RoleName": role_name,
                        "Arn": role["Arn"],
                        "CreateDate": role["CreateDate"].isoformat(),
                        "Path": role["Path"],
                        "HasBoundary": False,
                    })

        return roles_without_boundary

    def list_roles_with_boundary(self):
        """List all roles that have a permission boundary and which boundary."""
        roles_with_boundary = []
        paginator = self.iam.get_paginator("list_roles")

        for page in paginator.paginate():
            for role in page["Roles"]:
                boundary = role.get("PermissionsBoundary")
                if boundary:
                    roles_with_boundary.append({
                        "RoleName": role["RoleName"],
                        "Arn": role["Arn"],
                        "BoundaryArn": boundary["PermissionsBoundaryArn"],
                        "BoundaryType": boundary["PermissionsBoundaryType"],
                    })

        return roles_with_boundary

    def verify_boundary_denies_escalation(self, boundary_policy_arn):
        """Check that a permission boundary includes anti-escalation deny statements."""
        try:
            policy = self.iam.get_policy(PolicyArn=boundary_policy_arn)
            version_id = policy["Policy"]["DefaultVersionId"]
            policy_version = self.iam.get_policy_version(
                PolicyArn=boundary_policy_arn, VersionId=version_id
            )
            document = policy_version["PolicyVersion"]["Document"]
        except ClientError as e:
            return {"error": str(e)}

        if isinstance(document, str):
            document = json.loads(document)

        escalation_actions = {
            "iam:DeleteRolePermissionsBoundary",
            "iam:DeleteUserPermissionsBoundary",
            "iam:CreatePolicyVersion",
            "iam:SetDefaultPolicyVersion",
        }

        denied_actions = set()
        for statement in document.get("Statement", []):
            if statement.get("Effect") == "Deny":
                actions = statement.get("Action", [])
                if isinstance(actions, str):
                    actions = [actions]
                denied_actions.update(actions)

        missing_denies = escalation_actions - denied_actions
        return {
            "boundary_arn": boundary_policy_arn,
            "has_escalation_protection": len(missing_denies) == 0,
            "denied_actions": list(denied_actions),
            "missing_deny_actions": list(missing_denies),
        }

    def attach_boundary_to_role(self, role_name, boundary_policy_arn):
        """Attach a permission boundary to an existing IAM role."""
        try:
            self.iam.put_role_permissions_boundary(
                RoleName=role_name,
                PermissionsBoundary=boundary_policy_arn
            )
            return {"success": True, "role": role_name, "boundary": boundary_policy_arn}
        except ClientError as e:
            return {"success": False, "role": role_name, "error": str(e)}

    def generate_audit_report(self):
        """Generate a full compliance report for permission boundaries."""
        roles_without = self.list_roles_without_boundary()
        roles_with = self.list_roles_with_boundary()

        # Check escalation protection for each unique boundary
        boundary_arns = set(r["BoundaryArn"] for r in roles_with)
        boundary_checks = {}
        for arn in boundary_arns:
            boundary_checks[arn] = self.verify_boundary_denies_escalation(arn)

        report = {
            "report_title": "IAM Permission Boundary Compliance Audit",
            "account_id": self.account_id,
            "generated_at": datetime.now(timezone.utc).isoformat(),
            "summary": {
                "total_roles_with_boundary": len(roles_with),
                "total_roles_without_boundary": len(roles_without),
                "unique_boundaries": len(boundary_arns),
                "boundaries_with_escalation_protection": sum(
                    1 for v in boundary_checks.values()
                    if v.get("has_escalation_protection")
                ),
            },
            "roles_without_boundary": roles_without,
            "boundary_escalation_analysis": boundary_checks,
        }

        return report


def generate_boundary_policy(allowed_services, role_prefix="app-",
                              boundary_name="DeveloperBoundary"):
    """Generate a permission boundary policy JSON for given allowed services."""
    service_action_map = {
        "s3": "s3:*",
        "dynamodb": "dynamodb:*",
        "lambda": "lambda:*",
        "sqs": "sqs:*",
        "sns": "sns:*",
        "logs": "logs:*",
        "cloudwatch": "cloudwatch:*",
        "events": "events:*",
        "states": "states:*",
        "xray": "xray:*",
        "ec2": ["ec2:Describe*", "ec2:CreateTags"],
        "ecs": ["ecs:*", "ecr:*"],
        "rds": "rds:*",
        "secretsmanager": "secretsmanager:GetSecretValue",
        "kms": ["kms:Decrypt", "kms:GenerateDataKey", "kms:DescribeKey"],
        "ssm": ["ssm:GetParameter", "ssm:GetParameters", "ssm:GetParametersByPath"],
    }

    actions = []
    for service in allowed_services:
        service_lower = service.lower()
        if service_lower in service_action_map:
            action = service_action_map[service_lower]
            if isinstance(action, list):
                actions.extend(action)
            else:
                actions.append(action)

    policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "AllowedServices",
                "Effect": "Allow",
                "Action": actions,
                "Resource": "*"
            },
            {
                "Sid": "AllowPassRole",
                "Effect": "Allow",
                "Action": "iam:PassRole",
                "Resource": f"arn:aws:iam::*:role/{role_prefix}*",
                "Condition": {
                    "StringEquals": {
                        "iam:PassedToService": [
                            "lambda.amazonaws.com",
                            "ecs-tasks.amazonaws.com",
                            "states.amazonaws.com"
                        ]
                    }
                }
            },
            {
                "Sid": "DenyBoundaryModification",
                "Effect": "Deny",
                "Action": [
                    "iam:DeletePolicy",
                    "iam:DeletePolicyVersion",
                    "iam:CreatePolicyVersion",
                    "iam:SetDefaultPolicyVersion"
                ],
                "Resource": f"arn:aws:iam::*:policy/{boundary_name}"
            },
            {
                "Sid": "DenyBoundaryRemoval",
                "Effect": "Deny",
                "Action": [
                    "iam:DeleteUserPermissionsBoundary",
                    "iam:DeleteRolePermissionsBoundary"
                ],
                "Resource": "*"
            }
        ]
    }

    return json.dumps(policy, indent=2)


if __name__ == "__main__":
    print("=" * 60)
    print("AWS IAM Permission Boundary Management Tool")
    print("=" * 60)
    print()
    print("Usage:")
    print("  auditor = PermissionBoundaryAuditor(profile='prod')")
    print("  report = auditor.generate_audit_report()")
    print()
    print("Generate a boundary policy:")
    services = ["s3", "dynamodb", "lambda", "sqs", "logs", "cloudwatch"]
    policy = generate_boundary_policy(services)
    print(policy)