AWS Security Best Practices: Comprehensive Guide for 2026

mkdir -p aws-security/{src,config,scripts}
cd aws-security
python3 -m venv venv
source venv/bin/activate
pip install --upgrade pip

pip install boto3==1.34.0 botocore==1.34.0 requests==2.31.0 python-dotenv==1.0.0

# Configure AWS CLI (if not already done)
aws configure

# Or set environment variables
export AWS_ACCESS_KEY_ID=your_access_key
export AWS_SECRET_ACCESS_KEY=your_secret_key
export AWS_DEFAULT_REGION=us-east-1

# src/iam_security.py
"""IAM security checks and best practices."""
import boto3
from botocore.exceptions import ClientError
from typing import List, Dict
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class IAMSecurityChecker:
    """Checks IAM security configurations."""
    
    def __init__(self):
        """Initialize IAM client."""
        self.iam = boto3.client('iam')
    
    def check_mfa_enabled(self) -> Dict:
        """Check if MFA is enabled for users."""
        try:
            users = self.iam.list_users()
            results = {
                "users_without_mfa": [],
                "users_with_mfa": [],
                "total_users": len(users['Users'])
            }
            
            for user in users['Users']:
                username = user['UserName']
                mfa_devices = self.iam.list_mfa_devices(UserName=username)
                
                if not mfa_devices['MFADevices']:
                    results["users_without_mfa"].append(username)
                else:
                    results["users_with_mfa"].append(username)
            
            return results
            
        except ClientError as e:
            logger.error(f"Error checking MFA: {e}")
            return {"error": str(e)}
    
    def check_overly_permissive_policies(self) -> List[Dict]:
        """Check for overly permissive IAM policies."""
        issues = []
        
        try:
            policies = self.iam.list_policies(Scope='Local')
            
            for policy in policies['Policies']:
                policy_version = self.iam.get_policy_version(
                    PolicyArn=policy['Arn'],
                    VersionId=policy['DefaultVersionId']
                )
                
                document = policy_version['PolicyVersion']['Document']
                
                # Check for wildcard actions
                for statement in document.get('Statement', []):
                    actions = statement.get('Action', [])
                    if isinstance(actions, str):
                        actions = [actions]
                    
                    if '*' in actions or 'Action' in statement and statement['Action'] == '*':
                        issues.append({
                            "policy": policy['PolicyName'],
                            "arn": policy['Arn'],
                            "issue": "Wildcard action found",
                            "statement": statement
                        })
            
            return issues
            
        except ClientError as e:
            logger.error(f"Error checking policies: {e}")
            return []
    
    def check_unused_access_keys(self) -> List[Dict]:
        """Check for unused access keys."""
        unused_keys = []
        
        try:
            users = self.iam.list_users()
            
            for user in users['Users']:
                username = user['UserName']
                access_keys = self.iam.list_access_keys(UserName=username)
                
                for key_metadata in access_keys['AccessKeyMetadata']:
                    key_id = key_metadata['AccessKeyId']
                    last_used = self.iam.get_access_key_last_used(AccessKeyId=key_id)
                    
                    if 'LastUsedDate' not in last_used['AccessKeyLastUsed']:
                        unused_keys.append({
                            "username": username,
                            "access_key_id": key_id,
                            "status": "Never used",
                            "created": key_metadata['CreateDate'].isoformat()
                        })
            
            return unused_keys
            
        except ClientError as e:
            logger.error(f"Error checking access keys: {e}")
            return []

# src/s3_security.py
"""S3 security checks and best practices."""
import boto3
from botocore.exceptions import ClientError
from typing import List, Dict
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class S3SecurityChecker:
    """Checks S3 security configurations."""
    
    def __init__(self):
        """Initialize S3 client."""
        self.s3 = boto3.client('s3')
    
    def check_public_buckets(self) -> List[Dict]:
        """Check for publicly accessible buckets."""
        public_buckets = []
        
        try:
            buckets = self.s3.list_buckets()
            
            for bucket in buckets['Buckets']:
                bucket_name = bucket['Name']
                
                try:
                    # Check bucket policy
                    policy = self.s3.get_bucket_policy(Bucket=bucket_name)
                    # Check for public access
                    if 'Public' in policy.get('Policy', ''):
                        public_buckets.append({
                            "bucket": bucket_name,
                            "issue": "Public bucket policy",
                            "policy": policy['Policy']
                        })
                except ClientError as e:
                    if e.response['Error']['Code'] != 'NoSuchBucketPolicy':
                        logger.warning(f"Error checking bucket {bucket_name}: {e}")
                
                # Check public access block
                try:
                    public_access = self.s3.get_public_access_block(Bucket=bucket_name)
                    if not all([
                        public_access['PublicAccessBlockConfiguration'].get('BlockPublicAcls', False),
                        public_access['PublicAccessBlockConfiguration'].get('BlockPublicPolicy', False)
                    ]):
                        public_buckets.append({
                            "bucket": bucket_name,
                            "issue": "Public access not blocked"
                        })
                except ClientError as e:
                    if e.response['Error']['Code'] == 'NoSuchPublicAccessBlockConfiguration':
                        public_buckets.append({
                            "bucket": bucket_name,
                            "issue": "No public access block configured"
                        })
            
            return public_buckets
            
        except ClientError as e:
            logger.error(f"Error checking buckets: {e}")
            return []
    
    def check_encryption(self) -> List[Dict]:
        """Check for encryption configuration."""
        unencrypted_buckets = []
        
        try:
            buckets = self.s3.list_buckets()
            
            for bucket in buckets['Buckets']:
                bucket_name = bucket['Name']
                
                try:
                    encryption = self.s3.get_bucket_encryption(Bucket=bucket_name)
                    # Encryption is configured
                except ClientError as e:
                    if e.response['Error']['Code'] == 'ServerSideEncryptionConfigurationNotFoundError':
                        unencrypted_buckets.append({
                            "bucket": bucket_name,
                            "issue": "No encryption configured"
                        })
            
            return unencrypted_buckets
            
        except ClientError as e:
            logger.error(f"Error checking encryption: {e}")
            return []

# src/github_credential_scanner.py
"""Scan GitHub repositories for exposed AWS credentials."""
import requests
import re
import json
from typing import List, Dict, Optional
import logging
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class GitHubCredentialScanner:
    """Scans GitHub for exposed AWS credentials."""
    
    def __init__(self, github_token: Optional[str] = None):
        """
        Initialize scanner.
        
        Args:
            github_token: GitHub personal access token (optional, increases rate limit)
        """
        self.github_token = github_token
        self.session = requests.Session()
        if github_token:
            self.session.headers.update({
                'Authorization': f'token {github_token}',
                'Accept': 'application/vnd.github.v3+json'
            })
        
        # AWS credential patterns
        self.aws_key_pattern = re.compile(
            r'AKIA[0-9A-Z]{16}',  # AWS Access Key ID
            re.IGNORECASE
        )
        self.aws_secret_pattern = re.compile(
            r'["\']?(AWS|aws)[_\-]?SECRET[_\-]?ACCESS[_\-]?KEY["\']?\s*[:=]\s*["\']?([A-Za-z0-9/+=]{40})["\']?',
            re.IGNORECASE
        )
    
    def search_repositories(self, query: str, max_results: int = 100) -> List[Dict]:
        """
        Search GitHub repositories.
        
        Args:
            query: Search query
            max_results: Maximum results to return
            
        Returns:
            List of repository information
        """
        repos = []
        page = 1
        per_page = 100
        
        try:
            while len(repos) < max_results:
                url = f"https://api.github.com/search/repositories"
                params = {
                    "q": query,
                    "page": page,
                    "per_page": min(per_page, max_results - len(repos))
                }
                
                response = self.session.get(url, params=params)
                response.raise_for_status()
                
                data = response.json()
                
                if not data.get('items'):
                    break
                
                repos.extend(data['items'])
                
                if len(data['items']) < per_page:
                    break
                
                page += 1
            
            return repos[:max_results]
            
        except requests.RequestException as e:
            logger.error(f"Error searching repositories: {e}")
            return []
    
    def scan_repository(self, owner: str, repo: str) -> List[Dict]:
        """
        Scan a repository for AWS credentials.
        
        Args:
            owner: Repository owner
            repo: Repository name
            
        Returns:
            List of potential credential leaks
        """
        findings = []
        
        try:
            # Search code in repository
            url = f"https://api.github.com/search/code"
            params = {
                "q": f"AKIA repo:{owner}/{repo}",
                "per_page": 100
            }
            
            response = self.session.get(url, params=params)
            response.raise_for_status()
            
            results = response.json()
            
            for item in results.get('items', []):
                file_path = item['path']
                file_url = item['html_url']
                
                # Get file content
                content_response = self.session.get(item['url'])
                content_response.raise_for_status()
                content_data = content_response.json()
                
                content = content_data.get('content', '')
                # Decode base64 content
                import base64
                try:
                    decoded_content = base64.b64decode(content).decode('utf-8', errors='ignore')
                except:
                    decoded_content = content
                
                # Check for AWS keys
                key_matches = self.aws_key_pattern.findall(decoded_content)
                secret_matches = self.aws_secret_pattern.findall(decoded_content)
                
                if key_matches or secret_matches:
                    findings.append({
                        "repository": f"{owner}/{repo}",
                        "file_path": file_path,
                        "file_url": file_url,
                        "aws_keys_found": len(key_matches),
                        "aws_secrets_found": len(secret_matches),
                        "timestamp": datetime.utcnow().isoformat()
                    })
            
            return findings
            
        except requests.RequestException as e:
            logger.error(f"Error scanning repository {owner}/{repo}: {e}")
            return []
    
    def validate_aws_key(self, access_key_id: str) -> Dict:
        """
        Validate if an AWS access key is active.
        
        Args:
            access_key_id: AWS access key ID
            
        Returns:
            Validation result
        """
        # Note: This is a simplified check. In production, use AWS API properly.
        # Never use real credentials in validation without proper authorization.
        return {
            "access_key_id": access_key_id,
            "format_valid": len(access_key_id) == 20 and access_key_id.startswith('AKIA'),
            "warning": "Do not use real credentials for validation without authorization"
        }
    
    def generate_report(self, findings: List[Dict]) -> str:
        """
        Generate a report of findings.
        
        Args:
            findings: List of credential leaks found
            
        Returns:
            Report as string
        """
        report = f"""
# AWS Credential Leak Report
Generated: {datetime.utcnow().isoformat()}

## Summary
Total repositories scanned: {len(set(f['repository'] for f in findings))}
Total files with potential leaks: {len(findings)}

## Findings
"""
        for finding in findings:
            report += f"""
### {finding['repository']}
- File: {finding['file_path']}
- URL: {finding['file_url']}
- AWS Keys Found: {finding['aws_keys_found']}
- AWS Secrets Found: {finding['aws_secrets_found']}
- Timestamp: {finding['timestamp']}
"""
        
        return report


# Usage example
if __name__ == "__main__":
    scanner = GitHubCredentialScanner()
    
    # Search for repositories
    repos = scanner.search_repositories("language:python", max_results=10)
    
    # Scan repositories
    all_findings = []
    for repo in repos[:5]:  # Limit to first 5 for demo
        owner = repo['owner']['login']
        repo_name = repo['name']
        findings = scanner.scan_repository(owner, repo_name)
        all_findings.extend(findings)
    
    # Generate report
    report = scanner.generate_report(all_findings)
    print(report)