diff --git a/local-app/python-tools/gfl-resource-actions/README.md b/local-app/python-tools/gfl-resource-actions/README.md index 306f380b..9ccf10d6 100644 --- a/local-app/python-tools/gfl-resource-actions/README.md +++ b/local-app/python-tools/gfl-resource-actions/README.md @@ -1,171 +1,101 @@ -# AWS Resource Management Utilities +# AWS Resource Management Tool -A collection of Python utilities for managing and interacting with AWS resources across multiple accounts, with special support for GovCloud environments and AWS SSO. +## Overview +The AWS Resource Management Tool is a Python-based utility designed to discover, start, and stop AWS resources across multiple accounts, primarily in GovCloud environments. Its primary purpose is cost management by automating resource state management. ## Features - -- Cross-account resource management -- Dynamic AWS partition detection (supports commercial AWS, GovCloud, AWS China) -- AWS SSO profile integration -- Fallback mechanisms for authentication -- Support for AWS Organizations - -## Requirements - -- Python 3.6+ -- boto3 -- AWS credentials configured through AWS SSO or with standard credentials - -## Installation - -1. Clone this repository or copy the files to your desired location -2. Install required dependencies: +- Multi-account and multi-region support +- Resource discovery and management for EC2, RDS, EKS, EMR, and ECR +- AWS SSO and role-based authentication +- CLI for easy interaction +- Dry-run mode for safe testing +- CSV reporting for resource actions + +## Package Structure ``` -pip install boto3 +aws_resource_management/ # Main package +├── __init__.py +├── aws_utils.py # AWS credential/authentication utilities +├── cli.py # Command-line interface +├── core.py # Main business logic (ResourceManager) +├── discovery.py # Resource discovery logic +├── managers/ # Resource-specific managers +│ ├── base.py # Base resource manager class +│ ├── ec2.py # EC2-specific implementation +│ ├── eks.py # EKS-specific implementation +│ ├── emr.py # EMR-specific implementation +│ ├── rds.py # RDS-specific implementation +│ └── ecr.py # ECR-specific implementation +├── reporting.py # Reporting utilities +└── logging_setup.py # Logging configuration ``` -## Command-Line Usage - -The package provides a command-line interface for managing AWS resources: - -```bash -# Stop all resources in dry-run mode -aws-resource-mgmt --stop --dry-run --resource-type all +## Installation -# Start EC2 instances -aws-resource-mgmt --start --resource-type ec2 +1. Clone the repository: + ```bash + git clone + cd + ``` -# Stop RDS instances in specific regions -aws-resource-mgmt --stop --resource-type rds --region us-east-1 --region us-west-2 -``` +2. Install dependencies: + ```bash + make install-dev + ``` ## Usage -### Basic Usage - -```python -import aws_resource_management.aws_utils as aws_utils - -# Get an EC2 client for a specific account -ec2_client = aws_utils.create_boto3_client_for_account( - '123456789012', # AWS account ID - 'ec2', # AWS service - 'us-gov-west-1' # AWS region -) - -# Use the client -instances = ec2_client.describe_instances() -``` - -### Using AWS SSO Profiles - -The library will automatically find and use appropriate AWS SSO profiles from your AWS config file (`~/.aws/config`). It will: - -1. Look for profiles matching the requested account ID -2. If a role_name is specified, it will find a profile with that role -3. Otherwise, it will prefer profiles with admin permissions -4. Fall back to role assumption if no profile is found or profile authentication fails - -### Key Functions - -#### `create_boto3_client_for_account(account_id, service, region=None, role_name=None)` - -Creates a boto3 client for the specified AWS service in the target account. - -- `account_id`: Target AWS account ID -- `service`: AWS service name (e.g., 'ec2', 'rds') -- `region`: AWS region (defaults to config.DEFAULT_REGION) -- `role_name`: Specific role to assume or look for in SSO profiles - -#### `get_partition_for_region(region)` - -Determines the correct AWS partition for a given region. - -- `region`: AWS region name -- Returns: 'aws', 'aws-us-gov', or 'aws-cn' - -#### `find_profile_for_account(account_id, role_name=None)` - -Finds an SSO profile in the AWS config file for the specified account. - -- `account_id`: AWS account ID to find profile for -- `role_name`: Optional preferred role name -- Returns: Profile name if found, None otherwise - -#### `assume_role(account_id, region=None, role_name=None)` - -Gets credentials for the specified account, trying SSO profiles first. - -- `account_id`: AWS account ID -- `region`: AWS region for determining partition -- `role_name`: Role name to assume -- Returns: Credentials dictionary - -#### `get_organization_accounts(exclude_accounts=None)` - -Gets a list of accounts in the AWS organization. - -- `exclude_accounts`: List of account IDs to exclude -- Returns: List of dictionaries with account ID and name - -## Configuration - -Create a `config.py` file with the following variables: - -```python -# Default AWS region to use -DEFAULT_REGION = 'us-gov-east-1' - -# Default role name to assume if one is not provided -ASSUME_ROLE_NAME = 'OrganizationAccountAccessRole' -``` - -## Makefile Usage - -The project includes a Makefile to simplify common operations: - -``` -# Install dependencies -make install - -# Run tests -make test - -# Run linting -make lint - -# Format code -make format - -# Clean up temporary files -make clean - -# Build distribution package -make dist - -# Deploy to development environment -make deploy-dev - -# Deploy to production environment -make deploy-prod -``` - -The Makefile helps standardize development workflows and ensures consistent environment setup across different machines. To view all available commands: - -``` -make help -``` - -## Tips for GovCloud Usage - -When working with GovCloud: - -1. Ensure your AWS SSO is properly configured -2. AWS SSO profiles should include the GovCloud regions in their configuration -3. The library will automatically detect GovCloud regions and use the correct partition ('aws-us-gov') - -## Troubleshooting - -- **Authentication failures**: Check that your SSO session is active (`aws sso login --profile your-profile`) -- **Profile not found**: Verify your `~/.aws/config` file contains the correct account IDs -- **Permission issues**: Ensure the roles in your SSO profiles have the necessary permissions +### CLI Commands +The tool exposes a CLI through `aws-resource-mgmt` with the following options: + +- `--start`/`--stop`: Specify the action to perform. +- `--resource-type`: Select resource type (ec2, rds, eks, emr, ecr, all). +- `--region`/`--exclude-region`: Specify regions to include or exclude. +- `--account`/`--exclude-account`: Specify accounts to include or exclude. +- `--dry-run`: Simulate actions without making changes. + +#### Examples +- Run in dry-run mode: + ```bash + python -m aws_resource_management.cli --stop --dry-run --resource-type ec2 + ``` + +- Stop all resources: + ```bash + python -m aws_resource_management.cli --stop --resource-type all + ``` + +- Start EC2 instances only: + ```bash + python -m aws_resource_management.cli --start --resource-type ec2 + ``` + +## Development + +### Code Quality +- Run linters: + ```bash + make lint + ``` +- Format code: + ```bash + make format + ``` + +### Testing +- Run tests: + ```bash + make test + ``` + +### Build Distribution +- Build the package: + ```bash + make dist + ``` + +## Contributing +Contributions are welcome! Please see the `CONTRIBUTING.md` file for guidelines. + +## License +This project is licensed under the MIT License. See the LICENSE file for details. diff --git a/local-app/python-tools/gfl-resource-actions/analysis.md b/local-app/python-tools/gfl-resource-actions/analysis.md new file mode 100644 index 00000000..717eea6f --- /dev/null +++ b/local-app/python-tools/gfl-resource-actions/analysis.md @@ -0,0 +1,163 @@ +# AWS Resource Management Code Analysis + +## Current Structure Analysis + +The current codebase contains several utility modules with overlapping functionality. This analysis identifies opportunities for consolidation and suggests improvements to the project structure. + +### Utility Files Overview + +1. **aws_utils.py** + - Contains AWS credential management, session handling, region detection + - Has caching mechanisms for performance optimization + - Contains partition detection logic + +2. **utils.py** + - Already marked as deprecated, imports from aws_utils.py + - Maintained for backward compatibility + +3. **partition_utils.py** + - Contains AWS partition information and utilities + - Defines constants for partition regions + - Provides functions for region/partition validation and conversion + +4. **discovery_utils.py** + - Contains pagination utilities for AWS API responses + - Provides common functions for resource discovery + - Handles tag formatting and resource state normalization + +5. **logging_setup.py** + - Contains logging configuration utilities + - Includes CSV logging functionality + - Defines context logging utilities + +6. **csv_utils.py** + - Contains utilities for CSV file operations + - Has functions for initializing and writing to CSV files + - Provides tag formatting for CSV output + +7. **config_manager.py** + - Handles configuration loading and merging + - Provides default configuration values + - Implements deep dictionary merging + +## Identified Issues + +1. **Functionality Duplication** + - AWS credential handling is split between multiple files + - Region/partition handling logic is duplicated between aws_utils.py and partition_utils.py + - CSV handling appears in both logging_setup.py and csv_utils.py + +2. **Circular Dependencies** + - logging_setup.py imports from csv_utils.py and config_manager.py + - aws_utils.py imports from partition_utils.py + - Potential for circular import issues + +3. **Inconsistent Abstractions** + - Some files mix high-level and low-level functions + - Unclear boundaries between modules + +## Consolidation Plan + +### 1. Core Utility Modules + +Restructure the utilities into these core modules: + +1. **aws_core_utils.py** + - AWS credential management + - Session handling + - Account discovery + - Move core AWS authentication functions here + +2. **region_utils.py** + - Merge partition_utils.py functionality here + - Region validation and filtering + - Partition detection and mapping + +3. **api_utils.py** + - Move pagination utilities from discovery_utils.py here + - Add general AWS API helpers + - Centralize error handling patterns + +4. **logging_utils.py** + - Keep logging configuration separated + - Remove CSV dependencies + +5. **file_utils.py** + - Consolidate CSV handling here + - Add other file operations + - Include reporting file utilities + +### 2. Implementation Strategy + +1. **Phase 1 - Core Module Creation** + - Create the consolidated modules without removing existing ones + - Redirect imports from old modules to new consolidated ones + - Add deprecation warnings to old modules + +2. **Phase 2 - Migration** + - Update imports throughout the codebase + - Ensure backward compatibility + - Add comprehensive testing + +3. **Phase 3 - Cleanup** + - Remove deprecated modules + - Update documentation + - Clean up any remaining references + +### 3. Proposed Package Structure + +``` +aws_resource_management/ +├── __init__.py +├── utils/ +│ ├── __init__.py +│ ├── aws_core_utils.py +│ ├── region_utils.py +│ ├── api_utils.py +│ ├── logging_utils.py +│ ├── file_utils.py +│ └── config_utils.py +├── cli.py +├── core.py +├── discovery.py +├── reporting.py +├── resource_registry.py +├── managers/ +│ ├── __init__.py +│ ├── base.py +│ ├── ec2.py +│ ├── ecr.py +│ ├── eks.py +│ ├── emr.py +│ └── rds.py +``` + +## Implementation Recommendations + +1. **Maintain GovCloud Support** + - Ensure all utilities properly handle different AWS partitions + - Keep partition detection logic robust + +2. **Preserve Caching Mechanisms** + - Maintain the existing caching patterns during consolidation + - Add more consistent cache invalidation + +3. **Standardize Error Handling** + - Create consistent error handling patterns across utilities + - Implement more detailed error logging + +4. **Centralize Configuration** + - Use a single approach to configuration management + - Make configuration injection more explicit + +5. **Improve Type Annotations** + - Add comprehensive typing information + - Use modern typing features like TypedDict and Protocol + +## Next Steps + +1. Create the new utility modules following the plan above +2. Write comprehensive tests for the new modules +3. Update existing code to use the new modules +4. Gradually deprecate and remove old modules +5. Update documentation to reflect the new structure diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/__init__.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/__init__.py index 9b891146..6636f3f8 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/__init__.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/__init__.py @@ -2,17 +2,20 @@ AWS Resource Management package. """ -# Re-export key modules for easier imports -from aws_resource_management.logging_setup import ( +# Re-export key modules from consolidated utility modules +from aws_resource_management.utils.logging_utils import ( LoggingContext, configure_logging, - initialize_csv_log, - log_action_to_csv, log_operation, log_with_context, setup_logging, ) +from aws_resource_management.utils.file_utils import ( + initialize_action_log_csv as initialize_csv_log, + log_action_to_csv, +) + # Set up a default logger logger = setup_logging() diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/config_manager.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/config_manager.py deleted file mode 100644 index 5a00ba46..00000000 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/config_manager.py +++ /dev/null @@ -1,83 +0,0 @@ -""" -Configuration manager for AWS Resource Management. -""" - -import os -from pathlib import Path -from typing import Any, Dict, Optional - -import yaml - -# Default configuration -DEFAULT_CONFIG = { - "action_csv_file": "resource_actions.csv", - "log_level": "INFO", - "max_workers": 10, - "default_regions": { - "aws": ["us-east-1", "us-east-2", "us-west-1", "us-west-2"], - "aws-us-gov": ["us-gov-east-1", "us-gov-west-1"], - "aws-cn": ["cn-north-1", "cn-northwest-1"], - }, -} - -# Global config cache -_config = None - - -def get_config() -> Dict[str, Any]: - """ - Get configuration with defaults merged with user settings. - - Returns: - Configuration dictionary - """ - global _config - - # Return cached config if available - if _config is not None: - return _config - - # Start with default config - config = DEFAULT_CONFIG.copy() - - # Look for config files in multiple locations - config_paths = [ - os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "config.yaml"), - os.path.expanduser("~/.aws-resource-management/config.yaml"), - ] - - # Try to load and merge configs from files - for path in config_paths: - try: - if os.path.exists(path): - with open(path, "r") as f: - user_config = yaml.safe_load(f) - if user_config and isinstance(user_config, dict): - # Merge with default config - _deep_merge(config, user_config) - except Exception: - # Ignore errors reading config files - pass - - # Cache the config - _config = config - return config - - -def _deep_merge(base: Dict, update: Dict) -> Dict: - """ - Recursively merge dictionaries. - - Args: - base: Base dictionary to update - update: Dictionary with values to merge - - Returns: - Updated base dictionary - """ - for key, value in update.items(): - if isinstance(value, dict) and key in base and isinstance(base[key], dict): - _deep_merge(base[key], value) - else: - base[key] = value - return base diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/csv_utils.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/csv_utils.py deleted file mode 100644 index babf0a72..00000000 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/csv_utils.py +++ /dev/null @@ -1,106 +0,0 @@ -""" -Shared CSV utilities for AWS resource management. -""" - -import csv -import os -from datetime import datetime -from pathlib import Path -from typing import Any, Dict, List, Optional, Set - -def ensure_csv_dir(directory: str) -> str: - """ - Ensure the CSV directory exists and return its path. - - Args: - directory: Directory path to ensure exists - - Returns: - Absolute path to the directory - """ - path = Path(directory) - path.mkdir(parents=True, exist_ok=True) - return str(path.absolute()) - -def initialize_csv_file( - filepath: str, - headers: List[str], - overwrite: bool = False -) -> bool: - """ - Initialize a CSV file with headers if it doesn't exist or overwrite is True. - - Args: - filepath: Path to CSV file - headers: List of column headers - overwrite: Whether to overwrite existing file - - Returns: - True if file was created/initialized, False if file already existed - """ - # Create parent directories if they don't exist - os.makedirs(os.path.dirname(filepath), exist_ok=True) - - # Check if file exists - file_exists = os.path.isfile(filepath) - - # Create new file or overwrite existing file - if not file_exists or overwrite: - with open(filepath, 'w', newline='') as csvfile: - writer = csv.writer(csvfile) - writer.writerow(headers) - return True - - return False - -def write_csv_row(filepath: str, row_data: Dict[str, Any], headers: List[str]) -> None: - """ - Write a row to a CSV file. - - Args: - filepath: Path to CSV file - row_data: Dictionary containing row data - headers: List of column headers in correct order - """ - # Create file with headers if it doesn't exist - if not os.path.isfile(filepath): - initialize_csv_file(filepath, headers) - - # Write the row - with open(filepath, 'a', newline='') as csvfile: - writer = csv.DictWriter(csvfile, fieldnames=headers) - writer.writerow(row_data) - -def format_tags_for_csv(tags: Dict[str, str]) -> str: - """ - Format AWS resource tags for CSV output. - - Args: - tags: Dictionary of tag key-value pairs - - Returns: - Formatted string representation of tags - """ - if not tags: - return "" - return "; ".join([f"{k}={v}" for k, v in tags.items()]) - -def generate_timestamp_filename( - base_name: str, - prefix: Optional[str] = None, - extension: str = "csv" -) -> str: - """ - Generate a filename with timestamp. - - Args: - base_name: Base name for the file - prefix: Optional prefix - extension: File extension without dot - - Returns: - Timestamped filename - """ - timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - prefix_str = f"{prefix}_" if prefix else "" - return f"{prefix_str}{base_name}_{timestamp}.{extension}" diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/partition_utils.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/partition_utils.py deleted file mode 100644 index 59cf2721..00000000 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/partition_utils.py +++ /dev/null @@ -1,172 +0,0 @@ -""" -Centralized AWS partition information and utilities. - -This module provides consistent partition information and utilities -for working with different AWS partitions (commercial AWS, GovCloud, China) -throughout the application. -""" - -from typing import Any, Dict, List, Optional, Set, Tuple - - -# Centralized mapping of partitions to their regions -PARTITION_REGIONS = { - "aws-us-gov": ["us-gov-east-1", "us-gov-west-1"], - "aws-cn": ["cn-north-1", "cn-northwest-1"], - "aws": [ - "us-east-1", - "us-east-2", - "us-west-1", - "us-west-2", - "eu-west-1", - "eu-central-1", - "eu-west-2", - "eu-west-3", - "ap-northeast-1", - "ap-northeast-2", - "ap-southeast-1", - "ap-southeast-2", - "ca-central-1", - "sa-east-1", - "eu-north-1", - "eu-south-1", - "af-south-1", - "ap-east-1", - "ap-south-1", - "me-south-1" - ], -} - -# Sensible defaults for each partition when no regions are specified -DEFAULT_REGIONS = { - "aws-us-gov": ["us-gov-east-1", "us-gov-west-1"], - "aws-cn": ["cn-north-1", "cn-northwest-1"], - "aws": ["us-east-1", "us-east-2", "us-west-1", "us-west-2"], -} - -# The default partition to use if no partition is detected -DEFAULT_PARTITION = "aws-us-gov" - -# Map of region prefixes to partitions for quick lookups -REGION_PREFIX_TO_PARTITION = { - "us-gov-": "aws-us-gov", - "gov-": "aws-us-gov", - "cn-": "aws-cn" -} - -# All known AWS regions (flattened from PARTITION_REGIONS) -ALL_KNOWN_REGIONS = [] -for regions in PARTITION_REGIONS.values(): - ALL_KNOWN_REGIONS.extend(regions) - - -def get_partition_for_region(region_name: str) -> str: - """ - Determine AWS partition based on region name. - - Args: - region_name: AWS region name - - Returns: - AWS partition name (e.g., 'aws', 'aws-us-gov', 'aws-cn') - """ - # Check common region prefixes first - for prefix, partition in REGION_PREFIX_TO_PARTITION.items(): - if region_name.startswith(prefix): - return partition - - # Default to standard AWS partition - return "aws" - - -def get_regions_for_partition(partition: str) -> List[str]: - """ - Get the list of regions for a specific partition. - - Args: - partition: AWS partition name - - Returns: - List of region names in the partition - """ - return PARTITION_REGIONS.get(partition, PARTITION_REGIONS["aws"]) - - -def get_default_regions_for_partition(partition: str) -> List[str]: - """ - Get the default regions commonly used for a partition. - - Args: - partition: AWS partition name - - Returns: - List of default region names - """ - return DEFAULT_REGIONS.get(partition, DEFAULT_REGIONS["aws"]) - - -def is_region_in_partition(region: str, partition: str) -> bool: - """ - Check if a region belongs to the specified partition. - - Args: - region: AWS region name - partition: AWS partition name - - Returns: - True if the region is in the partition, False otherwise - """ - return get_partition_for_region(region) == partition - - -def is_valid_region(region: str) -> bool: - """ - Check if a region is known to be valid without making an API call. - - Args: - region: Region name to validate - - Returns: - True if the region is valid, False otherwise - """ - if not isinstance(region, str): - return False - - # First check if it's in our known regions list - if region in ALL_KNOWN_REGIONS: - return True - - # Check if it follows expected naming patterns for newer regions - # Commercial regions follow patterns like us-east-1, eu-west-2 - if region.startswith(("us-", "eu-", "ap-", "sa-", "ca-", "me-", "af-")) and len(region.split("-")) >= 3: - return True - - # GovCloud regions - if region.startswith("us-gov-") and len(region.split("-")) >= 3: - return True - - # China regions - if region.startswith("cn-") and len(region.split("-")) >= 3: - return True - - return False - - -def filter_regions_for_partition(regions: List[str], partition: str) -> List[str]: - """ - Filter regions to include only those belonging to the specified partition. - - Args: - regions: List of region names - partition: AWS partition name - - Returns: - List of valid region names within the partition - """ - # First validate the regions and filter out invalid ones - valid_regions = [r for r in regions if is_valid_region(r)] - - # Then filter for the specific partition - partition_regions = [r for r in valid_regions if is_region_in_partition(r, partition)] - - return partition_regions diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils.py deleted file mode 100644 index 4f583af5..00000000 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils.py +++ /dev/null @@ -1,18 +0,0 @@ -""" -AWS utility functions for resource management. -""" - -# This file is being deprecated in favor of aws_utils.py which provides more -# comprehensive implementations with better caching and error handling. -# See aws_utils.py for all functionality. - -import warnings - -warnings.warn( - "The utils.py module is deprecated. Please use aws_utils.py instead.", - DeprecationWarning, - stacklevel=2 -) - -# Import everything from aws_utils to maintain backward compatibility -from aws_resource_management.aws_utils import * diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/README.md b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/README.md new file mode 100644 index 00000000..6d949f91 --- /dev/null +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/README.md @@ -0,0 +1,105 @@ +# AWS Resource Management Utility Modules + +This package contains consolidated utility modules for AWS resource management operations, providing a more maintainable and organized codebase. + +## Modules Overview + +### `aws_core_utils.py` + +Core AWS authentication and session management: + +- `get_credentials()` - Get AWS credentials for an account +- `get_session_for_account()` - Get a boto3 session for an AWS account +- `get_account_list()` - Get list of AWS accounts from Organizations API +- `get_organization_accounts()` - Get accounts from AWS Organizations + +### `region_utils.py` + +AWS region and partition utilities: + +- `get_all_regions()` - Get all AWS regions +- `get_enabled_regions()` - Get enabled regions for an account +- `is_valid_region()` - Check if a region is valid +- `get_partition_for_region()` - Get AWS partition for a region +- `filter_regions_for_partition()` - Filter regions by AWS partition + +### `api_utils.py` + +AWS API interaction patterns: + +- `paginate_aws_response()` - Paginate through AWS API responses +- `safe_api_call()` - Safely execute API calls with error handling +- `retry_with_backoff()` - Retry API calls with exponential backoff +- `format_tags()` - Convert AWS tag format to dictionary +- `normalize_resource_state()` - Normalize resource state fields + +### `logging_utils.py` + +Enhanced logging capabilities: + +- `setup_logging()` - Set up basic logging +- `configure_logging()` - Configure advanced logging options +- `log_with_context()` - Log with additional context data +- `log_operation()` - Context manager for operation logging +- `LoggingContext` - Class for managing logging context + +### `file_utils.py` + +File operations, particularly for CSV handling: + +- `initialize_csv_file()` - Initialize CSV file with headers +- `write_csv_row()` - Write a row to a CSV file +- `log_action_to_csv()` - Log resource action to CSV +- `write_json_file()` - Write data to JSON file +- `read_json_file()` - Read data from JSON file + +### `config_utils.py` + +Configuration management: + +- `get_config()` - Get configuration with defaults +- `get_config_value()` - Get specific configuration value +- `update_config()` - Update configuration value and save +- `save_config()` - Save configuration to file + +## Migration Guide + +To migrate from the old utility modules to the new consolidated ones: + +### Option 1: Import from the main utils package + +```python +# Old approach +from aws_resource_management.aws_utils import get_credentials +from aws_resource_management.discovery_utils import paginate_aws_response + +# New approach - main utils package re-exports common functions +from aws_resource_management.utils import get_credentials, paginate_aws_response +``` + +### Option 2: Import from specific utility modules + +```python +# For more specialized functions +from aws_resource_management.utils.region_utils import filter_regions_for_partition +from aws_resource_management.utils.api_utils import retry_with_backoff +``` + +### Logger Usage + +```python +# Old approach +from aws_resource_management.logging_setup import logger, log_operation + +# New approach +from aws_resource_management.utils import logger, log_operation + +# Usage remains the same +with log_operation(logger, "Operation name"): + # Your code here + pass +``` + +## Backward Compatibility + +The old utility modules are still available but will issue deprecation warnings. They now import from the consolidated modules, ensuring functionality remains consistent during the transition. diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/__init__.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/__init__.py new file mode 100644 index 00000000..613e8ee8 --- /dev/null +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/__init__.py @@ -0,0 +1,53 @@ +""" +Consolidated utility modules for AWS Resource Management. + +This package contains the refactored utility modules for AWS resource management +to reduce duplication and improve maintainability. +""" + +# Import frequently used functions for convenience +from aws_resource_management.utils.api_utils import ( + paginate_aws_response, + safe_api_call, + format_tags, + normalize_resource_state, + add_account_info +) + +from aws_resource_management.utils.aws_core_utils import ( + get_credentials, + get_session_for_account, + get_account_list, + create_boto3_client +) + +from aws_resource_management.utils.config_utils import ( + get_config, + get_config_value, + update_config +) + +from aws_resource_management.utils.file_utils import ( + log_action_to_csv, + write_csv_row, + write_json_file, + read_json_file +) + +from aws_resource_management.utils.logging_utils import ( + setup_logging, + configure_logging, + log_operation, + LoggingContext +) + +from aws_resource_management.utils.region_utils import ( + get_all_regions, + get_enabled_regions, + is_valid_region, + get_partition_for_region, + DEFAULT_PARTITION +) + +# Make logger available for direct import +from aws_resource_management.utils.logging_utils import logger diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/discovery_utils.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/api_utils.py similarity index 62% rename from local-app/python-tools/gfl-resource-actions/aws_resource_management/discovery_utils.py rename to local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/api_utils.py index dff44416..0dc7c3fa 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/discovery_utils.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/api_utils.py @@ -1,10 +1,9 @@ """ -Shared discovery utilities for AWS resource management. +AWS API utilities for consistent AWS API interaction patterns. -This module provides common functions used in resource discovery -across different resource types. +This module provides utilities for working with AWS APIs, including pagination, +error handling, and resource normalization. """ - import logging from typing import Any, Dict, List, Optional, Callable, Iterator, Tuple, Union @@ -12,6 +11,7 @@ from botocore.exceptions import ClientError from botocore.paginate import PageIterator +# Configure logger logger = logging.getLogger(__name__) def paginate_aws_response( @@ -82,6 +82,28 @@ def paginate_aws_response( logger.error(f"Error paginating {operation}: {str(e)}") return [] +def safe_api_call(func: Callable, error_message: str, default_return: Any = None) -> Any: + """ + Safely execute an AWS API call with error handling. + + Args: + func: Function to execute (usually a lambda) + error_message: Error message to log on failure + default_return: Value to return on failure + + Returns: + API call result or default value on failure + """ + try: + return func() + except ClientError as e: + error_code = e.response.get("Error", {}).get("Code", "Unknown") + logger.error(f"{error_message}: {error_code} - {str(e)}") + except Exception as e: + logger.error(f"{error_message}: {str(e)}") + + return default_return + def format_tags(tags_list: List[Dict[str, str]]) -> Dict[str, str]: """ Convert AWS tags list format to dictionary. @@ -97,10 +119,7 @@ def format_tags(tags_list: List[Dict[str, str]]) -> Dict[str, str]: return {tag.get("Key", ""): tag.get("Value", "") for tag in tags_list} -def should_exclude_resource( - tags: Dict[str, str], - exclusion_tag: str -) -> bool: +def should_exclude_resource(tags: Dict[str, str], exclusion_tag: str) -> bool: """ Check if a resource should be excluded based on tags. @@ -146,24 +165,81 @@ def add_account_info( if account_name: resource["accountName"] = account_name -def safe_api_call(func, error_message: str, default_return: Any = None) -> Any: +def create_boto3_client( + service: str, + credentials: Dict[str, str], + region: Optional[str] = None +) -> boto3.client: """ - Safely execute an AWS API call with error handling. + Create a boto3 client with provided credentials. Args: - func: Function to execute (usually a lambda) - error_message: Error message to log on failure - default_return: Value to return on failure + service: AWS service name + credentials: Dict containing AWS credentials + region: AWS region name Returns: - API call result or default value on failure + Boto3 client """ - try: - return func() - except ClientError as e: - error_code = e.response.get("Error", {}).get("Code", "Unknown") - logger.error(f"{error_message}: {error_code} - {str(e)}") - except Exception as e: - logger.error(f"{error_message}: {str(e)}") + return boto3.client( + service, + region_name=region, + aws_access_key_id=credentials.get("aws_access_key_id"), + aws_secret_access_key=credentials.get("aws_secret_access_key"), + aws_session_token=credentials.get("aws_session_token"), + ) + +def retry_with_backoff( + func: Callable, + max_retries: int = 3, + initial_backoff: float = 1.0, + backoff_multiplier: float = 2.0, + retryable_exceptions: Optional[List[Exception]] = None, +) -> Any: + """ + Retry a function with exponential backoff. - return default_return + Args: + func: Function to call + max_retries: Maximum number of retries + initial_backoff: Initial backoff time in seconds + backoff_multiplier: Multiplier for backoff time after each retry + retryable_exceptions: List of exception types to retry on + + Returns: + Result of the function call + """ + import time + + if retryable_exceptions is None: + retryable_exceptions = [ClientError] + + retries = 0 + backoff = initial_backoff + + while True: + try: + return func() + except tuple(retryable_exceptions) as e: + # Check if we should retry based on error + if retries >= max_retries: + logger.warning(f"Max retries ({max_retries}) reached: {str(e)}") + raise + + if isinstance(e, ClientError): + error_code = e.response.get("Error", {}).get("Code", "") + if error_code in ["ThrottlingException", "RequestLimitExceeded", "Throttling"]: + logger.info(f"Request throttled, retrying ({retries+1}/{max_retries}) after {backoff}s") + else: + # Don't retry on permission errors + if error_code in ["AccessDenied", "UnauthorizedOperation"]: + logger.warning(f"Permission error, not retrying: {error_code}") + raise + logger.info(f"Retrying on error {error_code} ({retries+1}/{max_retries}) after {backoff}s") + else: + logger.info(f"Retrying on error: {str(e)} ({retries+1}/{max_retries}) after {backoff}s") + + # Sleep and retry + time.sleep(backoff) + retries += 1 + backoff *= backoff_multiplier diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/aws_utils.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/aws_core_utils.py similarity index 58% rename from local-app/python-tools/gfl-resource-actions/aws_resource_management/aws_utils.py rename to local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/aws_core_utils.py index 08f9e2a8..57a95f0d 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/aws_utils.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/aws_core_utils.py @@ -1,8 +1,9 @@ """ -AWS utility functions for credential management and account discovery. -Simplified version with reduced API calls and streamlined functionality. -""" +Core AWS utilities for credential management and session handling. +This module provides centralized functionality for AWS authentication, credential management, +and session handling with optimized caching. +""" import configparser import logging import os @@ -13,20 +14,19 @@ import boto3 import botocore -from aws_resource_management.logging_setup import setup_logging -from aws_resource_management.partition_utils import ( +from botocore.exceptions import ClientError, NoCredentialsError, ProfileNotFound + +# Import utilities from other modules +from aws_resource_management.utils.region_utils import ( DEFAULT_PARTITION, - DEFAULT_REGIONS, - PARTITION_REGIONS, + detect_partition_from_credentials, + get_all_regions, get_default_regions_for_partition, - get_partition_for_region, - get_regions_for_partition, - is_region_in_partition, - is_valid_region, ) -from botocore.exceptions import ClientError, NoCredentialsError, ProfileNotFound +from aws_resource_management.utils.logging_utils import setup_logging -logger = setup_logging() +# Configure logger +logger = setup_logging(__name__) # --------------------------------------------------------------------------- # Global caches to reduce API calls @@ -35,182 +35,140 @@ _session_cache = {} _session_cache_lock = threading.Lock() -# Region and partition information cache -_region_cache = { - "timestamp": 0, - "all_regions": {}, - "enabled_regions": {}, - # Use PARTITION_REGIONS from partition_utils for partition region data - "partition_regions": PARTITION_REGIONS, -} - # Cache expiration time in seconds (1 hour) CACHE_EXPIRY = 3600 # --------------------------------------------------------------------------- # String utility functions for safer string handling # --------------------------------------------------------------------------- - - def is_string(value: Any) -> bool: """Check if a value is a string.""" return isinstance(value, str) - def safe_startswith(value: Any, prefix: Union[str, Tuple[str, ...]]) -> bool: """Safely check if a value starts with a prefix, handling None values.""" if not is_string(value): return False - if isinstance(prefix, str): return value.startswith(prefix) elif isinstance(prefix, tuple) and all(is_string(p) for p in prefix): return value.startswith(prefix) - return False - def safe_in(substring: Any, container: Any) -> bool: """Safely check if a substring is in a container, handling None values.""" if substring is None or container is None: return False - try: return substring in container except (TypeError, ValueError): return False - -# --------------------------------------------------------------------------- -# Simplified region and partition handling -# --------------------------------------------------------------------------- - -# Export partition functions directly from partition_utils for backwards compatibility -# These explicit re-exports make it clearer that they're from partition_utils -# rather than just happening to have the same name - -def get_all_regions(partition: Optional[str] = None) -> List[str]: - """Get all regions, using cache to minimize API calls.""" - # Use predefined regions for special partitions - if partition in ("aws-us-gov", "aws-cn"): - return get_regions_for_partition(partition) - - # Check cache first - current_time = time.time() - cache_key = partition or "all" - - if ( - cache_key in _region_cache["all_regions"] - and current_time - _region_cache["timestamp"] < CACHE_EXPIRY - ): - return _region_cache["all_regions"][cache_key] - - # Cache miss: fetch regions - try: - ec2 = boto3.client("ec2", region_name="us-east-1") - all_regions = [ - region["RegionName"] for region in ec2.describe_regions()["Regions"] - ] - - # Cache the full list - _region_cache["all_regions"]["all"] = all_regions - _region_cache["timestamp"] = current_time - - # If partition specified, filter and cache that too - if partition: - filtered_regions = [ - r - for r in all_regions - if is_string(r) and get_partition_for_region(r) == partition - ] - _region_cache["all_regions"][partition] = filtered_regions - return filtered_regions - - return all_regions - except Exception as e: - logger.warning(f"Error getting regions: {str(e)}") - return get_regions_for_partition(partition or DEFAULT_PARTITION) - - # --------------------------------------------------------------------------- # Simplified credential management # --------------------------------------------------------------------------- - - def get_credentials( account_id: str, profile_name: Optional[str] = None, region: Optional[str] = None ) -> Optional[Dict[str, Any]]: - """Get AWS credentials with simplified logic and better caching.""" + """ + Get AWS credentials with simplified logic and better caching. + + Args: + account_id: AWS account ID + profile_name: Optional AWS profile name + region: Optional AWS region + + Returns: + Dictionary of credentials or None if not found + """ cache_key = f"creds:{account_id}:{profile_name or 'default'}" - + with _session_cache_lock: if ( cache_key in _session_cache and _session_cache[cache_key]["timestamp"] > time.time() - CACHE_EXPIRY ): return _session_cache[cache_key]["credentials"] - + # Try profile approach first credentials = _get_credentials_from_profile(account_id, profile_name) if credentials: _cache_credentials(cache_key, credentials) return credentials - + # Fall back to role assumption credentials = _assume_role_for_account(account_id) if credentials: _cache_credentials(cache_key, credentials) return credentials - + return None - def _cache_credentials(cache_key: str, credentials: Dict[str, Any]) -> None: - """Cache credentials with timestamp.""" + """ + Cache credentials with timestamp. + + Args: + cache_key: Cache key + credentials: Credentials dictionary to cache + """ with _session_cache_lock: _session_cache[cache_key] = { "credentials": credentials, "timestamp": time.time(), } - def _get_credentials_from_profile( account_id: str, profile_name: Optional[str] = None ) -> Optional[Dict[str, Any]]: - """Get credentials from a profile.""" + """ + Get credentials from an AWS profile. + + Args: + account_id: AWS account ID + profile_name: Optional profile name to use + + Returns: + Dictionary of credentials or None if not found + """ try: # Use specified profile or find one profile_to_use = profile_name or _find_profile_for_account(account_id) if not profile_to_use: return None - + # Get credentials from the profile session = boto3.Session(profile_name=profile_to_use) if session.get_credentials() is None: return None - + return { "aws_access_key_id": session.get_credentials().access_key, "aws_secret_access_key": session.get_credentials().secret_key, "aws_session_token": session.get_credentials().token, } except Exception as e: - logger.debug( - f"Error getting credentials from profile for {account_id}: {str(e)}" - ) + logger.debug(f"Error getting credentials from profile for {account_id}: {str(e)}") return None - def _find_profile_for_account(account_id: str) -> Optional[str]: - """Find an AWS profile for an account with simplified logic.""" + """ + Find an AWS profile for a specific account. + + Args: + account_id: AWS account ID + + Returns: + Profile name or None if not found + """ try: config_path = os.path.expanduser("~/.aws/config") if not os.path.exists(config_path): return None - + config = configparser.ConfigParser() config.read(config_path) - + # Prioritized profile patterns profile_patterns = [ f"{account_id}.AdministratorAccess", @@ -218,16 +176,13 @@ def _find_profile_for_account(account_id: str) -> Optional[str]: f"{account_id}-gov.administratoraccess", f"{account_id}-gov.inf-admin-t2", ] - + # Check for exact matches of preferred profiles first for section in config.sections(): - section_name = ( - section[8:] if safe_startswith(section, "profile ") else section - ) - + section_name = section[8:] if safe_startswith(section, "profile ") else section if section_name.lower() in [p.lower() for p in profile_patterns]: return section_name - + # Then check for any profile with matching account ID for section in config.sections(): if ( @@ -235,14 +190,21 @@ def _find_profile_for_account(account_id: str) -> Optional[str]: and config.get(section, "sso_account_id") == account_id ): return section[8:] if safe_startswith(section, "profile ") else section - + return None except Exception: return None - def _assume_role_for_account(account_id: str) -> Optional[Dict[str, Any]]: - """Assume a role to get credentials.""" + """ + Assume a role to get credentials for an account. + + Args: + account_id: AWS account ID + + Returns: + Dictionary of credentials or None if failed + """ try: # Try common role names in order of preference for role_name in ["OrganizationAccountAccessRole", "AWSControlTowerExecution"]: @@ -251,7 +213,6 @@ def _assume_role_for_account(account_id: str) -> Optional[Dict[str, Any]]: response = boto3.client("sts").assume_role( RoleArn=role_arn, RoleSessionName="ResourceManagementSession" ) - credentials = response["Credentials"] return { "aws_access_key_id": credentials["AccessKeyId"], @@ -260,85 +221,30 @@ def _assume_role_for_account(account_id: str) -> Optional[Dict[str, Any]]: } except botocore.exceptions.ClientError: continue - return None except Exception: return None - -def detect_partition_from_credentials(credentials: Dict[str, str]) -> str: - """Detect partition from credentials with minimal API calls.""" - if not credentials: - return DEFAULT_PARTITION # Default for environment from partition_utils - - # Check credential format first (fast, no API calls) - access_key = str(credentials.get("aws_access_key_id", "")).lower() - session_token = str(credentials.get("aws_session_token", "")).lower() - - # Check for GovCloud indicators - if any( - safe_in(indicator, access_key) or safe_in(indicator, session_token) - for indicator in ["gov", "usgovcloud"] - ): - return "aws-us-gov" - - # Check for China indicators - if any( - safe_in(indicator, access_key) or safe_in(indicator, session_token) - for indicator in ["cn-", "china"] - ): - return "aws-cn" - - # Try one API call to most likely partition based on environment - try: - boto3.client( - "sts", - region_name="us-gov-west-1", # Try GovCloud first based on environment - aws_access_key_id=credentials.get("aws_access_key_id"), - aws_secret_access_key=credentials.get("aws_secret_access_key"), - aws_session_token=credentials.get("aws_session_token"), - ).get_caller_identity() - return "aws-us-gov" - except Exception: - # If GovCloud fails, try standard AWS - try: - boto3.client( - "sts", - region_name="us-east-1", - aws_access_key_id=credentials.get("aws_access_key_id"), - aws_secret_access_key=credentials.get("aws_secret_access_key"), - aws_session_token=credentials.get("aws_session_token"), - ).get_caller_identity() - return "aws" - except Exception: - # Default to GovCloud for environment - return DEFAULT_PARTITION - - # --------------------------------------------------------------------------- # Simplified session management # --------------------------------------------------------------------------- - - @lru_cache(maxsize=128) def get_session_for_account( account_id: str, region: Optional[str] = None, profile_name: Optional[str] = None ) -> Optional[boto3.Session]: """ Get a boto3 session for the specified account with proper role assumption. - + Args: account_id: AWS account ID region: AWS region name (optional) profile_name: AWS profile name to use (optional) - + Returns: Boto3 session with appropriate credentials """ - cache_key = ( - f"session:{account_id}:{region or 'default'}:{profile_name or 'default'}" - ) - + cache_key = f"session:{account_id}:{region or 'default'}:{profile_name or 'default'}" + # Return cached session if available with _session_cache_lock: if ( @@ -346,34 +252,28 @@ def get_session_for_account( and _session_cache[cache_key]["timestamp"] > time.time() - CACHE_EXPIRY ): return _session_cache[cache_key]["session"] - + # Try getting a session using standard methods try: # Method 1: Try using a profile named after the account ID or the specified profile try: - session = boto3.Session( - profile_name=profile_name or account_id, region_name=region - ) + session = boto3.Session(profile_name=profile_name or account_id, region_name=region) # Validate session by checking identity sts = session.client("sts") identity = sts.get_caller_identity() if identity.get("Account") == account_id: - logger.debug( - f"Using profile {profile_name or account_id} for authentication" - ) - + logger.debug(f"Using profile {profile_name or account_id} for authentication") # Cache the session with _session_cache_lock: _session_cache[cache_key] = { "session": session, "timestamp": time.time(), } - return session except (ProfileNotFound, NoCredentialsError, ClientError): # Profile not found or invalid, continue to next method pass - + # Method 2: Use credentials to create a session credentials = get_credentials(account_id, profile_name) if credentials: @@ -383,21 +283,19 @@ def get_session_for_account( aws_session_token=credentials.get("aws_session_token"), region_name=region, ) - # Cache the session with _session_cache_lock: _session_cache[cache_key] = { "session": session, "timestamp": time.time(), } - logger.debug(f"Created session for account {account_id} using credentials") return session - + # Method 3: Try to assume role in target account using default session default_session = boto3.Session(region_name=region) sts = default_session.client("sts") - + # Try common cross-account roles for role_name in ["OrganizationAccountAccessRole", "AWSControlTowerExecution"]: try: @@ -405,7 +303,7 @@ def get_session_for_account( response = sts.assume_role( RoleArn=role_arn, RoleSessionName="ResourceManagementSession" ) - + # Create session with temporary credentials credentials = response["Credentials"] session = boto3.Session( @@ -414,37 +312,37 @@ def get_session_for_account( aws_session_token=credentials["SessionToken"], region_name=region, ) - + # Cache the session with _session_cache_lock: _session_cache[cache_key] = { "session": session, "timestamp": time.time(), } - logger.debug(f"Assumed {role_name} in account {account_id}") return session except ClientError: # Role assumption failed, try the next role continue - + # If all methods fail, raise exception raise Exception(f"Could not authenticate to account {account_id}") - except Exception as e: logger.error(f"Failed to get session for account {account_id}: {str(e)}") raise - # --------------------------------------------------------------------------- # Account discovery with caching # --------------------------------------------------------------------------- - - def get_account_list() -> List[Dict[str, str]]: - """Get AWS accounts with caching to minimize API calls.""" + """ + Get AWS accounts with caching to minimize API calls. + + Returns: + List of dictionaries with account information + """ cache_key = "accounts" - + # Check cache first with _session_cache_lock: if ( @@ -452,14 +350,12 @@ def get_account_list() -> List[Dict[str, str]]: and _session_cache[cache_key]["timestamp"] > time.time() - CACHE_EXPIRY ): return _session_cache[cache_key]["accounts"] - + # Try Organizations API first try: session = boto3.Session() accounts = [] - for account in ( - session.client("organizations").get_paginator("list_accounts").paginate() - ): + for account in session.client("organizations").get_paginator("list_accounts").paginate(): accounts.extend( [ {"account_id": acc["Id"], "account_name": acc["Name"]} @@ -467,28 +363,26 @@ def get_account_list() -> List[Dict[str, str]]: if acc["Status"] == "ACTIVE" ] ) - + # Cache the accounts with _session_cache_lock: _session_cache[cache_key] = {"accounts": accounts, "timestamp": time.time()} - return accounts except Exception as e: logger.warning(f"Failed to get accounts from Organizations API: {str(e)}") + # Fall back to profiles accounts = _get_accounts_from_profiles() - + # Cache these accounts too with _session_cache_lock: _session_cache[cache_key] = {"accounts": accounts, "timestamp": time.time()} - return accounts - def get_organization_accounts() -> List[Dict[str, str]]: """ Get list of accounts from AWS Organization. - + Returns: List of dictionaries with account information """ @@ -497,7 +391,7 @@ def get_organization_accounts() -> List[Dict[str, str]]: session = boto3.Session() org_client = session.client("organizations") accounts = [] - + # Use pagination to get all accounts paginator = org_client.get_paginator("list_accounts") for page in paginator.paginate(): @@ -506,7 +400,7 @@ def get_organization_accounts() -> List[Dict[str, str]]: accounts.append( {"account_id": account["Id"], "account_name": account["Name"]} ) - + logger.info(f"Found {len(accounts)} accounts in Organizations API") return accounts except Exception as e: @@ -514,39 +408,42 @@ def get_organization_accounts() -> List[Dict[str, str]]: logger.info("Falling back to accounts from profiles") return _get_accounts_from_profiles() - def _get_accounts_from_profiles() -> List[Dict[str, str]]: - """Get accounts from local AWS config with simplified logic.""" + """ + Get accounts from local AWS config. + + Returns: + List of dictionaries with account information + """ accounts_by_id = {} # Use dict to avoid duplicates - try: config_path = os.path.expanduser("~/.aws/config") if not os.path.exists(config_path): return [] - + config = configparser.ConfigParser() config.read(config_path) - + for section in config.sections(): if not config.has_option(section, "sso_account_id"): continue - + account_id = config.get(section, "sso_account_id") account_name = ( config.get(section, "sso_account_name") if config.has_option(section, "sso_account_name") else account_id ) - + # Get profile name profile = section[8:] if safe_startswith(section, "profile ") else section - + # Keep track if we should replace an existing entry replace_existing = account_id in accounts_by_id if replace_existing and config.has_option(section, "sso_role_name"): role_name = config.get(section, "sso_role_name") replace_existing = role_name in ["AdministratorAccess", "inf-admin-t2"] - + # Store or replace account info if replace_existing or account_id not in accounts_by_id: accounts_by_id[account_id] = { @@ -554,151 +451,17 @@ def _get_accounts_from_profiles() -> List[Dict[str, str]]: "account_name": account_name, "profile": profile, } - + return list(accounts_by_id.values()) except Exception as e: logger.error(f"Error reading AWS profiles: {str(e)}") return [] - -# --------------------------------------------------------------------------- -# Simplified region utilities -# --------------------------------------------------------------------------- - - -def get_enabled_regions( - credentials: Dict[str, str], partition: Optional[str] = None -) -> List[str]: - """Get enabled regions with minimized API calls using cache.""" - # Determine partition if not specified - if not partition: - partition = detect_partition_from_credentials(credentials) - - # For GovCloud/China, return predefined regions to avoid API calls - if partition in ("aws-us-gov", "aws-cn"): - return get_regions_for_partition(partition) - - # Generate cache key from credentials - creds_hash = hash( - f"{credentials.get('aws_access_key_id', '')}:{credentials.get('aws_secret_access_key', '')}" - ) - cache_key = f"enabled_regions:{creds_hash}:{partition}" - - # Check cache - with _session_cache_lock: - if ( - cache_key in _region_cache["enabled_regions"] - and time.time() - _region_cache["timestamp"] < CACHE_EXPIRY - ): - return _region_cache["enabled_regions"][cache_key] - - # Make a single API call to describe_regions rather than checking each region - try: - session = boto3.Session( - aws_access_key_id=credentials.get("aws_access_key_id"), - aws_secret_access_key=credentials.get("aws_secret_access_key"), - aws_session_token=credentials.get("aws_session_token"), - ) - - regions = [ - region["RegionName"] - for region in session.client( - "ec2", region_name="us-east-1" - ).describe_regions()["Regions"] - ] - - # Cache the result - with _session_cache_lock: - _region_cache["enabled_regions"][cache_key] = regions - _region_cache["timestamp"] = time.time() - - return regions - except Exception: - # Fall back to default regions from partition_utils - return get_default_regions_for_partition(partition) - - -def list_enabled_regions( - session: boto3.Session, exclude_regions: Optional[List[str]] = None -) -> List[str]: - """ - List all enabled regions for the account, respecting partition. - - Args: - session: Boto3 session - exclude_regions: List of regions to exclude - - Returns: - List of enabled region names - """ - if exclude_regions is None: - exclude_regions = [] - - try: - ec2 = session.client( - "ec2", region_name="us-east-1" - ) # Use US East 1 for global operations - regions_response = ec2.describe_regions( - AllRegions=False - ) # Only get enabled regions - - regions = [ - r["RegionName"] - for r in regions_response["Regions"] - if r["RegionName"] not in exclude_regions - ] - return regions - except Exception as e: - logger.warning(f"Could not retrieve regions, using defaults: {str(e)}") - - # For GovCloud, provide sensible defaults - if session.region_name and session.region_name.startswith("us-gov-"): - return ["us-gov-east-1", "us-gov-west-1"] - - # Default to common commercial regions - return ["us-east-1", "us-east-2", "us-west-1", "us-west-2"] - - -# Replace is_valid_region with import from partition_utils -def is_valid_region(region: str) -> bool: - """ - Check if a region is valid without making an API call. - Legacy function - delegates to partition_utils.is_valid_region. - - Args: - region: Region name to check - - Returns: - True if the region is valid, False otherwise - """ - from aws_resource_management.partition_utils import is_valid_region as _is_valid_region - return _is_valid_region(region) - - -# --------------------------------------------------------------------------- -# Function aliases for backward compatibility -# --------------------------------------------------------------------------- - -# Explicitly document this is an alias for get_enabled_regions -get_available_regions = get_enabled_regions - - -# Ensure create_boto3_client function exists -def create_boto3_client(service, credentials, region=None): - """Create a boto3 client with provided credentials. - - Args: - service: AWS service name - credentials: Dict containing AWS credentials - region: AWS region name - - Returns: - Boto3 client - """ - return boto3.client( - service, - region_name=region, - aws_access_key_id=credentials.get("aws_access_key_id"), - aws_secret_access_key=credentials.get("aws_secret_access_key"), - aws_session_token=credentials.get("aws_session_token"), - ) +# Function alias for backward compatibility +create_boto3_client = lambda service, credentials, region=None: boto3.client( + service, + region_name=region, + aws_access_key_id=credentials.get("aws_access_key_id"), + aws_secret_access_key=credentials.get("aws_secret_access_key"), + aws_session_token=credentials.get("aws_session_token"), +) diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/config_utils.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/config_utils.py new file mode 100644 index 00000000..cb1e17c1 --- /dev/null +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/config_utils.py @@ -0,0 +1,165 @@ +""" +Configuration management utilities for AWS Resource Management. + +This module provides functions for loading and managing configuration +from multiple sources. +""" +import os +from pathlib import Path +from typing import Any, Dict, List, Optional +import yaml + +# Default configuration +DEFAULT_CONFIG = { + "action_csv_file": "resource_actions.csv", + "log_level": "INFO", + "max_workers": 10, + "default_regions": { + "aws": ["us-east-1", "us-east-2", "us-west-1", "us-west-2"], + "aws-us-gov": ["us-gov-east-1", "us-gov-west-1"], + "aws-cn": ["cn-north-1", "cn-northwest-1"], + }, +} + +# Global config cache +_config = None + +def get_config() -> Dict[str, Any]: + """ + Get configuration with defaults merged with user settings. + + Returns: + Configuration dictionary + """ + global _config + + # Return cached config if available + if _config is not None: + return _config + + # Start with default config + config = DEFAULT_CONFIG.copy() + + # Look for config files in multiple locations + config_paths = [ + os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "..", "config.yaml"), + os.path.expanduser("~/.aws-resource-management/config.yaml"), + ] + + # Try to load and merge configs from files + for path in config_paths: + try: + if os.path.exists(path): + with open(path, "r") as f: + user_config = yaml.safe_load(f) + if user_config and isinstance(user_config, dict): + # Merge with default config + _deep_merge(config, user_config) + except Exception: + # Ignore errors reading config files + pass + + # Cache the config + _config = config + return config + +def _deep_merge(base: Dict, update: Dict) -> Dict: + """ + Recursively merge dictionaries. + + Args: + base: Base dictionary to update + update: Dictionary with values to merge + + Returns: + Updated base dictionary + """ + for key, value in update.items(): + if isinstance(value, dict) and key in base and isinstance(base[key], dict): + _deep_merge(base[key], value) + else: + base[key] = value + return base + +def get_config_value(key: str, default: Any = None) -> Any: + """ + Get a specific configuration value. + + Args: + key: Configuration key to retrieve + default: Default value if key not found + + Returns: + Configuration value or default + """ + config = get_config() + + # Handle nested keys with dot notation (e.g., "aws.region") + if "." in key: + parts = key.split(".") + current = config + for part in parts: + if not isinstance(current, dict) or part not in current: + return default + current = current[part] + return current + + return config.get(key, default) + +def save_config(config: Dict[str, Any], config_path: Optional[str] = None) -> bool: + """ + Save configuration to a file. + + Args: + config: Configuration dictionary to save + config_path: Path to save config to (default: user config path) + + Returns: + True if saved successfully, False otherwise + """ + if config_path is None: + config_path = os.path.expanduser("~/.aws-resource-management/config.yaml") + + try: + # Ensure directory exists + os.makedirs(os.path.dirname(config_path), exist_ok=True) + + # Write config + with open(config_path, "w") as f: + yaml.safe_dump(config, f, default_flow_style=False) + + # Update cache + global _config + _config = config + + return True + except Exception: + return False + +def update_config(key: str, value: Any, config_path: Optional[str] = None) -> bool: + """ + Update a specific configuration value and save. + + Args: + key: Configuration key to update + value: New value + config_path: Path to save config to (default: user config path) + + Returns: + True if updated successfully, False otherwise + """ + config = get_config() + + # Handle nested keys with dot notation + if "." in key: + parts = key.split(".") + current = config + for i, part in enumerate(parts[:-1]): + if part not in current or not isinstance(current[part], dict): + current[part] = {} + current = current[part] + current[parts[-1]] = value + else: + config[key] = value + + return save_config(config, config_path) diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/file_utils.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/file_utils.py new file mode 100644 index 00000000..2a91d8b0 --- /dev/null +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/file_utils.py @@ -0,0 +1,328 @@ +""" +File handling utilities for AWS Resource Management. + +This module provides functions for file operations, including CSV file handling +and reporting file management. +""" +import csv +import json +import os +from datetime import datetime +from pathlib import Path +from typing import Any, Dict, List, Optional, Set, Union + +# Logger for this module +import logging +logger = logging.getLogger(__name__) + +def ensure_directory(directory_path: Union[str, Path]) -> str: + """ + Ensure a directory exists and create it if it doesn't. + + Args: + directory_path: Directory path + + Returns: + Absolute path to the directory + """ + path = Path(directory_path) + path.mkdir(parents=True, exist_ok=True) + return str(path.absolute()) + +def generate_timestamp_filename( + base_name: str, + prefix: Optional[str] = None, + extension: str = "csv" +) -> str: + """ + Generate a filename with timestamp. + + Args: + base_name: Base name for the file + prefix: Optional prefix + extension: File extension without dot + + Returns: + Timestamped filename + """ + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + prefix_str = f"{prefix}_" if prefix else "" + return f"{prefix_str}{base_name}_{timestamp}.{extension}" + +def get_logs_directory(base_dir: Optional[str] = None) -> str: + """ + Get the logs directory path, creating it if it doesn't exist. + + Args: + base_dir: Base directory to place logs in, defaults to module path + + Returns: + Absolute path to logs directory + """ + if base_dir is None: + # Default to a logs directory in the parent of the current module + base_dir = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "logs" + ) + + return ensure_directory(base_dir) + +# CSV Handling Functions +def initialize_csv_file( + filepath: str, + headers: List[str], + overwrite: bool = False +) -> bool: + """ + Initialize a CSV file with headers if it doesn't exist or overwrite is True. + + Args: + filepath: Path to CSV file + headers: List of column headers + overwrite: Whether to overwrite existing file + + Returns: + True if file was created/initialized, False if file already existed + """ + # Create parent directories if they don't exist + os.makedirs(os.path.dirname(filepath), exist_ok=True) + + # Check if file exists + file_exists = os.path.isfile(filepath) + + # Create new file or overwrite existing file + if not file_exists or overwrite: + with open(filepath, 'w', newline='') as csvfile: + writer = csv.writer(csvfile) + writer.writerow(headers) + return True + + return False + +def write_csv_row(filepath: str, row_data: Dict[str, Any], headers: List[str]) -> None: + """ + Write a row to a CSV file. + + Args: + filepath: Path to CSV file + row_data: Dictionary containing row data + headers: List of column headers in correct order + """ + # Create file with headers if it doesn't exist + if not os.path.isfile(filepath): + initialize_csv_file(filepath, headers) + + # Write the row + with open(filepath, 'a', newline='') as csvfile: + writer = csv.DictWriter(csvfile, fieldnames=headers) + writer.writerow(row_data) + +def write_multiple_csv_rows( + filepath: str, + rows: List[Dict[str, Any]], + headers: List[str] +) -> None: + """ + Write multiple rows to a CSV file. + + Args: + filepath: Path to CSV file + rows: List of dictionaries containing row data + headers: List of column headers in correct order + """ + # Create file with headers if it doesn't exist + if not os.path.isfile(filepath): + initialize_csv_file(filepath, headers) + + # Write the rows + with open(filepath, 'a', newline='') as csvfile: + writer = csv.DictWriter(csvfile, fieldnames=headers) + writer.writerows(rows) + +def read_csv_file(filepath: str) -> List[Dict[str, str]]: + """ + Read a CSV file and return rows as dictionaries. + + Args: + filepath: Path to CSV file + + Returns: + List of dictionaries containing row data + """ + if not os.path.exists(filepath): + logger.warning(f"CSV file not found: {filepath}") + return [] + + try: + with open(filepath, 'r', newline='') as csvfile: + reader = csv.DictReader(csvfile) + return list(reader) + except Exception as e: + logger.error(f"Error reading CSV file {filepath}: {str(e)}") + return [] + +def format_tags_for_csv(tags: Dict[str, str]) -> str: + """ + Format AWS resource tags for CSV output. + + Args: + tags: Dictionary of tag key-value pairs + + Returns: + Formatted string representation of tags + """ + if not tags: + return "" + return "; ".join([f"{k}={v}" for k, v in tags.items()]) + +def initialize_action_log_csv( + csv_file: Optional[str] = None, + log_dir: Optional[str] = None +) -> str: + """ + Initialize CSV log file for resource actions with headers. + + Args: + csv_file: CSV file name (default: resource_actions.csv) + log_dir: Directory to place log file (default: project logs dir) + + Returns: + Path to the CSV log file + """ + if csv_file is None: + csv_file = "resource_actions.csv" + + # Configure CSV logging directory + if log_dir is None: + log_dir = get_logs_directory() + + csv_path = os.path.join(log_dir, csv_file) + + # Initialize with headers + headers = [ + "timestamp", + "account_id", + "account_name", + "resource_type", + "resource_id", + "resource_name", + "action", + "region", + "status", + "details", + "dry_run", + "schedule", + ] + initialize_csv_file(csv_path, headers, overwrite=False) + return csv_path + +def log_action_to_csv( + account_id: str, + account_name: str, + resource_type: str, + resource_id: str, + resource_name: str, + action: str, + region: str, + status: str, + details: str = "", + dry_run: bool = False, + existing_schedule: Optional[str] = None, + csv_file: Optional[str] = None, + log_dir: Optional[str] = None, +) -> None: + """ + Log resource action to CSV file for tracking. + + Args: + account_id: AWS account ID + account_name: AWS account name + resource_type: Resource type (ec2, rds, eks, emr) + resource_id: Resource ID + resource_name: Resource name + action: Action performed (stop, start) + region: AWS region + status: Action status + details: Additional details + dry_run: Whether this was a dry run + existing_schedule: Existing schedule for the resource + csv_file: Custom CSV file name + log_dir: Custom log directory + """ + timestamp = datetime.now().isoformat() + + # Initialize CSV log file + csv_path = initialize_action_log_csv(csv_file, log_dir) + + # Prepare row data + row_data = { + "timestamp": timestamp, + "account_id": account_id, + "account_name": account_name, + "resource_type": resource_type, + "resource_id": resource_id, + "resource_name": resource_name, + "action": action, + "region": region, + "status": status, + "details": details, + "dry_run": "Yes" if dry_run else "No", + "schedule": existing_schedule or "", + } + + # Headers must match the keys in row_data + headers = list(row_data.keys()) + + # Write to CSV + write_csv_row(csv_path, row_data, headers) + +def write_json_file( + filepath: str, + data: Any, + indent: int = 2, + ensure_dir: bool = True +) -> bool: + """ + Write data to a JSON file. + + Args: + filepath: Path to JSON file + data: Data to write + indent: Indentation level for JSON + ensure_dir: Whether to ensure the directory exists + + Returns: + True if successful, False if failed + """ + try: + if ensure_dir: + os.makedirs(os.path.dirname(filepath), exist_ok=True) + + with open(filepath, 'w') as f: + json.dump(data, f, indent=indent) + return True + except Exception as e: + logger.error(f"Error writing JSON file {filepath}: {str(e)}") + return False + +def read_json_file(filepath: str, default: Any = None) -> Any: + """ + Read data from a JSON file. + + Args: + filepath: Path to JSON file + default: Default value to return if file doesn't exist or can't be parsed + + Returns: + Parsed JSON data or default value + """ + if not os.path.exists(filepath): + return default + + try: + with open(filepath, 'r') as f: + return json.load(f) + except Exception as e: + logger.error(f"Error reading JSON file {filepath}: {str(e)}") + return default diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/logging_setup.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/logging_utils.py similarity index 68% rename from local-app/python-tools/gfl-resource-actions/aws_resource_management/logging_setup.py rename to local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/logging_utils.py index 7e4d73d5..a1894a07 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/logging_setup.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/logging_utils.py @@ -1,5 +1,8 @@ """ Enhanced logging utilities for AWS Resource Management. + +This module provides a centralized logging configuration and context-aware logging +functionality. """ import csv import datetime @@ -12,32 +15,36 @@ from pathlib import Path from typing import Any, Callable, Dict, Optional -from aws_resource_management.config_manager import get_config -from aws_resource_management.csv_utils import ensure_csv_dir, initialize_csv_file, write_csv_row +# Import our file utilities +from aws_resource_management.utils.file_utils import ( + ensure_directory, + initialize_action_log_csv, + log_action_to_csv +) # Global context data that can be included in log messages class LoggingContext: + """Context manager for logging additional data with log messages.""" _context = {} - + @classmethod def get(cls) -> Dict[str, Any]: """Get the current context dictionary.""" return cls._context.copy() - + @classmethod def set(cls, **kwargs) -> None: """Set context values.""" cls._context.update(kwargs) - + @classmethod def clear(cls) -> None: """Clear the context.""" cls._context.clear() - class JSONFormatter(logging.Formatter): """Format log records as JSON.""" - + def format(self, record: logging.LogRecord) -> str: """Format the log record as JSON.""" log_data = { @@ -46,41 +53,47 @@ def format(self, record: logging.LogRecord) -> str: "logger": record.name, "message": record.getMessage(), } - + # Add exception info if present if record.exc_info: log_data["exception"] = self.formatException(record.exc_info) - + # Add context data if present if hasattr(record, "context") and record.context: log_data.update(record.context) - + # Add LoggingContext data log_data.update(LoggingContext.get()) - + return json.dumps(log_data) - -def setup_logging(name="aws_resource_management", level=logging.INFO) -> logging.Logger: - """Set up basic logging.""" +def setup_logging(name: str = "aws_resource_management", level: int = logging.INFO) -> logging.Logger: + """ + Set up basic logging with simple configuration. + + Args: + name: Logger name + level: Logging level + + Returns: + Configured logger + """ logger = logging.getLogger(name) - + # Only configure if handlers aren't already set up if not logger.handlers: logger.setLevel(level) - + # Create console handler with formatter handler = logging.StreamHandler(sys.stdout) handler.setFormatter( logging.Formatter("%(asctime)s [%(levelname)s] %(name)s - %(message)s") ) - logger.addHandler(handler) logger.propagate = False - + return logger - def configure_logging( app_name: str, log_level: str = "INFO", @@ -91,7 +104,7 @@ def configure_logging( ) -> logging.Logger: """ Configure logging with advanced features. - + Args: app_name: Application name log_level: Log level name @@ -99,22 +112,22 @@ def configure_logging( log_file: Path to log file console_output: Whether to log to console aws_context: AWS context information - + Returns: Configured logger instance """ # Convert log level string to logging level level = getattr(logging, log_level.upper(), logging.INFO) - + # Create logger logger = logging.getLogger(app_name) logger.setLevel(level) logger.handlers = [] # Remove any existing handlers - + # Set global AWS context if provided if aws_context: LoggingContext.set(**aws_context) - + # Create formatter if json_format: formatter = JSONFormatter() @@ -122,29 +135,33 @@ def configure_logging( formatter = logging.Formatter( "%(asctime)s [%(levelname)s] %(name)s - %(message)s" ) - + # Add console handler if console_output: console_handler = logging.StreamHandler(sys.stdout) console_handler.setFormatter(formatter) logger.addHandler(console_handler) - + # Add file handler if log_file: + # Ensure directory exists + log_dir = os.path.dirname(log_file) + if log_dir: + ensure_directory(log_dir) + file_handler = logging.FileHandler(log_file) file_handler.setFormatter(formatter) logger.addHandler(file_handler) - + # Prevent propagation to avoid duplicate logs logger.propagate = False - + return logger - def log_with_context(logger: logging.Logger, level: int, msg: str, **context) -> None: """ Log with additional context data. - + Args: logger: Logger to use level: Log level @@ -154,14 +171,13 @@ def log_with_context(logger: logging.Logger, level: int, msg: str, **context) -> # Add logging context combined_context = LoggingContext.get() combined_context.update(context) - + # Create a record with extra context extra = {"context": combined_context} - + # Log with context logger.log(level, msg, extra=extra) - @contextmanager def log_operation( logger: logging.Logger, @@ -173,7 +189,7 @@ def log_operation( ): """ Context manager to log the start, end, and any exceptions for an operation. - + Args: logger: Logger to use operation: Operation name @@ -184,21 +200,21 @@ def log_operation( """ success_level = success_level or level start_time = time.time() - + # Set operation context operation_context = { "operation": operation, "operation_status": "started", } operation_context.update(context) - + # Log start with context log_with_context(logger, level, f"Started {operation}", **operation_context) - + try: # Execute the operation yield - + # Log success with duration duration = time.time() - start_time operation_context.update( @@ -213,7 +229,6 @@ def log_operation( f"Completed {operation} in {duration:.3f}s", **operation_context, ) - except Exception as e: # Log failure with exception and duration duration = time.time() - start_time @@ -230,103 +245,5 @@ def log_operation( ) raise - -def initialize_csv_log(csv_file: Optional[str] = None) -> str: - """ - Initialize CSV log file with headers if it doesn't exist. - - Args: - csv_file: CSV file name (default: resource_actions.csv from config) - - Returns: - Path to the CSV log file - """ - config = get_config() - - if csv_file is None: - csv_file = config.get("action_csv_file") - - # Configure CSV logging directory - csv_log_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "logs") - csv_log_dir = ensure_csv_dir(csv_log_dir) - csv_path = os.path.join(csv_log_dir, csv_file) - - # Initialize with headers - headers = [ - "timestamp", - "account_id", - "account_name", - "resource_type", - "resource_id", - "resource_name", - "action", - "region", - "status", - "details", - "dry_run", - "schedule", - ] - - initialize_csv_file(csv_path, headers, overwrite=False) - return csv_path - - -def log_action_to_csv( - account_id: str, - account_name: str, - resource_type: str, - resource_id: str, - resource_name: str, - action: str, - region: str, - status: str, - details: str = "", - dry_run: bool = False, - existing_schedule: Optional[str] = None, -) -> None: - """ - Log resource action to CSV file for tracking. - - Args: - account_id: AWS account ID - account_name: AWS account name - resource_type: Resource type (ec2, rds, eks, emr) - resource_id: Resource ID - resource_name: Resource name - action: Action performed (stop, start) - region: AWS region - status: Action status - details: Additional details - dry_run: Whether this was a dry run - existing_schedule: Existing schedule for the resource - """ - timestamp = datetime.datetime.now().isoformat() - - # Initialize CSV log file - csv_path = initialize_csv_log() - - # Prepare row data - row_data = { - "timestamp": timestamp, - "account_id": account_id, - "account_name": account_name, - "resource_type": resource_type, - "resource_id": resource_id, - "resource_name": resource_name, - "action": action, - "region": region, - "status": status, - "details": details, - "dry_run": "Yes" if dry_run else "No", - "schedule": existing_schedule or "", - } - - # Headers must match the keys in row_data - headers = list(row_data.keys()) - - # Write to CSV - write_csv_row(csv_path, row_data, headers) - - -# Create a logger instance for direct import +# Create a default logger instance for direct import logger = setup_logging() diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/region_utils.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/region_utils.py new file mode 100644 index 00000000..50356719 --- /dev/null +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/region_utils.py @@ -0,0 +1,363 @@ +""" +AWS region and partition utilities. + +This module provides centralized functionality for working with AWS regions and partitions, +consolidating functionality previously spread across multiple modules. +""" +import logging +import time +from typing import Any, Dict, List, Optional, Set, Tuple, Union + +import boto3 + +# Centralized mapping of partitions to their regions +PARTITION_REGIONS = { + "aws-us-gov": ["us-gov-east-1", "us-gov-west-1"], + "aws-cn": ["cn-north-1", "cn-northwest-1"], + "aws": [ + "us-east-1", "us-east-2", "us-west-1", "us-west-2", + "eu-west-1", "eu-central-1", "eu-west-2", "eu-west-3", + "ap-northeast-1", "ap-northeast-2", "ap-southeast-1", "ap-southeast-2", + "ca-central-1", "sa-east-1", "eu-north-1", "eu-south-1", + "af-south-1", "ap-east-1", "ap-south-1", "me-south-1" + ], +} + +# Sensible defaults for each partition when no regions are specified +DEFAULT_REGIONS = { + "aws-us-gov": ["us-gov-east-1", "us-gov-west-1"], + "aws-cn": ["cn-north-1", "cn-northwest-1"], + "aws": ["us-east-1", "us-east-2", "us-west-1", "us-west-2"], +} + +# The default partition to use if no partition is detected +DEFAULT_PARTITION = "aws-us-gov" + +# Map of region prefixes to partitions for quick lookups +REGION_PREFIX_TO_PARTITION = { + "us-gov-": "aws-us-gov", + "gov-": "aws-us-gov", + "cn-": "aws-cn" +} + +# All known AWS regions (flattened from PARTITION_REGIONS) +ALL_KNOWN_REGIONS = [] +for regions in PARTITION_REGIONS.values(): + ALL_KNOWN_REGIONS.extend(regions) + +# Cache for region information +_region_cache = { + "timestamp": 0, + "all_regions": {}, + "enabled_regions": {}, + "partition_regions": PARTITION_REGIONS, +} + +# Cache expiration time in seconds (1 hour) +CACHE_EXPIRY = 3600 + +# Logger for this module +logger = logging.getLogger(__name__) + +def get_partition_for_region(region_name: str) -> str: + """ + Determine AWS partition based on region name. + + Args: + region_name: AWS region name + + Returns: + AWS partition name (e.g., 'aws', 'aws-us-gov', 'aws-cn') + """ + # Check common region prefixes first + for prefix, partition in REGION_PREFIX_TO_PARTITION.items(): + if region_name.startswith(prefix): + return partition + + # Default to standard AWS partition + return "aws" + +def get_regions_for_partition(partition: str) -> List[str]: + """ + Get the list of regions for a specific partition. + + Args: + partition: AWS partition name + + Returns: + List of region names in the partition + """ + return PARTITION_REGIONS.get(partition, PARTITION_REGIONS["aws"]) + +def get_default_regions_for_partition(partition: str) -> List[str]: + """ + Get the default regions commonly used for a partition. + + Args: + partition: AWS partition name + + Returns: + List of default region names + """ + return DEFAULT_REGIONS.get(partition, DEFAULT_REGIONS["aws"]) + +def is_region_in_partition(region: str, partition: str) -> bool: + """ + Check if a region belongs to the specified partition. + + Args: + region: AWS region name + partition: AWS partition name + + Returns: + True if the region is in the partition, False otherwise + """ + return get_partition_for_region(region) == partition + +def is_valid_region(region: str) -> bool: + """ + Check if a region is known to be valid without making an API call. + + Args: + region: Region name to validate + + Returns: + True if the region is valid, False otherwise + """ + if not isinstance(region, str): + return False + + # First check if it's in our known regions list + if region in ALL_KNOWN_REGIONS: + return True + + # Check if it follows expected naming patterns for newer regions + # Commercial regions follow patterns like us-east-1, eu-west-2 + if region.startswith(("us-", "eu-", "ap-", "sa-", "ca-", "me-", "af-")) and len(region.split("-")) >= 3: + return True + + # GovCloud regions + if region.startswith("us-gov-") and len(region.split("-")) >= 3: + return True + + # China regions + if region.startswith("cn-") and len(region.split("-")) >= 3: + return True + + return False + +def filter_regions_for_partition(regions: List[str], partition: str) -> List[str]: + """ + Filter regions to include only those belonging to the specified partition. + + Args: + regions: List of region names + partition: AWS partition name + + Returns: + List of valid region names within the partition + """ + # First validate the regions and filter out invalid ones + valid_regions = [r for r in regions if is_valid_region(r)] + + # Then filter for the specific partition + partition_regions = [r for r in valid_regions if is_region_in_partition(r, partition)] + + return partition_regions + +def detect_partition_from_credentials(credentials: Dict[str, str]) -> str: + """ + Detect partition from credentials with minimal API calls. + + Args: + credentials: Dictionary containing AWS credentials + + Returns: + AWS partition name + """ + if not credentials: + return DEFAULT_PARTITION # Default for environment + + # Check credential format first (fast, no API calls) + access_key = str(credentials.get("aws_access_key_id", "")).lower() + session_token = str(credentials.get("aws_session_token", "")).lower() + + # Check for GovCloud indicators + if any(indicator in access_key or indicator in session_token + for indicator in ["gov", "usgovcloud"]): + return "aws-us-gov" + + # Check for China indicators + if any(indicator in access_key or indicator in session_token + for indicator in ["cn-", "china"]): + return "aws-cn" + + # Try one API call to most likely partition based on environment + try: + boto3.client( + "sts", + region_name="us-gov-west-1", # Try GovCloud first based on environment + aws_access_key_id=credentials.get("aws_access_key_id"), + aws_secret_access_key=credentials.get("aws_secret_access_key"), + aws_session_token=credentials.get("aws_session_token"), + ).get_caller_identity() + return "aws-us-gov" + except Exception: + # If GovCloud fails, try standard AWS + try: + boto3.client( + "sts", + region_name="us-east-1", + aws_access_key_id=credentials.get("aws_access_key_id"), + aws_secret_access_key=credentials.get("aws_secret_access_key"), + aws_session_token=credentials.get("aws_session_token"), + ).get_caller_identity() + return "aws" + except Exception: + # Default to the environment's default partition + return DEFAULT_PARTITION + +def get_all_regions(partition: Optional[str] = None) -> List[str]: + """ + Get all AWS regions, using cache to minimize API calls. + + Args: + partition: Optional partition to filter regions by + + Returns: + List of region names + """ + # Use predefined regions for special partitions + if partition in ("aws-us-gov", "aws-cn"): + return get_regions_for_partition(partition) + + # Check cache first + current_time = time.time() + cache_key = partition or "all" + if ( + cache_key in _region_cache["all_regions"] + and current_time - _region_cache["timestamp"] < CACHE_EXPIRY + ): + return _region_cache["all_regions"][cache_key] + + # Cache miss: fetch regions + try: + ec2 = boto3.client("ec2", region_name="us-east-1") + all_regions = [ + region["RegionName"] for region in ec2.describe_regions()["Regions"] + ] + + # Cache the full list + _region_cache["all_regions"]["all"] = all_regions + _region_cache["timestamp"] = current_time + + # If partition specified, filter and cache that too + if partition: + filtered_regions = [ + r for r in all_regions + if isinstance(r, str) and get_partition_for_region(r) == partition + ] + _region_cache["all_regions"][partition] = filtered_regions + return filtered_regions + + return all_regions + except Exception as e: + logger.warning(f"Error getting regions: {str(e)}") + return get_regions_for_partition(partition or DEFAULT_PARTITION) + +def get_enabled_regions( + credentials: Dict[str, str], + partition: Optional[str] = None +) -> List[str]: + """ + Get enabled regions with minimized API calls using cache. + + Args: + credentials: Dictionary containing AWS credentials + partition: Optional partition to filter regions by + + Returns: + List of enabled region names + """ + # Determine partition if not specified + if not partition: + partition = detect_partition_from_credentials(credentials) + + # For GovCloud/China, return predefined regions to avoid API calls + if partition in ("aws-us-gov", "aws-cn"): + return get_regions_for_partition(partition) + + # Generate cache key from credentials + creds_hash = hash( + f"{credentials.get('aws_access_key_id', '')}:{credentials.get('aws_secret_access_key', '')}" + ) + cache_key = f"enabled_regions:{creds_hash}:{partition}" + + # Check cache + if ( + cache_key in _region_cache["enabled_regions"] + and time.time() - _region_cache["timestamp"] < CACHE_EXPIRY + ): + return _region_cache["enabled_regions"][cache_key] + + # Make a single API call to describe_regions rather than checking each region + try: + session = boto3.Session( + aws_access_key_id=credentials.get("aws_access_key_id"), + aws_secret_access_key=credentials.get("aws_secret_access_key"), + aws_session_token=credentials.get("aws_session_token"), + ) + regions = [ + region["RegionName"] + for region in session.client( + "ec2", region_name="us-east-1" + ).describe_regions()["Regions"] + ] + + # Cache the result + _region_cache["enabled_regions"][cache_key] = regions + _region_cache["timestamp"] = time.time() + + return regions + except Exception: + # Fall back to default regions + return get_default_regions_for_partition(partition) + +def list_enabled_regions( + session: boto3.Session, + exclude_regions: Optional[List[str]] = None +) -> List[str]: + """ + List all enabled regions for the account, respecting partition. + + Args: + session: Boto3 session + exclude_regions: List of regions to exclude + + Returns: + List of enabled region names + """ + if exclude_regions is None: + exclude_regions = [] + + try: + ec2 = session.client("ec2", region_name="us-east-1") # Use US East 1 for global operations + regions_response = ec2.describe_regions(AllRegions=False) # Only get enabled regions + regions = [ + r["RegionName"] + for r in regions_response["Regions"] + if r["RegionName"] not in exclude_regions + ] + return regions + except Exception as e: + logger.warning(f"Could not retrieve regions, using defaults: {str(e)}") + + # For GovCloud, provide sensible defaults + if session.region_name and session.region_name.startswith("us-gov-"): + return ["us-gov-east-1", "us-gov-west-1"] + + # Default to common commercial regions + return ["us-east-1", "us-east-2", "us-west-1", "us-west-2"] + +# Alias for backward compatibility +get_available_regions = get_enabled_regions diff --git a/local-app/python-tools/gfl-resource-actions/plan.md b/local-app/python-tools/gfl-resource-actions/plan.md index 7874143a..6276a040 100644 --- a/local-app/python-tools/gfl-resource-actions/plan.md +++ b/local-app/python-tools/gfl-resource-actions/plan.md @@ -1,3 +1,124 @@ +# AWS Resource Management Refactoring Plan + +## Phase 1: Core Module Creation ✅ + +We've successfully completed the first phase of the refactoring plan by creating a consolidated `utils` package with specialized utility modules: + +1. **aws_core_utils.py** - AWS credential management and session handling + - Credential acquisition and caching + - Session management + - Account discovery + +2. **region_utils.py** - Region and partition handling + - Region validation and filtering + - Partition detection + - Region/partition mapping + +3. **api_utils.py** - AWS API interaction patterns + - Pagination utilities + - Error handling helpers + - Resource normalization + +4. **file_utils.py** - File operations + - CSV handling + - JSON file operations + - Directory management + +5. **logging_utils.py** - Logging configuration + - Context-aware logging + - Operation logging with timing + - JSON formatting + +6. **config_utils.py** - Configuration management + - Config loading and merging + - Default configuration + - Config updates + +## Phase 2: Migration (Next Steps) + +The second phase involves updating imports throughout the codebase to use the new consolidated modules: + +1. Update import statements in: + - `core.py` + - `discovery.py` + - `reporting.py` + - `resource_registry.py` + - `managers/*.py` + +2. Add deprecation warnings to the old utility modules: + - Update `aws_utils.py` to import from new modules and add deprecation warning + - Update `discovery_utils.py` to import from new modules and add deprecation warning + - Update `csv_utils.py` to import from new modules and add deprecation warning + - Update `logging_setup.py` to import from new modules and add deprecation warning + - Update `config_manager.py` to import from new modules and add deprecation warning + +3. Add comprehensive tests: + - Create test cases for each new utility module + - Verify backward compatibility of functions + - Test in multiple AWS environments (Commercial, GovCloud, China) + +## Phase 3: Cleanup + +1. Remove deprecated modules: + - Remove original utility modules once all code has been updated to use the new modules + - Update documentation to reflect the new structure + - Clean up any remaining references + +2. Final validation: + - Run full test suite + - Verify all functionality in actual AWS environments + - Test with multiple AWS partitions + +## Implementation Notes + +1. **Backward Compatibility**: + - The new modules maintain the same function signatures as the original functions + - The `utils/__init__.py` file exposes commonly used functions for convenient import + - Old modules are kept but will import from new modules for backward compatibility + +2. **Improved Features**: + - Better error handling in API calls + - More consistent caching mechanisms + - Optimized pagination handling + - Clearer separation of concerns + - More comprehensive type annotations + +3. **Structure Benefits**: + - Eliminates circular dependencies + - Reduces code duplication + - Improves maintainability + - Simplifies future enhancements + +## Migration Guide for Developers + +When working with the AWS Resource Management codebase: + +1. **Prefer importing from `utils` package**: + ```python + # Old approach + from aws_resource_management.aws_utils import get_credentials + + # New approach + from aws_resource_management.utils import get_credentials + ``` + +2. **For specialized functions, import from specific module**: + ```python + # For region-specific functionality + from aws_resource_management.utils.region_utils import filter_regions_for_partition + + # For API-specific functionality + from aws_resource_management.utils.api_utils import retry_with_backoff + ``` + +3. **Use the comprehensive logging system**: + ```python + from aws_resource_management.utils import logger, log_operation + + with log_operation(logger, "Creating AWS session"): + session = get_session_for_account(account_id, region) + ``` + Step-by-Step Improvement Plan for AWS Resource Management Tool 1. ✅ Code Organization and Structure (COMPLETED) Currently, the codebase is well-organized with separate modules for different resource managers: