AWS Roles Anywhere: Extending IAM Beyond AWS from Laptop to Production

# Old way: Hard-coded AWS credentials
import boto3

# Bad: Credentials in code, version control, everywhere
AWS_ACCESS_KEY = "AKIAIOSFODNN7EXAMPLE"
AWS_SECRET_KEY = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"

s3 = boto3.client('s3',
    aws_access_key_id=AWS_ACCESS_KEY,
    aws_secret_access_key=AWS_SECRET_KEY
)

#!/usr/bin/env python3
"""
Generate self-signed X.509 certificates for AWS Roles Anywhere.
This creates the certificate chain: Root CA → Intermediate CA → Client Certificate
"""

from cryptography import x509
from cryptography.x509.oid import NameOID, ExtensionOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from datetime import datetime, timedelta
import argparse

def generate_root_ca(common_name="RolesAnywhere Root CA", validity_days=3650):
    """Generate root CA certificate (valid 10 years)."""
    
    # Generate RSA key pair
    private_key = rsa.generate_private_key(
        public_exponent=65537,
        key_size=2048
    )
    
    # Build certificate subject
    subject = issuer = x509.Name([
        x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
        x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, "California"),
        x509.NameAttribute(NameOID.LOCALITY_NAME, "San Francisco"),
        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Acme Corp"),
        x509.NameAttribute(NameOID.COMMON_NAME, common_name),
    ])
    
    # Create certificate
    cert = x509.CertificateBuilder().subject_name(
        subject
    ).issuer_name(
        issuer
    ).public_key(
        private_key.public_key()
    ).serial_number(
        x509.random_serial_number()
    ).not_valid_before(
        datetime.utcnow()
    ).not_valid_after(
        datetime.utcnow() + timedelta(days=validity_days)
    ).add_extension(
        # Mark as CA (self-signed root)
        x509.BasicConstraints(ca=True, path_length=1),
        critical=True,
    ).add_extension(
        # Key usage
        x509.KeyUsage(
            digital_signature=True,
            content_commitment=False,
            key_encipherment=False,
            data_encipherment=False,
            key_agreement=False,
            key_cert_sign=True,
            crl_sign=True,
            encipher_only=False,
            decipher_only=False,
        ),
        critical=True,
    ).sign(
        private_key=private_key,
        algorithm=hashes.SHA256()
    )
    
    return cert, private_key

def generate_client_certificate(
    ca_cert,
    ca_private_key,
    common_name="user@example.com",
    validity_days=365
):
    """Generate client certificate signed by CA."""
    
    # Generate RSA key pair for client
    private_key = rsa.generate_private_key(
        public_exponent=65537,
        key_size=2048
    )
    
    # Build certificate subject
    subject = x509.Name([
        x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
        x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, "California"),
        x509.NameAttribute(NameOID.LOCALITY_NAME, "San Francisco"),
        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Acme Corp"),
        x509.NameAttribute(NameOID.COMMON_NAME, common_name),
    ])
    
    # Create certificate (signed by CA)
    cert = x509.CertificateBuilder().subject_name(
        subject
    ).issuer_name(
        ca_cert.issuer  # Signed by CA
    ).public_key(
        private_key.public_key()
    ).serial_number(
        x509.random_serial_number()
    ).not_valid_before(
        datetime.utcnow()
    ).not_valid_after(
        datetime.utcnow() + timedelta(days=validity_days)
    ).add_extension(
        # NOT a CA (end entity)
        x509.BasicConstraints(ca=False, path_length=None),
        critical=True,
    ).add_extension(
        # Key usage
        x509.KeyUsage(
            digital_signature=True,
            content_commitment=False,
            key_encipherment=False,
            data_encipherment=False,
            key_agreement=False,
            key_cert_sign=False,
            crl_sign=False,
            encipher_only=False,
            decipher_only=False,
        ),
        critical=True,
    ).sign(
        private_key=ca_private_key,
        algorithm=hashes.SHA256()
    )
    
    return cert, private_key

def save_certificate(cert, filepath):
    """Save certificate to PEM file."""
    
    pem = cert.public_bytes(serialization.Encoding.PEM)
    with open(filepath, 'wb') as f:
        f.write(pem)

def save_private_key(private_key, filepath, password=None):
    """Save private key to PEM file (optionally encrypted)."""
    
    encryption = serialization.NoEncryption()
    if password:
        encryption = serialization.BestAvailableEncryption(password.encode())
    
    pem = private_key.private_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PrivateFormat.PKCS8,
        encryption_algorithm=encryption
    )
    
    with open(filepath, 'wb') as f:
        f.write(pem)

if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="Generate X.509 certificates for AWS Roles Anywhere"
    )
    parser.add_argument(
        '--output-dir',
        default='./certs',
        help='Output directory for certificates'
    )
    
    args = parser.parse_args()
    
    import os
    os.makedirs(args.output_dir, exist_ok=True)
    
    print("🔐 Generating Root CA certificate...")
    root_cert, root_key = generate_root_ca()
    save_certificate(root_cert, f'{args.output_dir}/root-cert.pem')
    save_private_key(root_key, f'{args.output_dir}/root-key.pem')
    print(f"✅ Root CA: {args.output_dir}/root-cert.pem")
    
    print("\n🔐 Generating Client certificate...")
    client_cert, client_key = generate_client_certificate(
        root_cert,
        root_key,
        common_name="developer@acme.com"
    )
    save_certificate(client_cert, f'{args.output_dir}/client-cert.pem')
    save_private_key(client_key, f'{args.output_dir}/client-key.pem')
    print(f"✅ Client cert: {args.output_dir}/client-cert.pem")
    
    print("\n📋 Certificate Details:")
    print(f"Root CA valid until: {root_cert.not_valid_after}")
    print(f"Client cert valid until: {client_cert.not_valid_after}")
    print("\n⚠️  IMPORTANT: Store private keys securely!")

import boto3

iam = boto3.client('iam')

# Read root CA certificate
with open('certs/root-cert.pem', 'r') as f:
    ca_cert = f.read()

# Create trust anchor
response = iam.create_trust_anchor(
    Name='company-root-ca',
    Source={
        'SourceType': 'CERTIFICATE_BUNDLE',
        'SourceData': {
            'X509CertificateData': ca_cert
        }
    },
    Tags=[
        {
            'Key': 'Environment',
            'Value': 'development'
        }
    ]
)

trust_anchor_arn = response['TrustAnchorDetail']['Arn']
print(f"✅ Trust Anchor created: {trust_anchor_arn}")

# Create role with Roles Anywhere trust policy
trust_policy = {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "rolesanywhere.amazonaws.com"
            },
            "Action": [
                "sts:AssumeRole",
                "sts:TagSession",
                "sts:SetSourceIdentity"
            ]
        }
    ]
}

iam.create_role(
    RoleName='RolesAnywhereS3Access',
    AssumeRolePolicyDocument=json.dumps(trust_policy),
    Description='Role for workloads using Roles Anywhere'
)

# Attach permissions
iam.attach_role_policy(
    RoleName='RolesAnywhereS3Access',
    PolicyArn='arn:aws:iam::aws:policy/AmazonS3ReadOnlyAccess'
)

# Create profile
response = iam.create_profile(
    Name='developer-profile',
    RoleArns=[
        'arn:aws:iam::123456789012:role/RolesAnywhereS3Access'
    ]
)

profile_arn = response['ProfileDetail']['Arn']
print(f"✅ Profile created: {profile_arn}")

"""
AWS Roles Anywhere Credential Helper
Generates temporary AWS credentials using X.509 certificate.
"""

import boto3
import hashlib
import json
from datetime import datetime, timezone
from urllib.parse import urlencode
import requests
from cryptography import x509
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding
from cryptography.hazmat.backends import default_backend

class RolesAnywhereAuth:
    def __init__(self, cert_path, key_path, trust_anchor_arn, role_arn, profile_arn):
        """
        Initialize Roles Anywhere authenticator.
        
        Args:
            cert_path: Path to client certificate (client-cert.pem)
            key_path: Path to client private key (client-key.pem)
            trust_anchor_arn: ARN of the trust anchor
            role_arn: ARN of the role to assume
            profile_arn: ARN of the profile
        """
        
        self.cert_path = cert_path
        self.key_path = key_path
        self.trust_anchor_arn = trust_anchor_arn
        self.role_arn = role_arn
        self.profile_arn = profile_arn
        
        # Load certificate and key
        self.cert = self._load_certificate()
        self.key = self._load_private_key()
    
    def _load_certificate(self):
        """Load X.509 certificate."""
        with open(self.cert_path, 'rb') as f:
            cert_data = f.read()
        return x509.load_pem_x509_certificate(cert_data, default_backend())
    
    def _load_private_key(self):
        """Load private key."""
        with open(self.key_path, 'rb') as f:
            key_data = f.read()
        return serialization.load_pem_private_key(
            key_data,
            password=None,
            backend=default_backend()
        )
    
    def get_credentials(self):
        """
        Get temporary AWS credentials.
        
        Returns:
            dict: AWS credentials (AccessKeyId, SecretAccessKey, SessionToken)
        """
        
        # Step 1: Create credential request
        credential_request = self._create_credential_request()
        
        # Step 2: Sign the request
        signed_request = self._sign_request(credential_request)
        
        # Step 3: Send to AWS STS
        response = self._call_sts(signed_request)
        
        # Step 4: Extract credentials
        credentials = self._extract_credentials(response)
        
        return credentials
    
    def _create_credential_request(self):
        """Create credential request with certificate and headers."""
        
        timestamp = datetime.now(timezone.utc).isoformat()
        
        request = {
            'roleArn': self.role_arn,
            'trustAnchorArn': self.trust_anchor_arn,
            'profileArn': self.profile_arn,
            'durationSeconds': 3600,  # 1 hour
            'timestamp': timestamp,
            'x509Certificate': self._cert_to_pem(),
            'certificateChain': self._cert_to_pem()  # For now, just the cert
        }
        
        return request
    
    def _cert_to_pem(self):
        """Convert certificate to PEM string."""
        pem_bytes = self.cert.public_bytes(serialization.Encoding.PEM)
        return pem_bytes.decode('utf-8')
    
    def _sign_request(self, request):
        """Sign the request using certificate private key."""
        
        # Create canonical string to sign
        request_json = json.dumps(request, sort_keys=True)
        
        # Hash the request
        message_hash = hashes.Hash(hashes.SHA256())
        message_hash.update(request_json.encode('utf-8'))
        digest = message_hash.finalize()
        
        # Sign with private key (PKCS#1 v1.5 padding)
        signature = self.key.sign(
            digest,
            padding.PKCS1v15(),
            hashes.SHA256()
        )
        
        # Return signed request
        request['signature'] = signature.hex()
        request['signatureAlgorithm'] = 'SHA256WithRSAEncryption'
        
        return request
    
    def _call_sts(self, signed_request):
        """Call AWS STS CreateSession to exchange certificate for credentials."""
        
        # AWS Roles Anywhere endpoint
        sts_endpoint = 'https://sts.amazonaws.com/'
        
        # Prepare request
        headers = {
            'Authorization': f'x509',
            'Content-Type': 'application/json'
        }
        
        params = {
            'Action': 'CreateSession',
            'Version': '2011-06-15',
            'RoleArn': self.role_arn,
            'RoleSessionName': 'roles-anywhere-session',
            'DurationSeconds': 3600,
        }
        
        # Make request using certificate for mTLS
        response = requests.post(
            sts_endpoint,
            headers=headers,
            params=params,
            json=signed_request,
            cert=(self.cert_path, self.key_path),  # mTLS
            verify=True
        )
        
        if response.status_code != 200:
            raise Exception(f"STS error: {response.text}")
        
        return response.json()
    
    def _extract_credentials(self, response):
        """Extract AWS credentials from STS response."""
        
        credentials = response.get('Credentials', {})
        
        return {
            'AccessKeyId': credentials.get('AccessKeyId'),
            'SecretAccessKey': credentials.get('SecretAccessKey'),
            'SessionToken': credentials.get('SessionToken'),
            'Expiration': credentials.get('Expiration')
        }

# Usage Example
if __name__ == '__main__':
    auth = RolesAnywhereAuth(
        cert_path='certs/client-cert.pem',
        key_path='certs/client-key.pem',
        trust_anchor_arn='arn:aws:iam::123456789012:trust-anchor/abc123',
        role_arn='arn:aws:iam::123456789012:role/RolesAnywhereS3Access',
        profile_arn='arn:aws:iam::123456789012:profile/abc123'
    )
    
    print("🔑 Fetching temporary AWS credentials...")
    creds = auth.get_credentials()
    
    print(f"✅ Credentials obtained!")
    print(f"Access Key: {creds['AccessKeyId']}")
    print(f"Expiration: {creds['Expiration']}")
    
    # Now use boto3 with these credentials
    import boto3
    
    s3 = boto3.client(
        's3',
        aws_access_key_id=creds['AccessKeyId'],
        aws_secret_access_key=creds['SecretAccessKey'],
        aws_session_token=creds['SessionToken']
    )
    
    # List S3 buckets
    response = s3.list_buckets()
    print(f"\n📦 Your S3 buckets: {[b['Name'] for b in response['Buckets']]}")

"""
Django settings that automatically use Roles Anywhere on non-AWS systems.
"""

import os
import boto3
from roles_anywhere_auth import RolesAnywhereAuth

def get_aws_credentials():
    """Get AWS credentials (from IAM role if on EC2, from Roles Anywhere if local)."""
    
    # Check if running on EC2 (has instance metadata)
    try:
        import requests
        metadata = requests.get(
            'http://169.254.169.254/latest/meta-data/iam/security-credentials/',
            timeout=1
        )
        if metadata.status_code == 200:
            print("✅ Running on EC2, using IAM role")
            return None  # Use IAM role automatically
    except:
        pass
    
    # Not on EC2 - use Roles Anywhere
    if os.path.exists('certs/client-cert.pem'):
        print("🔐 Running locally, using Roles Anywhere certificates")
        
        auth = RolesAnywhereAuth(
            cert_path='certs/client-cert.pem',
            key_path='certs/client-key.pem',
            trust_anchor_arn=os.environ['TRUST_ANCHOR_ARN'],
            role_arn=os.environ['ROLE_ARN'],
            profile_arn=os.environ['PROFILE_ARN']
        )
        
        return auth.get_credentials()
    
    raise Exception("No AWS credentials found")

# In Django settings.py
AWS_CREDENTIALS = get_aws_credentials()

if AWS_CREDENTIALS:
    os.environ['AWS_ACCESS_KEY_ID'] = AWS_CREDENTIALS['AccessKeyId']
    os.environ['AWS_SECRET_ACCESS_KEY'] = AWS_CREDENTIALS['SecretAccessKey']
    os.environ['AWS_SESSION_TOKEN'] = AWS_CREDENTIALS['SessionToken']

# boto3 will now use these credentials

from cryptography import x509
from datetime import datetime

def check_certificate_expiry(cert_path, warn_days=30):
    """Check if certificate is expiring soon."""
    
    with open(cert_path, 'rb') as f:
        cert = x509.load_pem_x509_certificate(f.read())
    
    days_until_expiry = (cert.not_valid_after - datetime.utcnow()).days
    
    if days_until_expiry < warn_days:
        print(f"⚠️  Certificate expires in {days_until_expiry} days!")
        print("Generate new certificate: python generate_certificates.py")
    
    return days_until_expiry

import json
from datetime import datetime, timezone, timedelta

class CachedRolesAnywhereAuth(RolesAnywhereAuth):
    def __init__(self, *args, cache_file='~/.aws/roles-anywhere-cache.json', **kwargs):
        super().__init__(*args, **kwargs)
        self.cache_file = cache_file
    
    def get_credentials(self):
        """Get credentials from cache if valid, otherwise fetch new ones."""
        
        # Check cache
        if os.path.exists(self.cache_file):
            with open(self.cache_file, 'r') as f:
                cached = json.load(f)
            
            expiry = datetime.fromisoformat(cached['Expiration'])
            if expiry > datetime.now(timezone.utc):
                print("✅ Using cached credentials")
                return cached
        
        # Fetch new credentials
        print("🔑 Fetching new credentials...")
        creds = super().get_credentials()
        
        # Cache them
        os.makedirs(os.path.dirname(self.cache_file), exist_ok=True)
        with open(self.cache_file, 'w') as f:
            json.dump(creds, f)
        
        return creds

trust_policy = {
    "Statement": [{
        "Effect": "Allow",
        "Principal": {"Service": "rolesanywhere.amazonaws.com"},
        "Action": [
            "sts:AssumeRole",
            "sts:TagSession",
            "sts:SetSourceIdentity"
        ]
    }]
}