authentication.md

Authentication and Authorization

This document covers the authentication and authorization mechanisms used by the FinOps Cloud Resource Cleanup API.

Overview

The FinOps API uses a multi-layered security approach combining AWS IAM, API Gateway authentication, and custom authorization logic to ensure secure access to resources and operations.

Authentication Methods

1. AWS IAM Authentication

IAM Roles for Applications

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "arn:aws:iam::123456789012:role/FinOpsApplicationRole"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

SigV4 Request Signing

# Example: Python client with AWS SigV4
import boto3
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
import requests

def make_authenticated_request(url, method='GET', payload=None):
    """Make authenticated request using AWS SigV4."""

    session = boto3.Session()
    credentials = session.get_credentials()
    region = session.region_name

    request = AWSRequest(
        method=method,
        url=url,
        data=payload,
        headers={'Content-Type': 'application/json'}
    )

    SigV4Auth(credentials, 'execute-api', region).add_auth(request)

    return requests.request(
        method=method,
        url=url,
        headers=dict(request.headers),
        data=payload
    )

# Usage
response = make_authenticated_request(
    'https://api.finops.company.com/resources',
    method='GET'
)

2. API Key Authentication

API Key Generation

# Create API key via AWS CLI
aws apigateway create-api-key \
    --name "finops-client-key" \
    --description "API key for FinOps client applications" \
    --enabled

# Associate with usage plan
aws apigateway create-usage-plan-key \
    --usage-plan-id "abcdef123456" \
    --key-id "1234567890abcdef" \
    --key-type "API_KEY"

API Key Usage

# Python example
import requests

headers = {
    'X-API-Key': 'your-api-key-here',
    'Content-Type': 'application/json'
}

response = requests.get(
    'https://api.finops.company.com/resources',
    headers=headers
)

// JavaScript example
const response = await fetch("https://api.finops.company.com/resources", {
  headers: {
    "X-API-Key": "your-api-key-here",
    "Content-Type": "application/json",
  },
});

# curl example
curl -X GET \
  https://api.finops.company.com/resources \
  -H 'X-API-Key: your-api-key-here' \
  -H 'Content-Type: application/json'

3. JWT Token Authentication

Token Structure

{
  "header": {
    "alg": "RS256",
    "typ": "JWT",
    "kid": "finops-key-id"
  },
  "payload": {
    "sub": "user@company.com",
    "iss": "finops-auth-service",
    "aud": "finops-api",
    "exp": 1706443200,
    "iat": 1706356800,
    "roles": ["finops:viewer", "finops:operator"],
    "permissions": ["resources:read", "reports:read", "scans:execute"],
    "account_id": "123456789012",
    "departments": ["engineering", "finance"]
  }
}

Token Validation

# Lambda authorizer for JWT validation
import jwt
import json
import os
from typing import Dict, Any

def lambda_handler(event: Dict[str, Any], context) -> Dict[str, Any]:
    """JWT token validation Lambda authorizer."""

    token = extract_token(event)

    try:
        # Validate token
        decoded_token = jwt.decode(
            token,
            key=get_public_key(),
            algorithms=['RS256'],
            audience='finops-api',
            issuer='finops-auth-service'
        )

        # Generate policy
        policy = generate_policy(
            principal_id=decoded_token['sub'],
            effect='Allow',
            resource=event['methodArn'],
            context=decoded_token
        )

        return policy

    except jwt.InvalidTokenError as e:
        raise Exception('Unauthorized')

def extract_token(event: Dict[str, Any]) -> str:
    """Extract JWT token from Authorization header."""
    auth_header = event.get('authorizationToken', '')

    if not auth_header.startswith('Bearer '):
        raise Exception('Invalid token format')

    return auth_header[7:]  # Remove 'Bearer ' prefix

def get_public_key() -> str:
    """Get public key for JWT verification."""
    # In production, fetch from JWKS endpoint or AWS Secrets Manager
    return os.environ.get('JWT_PUBLIC_KEY')

def generate_policy(principal_id: str, effect: str, resource: str, context: Dict[str, Any]) -> Dict[str, Any]:
    """Generate IAM policy for API Gateway."""
    return {
        'principalId': principal_id,
        'policyDocument': {
            'Version': '2012-10-17',
            'Statement': [
                {
                    'Action': 'execute-api:Invoke',
                    'Effect': effect,
                    'Resource': resource
                }
            ]
        },
        'context': {
            'user_id': context.get('sub'),
            'roles': ','.join(context.get('roles', [])),
            'permissions': ','.join(context.get('permissions', [])),
            'account_id': context.get('account_id')
        }
    }

Authorization Framework

1. Role-Based Access Control (RBAC)

Role Definitions

# Role and permission definitions
ROLES = {
    'finops:admin': {
        'description': 'Full administrative access',
        'permissions': [
            'resources:*',
            'reports:*',
            'scans:*',
            'settings:*',
            'users:*'
        ]
    },
    'finops:operator': {
        'description': 'Operational access for FinOps team',
        'permissions': [
            'resources:read',
            'resources:update',
            'reports:read',
            'reports:create',
            'scans:execute',
            'scans:read'
        ]
    },
    'finops:viewer': {
        'description': 'Read-only access for stakeholders',
        'permissions': [
            'resources:read',
            'reports:read',
            'scans:read'
        ]
    },
    'finops:auditor': {
        'description': 'Audit and compliance access',
        'permissions': [
            'resources:read',
            'reports:read',
            'audit:read',
            'compliance:read'
        ]
    }
}

PERMISSIONS = {
    'resources:read': 'View resource information',
    'resources:update': 'Update resource metadata',
    'resources:delete': 'Delete or archive resources',
    'reports:read': 'View reports and analytics',
    'reports:create': 'Generate new reports',
    'reports:delete': 'Delete reports',
    'scans:read': 'View scan results',
    'scans:execute': 'Execute resource scans',
    'scans:configure': 'Configure scan parameters',
    'settings:read': 'View system settings',
    'settings:update': 'Modify system settings',
    'users:read': 'View user information',
    'users:manage': 'Manage user accounts'
}

Permission Checking

# Permission validation decorator
from functools import wraps
import json

def require_permission(permission: str):
    """Decorator to enforce permission requirements."""

    def decorator(func):
        @wraps(func)
        def wrapper(event, context):
            # Extract user context from API Gateway authorizer
            user_context = event.get('requestContext', {}).get('authorizer', {})
            user_permissions = user_context.get('permissions', '').split(',')

            # Check permission
            if not has_permission(user_permissions, permission):
                return {
                    'statusCode': 403,
                    'body': json.dumps({
                        'error': 'Forbidden',
                        'message': f'Required permission: {permission}'
                    })
                }

            return func(event, context)
        return wrapper
    return decorator

def has_permission(user_permissions: list, required_permission: str) -> bool:
    """Check if user has required permission."""

    # Check exact match
    if required_permission in user_permissions:
        return True

    # Check wildcard permissions
    resource, action = required_permission.split(':')
    wildcard_permission = f"{resource}:*"

    if wildcard_permission in user_permissions:
        return True

    # Check admin wildcard
    if '*:*' in user_permissions:
        return True

    return False

# Usage example
@require_permission('resources:read')
def get_resources(event, context):
    """Get resources endpoint with permission check."""
    # Implementation here
    pass

2. Attribute-Based Access Control (ABAC)

Resource-Level Access Control

# Resource access control based on attributes
def check_resource_access(user_context: Dict[str, Any], resource: Dict[str, Any], action: str) -> bool:
    """Check if user can access specific resource based on attributes."""

    user_account = user_context.get('account_id')
    user_departments = user_context.get('departments', [])
    user_roles = user_context.get('roles', [])

    resource_account = resource.get('accountId')
    resource_department = resource.get('tags', {}).get('Department')
    resource_environment = resource.get('tags', {}).get('Environment')

    # Account-level access control
    if user_account != resource_account and 'finops:admin' not in user_roles:
        return False

    # Department-level access control
    if resource_department and resource_department not in user_departments:
        if 'finops:admin' not in user_roles:
            return False

    # Environment-level access control
    if action in ['resources:delete', 'resources:cleanup']:
        if resource_environment == 'production' and 'finops:operator' not in user_roles:
            return False

    return True

# Usage in Lambda function
@require_permission('resources:read')
def get_resource(event, context):
    """Get specific resource with attribute-based access control."""

    resource_id = event['pathParameters']['resourceId']
    user_context = event['requestContext']['authorizer']

    # Get resource from database
    resource = get_resource_from_db(resource_id)

    # Check attribute-based access
    if not check_resource_access(user_context, resource, 'resources:read'):
        return {
            'statusCode': 403,
            'body': json.dumps({
                'error': 'Forbidden',
                'message': 'Access denied to this resource'
            })
        }

    return {
        'statusCode': 200,
        'body': json.dumps(resource)
    }

3. Multi-Account Access Control

Cross-Account Role Assumption

# Cross-account access management
class MultiAccountAccessManager:

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

    def assume_role_for_account(self, account_id: str, role_name: str = 'FinOpsResourceCleanupRole') -> Dict[str, Any]:
        """Assume role in target account for resource access."""

        role_arn = f"arn:aws:iam::{account_id}:role/{role_name}"

        try:
            response = self.sts_client.assume_role(
                RoleArn=role_arn,
                RoleSessionName=f"finops-session-{int(time.time())}",
                DurationSeconds=3600  # 1 hour
            )

            return response['Credentials']

        except ClientError as e:
            if e.response['Error']['Code'] == 'AccessDenied':
                raise Exception(f"Cannot assume role in account {account_id}")
            raise

    def get_account_client(self, service: str, account_id: str) -> boto3.client:
        """Get boto3 client for specific account."""

        if account_id == self.get_current_account():
            return boto3.client(service)

        credentials = self.assume_role_for_account(account_id)

        return boto3.client(
            service,
            aws_access_key_id=credentials['AccessKeyId'],
            aws_secret_access_key=credentials['SecretAccessKey'],
            aws_session_token=credentials['SessionToken']
        )

    def get_current_account(self) -> str:
        """Get current AWS account ID."""
        return self.sts_client.get_caller_identity()['Account']

# Usage
access_manager = MultiAccountAccessManager()
ec2_client = access_manager.get_account_client('ec2', '123456789012')

API Gateway Security Configuration

1. Resource Policies

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": "*",
      "Action": "execute-api:Invoke",
      "Resource": "arn:aws:execute-api:us-east-1:123456789012:*/*/GET/resources",
      "Condition": {
        "StringEquals": {
          "aws:PrincipalAccount": "123456789012"
        }
      }
    },
    {
      "Effect": "Allow",
      "Principal": "*",
      "Action": "execute-api:Invoke",
      "Resource": "arn:aws:execute-api:us-east-1:123456789012:*/*/POST/scans",
      "Condition": {
        "StringLike": {
          "aws:PrincipalArn": "arn:aws:iam::123456789012:role/FinOps*"
        }
      }
    },
    {
      "Effect": "Deny",
      "Principal": "*",
      "Action": "execute-api:Invoke",
      "Resource": "arn:aws:execute-api:us-east-1:123456789012:*/*/DELETE/*",
      "Condition": {
        "StringNotEquals": {
          "aws:PrincipalTag/Department": ["FinOps", "Security"]
        }
      }
    }
  ]
}

2. Throttling and Rate Limiting

# CloudFormation template for API Gateway throttling
ApiGatewayUsagePlan:
  Type: AWS::ApiGateway::UsagePlan
  Properties:
    UsagePlanName: FinOpsAPIUsagePlan
    Description: Usage plan for FinOps API
    ApiStages:
      - ApiId: !Ref FinOpsAPI
        Stage: !Ref FinOpsAPIStage
    Throttle:
      BurstLimit: 100
      RateLimit: 50
    Quota:
      Limit: 10000
      Period: DAY

# Method-specific throttling
ResourcesGetMethod:
  Type: AWS::ApiGateway::Method
  Properties:
    HttpMethod: GET
    ResourceId: !Ref ResourcesResource
    RestApiId: !Ref FinOpsAPI
    AuthorizationType: AWS_IAM
    MethodResponses:
      - StatusCode: 200
      - StatusCode: 403
      - StatusCode: 429 # Too Many Requests
    Integration:
      Type: AWS_PROXY
      IntegrationHttpMethod: POST
      Uri: !Sub "arn:aws:apigateway:${AWS::Region}:lambda:path/2015-03-31/functions/${GetResourcesFunction.Arn}/invocations"

Security Best Practices

1. Token Management

JWT Token Refresh

# Token refresh mechanism
class TokenManager:

    def __init__(self):
        self.current_token = None
        self.refresh_token = None
        self.token_expiry = None

    def get_valid_token(self) -> str:
        """Get valid token, refreshing if necessary."""

        if self.is_token_expired():
            self.refresh_access_token()

        return self.current_token

    def is_token_expired(self) -> bool:
        """Check if current token is expired."""

        if not self.token_expiry:
            return True

        # Refresh 5 minutes before expiry
        buffer_time = 300
        return time.time() > (self.token_expiry - buffer_time)

    def refresh_access_token(self):
        """Refresh access token using refresh token."""

        response = requests.post(
            'https://auth.finops.company.com/token/refresh',
            json={'refresh_token': self.refresh_token}
        )

        if response.status_code == 200:
            token_data = response.json()
            self.current_token = token_data['access_token']
            self.token_expiry = time.time() + token_data['expires_in']
        else:
            raise Exception('Token refresh failed')

Secure Token Storage

# Secure token storage using AWS Secrets Manager
import boto3
import json

class SecureTokenStorage:

    def __init__(self):
        self.secrets_client = boto3.client('secretsmanager')

    def store_token(self, user_id: str, token_data: Dict[str, Any]):
        """Store token securely in Secrets Manager."""

        secret_name = f"finops/user-tokens/{user_id}"

        try:
            self.secrets_client.create_secret(
                Name=secret_name,
                SecretString=json.dumps(token_data),
                Description=f"API tokens for user {user_id}"
            )
        except self.secrets_client.exceptions.ResourceExistsException:
            self.secrets_client.update_secret(
                SecretId=secret_name,
                SecretString=json.dumps(token_data)
            )

    def get_token(self, user_id: str) -> Dict[str, Any]:
        """Retrieve token from Secrets Manager."""

        secret_name = f"finops/user-tokens/{user_id}"

        try:
            response = self.secrets_client.get_secret_value(SecretId=secret_name)
            return json.loads(response['SecretString'])
        except self.secrets_client.exceptions.ResourceNotFoundException:
            return None

2. Audit Logging

Authentication Events

# Authentication audit logging
import json
from datetime import datetime

def log_authentication_event(event_type: str, user_id: str, success: bool, details: Dict[str, Any] = None):
    """Log authentication events for audit trail."""

    audit_event = {
        'timestamp': datetime.utcnow().isoformat(),
        'event_type': event_type,
        'user_id': user_id,
        'success': success,
        'source_ip': details.get('source_ip') if details else None,
        'user_agent': details.get('user_agent') if details else None,
        'api_endpoint': details.get('api_endpoint') if details else None,
        'session_id': details.get('session_id') if details else None
    }

    # Send to CloudWatch Logs
    cloudwatch_logs = boto3.client('logs')

    cloudwatch_logs.put_log_events(
        logGroupName='/aws/lambda/finops-auth-audit',
        logStreamName=f"auth-events-{datetime.now().strftime('%Y-%m-%d')}",
        logEvents=[
            {
                'timestamp': int(time.time() * 1000),
                'message': json.dumps(audit_event)
            }
        ]
    )

# Usage in authentication functions
def authenticate_user(event, context):
    try:
        # Authentication logic
        user = validate_credentials(event)

        log_authentication_event(
            'login_success',
            user['user_id'],
            True,
            {
                'source_ip': event.get('requestContext', {}).get('identity', {}).get('sourceIp'),
                'user_agent': event.get('headers', {}).get('User-Agent')
            }
        )

        return generate_success_response(user)

    except AuthenticationError as e:
        log_authentication_event(
            'login_failure',
            event.get('username', 'unknown'),
            False,
            {'error': str(e)}
        )

        return generate_error_response(str(e))

3. Security Headers

# Security headers for API responses
def add_security_headers(response: Dict[str, Any]) -> Dict[str, Any]:
    """Add security headers to API responses."""

    security_headers = {
        'Strict-Transport-Security': 'max-age=31536000; includeSubDomains',
        'X-Content-Type-Options': 'nosniff',
        'X-Frame-Options': 'DENY',
        'X-XSS-Protection': '1; mode=block',
        'Content-Security-Policy': "default-src 'self'",
        'Referrer-Policy': 'strict-origin-when-cross-origin'
    }

    if 'headers' not in response:
        response['headers'] = {}

    response['headers'].update(security_headers)

    return response

# Lambda function wrapper
def secure_lambda_handler(func):
    """Decorator to add security headers to Lambda responses."""

    @wraps(func)
    def wrapper(event, context):
        response = func(event, context)
        return add_security_headers(response)

    return wrapper

Error Handling

1. Authentication Errors

# Standard authentication error responses
class AuthenticationError(Exception):
    """Base authentication error."""
    pass

class TokenExpiredError(AuthenticationError):
    """Token has expired."""
    pass

class InvalidTokenError(AuthenticationError):
    """Token is invalid or malformed."""
    pass

class InsufficientPermissionsError(AuthenticationError):
    """User lacks required permissions."""
    pass

def handle_auth_error(error: Exception) -> Dict[str, Any]:
    """Handle authentication errors with appropriate responses."""

    error_responses = {
        TokenExpiredError: {
            'statusCode': 401,
            'body': json.dumps({
                'error': 'token_expired',
                'message': 'Token has expired, please refresh'
            })
        },
        InvalidTokenError: {
            'statusCode': 401,
            'body': json.dumps({
                'error': 'invalid_token',
                'message': 'Token is invalid or malformed'
            })
        },
        InsufficientPermissionsError: {
            'statusCode': 403,
            'body': json.dumps({
                'error': 'insufficient_permissions',
                'message': 'User lacks required permissions'
            })
        }
    }

    error_type = type(error)
    if error_type in error_responses:
        return error_responses[error_type]

    # Default error response
    return {
        'statusCode': 401,
        'body': json.dumps({
            'error': 'authentication_failed',
            'message': 'Authentication failed'
        })
    }

Testing Authentication

1. Unit Tests

# tests/test_authentication.py
import pytest
from unittest.mock import Mock, patch
from src.auth.jwt_validator import validate_jwt_token

class TestAuthentication:

    def test_valid_jwt_token(self):
        """Test validation of valid JWT token."""

        valid_token = "eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9..."

        with patch('src.auth.jwt_validator.get_public_key') as mock_key:
            mock_key.return_value = "mock_public_key"

            with patch('jwt.decode') as mock_decode:
                mock_decode.return_value = {
                    'sub': 'user@company.com',
                    'roles': ['finops:viewer'],
                    'permissions': ['resources:read']
                }

                result = validate_jwt_token(valid_token)

                assert result['sub'] == 'user@company.com'
                assert 'finops:viewer' in result['roles']

    def test_expired_jwt_token(self):
        """Test handling of expired JWT token."""

        expired_token = "eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9..."

        with patch('jwt.decode') as mock_decode:
            mock_decode.side_effect = jwt.ExpiredSignatureError()

            with pytest.raises(TokenExpiredError):
                validate_jwt_token(expired_token)

Next Steps

After implementing authentication and authorization:

1. API Documentation: Review API Endpoints for complete API reference
2. Error Handling: Implement Error Handling patterns
3. Security: Review Security Policies
4. Monitoring: Set up CloudWatch Integration for auth events